package spindle import ( "context" "encoding/json" "fmt" "time" "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/tangled" "tangled.org/core/eventconsumer" "tangled.org/core/spindle/db" "tangled.org/core/spindle/git" "tangled.org/core/tap" ) func (s *Spindle) processEvent(ctx context.Context, evt tap.Event) error { l := s.l.With("component", "tapIndexer") var err error switch evt.Type { case tap.EvtRecord: switch evt.Record.Collection.String() { case tangled.SpindleMemberNSID: err = s.processMember(ctx, evt) case tangled.RepoNSID: err = s.processRepo(ctx, evt) case tangled.RepoCollaboratorNSID: err = s.processCollaborator(ctx, evt) case tangled.RepoPullNSID: err = s.processPull(ctx, evt) } case tap.EvtIdentity: // no-op } if err != nil { l.Error("failed to process message. will retry later", "event.ID", evt.ID, "err", err) return err } return nil } // NOTE: make sure to return nil if we don't need to retry (e.g. forbidden, unrelated) func (s *Spindle) processMember(ctx context.Context, evt tap.Event) error { l := s.l.With("component", "tapIndexer", "record", evt.Record.AtUri()) l.Info("processing spindle.member record") // only listen to members if ok, err := s.e.IsSpindleMemberInviteAllowed(evt.Record.Did, s.cfg.Server.Did()); !ok || err != nil { l.Warn("forbidden request: member invite not allowed", "did", evt.Record.Did, "error", err) return nil } switch evt.Record.Action { case tap.RecordCreateAction, tap.RecordUpdateAction: record := tangled.SpindleMember{} if err := json.Unmarshal(evt.Record.Record, &record); err != nil { return fmt.Errorf("parsing record: %w", err) } domain := s.cfg.Server.Hostname if record.Instance != domain { l.Info("domain mismatch", "domain", record.Instance, "expected", domain) return nil } created, err := time.Parse(record.CreatedAt, time.RFC3339) if err != nil { created = time.Now() } if err := db.AddSpindleMember(s.db, db.SpindleMember{ Did: evt.Record.Did, Rkey: evt.Record.Rkey.String(), Instance: record.Instance, Subject: syntax.DID(record.Subject), Created: created, }); err != nil { l.Error("failed to add member", "error", err) return fmt.Errorf("adding member to db: %w", err) } if err := s.e.AddSpindleMember(syntax.DID(record.Subject), s.cfg.Server.Did()); err != nil { return fmt.Errorf("adding member to rbac: %w", err) } if err := s.tap.AddRepos(ctx, []syntax.DID{syntax.DID(record.Subject)}); err != nil { return fmt.Errorf("adding did to tap", err) } l.Info("added member", "member", record.Subject) return nil case tap.RecordDeleteAction: var ( did = evt.Record.Did.String() rkey = evt.Record.Rkey.String() ) member, err := db.GetSpindleMember(s.db, did, rkey) if err != nil { return fmt.Errorf("finding member: %w", err) } if err := db.RemoveSpindleMember(s.db, did, rkey); err != nil { return fmt.Errorf("removing member from db: %w", err) } if err := s.e.RemoveSpindleMember(member.Subject, s.cfg.Server.Did()); err != nil { return fmt.Errorf("removing member from rbac: %w", err) } if err := s.tapSafeRemoveDid(ctx, member.Subject); err != nil { return fmt.Errorf("removing did from tap: %w", err) } l.Info("removed member", "member", member.Subject) return nil } return nil } func (s *Spindle) processCollaborator(ctx context.Context, evt tap.Event) error { l := s.l.With("component", "tapIndexer", "record", evt.Record.AtUri()) l.Info("processing repo.collaborator record") // only listen to members if ok, err := s.e.IsSpindleMember(evt.Record.Did, s.cfg.Server.Did()); !ok || err != nil { l.Warn("forbidden request: not spindle member", "did", evt.Record.Did, "err", err) return nil } switch evt.Record.Action { case tap.RecordCreateAction, tap.RecordUpdateAction: record := tangled.RepoCollaborator{} if err := json.Unmarshal(evt.Record.Record, &record); err != nil { l.Error("invalid record", "err", err) return fmt.Errorf("parsing record: %w", err) } // retry later if target repo is not ingested yet if _, err := s.db.GetRepo(syntax.ATURI(record.Repo)); err != nil { l.Warn("target repo is not ingested yet", "repo", record.Repo, "err", err) return fmt.Errorf("target repo is unknown") } // check perms for this user if ok, err := s.e.IsRepoCollaboratorInviteAllowed(evt.Record.Did, syntax.ATURI(record.Repo)); !ok || err != nil { l.Warn("forbidden request collaborator invite not allowed", "did", evt.Record.Did, "err", err) return nil } if err := s.db.PutRepoCollaborator(&db.RepoCollaborator{ Did: evt.Record.Did, Rkey: evt.Record.Rkey, Repo: syntax.ATURI(record.Repo), Subject: syntax.DID(record.Subject), }); err != nil { return fmt.Errorf("adding collaborator to db: %w", err) } if err := s.e.AddRepoCollaborator(syntax.DID(record.Subject), syntax.ATURI(record.Repo)); err != nil { return fmt.Errorf("adding collaborator to rbac: %w", err) } if err := s.tap.AddRepos(ctx, []syntax.DID{syntax.DID(record.Subject)}); err != nil { return fmt.Errorf("adding did to tap: %w", err) } l.Info("add repo collaborator", "subejct", record.Subject, "repo", record.Repo) return nil case tap.RecordDeleteAction: // get existing collaborator collaborator, err := s.db.GetRepoCollaborator(evt.Record.Did, evt.Record.Rkey) if err != nil { return fmt.Errorf("failed to get existing collaborator info: %w", err) } // check perms for this user if ok, err := s.e.IsRepoCollaboratorInviteAllowed(evt.Record.Did, collaborator.Repo); !ok || err != nil { l.Warn("forbidden request collaborator invite not allowed", "did", evt.Record.Did, "err", err) return nil } if err := s.db.RemoveRepoCollaborator(collaborator.Subject, collaborator.Rkey); err != nil { return fmt.Errorf("removing collaborator from db: %w", err) } if err := s.e.RemoveRepoCollaborator(collaborator.Subject, collaborator.Repo); err != nil { return fmt.Errorf("removing collaborator from rbac: %w", err) } if err := s.tapSafeRemoveDid(ctx, collaborator.Subject); err != nil { return fmt.Errorf("removing did from tap: %w", err) } l.Info("removed repo collaborator", "subejct", collaborator.Subject, "repo", collaborator.Repo) return nil } return nil } func (s *Spindle) processRepo(ctx context.Context, evt tap.Event) error { l := s.l.With("component", "tapIndexer", "record", evt.Record.AtUri()) l.Info("processing repo record") // only listen to members if ok, err := s.e.IsSpindleMember(evt.Record.Did, s.cfg.Server.Did()); !ok || err != nil { l.Warn("forbidden request: not spindle member", "did", evt.Record.Did, "err", err) return nil } switch evt.Record.Action { case tap.RecordCreateAction, tap.RecordUpdateAction: record := tangled.Repo{} if err := json.Unmarshal(evt.Record.Record, &record); err != nil { return fmt.Errorf("parsing record: %w", err) } domain := s.cfg.Server.Hostname if record.Spindle == nil || *record.Spindle != domain { if record.Spindle == nil { l.Info("spindle isn't configured", "name", record.Name) } else { l.Info("different spindle configured", "name", record.Name, "spindle", *record.Spindle, "domain", domain) } if err := s.db.DeleteRepo(evt.Record.Did, evt.Record.Rkey); err != nil { return fmt.Errorf("deleting repo from db: %w", err) } return nil } repo := &db.Repo{ Did: evt.Record.Did, Rkey: evt.Record.Rkey, Name: record.Name, Knot: record.Knot, } if err := s.db.PutRepo(repo); err != nil { return fmt.Errorf("adding repo to db: %w", err) } if err := s.e.AddRepo(evt.Record.AtUri()); err != nil { return fmt.Errorf("adding repo to rbac") } // add this knot to the event consumer src := eventconsumer.NewKnotSource(record.Knot) s.ks.AddSource(context.Background(), src) // setup sparse sync repoCloneUri := s.newRepoCloneUrl(repo.Knot, repo.Did.String(), repo.Name) repoPath := s.newRepoPath(repo.Did, repo.Rkey) if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, ""); err != nil { return fmt.Errorf("setting up sparse-clone git repo: %w", err) } l.Info("added repo", "repo", evt.Record.AtUri()) return nil case tap.RecordDeleteAction: // check perms for this user if ok, err := s.e.IsRepoOwner(evt.Record.Did, evt.Record.AtUri()); !ok || err != nil { l.Warn("forbidden request: not repo owner", "did", evt.Record.Did, "err", err) return nil } if err := s.db.DeleteRepo(evt.Record.Did, evt.Record.Rkey); err != nil { return fmt.Errorf("deleting repo from db: %w", err) } if err := s.e.DeleteRepo(evt.Record.AtUri()); err != nil { return fmt.Errorf("deleting repo from rbac: %w", err) } l.Info("deleted repo", "repo", evt.Record.AtUri()) return nil } return nil } func (s *Spindle) processPull(ctx context.Context, evt tap.Event) error { l := s.l.With("component", "tapIndexer", "record", evt.Record.AtUri()) l.Info("processing pull record") switch evt.Record.Action { case tap.RecordCreateAction, tap.RecordUpdateAction: // TODO case tap.RecordDeleteAction: // TODO } return nil } func (s *Spindle) tapSafeRemoveDid(ctx context.Context, did syntax.DID) error { known, err := s.db.IsKnownDid(syntax.DID(did)) if err != nil { return fmt.Errorf("ensuring did known state: %w", err) } if !known { if err := s.tap.RemoveRepos(ctx, []syntax.DID{did}); err != nil { return fmt.Errorf("removing did from tap: %w", err) } } return nil }