Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505package ssh
import ( "context" "errors" "fmt" "strings" "time"
"github.com/charmbracelet/bubbles/spinner" "github.com/charmbracelet/bubbles/viewport" tea "github.com/charmbracelet/bubbletea" "github.com/charmbracelet/lipgloss" "github.com/gorilla/websocket" "tangled.org/core/api/tangled" extlexutil "tangled.org/core/lexutil")
var ( colorFg lipgloss.NoColor = lipgloss.NoColor{} colorBlue lipgloss.ANSIColor = 4 colorBrightBlack lipgloss.ANSIColor = 8)
type tickMsg time.Time
type statusUpdateMsg struct { pipeline *tangled.CiPipeline}
type statusUpdateErrMsg struct{ err error }
type pipelineModel struct { renderer *lipgloss.Renderer xrpcc *extlexutil.Client pipeline *tangled.CiPipeline selected int logs map[string]*workflowLogs
// pipeline log stream: cancel tears down the consumer goroutine on quit. // the event/done channels are threaded through log messages, not stored here. cancel context.CancelFunc streamDone bool streamErr error
spinner spinner.Model width int height int}
type workflowLogs struct { steps []step stepIndex map[int64]int // stepId -> index map vp viewport.Model ready bool}
func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *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)} } sp := spinner.New(spinner.WithSpinner(spinner.Line)) return &pipelineModel{ renderer: renderer, xrpcc: xrpcc, pipeline: pipeline, logs: logs, spinner: sp, width: width, height: height, }}
func (m *pipelineModel) Init() tea.Cmd { return tea.Batch(tick(), m.spinner.Tick, m.subscribeCmd())}
func tick() tea.Cmd { return tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) })}
// subscribeCmd opens the ci.pipeline.subscribeLogs stream for current pipeline.// A consumer goroutine pushes decoded events onto the scheduler channel; the// returned command yields the first event into the bubbletea loop.func (m *pipelineModel) subscribeCmd() tea.Cmd { // cancel existing subscriptions just in case if m.cancel != nil { m.cancel() } sched := newEventScheduler() done := make(chan error, 1) ctx, cancel := context.WithCancel(context.Background()) m.cancel = cancel
pipelineId := m.pipeline.Id go func() { err := tangled.CiSubscribePipelineLogs(ctx, m.xrpcc, pipelineId, nil, sched) done <- err }()
return readEventCmd(sched.ch, done)}
func readEventCmd(events chan *tangled.CiSubscribePipelineLogs_Event, done chan error) tea.Cmd { return func() tea.Msg { ev, ok := <-events if !ok { return logDoneMsg{err: <-done} } return logEventMsg{ev: ev, events: events, done: done} }}
// fetchStatusCmd re-fetches the pipelinefunc (m *pipelineModel) fetchStatusCmd() tea.Cmd { pipelineId := 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) if err != nil { return statusUpdateErrMsg{err: fmt.Errorf("refreshing pipeline: %w", err)} } return statusUpdateMsg{pipeline: out} }}
func (m *pipelineModel) vpHeight() int { return max(m.height-2, 1) // topbar + empty line take 2 lines}
// resizeViewports updates all viewport dimensions and re-renders their content after a terminal resize.//// TODO: can be tedious if we have logs of logsfunc (m *pipelineModel) resizeViewports() { for _, wl := range m.logs { if !wl.ready { continue } atBottom := wl.vp.AtBottom() wl.vp.Width = m.width wl.vp.Height = m.vpHeight() wl.vp.SetContent(renderLogs(m.renderer, wl, m.width)) if atBottom { wl.vp.GotoBottom() } }}
func (m *pipelineModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) { switch msg := msg.(type) { case tea.WindowSizeMsg: m.width, m.height = msg.Width, msg.Height m.resizeViewports()
case tickMsg: // re-render running workflows so elapsed times advance m.refreshRunning() return m, tick()
case spinner.TickMsg: var cmd tea.Cmd m.spinner, cmd = m.spinner.Update(msg) return m, cmd
case tea.KeyMsg: switch msg.String() { case "q", "ctrl+c": if m.cancel != nil { m.cancel() } return m, tea.Quit case "tab", "right", "l": m.selected = (m.selected + 1) % len(m.pipeline.Workflows) return m, nil case "shift+tab", "left", "h": m.selected = (m.selected - 1 + len(m.pipeline.Workflows)) % len(m.pipeline.Workflows) return m, nil } if wl := m.selectedLogs(); wl != nil && wl.ready { switch msg.String() { case "g": wl.vp.GotoTop() return m, nil case "G": wl.vp.GotoBottom() return m, nil case "ctrl+e": wl.vp.ScrollDown(1) return m, nil case "ctrl+y": wl.vp.ScrollUp(1) return m, nil } var cmd tea.Cmd wl.vp, cmd = wl.vp.Update(msg) return m, cmd }
case logEventMsg: m.applyEvent(msg.ev) return m, readEventCmd(msg.events, msg.done)
case logDoneMsg: m.streamDone = true if !isExpectedClose(msg.err) { m.streamErr = msg.err } m.refreshAll() // resolve final workflow statuses once now that the stream has ended return m, m.fetchStatusCmd()
case statusUpdateMsg: m.pipeline = msg.pipeline known := make(map[string]bool, len(m.pipeline.Workflows)) for _, wf := range m.pipeline.Workflows { known[wf.Name] = true } for name := range m.logs { if !known[name] { delete(m.logs, name) } } if m.selected >= len(m.pipeline.Workflows) { m.selected = max(len(m.pipeline.Workflows)-1, 0) } m.refreshAll()
case statusUpdateErrMsg: // best-effort final status refresh; ignore failures }
return m, nil}
func (m *pipelineModel) selectedLogs() *workflowLogs { if len(m.pipeline.Workflows) < 1+m.selected { return nil } return m.logs[m.pipeline.Workflows[m.selected].Name]}
func (m *pipelineModel) initViewport(wl *workflowLogs) { if wl.ready { return } wl.vp = viewport.New(m.width, m.vpHeight()) wl.ready = true}
// ensureWorkflow returns the log state for a workflow, lazily creating its routing entry.func (m *pipelineModel) ensureWorkflow(name string) *workflowLogs { wl, ok := m.logs[name] if !ok { wl = &workflowLogs{stepIndex: make(map[int64]int)} m.logs[name] = wl } return wl}
// renderWorkflow re-renders a workflow's viewport, preserving bottom-stickiness.func (m *pipelineModel) renderWorkflow(wl *workflowLogs) { m.initViewport(wl) atBottom := wl.vp.AtBottom() wl.vp.SetContent(renderLogs(m.renderer, wl, m.width)) if atBottom { wl.vp.GotoBottom() }}
// refreshAll re-renders every initialized viewport.func (m *pipelineModel) refreshAll() { for _, wl := range m.logs { m.renderWorkflow(wl) }}
// refreshRunning re-renders workflows with unfinished steps so elapsed times advance.func (m *pipelineModel) refreshRunning() { if m.streamDone { return } for _, wl := range m.logs { if !wl.ready { continue } for i := range wl.steps { if !wl.steps[i].finished { m.renderWorkflow(wl) break } } }}
// applyEvent routes a decoded subscribeLogs event into the matching workflow.func (m *pipelineModel) applyEvent(ev *tangled.CiSubscribePipelineLogs_Event) { switch { case ev.Error != nil: if ev.Error.Message != "" { m.streamErr = fmt.Errorf("%s: %s", ev.Error.Error, ev.Error.Message) } else { m.streamErr = fmt.Errorf("%s", ev.Error.Error) }
case ev.Control != nil: c := ev.Control wl := m.ensureWorkflow(c.Workflow) switch derefStr(c.Status) { case "start": wl.stepIndex[c.Step] = len(wl.steps) wl.steps = append(wl.steps, step{ id: c.Step, name: c.Content, command: derefStr(c.Command), startTime: parseRFC3339(c.Time), }) case "end": if idx, ok := wl.stepIndex[c.Step]; ok { wl.steps[idx].endTime, wl.steps[idx].finished = parseRFC3339(c.Time), true } } m.renderWorkflow(wl)
case ev.Data != nil: d := ev.Data wl := m.ensureWorkflow(d.Workflow) if idx, ok := wl.stepIndex[d.Step]; ok { wl.steps[idx].lines = append(wl.steps[idx].lines, d.Content) } m.renderWorkflow(wl) }}
// isExpectedClose reports whether err is a clean websocket close (or nil).func isExpectedClose(err error) bool { if err == nil { return true } var ce *websocket.CloseError if errors.As(err, &ce) { switch ce.Code { case websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure: return true } } return false}
// renderLogs builds the full log content string for a workflow, used as viewport content.func renderLogs(r *lipgloss.Renderer, wl *workflowLogs, width int) string { headerStyle := r.NewStyle().Foreground(colorFg).Bold(true) cmdStyle := r.NewStyle().Foreground(colorBlue).Width(width) dimStyle := r.NewStyle().Faint(true) var sb strings.Builder for i := range wl.steps { st := &wl.steps[i] dur := "" if st.finished { dur = st.endTime.Sub(st.startTime).Round(time.Millisecond).String() } else if !st.startTime.IsZero() { dur = time.Since(st.startTime).Round(time.Second).String() } // build overlay: "── name ──...── dur ──" nameStr := headerStyle.Render(st.name + " ") durStr := headerStyle.Render(" " + dur + " ") nameW := lipgloss.Width(nameStr) durW := lipgloss.Width(durStr) fillW := max(width-nameW-durW, 0) fill := dimStyle.Render(strings.Repeat("─", fillW)) header := nameStr + fill + durStr sb.WriteString(header + "\n") if st.command != "" { sb.WriteString(cmdStyle.Render(st.command) + "\n") } for _, l := range st.lines { sb.WriteString(l + "\n") } sb.WriteString("\n") } return sb.String()}
func (m *pipelineModel) View() string { body := "" if wl := m.selectedLogs(); wl != nil && wl.ready { body = wl.vp.View() } if m.streamErr != nil { body = lipgloss.JoinVertical(lipgloss.Left, body, m.renderer.NewStyle().Foreground(colorBlue).Render("stream error: "+m.streamErr.Error())) } return lipgloss.JoinVertical(lipgloss.Left, m.topbarView(), "", body)}
// topbarView renders the single-line tab bar with workflow tabs left and trigger info + help right.func (m *pipelineModel) topbarView() string { r := m.renderer activeStyle := r.NewStyle().Background(colorBlue).Foreground(colorFg).Bold(true)
now := time.Now()
var tabs strings.Builder 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 if i == m.selected { tab := base if elapsed != "" { tab += " " + elapsed } tab += " " tabs.WriteString(activeStyle.Render(tab)) } else { tabs.WriteString(base) if elapsed != "" { dim := r.NewStyle().Faint(true) tabs.WriteString(" " + dim.Render(elapsed)) } tabs.WriteString(" ") } }
tabsStr := tabs.String() infoStr := triggerLine(r, m.pipeline.Trigger, m.pipeline.Commit) + " · " + helpText(r)
gap := max(m.width-lipgloss.Width(tabsStr)-lipgloss.Width(infoStr), 1)
return tabsStr + strings.Repeat(" ", gap) + infoStr}
func helpText(r *lipgloss.Renderer) string { key := r.NewStyle().Foreground(colorFg) action := r.NewStyle().Faint(true) sep := action.Render(" · ")
items := []string{ key.Render("←/→") + " " + action.Render("switch"), key.Render("↑/↓") + " " + action.Render("scroll"), key.Render("q") + " " + action.Render("quit"), } return strings.Join(items, sep)}
func shortSha(sha string) string { if len(sha) >= 8 { return sha[:8] } return sha}
func triggerLine(r *lipgloss.Renderer, t *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)") } return dim.Render(hash)}
func statusIcon(status string, spinnerFrame string) string { switch status { case "success": return "✓" case "failed": return "×" case "running": return spinnerFrame case "pending": return "·" case "timeout": return "⌀" case "cancelled": return "-" default: return "?" }}
func derefStr(s *string) string { if s == nil { return "" } return *s}
func parseRFC3339(s string) time.Time { t, err := time.Parse(time.RFC3339, s) if err != nil { return time.Time{} } return t}