lewis.moe
Monorepo for Tangled
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 inapplicable 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 var notifiers []notify.Notifier
155
156 // Always add the database notifier
157 notifiers = append(notifiers, dbnotify.NewDatabaseNotifier(d, res))
158
159 // Add other notifiers in production only
160 if !config.Core.Dev {
161 notifiers = append(notifiers, phnotify.NewPosthogNotifier(posthog))
162 }
163 notifiers = append(notifiers, indexer)
164
165 // Add webhook notifier
166 notifiers = append(notifiers, notify.NewWebhookNotifier(d))
167
168 notifier := notify.NewMergedNotifier(notifiers)
169 notifier = notify.NewLoggingNotifier(notifier, tlog.SubLogger(logger, "notify"))
170
171 knotstream, err := Knotstream(ctx, config, d, enforcer, posthog, notifier)
172 if err != nil {
173 return nil, fmt.Errorf("failed to start knotstream consumer: %w", err)
174 }
175 knotstream.Start(ctx)
176
177 spindlestream, err := Spindlestream(ctx, config, d, enforcer)
178 if err != nil {
179 return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err)
180 }
181 spindlestream.Start(ctx)
182
183 state := &State{
184 d,
185 notifier,
186 indexer,
187 oauth,
188 enforcer,
189 pages,
190 res,
191 mentionsResolver,
192 posthog,
193 jc,
194 config,
195 repoResolver,
196 knotstream,
197 spindlestream,
198 logger,
199 validator,
200 }
201
202 return state, nil
203}
204
205func (s *State) Close() error {
206 // other close up logic goes here
207 return s.db.Close()
208}
209
210func (s *State) RobotsTxt(w http.ResponseWriter, r *http.Request) {
211 w.Header().Set("Content-Type", "text/plain")
212 w.Header().Set("Cache-Control", "public, max-age=86400") // one day
213
214 robotsTxt := `# Hello, Tanglers!
215User-agent: *
216Allow: /
217Disallow: /*/*/settings
218Disallow: /settings
219Disallow: /*/*/compare
220Disallow: /*/*/fork
221
222Crawl-delay: 1
223`
224 w.Write([]byte(robotsTxt))
225}
226
227func (s *State) TermsOfService(w http.ResponseWriter, r *http.Request) {
228 user := s.oauth.GetMultiAccountUser(r)
229 s.pages.TermsOfService(w, pages.TermsOfServiceParams{
230 LoggedInUser: user,
231 })
232}
233
234func (s *State) PrivacyPolicy(w http.ResponseWriter, r *http.Request) {
235 user := s.oauth.GetMultiAccountUser(r)
236 s.pages.PrivacyPolicy(w, pages.PrivacyPolicyParams{
237 LoggedInUser: user,
238 })
239}
240
241func (s *State) Brand(w http.ResponseWriter, r *http.Request) {
242 user := s.oauth.GetMultiAccountUser(r)
243 s.pages.Brand(w, pages.BrandParams{
244 LoggedInUser: user,
245 })
246}
247
248func (s *State) HomeOrTimeline(w http.ResponseWriter, r *http.Request) {
249 if s.oauth.GetMultiAccountUser(r) != nil {
250 s.Timeline(w, r)
251 return
252 }
253 s.Home(w, r)
254}
255
256func (s *State) Timeline(w http.ResponseWriter, r *http.Request) {
257 user := s.oauth.GetMultiAccountUser(r)
258
259 // TODO: set this flag based on the UI
260 filtered := false
261
262 var userDid string
263 if user != nil && user.Active != nil {
264 userDid = user.Active.Did
265 }
266 timeline, err := db.MakeTimeline(s.db, 50, userDid, filtered)
267 if err != nil {
268 s.logger.Error("failed to make timeline", "err", err)
269 s.pages.Notice(w, "timeline", "Uh oh! Failed to load timeline.")
270 }
271
272 repos, err := db.GetTopStarredReposLastWeek(s.db)
273 if err != nil {
274 s.logger.Error("failed to get top starred repos", "err", err)
275 s.pages.Notice(w, "topstarredrepos", "Unable to load.")
276 return
277 }
278
279 gfiLabel, err := db.GetLabelDefinition(s.db, orm.FilterEq("at_uri", s.config.Label.GoodFirstIssue))
280 if err != nil {
281 // non-fatal
282 }
283
284 s.pages.Timeline(w, pages.TimelineParams{
285 LoggedInUser: user,
286 Timeline: timeline,
287 Repos: repos,
288 GfiLabel: gfiLabel,
289 })
290}
291
292func (s *State) UpgradeBanner(w http.ResponseWriter, r *http.Request) {
293 user := s.oauth.GetMultiAccountUser(r)
294 if user == nil {
295 return
296 }
297
298 l := s.logger.With("handler", "UpgradeBanner")
299 l = l.With("did", user.Active.Did)
300
301 regs, err := db.GetRegistrations(
302 s.db,
303 orm.FilterEq("did", user.Active.Did),
304 orm.FilterEq("needs_upgrade", 1),
305 )
306 if err != nil {
307 l.Error("non-fatal: failed to get registrations", "err", err)
308 }
309
310 spindles, err := db.GetSpindles(
311 s.db,
312 orm.FilterEq("owner", user.Active.Did),
313 orm.FilterEq("needs_upgrade", 1),
314 )
315 if err != nil {
316 l.Error("non-fatal: failed to get spindles", "err", err)
317 }
318
319 if regs == nil && spindles == nil {
320 return
321 }
322
323 s.pages.UpgradeBanner(w, pages.UpgradeBannerParams{
324 Registrations: regs,
325 Spindles: spindles,
326 })
327}
328
329func (s *State) Home(w http.ResponseWriter, r *http.Request) {
330 // TODO: set this flag based on the UI
331 filtered := false
332
333 timeline, err := db.MakeTimeline(s.db, 5, "", filtered)
334 if err != nil {
335 s.logger.Error("failed to make timeline", "err", err)
336 s.pages.Notice(w, "timeline", "Uh oh! Failed to load timeline.")
337 return
338 }
339
340 repos, err := db.GetTopStarredReposLastWeek(s.db)
341 if err != nil {
342 s.logger.Error("failed to get top starred repos", "err", err)
343 s.pages.Notice(w, "topstarredrepos", "Unable to load.")
344 return
345 }
346
347 s.pages.Home(w, pages.TimelineParams{
348 LoggedInUser: nil,
349 Timeline: timeline,
350 Repos: repos,
351 })
352}
353
354func (s *State) Keys(w http.ResponseWriter, r *http.Request) {
355 user := chi.URLParam(r, "user")
356 user = strings.TrimPrefix(user, "@")
357
358 if user == "" {
359 w.WriteHeader(http.StatusBadRequest)
360 return
361 }
362
363 id, err := s.idResolver.ResolveIdent(r.Context(), user)
364 if err != nil {
365 w.WriteHeader(http.StatusInternalServerError)
366 return
367 }
368
369 pubKeys, err := db.GetPublicKeysForDid(s.db, id.DID.String())
370 if err != nil {
371 s.logger.Error("failed to get public keys", "err", err)
372 http.Error(w, "failed to get public keys", http.StatusInternalServerError)
373 return
374 }
375
376 if len(pubKeys) == 0 {
377 w.WriteHeader(http.StatusNoContent)
378 return
379 }
380
381 for _, k := range pubKeys {
382 key := strings.TrimRight(k.Key, "\n")
383 fmt.Fprintln(w, key)
384 }
385}
386
387func validateRepoName(name string) error {
388 // check for path traversal attempts
389 if name == "." || name == ".." ||
390 strings.Contains(name, "/") || strings.Contains(name, "\\") {
391 return fmt.Errorf("Repository name contains invalid path characters")
392 }
393
394 // check for sequences that could be used for traversal when normalized
395 if strings.Contains(name, "./") || strings.Contains(name, "../") ||
396 strings.HasPrefix(name, ".") || strings.HasSuffix(name, ".") {
397 return fmt.Errorf("Repository name contains invalid path sequence")
398 }
399
400 // then continue with character validation
401 for _, char := range name {
402 if !((char >= 'a' && char <= 'z') ||
403 (char >= 'A' && char <= 'Z') ||
404 (char >= '0' && char <= '9') ||
405 char == '-' || char == '_' || char == '.') {
406 return fmt.Errorf("Repository name can only contain alphanumeric characters, periods, hyphens, and underscores")
407 }
408 }
409
410 // additional check to prevent multiple sequential dots
411 if strings.Contains(name, "..") {
412 return fmt.Errorf("Repository name cannot contain sequential dots")
413 }
414
415 // if all checks pass
416 return nil
417}
418
419func stripGitExt(name string) string {
420 return strings.TrimSuffix(name, ".git")
421}
422
423func (s *State) NewRepo(w http.ResponseWriter, r *http.Request) {
424 switch r.Method {
425 case http.MethodGet:
426 user := s.oauth.GetMultiAccountUser(r)
427 knots, err := s.enforcer.GetKnotsForUser(user.Active.Did)
428 if err != nil {
429 s.pages.Notice(w, "repo", "Invalid user account.")
430 return
431 }
432
433 s.pages.NewRepo(w, pages.NewRepoParams{
434 LoggedInUser: user,
435 Knots: knots,
436 })
437
438 case http.MethodPost:
439 l := s.logger.With("handler", "NewRepo")
440
441 user := s.oauth.GetMultiAccountUser(r)
442 l = l.With("did", user.Active.Did)
443
444 // form validation
445 domain := r.FormValue("domain")
446 if domain == "" {
447 s.pages.Notice(w, "repo", "Invalid form submission—missing knot domain.")
448 return
449 }
450 l = l.With("knot", domain)
451
452 repoName := r.FormValue("name")
453 if repoName == "" {
454 s.pages.Notice(w, "repo", "Repository name cannot be empty.")
455 return
456 }
457
458 if err := validateRepoName(repoName); err != nil {
459 s.pages.Notice(w, "repo", err.Error())
460 return
461 }
462 repoName = stripGitExt(repoName)
463 l = l.With("repoName", repoName)
464
465 defaultBranch := r.FormValue("branch")
466 if defaultBranch == "" {
467 defaultBranch = "main"
468 }
469 l = l.With("defaultBranch", defaultBranch)
470
471 description := r.FormValue("description")
472
473 // ACL validation
474 ok, err := s.enforcer.E.Enforce(user.Active.Did, domain, domain, "repo:create")
475 if err != nil || !ok {
476 l.Info("unauthorized")
477 s.pages.Notice(w, "repo", "You do not have permission to create a repo in this knot.")
478 return
479 }
480
481 // Check for existing repos
482 existingRepo, err := db.GetRepo(
483 s.db,
484 orm.FilterEq("did", user.Active.Did),
485 orm.FilterEq("name", repoName),
486 )
487 if err == nil && existingRepo != nil {
488 l.Info("repo exists")
489 s.pages.Notice(w, "repo", fmt.Sprintf("You already have a repository by this name on %s", existingRepo.Knot))
490 return
491 }
492
493 // create atproto record for this repo
494 rkey := tid.TID()
495 repo := &models.Repo{
496 Did: user.Active.Did,
497 Name: repoName,
498 Knot: domain,
499 Rkey: rkey,
500 Description: description,
501 Created: time.Now(),
502 Labels: s.config.Label.DefaultLabelDefs,
503 }
504 record := repo.AsRecord()
505
506 atpClient, err := s.oauth.AuthorizedClient(r)
507 if err != nil {
508 l.Info("PDS write failed", "err", err)
509 s.pages.Notice(w, "repo", "Failed to write record to PDS.")
510 return
511 }
512
513 atresp, err := comatproto.RepoPutRecord(r.Context(), atpClient, &comatproto.RepoPutRecord_Input{
514 Collection: tangled.RepoNSID,
515 Repo: user.Active.Did,
516 Rkey: rkey,
517 Record: &lexutil.LexiconTypeDecoder{
518 Val: &record,
519 },
520 })
521 if err != nil {
522 l.Info("PDS write failed", "err", err)
523 s.pages.Notice(w, "repo", "Failed to announce repository creation.")
524 return
525 }
526
527 aturi := atresp.Uri
528 l = l.With("aturi", aturi)
529 l.Info("wrote to PDS")
530
531 tx, err := s.db.BeginTx(r.Context(), nil)
532 if err != nil {
533 l.Info("txn failed", "err", err)
534 s.pages.Notice(w, "repo", "Failed to save repository information.")
535 return
536 }
537
538 // The rollback function reverts a few things on failure:
539 // - the pending txn
540 // - the ACLs
541 // - the atproto record created
542 rollback := func() {
543 err1 := tx.Rollback()
544 err2 := s.enforcer.E.LoadPolicy()
545 err3 := rollbackRecord(context.Background(), aturi, atpClient)
546
547 // ignore txn complete errors, this is okay
548 if errors.Is(err1, sql.ErrTxDone) {
549 err1 = nil
550 }
551
552 if errs := errors.Join(err1, err2, err3); errs != nil {
553 l.Error("failed to rollback changes", "errs", errs)
554 return
555 }
556 }
557 defer rollback()
558
559 client, err := s.oauth.ServiceClient(
560 r,
561 oauth.WithService(domain),
562 oauth.WithLxm(tangled.RepoCreateNSID),
563 oauth.WithDev(s.config.Core.Dev),
564 )
565 if err != nil {
566 l.Error("service auth failed", "err", err)
567 s.pages.Notice(w, "repo", "Failed to reach PDS.")
568 return
569 }
570
571 xe := tangled.RepoCreate(
572 r.Context(),
573 client,
574 &tangled.RepoCreate_Input{
575 Rkey: rkey,
576 },
577 )
578 if err := xrpcclient.HandleXrpcErr(xe); err != nil {
579 l.Error("xrpc error", "xe", xe)
580 s.pages.Notice(w, "repo", err.Error())
581 return
582 }
583
584 err = db.AddRepo(tx, repo)
585 if err != nil {
586 l.Error("db write failed", "err", err)
587 s.pages.Notice(w, "repo", "Failed to save repository information.")
588 return
589 }
590
591 // acls
592 p, _ := securejoin.SecureJoin(user.Active.Did, repoName)
593 err = s.enforcer.AddRepo(user.Active.Did, domain, p)
594 if err != nil {
595 l.Error("acl setup failed", "err", err)
596 s.pages.Notice(w, "repo", "Failed to set up repository permissions.")
597 return
598 }
599
600 err = tx.Commit()
601 if err != nil {
602 l.Error("txn commit failed", "err", err)
603 http.Error(w, err.Error(), http.StatusInternalServerError)
604 return
605 }
606
607 err = s.enforcer.E.SavePolicy()
608 if err != nil {
609 l.Error("acl save failed", "err", err)
610 http.Error(w, err.Error(), http.StatusInternalServerError)
611 return
612 }
613
614 // reset the ATURI because the transaction completed successfully
615 aturi = ""
616
617 s.notifier.NewRepo(r.Context(), repo)
618 s.pages.HxLocation(w, fmt.Sprintf("/%s/%s", user.Active.Did, repoName))
619 }
620}
621
622// this is used to rollback changes made to the PDS
623//
624// it is a no-op if the provided ATURI is empty
625func rollbackRecord(ctx context.Context, aturi string, client *atpclient.APIClient) error {
626 if aturi == "" {
627 return nil
628 }
629
630 parsed := syntax.ATURI(aturi)
631
632 collection := parsed.Collection().String()
633 repo := parsed.Authority().String()
634 rkey := parsed.RecordKey().String()
635
636 _, err := comatproto.RepoDeleteRecord(ctx, client, &comatproto.RepoDeleteRecord_Input{
637 Collection: collection,
638 Repo: repo,
639 Rkey: rkey,
640 })
641 return err
642}
643
644func BackfillDefaultDefs(e db.Execer, r *idresolver.Resolver, defaults []string) error {
645 defaultLabels, err := db.GetLabelDefinitions(e, orm.FilterIn("at_uri", defaults))
646 if err != nil {
647 return err
648 }
649 // already present
650 if len(defaultLabels) == len(defaults) {
651 return nil
652 }
653
654 labelDefs, err := models.FetchLabelDefs(r, defaults)
655 if err != nil {
656 return err
657 }
658
659 // Insert each label definition to the database
660 for _, labelDef := range labelDefs {
661 _, err = db.AddLabelDefinition(e, &labelDef)
662 if err != nil {
663 return fmt.Errorf("failed to add label definition %s: %v", labelDef.Name, err)
664 }
665 }
666
667 return nil
668}