Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747package pipelines
import ( "bytes" "context" "fmt" "log/slog" "net" "net/http" "sync" "time"
"tangled.org/core/api/tangled" "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/hostutil" "tangled.org/core/idresolver" "tangled.org/core/lexutil" "tangled.org/core/rbac" "tangled.org/core/types"
"github.com/bluesky-social/indigo/atproto/syntax" indigoxrpc "github.com/bluesky-social/indigo/xrpc" "github.com/go-chi/chi/v5" "github.com/gorilla/websocket")
type Pipelines struct { repoResolver *reporesolver.RepoResolver idResolver *idresolver.Resolver config *config.Config oauth *oauth.OAuth pages *pages.Pages db *db.DB enforcer *rbac.Enforcer logger *slog.Logger}
func (p *Pipelines) Router(mw *middleware.Middleware) http.Handler { r := chi.NewRouter() r.Get("/", p.Index) r.Get("/{pipeline}/workflow/{workflow}", p.Workflow) r.Get("/{pipeline}/workflow/{workflow}/logs", p.Logs) r.Group(func(r chi.Router) { r.Use(mw.RepoPermissionMiddleware("repo:push")) r.Post("/{pipeline}/cancel", p.CancelPipeline) r.Post("/{pipeline}/workflow/{workflow}/cancel", p.CancelWorkflow) r.Post("/{pipeline}/retry", p.RetryPipeline) r.Post("/{pipeline}/workflow/{workflow}/retry", p.RetryWorkflow) })
return r}
func New( oauth *oauth.OAuth, repoResolver *reporesolver.RepoResolver, pages *pages.Pages, idResolver *idresolver.Resolver, db *db.DB, config *config.Config, enforcer *rbac.Enforcer, logger *slog.Logger,) *Pipelines { return &Pipelines{ oauth: oauth, repoResolver: repoResolver, pages: pages, idResolver: idResolver, config: config, db: db, enforcer: enforcer, logger: logger, }}
// FetchStatuses fetches pipelines from the repo's spindle, keyed by commit sha.func FetchStatuses( ctx context.Context, repo *models.Repo, shas []string,) (map[string]types.Pipeline, error) { m := make(map[string]types.Pipeline)
if len(shas) == 0 || repo.Spindle == "" { return m, nil }
spindleUrl, err := hostutil.EnsureHttpScheme(repo.Spindle) if err != nil { return m, nil }
xrpcc := &indigoxrpc.Client{Host: spindleUrl} out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", nil, 0, repo.RepoDid) if err != nil { return nil, err }
return types.PipelinesByCommit(out.Pipelines), nil}
// StatusesHandler renders the pipeline-statuses fragment for a set of commit// shas (given as repeated "sha" query params). It backs both the repo's// commit-statuses endpoint and the pull request's pipeline-statuses endpoint.func StatusesHandler( oauth *oauth.OAuth, repoResolver *reporesolver.RepoResolver, pgs *pages.Pages, logger *slog.Logger,) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { l := logger.With("handler", "StatusesHandler")
f, err := repoResolver.Resolve(r) if err != nil { l.Error("failed to resolve repo", "err", err) return }
user := oauth.GetMultiAccountUser(r) shas := r.URL.Query()["sha"]
statuses, err := FetchStatuses(r.Context(), f, shas) if err != nil { l.Error("failed to fetch pipeline statuses", "err", err) return }
err = pgs.PipelineStatusesFragment(w, pages.PipelineStatusesParams{ RepoInfo: repoResolver.GetRepoInfo(r, user), Pipelines: statuses, }) if err != nil { l.Error("failed to render pipeline statuses", "err", err) } }}
func (p *Pipelines) Index(w http.ResponseWriter, r *http.Request) { user := p.oauth.GetMultiAccountUser(r) l := p.logger.With("handler", "Index")
f, err := p.repoResolver.Resolve(r) if err != nil { l.Error("failed to get repo and knot", "err", err) return }
filterKind := r.URL.Query().Get("trigger") var kinds []string switch filterKind { case "push", "pull_request": kinds = []string{filterKind} default: filterKind = "all" }
if f.Spindle == "" { p.pages.Pipelines(w, pages.PipelinesParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), Pipelines: nil, FilterKind: filterKind, Total: 0, }) return }
spindleUrl, err := hostutil.EnsureHttpScheme(f.Spindle) if err != nil { l.Error("invalid spindle host", "host", f.Spindle, "err", err) p.pages.Pipelines(w, pages.PipelinesParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), Pipelines: nil, FilterKind: filterKind, Total: 0, }) return }
xrpcc := indigoxrpc.Client{Host: spindleUrl} out, err := tangled.CiQueryPipelines(r.Context(), &xrpcc, nil, "", kinds, 30, f.RepoDid) if err != nil { l.Error("failed to fetch pipelines", "err", err) p.pages.Pipelines(w, pages.PipelinesParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), Pipelines: nil, FilterKind: filterKind, Total: 0, }) return }
var pipelines []types.Pipeline for _, pipeline := range out.Pipelines { if pipeline.Repo == nil || *pipeline.Repo != f.RepoDid { l.Warn("spindle returned pipeline for unexpected repo", "want", f.RepoDid, "got", pipeline.Repo, "spindle", f.Spindle) continue } pipelines = append(pipelines, types.Pipeline{CiPipeline: pipeline}) }
p.pages.Pipelines(w, pages.PipelinesParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), Pipelines: pipelines, FilterKind: filterKind, Total: out.Total, })}
func (p *Pipelines) Workflow(w http.ResponseWriter, r *http.Request) { user := p.oauth.GetMultiAccountUser(r) l := p.logger.With("handler", "Workflow")
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, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) if err != nil { l.Debug("invalid pipeline id", "id", pipelineId) p.pages.Error404(w) return }
workflowName := chi.URLParam(r, "workflow") if workflowName == "" { l.Debug("empty workflow name") p.pages.Error404(w) return }
l = l.With("pipeline", pipelineId, "workflow", workflowName)
// TODO: change url path to: // /{owner}/{slug}/pipelines/{spindle-did}/{pipeline-id}/workflow/{workflow-id}
if f.Spindle == "" { p.pages.Error404(w) return }
spindleUrl, err := hostutil.EnsureHttpScheme(f.Spindle) if err != nil { l.Error("invalid spindle host", "host", f.Spindle, "err", err) p.pages.Error404(w) return }
xrpcc := &indigoxrpc.Client{Host: spindleUrl} out, err := tangled.CiGetPipeline(r.Context(), xrpcc, pipelineId.String()) if err != nil { // TODO(boltless): change behavior based on error l.Debug("failed to get pipeline", "err", err) p.pages.Error404(w) return }
if out.Repo == nil || *out.Repo != f.RepoDid { l.Debug("spindle returned pipeline for unexpected repo", "want", f.RepoDid, "got", out.Repo) p.pages.Error404(w) return } // 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 }
p.pages.Workflow(w, pages.WorkflowParams{ BaseParams: pages.BaseParamsFromContext(r.Context()), RepoInfo: p.repoResolver.GetRepoInfo(r, user), Pipeline: types.Pipeline{CiPipeline: out}, Workflow: workflowName, SSHLogCommand: p.sshLogCommand(f.RepoDid, out.Commit), })}
func (p *Pipelines) sshLogCommand(repoDid, sha string) string { if p.config == nil || !p.config.SSH.Enabled || sha == "" { return "" } _, port, err := net.SplitHostPort(p.config.SSH.ListenAddr) if err != nil || port == "" { return "" } return fmt.Sprintf("ssh -t -p %s %s %s %s", port, p.config.Core.Hostname(), repoDid, sha)}
var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024,}
type webLogScheduler struct { ch chan *tangled.CiSubscribePipelineLogs_Event}
var _ lexutil.Scheduler[tangled.CiSubscribePipelineLogs_Event] = (*webLogScheduler)(nil)
// AddWork implements [lexutil.Scheduler].func (w *webLogScheduler) AddWork(ctx context.Context, _ string, val *tangled.CiSubscribePipelineLogs_Event) error { select { case w.ch <- val: return nil case <-ctx.Done(): return ctx.Err() }}
// Shutdown implements [lexutil.Scheduler].func (w *webLogScheduler) Shutdown() { close(w.ch) }
func retryPipelineTrigger(orig *tangled.CiPipeline) *tangled.CiTriggerPipeline_Input_Trigger { if orig.Trigger != nil && orig.Trigger.CiTrigger_PullRequest != nil { pr := orig.Trigger.CiTrigger_PullRequest sourceSha := pr.SourceSha if sourceSha == "" { sourceSha = orig.Commit } sourceRepo := pr.SourceRepo if sourceRepo == nil { sourceRepo = orig.SourceRepo }
return &tangled.CiTriggerPipeline_Input_Trigger{ CiTrigger_PullRequest: &tangled.CiTrigger_PullRequest{ Pull: pr.Pull, SourceBranch: pr.SourceBranch, SourceRepo: sourceRepo, SourceSha: sourceSha, TargetBranch: pr.TargetBranch, }, } }
manual := &tangled.CiTrigger_Manual{ Sha: orig.Commit, SourceRepo: orig.SourceRepo, } if orig.Trigger != nil && orig.Trigger.CiTrigger_Manual != nil { origManual := orig.Trigger.CiTrigger_Manual if origManual.Sha != "" { manual.Sha = origManual.Sha } manual.Ref = origManual.Ref manual.Inputs = origManual.Inputs if origManual.SourceRepo != nil { manual.SourceRepo = origManual.SourceRepo } }
return &tangled.CiTriggerPipeline_Input_Trigger{ CiTrigger_Manual: manual, }}
func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { l := p.logger.With("handler", "logs")
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) return }
if f.Spindle == "" { http.Error(w, "invalid repo info", http.StatusBadRequest) return }
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 }
workflowName := chi.URLParam(r, "workflow") if workflowName == "" { l.Debug("empty workflow name") http.Error(w, "invalid workflow name", http.StatusBadRequest) return }
clientConn, err := upgrader.Upgrade(w, r, nil) if err != nil { l.Error("websocket upgrade failed", "err", err) return } defer clientConn.Close()
ctx, cancel := context.WithCancel(r.Context()) defer cancel()
spindleUrl, err := hostutil.EnsureHttpScheme(f.Spindle) if err != nil { l.Error("invalid spindle host", "host", f.Spindle, "err", err) return }
evChan := make(chan *tangled.CiSubscribePipelineLogs_Event, 100) done := make(chan error, 1) sched := &webLogScheduler{ch: evChan} xrpcc := &lexutil.Client{Client: indigoxrpc.Client{Host: spindleUrl}} go func() { done <- tangled.CiSubscribePipelineLogs(ctx, xrpcc, pipelineId.String(), []string{workflowName}, sched) }()
var lastWriteLk sync.Mutex lastWrite := time.Now()
// Start a goroutine to ping the client periodically to check if it's still // alive. If the client doesn't respond to a ping within 5 seconds, we'll // close the connection and teardown the consumer. go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for { select { case <-ticker.C: lastWriteLk.Lock() lw := lastWrite lastWriteLk.Unlock() if time.Since(lw) < 30*time.Second { continue } if err := clientConn.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil { l.Warn("failed to ping client", "err", err) cancel() return } case <-ctx.Done(): return } } }()
clientConn.SetPingHandler(func(message string) error { err := clientConn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(60*time.Second)) if err == websocket.ErrCloseSent { return nil } return err })
// Start a goroutine to read messages from the client and discard them. go func() { for { if _, _, err := clientConn.ReadMessage(); err != nil { cancel() return } } }()
// Main loop: sole writer of data frames to the client. stepStartTimes := make(map[int]time.Time) stepAnsi := make(map[int]*ansiState) var fragment bytes.Buffer for { select { case <-ctx.Done(): l.Info("client disconnected") return
case ev, ok := <-evChan: if !ok { // Stream ended: Shutdown closed the upstream channel. if err := <-done; !isExpectedClose(err) { l.Error("spindle stream error", "err", err) } msg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "finished") _ = clientConn.WriteMessage(websocket.CloseMessage, msg) return }
fragment.Reset()
switch { case ev.Error != nil: l.Error("spindle error frame", "err", ev.Error.Error, "msg", ev.Error.Message) return
case ev.Control != nil: c := ev.Control step := int(c.Step) switch derefStr(c.Status) { case "start": t := parseRFC3339(c.Time) stepStartTimes[step] = t // "system" steps are injected by the CI runner; collapse them. collapsed := derefStr(c.Kind) == "system" err = p.pages.LogBlock(&fragment, pages.LogBlockParams{ Id: step, Name: c.Content, Command: derefStr(c.Command), Collapsed: collapsed, StartTime: t, }) case "end": err = p.pages.LogBlockEnd(&fragment, pages.LogBlockEndParams{ Id: step, StartTime: stepStartTimes[step], EndTime: parseRFC3339(c.Time), }) }
case ev.Data != nil: d := ev.Data step := int(d.Step) ansi, ok := stepAnsi[step] if !ok { ansi = NewAnsiState() stepAnsi[step] = ansi } err = p.pages.LogLine(&fragment, pages.LogLineParams{ Id: step, Content: ansi.Render(d.Content), }) } if err != nil { l.Error("failed to render log line", "err", err) return }
if err = clientConn.WriteMessage(websocket.TextMessage, fragment.Bytes()); err != nil { l.Error("error writing to client", "err", err) return } lastWriteLk.Lock() lastWrite = time.Now() lastWriteLk.Unlock() } }}
func (p *Pipelines) CancelPipeline(w http.ResponseWriter, r *http.Request) { p.cancel(w, r, nil)}
func (p *Pipelines) CancelWorkflow(w http.ResponseWriter, r *http.Request) { workflowName := chi.URLParam(r, "workflow") if workflowName == "" { p.logger.With("handler", "CancelWorkflow").Debug("empty workflow name") p.pages.Error404(w) return } p.cancel(w, r, []string{workflowName})}
func (p *Pipelines) cancel(w http.ResponseWriter, r *http.Request, workflows []string) { l := p.logger.With("handler", "cancel", "workflows", workflows) errorId := "workflow-error" target := "pipeline" if len(workflows) == 1 { target = "workflow" } fail := "Failed to cancel " + target
f, err := p.repoResolver.Resolve(r) if err != nil { l.Error("failed to get repo and knot", "err", err) p.pages.Notice(w, errorId, fail) return } l = l.With("repo", f.RepoDid)
if f.Spindle == "" { l.Debug("spindle is empty") p.pages.Notice(w, errorId, fail) return }
pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) if err != nil { l.Debug("invalid pipeline id", "id", pipelineId) p.pages.Error404(w) return } l = l.With("pipeline", pipelineId)
spindleClient, err := p.oauth.SpindleServiceClient(r, f.Spindle, tangled.CiCancelPipelineNSID) if err != nil { l.Error("failed to prepare spindle client", "err", err) p.pages.Notice(w, errorId, fail) return }
if err := tangled.CiCancelPipeline( r.Context(), spindleClient, &tangled.CiCancelPipeline_Input{ Repo: f.RepoDid, Pipeline: pipelineId.String(), Workflows: workflows, }, ); err != nil { l.Error("failed to cancel pipeline", "err", err) p.pages.Notice(w, errorId, fail) return } l.Debug("canceled pipeline")}
// RetryPipeline retries all workflows in a pipelinefunc (p *Pipelines) RetryPipeline(w http.ResponseWriter, r *http.Request) { p.retry(w, r, "")}
// RetryWorkflow retries a single workflow in a pipelinefunc (p *Pipelines) RetryWorkflow(w http.ResponseWriter, r *http.Request) { p.retry(w, r, chi.URLParam(r, "workflow"))}
// retry triggers a new pipeline run for the original commit, either for all or a single workflowfunc (p *Pipelines) retry(w http.ResponseWriter, r *http.Request, only string) { user := p.oauth.GetMultiAccountUser(r) l := p.logger.With("handler", "retry", "only", only) errorId := "workflow-error"
// fail logs the error and shows a notice to the user fail := func(msg string, err error) { if err != nil { l.Error(msg, "err", err) p.pages.Notice(w, errorId, fmt.Sprintf("%s: %v", msg, err)) } else { l.Error(msg) p.pages.Notice(w, errorId, msg) } }
f, err := p.repoResolver.Resolve(r) if err != nil { fail("failed to resolve repository", err) return } l = l.With("repo", f.RepoDid)
if f.Spindle == "" { fail("this repository has no spindle configured", nil) return }
pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) if err != nil { l.Debug("invalid pipeline id", "id", pipelineId) p.pages.Error404(w) return } l = l.With("pipeline", pipelineId)
spindleUrl, err := hostutil.EnsureHttpScheme(f.Spindle) if err != nil { fail("invalid spindle host", err) return }
// fetch the original pipeline to replay the same commit and workflows queryClient := &indigoxrpc.Client{Host: spindleUrl} orig, err := tangled.CiGetPipeline(r.Context(), queryClient, pipelineId.String()) if err != nil { fail("failed to load the original pipeline", err) return } if orig.Commit == "" { fail("cannot retry: the original pipeline has no commit", nil) return }
// figure out which workflows to run and where to redirect var workflows []string if only != "" { workflows = []string{only} } else { for _, wf := range orig.Workflows { workflows = append(workflows, wf.Name) } } if len(workflows) == 0 { fail("cannot retry: the original pipeline has no workflows", nil) return } redirectWf := workflows[0]
spindleClient, err := p.oauth.SpindleServiceClient(r, f.Spindle, tangled.CiTriggerPipelineNSID) if err != nil { fail("failed to authorize with spindle", err) return }
out, err := tangled.CiTriggerPipeline( r.Context(), spindleClient, &tangled.CiTriggerPipeline_Input{ Repo: f.RepoDid, Trigger: retryPipelineTrigger(orig), Workflows: workflows, }, ) if err != nil { fail("spindle rejected the trigger", err) return }
newAt, err := syntax.ParseATURI(out.Pipeline) if err != nil { fail("pipeline triggered, but the response was malformed", err) return } newId := newAt.RecordKey().String() l = l.With("new", newId) l.Info("pipeline retried")
repoInfo := p.repoResolver.GetRepoInfo(r, user) dest := fmt.Sprintf("/%s/pipelines/%s/workflow/%s", repoInfo.FullName(), newId, redirectWf)
if r.Header.Get("HX-Request") == "true" { w.Header().Set("HX-Redirect", dest) w.WriteHeader(http.StatusOK) return } http.Redirect(w, r, dest, http.StatusSeeOther)}