Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485package spindle
import ( "context" "database/sql" "encoding/json" "errors" "fmt" "io" "log/slog" "math/rand/v2" "net/http" "net/url" "strings" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/tangled" "tangled.org/core/gitutil" "tangled.org/core/hostutil" knotfeed "tangled.org/core/knotfeed" kgit "tangled.org/core/knotserver/git" "tangled.org/core/log" "tangled.org/core/rbac" "tangled.org/core/spindle/db" "tangled.org/core/workflow")
const collaboratorInviteNSID syntax.NSID = tangled.RepoCollaboratorInviteNSID
const ( reconcileInterval = 10 * time.Minute reconcileJitter = time.Minute maxListPages = 256
feedWipeAttempts = 3 feedWipeRetryWait = 100 * time.Millisecond)
const ( changedFileOverheadBytes = 48 changedFilesMaxBytes = 524288 changedFilesMaxCount = 8192)
var knotHTTPClient = &http.Client{Timeout: 30 * time.Second}
type refRecord struct { Rkey syntax.RecordKey Sha knotfeed.ObjectID}
func (s *Spindle) handleKnotFeed(ctx context.Context, knot string, msg knotfeed.Message) error { if msg.Type != knotfeed.TypeCommit || msg.Commit == nil { return nil } var errs []error var invites []syntax.DID seen := make(map[syntax.DID]struct{}) for _, op := range msg.Commit.Records { repoDid := msg.Commit.Repo switch op.Collection { case knotfeed.GitRefCollection: if err := s.handleRefOp(ctx, knot, repoDid, op); err != nil { errs = append(errs, err) } case collaboratorInviteNSID: if _, ok := seen[repoDid]; !ok { seen[repoDid] = struct{}{} invites = append(invites, repoDid) } } } for _, repoDid := range invites { if err := s.reconcileCollaborators(ctx, knot, repoDid); err != nil { errs = append(errs, err) } } return errors.Join(errs...)}
func (s *Spindle) owningRepo(l *slog.Logger, knot string, repoDid syntax.DID, event string) (*db.Repo, bool, error) { repo, err := s.db.GetRepoByDid(repoDid) switch { case errors.Is(err, sql.ErrNoRows): l.Info(fmt.Sprintf("skipping %s for unknown repo", event)) return nil, false, nil case err != nil: return nil, false, fmt.Errorf("lookup repo %s: %w", repoDid, err) case repo.Knot != knot: l.Info(fmt.Sprintf("dropping %s from non-owning knot", event), "repoKnot", repo.Knot) return nil, false, nil } return repo, true, nil}
func (s *Spindle) handleRefOp(ctx context.Context, knot string, repoDid syntax.DID, op knotfeed.RecordOp) error { l := log.FromContext(ctx).With("repo", repoDid, "rkey", op.Rkey)
repo, proceed, err := s.owningRepo(l, knot, repoDid, "ref update") if !proceed || err != nil { return err }
refname, ok := knotfeed.UnescapeRkey(op.Rkey) if !ok { l.Info("skipping ref update with undecodable rkey") return nil } if !isMaterializedRef(refname) { l.Info("skipping ref update outside heads and tags", "ref", refname) return nil }
if op.Deleted() { if err := s.db.DeleteFeedRef(repoDid, op.Rkey); err != nil { return fmt.Errorf("forgetting ref state: %w", err) } l.Info("ref deleted, nothing to trigger", "ref", refname) return nil }
record, err := knotfeed.DecodeRefRecord(op.Bytes) if err != nil { return fmt.Errorf("decoding ref record: %w", err) }
oldSha, err := s.priorSha(repoDid, op) if err != nil { return err } // deliver push webhooks independently of CI; a skip-ci push option // should not suppress webhook notifications s.wh.FirePush(ctx, repo, string(record.Editor), refname, oldSha.String(), record.Sha.String())
if kgit.HasSkipCIPushOption(record.PushOptions) { if err := s.db.PutFeedRef(repoDid, op.Rkey, record.Sha); err != nil { return fmt.Errorf("recording ref state: %w", err) } l.Info("push requested ci skip, skipping the event", "ref", refname) return nil }
repoCloneUri := s.newRepoCloneUrl(knot, repoDid) repoPath := s.newRepoPath(repoDid) if err := gitutil.SparseSync(ctx, repoCloneUri, repoPath, record.Sha.String(), sparseWorkflowDir); err != nil { return fmt.Errorf("sync git repo: %w", err) } l.Info("synced git repo")
changedFiles := changedFilesUnderBudget(l, repoPath, oldSha, record.Sha)
triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { return fmt.Errorf("building trigger repo: %w", err) } if s.scheduler != nil && refname == "refs/heads/"+triggerRepo.DefaultBranch { if err := s.scheduler.RefreshRepo(ctx, *repo); err != nil { l.Warn("failed to refresh schedules after default-branch push", "err", err) } }
trigger := tangled.Pipeline_TriggerMetadata{ Kind: string(workflow.TriggerKindPush), Push: &tangled.Pipeline_PushTriggerData{ Ref: refname, OldSha: oldSha.String(), NewSha: record.Sha.String(), }, Repo: triggerRepo, }
pipelineId, err := s.runPipeline(ctx, repoDid, trigger, changedFiles, repoCloneUri, repoPath, record.Sha.String(), nil, triggerRepo) if err != nil { return err } if err := s.db.PutFeedRef(repoDid, op.Rkey, record.Sha); err != nil { return fmt.Errorf("recording ref state: %w", err) } if pipelineId == "" { l.Info("no workflow matched 'push' trigger, skipping the event") return nil } l.Info("pipeline triggered", "pipeline", pipelineId) return nil}
func isMaterializedRef(refname string) bool { return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/")}
// priorSha is where the ref pointed before this op: reported by the feed, recorded// from the previous op, or zero when we have never seen the ref, which is not an// error.func (s *Spindle) priorSha(repoDid syntax.DID, op knotfeed.RecordOp) (knotfeed.ObjectID, error) { if sha, ok := op.Prior.Sha(); ok { return sha, nil } sha, found, err := s.db.FeedRefSha(repoDid, op.Rkey) if err != nil { return knotfeed.ObjectID{}, fmt.Errorf("reading ref state: %w", err) } if !found { return knotfeed.ObjectID{}, nil } return sha, nil}
func changedFilesUnderBudget(l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) []string { if oldSha.IsZero() { return nil } gr, err := kgit.Open(repoPath, newSha.String()) if err != nil { l.Warn("cannot open synced repo for changed files", "err", err) return nil } paths, err := gr.ChangedFilesBetween(oldSha.String(), newSha.String()) if err != nil { l.Warn("changed files unavailable between revisions", "oldSha", oldSha, "newSha", newSha, "err", err) return nil } return admitChangedFiles(paths)}
func admitChangedFiles(paths []string) []string { spent := 0 admitted := make([]string, 0, min(len(paths), changedFilesMaxCount)) for _, path := range paths { spent += changedFileOverheadBytes + len(path) if spent > changedFilesMaxBytes || len(admitted) == changedFilesMaxCount { return nil } admitted = append(admitted, path) } return admitted}
func (s *Spindle) reconcileCollaborators(ctx context.Context, knot string, repoDid syntax.DID) error { l := log.FromContext(ctx).With("repo", repoDid)
_, proceed, err := s.owningRepo(l, knot, repoDid, "collaborator reconcile") if !proceed || err != nil { return err }
desired, err := s.collaboratorList(ctx, knot, repoDid) if err != nil { return fmt.Errorf("listing collaborators: %w", err) } tracked, err := s.db.ListKnotCollaboratorsByRepoDid(repoDid) if err != nil { return fmt.Errorf("listing tracked collaborators: %w", err) }
desiredSet := make(map[syntax.DID]struct{}, len(desired)) for _, subject := range desired { desiredSet[subject] = struct{}{} } trackedSet := make(map[syntax.DID]struct{}, len(tracked)) for _, c := range tracked { trackedSet[c.Subject] = struct{}{} }
for subject := range desiredSet { if _, ok := trackedSet[subject]; ok { continue } if err := s.e.AddCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil { l.Error("failed to add collaborator policy", "subject", subject, "err", err) continue } if err := s.db.AddKnotCollaborator(repoDid, subject); err != nil { l.Error("failed to track collaborator", "subject", subject, "err", err) continue } l.Info("added knot-managed collaborator", "subject", subject) } for subject := range trackedSet { if _, ok := desiredSet[subject]; ok { continue } if err := s.e.RemoveCollaborator(subject.String(), rbac.ThisServer, repoDid.String()); err != nil { l.Error("failed to remove collaborator policy", "subject", subject, "err", err) continue } if err := s.db.DeleteRepoCollaboratorBySubjectRepo(subject, repoDid); err != nil { l.Error("failed to delete collaborator row", "subject", subject, "err", err) continue } l.Info("removed knot-managed collaborator", "subject", subject) } return nil}
func (s *Spindle) reconcileCollaboratorsLoop(ctx context.Context) { s.reconcileAllCollaborators(ctx) for { delay := reconcileInterval + time.Duration(rand.Int64N(int64(reconcileJitter))) select { case <-ctx.Done(): return case <-time.After(delay): } s.reconcileAllCollaborators(ctx) }}
func (s *Spindle) reconcileAllCollaborators(ctx context.Context) { repos, err := s.db.AllRepos() if err != nil { s.l.Warn("failed to load repos for collaborator reconcile", "err", err) return } for _, repo := range repos { if repo.RepoDid == "" { continue } if err := s.reconcileCollaborators(ctx, repo.Knot, repo.RepoDid); err != nil { s.l.Warn("collaborator reconcile failed", "repo", repo.RepoDid, "err", err) } }}
func (s *Spindle) feedOutdatedReplay(ctx context.Context, knot string, feed knotfeed.Feed) knotfeed.Cursor { live := feed.Live(time.Now()) repos, err := s.db.AllRepos() if err != nil { s.l.Warn("failed to load repos after outdated cursor", "knot", knot, "err", err) return live } reset := 0 for _, repo := range repos { if repo.Knot != knot || repo.RepoDid == "" { continue } if err := s.wipeFeedRefs(ctx, repo.RepoDid); err != nil { s.l.Error("failed to reset ref state, old shas may be stale until the next push", "repo", repo.RepoDid, "err", err) continue } reset++ } if reset > 0 { s.l.Warn("knot cannot replay from our cursor, ref state reset", "knot", knot, "repos", reset) } return live}
func (s *Spindle) wipeFeedRefs(ctx context.Context, repoDid syntax.DID) error { var err error for attempt := range feedWipeAttempts { if attempt > 0 { select { case <-ctx.Done(): return ctx.Err() case <-time.After(feedWipeRetryWait << attempt): } } err = s.db.DeleteFeedRefsByRepoDid(repoDid) if err == nil { return nil } } return err}
func (s *Spindle) collaboratorList(ctx context.Context, knot string, repoDid syntax.DID) ([]syntax.DID, error) { if s.listCollaborators != nil { return s.listCollaborators(ctx, knot, repoDid) }
base, err := knotEndpoint(knot, "/xrpc/sh.tangled.repo.listCollaborators") if err != nil { return nil, err } var subjects []syntax.DID cursor := "" for page := 0; ; page++ { if page >= maxListPages { return nil, fmt.Errorf("collaborators for %s exceed %d pages", repoDid, maxListPages) } q := url.Values{} q.Set("subject", repoDid.String()) q.Set("limit", "1000") if cursor != "" { q.Set("cursor", cursor) } var out struct { Items []struct { Subject string `json:"subject"` } `json:"items"` Cursor string `json:"cursor"` } if err := getJSON(ctx, base+"?"+q.Encode(), &out); err != nil { return nil, err } for _, item := range out.Items { did, err := syntax.ParseDID(item.Subject) if err != nil { return nil, fmt.Errorf("parsing collaborator subject %s: %w", item.Subject, err) } subjects = append(subjects, did) } if out.Cursor == "" { return subjects, nil } cursor = out.Cursor }}
func (s *Spindle) refRecords(ctx context.Context, knot string, repoDid syntax.DID) ([]refRecord, error) { if s.listRefRecords != nil { return s.listRefRecords(ctx, knot, repoDid) }
base, err := knotEndpoint(knot, "/xrpc/sh.tangled.git.listRefs") if err != nil { return nil, err } var refs []refRecord cursor := "" for page := 0; ; page++ { if page >= maxListPages { return nil, fmt.Errorf("refs for %s exceed %d pages", repoDid, maxListPages) } q := url.Values{} q.Set("repo", repoDid.String()) q.Set("limit", "1000") if cursor != "" { q.Set("cursor", cursor) } var out struct { Refs []struct { Ref string `json:"ref"` Sha string `json:"sha"` } `json:"refs"` Cursor string `json:"cursor"` } if err := getJSON(ctx, base+"?"+q.Encode(), &out); err != nil { return nil, err } for _, ref := range out.Refs { rkey, named := knotfeed.EscapeRefname(ref.Ref) sha, isObject := knotfeed.ParseObjectID(ref.Sha) if !named || !isObject { continue } refs = append(refs, refRecord{Rkey: rkey, Sha: sha}) } if out.Cursor == "" { return refs, nil } cursor = out.Cursor }}
func knotEndpoint(knot, path string) (string, error) { host, noTLS, err := hostutil.ParseHostname(knot) if err != nil { return "", fmt.Errorf("parsing knot host %s: %w", knot, err) } scheme := "https" if noTLS { scheme = "http" } return fmt.Sprintf("%s://%s%s", scheme, host, path), nil}
func getJSON(ctx context.Context, u string, out any) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) if err != nil { return err } resp, err := knotHTTPClient.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) return fmt.Errorf("%s %s answered %s: %s", http.MethodGet, u, resp.Status, body) } return json.NewDecoder(resp.Body).Decode(out)}