Monorepo for Tangled
appview/pipelines: live status updates in workflows page
Signed-off-by: oppiliappan <me@oppi.li>
authored by
oppiliappan
and committed by
Tangled
(May 26, 2026, 9:08 AM +0300)
0cb049f1
2c52d785
+68
-19
+9
appview/pages/pages.go
+9
appview/pages/pages.go
···
1543
1543
return p.executePlain("repo/pipelines/fragments/logLine", w, params)
1544
1544
}
1545
1545
1546
+
type WorkflowSymbolOOBParams struct {
1547
+
Name string
1548
+
Statuses models.WorkflowStatus
1549
+
}
1550
+
1551
+
func (p *Pages) WorkflowSymbolOOB(w io.Writer, params WorkflowSymbolOOBParams) error {
1552
+
return p.executePlain("repo/pipelines/fragments/workflowSymbolOOB", w, params)
1553
+
}
1554
+
1546
1555
type WorkflowParams struct {
1547
1556
LoggedInUser *oauth.MultiAccountUser
1548
1557
RepoInfo repoinfo.RepoInfo
+5
appview/pages/templates/repo/pipelines/fragments/workflowSymbolOOB.html
+5
appview/pages/templates/repo/pipelines/fragments/workflowSymbolOOB.html
···
1
+
{{ define "repo/pipelines/fragments/workflowSymbolOOB" }}
2
+
<div id="workflow-symbol-{{ normalizeForHtmlId .Name }}" class="flex-shrink-0" hx-swap-oob="outerHTML:#workflow-symbol-{{ normalizeForHtmlId .Name }}">
3
+
{{ template "repo/pipelines/fragments/workflowSymbol" .Statuses }}
4
+
</div>
5
+
{{ end }}
+1
-1
appview/pages/templates/repo/pipelines/workflow.html
+1
-1
appview/pages/templates/repo/pipelines/workflow.html
···
49
49
{{ $kind := $lastStatus.Status.String }}
50
50
51
51
<div id="left" class="flex items-center gap-2 flex-1 min-w-0">
52
-
<div class="flex-shrink-0">
52
+
<div id="workflow-symbol-{{ normalizeForHtmlId $name }}" class="flex-shrink-0">
53
53
{{ template "repo/pipelines/fragments/workflowSymbol" $all }}
54
54
</div>
55
55
<span class="truncate" title="{{ $name }}">
+52
-18
appview/pipelines/pipelines.go
+52
-18
appview/pipelines/pipelines.go
···
29
29
)
30
30
31
31
type Pipelines struct {
32
-
repoResolver *reporesolver.RepoResolver
33
-
idResolver *idresolver.Resolver
34
-
config *config.Config
35
-
oauth *oauth.OAuth
36
-
pages *pages.Pages
37
-
spindlestream *eventconsumer.Consumer
38
-
db *db.DB
39
-
enforcer *rbac.Enforcer
40
-
logger *slog.Logger
32
+
repoResolver *reporesolver.RepoResolver
33
+
idResolver *idresolver.Resolver
34
+
config *config.Config
35
+
oauth *oauth.OAuth
36
+
pages *pages.Pages
37
+
spindlestream *eventconsumer.Consumer
38
+
pipelineNotifier *StatusNotifier
39
+
db *db.DB
40
+
enforcer *rbac.Enforcer
41
+
logger *slog.Logger
41
42
}
42
43
43
44
func (p *Pipelines) Router(mw *middleware.Middleware) http.Handler {
···
57
58
repoResolver *reporesolver.RepoResolver,
58
59
pages *pages.Pages,
59
60
spindlestream *eventconsumer.Consumer,
61
+
pipelineNotifier *StatusNotifier,
60
62
idResolver *idresolver.Resolver,
61
63
db *db.DB,
62
64
config *config.Config,
···
64
66
logger *slog.Logger,
65
67
) *Pipelines {
66
68
return &Pipelines{
67
-
oauth: oauth,
68
-
repoResolver: repoResolver,
69
-
pages: pages,
70
-
idResolver: idResolver,
71
-
config: config,
72
-
spindlestream: spindlestream,
73
-
db: db,
74
-
enforcer: enforcer,
75
-
logger: logger,
69
+
oauth: oauth,
70
+
repoResolver: repoResolver,
71
+
pages: pages,
72
+
idResolver: idResolver,
73
+
config: config,
74
+
spindlestream: spindlestream,
75
+
pipelineNotifier: pipelineNotifier,
76
+
db: db,
77
+
enforcer: enforcer,
78
+
logger: logger,
76
79
}
77
80
}
78
81
···
212
215
knot := f.Knot
213
216
rkey := singlePipeline.Rkey
214
217
218
+
statusCh := p.pipelineNotifier.Subscribe(singlePipeline.AtUri())
219
+
defer p.pipelineNotifier.Unsubscribe(singlePipeline.AtUri(), statusCh)
220
+
215
221
if spindle == "" || knot == "" || rkey == "" {
216
222
http.Error(w, "invalid repo info", http.StatusBadRequest)
217
223
return
···
327
333
if err = clientConn.WriteMessage(websocket.TextMessage, fragment.Bytes()); err != nil {
328
334
l.Error("error writing to client", "err", err)
329
335
return
336
+
}
337
+
338
+
case _, ok := <-statusCh:
339
+
if !ok {
340
+
continue
341
+
}
342
+
fresh, err := db.GetPipelineStatuses(
343
+
p.db,
344
+
1,
345
+
orm.FilterEq("p.repo_did", f.RepoDid),
346
+
orm.FilterEq("p.id", pipelineId),
347
+
)
348
+
if err != nil || len(fresh) == 0 {
349
+
continue
350
+
}
351
+
for name, ws := range fresh[0].Statuses {
352
+
fragment.Reset()
353
+
if err = p.pages.WorkflowSymbolOOB(&fragment, pages.WorkflowSymbolOOBParams{
354
+
Name: name,
355
+
Statuses: ws,
356
+
}); err != nil {
357
+
l.Error("failed to render workflow symbol OOB", "err", err)
358
+
continue
359
+
}
360
+
if err = clientConn.WriteMessage(websocket.TextMessage, fragment.Bytes()); err != nil {
361
+
l.Error("error writing workflow symbol to client", "err", err)
362
+
return
363
+
}
330
364
}
331
365
332
366
case <-time.After(30 * time.Second):