Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
Go
at sl/rbac2test
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392package 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/spindle/models" "tangled.org/core/tap" "tangled.org/core/tid" "tangled.org/core/workflow")
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: %w", 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")
// only listen to live events if !evt.Record.Live { l.Info("skipping backfill event", "event", evt.Record.AtUri()) return nil }
switch evt.Record.Action { case tap.RecordCreateAction, tap.RecordUpdateAction: record := tangled.RepoPull{} if err := json.Unmarshal(evt.Record.Record, &record); err != nil { l.Error("invalid record", "err", err) return fmt.Errorf("parsing record: %w", err) }
// ignore legacy records if record.Target == nil { l.Info("ignoring pull record: target repo is nil") return nil }
// ignore patch-based and fork-based PRs if record.Source == nil || record.Source.Repo != nil { l.Info("ignoring pull record: not a branch-based pull request") return nil }
// skip if target repo is unknown repo, err := s.db.GetRepo(syntax.ATURI(record.Target.Repo)) if err != nil { l.Warn("target repo is not ingested yet", "repo", record.Target.Repo, "err", err) return fmt.Errorf("target repo is unknown") }
compiler := workflow.Compiler{ Trigger: tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPullRequest), PullRequest: &tangled.Pipeline_PullRequestTriggerData{ Action: "create", SourceBranch: record.Source.Branch, SourceSha: record.Source.Sha, TargetBranch: record.Target.Branch, }, Repo: &tangled.Pipeline_TriggerRepo{ Did: repo.Did.String(), Knot: repo.Knot, Repo: repo.Name, }, }, }
repoUri := s.newRepoCloneUrl(repo.Knot, repo.Did.String(), repo.Name) repoPath := s.newRepoPath(repo.Did, repo.Rkey)
// load workflow definitions from rev (without spindle context) rawPipeline, err := s.loadPipeline(ctx, repoUri, repoPath, record.Source.Sha) if err != nil { // don't retry l.Error("failed loading pipeline", "err", err) return nil } if len(rawPipeline) == 0 { l.Info("no workflow definition find for the repo. skipping the event") return nil } tpl := compiler.Compile(compiler.Parse(rawPipeline)) // TODO: pass compile error to workflow log for _, w := range compiler.Diagnostics.Errors { l.Error(w.String()) } for _, w := range compiler.Diagnostics.Warnings { l.Warn(w.String()) }
pipelineId := models.PipelineId{ Knot: tpl.TriggerMetadata.Repo.Knot, Rkey: tid.TID(), } if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { l.Error("failed to create pipeline event", "err", err) return nil } err = s.processPipeline(ctx, tpl, pipelineId) if err != nil { // don't retry l.Error("failed processing pipeline", "err", err) return nil } case tap.RecordDeleteAction: // no-op } 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}