dzming.li
A vibe coded tangled fork which supports pijul.
1package state
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "log/slog"
9 "net/http"
10 "strings"
11 "time"
12
13 "tangled.org/core/api/tangled"
14 "tangled.org/core/appview"
15 "tangled.org/core/appview/config"
16 "tangled.org/core/appview/db"
17 "tangled.org/core/appview/indexer"
18 "tangled.org/core/appview/mentions"
19 "tangled.org/core/appview/models"
20 "tangled.org/core/appview/notify"
21 dbnotify "tangled.org/core/appview/notify/db"
22 phnotify "tangled.org/core/appview/notify/posthog"
23 "tangled.org/core/appview/oauth"
24 "tangled.org/core/appview/pages"
25 "tangled.org/core/appview/reporesolver"
26 "tangled.org/core/appview/validator"
27 xrpcclient "tangled.org/core/appview/xrpcclient"
28 "tangled.org/core/eventconsumer"
29 "tangled.org/core/idresolver"
30 "tangled.org/core/jetstream"
31 "tangled.org/core/log"
32 tlog "tangled.org/core/log"
33 "tangled.org/core/orm"
34 "tangled.org/core/rbac"
35 "tangled.org/core/tid"
36
37 comatproto "github.com/bluesky-social/indigo/api/atproto"
38 atpclient "github.com/bluesky-social/indigo/atproto/client"
39 "github.com/bluesky-social/indigo/atproto/syntax"
40 lexutil "github.com/bluesky-social/indigo/lex/util"
41 securejoin "github.com/cyphar/filepath-securejoin"
42 "github.com/go-chi/chi/v5"
43 "github.com/posthog/posthog-go"
44)
45
46type State struct {
47 db *db.DB
48 notifier notify.Notifier
49 indexer *indexer.Indexer
50 oauth *oauth.OAuth
51 enforcer *rbac.Enforcer
52 pages *pages.Pages
53 idResolver *idresolver.Resolver
54 mentionsResolver *mentions.Resolver
55 posthog posthog.Client
56 jc *jetstream.JetstreamClient
57 config *config.Config
58 repoResolver *reporesolver.RepoResolver
59 knotstream *eventconsumer.Consumer
60 spindlestream *eventconsumer.Consumer
61 logger *slog.Logger
62 validator *validator.Validator
63}
64
65func Make(ctx context.Context, config *config.Config) (*State, error) {
66 logger := tlog.FromContext(ctx)
67
68 d, err := db.Make(ctx, config.Core.DbPath)
69 if err != nil {
70 return nil, fmt.Errorf("failed to create db: %w", err)
71 }
72
73 indexer := indexer.New(log.SubLogger(logger, "indexer"))
74 err = indexer.Init(ctx, d)
75 if err != nil {
76 return nil, fmt.Errorf("failed to create indexer: %w", err)
77 }
78
79 enforcer, err := rbac.NewEnforcer(config.Core.DbPath)
80 if err != nil {
81 return nil, fmt.Errorf("failed to create enforcer: %w", err)
82 }
83
84 res, err := idresolver.RedisResolver(config.Redis.ToURL(), config.Plc.PLCURL)
85 if err != nil {
86 logger.Error("failed to create redis resolver", "err", err)
87 res = idresolver.DefaultResolver(config.Plc.PLCURL)
88 }
89
90 posthog, err := posthog.NewWithConfig(config.Posthog.ApiKey, posthog.Config{Endpoint: config.Posthog.Endpoint})
91 if err != nil {
92 return nil, fmt.Errorf("failed to create posthog client: %w", err)
93 }
94
95 pages := pages.NewPages(config, res, d, log.SubLogger(logger, "pages"))
96 oauth, err := oauth.New(config, posthog, d, enforcer, res, log.SubLogger(logger, "oauth"))
97 if err != nil {
98 return nil, fmt.Errorf("failed to start oauth handler: %w", err)
99 }
100 validator := validator.New(d, res, enforcer)
101
102 repoResolver := reporesolver.New(config, enforcer, d)
103
104 mentionsResolver := mentions.New(config, res, d, log.SubLogger(logger, "mentionsResolver"))
105
106 wrapper := db.DbWrapper{Execer: d}
107 jc, err := jetstream.NewJetstreamClient(
108 config.Jetstream.Endpoint,
109 "appview",
110 []string{
111 tangled.GraphFollowNSID,
112 tangled.FeedStarNSID,
113 tangled.PublicKeyNSID,
114 tangled.RepoArtifactNSID,
115 tangled.ActorProfileNSID,
116 tangled.SpindleMemberNSID,
117 tangled.SpindleNSID,
118 tangled.StringNSID,
119 tangled.RepoIssueNSID,
120 tangled.RepoIssueCommentNSID,
121 tangled.LabelDefinitionNSID,
122 tangled.LabelOpNSID,
123 },
124 nil,
125 tlog.SubLogger(logger, "jetstream"),
126 wrapper,
127 false,
128
129 // in-memory filter is inapplicalble to appview so
130 // we'll never log dids anyway.
131 false,
132 )
133 if err != nil {
134 return nil, fmt.Errorf("failed to create jetstream client: %w", err)
135 }
136
137 if err := BackfillDefaultDefs(d, res, config.Label.DefaultLabelDefs); err != nil {
138 return nil, fmt.Errorf("failed to backfill default label defs: %w", err)
139 }
140
141 ingester := appview.Ingester{
142 Db: wrapper,
143 Enforcer: enforcer,
144 IdResolver: res,
145 Config: config,
146 Logger: log.SubLogger(logger, "ingester"),
147 Validator: validator,
148 }
149 err = jc.StartJetstream(ctx, ingester.Ingest())
150 if err != nil {
151 return nil, fmt.Errorf("failed to start jetstream watcher: %w", err)
152 }
153
154 knotstream, err := Knotstream(ctx, config, d, enforcer, posthog)
155 if err != nil {
156 return nil, fmt.Errorf("failed to start knotstream consumer: %w", err)
157 }
158 knotstream.Start(ctx)
159
160 spindlestream, err := Spindlestream(ctx, config, d, enforcer)
161 if err != nil {
162 return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err)
163 }
164 spindlestream.Start(ctx)
165
166 var notifiers []notify.Notifier
167
168 // Always add the database notifier
169 notifiers = append(notifiers, dbnotify.NewDatabaseNotifier(d, res))
170
171 // Add other notifiers in production only
172 if !config.Core.Dev {
173 notifiers = append(notifiers, phnotify.NewPosthogNotifier(posthog))
174 }
175 notifiers = append(notifiers, indexer)
176 notifier := notify.NewMergedNotifier(notifiers, tlog.SubLogger(logger, "notify"))
177
178 state := &State{
179 d,
180 notifier,
181 indexer,
182 oauth,
183 enforcer,
184 pages,
185 res,
186 mentionsResolver,
187 posthog,
188 jc,
189 config,
190 repoResolver,
191 knotstream,
192 spindlestream,
193 logger,
194 validator,
195 }
196
197 return state, nil
198}
199
200func (s *State) Close() error {
201 // other close up logic goes here
202 return s.db.Close()
203}
204
205func (s *State) RobotsTxt(w http.ResponseWriter, r *http.Request) {
206 w.Header().Set("Content-Type", "text/plain")
207 w.Header().Set("Cache-Control", "public, max-age=86400") // one day
208
209 robotsTxt := `User-agent: *
210Allow: /
211Disallow: /settings
212Disallow: /notifications
213Disallow: /login
214Disallow: /logout
215Disallow: /signup
216Disallow: /oauth
217Disallow: */settings$
218Disallow: */settings/*
219
220Crawl-delay: 1
221
222Sitemap: https://tangled.org/sitemap.xml
223`
224 w.Write([]byte(robotsTxt))
225}
226
227func (s *State) Sitemap(w http.ResponseWriter, r *http.Request) {
228 w.Header().Set("Content-Type", "application/xml; charset=utf-8")
229 w.Header().Set("Cache-Control", "public, max-age=3600")
230
231 // basic sitemap with static pages
232 sitemap := `<?xml version="1.0" encoding="UTF-8"?>
233<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
234 <url>
235 <loc>https://tangled.org</loc>
236 <changefreq>daily</changefreq>
237 <priority>1.0</priority>
238 </url>
239 <url>
240 <loc>https://tangled.org/timeline</loc>
241 <changefreq>hourly</changefreq>
242 <priority>0.9</priority>
243 </url>
244 <url>
245 <loc>https://tangled.org/goodfirstissues</loc>
246 <changefreq>daily</changefreq>
247 <priority>0.8</priority>
248 </url>
249 <url>
250 <loc>https://tangled.org/terms</loc>
251 <changefreq>monthly</changefreq>
252 <priority>0.3</priority>
253 </url>
254 <url>
255 <loc>https://tangled.org/privacy</loc>
256 <changefreq>monthly</changefreq>
257 <priority>0.3</priority>
258 </url>
259 <url>
260 <loc>https://tangled.org/brand</loc>
261 <changefreq>monthly</changefreq>
262 <priority>0.5</priority>
263 </url>
264</urlset>`
265 w.Write([]byte(sitemap))
266}
267
268// https://developer.mozilla.org/en-US/docs/Web/Progressive_web_apps/Manifest
269const manifestJson = `{
270 "name": "tangled",
271 "description": "tightly-knit social coding.",
272 "icons": [
273 {
274 "src": "/favicon.svg",
275 "sizes": "144x144"
276 }
277 ],
278 "start_url": "/",
279 "id": "org.tangled",
280
281 "display": "standalone",
282 "background_color": "#111827",
283 "theme_color": "#111827"
284}`
285
286func (p *State) PWAManifest(w http.ResponseWriter, r *http.Request) {
287 w.Header().Set("Content-Type", "application/json")
288 w.Write([]byte(manifestJson))
289}
290
291func (s *State) TermsOfService(w http.ResponseWriter, r *http.Request) {
292 user := s.oauth.GetMultiAccountUser(r)
293 s.pages.TermsOfService(w, pages.TermsOfServiceParams{
294 LoggedInUser: user,
295 })
296}
297
298func (s *State) PrivacyPolicy(w http.ResponseWriter, r *http.Request) {
299 user := s.oauth.GetMultiAccountUser(r)
300 s.pages.PrivacyPolicy(w, pages.PrivacyPolicyParams{
301 LoggedInUser: user,
302 })
303}
304
305func (s *State) Brand(w http.ResponseWriter, r *http.Request) {
306 user := s.oauth.GetMultiAccountUser(r)
307 s.pages.Brand(w, pages.BrandParams{
308 LoggedInUser: user,
309 })
310}
311
312func (s *State) HomeOrTimeline(w http.ResponseWriter, r *http.Request) {
313 if s.oauth.GetMultiAccountUser(r) != nil {
314 s.Timeline(w, r)
315 return
316 }
317 s.Home(w, r)
318}
319
320func (s *State) Timeline(w http.ResponseWriter, r *http.Request) {
321 user := s.oauth.GetMultiAccountUser(r)
322
323 // TODO: set this flag based on the UI
324 filtered := false
325
326 var userDid string
327 if user != nil && user.Active != nil {
328 userDid = user.Active.Did
329 }
330 timeline, err := db.MakeTimeline(s.db, 50, userDid, filtered)
331 if err != nil {
332 s.logger.Error("failed to make timeline", "err", err)
333 s.pages.Notice(w, "timeline", "Uh oh! Failed to load timeline.")
334 }
335
336 repos, err := db.GetTopStarredReposLastWeek(s.db)
337 if err != nil {
338 s.logger.Error("failed to get top starred repos", "err", err)
339 s.pages.Notice(w, "topstarredrepos", "Unable to load.")
340 return
341 }
342
343 gfiLabel, err := db.GetLabelDefinition(s.db, orm.FilterEq("at_uri", s.config.Label.GoodFirstIssue))
344 if err != nil {
345 // non-fatal
346 }
347
348 s.pages.Timeline(w, pages.TimelineParams{
349 LoggedInUser: user,
350 Timeline: timeline,
351 Repos: repos,
352 GfiLabel: gfiLabel,
353 })
354}
355
356func (s *State) UpgradeBanner(w http.ResponseWriter, r *http.Request) {
357 user := s.oauth.GetMultiAccountUser(r)
358 if user == nil {
359 return
360 }
361
362 l := s.logger.With("handler", "UpgradeBanner")
363 l = l.With("did", user.Active.Did)
364
365 regs, err := db.GetRegistrations(
366 s.db,
367 orm.FilterEq("did", user.Active.Did),
368 orm.FilterEq("needs_upgrade", 1),
369 )
370 if err != nil {
371 l.Error("non-fatal: failed to get registrations", "err", err)
372 }
373
374 spindles, err := db.GetSpindles(
375 s.db,
376 orm.FilterEq("owner", user.Active.Did),
377 orm.FilterEq("needs_upgrade", 1),
378 )
379 if err != nil {
380 l.Error("non-fatal: failed to get spindles", "err", err)
381 }
382
383 if regs == nil && spindles == nil {
384 return
385 }
386
387 s.pages.UpgradeBanner(w, pages.UpgradeBannerParams{
388 Registrations: regs,
389 Spindles: spindles,
390 })
391}
392
393func (s *State) Home(w http.ResponseWriter, r *http.Request) {
394 // TODO: set this flag based on the UI
395 filtered := false
396
397 timeline, err := db.MakeTimeline(s.db, 5, "", filtered)
398 if err != nil {
399 s.logger.Error("failed to make timeline", "err", err)
400 s.pages.Notice(w, "timeline", "Uh oh! Failed to load timeline.")
401 return
402 }
403
404 repos, err := db.GetTopStarredReposLastWeek(s.db)
405 if err != nil {
406 s.logger.Error("failed to get top starred repos", "err", err)
407 s.pages.Notice(w, "topstarredrepos", "Unable to load.")
408 return
409 }
410
411 s.pages.Home(w, pages.TimelineParams{
412 LoggedInUser: nil,
413 Timeline: timeline,
414 Repos: repos,
415 })
416}
417
418func (s *State) Keys(w http.ResponseWriter, r *http.Request) {
419 user := chi.URLParam(r, "user")
420 user = strings.TrimPrefix(user, "@")
421
422 if user == "" {
423 w.WriteHeader(http.StatusBadRequest)
424 return
425 }
426
427 id, err := s.idResolver.ResolveIdent(r.Context(), user)
428 if err != nil {
429 w.WriteHeader(http.StatusInternalServerError)
430 return
431 }
432
433 pubKeys, err := db.GetPublicKeysForDid(s.db, id.DID.String())
434 if err != nil {
435 s.logger.Error("failed to get public keys", "err", err)
436 http.Error(w, "failed to get public keys", http.StatusInternalServerError)
437 return
438 }
439
440 if len(pubKeys) == 0 {
441 w.WriteHeader(http.StatusNoContent)
442 return
443 }
444
445 for _, k := range pubKeys {
446 key := strings.TrimRight(k.Key, "\n")
447 fmt.Fprintln(w, key)
448 }
449}
450
451func validateRepoName(name string) error {
452 // check for path traversal attempts
453 if name == "." || name == ".." ||
454 strings.Contains(name, "/") || strings.Contains(name, "\\") {
455 return fmt.Errorf("Repository name contains invalid path characters")
456 }
457
458 // check for sequences that could be used for traversal when normalized
459 if strings.Contains(name, "./") || strings.Contains(name, "../") ||
460 strings.HasPrefix(name, ".") || strings.HasSuffix(name, ".") {
461 return fmt.Errorf("Repository name contains invalid path sequence")
462 }
463
464 // then continue with character validation
465 for _, char := range name {
466 if !((char >= 'a' && char <= 'z') ||
467 (char >= 'A' && char <= 'Z') ||
468 (char >= '0' && char <= '9') ||
469 char == '-' || char == '_' || char == '.') {
470 return fmt.Errorf("Repository name can only contain alphanumeric characters, periods, hyphens, and underscores")
471 }
472 }
473
474 // additional check to prevent multiple sequential dots
475 if strings.Contains(name, "..") {
476 return fmt.Errorf("Repository name cannot contain sequential dots")
477 }
478
479 // if all checks pass
480 return nil
481}
482
483func stripGitExt(name string) string {
484 return strings.TrimSuffix(name, ".git")
485}
486
487func (s *State) NewRepo(w http.ResponseWriter, r *http.Request) {
488 switch r.Method {
489 case http.MethodGet:
490 user := s.oauth.GetMultiAccountUser(r)
491 knots, err := s.enforcer.GetKnotsForUser(user.Active.Did)
492 if err != nil {
493 s.pages.Notice(w, "repo", "Invalid user account.")
494 return
495 }
496
497 s.pages.NewRepo(w, pages.NewRepoParams{
498 LoggedInUser: user,
499 Knots: knots,
500 })
501
502 case http.MethodPost:
503 l := s.logger.With("handler", "NewRepo")
504
505 user := s.oauth.GetMultiAccountUser(r)
506 l = l.With("did", user.Active.Did)
507
508 // form validation
509 domain := r.FormValue("domain")
510 if domain == "" {
511 s.pages.Notice(w, "repo", "Invalid form submission—missing knot domain.")
512 return
513 }
514 l = l.With("knot", domain)
515
516 repoName := r.FormValue("name")
517 if repoName == "" {
518 s.pages.Notice(w, "repo", "Repository name cannot be empty.")
519 return
520 }
521
522 if err := validateRepoName(repoName); err != nil {
523 s.pages.Notice(w, "repo", err.Error())
524 return
525 }
526 repoName = stripGitExt(repoName)
527 l = l.With("repoName", repoName)
528
529 defaultBranch := r.FormValue("branch")
530 if defaultBranch == "" {
531 defaultBranch = "main"
532 }
533 l = l.With("defaultBranch", defaultBranch)
534
535 description := r.FormValue("description")
536
537 // ACL validation
538 ok, err := s.enforcer.E.Enforce(user.Active.Did, domain, domain, "repo:create")
539 if err != nil || !ok {
540 l.Info("unauthorized")
541 s.pages.Notice(w, "repo", "You do not have permission to create a repo in this knot.")
542 return
543 }
544
545 // Check for existing repos
546 existingRepo, err := db.GetRepo(
547 s.db,
548 orm.FilterEq("did", user.Active.Did),
549 orm.FilterEq("name", repoName),
550 )
551 if err == nil && existingRepo != nil {
552 l.Info("repo exists")
553 s.pages.Notice(w, "repo", fmt.Sprintf("You already have a repository by this name on %s", existingRepo.Knot))
554 return
555 }
556
557 // create atproto record for this repo
558 rkey := tid.TID()
559 repo := &models.Repo{
560 Did: user.Active.Did,
561 Name: repoName,
562 Knot: domain,
563 Rkey: rkey,
564 Description: description,
565 Created: time.Now(),
566 Labels: s.config.Label.DefaultLabelDefs,
567 }
568 record := repo.AsRecord()
569
570 atpClient, err := s.oauth.AuthorizedClient(r)
571 if err != nil {
572 l.Info("PDS write failed", "err", err)
573 s.pages.Notice(w, "repo", "Failed to write record to PDS.")
574 return
575 }
576
577 atresp, err := comatproto.RepoPutRecord(r.Context(), atpClient, &comatproto.RepoPutRecord_Input{
578 Collection: tangled.RepoNSID,
579 Repo: user.Active.Did,
580 Rkey: rkey,
581 Record: &lexutil.LexiconTypeDecoder{
582 Val: &record,
583 },
584 })
585 if err != nil {
586 l.Info("PDS write failed", "err", err)
587 s.pages.Notice(w, "repo", "Failed to announce repository creation.")
588 return
589 }
590
591 aturi := atresp.Uri
592 l = l.With("aturi", aturi)
593 l.Info("wrote to PDS")
594
595 tx, err := s.db.BeginTx(r.Context(), nil)
596 if err != nil {
597 l.Info("txn failed", "err", err)
598 s.pages.Notice(w, "repo", "Failed to save repository information.")
599 return
600 }
601
602 // The rollback function reverts a few things on failure:
603 // - the pending txn
604 // - the ACLs
605 // - the atproto record created
606 rollback := func() {
607 err1 := tx.Rollback()
608 err2 := s.enforcer.E.LoadPolicy()
609 err3 := rollbackRecord(context.Background(), aturi, atpClient)
610
611 // ignore txn complete errors, this is okay
612 if errors.Is(err1, sql.ErrTxDone) {
613 err1 = nil
614 }
615
616 if errs := errors.Join(err1, err2, err3); errs != nil {
617 l.Error("failed to rollback changes", "errs", errs)
618 return
619 }
620 }
621 defer rollback()
622
623 client, err := s.oauth.ServiceClient(
624 r,
625 oauth.WithService(domain),
626 oauth.WithLxm(tangled.RepoCreateNSID),
627 oauth.WithDev(s.config.Core.Dev),
628 )
629 if err != nil {
630 l.Error("service auth failed", "err", err)
631 s.pages.Notice(w, "repo", "Failed to reach PDS.")
632 return
633 }
634
635 xe := tangled.RepoCreate(
636 r.Context(),
637 client,
638 &tangled.RepoCreate_Input{
639 Rkey: rkey,
640 },
641 )
642 if err := xrpcclient.HandleXrpcErr(xe); err != nil {
643 l.Error("xrpc error", "xe", xe)
644 s.pages.Notice(w, "repo", err.Error())
645 return
646 }
647
648 err = db.AddRepo(tx, repo)
649 if err != nil {
650 l.Error("db write failed", "err", err)
651 s.pages.Notice(w, "repo", "Failed to save repository information.")
652 return
653 }
654
655 // acls
656 p, _ := securejoin.SecureJoin(user.Active.Did, repoName)
657 err = s.enforcer.AddRepo(user.Active.Did, domain, p)
658 if err != nil {
659 l.Error("acl setup failed", "err", err)
660 s.pages.Notice(w, "repo", "Failed to set up repository permissions.")
661 return
662 }
663
664 err = tx.Commit()
665 if err != nil {
666 l.Error("txn commit failed", "err", err)
667 http.Error(w, err.Error(), http.StatusInternalServerError)
668 return
669 }
670
671 err = s.enforcer.E.SavePolicy()
672 if err != nil {
673 l.Error("acl save failed", "err", err)
674 http.Error(w, err.Error(), http.StatusInternalServerError)
675 return
676 }
677
678 // reset the ATURI because the transaction completed successfully
679 aturi = ""
680
681 s.notifier.NewRepo(r.Context(), repo)
682 s.pages.HxLocation(w, fmt.Sprintf("/%s/%s", user.Active.Did, repoName))
683 }
684}
685
686// this is used to rollback changes made to the PDS
687//
688// it is a no-op if the provided ATURI is empty
689func rollbackRecord(ctx context.Context, aturi string, client *atpclient.APIClient) error {
690 if aturi == "" {
691 return nil
692 }
693
694 parsed := syntax.ATURI(aturi)
695
696 collection := parsed.Collection().String()
697 repo := parsed.Authority().String()
698 rkey := parsed.RecordKey().String()
699
700 _, err := comatproto.RepoDeleteRecord(ctx, client, &comatproto.RepoDeleteRecord_Input{
701 Collection: collection,
702 Repo: repo,
703 Rkey: rkey,
704 })
705 return err
706}
707
708func BackfillDefaultDefs(e db.Execer, r *idresolver.Resolver, defaults []string) error {
709 defaultLabels, err := db.GetLabelDefinitions(e, orm.FilterIn("at_uri", defaults))
710 if err != nil {
711 return err
712 }
713 // already present
714 if len(defaultLabels) == len(defaults) {
715 return nil
716 }
717
718 labelDefs, err := models.FetchLabelDefs(r, defaults)
719 if err != nil {
720 return err
721 }
722
723 // Insert each label definition to the database
724 for _, labelDef := range labelDefs {
725 _, err = db.AddLabelDefinition(e, &labelDef)
726 if err != nil {
727 return fmt.Errorf("failed to add label definition %s: %v", labelDef.Name, err)
728 }
729 }
730
731 return nil
732}