alice.pds.demo.boltless.dev
This repository has no description
1package ssh
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8 "time"
9
10 "github.com/charmbracelet/bubbles/spinner"
11 "github.com/charmbracelet/bubbles/viewport"
12 tea "github.com/charmbracelet/bubbletea"
13 "github.com/charmbracelet/lipgloss"
14 "github.com/gorilla/websocket"
15 "tangled.org/core/api/tangled"
16 extlexutil "tangled.org/core/lexutil"
17)
18
19var (
20 colorFg lipgloss.NoColor = lipgloss.NoColor{}
21 colorBlue lipgloss.ANSIColor = 4
22 colorBrightBlack lipgloss.ANSIColor = 8
23)
24
25type tickMsg time.Time
26
27type statusUpdateMsg struct {
28 pipeline *tangled.CiDefs_Pipeline
29}
30
31type statusUpdateErrMsg struct{ err error }
32
33type pipelineModel struct {
34 renderer *lipgloss.Renderer
35 xrpcc *extlexutil.Client
36 pipeline *tangled.CiDefs_Pipeline
37 selected int
38 logs map[string]*workflowLogs
39
40 // pipeline log stream: cancel tears down the consumer goroutine on quit.
41 // the event/done channels are threaded through log messages, not stored here.
42 cancel context.CancelFunc
43 streamDone bool
44 streamErr error
45
46 spinner spinner.Model
47 width int
48 height int
49}
50
51type workflowLogs struct {
52 steps []step
53 stepIndex map[int64]int // stepId -> index map
54 vp viewport.Model
55 ready bool
56}
57
58func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *tangled.CiDefs_Pipeline, width, height int) *pipelineModel {
59 logs := make(map[string]*workflowLogs, len(pipeline.Workflows))
60 for _, wf := range pipeline.Workflows {
61 logs[wf.Name] = &workflowLogs{stepIndex: make(map[int64]int)}
62 }
63 sp := spinner.New(spinner.WithSpinner(spinner.Line))
64 return &pipelineModel{
65 renderer: renderer,
66 xrpcc: xrpcc,
67 pipeline: pipeline,
68 logs: logs,
69 spinner: sp,
70 width: width,
71 height: height,
72 }
73}
74
75func (m *pipelineModel) Init() tea.Cmd {
76 return tea.Batch(tick(), m.spinner.Tick, m.subscribeCmd())
77}
78
79func tick() tea.Cmd {
80 return tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) })
81}
82
83// subscribeCmd opens the ci.pipeline.subscribeLogs stream for current pipeline.
84// A consumer goroutine pushes decoded events onto the scheduler channel; the
85// returned command yields the first event into the bubbletea loop.
86func (m *pipelineModel) subscribeCmd() tea.Cmd {
87 // cancel existing subscriptions just in case
88 if m.cancel != nil {
89 m.cancel()
90 }
91 sched := newEventScheduler()
92 done := make(chan error, 1)
93 ctx, cancel := context.WithCancel(context.Background())
94 m.cancel = cancel
95
96 pipelineId := m.pipeline.Id
97 go func() {
98 err := tangled.CiPipelineSubscribeLogs(ctx, m.xrpcc, pipelineId, nil, sched)
99 done <- err
100 }()
101
102 return readEventCmd(sched.ch, done)
103}
104
105func readEventCmd(events chan *tangled.CiPipelineSubscribeLogs_Event, done chan error) tea.Cmd {
106 return func() tea.Msg {
107 ev, ok := <-events
108 if !ok {
109 return logDoneMsg{err: <-done}
110 }
111 return logEventMsg{ev: ev, events: events, done: done}
112 }
113}
114
115// fetchStatusCmd re-fetches the pipeline
116func (m *pipelineModel) fetchStatusCmd() tea.Cmd {
117 pipelineId := m.pipeline.Id
118 return func() tea.Msg {
119 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
120 defer cancel()
121 out, err := tangled.CiGetPipeline(ctx, m.xrpcc, pipelineId)
122 if err != nil {
123 return statusUpdateErrMsg{err: fmt.Errorf("refreshing pipeline: %w", err)}
124 }
125 return statusUpdateMsg{pipeline: out}
126 }
127}
128
129func (m *pipelineModel) vpHeight() int {
130 return max(m.height-2, 1) // topbar + empty line take 2 lines
131}
132
133// resizeViewports updates all viewport dimensions and re-renders their content after a terminal resize.
134//
135// TODO: can be tedious if we have logs of logs
136func (m *pipelineModel) resizeViewports() {
137 for _, wl := range m.logs {
138 if !wl.ready {
139 continue
140 }
141 atBottom := wl.vp.AtBottom()
142 wl.vp.Width = m.width
143 wl.vp.Height = m.vpHeight()
144 wl.vp.SetContent(renderLogs(m.renderer, wl, m.width))
145 if atBottom {
146 wl.vp.GotoBottom()
147 }
148 }
149}
150
151func (m *pipelineModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
152 switch msg := msg.(type) {
153 case tea.WindowSizeMsg:
154 m.width, m.height = msg.Width, msg.Height
155 m.resizeViewports()
156
157 case tickMsg:
158 // re-render running workflows so elapsed times advance
159 m.refreshRunning()
160 return m, tick()
161
162 case spinner.TickMsg:
163 var cmd tea.Cmd
164 m.spinner, cmd = m.spinner.Update(msg)
165 return m, cmd
166
167 case tea.KeyMsg:
168 switch msg.String() {
169 case "q", "ctrl+c":
170 if m.cancel != nil {
171 m.cancel()
172 }
173 return m, tea.Quit
174 case "tab", "right", "l":
175 m.selected = (m.selected + 1) % len(m.pipeline.Workflows)
176 return m, nil
177 case "shift+tab", "left", "h":
178 m.selected = (m.selected - 1 + len(m.pipeline.Workflows)) % len(m.pipeline.Workflows)
179 return m, nil
180 }
181 if wl := m.selectedLogs(); wl != nil && wl.ready {
182 switch msg.String() {
183 case "g":
184 wl.vp.GotoTop()
185 return m, nil
186 case "G":
187 wl.vp.GotoBottom()
188 return m, nil
189 case "ctrl+e":
190 wl.vp.ScrollDown(1)
191 return m, nil
192 case "ctrl+y":
193 wl.vp.ScrollUp(1)
194 return m, nil
195 }
196 var cmd tea.Cmd
197 wl.vp, cmd = wl.vp.Update(msg)
198 return m, cmd
199 }
200
201 case logEventMsg:
202 m.applyEvent(msg.ev)
203 return m, readEventCmd(msg.events, msg.done)
204
205 case logDoneMsg:
206 m.streamDone = true
207 if !isExpectedClose(msg.err) {
208 m.streamErr = msg.err
209 }
210 m.refreshAll()
211 // resolve final workflow statuses once now that the stream has ended
212 return m, m.fetchStatusCmd()
213
214 case statusUpdateMsg:
215 m.pipeline = msg.pipeline
216 known := make(map[string]bool, len(m.pipeline.Workflows))
217 for _, wf := range m.pipeline.Workflows {
218 known[wf.Name] = true
219 }
220 for name := range m.logs {
221 if !known[name] {
222 delete(m.logs, name)
223 }
224 }
225 if m.selected >= len(m.pipeline.Workflows) {
226 m.selected = max(len(m.pipeline.Workflows)-1, 0)
227 }
228 m.refreshAll()
229
230 case statusUpdateErrMsg:
231 // best-effort final status refresh; ignore failures
232 }
233
234 return m, nil
235}
236
237func (m *pipelineModel) selectedLogs() *workflowLogs {
238 if len(m.pipeline.Workflows) < 1+m.selected {
239 return nil
240 }
241 return m.logs[m.pipeline.Workflows[m.selected].Name]
242}
243
244func (m *pipelineModel) initViewport(wl *workflowLogs) {
245 if wl.ready {
246 return
247 }
248 wl.vp = viewport.New(m.width, m.vpHeight())
249 wl.ready = true
250}
251
252// ensureWorkflow returns the log state for a workflow, lazily creating its routing entry.
253func (m *pipelineModel) ensureWorkflow(name string) *workflowLogs {
254 wl, ok := m.logs[name]
255 if !ok {
256 wl = &workflowLogs{stepIndex: make(map[int64]int)}
257 m.logs[name] = wl
258 }
259 return wl
260}
261
262// renderWorkflow re-renders a workflow's viewport, preserving bottom-stickiness.
263func (m *pipelineModel) renderWorkflow(wl *workflowLogs) {
264 m.initViewport(wl)
265 atBottom := wl.vp.AtBottom()
266 wl.vp.SetContent(renderLogs(m.renderer, wl, m.width))
267 if atBottom {
268 wl.vp.GotoBottom()
269 }
270}
271
272// refreshAll re-renders every initialized viewport.
273func (m *pipelineModel) refreshAll() {
274 for _, wl := range m.logs {
275 m.renderWorkflow(wl)
276 }
277}
278
279// refreshRunning re-renders workflows with unfinished steps so elapsed times advance.
280func (m *pipelineModel) refreshRunning() {
281 if m.streamDone {
282 return
283 }
284 for _, wl := range m.logs {
285 if !wl.ready {
286 continue
287 }
288 for i := range wl.steps {
289 if !wl.steps[i].finished {
290 m.renderWorkflow(wl)
291 break
292 }
293 }
294 }
295}
296
297// applyEvent routes a decoded subscribeLogs event into the matching workflow.
298func (m *pipelineModel) applyEvent(ev *tangled.CiPipelineSubscribeLogs_Event) {
299 switch {
300 case ev.Error != nil:
301 if ev.Error.Message != "" {
302 m.streamErr = fmt.Errorf("%s: %s", ev.Error.Error, ev.Error.Message)
303 } else {
304 m.streamErr = fmt.Errorf("%s", ev.Error.Error)
305 }
306
307 case ev.Control != nil:
308 c := ev.Control
309 wl := m.ensureWorkflow(c.Workflow)
310 switch derefStr(c.Status) {
311 case "start":
312 wl.stepIndex[c.Step] = len(wl.steps)
313 wl.steps = append(wl.steps, step{
314 id: c.Step, name: c.Content, command: derefStr(c.Command), startTime: parseRFC3339(c.Time),
315 })
316 case "end":
317 if idx, ok := wl.stepIndex[c.Step]; ok {
318 wl.steps[idx].endTime, wl.steps[idx].finished = parseRFC3339(c.Time), true
319 }
320 }
321 m.renderWorkflow(wl)
322
323 case ev.Data != nil:
324 d := ev.Data
325 wl := m.ensureWorkflow(d.Workflow)
326 if idx, ok := wl.stepIndex[d.Step]; ok {
327 wl.steps[idx].lines = append(wl.steps[idx].lines, d.Content)
328 }
329 m.renderWorkflow(wl)
330 }
331}
332
333// isExpectedClose reports whether err is a clean websocket close (or nil).
334func isExpectedClose(err error) bool {
335 if err == nil {
336 return true
337 }
338 var ce *websocket.CloseError
339 if errors.As(err, &ce) {
340 switch ce.Code {
341 case websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure:
342 return true
343 }
344 }
345 return false
346}
347
348// renderLogs builds the full log content string for a workflow, used as viewport content.
349func renderLogs(r *lipgloss.Renderer, wl *workflowLogs, width int) string {
350 headerStyle := r.NewStyle().Foreground(colorFg).Bold(true)
351 cmdStyle := r.NewStyle().Foreground(colorBlue).Width(width)
352 dimStyle := r.NewStyle().Faint(true)
353 var sb strings.Builder
354 for i := range wl.steps {
355 st := &wl.steps[i]
356 dur := ""
357 if st.finished {
358 dur = st.endTime.Sub(st.startTime).Round(time.Millisecond).String()
359 } else if !st.startTime.IsZero() {
360 dur = time.Since(st.startTime).Round(time.Second).String()
361 }
362 // build overlay: "── name ──...── dur ──"
363 nameStr := headerStyle.Render(st.name + " ")
364 durStr := headerStyle.Render(" " + dur + " ")
365 nameW := lipgloss.Width(nameStr)
366 durW := lipgloss.Width(durStr)
367 fillW := max(width-nameW-durW, 0)
368 fill := dimStyle.Render(strings.Repeat("─", fillW))
369 header := nameStr + fill + durStr
370 sb.WriteString(header + "\n")
371 if st.command != "" {
372 sb.WriteString(cmdStyle.Render(st.command) + "\n")
373 }
374 for _, l := range st.lines {
375 sb.WriteString(l + "\n")
376 }
377 sb.WriteString("\n")
378 }
379 return sb.String()
380}
381
382func (m *pipelineModel) View() string {
383 body := ""
384 if wl := m.selectedLogs(); wl != nil && wl.ready {
385 body = wl.vp.View()
386 }
387 if m.streamErr != nil {
388 body = lipgloss.JoinVertical(lipgloss.Left, body, m.renderer.NewStyle().Foreground(colorBlue).Render("stream error: "+m.streamErr.Error()))
389 }
390 return lipgloss.JoinVertical(lipgloss.Left, m.topbarView(), "", body)
391}
392
393// topbarView renders the single-line tab bar with workflow tabs left and trigger info + help right.
394func (m *pipelineModel) topbarView() string {
395 r := m.renderer
396 activeStyle := r.NewStyle().Background(colorBlue).Foreground(colorFg).Bold(true)
397
398 now := time.Now()
399
400 var tabs strings.Builder
401 for i, wf := range m.pipeline.Workflows {
402 status := wf.Status
403 elapsed := workflowElapsed(wf, now).Round(time.Second).String()
404 base := " " + statusIcon(status, m.spinner.View()) + " " + wf.Name
405 if i == m.selected {
406 tab := base
407 if elapsed != "" {
408 tab += " " + elapsed
409 }
410 tab += " "
411 tabs.WriteString(activeStyle.Render(tab))
412 } else {
413 tabs.WriteString(base)
414 if elapsed != "" {
415 dim := r.NewStyle().Faint(true)
416 tabs.WriteString(" " + dim.Render(elapsed))
417 }
418 tabs.WriteString(" ")
419 }
420 }
421
422 tabsStr := tabs.String()
423 infoStr := triggerLine(r, m.pipeline.Trigger, m.pipeline.Commit) + " · " + helpText(r)
424
425 gap := max(m.width-lipgloss.Width(tabsStr)-lipgloss.Width(infoStr), 1)
426
427 return tabsStr + strings.Repeat(" ", gap) + infoStr
428}
429
430func helpText(r *lipgloss.Renderer) string {
431 key := r.NewStyle().Foreground(colorFg)
432 action := r.NewStyle().Faint(true)
433 sep := action.Render(" · ")
434
435 items := []string{
436 key.Render("←/→") + " " + action.Render("switch"),
437 key.Render("↑/↓") + " " + action.Render("scroll"),
438 key.Render("q") + " " + action.Render("quit"),
439 }
440 return strings.Join(items, sep)
441}
442
443func shortSha(sha string) string {
444 if len(sha) >= 8 {
445 return sha[:8]
446 }
447 return sha
448}
449
450func triggerLine(r *lipgloss.Renderer, t *tangled.CiDefs_Pipeline_Trigger, sha string) string {
451 hash := shortSha(sha)
452 dim := r.NewStyle().Faint(true)
453 if t == nil {
454 return dim.Render(hash)
455 }
456 if t.CiTrigger_Push != nil {
457 return t.CiTrigger_Push.Ref + dim.Render("@"+hash) + dim.Render(" (push)")
458 }
459 if t.CiTrigger_PullRequest != nil {
460 source := ""
461 if t.CiTrigger_PullRequest.SourceBranch != nil {
462 source = *t.CiTrigger_PullRequest.SourceBranch
463 }
464 return t.CiTrigger_PullRequest.TargetBranch + dim.Render(" <- "+source+"@"+hash) + dim.Render(" (pull-request)")
465 }
466 return dim.Render(hash)
467}
468
469func statusIcon(status string, spinnerFrame string) string {
470 switch status {
471 case "success":
472 return "✓"
473 case "failed":
474 return "×"
475 case "running":
476 return spinnerFrame
477 case "pending":
478 return "·"
479 case "timeout":
480 return "⌀"
481 case "cancelled":
482 return "-"
483 default:
484 return "?"
485 }
486}
487
488func derefStr(s *string) string {
489 if s == nil {
490 return ""
491 }
492 return *s
493}
494
495func parseRFC3339(s string) time.Time {
496 t, err := time.Parse(time.RFC3339, s)
497 if err != nil {
498 return time.Time{}
499 }
500 return t
501}