From 5de245f25260e3bbec9948311fcb6928c4275e7b Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Sat, 6 Jun 2026 00:48:30 +0900 Subject: [PATCH] appview, spindle: remove spindlestream Signed-off-by: Seongmin Lee --- appview/db/pipeline.go | 447 ---------------------------- appview/db/pipeline_test.go | 116 -------- appview/pages/pages.go | 14 +- appview/pipelines/logs.go | 17 ++ appview/pipelines/pipelines.go | 224 ++++++-------- appview/pipelines/ssh/session.go | 21 +- appview/pulls/list.go | 35 ++- appview/pulls/single.go | 35 ++- appview/repo/index.go | 2 +- appview/repo/log.go | 9 +- appview/repo/repo.go | 21 -- appview/repo/repo_util.go | 23 +- appview/state/router.go | 4 +- appview/state/spindlestream.go | 182 ----------- appview/state/spindlestream_test.go | 144 --------- appview/state/state.go | 10 +- 16 files changed, 178 insertions(+), 1126 deletions(-) delete mode 100644 appview/db/pipeline.go delete mode 100644 appview/db/pipeline_test.go delete mode 100644 appview/state/spindlestream.go delete mode 100644 appview/state/spindlestream_test.go diff --git a/appview/db/pipeline.go b/appview/db/pipeline.go deleted file mode 100644 index e2084d79..00000000 --- a/appview/db/pipeline.go +++ /dev/null @@ -1,447 +0,0 @@ -package db - -import ( - "context" - "database/sql" - "fmt" - "slices" - "strings" - "time" - - "github.com/bluesky-social/indigo/atproto/syntax" - "tangled.org/core/appview/models" - "tangled.org/core/orm" -) - -func GetPipelines(e Execer, filters ...orm.Filter) ([]models.Pipeline, error) { - var pipelines []models.Pipeline - - var conditions []string - var args []any - for _, filter := range filters { - conditions = append(conditions, filter.Condition()) - args = append(args, filter.Arg()...) - } - - whereClause := "" - if conditions != nil { - whereClause = " where " + strings.Join(conditions, " and ") - } - - query := fmt.Sprintf(`select id, rkey, knot, repo_owner, repo_name, sha, created, repo_did from pipelines %s`, whereClause) - - rows, err := e.Query(query, args...) - - if err != nil { - return nil, err - } - defer rows.Close() - - for rows.Next() { - var pipeline models.Pipeline - var createdAt string - var repoDid sql.NullString - err = rows.Scan( - &pipeline.Id, - &pipeline.Rkey, - &pipeline.Knot, - &pipeline.RepoOwner, - &pipeline.RepoName, - &pipeline.Sha, - &createdAt, - &repoDid, - ) - if err != nil { - return nil, err - } - - if t, err := time.Parse(time.RFC3339, createdAt); err == nil { - pipeline.Created = t - } - if repoDid.Valid { - pipeline.RepoDid = repoDid.String - } - - pipelines = append(pipelines, pipeline) - } - - if err = rows.Err(); err != nil { - return nil, err - } - - return pipelines, nil -} - -func AddPipeline(e Execer, pipeline models.Pipeline) error { - var repoDid *string - if pipeline.RepoDid != "" { - repoDid = &pipeline.RepoDid - } - - args := []any{ - pipeline.Rkey, - pipeline.Knot, - pipeline.RepoOwner, - pipeline.RepoName, - pipeline.TriggerId, - pipeline.Sha, - repoDid, - } - - placeholders := make([]string, len(args)) - for i := range placeholders { - placeholders[i] = "?" - } - - query := fmt.Sprintf(` - insert or ignore into pipelines ( - rkey, - knot, - repo_owner, - repo_name, - trigger_id, - sha, - repo_did - ) values (%s) - `, strings.Join(placeholders, ",")) - - _, err := e.Exec(query, args...) - - return err -} - -func AddTrigger(e Execer, trigger models.Trigger) (int64, error) { - args := []any{ - trigger.Kind, - trigger.PushRef, - trigger.PushNewSha, - trigger.PushOldSha, - trigger.PRSourceBranch, - trigger.PRTargetBranch, - trigger.PRSourceSha, - trigger.PRAction, - } - - placeholders := make([]string, len(args)) - for i := range placeholders { - placeholders[i] = "?" - } - - query := fmt.Sprintf(`insert or ignore into triggers ( - kind, - push_ref, - push_new_sha, - push_old_sha, - pr_source_branch, - pr_target_branch, - pr_source_sha, - pr_action - ) values (%s)`, strings.Join(placeholders, ",")) - - res, err := e.Exec(query, args...) - if err != nil { - return 0, err - } - - return res.LastInsertId() -} - -func AddPipelineStatus(ctx context.Context, e Execer, status models.PipelineStatus) error { - args := []any{ - status.Spindle, - status.Rkey, - status.PipelineKnot, - status.PipelineRkey, - status.Workflow, - status.Status, - status.Error, - status.ExitCode, - status.Created.Format(time.RFC3339), - } - - placeholders := make([]string, len(args)) - for i := range placeholders { - placeholders[i] = "?" - } - - query := fmt.Sprintf(` - insert or ignore into pipeline_statuses ( - spindle, - rkey, - pipeline_knot, - pipeline_rkey, - workflow, - status, - error, - exit_code, - created - ) values (%s) - `, strings.Join(placeholders, ",")) - - _, err := e.ExecContext(ctx, query, args...) - return err -} - -// this is a mega query, but the most useful one: -// get N pipelines, for each one get the latest status of its N workflows -// -// the pipelines table is aliased to `p` -// the triggers table is aliased to `t` -func GetPipelineStatuses(e Execer, limit int, filters ...orm.Filter) ([]models.Pipeline, error) { - var conditions []string - var args []any - for _, filter := range filters { - conditions = append(conditions, filter.Condition()) - args = append(args, filter.Arg()...) - } - - whereClause := "" - if conditions != nil { - whereClause = " where " + strings.Join(conditions, " and ") - } - - query := fmt.Sprintf(` - select - p.id, - p.knot, - p.rkey, - p.repo_owner, - p.repo_name, - p.sha, - p.created, - p.repo_did, - t.id, - t.kind, - t.push_ref, - t.push_new_sha, - t.push_old_sha, - t.pr_source_branch, - t.pr_target_branch, - t.pr_source_sha, - t.pr_action - from - pipelines p - join - triggers t ON p.trigger_id = t.id - %s - order by p.created desc - limit %d - `, whereClause, limit) - - rows, err := e.Query(query, args...) - if err != nil { - return nil, err - } - defer rows.Close() - - pipelines := make(map[syntax.ATURI]models.Pipeline) - for rows.Next() { - var p models.Pipeline - var t models.Trigger - var created string - var repoDid sql.NullString - - err := rows.Scan( - &p.Id, - &p.Knot, - &p.Rkey, - &p.RepoOwner, - &p.RepoName, - &p.Sha, - &created, - &repoDid, - &p.TriggerId, - &t.Kind, - &t.PushRef, - &t.PushNewSha, - &t.PushOldSha, - &t.PRSourceBranch, - &t.PRTargetBranch, - &t.PRSourceSha, - &t.PRAction, - ) - if err != nil { - return nil, err - } - - p.Created, err = time.Parse(time.RFC3339, created) - if err != nil { - return nil, fmt.Errorf("invalid pipeline created timestamp %q: %w", created, err) - } - if repoDid.Valid { - p.RepoDid = repoDid.String - } - - t.Id = p.TriggerId - p.Trigger = &t - p.Statuses = make(map[string]models.WorkflowStatus) - - pipelines[p.AtUri()] = p - } - - // get all statuses - // the where clause here is of the form: - // - // and ( - // (ps.pipeline_knot = k1 and ps.pipeline_rkey = r1) - // or (ps.pipeline_knot = k2 and ps.pipeline_rkey = r2) - // ) - // - // the join on pipelines and repos enforces that the status was emitted - // by the spindle that is actually registered for the pipeline's repo. - conditions = nil - args = nil - for _, p := range pipelines { - knotFilter := orm.FilterEq("ps.pipeline_knot", p.Knot) - rkeyFilter := orm.FilterEq("ps.pipeline_rkey", p.Rkey) - conditions = append(conditions, fmt.Sprintf("(%s and %s)", knotFilter.Condition(), rkeyFilter.Condition())) - args = append(args, p.Knot) - args = append(args, p.Rkey) - } - whereClause = "" - if conditions != nil { - whereClause = "and (" + strings.Join(conditions, " or ") + ")" - } - query = fmt.Sprintf(` - select - ps.id, ps.spindle, ps.rkey, ps.pipeline_knot, ps.pipeline_rkey, - ps.created, ps.workflow, ps.status, ps.error, ps.exit_code - from - pipeline_statuses ps - join - pipelines p on p.knot = ps.pipeline_knot and p.rkey = ps.pipeline_rkey - join - repos r on r.repo_did = p.repo_did - where - ps.spindle = r.spindle - %s - `, whereClause) - - rows, err = e.Query(query, args...) - if err != nil { - return nil, err - } - defer rows.Close() - - for rows.Next() { - var ps models.PipelineStatus - var created string - - err := rows.Scan( - &ps.ID, - &ps.Spindle, - &ps.Rkey, - &ps.PipelineKnot, - &ps.PipelineRkey, - &created, - &ps.Workflow, - &ps.Status, - &ps.Error, - &ps.ExitCode, - ) - if err != nil { - return nil, err - } - - ps.Created, err = time.Parse(time.RFC3339, created) - if err != nil { - return nil, fmt.Errorf("invalid status created timestamp %q: %w", created, err) - } - - pipelineAt := ps.PipelineAt() - - // extract - pipeline, ok := pipelines[pipelineAt] - if !ok { - continue - } - statuses, _ := pipeline.Statuses[ps.Workflow] - if !ok { - pipeline.Statuses[ps.Workflow] = models.WorkflowStatus{} - } - - // append - statuses.Data = append(statuses.Data, ps) - - // reassign - pipeline.Statuses[ps.Workflow] = statuses - pipelines[pipelineAt] = pipeline - } - - var all []models.Pipeline - for _, p := range pipelines { - for _, s := range p.Statuses { - slices.SortFunc(s.Data, func(a, b models.PipelineStatus) int { - if a.Created.After(b.Created) { - return 1 - } - if a.Created.Before(b.Created) { - return -1 - } - if a.ID > b.ID { - return 1 - } - if a.ID < b.ID { - return -1 - } - return 0 - }) - } - all = append(all, p) - } - - // sort pipelines by date - slices.SortFunc(all, func(a, b models.Pipeline) int { - if a.Created.After(b.Created) { - return -1 - } - return 1 - }) - - return all, nil -} - -// the pipelines table is aliased to `p` -// the triggers table is aliased to `t` -func GetPipelineCount(e Execer, filters ...orm.Filter) (int64, error) { - var conditions []string - var args []any - for _, filter := range filters { - conditions = append(conditions, filter.Condition()) - args = append(args, filter.Arg()...) - } - - whereClause := "" - if conditions != nil { - whereClause = " where " + strings.Join(conditions, " and ") - } - - query := fmt.Sprintf(` - select - count(1) - from - pipelines p - join - triggers t ON p.trigger_id = t.id - %s - `, whereClause) - - rows, err := e.Query(query, args...) - if err != nil { - return 0, err - } - defer rows.Close() - - for rows.Next() { - var count int64 - err := rows.Scan(&count) - if err != nil { - return 0, err - } - - return count, nil - } - - // unreachable - return 0, nil -} diff --git a/appview/db/pipeline_test.go b/appview/db/pipeline_test.go deleted file mode 100644 index 0c49747b..00000000 --- a/appview/db/pipeline_test.go +++ /dev/null @@ -1,116 +0,0 @@ -package db - -import ( - "context" - "strings" - "testing" - "time" - - "github.com/bluesky-social/indigo/atproto/syntax" - "tangled.org/core/appview/models" - "tangled.org/core/orm" - spindle "tangled.org/core/spindle/models" - "tangled.org/core/workflow" -) - -// seedPipeline inserts a trigger + pipeline row and returns the pipeline. -func seedPipeline(t *testing.T, d *DB, knot, rkey, repoDid string) models.Pipeline { - t.Helper() - sha := strings.Repeat("a", 40) - ref := "refs/heads/main" - newSha := sha - oldSha := strings.Repeat("0", 40) - trigger := models.Trigger{ - Kind: workflow.TriggerKindPush, - PushRef: &ref, - PushNewSha: &newSha, - PushOldSha: &oldSha, - } - tx, err := d.Begin() - if err != nil { - t.Fatalf("Begin: %v", err) - } - triggerID, err := AddTrigger(tx, trigger) - if err != nil { - tx.Rollback() - t.Fatalf("AddTrigger: %v", err) - } - pipeline := models.Pipeline{ - Knot: knot, - Rkey: rkey, - RepoOwner: syntax.DID("did:plc:owner"), - RepoName: "repo", - RepoDid: repoDid, - TriggerId: int(triggerID), - Sha: sha, - } - if err := AddPipeline(tx, pipeline); err != nil { - tx.Rollback() - t.Fatalf("AddPipeline: %v", err) - } - if err := tx.Commit(); err != nil { - t.Fatalf("Commit: %v", err) - } - return pipeline -} - -// seedStatus inserts a pipeline_status row directly. -func seedStatus(t *testing.T, d *DB, spindleInstance, rkey, pipelineKnot, pipelineRkey, workflow string) { - t.Helper() - status := models.PipelineStatus{ - Spindle: spindleInstance, - Rkey: rkey, - PipelineKnot: pipelineKnot, - PipelineRkey: pipelineRkey, - Workflow: workflow, - Status: spindle.StatusKindSuccess, - Created: time.Now(), - } - if err := AddPipelineStatus(context.Background(), d, status); err != nil { - t.Fatalf("AddPipelineStatus: %v", err) - } -} - -// TestGetPipelineStatuses_SpindleValidation verifies that GetPipelineStatuses -// only returns statuses emitted by the spindle registered for the pipeline's -// repo, and silently drops statuses from a rogue spindle. -func TestGetPipelineStatuses_SpindleValidation(t *testing.T) { - d := newTestDB(t) - - const ( - knot = "knot.example.com" - correctSpindle = "spindle.example.com" - rogueSpindle = "evil.example.com" - repoDid = "did:plc:testrepo" - pipelineRkey = "pipeline1" - ) - - // seed repo with the correct spindle - repo := seedRepo(t, d, "did:plc:owner", knot, "repo", "repo", repoDid) - if err := UpdateSpindle(d, repo.RepoDid, &[]string{correctSpindle}[0]); err != nil { - t.Fatalf("UpdateSpindle: %v", err) - } - - // seed the pipeline for this repo - seedPipeline(t, d, knot, pipelineRkey, repoDid) - - // insert one status from the correct spindle, one from a rogue spindle - seedStatus(t, d, correctSpindle, "status-valid", knot, pipelineRkey, "build") - seedStatus(t, d, rogueSpindle, "status-rogue", knot, pipelineRkey, "build") - - pipelines, err := GetPipelineStatuses(d, 10, orm.FilterEq("p.repo_did", repoDid)) - if err != nil { - t.Fatalf("GetPipelineStatuses: %v", err) - } - if len(pipelines) != 1 { - t.Fatalf("expected 1 pipeline, got %d", len(pipelines)) - } - - statuses := pipelines[0].Statuses["build"].Data - if len(statuses) != 1 { - t.Fatalf("expected 1 status (from correct spindle), got %d", len(statuses)) - } - if statuses[0].Spindle != correctSpindle { - t.Errorf("expected spindle %q, got %q", correctSpindle, statuses[0].Spindle) - } -} diff --git a/appview/pages/pages.go b/appview/pages/pages.go index 5d8aafa7..95b402a7 100644 --- a/appview/pages/pages.go +++ b/appview/pages/pages.go @@ -933,7 +933,7 @@ type RepoIndexParams struct { EmailToDid map[string]string VerifiedCommits commitverify.VerifiedCommits Languages []types.RepoLanguageDetails - Pipelines map[string]models.Pipeline + Pipelines map[string]*tangled.CiDefs_Pipeline NeedsKnotUpgrade bool KnotUnreachable bool types.RepoIndexResponse @@ -980,7 +980,7 @@ type RepoLogParams struct { Active string EmailToDid map[string]string VerifiedCommits commitverify.VerifiedCommits - Pipelines map[string]models.Pipeline + Pipelines map[string]*tangled.CiDefs_Pipeline types.RepoLogResponse } @@ -995,7 +995,7 @@ type RepoCommitParams struct { RepoInfo repoinfo.RepoInfo Active string EmailToDid map[string]string - Pipeline *models.Pipeline + Pipeline *tangled.CiDefs_Pipeline DiffOpts types.DiffOpts // singular because it's always going to be just one @@ -1393,7 +1393,7 @@ type RepoPullsParams struct { FilterQuery string BaseFilterQuery string Stacks []models.Stack - Pipelines map[string]models.Pipeline + Pipelines map[string]tangled.CiDefs_Pipeline LabelDefs map[string]*models.LabelDefinition Page pagination.Page PullCount int @@ -1433,7 +1433,7 @@ type RepoSinglePullParams struct { BranchDeleteStatus *models.BranchDeleteStatus MergeCheck types.MergeCheckResponse ResubmitCheck ResubmitResult - Pipelines map[string]models.Pipeline + Pipelines map[string]tangled.CiDefs_Pipeline Diff types.DiffRenderer DiffOpts types.DiffOpts ActiveRound int @@ -1601,7 +1601,7 @@ func (p *Pages) RepoForks(w io.Writer, params RepoForksParams) error { type PipelinesParams struct { BaseParams RepoInfo repoinfo.RepoInfo - Pipelines []models.Pipeline + Pipelines []*tangled.CiDefs_Pipeline Active string FilterKind string Total int64 @@ -1655,7 +1655,7 @@ func (p *Pages) WorkflowSymbolOOB(w io.Writer, params WorkflowSymbolOOBParams) e type WorkflowParams struct { BaseParams RepoInfo repoinfo.RepoInfo - Pipeline models.Pipeline + Pipeline *tangled.CiDefs_Pipeline Workflow string LogUrl string Active string diff --git a/appview/pipelines/logs.go b/appview/pipelines/logs.go index 172e501e..5ca9a8dd 100644 --- a/appview/pipelines/logs.go +++ b/appview/pipelines/logs.go @@ -1,13 +1,17 @@ package pipelines import ( + "fmt" "html/template" + "net/url" "path" "regexp" "strings" + "github.com/bluesky-social/indigo/atproto/syntax" terminal "github.com/buildkite/terminal-to-html/v3" "github.com/gorilla/websocket" + "tangled.org/core/api/tangled" "tangled.org/core/appview/pages/markup" "tangled.org/core/hostutil" ) @@ -81,6 +85,19 @@ func ReadLogs(conn *websocket.Conn, ch chan LogEvent) { } } +func SubscribeLogsUrl(spindle string, pipeline syntax.TID, workflow string) string { + u, err := hostutil.EnsureWsScheme(spindle) + if err != nil { + return "" + } + + query := url.Values{} + query.Set("pipeline", pipeline.String()) + query.Set("workflow", workflow) + return fmt.Sprintf("%s/xrpc/%s?%s", u, tangled.CiWorkflowSubscribeLogsNSID, query.Encode()) +} + +// TODO(boltless): deprecate this func SpindleURL(spindle, knot, rkey, workflow string) string { url, err := hostutil.EnsureWsScheme(spindle) if err != nil { diff --git a/appview/pipelines/pipelines.go b/appview/pipelines/pipelines.go index a22de03f..2e760e41 100644 --- a/appview/pipelines/pipelines.go +++ b/appview/pipelines/pipelines.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "encoding/json" - "fmt" "log/slog" "net/http" "time" @@ -13,17 +12,17 @@ import ( "tangled.org/core/appview/config" "tangled.org/core/appview/db" "tangled.org/core/appview/middleware" - "tangled.org/core/appview/models" "tangled.org/core/appview/oauth" "tangled.org/core/appview/pages" "tangled.org/core/appview/reporesolver" - "tangled.org/core/eventconsumer" "tangled.org/core/hostutil" "tangled.org/core/idresolver" "tangled.org/core/orm" "tangled.org/core/rbac" spindlemodel "tangled.org/core/spindle/models" + "github.com/bluesky-social/indigo/atproto/syntax" + indigoxrpc "github.com/bluesky-social/indigo/xrpc" "github.com/go-chi/chi/v5" "github.com/gorilla/websocket" ) @@ -34,7 +33,6 @@ type Pipelines struct { config *config.Config oauth *oauth.OAuth pages *pages.Pages - spindlestream *eventconsumer.Consumer pipelineNotifier *StatusNotifier db *db.DB enforcer *rbac.Enforcer @@ -48,7 +46,7 @@ func (p *Pipelines) Router(mw *middleware.Middleware) http.Handler { r.Get("/{pipeline}/workflow/{workflow}/logs", p.Logs) r. With(mw.RepoPermissionMiddleware("repo:owner")). - Post("/{pipeline}/workflow/{workflow}/cancel", p.Cancel) + Post("/{pipeline}/workflow/{workflow}/cancel", p.CancelWorkflow) return r } @@ -57,7 +55,6 @@ func New( oauth *oauth.OAuth, repoResolver *reporesolver.RepoResolver, pages *pages.Pages, - spindlestream *eventconsumer.Consumer, pipelineNotifier *StatusNotifier, idResolver *idresolver.Resolver, db *db.DB, @@ -71,7 +68,6 @@ func New( pages: pages, idResolver: idResolver, config: config, - spindlestream: spindlestream, pipelineNotifier: pipelineNotifier, db: db, enforcer: enforcer, @@ -103,28 +99,20 @@ func (p *Pipelines) Index(w http.ResponseWriter, r *http.Request) { filterKind = "all" } - ps, err := db.GetPipelineStatuses( - p.db, - 30, - filters..., - ) + // sh.tangled.ci.queryPipelines(repo, kind, limit=30) + xrpcc := indigoxrpc.Client{Host: f.Spindle} + out, err := tangled.CiQueryPipelines(r.Context(), &xrpcc, nil, "", 1, f.RepoDid) if err != nil { - l.Error("failed to query db", "err", err) - return - } - - total, err := db.GetPipelineCount(p.db, filters...) - if err != nil { - l.Error("failed to query db", "err", err) - return + l.Error("failed to fetch pipelines", "err", err) + panic("unimplemented") // spindle failure, appview should not fail. } p.pages.Pipelines(w, pages.PipelinesParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), - Pipelines: ps, + Pipelines: out.Pipelines, FilterKind: filterKind, - Total: total, + Total: out.Total, }) } @@ -135,44 +123,58 @@ func (p *Pipelines) Workflow(w http.ResponseWriter, r *http.Request) { f, err := p.repoResolver.Resolve(r) if err != nil { l.Error("failed to get repo and knot", "err", err) + p.pages.Error404(w) return } - pipelineId := chi.URLParam(r, "pipeline") - if pipelineId == "" { - l.Error("empty pipeline ID") + pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) + if err != nil { + l.Debug("invalid pipeline id", "id", pipelineId) + p.pages.Error404(w) return } - workflow := chi.URLParam(r, "workflow") - if workflow == "" { - l.Error("empty workflow name") + workflowName := chi.URLParam(r, "workflow") + if workflowName == "" { + l.Debug("empty workflow name") + p.pages.Error404(w) return } - ps, err := db.GetPipelineStatuses( - p.db, - 1, - orm.FilterEq("p.repo_did", f.RepoDid), - orm.FilterEq("p.id", pipelineId), - ) + l = l.With("pipeline", pipelineId, "workflow", workflowName) + + // TODO: change url path to: + // /{owner}/{slug}/pipelines/{spindle}/{pipeline(tid)}/workflow/{workflow} + // do we want spindle DID or just Hostname? <- DID will make more sense since it's not defined in DID doc. + + xrpcc := &indigoxrpc.Client{Host: f.Spindle} + out, err := tangled.CiGetPipeline(r.Context(), xrpcc, pipelineId.String()) if err != nil { - l.Error("failed to query db", "err", err) + // TODO(boltless): change behavior based on error + l.Debug("failed to get pipeline", "err", err) + p.pages.Error404(w) return } - if len(ps) != 1 { - l.Error("invalid number of pipelines", "len", len(ps)) + // ensure workflow exists + exist := false + for _, workflow := range out.Workflows { + if workflow.Name == workflowName { + exist = true + break + } + } + if !exist { + l.Debug("workflow doesn't exist in pipeline") + p.pages.Error404(w) return } - singlePipeline := ps[0] - p.pages.Workflow(w, pages.WorkflowParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), - Pipeline: singlePipeline, - Workflow: workflow, + Pipeline: out, + Workflow: workflowName, }) } @@ -191,44 +193,31 @@ func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { return } - pipelineId := chi.URLParam(r, "pipeline") - workflow := chi.URLParam(r, "workflow") - if pipelineId == "" || workflow == "" { - http.Error(w, "missing pipeline ID or workflow", http.StatusBadRequest) + if f.Spindle == "" { + http.Error(w, "invalid repo info", http.StatusBadRequest) return } - ps, err := db.GetPipelineStatuses( - p.db, - 1, - orm.FilterEq("p.repo_did", f.RepoDid), - orm.FilterEq("p.id", pipelineId), - ) - if err != nil || len(ps) != 1 { - l.Error("pipeline query failed", "err", err, "count", len(ps)) - http.Error(w, "pipeline not found", http.StatusNotFound) + pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) + if err != nil { + l.Debug("invalid pipeline id", "id", pipelineId) + http.Error(w, "invalid pipeline id", http.StatusBadRequest) return } - singlePipeline := ps[0] - spindle := f.Spindle - knot := f.Knot - rkey := singlePipeline.Rkey - - statusCh := p.pipelineNotifier.Subscribe(singlePipeline.AtUri()) - defer p.pipelineNotifier.Unsubscribe(singlePipeline.AtUri(), statusCh) - - if spindle == "" || knot == "" || rkey == "" { - http.Error(w, "invalid repo info", http.StatusBadRequest) + workflowName := chi.URLParam(r, "workflow") + if workflowName == "" { + l.Debug("empty workflow name") + http.Error(w, "invalid workflow name", http.StatusBadRequest) return } - url := SpindleURL(spindle, knot, rkey, workflow) - if url == "" { + logsUrl := SubscribeLogsUrl(f.Spindle, pipelineId, workflowName) + if logsUrl == "" { http.Error(w, "invalid spindle hostname", http.StatusBadRequest) return } - l = l.With("url", url) + l = l.With("url", logsUrl) clientConn, err := upgrader.Upgrade(w, r, nil) if err != nil { @@ -249,7 +238,7 @@ func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { l.Info("logs endpoint hit") - spindleConn, _, err := websocket.DefaultDialer.Dial(url, nil) + spindleConn, _, err := websocket.DefaultDialer.Dial(logsUrl, nil) if err != nil { l.Error("websocket dial failed", "err", err) return @@ -339,34 +328,6 @@ func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { return } - case _, ok := <-statusCh: - if !ok { - continue - } - fresh, err := db.GetPipelineStatuses( - p.db, - 1, - orm.FilterEq("p.repo_did", f.RepoDid), - orm.FilterEq("p.id", pipelineId), - ) - if err != nil || len(fresh) == 0 { - continue - } - for name, ws := range fresh[0].Statuses { - fragment.Reset() - if err = p.pages.WorkflowSymbolOOB(&fragment, pages.WorkflowSymbolOOBParams{ - Name: name, - Statuses: ws, - }); err != nil { - l.Error("failed to render workflow symbol OOB", "err", err) - continue - } - if err = clientConn.WriteMessage(websocket.TextMessage, fragment.Bytes()); err != nil { - l.Error("error writing workflow symbol to client", "err", err) - return - } - } - case <-time.After(30 * time.Second): l.Debug("sent keepalive") if err = clientConn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil { @@ -377,56 +338,41 @@ func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { } } -func (p *Pipelines) Cancel(w http.ResponseWriter, r *http.Request) { - l := p.logger.With("handler", "Cancel") - - var ( - pipelineId = chi.URLParam(r, "pipeline") - workflow = chi.URLParam(r, "workflow") - ) - if pipelineId == "" || workflow == "" { - http.Error(w, "missing pipeline ID or workflow", http.StatusBadRequest) - return - } +func (p *Pipelines) CancelWorkflow(w http.ResponseWriter, r *http.Request) { + l := p.logger.With("handler", "CancelWorkflow") + errorId := "workflow-error" f, err := p.repoResolver.Resolve(r) if err != nil { l.Error("failed to get repo and knot", "err", err) - http.Error(w, "bad repo/knot", http.StatusBadRequest) + p.pages.Notice(w, errorId, "Failed to cancel workflow") return } + l = l.With("repo", f.RepoDid) - pipeline, err := func() (models.Pipeline, error) { - ps, err := db.GetPipelineStatuses( - p.db, - 1, - orm.FilterEq("p.repo_did", f.RepoDid), - orm.FilterEq("p.id", pipelineId), - ) - if err != nil { - return models.Pipeline{}, err - } - if len(ps) != 1 { - return models.Pipeline{}, fmt.Errorf("wrong pipeline count %d", len(ps)) - } - return ps[0], nil - }() + if f.Spindle == "" { + l.Debug("spindle is empty") + p.pages.Notice(w, errorId, "Failed to cancel workflow") + return + } + + pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) if err != nil { - l.Error("pipeline query failed", "err", err) - http.Error(w, "pipeline not found", http.StatusNotFound) + l.Debug("invalid pipeline id", "id", pipelineId) + p.pages.Error404(w) + return } - var ( - spindle = f.Spindle - knot = f.Knot - rkey = pipeline.Rkey - ) - if spindle == "" || knot == "" || rkey == "" { - http.Error(w, "invalid repo info", http.StatusBadRequest) + workflowName := chi.URLParam(r, "workflow") + if workflowName == "" { + l.Debug("empty workflow name") + p.pages.Error404(w) return } - hostname, noTLS, err := hostutil.ParseHostname(spindle) + l = l.With("pipeline", pipelineId, "workflow", workflowName) + + hostname, noTLS, err := hostutil.ParseHostname(f.Spindle) if err != nil { http.Error(w, "invalid spindle hostname", http.StatusBadRequest) return @@ -440,20 +386,18 @@ func (p *Pipelines) Cancel(w http.ResponseWriter, r *http.Request) { oauth.WithTimeout(time.Second*30), // workflow cleanup usually takes time ) - err = tangled.PipelineCancelPipeline( + if err := tangled.PipelineCancelPipeline( r.Context(), spindleClient, &tangled.PipelineCancelPipeline_Input{ Repo: string(f.RepoAt()), - Pipeline: pipeline.AtUri().String(), - Workflow: workflow, + Pipeline: pipelineId.String(), + Workflow: workflowName, }, - ) - errorId := "workflow-error" - if err != nil { + ); err != nil { l.Error("failed to cancel workflow", "err", err) p.pages.Notice(w, errorId, "Failed to cancel workflow") return } - l.Debug("canceled pipeline", "uri", pipeline.AtUri()) + l.Debug("canceled workflow") } diff --git a/appview/pipelines/ssh/session.go b/appview/pipelines/ssh/session.go index b5328a63..7cf036b0 100644 --- a/appview/pipelines/ssh/session.go +++ b/appview/pipelines/ssh/session.go @@ -1,13 +1,14 @@ package ssh import ( + "context" "fmt" tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/ssh" wishtea "github.com/charmbracelet/wish/bubbletea" - "tangled.org/core/appview/db" - "tangled.org/core/orm" + indigoxrpc "github.com/bluesky-social/indigo/xrpc" + "tangled.org/core/api/tangled" ) func (s *Server) teaHandler(sess ssh.Session) (tea.Model, []tea.ProgramOption) { @@ -29,17 +30,19 @@ func (s *Server) teaHandler(sess ssh.Session) (tea.Model, []tea.ProgramOption) { l = l.With("repoDID", repoDID, "sha", sha) - pipelines, err := db.GetPipelineStatuses(s.db, 1, - orm.FilterEq("p.repo_did", repoDID), - orm.FilterEq("p.sha", sha), - ) - if err != nil || len(pipelines) == 0 { + spindle := "" + + l = l.With("spindle", spindle) + + xrpcc := indigoxrpc.Client{Host: spindle} + out, err := tangled.CiQueryPipelines(context.TODO(), &xrpcc, []string{sha}, "", 1, repoDID) + if err != nil || len(out.Pipelines) == 0 { l.Warn("pipeline not found", "err", err) return newErrorModel(renderer, fmt.Sprintf("pipeline not found for repo %s @ %s", repoDID, sha)), wishtea.MakeOptions(sess) } - pipeline := pipelines[0] - l.Info("serving pipeline", "workflows", len(pipeline.Statuses)) + pipeline := out.Pipelines[0] + l.Info("serving pipeline", "pipeline", pipeline.Id, "workflows", len(pipeline.Workflows)) pty, _, _ := sess.Pty() opts := append(wishtea.MakeOptions(sess), tea.WithAltScreen()) return newPipelineModel(renderer, s, pipeline, pty.Window.Width, pty.Window.Height), opts diff --git a/appview/pulls/list.go b/appview/pulls/list.go index 4e1b1305..9c4ba84a 100644 --- a/appview/pulls/list.go +++ b/appview/pulls/list.go @@ -14,6 +14,7 @@ import ( "tangled.org/core/orm" "github.com/bluesky-social/indigo/atproto/syntax" + indigoxrpc "github.com/bluesky-social/indigo/xrpc" ) func (s *Pulls) RepoPulls(w http.ResponseWriter, r *http.Request) { @@ -261,20 +262,24 @@ func (s *Pulls) RepoPulls(w http.ResponseWriter, r *http.Request) { stacks = append(stacks, stack) } - ps, err := db.GetPipelineStatuses( - s.db, - len(shas), - orm.FilterEq("p.repo_did", f.RepoDid), - orm.FilterIn("p.sha", shas), - ) - if err != nil { - l.Warn("failed to fetch pipeline statuses", "err", err) - // non-fatal - } - m := make(map[string]models.Pipeline) - for _, p := range ps { - m[p.Sha] = p - } + // commitId -> latest pipeline + pipelines := func(ctx context.Context, shas []string) map[string]tangled.CiDefs_Pipeline { + xrpcc := &indigoxrpc.Client{Host: f.Spindle} + out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", 0, f.RepoDid) + if err != nil { + l.Error("failed to fetch pipelines", "err", err) + } + + m := make(map[string]tangled.CiDefs_Pipeline) + + for _, pipeline := range out.Pipelines { + if pipeline == nil { + continue + } + m[pipeline.Commit] = *pipeline + } + return m + }(r.Context(), shas) labelDefs, err := db.GetLabelDefinitions( s.db, @@ -317,7 +322,7 @@ func (s *Pulls) RepoPulls(w http.ResponseWriter, r *http.Request) { FilterState: filterState, FilterQuery: query.String(), Stacks: stacks, - Pipelines: m, + Pipelines: pipelines, Page: page, PullCount: totalPulls, VouchRelationships: vouchRelationships, diff --git a/appview/pulls/single.go b/appview/pulls/single.go index dd1337e0..ec8514c4 100644 --- a/appview/pulls/single.go +++ b/appview/pulls/single.go @@ -1,6 +1,7 @@ package pulls import ( + "context" "fmt" "net/http" "strconv" @@ -152,8 +153,6 @@ func (s *Pulls) repoPullHelper(w http.ResponseWriter, r *http.Request, interdiff resubmitResult = s.resubmitCheck(r, f, pull, stack) } - m := make(map[string]models.Pipeline) - var shas []string for _, s := range pull.Submissions { shas = append(shas, s.SourceRev) @@ -162,20 +161,24 @@ func (s *Pulls) repoPullHelper(w http.ResponseWriter, r *http.Request, interdiff shas = append(shas, p.LatestSha()) } - ps, err := db.GetPipelineStatuses( - s.db, - len(shas), - orm.FilterEq("p.repo_did", f.RepoDid), - orm.FilterIn("p.sha", shas), - ) - if err != nil { - l.Error("failed to fetch pipeline statuses", "err", err) - // non-fatal - } + // commitId -> latest pipeline + pipelines := func(ctx context.Context) map[string]tangled.CiDefs_Pipeline { + xrpcc := &indigoxrpc.Client{Host: f.Spindle} + out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", 0, f.RepoDid) + if err != nil { + l.Error("failed to fetch pipelines", "err", err) + } - for _, p := range ps { - m[p.Sha] = p - } + m := make(map[string]tangled.CiDefs_Pipeline) + + for _, pipeline := range out.Pipelines { + if pipeline == nil { + continue + } + m[pipeline.Commit] = *pipeline + } + return m + }(r.Context()) entities := []syntax.ATURI{pull.AtUri()} for _, s := range pull.Submissions { @@ -259,7 +262,7 @@ func (s *Pulls) repoPullHelper(w http.ResponseWriter, r *http.Request, interdiff BranchDeleteStatus: branchDeleteStatus, MergeCheck: mergeCheckResponse, ResubmitCheck: resubmitResult, - Pipelines: m, + Pipelines: pipelines, Diff: diff, DiffOpts: diffOpts, ActiveRound: roundIdInt, diff --git a/appview/repo/index.go b/appview/repo/index.go index 2348e2d3..22d90d48 100644 --- a/appview/repo/index.go +++ b/appview/repo/index.go @@ -164,7 +164,7 @@ func (rp *Repo) Index(w http.ResponseWriter, r *http.Request) { for _, c := range commitsTrunc { shas = append(shas, c.Hash.String()) } - pipelines, err := getPipelineStatuses(rp.db, f, shas) + pipelines, err := getPipelineStatuses(r.Context(), f, shas) if err != nil { l.Error("failed to fetch pipeline statuses", "err", err) // non-fatal diff --git a/appview/repo/log.go b/appview/repo/log.go index c7e44503..fb1852b2 100644 --- a/appview/repo/log.go +++ b/appview/repo/log.go @@ -11,7 +11,6 @@ import ( "tangled.org/core/api/tangled" "tangled.org/core/appview/commitverify" "tangled.org/core/appview/db" - "tangled.org/core/appview/models" "tangled.org/core/appview/pages" xrpcclient "tangled.org/core/appview/xrpcclient" "tangled.org/core/types" @@ -178,7 +177,7 @@ func (rp *Repo) Log(w http.ResponseWriter, r *http.Request) { for _, c := range xrpcResp.Commits { shas = append(shas, c.Hash.String()) } - pipelines, err := getPipelineStatuses(rp.db, f, shas) + pipelines, err := getPipelineStatuses(r.Context(), f, shas) if err != nil { l.Error("failed to getPipelineStatuses", "err", err) // non-fatal @@ -250,14 +249,14 @@ func (rp *Repo) Commit(w http.ResponseWriter, r *http.Request) { } user := rp.oauth.GetMultiAccountUser(r) - pipelines, err := getPipelineStatuses(rp.db, f, []string{result.Diff.Commit.This}) + pipelines, err := getPipelineStatuses(r.Context(), f, []string{result.Diff.Commit.This}) if err != nil { l.Error("failed to getPipelineStatuses", "err", err) // non-fatal } - var pipeline *models.Pipeline + var pipeline *tangled.CiDefs_Pipeline if p, ok := pipelines[result.Diff.Commit.This]; ok { - pipeline = &p + pipeline = p } rp.pages.RepoCommit(w, pages.RepoCommitParams{ diff --git a/appview/repo/repo.go b/appview/repo/repo.go index 3e7975a2..690a49cb 100644 --- a/appview/repo/repo.go +++ b/appview/repo/repo.go @@ -29,7 +29,6 @@ import ( "tangled.org/core/appview/validator" xrpcclient "tangled.org/core/appview/xrpcclient" "tangled.org/core/consts" - "tangled.org/core/eventconsumer" "tangled.org/core/idresolver" "tangled.org/core/ogre" "tangled.org/core/orm" @@ -51,7 +50,6 @@ type Repo struct { config *config.Config oauth *oauth.OAuth pages *pages.Pages - spindlestream *eventconsumer.Consumer db *db.DB enforcer *rbac.Enforcer acl *knotacl.Service @@ -67,7 +65,6 @@ func New( oauth *oauth.OAuth, repoResolver *reporesolver.RepoResolver, pages *pages.Pages, - spindlestream *eventconsumer.Consumer, idResolver *idresolver.Resolver, db *db.DB, config *config.Config, @@ -84,7 +81,6 @@ func New( pages: pages, idResolver: idResolver, config: config, - spindlestream: spindlestream, db: db, notifier: notifier, enforcer: enforcer, @@ -173,23 +169,6 @@ func (rp *Repo) EditSpindle(w http.ResponseWriter, r *http.Request) { return } - oldSpindle := f.Spindle - if oldSpindle != "" && oldSpindle != newSpindle { - remaining, qErr := db.GetRepos(rp.db, orm.FilterEq("spindle", oldSpindle)) - if qErr != nil { - l.Warn("failed to count repos using old spindle", "err", qErr) - } else if len(remaining) == 0 { - rp.spindlestream.RemoveSource(eventconsumer.NewSpindleSource(oldSpindle)) - } - } - - if !removingSpindle { - rp.spindlestream.AddSource( - context.Background(), - eventconsumer.NewSpindleSource(newSpindle), - ) - } - rp.pages.HxRefresh(w) } diff --git a/appview/repo/repo_util.go b/appview/repo/repo_util.go index c8f49b17..4953f7e5 100644 --- a/appview/repo/repo_util.go +++ b/appview/repo/repo_util.go @@ -1,14 +1,15 @@ package repo import ( + "context" "maps" "slices" "sort" "strings" - "tangled.org/core/appview/db" + indigoxrpc "github.com/bluesky-social/indigo/xrpc" + "tangled.org/core/api/tangled" "tangled.org/core/appview/models" - "tangled.org/core/orm" "tangled.org/core/types" ) @@ -90,28 +91,24 @@ func balanceIndexItems(commitCount, branchCount, tagCount, fileCount int) (commi // // golang is so blessed that it requires 35 lines of imperative code for this func getPipelineStatuses( - d *db.DB, + ctx context.Context, repo *models.Repo, shas []string, -) (map[string]models.Pipeline, error) { - m := make(map[string]models.Pipeline) +) (map[string]*tangled.CiDefs_Pipeline, error) { + m := make(map[string]*tangled.CiDefs_Pipeline) if len(shas) == 0 { return m, nil } - ps, err := db.GetPipelineStatuses( - d, - len(shas), - orm.FilterEq("p.repo_did", repo.RepoDid), - orm.FilterIn("p.sha", shas), - ) + xrpcc := &indigoxrpc.Client{Host: repo.Spindle} + out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", 0, repo.RepoDid) if err != nil { return nil, err } - for _, p := range ps { - m[p.Sha] = p + for _, p := range out.Pipelines { + m[p.Commit] = p } return m, nil diff --git a/appview/state/router.go b/appview/state/router.go index c467acb3..927b92a5 100644 --- a/appview/state/router.go +++ b/appview/state/router.go @@ -130,6 +130,8 @@ func (s *State) UserRouter(mw *middleware.Middleware) http.Handler { r := chi.NewRouter() r.Use(mw.InjectBaseParams) + // TODO: workflow status update requests (30s polling) + r.With(mw.ResolveIdent()).Route("/{user}", func(r chi.Router) { r.Get("/", s.Profile) r.Get("/feed.atom", s.AtomFeedPage) @@ -386,7 +388,6 @@ func (s *State) RepoRouter(mw *middleware.Middleware) http.Handler { s.oauth, s.repoResolver, s.pages, - s.spindlestream, s.idResolver, s.db, s.config, @@ -405,7 +406,6 @@ func (s *State) PipelinesRouter(mw *middleware.Middleware) http.Handler { s.oauth, s.repoResolver, s.pages, - s.spindlestream, s.pipelineNotifier, s.idResolver, s.db, diff --git a/appview/state/spindlestream.go b/appview/state/spindlestream.go deleted file mode 100644 index 2328753c..00000000 --- a/appview/state/spindlestream.go +++ /dev/null @@ -1,182 +0,0 @@ -package state - -import ( - "context" - "encoding/json" - "fmt" - "strings" - "time" - - "github.com/bluesky-social/indigo/atproto/syntax" - "tangled.org/core/api/tangled" - "tangled.org/core/appview/config" - "tangled.org/core/appview/db" - "tangled.org/core/appview/models" - "tangled.org/core/appview/pipelines" - ec "tangled.org/core/eventconsumer" - "tangled.org/core/eventstream" - "tangled.org/core/log" - "tangled.org/core/orm" - "tangled.org/core/rbac" - spindle "tangled.org/core/spindle/models" - "tangled.org/core/workflow" -) - -func Spindlestream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, pn *pipelines.StatusNotifier) (*ec.Consumer, error) { - spindles, err := db.GetSpindles(ctx, d, orm.FilterIsNot("verified", "null")) - if err != nil { - return nil, err - } - - hosts := make([]string, len(spindles)) - for i, s := range spindles { - hosts[i] = s.Instance - } - - return bootstrapStream( - ctx, "spindlestream", ec.KindSpindle, hosts, c.Redis.Addr, - c.Spindlestream, - spindleIngester(d, pn), - ), nil -} - -func spindleIngester(d *db.DB, pn *pipelines.StatusNotifier) ec.ProcessFunc { - return func(ctx context.Context, source ec.Source, msg eventstream.Event) error { - switch msg.Nsid { - case tangled.PipelineNSID: - return ingestPipeline(ctx, d, source, msg) - case tangled.PipelineStatusNSID: - return ingestPipelineStatus(ctx, d, pn, source, msg) - } - return nil - } -} - -func ingestPipeline(ctx context.Context, d *db.DB, source ec.Source, msg eventstream.Event) error { - l := log.FromContext(ctx) - - var record tangled.Pipeline - if err := json.Unmarshal(msg.EventJson, &record); err != nil { - return fmt.Errorf("unmarshal pipeline: %w", err) - } - - if record.TriggerMetadata == nil { - return fmt.Errorf("empty trigger metadata: nsid %s, rkey %s", msg.Nsid, msg.Rkey) - } - - if record.TriggerMetadata.Repo == nil { - return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey) - } - - repoName := "" - if record.TriggerMetadata.Repo.Repo != nil { - repoName = *record.TriggerMetadata.Repo.Repo - } - - repo, lookupErr := resolveRepo(d, record.TriggerMetadata.Repo.RepoDid, record.TriggerMetadata.Repo.Did, repoName) - if lookupErr != nil { - return fmt.Errorf("failed to look up repo: %w", lookupErr) - } - if repo.Spindle == "" { - return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey) - } - - // trigger info - var trigger models.Trigger - var sha string - trigger.Kind = workflow.TriggerKind(record.TriggerMetadata.Kind) - switch trigger.Kind { - case workflow.TriggerKindPush: - trigger.PushRef = &record.TriggerMetadata.Push.Ref - trigger.PushNewSha = &record.TriggerMetadata.Push.NewSha - trigger.PushOldSha = &record.TriggerMetadata.Push.OldSha - sha = *trigger.PushNewSha - case workflow.TriggerKindPullRequest: - trigger.PRSourceBranch = &record.TriggerMetadata.PullRequest.SourceBranch - trigger.PRTargetBranch = &record.TriggerMetadata.PullRequest.TargetBranch - trigger.PRSourceSha = &record.TriggerMetadata.PullRequest.SourceSha - trigger.PRAction = &record.TriggerMetadata.PullRequest.Action - sha = *trigger.PRSourceSha - } - - tx, err := d.Begin() - if err != nil { - return fmt.Errorf("failed to start txn: %w", err) - } - - triggerId, err := db.AddTrigger(tx, trigger) - if err != nil { - return fmt.Errorf("failed to add trigger entry: %w", err) - } - - // TODO: we shouldn't even use knot to identify pipelines - knot := record.TriggerMetadata.Repo.Knot - pipeline := models.Pipeline{ - Rkey: msg.Rkey, - Knot: knot, - RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did), - RepoName: repoName, - RepoDid: repo.RepoDid, - TriggerId: int(triggerId), - Sha: sha, - } - - err = db.AddPipeline(tx, pipeline) - if err != nil { - return fmt.Errorf("failed to add pipeline: %w", err) - } - - err = tx.Commit() - if err != nil { - return fmt.Errorf("failed to commit txn: %w", err) - } - - l.Info("added pipeline", "pipeline", pipeline) - - return nil -} - -func ingestPipelineStatus(ctx context.Context, d *db.DB, pn *pipelines.StatusNotifier, source ec.Source, msg eventstream.Event) error { - var record tangled.PipelineStatus - err := json.Unmarshal(msg.EventJson, &record) - if err != nil { - return err - } - - pipelineUri, err := syntax.ParseATURI(record.Pipeline) - if err != nil { - return err - } - - exitCode := 0 - if record.ExitCode != nil { - exitCode = int(*record.ExitCode) - } - - // pick the record creation time if possible, or use time.Now - created := time.Now() - if t, err := time.Parse(time.RFC3339, record.CreatedAt); err == nil && created.After(t) { - created = t - } - - status := models.PipelineStatus{ - Spindle: source.Host, - Rkey: msg.Rkey, - PipelineKnot: strings.TrimPrefix(pipelineUri.Authority().String(), "did:web:"), - PipelineRkey: pipelineUri.RecordKey().String(), - Created: created, - Workflow: record.Workflow, - Status: spindle.StatusKind(record.Status), - Error: record.Error, - ExitCode: exitCode, - } - - err = db.AddPipelineStatus(ctx, d, status) - if err != nil { - return fmt.Errorf("failed to add pipeline status: %w", err) - } - - pn.Publish(pipelineUri) - - return nil -} diff --git a/appview/state/spindlestream_test.go b/appview/state/spindlestream_test.go deleted file mode 100644 index 8b515a19..00000000 --- a/appview/state/spindlestream_test.go +++ /dev/null @@ -1,144 +0,0 @@ -package state - -import ( - "context" - "io" - "log/slog" - "net/http" - "net/http/httptest" - "path/filepath" - "strings" - "testing" - "time" - - "tangled.org/core/appview/db" - "tangled.org/core/appview/pipelines" - ec "tangled.org/core/eventconsumer" - "tangled.org/core/eventconsumer/cursor" - "tangled.org/core/eventstream" - "tangled.org/core/notifier" - spindledb "tangled.org/core/spindle/db" - spindlemodels "tangled.org/core/spindle/models" -) - -func TestColdStart_SpindleEventsRebuildPipelineStatuses(t *testing.T) { - ctx := t.Context() - - spindleDB, err := spindledb.Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) - if err != nil { - t.Fatalf("spindle Make: %v", err) - } - t.Cleanup(func() { spindleDB.Close() }) - - n := notifier.New() - workflowId := spindlemodels.WorkflowId{ - PipelineId: spindlemodels.PipelineId{Knot: "knot.boltless.example", Rkey: "pipeline-rk1"}, - Name: "build", - } - for _, step := range []func() error{ - func() error { return spindleDB.StatusPending(workflowId, &n) }, - func() error { return spindleDB.StatusRunning(workflowId, &n) }, - func() error { return spindleDB.StatusSuccess(workflowId, &n) }, - } { - if err := step(); err != nil { - t.Fatalf("seed spindle event: %v", err) - } - } - - mux := http.NewServeMux() - mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) { - _ = eventstream.Stream(w, r, eventstream.StreamConfig{ - Backend: spindleDB, - Notifier: &n, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - }) - }) - srv := httptest.NewServer(mux) - t.Cleanup(srv.Close) - source := ec.Source{Kind: "test", Host: strings.TrimPrefix(srv.URL, "http://"), NoTLS: true} - - appviewDB, err := db.Make(ctx, filepath.Join(t.TempDir(), "appview.db")) - if err != nil { - t.Fatalf("appview Make: %v", err) - } - t.Cleanup(func() { appviewDB.Close() }) - - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - processFunc := spindleIngester(appviewDB, pipelines.NewStatusNotifier()) - - cfg := ec.ConsumerConfig{ - ProcessFunc: processFunc, - WorkerCount: 1, - QueueSize: 16, - ConnectionTimeout: 2 * time.Second, - CursorStore: &cursor.MemoryStore{}, - Logger: logger, - } - c := ec.NewConsumer(cfg) - - consumerCtx, cancel := context.WithCancel(ctx) - defer cancel() - c.Start(consumerCtx) - c.AddSource(consumerCtx, source) - - deadline := time.Now().Add(3 * time.Second) - for time.Now().Before(deadline) { - var n int - if err := appviewDB.QueryRow(`select count(*) from pipeline_statuses`).Scan(&n); err != nil { - t.Fatalf("count: %v", err) - } - if n >= 3 { - break - } - time.Sleep(20 * time.Millisecond) - } - - rows, err := appviewDB.Query(` - select spindle, pipeline_knot, pipeline_rkey, workflow, status - from pipeline_statuses - order by created asc - `) - if err != nil { - t.Fatalf("query: %v", err) - } - defer rows.Close() - - type rec struct { - spindle, knot, rkey, workflow, status string - } - var got []rec - for rows.Next() { - var r rec - if err := rows.Scan(&r.spindle, &r.knot, &r.rkey, &r.workflow, &r.status); err != nil { - t.Fatalf("scan: %v", err) - } - got = append(got, r) - } - - if len(got) != 3 { - t.Fatalf("pipeline_statuses rows = %d, want 3: %+v", len(got), got) - } - - wantStatuses := []string{"pending", "running", "success"} - gotStatuses := map[string]bool{} - for _, r := range got { - gotStatuses[r.status] = true - if r.spindle != source.Host { - t.Errorf("spindle = %q, want %q", r.spindle, source.Host) - } - if r.knot != workflowId.Knot { - t.Errorf("pipeline_knot = %q, want %q", r.knot, workflowId.Knot) - } - if r.rkey != workflowId.Rkey { - t.Errorf("pipeline_rkey = %q, want %q", r.rkey, workflowId.Rkey) - } - if r.workflow != workflowId.Name { - t.Errorf("workflow = %q, want %q", r.workflow, workflowId.Name) - } - } - for _, want := range wantStatuses { - if !gotStatuses[want] { - t.Errorf("missing status %q in projection", want) - } - } -} diff --git a/appview/state/state.go b/appview/state/state.go index 529abb33..ea48bc37 100644 --- a/appview/state/state.go +++ b/appview/state/state.go @@ -71,7 +71,6 @@ type State struct { repoResolver *reporesolver.RepoResolver aclService *knotacl.Service knotstream *eventconsumer.Consumer - spindlestream *eventconsumer.Consumer pipelineNotifier *pipelines.StatusNotifier logger *slog.Logger validator *validator.Validator @@ -221,14 +220,10 @@ func Make(ctx context.Context, config *config.Config) (*State, error) { } knotstream.Start(ctx) + // TODO: pipeline notifier serves as pipeline store with cache to prevent spamming spindles + // maybe, not super sure... pipelineNotifier := pipelines.NewStatusNotifier() - spindlestream, err := Spindlestream(ctx, config, d, enforcer, pipelineNotifier) - if err != nil { - return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err) - } - spindlestream.Start(ctx) - state := &State{ db: d, notifier: notifier, @@ -245,7 +240,6 @@ func Make(ctx context.Context, config *config.Config) (*State, error) { repoResolver: repoResolver, aclService: aclService, knotstream: knotstream, - spindlestream: spindlestream, pipelineNotifier: pipelineNotifier, logger: logger, validator: validator, -- 2.54.0