From 92166ee183c9474f736dc93d63b37b334e9ac006 Mon Sep 17 00:00:00 2001 From: dawn Date: Mon, 26 Oct 2026 07:17:42 +0300 Subject: [PATCH] {api/org_tangled,cilogs}: follow ci logs from spindles over org.tangled.ci Signed-off-by: dawn --- .../cisubscribePipelineLogs_ext.go | 3 +- cilogs/cihelpers.go | 13 +- cilogs/cilogs_test.go | 233 ++++++++++++++++++ cilogs/logstream.go | 12 +- cilogs/session.go | 6 +- cilogs/tui.go | 50 ++-- 6 files changed, 277 insertions(+), 40 deletions(-) create mode 100644 cilogs/cilogs_test.go diff --git a/api/org_tangled/cisubscribePipelineLogs_ext.go b/api/org_tangled/cisubscribePipelineLogs_ext.go index 5a0bfdca7..59183de8b 100644 --- a/api/org_tangled/cisubscribePipelineLogs_ext.go +++ b/api/org_tangled/cisubscribePipelineLogs_ext.go @@ -81,10 +81,11 @@ func (xevt *CiSubscribePipelineLogs_Event) Deserialize(r io.Reader) error { return nil } -func CiSubscribePipelineLogs(ctx context.Context, c extlexutil.LexClient, pipeline string, workflows []string, sched extlexutil.Scheduler[CiSubscribePipelineLogs_Event]) error { +func CiSubscribePipelineLogs(ctx context.Context, c extlexutil.LexClient, repo string, pipeline string, workflows []string, sched extlexutil.Scheduler[CiSubscribePipelineLogs_Event]) error { defer sched.Shutdown() params := map[string]any{} + params["repo"] = repo params["pipeline"] = pipeline params["workflows"] = workflows diff --git a/cilogs/cihelpers.go b/cilogs/cihelpers.go index b86cd4369..8473b0372 100644 --- a/cilogs/cihelpers.go +++ b/cilogs/cihelpers.go @@ -3,12 +3,12 @@ package cilogs import ( "time" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" ) // helper functions against generated code -func workflowElapsed(wf *tangled.CiPipeline_Workflow, now time.Time) time.Duration { +func workflowElapsed(wf *org_tangled.CiPipeline_Workflow, now time.Time) time.Duration { if wf.StartedAt == nil { return 0 } @@ -26,6 +26,13 @@ func workflowElapsed(wf *tangled.CiPipeline_Workflow, now time.Time) time.Durati return finished.Sub(started) } +func workflowName(wf *org_tangled.CiPipeline_Workflow) string { + if wf.Name != nil { + return *wf.Name + } + return wf.Id +} + var finishedStatuses = map[string]bool{ "failed": true, "timeout": true, @@ -33,7 +40,7 @@ var finishedStatuses = map[string]bool{ "success": true, } -func pipelineFinished(p *tangled.CiPipeline) bool { +func pipelineFinished(p *org_tangled.CiPipeline) bool { for _, wf := range p.Workflows { if !finishedStatuses[wf.Status] { return false diff --git a/cilogs/cilogs_test.go b/cilogs/cilogs_test.go new file mode 100644 index 000000000..f517024f6 --- /dev/null +++ b/cilogs/cilogs_test.go @@ -0,0 +1,233 @@ +package cilogs + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/bluesky-social/indigo/atproto/identity" + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/gorilla/websocket" + gossh "golang.org/x/crypto/ssh" + "tangled.org/core/api/org_tangled" + "tangled.org/core/repoident" +) + +const ( + testRepo = "did:plc:cilogsrepo" + testSha = "4f1c2b3a4f1c2b3a4f1c2b3a4f1c2b3a4f1c2b3a" +) + +func fakeSpindle(t *testing.T, pipeline map[string]any) *httptest.Server { + upgrader := websocket.Upgrader{} + mux := http.NewServeMux() + reply := func(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(v) + } + mux.HandleFunc("/xrpc/org.tangled.ci.queryPipelines", func(w http.ResponseWriter, r *http.Request) { + reply(w, map[string]any{"pipelines": []any{pipeline}}) + }) + mux.HandleFunc("/xrpc/org.tangled.ci.getPipeline", func(w http.ResponseWriter, r *http.Request) { + reply(w, pipeline) + }) + mux.HandleFunc("/xrpc/org.tangled.ci.subscribePipelineLogs", func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("repo") != testRepo { + http.Error(w, `{"error":"InvalidRequest","message":"repo is required"}`, http.StatusBadRequest) + return + } + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + status := func(s string) *string { return &s } + now := time.Now().Format(time.RFC3339) + for _, ev := range []org_tangled.CiSubscribePipelineLogs_Event{ + {Control: &org_tangled.CiSubscribePipelineLogs_Control{Time: now, Workflow: "build", Step: 0, Content: "compile", Status: status("start")}}, + {Data: &org_tangled.CiSubscribePipelineLogs_Data{Time: now, Workflow: "build", Step: 0, Content: "hello from the build\n", Stream: "stdout"}}, + } { + var buf bytes.Buffer + if err := ev.Serialize(&buf); err != nil { + t.Error(err) + return + } + conn.WriteMessage(websocket.BinaryMessage, buf.Bytes()) + } + <-r.Context().Done() + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv +} + +func fakeKnot(t *testing.T, spindleDid string) *httptest.Server { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/xrpc/com.atproto.repo.getRecord" || r.URL.Query().Get("collection") != org_tangled.RepoManifestNSID { + http.Error(w, `{"error":"RecordNotFound"}`, http.StatusNotFound) + return + } + json.NewEncoder(w).Encode(map[string]any{"value": map[string]any{ + "$type": org_tangled.RepoManifestNSID, + "declaration": map[string]any{"owner": "did:plc:owner", "name": "proj"}, + "spindle": spindleDid, + "createdAt": "2026-10-01T09:12:44Z", + }}) + })) + t.Cleanup(srv.Close) + return srv +} + +func directoryFor(knot *httptest.Server) identity.Directory { + dir := identity.NewMockDirectory() + dir.Insert(identity.Identity{DID: testRepo, Services: map[string]identity.ServiceEndpoint{ + repoident.KnotServiceID: {Type: repoident.KnotServiceType, URL: knot.URL}, + }}) + return dir +} + +func freeAddr(t *testing.T) string { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + return ln.Addr().String() +} + +func dialSSH(t *testing.T, addr string) *gossh.Client { + deadline := time.Now().Add(5 * time.Second) + for { + c, err := gossh.Dial("tcp", addr, &gossh.ClientConfig{User: "anyone", HostKeyCallback: gossh.InsecureIgnoreHostKey()}) + if err == nil { + return c + } + if time.Now().After(deadline) { + t.Fatal(err) + } + time.Sleep(50 * time.Millisecond) + } +} + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +func TestFollowsAWorkflowsLogsFromTheSpindleInTheRepoManifest(t *testing.T) { + name := "build" + pipeline := map[string]any{ + "id": syntax.NewTIDNow(0).String(), + "repo": testRepo, + "commit": testSha, + "trigger": map[string]any{ + "$type": "org.tangled.event.push", "ref": "refs/heads/main", "newSha": testSha, + "oldSha": strings.Repeat("0", 40), "repo": testRepo, + }, + "workflows": []any{map[string]any{"id": name, "name": name, "status": "running", "startedAt": time.Now().Format(time.RFC3339)}}, + } + spindle := fakeSpindle(t, pipeline) + spindleDid := "did:web:localhost%3A" + spindle.URL[strings.LastIndex(spindle.URL, ":")+1:] + knot := fakeKnot(t, spindleDid) + + dir := directoryFor(knot) + + addr := freeAddr(t) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + srv := New(Config{ListenAddr: addr, HostKeyPath: t.TempDir() + "/host_key"}, NewSpindles(dir, true), slog.New(slog.NewTextHandler(io.Discard, nil))) + go srv.ListenAndServe(ctx) + + client := dialSSH(t, addr) + defer client.Close() + sess, err := client.NewSession() + if err != nil { + t.Fatal(err) + } + defer sess.Close() + if err := sess.RequestPty("xterm-256color", 40, 160, gossh.TerminalModes{}); err != nil { + t.Fatal(err) + } + var out syncBuffer + sess.Stdout = &out + sess.Stderr = &out + stdin, err := sess.StdinPipe() + if err != nil { + t.Fatal(err) + } + if err := sess.Start(testRepo + " " + testSha); err != nil { + t.Fatal(err) + } + // wish asks the terminal for its background colour before drawing anything + stdin.Write([]byte("\x1b]11;rgb:0000/0000/0000\x07\x1b[?62c")) + + deadline := time.Now().Add(10 * time.Second) + for !strings.Contains(out.String(), "hello from the build") { + if strings.Contains(out.String(), "error:") || time.Now().After(deadline) { + t.Fatalf("never saw the log line, terminal got:\n%s", out.String()) + } + time.Sleep(50 * time.Millisecond) + } + if !strings.Contains(out.String(), "refs/heads/main") { + t.Errorf("the top bar doesn't name the pushed ref:\n%s", out.String()) + } + stdin.Write([]byte("q")) +} + +func TestSpindleURL(t *testing.T) { + for did, want := range map[string]string{ + "did:web:spindle.example.com": "https://spindle.example.com", + "did:web:localhost%3A6555": "http://localhost:6555", + "did:web:host:with:a:path": "", + "did:plc:abc": "", + } { + got, err := spindleURL(did) + if (err != nil) != (want == "") || got != want { + t.Errorf("spindleURL(%q) = %q, %v, want %q", did, got, err, want) + } + } +} + +func TestNoSpindleInTheManifestIsReportedAsSuch(t *testing.T) { + _, err := NewSpindles(directoryFor(fakeKnot(t, "")), true).Of(context.Background(), testRepo) + if !errors.Is(err, ErrNoSpindle) { + t.Errorf("err = %v, want ErrNoSpindle", err) + } +} + +func TestRefusesLoopbackSpindlesOutsideDev(t *testing.T) { + srv := httptest.NewServer(http.NotFoundHandler()) + defer srv.Close() + addr := strings.TrimPrefix(srv.URL, "http://") + for dev, wantErr := range map[bool]bool{false: true, true: false} { + conn, err := NewSpindles(identity.NewMockDirectory(), dev).dialer.DialContext(context.Background(), "tcp", addr) + if conn != nil { + conn.Close() + } + if (err != nil) != wantErr { + t.Errorf("dev=%v: dial err = %v, want error: %v", dev, err, wantErr) + } + } +} diff --git a/cilogs/logstream.go b/cilogs/logstream.go index 49d9be36c..a8d84db04 100644 --- a/cilogs/logstream.go +++ b/cilogs/logstream.go @@ -4,7 +4,7 @@ import ( "context" "time" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" ) type step struct { @@ -22,20 +22,20 @@ type logDoneMsg struct { } type logEventMsg struct { - ev *tangled.CiSubscribePipelineLogs_Event - events chan *tangled.CiSubscribePipelineLogs_Event + ev *org_tangled.CiSubscribePipelineLogs_Event + events chan *org_tangled.CiSubscribePipelineLogs_Event done chan error } type eventScheduler struct { - ch chan *tangled.CiSubscribePipelineLogs_Event + ch chan *org_tangled.CiSubscribePipelineLogs_Event } func newEventScheduler() *eventScheduler { - return &eventScheduler{ch: make(chan *tangled.CiSubscribePipelineLogs_Event, 1024)} + return &eventScheduler{ch: make(chan *org_tangled.CiSubscribePipelineLogs_Event, 1024)} } -func (s *eventScheduler) AddWork(ctx context.Context, _ string, v *tangled.CiSubscribePipelineLogs_Event) error { +func (s *eventScheduler) AddWork(ctx context.Context, _ string, v *org_tangled.CiSubscribePipelineLogs_Event) error { select { case s.ch <- v: return nil diff --git a/cilogs/session.go b/cilogs/session.go index 21732dcde..2f9284b0f 100644 --- a/cilogs/session.go +++ b/cilogs/session.go @@ -8,7 +8,7 @@ import ( tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/ssh" wishtea "github.com/charmbracelet/wish/bubbletea" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" ) func (s *Server) teaHandler(sess ssh.Session) (tea.Model, []tea.ProgramOption) { @@ -45,14 +45,14 @@ func (s *Server) teaHandler(sess ssh.Session) (tea.Model, []tea.ProgramOption) { } l = l.With("spindle", xrpcc.Host) - out, err := tangled.CiQueryPipelines(sess.Context(), xrpcc, []string{sha}, "", nil, 1, repoDID) + out, err := org_tangled.CiQueryPipelines(sess.Context(), xrpcc, []string{sha}, "", 1, repoDID, nil) 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 := out.Pipelines[0] - if _, err := syntax.ParseTID(pipeline.Id); err != nil { + if _, err := syntax.ParseRecordKey(pipeline.Id); err != nil { l.Warn("invalid pipeline id", "id", pipeline.Id, "err", err) return newErrorModel(renderer, fmt.Sprintf("invalid pipeline id %q", pipeline.Id)), wishtea.MakeOptions(sess) } diff --git a/cilogs/tui.go b/cilogs/tui.go index fcabde919..730b69458 100644 --- a/cilogs/tui.go +++ b/cilogs/tui.go @@ -12,7 +12,7 @@ import ( tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/lipgloss" "github.com/gorilla/websocket" - "tangled.org/core/api/tangled" + "tangled.org/core/api/org_tangled" extlexutil "tangled.org/core/lexutil" ) @@ -25,7 +25,7 @@ var ( type tickMsg time.Time type statusUpdateMsg struct { - pipeline *tangled.CiPipeline + pipeline *org_tangled.CiPipeline } type statusUpdateErrMsg struct{ err error } @@ -33,7 +33,7 @@ type statusUpdateErrMsg struct{ err error } type pipelineModel struct { renderer *lipgloss.Renderer xrpcc *extlexutil.Client - pipeline *tangled.CiPipeline + pipeline *org_tangled.CiPipeline selected int logs map[string]*workflowLogs @@ -55,10 +55,10 @@ type workflowLogs struct { ready bool } -func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *tangled.CiPipeline, width, height int) *pipelineModel { +func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *org_tangled.CiPipeline, width, height int) *pipelineModel { logs := make(map[string]*workflowLogs, len(pipeline.Workflows)) for _, wf := range pipeline.Workflows { - logs[wf.Name] = &workflowLogs{stepIndex: make(map[int64]int)} + logs[workflowName(wf)] = &workflowLogs{stepIndex: make(map[int64]int)} } sp := spinner.New(spinner.WithSpinner(spinner.Line)) return &pipelineModel{ @@ -93,16 +93,16 @@ func (m *pipelineModel) subscribeCmd() tea.Cmd { ctx, cancel := context.WithCancel(context.Background()) m.cancel = cancel - pipelineId := m.pipeline.Id + repo, pipelineId := m.pipeline.Repo, m.pipeline.Id go func() { - err := tangled.CiSubscribePipelineLogs(ctx, m.xrpcc, pipelineId, nil, sched) + err := org_tangled.CiSubscribePipelineLogs(ctx, m.xrpcc, repo, pipelineId, nil, sched) done <- err }() return readEventCmd(sched.ch, done) } -func readEventCmd(events chan *tangled.CiSubscribePipelineLogs_Event, done chan error) tea.Cmd { +func readEventCmd(events chan *org_tangled.CiSubscribePipelineLogs_Event, done chan error) tea.Cmd { return func() tea.Msg { ev, ok := <-events if !ok { @@ -114,11 +114,11 @@ func readEventCmd(events chan *tangled.CiSubscribePipelineLogs_Event, done chan // fetchStatusCmd re-fetches the pipeline func (m *pipelineModel) fetchStatusCmd() tea.Cmd { - pipelineId := m.pipeline.Id + repo, pipelineId := m.pipeline.Repo, m.pipeline.Id return func() tea.Msg { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - out, err := tangled.CiGetPipeline(ctx, m.xrpcc, pipelineId) + out, err := org_tangled.CiGetPipeline(ctx, m.xrpcc, pipelineId, repo) if err != nil { return statusUpdateErrMsg{err: fmt.Errorf("refreshing pipeline: %w", err)} } @@ -215,7 +215,7 @@ func (m *pipelineModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { m.pipeline = msg.pipeline known := make(map[string]bool, len(m.pipeline.Workflows)) for _, wf := range m.pipeline.Workflows { - known[wf.Name] = true + known[workflowName(wf)] = true } for name := range m.logs { if !known[name] { @@ -238,7 +238,7 @@ func (m *pipelineModel) selectedLogs() *workflowLogs { if len(m.pipeline.Workflows) < 1+m.selected { return nil } - return m.logs[m.pipeline.Workflows[m.selected].Name] + return m.logs[workflowName(m.pipeline.Workflows[m.selected])] } func (m *pipelineModel) initViewport(wl *workflowLogs) { @@ -295,7 +295,7 @@ func (m *pipelineModel) refreshRunning() { } // applyEvent routes a decoded subscribeLogs event into the matching workflow. -func (m *pipelineModel) applyEvent(ev *tangled.CiSubscribePipelineLogs_Event) { +func (m *pipelineModel) applyEvent(ev *org_tangled.CiSubscribePipelineLogs_Event) { switch { case ev.Error != nil: if ev.Error.Message != "" { @@ -401,7 +401,7 @@ func (m *pipelineModel) topbarView() string { for i, wf := range m.pipeline.Workflows { status := wf.Status elapsed := workflowElapsed(wf, now).Round(time.Second).String() - base := " " + statusIcon(status, m.spinner.View()) + " " + wf.Name + base := " " + statusIcon(status, m.spinner.View()) + " " + workflowName(wf) if i == m.selected { tab := base if elapsed != "" { @@ -447,24 +447,20 @@ func shortSha(sha string) string { return sha } -func triggerLine(r *lipgloss.Renderer, t *tangled.CiPipeline_Trigger, sha string) string { +func triggerLine(r *lipgloss.Renderer, t *org_tangled.CiPipeline_Trigger, sha string) string { hash := shortSha(sha) dim := r.NewStyle().Faint(true) if t == nil { return dim.Render(hash) } - if t.CiTrigger_Push != nil { - return t.CiTrigger_Push.Ref + dim.Render("@"+hash) + dim.Render(" (push)") - } - if t.CiTrigger_PullRequest != nil { - source := "" - if t.CiTrigger_PullRequest.SourceBranch != nil { - source = *t.CiTrigger_PullRequest.SourceBranch - } - return t.CiTrigger_PullRequest.TargetBranch + dim.Render(" <- "+source+"@"+hash) + dim.Render(" (pull-request)") - } - if t.CiTrigger_Schedule != nil { - return t.CiTrigger_Schedule.Ref + dim.Render("@"+hash) + dim.Render(" (schedule)") + switch { + case t.EventPush != nil: + return t.EventPush.Ref + dim.Render("@"+hash) + dim.Render(" (push)") + case t.EventPullRequest != nil: + pr := t.EventPullRequest + return derefStr(pr.TargetBranch) + dim.Render(" <- "+derefStr(pr.SourceBranch)+"@"+hash) + dim.Render(" (pull-request)") + case t.CiTriggerSchedule != nil: + return t.CiTriggerSchedule.Ref + dim.Render("@"+hash) + dim.Render(" (schedule)") } return dim.Render(hash) } -- knotmirror 0.0.1