diff --git a/spindle/embedtap.go b/spindle/embedtap.go --- a/spindle/embedtap.go +++ b/spindle/embedtap.go @@ -67,8 +67,8 @@ RepoFetchTimeout: 5 * time.Minute, IdentityCacheSize: 50_000, EventCacheSize: 10_000, - SignalCollection: tangled.RepoNSID, - CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID}, + SignalCollection: tangled.RepoPullNSID, // HACK: to ingest PRs from any users + CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID}, AdminPassword: cfg.Server.Tap.AdminPassword, RetryTimeout: 60 * time.Second, } diff --git a/spindle/server.go b/spindle/server.go --- a/spindle/server.go +++ b/spindle/server.go @@ -4,6 +4,7 @@ "context" _ "embed" "encoding/json" + "errors" "fmt" "log/slog" "maps" @@ -13,7 +14,9 @@ "time" "github.com/bluesky-social/indigo/atproto/syntax" + indigoxrpc "github.com/bluesky-social/indigo/xrpc" "github.com/go-chi/chi/v5" + "github.com/go-git/go-git/v5/plumbing/object" "github.com/hashicorp/go-version" "tangled.org/core/api/tangled" "tangled.org/core/eventconsumer" @@ -21,6 +24,7 @@ "tangled.org/core/eventstream" "tangled.org/core/idresolver" "tangled.org/core/jetstream" + kgit "tangled.org/core/knotserver/git" "tangled.org/core/log" "tangled.org/core/notifier" "tangled.org/core/rbac" @@ -35,6 +39,8 @@ "tangled.org/core/spindle/queue" "tangled.org/core/spindle/secrets" "tangled.org/core/spindle/xrpc" + "tangled.org/core/tid" + "tangled.org/core/workflow" "tangled.org/core/xrpc/serviceauth" ) @@ -175,12 +181,12 @@ return nil, fmt.Errorf("failed to start jetstream consumer: %w", err) } - // for each incoming sh.tangled.pipeline, we execute - // spindle.processPipeline, which in turn enqueues the pipeline - // job in the above registered queue. + // spindle listen to knot stream for sh.tangled.git.refUpdate + // which will sync the local workflow files in spindle and enqueues the + // pipeline job for on-push workflows ccfg := eventconsumer.NewConsumerConfig() ccfg.Logger = log.SubLogger(logger, "eventconsumer") - ccfg.ProcessFunc = spindle.processPipeline + ccfg.ProcessFunc = spindle.processKnotStream ccfg.CursorStore = cursorStore if cfg.Server.Dev { ccfg.RetryInterval = 5 * time.Second @@ -395,7 +401,7 @@ return x.Router() } -func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { +func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { l := log.FromContext(ctx).With("handler", "processKnotStream") l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) if msg.Nsid == tangled.PipelineNSID { @@ -429,83 +435,9 @@ Rkey: msg.Rkey, } - workflows := make(map[models.Engine][]models.Workflow) - - // Build pipeline environment variables once for all workflows - pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId) - - for _, w := range tpl.Workflows { - if w != nil { - if _, ok := s.engs[w.Engine]; !ok { - s.l.Error("workflow failed: unknown engine", - "pipeline", pipelineId, "workflow", w.Name, "engine", w.Engine) - err = s.db.StatusFailed(models.WorkflowId{ - PipelineId: pipelineId, - Name: w.Name, - }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) - if err != nil { - return fmt.Errorf("db.StatusFailed: %w", err) - } - - continue - } - - eng := s.engs[w.Engine] - - if _, ok := workflows[eng]; !ok { - workflows[eng] = []models.Workflow{} - } - - ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) - if err != nil { - s.l.Error("workflow failed: init workflow", - "pipeline", pipelineId, "workflow", w.Name, "engine", w.Engine, "err", err) - err = s.db.StatusFailed(models.WorkflowId{ - PipelineId: pipelineId, - Name: w.Name, - }, fmt.Sprintf("init workflow: %s", err), -1, s.n) - if err != nil { - return fmt.Errorf("db.StatusFailed: %w", err) - } - - continue - } - - // inject TANGLED_* env vars after InitWorkflow - // This prevents user-defined env vars from overriding them - if ewf.Environment == nil { - ewf.Environment = make(map[string]string) - } - maps.Copy(ewf.Environment, pipelineEnv) - - workflows[eng] = append(workflows[eng], *ewf) - - err = s.db.StatusPending(models.WorkflowId{ - PipelineId: pipelineId, - Name: w.Name, - }, s.n) - if err != nil { - return fmt.Errorf("db.StatusPending: %w", err) - } - } - } - - ok := s.jq.Enqueue(repoDid, queue.Job{ - Run: func() error { - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{ - RepoDid: repoDid, - Workflows: workflows, - }, pipelineId) - return nil - }, - OnFail: func(jobError error) { - s.l.Error("pipeline run failed", "error", jobError) - }, - }) - if ok { - s.l.Info("pipeline enqueued successfully", "id", msg.Rkey) - } else { - s.l.Error("failed to enqueue pipeline: queue is full") + err = s.processPipeline(ctx, repoDid, tpl, pipelineId) + if err != nil { + return err } } else if msg.Nsid == tangled.GitRefUpdateNSID { event := tangled.GitRefUpdate{} @@ -517,21 +449,212 @@ l.Debug("debug") repoDid := syntax.DID(event.Repo) - if _, err := s.db.GetRepoByDid(repoDid); err != nil { + repo, err := s.db.GetRepoByDid(repoDid) + if err != nil { return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) } // NOTE: we are blindly trusting the knot that it will return only repos it own - repoCloneUri := s.newRepoCloneUrl(src.Key(), syntax.DID(event.Repo)) - repoPath := s.newRepoPath(syntax.DID(event.Repo)) + repoCloneUri := s.newRepoCloneUrl(src.Key(), repoDid) + repoPath := s.newRepoPath(repoDid) if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { return fmt.Errorf("sync git repo: %w", err) } l.Info("synced git repo") - // TODO: plan the pipeline + scheme := "https" + if s.cfg.Server.Dev { + scheme = "http" + } + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} + + // HACK: fetch current default branch + // TODO: this should be included in refUpdate event + defaultBranch, _ := func(repo syntax.DID) (string, error) { + defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) + if err != nil { + return "", err + } + return defaultBranchOut.Name, nil + }(repoDid) + + compiler := workflow.Compiler{ + ChangedFiles: event.ChangedFiles, + Trigger: tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindPush), + Push: &tangled.Pipeline_PushTriggerData{ + Ref: event.Ref, + OldSha: event.OldSha, + NewSha: event.NewSha, + }, + Repo: &tangled.Pipeline_TriggerRepo{ + Did: repo.Owner.String(), + Knot: repo.Knot, + Repo: (*string)(&repo.Rkey), + RepoDid: (*string)(&repoDid), + DefaultBranch: defaultBranch, + }, + }, + } + + // load workflow definitions from rev (without spindle context) + rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, event.NewSha) + if err != nil { + return fmt.Errorf("loading pipeline: %w", err) + } + 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()) + } + if len(tpl.Workflows) == 0 { + l.Info("no workflow matching trigger 'push'. skipping the event") + return nil + } + + 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, repoDid, tpl, pipelineId) + if err != nil { + return err + } } + return nil +} + +func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, error) { + if err := git.SparseSyncGitRepo(ctx, repoUri, repoPath, rev); err != nil { + return nil, fmt.Errorf("syncing git repo: %w", err) + } + gr, err := kgit.Open(repoPath, rev) + if err != nil { + return nil, fmt.Errorf("opening git repo: %w", err) + } + + workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir) + if errors.Is(err, object.ErrDirectoryNotFound) { + // return empty RawPipeline when directory doesn't exist + return nil, nil + } else if err != nil { + return nil, fmt.Errorf("loading file tree: %w", err) + } + + var rawPipeline workflow.RawPipeline + for _, e := range workflowDir { + if !e.IsFile() { + continue + } + + fpath := filepath.Join(workflow.WorkflowDir, e.Name) + contents, err := gr.RawContent(fpath) + if err != nil { + return nil, fmt.Errorf("reading raw content of '%s': %w", fpath, err) + } + + rawPipeline = append(rawPipeline, workflow.RawWorkflow{ + Name: e.Name, + Contents: contents, + }) + } + + return rawPipeline, nil +} + +func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId) error { + // Build pipeline environment variables once for all workflows + pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId) + + // filter & init workflows + workflows := make(map[models.Engine][]models.Workflow) + for _, w := range tpl.Workflows { + if w == nil { + continue + } + if _, ok := s.engs[w.Engine]; !ok { + err := s.db.StatusFailed(models.WorkflowId{ + PipelineId: pipelineId, + Name: w.Name, + }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) + if err != nil { + return fmt.Errorf("db.StatusFailed: %w", err) + } + + continue + } + + eng := s.engs[w.Engine] + + if _, ok := workflows[eng]; !ok { + workflows[eng] = []models.Workflow{} + } + + ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) + if err != nil { + err = s.db.StatusFailed(models.WorkflowId{ + PipelineId: pipelineId, + Name: w.Name, + }, fmt.Sprintf("init workflow: %s", err), -1, s.n) + if err != nil { + return fmt.Errorf("db.StatusFailed: %w", err) + } + + continue + } + + // inject TANGLED_* env vars after InitWorkflow + // This prevents user-defined env vars from overriding them + if ewf.Environment == nil { + ewf.Environment = make(map[string]string) + } + maps.Copy(ewf.Environment, pipelineEnv) + + workflows[eng] = append(workflows[eng], *ewf) + } + + // enqueue pipeline + ok := s.jq.Enqueue(repoDid, queue.Job{ + Run: func() error { + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{ + RepoDid: repoDid, + Workflows: workflows, + }, pipelineId) + return nil + }, + OnFail: func(jobError error) { + s.l.Error("pipeline run failed", "error", jobError) + }, + }) + if !ok { + return fmt.Errorf("failed to enqueue pipeline: queue is full") + } + s.l.Info("pipeline enqueued successfully", "id", pipelineId) + + // after successful enqueue, emit StatusPending for all workflows + for _, ewfs := range workflows { + for _, ewf := range ewfs { + err := s.db.StatusPending(models.WorkflowId{ + PipelineId: pipelineId, + Name: ewf.Name, + }, s.n) + if err != nil { + return fmt.Errorf("db.StatusPending: %w", err) + } + } + } return nil } diff --git a/spindle/tapclient.go b/spindle/tapclient.go --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -6,18 +6,26 @@ "encoding/json" "errors" "fmt" + "io" "log/slog" + "net/http" + "net/url" "sync" "time" "github.com/bluesky-social/indigo/atproto/syntax" + indigoxrpc "github.com/bluesky-social/indigo/xrpc" "tangled.org/core/api/tangled" + avmodels "tangled.org/core/appview/models" "tangled.org/core/eventconsumer" "tangled.org/core/log" "tangled.org/core/rbac" "tangled.org/core/spindle/db" "tangled.org/core/spindle/git" + "tangled.org/core/spindle/models" "tangled.org/core/tapc" + "tangled.org/core/tid" + "tangled.org/core/workflow" ) const ( @@ -75,6 +83,8 @@ return t.processRepo(ctx, evt.Record) case tangled.RepoCollaboratorNSID: return t.processCollaborator(ctx, evt.Record) + case tangled.RepoPullNSID: + return t.processPull(ctx, evt.Record) } return nil } @@ -306,6 +316,135 @@ return nil } +func (t *Tap) processPull(ctx context.Context, evt *tapc.RecordEventData) error { + l := t.logger.With("collection", evt.Collection, "did", evt.Did, "rkey", evt.Rkey) + + // only listen to live events + if !evt.Live { + l.Info("skipping backfill event", "event", evt.AtUri()) + return nil + } + + switch evt.Action { + case tapc.RecordCreateAction, tapc.RecordUpdateAction: + record := tangled.RepoPull{} + if err := json.Unmarshal(evt.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 := t.spindle.db.GetRepoByDid(syntax.DID(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") + } + + // only accept branch-based PR (excluding patch-based and fork-based) + if record.Source == nil || record.Source.Repo != nil { + l.Warn("skipping non-branch-based PR") + return nil + } + + latestSubmission, err := t.fetchLatestSubmission(ctx, evt.Did.String(), evt.Rkey.String(), &record) + if err != nil { + return err + } + sourceSha := latestSubmission.SourceRev + + scheme := "https" + if t.spindle.cfg.Server.Dev { + scheme = "http" + } + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} + + // fetch current default branch + defaultBranch, _ := func(repo syntax.DID) (string, error) { + defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) + if err != nil { + return "", err + } + return defaultBranchOut.Name, nil + }(repo.RepoDid) + + compiler := workflow.Compiler{ + Trigger: tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindPullRequest), + PullRequest: &tangled.Pipeline_PullRequestTriggerData{ + Action: "create", + SourceBranch: record.Source.Branch, + SourceSha: sourceSha, + TargetBranch: record.Target.Branch, + }, + Repo: &tangled.Pipeline_TriggerRepo{ + Did: repo.Owner.String(), + Knot: repo.Knot, + Repo: (*string)(&repo.Rkey), + RepoDid: (*string)(&repo.RepoDid), + DefaultBranch: defaultBranch, + }, + }, + } + + repoUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) + repoPath := t.spindle.newRepoPath(repo.RepoDid) + + // load workflow definitions from rev (without spindle context) + rawPipeline, err := t.spindle.loadPipeline(ctx, repoUri, repoPath, sourceSha) + 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()) + } + if len(tpl.Workflows) == 0 { + l.Info("no workflow matching trigger 'pull_request'. skipping the event") + return nil + } + + pipelineId := models.PipelineId{ + Knot: tpl.TriggerMetadata.Repo.Knot, + Rkey: tid.TID(), + } + if err := t.spindle.db.CreatePipelineEvent(pipelineId.Rkey, tpl, t.spindle.n); err != nil { + l.Error("failed to create pipeline event", "err", err) + return nil + } + err = t.spindle.processPipeline(ctx, repo.RepoDid, tpl, pipelineId) + if err != nil { + // don't retry + l.Error("failed processing pipeline", "err", err) + return nil + } + case tapc.RecordDeleteAction: + // no-op + } + return nil +} + func (t *Tap) bufferCollab(repoDid syntax.DID, evt *tapc.RecordEventData) { t.pendingMu.Lock() defer t.pendingMu.Unlock() @@ -372,4 +511,50 @@ if expired > 0 { t.logger.Warn("expired buffered collaborator events without matching repo arrival", "count", expired, "ttl", pendingCollabTTL) } +} + +func (t *Tap) fetchLatestSubmission(ctx context.Context, did, rkey string, record *tangled.RepoPull) (*avmodels.PullSubmission, error) { + // resolve the PR owner's identity to fetch the blob from their PDS + prOwnerIdent, err := t.spindle.res.ResolveIdent(ctx, did) + if err != nil || prOwnerIdent.Handle.IsInvalidHandle() { + return nil, fmt.Errorf("failed to resolve PR owner handle: %w", err) + } + + if len(record.Rounds) == 0 { + return nil, fmt.Errorf("failed to fetch latest submission, no rounds in record") + } + + roundNumber := len(record.Rounds) - 1 + round := record.Rounds[roundNumber] + + // fetch the blob from the PR owner's PDS + prOwnerPds := prOwnerIdent.PDSEndpoint() + blobUrl, err := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", prOwnerPds)) + if err != nil { + return nil, fmt.Errorf("failed to construct blob URL: %w", err) + } + q := blobUrl.Query() + q.Set("cid", round.PatchBlob.Ref.String()) + q.Set("did", did) + blobUrl.RawQuery = q.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobUrl.String(), nil) + if err != nil { + return nil, fmt.Errorf("failed to create blob request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + blobResp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch blob: %w", err) + } + defer blobResp.Body.Close() + + blob := io.ReadCloser(blobResp.Body) + latestSubmission, err := avmodels.PullSubmissionFromRecord(did, rkey, roundNumber, round, &blob) + if err != nil { + return nil, fmt.Errorf("failed to parse submission: %w", err) + } + + return latestSubmission, nil } diff --git a/spindle/db/events.go b/spindle/db/events.go --- a/spindle/db/events.go +++ b/spindle/db/events.go @@ -19,6 +19,19 @@ return eventstream.List(d, cursor, limit) } +func (d *DB) CreatePipelineEvent(rkey string, pipeline tangled.Pipeline, n *notifier.Notifier) error { + eventJson, err := json.Marshal(pipeline) + if err != nil { + return err + } + event := eventstream.Event{ + Rkey: rkey, + Nsid: tangled.PipelineNSID, + EventJson: eventJson, + } + return d.insertEvent(event, n) +} + func (d *DB) createStatusEvent( workflowId models.WorkflowId, statusKind models.StatusKind,