-447

appview/db/pipeline.go

Reviewed

··· 1 1 - package db 2 2 - 3 3 - import ( 4 4 - "context" 5 5 - "database/sql" 6 6 - "fmt" 7 7 - "slices" 8 8 - "strings" 9 9 - "time" 10 10 - 11 11 - "github.com/bluesky-social/indigo/atproto/syntax" 12 12 - "tangled.org/core/appview/models" 13 13 - "tangled.org/core/orm" 14 14 - ) 15 15 - 16 16 - func GetPipelines(e Execer, filters ...orm.Filter) ([]models.Pipeline, error) { 17 17 - var pipelines []models.Pipeline 18 18 - 19 19 - var conditions []string 20 20 - var args []any 21 21 - for _, filter := range filters { 22 22 - conditions = append(conditions, filter.Condition()) 23 23 - args = append(args, filter.Arg()...) 24 24 - } 25 25 - 26 26 - whereClause := "" 27 27 - if conditions != nil { 28 28 - whereClause = " where " + strings.Join(conditions, " and ") 29 29 - } 30 30 - 31 31 - query := fmt.Sprintf(`select id, rkey, knot, repo_owner, repo_name, sha, created, repo_did from pipelines %s`, whereClause) 32 32 - 33 33 - rows, err := e.Query(query, args...) 34 34 - 35 35 - if err != nil { 36 36 - return nil, err 37 37 - } 38 38 - defer rows.Close() 39 39 - 40 40 - for rows.Next() { 41 41 - var pipeline models.Pipeline 42 42 - var createdAt string 43 43 - var repoDid sql.NullString 44 44 - err = rows.Scan( 45 45 - &pipeline.Id, 46 46 - &pipeline.Rkey, 47 47 - &pipeline.Knot, 48 48 - &pipeline.RepoOwner, 49 49 - &pipeline.RepoName, 50 50 - &pipeline.Sha, 51 51 - &createdAt, 52 52 - &repoDid, 53 53 - ) 54 54 - if err != nil { 55 55 - return nil, err 56 56 - } 57 57 - 58 58 - if t, err := time.Parse(time.RFC3339, createdAt); err == nil { 59 59 - pipeline.Created = t 60 60 - } 61 61 - if repoDid.Valid { 62 62 - pipeline.RepoDid = repoDid.String 63 63 - } 64 64 - 65 65 - pipelines = append(pipelines, pipeline) 66 66 - } 67 67 - 68 68 - if err = rows.Err(); err != nil { 69 69 - return nil, err 70 70 - } 71 71 - 72 72 - return pipelines, nil 73 73 - } 74 74 - 75 75 - func AddPipeline(e Execer, pipeline models.Pipeline) error { 76 76 - var repoDid *string 77 77 - if pipeline.RepoDid != "" { 78 78 - repoDid = &pipeline.RepoDid 79 79 - } 80 80 - 81 81 - args := []any{ 82 82 - pipeline.Rkey, 83 83 - pipeline.Knot, 84 84 - pipeline.RepoOwner, 85 85 - pipeline.RepoName, 86 86 - pipeline.TriggerId, 87 87 - pipeline.Sha, 88 88 - repoDid, 89 89 - } 90 90 - 91 91 - placeholders := make([]string, len(args)) 92 92 - for i := range placeholders { 93 93 - placeholders[i] = "?" 94 94 - } 95 95 - 96 96 - query := fmt.Sprintf(` 97 97 - insert or ignore into pipelines ( 98 98 - rkey, 99 99 - knot, 100 100 - repo_owner, 101 101 - repo_name, 102 102 - trigger_id, 103 103 - sha, 104 104 - repo_did 105 105 - ) values (%s) 106 106 - `, strings.Join(placeholders, ",")) 107 107 - 108 108 - _, err := e.Exec(query, args...) 109 109 - 110 110 - return err 111 111 - } 112 112 - 113 113 - func AddTrigger(e Execer, trigger models.Trigger) (int64, error) { 114 114 - args := []any{ 115 115 - trigger.Kind, 116 116 - trigger.PushRef, 117 117 - trigger.PushNewSha, 118 118 - trigger.PushOldSha, 119 119 - trigger.PRSourceBranch, 120 120 - trigger.PRTargetBranch, 121 121 - trigger.PRSourceSha, 122 122 - trigger.PRAction, 123 123 - } 124 124 - 125 125 - placeholders := make([]string, len(args)) 126 126 - for i := range placeholders { 127 127 - placeholders[i] = "?" 128 128 - } 129 129 - 130 130 - query := fmt.Sprintf(`insert or ignore into triggers ( 131 131 - kind, 132 132 - push_ref, 133 133 - push_new_sha, 134 134 - push_old_sha, 135 135 - pr_source_branch, 136 136 - pr_target_branch, 137 137 - pr_source_sha, 138 138 - pr_action 139 139 - ) values (%s)`, strings.Join(placeholders, ",")) 140 140 - 141 141 - res, err := e.Exec(query, args...) 142 142 - if err != nil { 143 143 - return 0, err 144 144 - } 145 145 - 146 146 - return res.LastInsertId() 147 147 - } 148 148 - 149 149 - func AddPipelineStatus(ctx context.Context, e Execer, status models.PipelineStatus) error { 150 150 - args := []any{ 151 151 - status.Spindle, 152 152 - status.Rkey, 153 153 - status.PipelineKnot, 154 154 - status.PipelineRkey, 155 155 - status.Workflow, 156 156 - status.Status, 157 157 - status.Error, 158 158 - status.ExitCode, 159 159 - status.Created.Format(time.RFC3339), 160 160 - } 161 161 - 162 162 - placeholders := make([]string, len(args)) 163 163 - for i := range placeholders { 164 164 - placeholders[i] = "?" 165 165 - } 166 166 - 167 167 - query := fmt.Sprintf(` 168 168 - insert or ignore into pipeline_statuses ( 169 169 - spindle, 170 170 - rkey, 171 171 - pipeline_knot, 172 172 - pipeline_rkey, 173 173 - workflow, 174 174 - status, 175 175 - error, 176 176 - exit_code, 177 177 - created 178 178 - ) values (%s) 179 179 - `, strings.Join(placeholders, ",")) 180 180 - 181 181 - _, err := e.ExecContext(ctx, query, args...) 182 182 - return err 183 183 - } 184 184 - 185 185 - // this is a mega query, but the most useful one: 186 186 - // get N pipelines, for each one get the latest status of its N workflows 187 187 - // 188 188 - // the pipelines table is aliased to `p` 189 189 - // the triggers table is aliased to `t` 190 190 - func GetPipelineStatuses(e Execer, limit int, filters ...orm.Filter) ([]models.Pipeline, error) { 191 191 - var conditions []string 192 192 - var args []any 193 193 - for _, filter := range filters { 194 194 - conditions = append(conditions, filter.Condition()) 195 195 - args = append(args, filter.Arg()...) 196 196 - } 197 197 - 198 198 - whereClause := "" 199 199 - if conditions != nil { 200 200 - whereClause = " where " + strings.Join(conditions, " and ") 201 201 - } 202 202 - 203 203 - query := fmt.Sprintf(` 204 204 - select 205 205 - p.id, 206 206 - p.knot, 207 207 - p.rkey, 208 208 - p.repo_owner, 209 209 - p.repo_name, 210 210 - p.sha, 211 211 - p.created, 212 212 - p.repo_did, 213 213 - t.id, 214 214 - t.kind, 215 215 - t.push_ref, 216 216 - t.push_new_sha, 217 217 - t.push_old_sha, 218 218 - t.pr_source_branch, 219 219 - t.pr_target_branch, 220 220 - t.pr_source_sha, 221 221 - t.pr_action 222 222 - from 223 223 - pipelines p 224 224 - join 225 225 - triggers t ON p.trigger_id = t.id 226 226 - %s 227 227 - order by p.created desc 228 228 - limit %d 229 229 - `, whereClause, limit) 230 230 - 231 231 - rows, err := e.Query(query, args...) 232 232 - if err != nil { 233 233 - return nil, err 234 234 - } 235 235 - defer rows.Close() 236 236 - 237 237 - pipelines := make(map[syntax.ATURI]models.Pipeline) 238 238 - for rows.Next() { 239 239 - var p models.Pipeline 240 240 - var t models.Trigger 241 241 - var created string 242 242 - var repoDid sql.NullString 243 243 - 244 244 - err := rows.Scan( 245 245 - &p.Id, 246 246 - &p.Knot, 247 247 - &p.Rkey, 248 248 - &p.RepoOwner, 249 249 - &p.RepoName, 250 250 - &p.Sha, 251 251 - &created, 252 252 - &repoDid, 253 253 - &p.TriggerId, 254 254 - &t.Kind, 255 255 - &t.PushRef, 256 256 - &t.PushNewSha, 257 257 - &t.PushOldSha, 258 258 - &t.PRSourceBranch, 259 259 - &t.PRTargetBranch, 260 260 - &t.PRSourceSha, 261 261 - &t.PRAction, 262 262 - ) 263 263 - if err != nil { 264 264 - return nil, err 265 265 - } 266 266 - 267 267 - p.Created, err = time.Parse(time.RFC3339, created) 268 268 - if err != nil { 269 269 - return nil, fmt.Errorf("invalid pipeline created timestamp %q: %w", created, err) 270 270 - } 271 271 - if repoDid.Valid { 272 272 - p.RepoDid = repoDid.String 273 273 - } 274 274 - 275 275 - t.Id = p.TriggerId 276 276 - p.Trigger = &t 277 277 - p.Statuses = make(map[string]models.WorkflowStatus) 278 278 - 279 279 - pipelines[p.AtUri()] = p 280 280 - } 281 281 - 282 282 - // get all statuses 283 283 - // the where clause here is of the form: 284 284 - // 285 285 - // and ( 286 286 - // (ps.pipeline_knot = k1 and ps.pipeline_rkey = r1) 287 287 - // or (ps.pipeline_knot = k2 and ps.pipeline_rkey = r2) 288 288 - // ) 289 289 - // 290 290 - // the join on pipelines and repos enforces that the status was emitted 291 291 - // by the spindle that is actually registered for the pipeline's repo. 292 292 - conditions = nil 293 293 - args = nil 294 294 - for _, p := range pipelines { 295 295 - knotFilter := orm.FilterEq("ps.pipeline_knot", p.Knot) 296 296 - rkeyFilter := orm.FilterEq("ps.pipeline_rkey", p.Rkey) 297 297 - conditions = append(conditions, fmt.Sprintf("(%s and %s)", knotFilter.Condition(), rkeyFilter.Condition())) 298 298 - args = append(args, p.Knot) 299 299 - args = append(args, p.Rkey) 300 300 - } 301 301 - whereClause = "" 302 302 - if conditions != nil { 303 303 - whereClause = "and (" + strings.Join(conditions, " or ") + ")" 304 304 - } 305 305 - query = fmt.Sprintf(` 306 306 - select 307 307 - ps.id, ps.spindle, ps.rkey, ps.pipeline_knot, ps.pipeline_rkey, 308 308 - ps.created, ps.workflow, ps.status, ps.error, ps.exit_code 309 309 - from 310 310 - pipeline_statuses ps 311 311 - join 312 312 - pipelines p on p.knot = ps.pipeline_knot and p.rkey = ps.pipeline_rkey 313 313 - join 314 314 - repos r on r.repo_did = p.repo_did 315 315 - where 316 316 - ps.spindle = r.spindle 317 317 - %s 318 318 - `, whereClause) 319 319 - 320 320 - rows, err = e.Query(query, args...) 321 321 - if err != nil { 322 322 - return nil, err 323 323 - } 324 324 - defer rows.Close() 325 325 - 326 326 - for rows.Next() { 327 327 - var ps models.PipelineStatus 328 328 - var created string 329 329 - 330 330 - err := rows.Scan( 331 331 - &ps.ID, 332 332 - &ps.Spindle, 333 333 - &ps.Rkey, 334 334 - &ps.PipelineKnot, 335 335 - &ps.PipelineRkey, 336 336 - &created, 337 337 - &ps.Workflow, 338 338 - &ps.Status, 339 339 - &ps.Error, 340 340 - &ps.ExitCode, 341 341 - ) 342 342 - if err != nil { 343 343 - return nil, err 344 344 - } 345 345 - 346 346 - ps.Created, err = time.Parse(time.RFC3339, created) 347 347 - if err != nil { 348 348 - return nil, fmt.Errorf("invalid status created timestamp %q: %w", created, err) 349 349 - } 350 350 - 351 351 - pipelineAt := ps.PipelineAt() 352 352 - 353 353 - // extract 354 354 - pipeline, ok := pipelines[pipelineAt] 355 355 - if !ok { 356 356 - continue 357 357 - } 358 358 - statuses, _ := pipeline.Statuses[ps.Workflow] 359 359 - if !ok { 360 360 - pipeline.Statuses[ps.Workflow] = models.WorkflowStatus{} 361 361 - } 362 362 - 363 363 - // append 364 364 - statuses.Data = append(statuses.Data, ps) 365 365 - 366 366 - // reassign 367 367 - pipeline.Statuses[ps.Workflow] = statuses 368 368 - pipelines[pipelineAt] = pipeline 369 369 - } 370 370 - 371 371 - var all []models.Pipeline 372 372 - for _, p := range pipelines { 373 373 - for _, s := range p.Statuses { 374 374 - slices.SortFunc(s.Data, func(a, b models.PipelineStatus) int { 375 375 - if a.Created.After(b.Created) { 376 376 - return 1 377 377 - } 378 378 - if a.Created.Before(b.Created) { 379 379 - return -1 380 380 - } 381 381 - if a.ID > b.ID { 382 382 - return 1 383 383 - } 384 384 - if a.ID < b.ID { 385 385 - return -1 386 386 - } 387 387 - return 0 388 388 - }) 389 389 - } 390 390 - all = append(all, p) 391 391 - } 392 392 - 393 393 - // sort pipelines by date 394 394 - slices.SortFunc(all, func(a, b models.Pipeline) int { 395 395 - if a.Created.After(b.Created) { 396 396 - return -1 397 397 - } 398 398 - return 1 399 399 - }) 400 400 - 401 401 - return all, nil 402 402 - } 403 403 - 404 404 - // the pipelines table is aliased to `p` 405 405 - // the triggers table is aliased to `t` 406 406 - func GetPipelineCount(e Execer, filters ...orm.Filter) (int64, error) { 407 407 - var conditions []string 408 408 - var args []any 409 409 - for _, filter := range filters { 410 410 - conditions = append(conditions, filter.Condition()) 411 411 - args = append(args, filter.Arg()...) 412 412 - } 413 413 - 414 414 - whereClause := "" 415 415 - if conditions != nil { 416 416 - whereClause = " where " + strings.Join(conditions, " and ") 417 417 - } 418 418 - 419 419 - query := fmt.Sprintf(` 420 420 - select 421 421 - count(1) 422 422 - from 423 423 - pipelines p 424 424 - join 425 425 - triggers t ON p.trigger_id = t.id 426 426 - %s 427 427 - `, whereClause) 428 428 - 429 429 - rows, err := e.Query(query, args...) 430 430 - if err != nil { 431 431 - return 0, err 432 432 - } 433 433 - defer rows.Close() 434 434 - 435 435 - for rows.Next() { 436 436 - var count int64 437 437 - err := rows.Scan(&count) 438 438 - if err != nil { 439 439 - return 0, err 440 440 - } 441 441 - 442 442 - return count, nil 443 443 - } 444 444 - 445 445 - // unreachable 446 446 - return 0, nil 447 447 - }

-116

appview/db/pipeline_test.go

Reviewed

··· 1 1 - package db 2 2 - 3 3 - import ( 4 4 - "context" 5 5 - "strings" 6 6 - "testing" 7 7 - "time" 8 8 - 9 9 - "github.com/bluesky-social/indigo/atproto/syntax" 10 10 - "tangled.org/core/appview/models" 11 11 - "tangled.org/core/orm" 12 12 - spindle "tangled.org/core/spindle/models" 13 13 - "tangled.org/core/workflow" 14 14 - ) 15 15 - 16 16 - // seedPipeline inserts a trigger + pipeline row and returns the pipeline. 17 17 - func seedPipeline(t *testing.T, d *DB, knot, rkey, repoDid string) models.Pipeline { 18 18 - t.Helper() 19 19 - sha := strings.Repeat("a", 40) 20 20 - ref := "refs/heads/main" 21 21 - newSha := sha 22 22 - oldSha := strings.Repeat("0", 40) 23 23 - trigger := models.Trigger{ 24 24 - Kind: workflow.TriggerKindPush, 25 25 - PushRef: &ref, 26 26 - PushNewSha: &newSha, 27 27 - PushOldSha: &oldSha, 28 28 - } 29 29 - tx, err := d.Begin() 30 30 - if err != nil { 31 31 - t.Fatalf("Begin: %v", err) 32 32 - } 33 33 - triggerID, err := AddTrigger(tx, trigger) 34 34 - if err != nil { 35 35 - tx.Rollback() 36 36 - t.Fatalf("AddTrigger: %v", err) 37 37 - } 38 38 - pipeline := models.Pipeline{ 39 39 - Knot: knot, 40 40 - Rkey: rkey, 41 41 - RepoOwner: syntax.DID("did:plc:owner"), 42 42 - RepoName: "repo", 43 43 - RepoDid: repoDid, 44 44 - TriggerId: int(triggerID), 45 45 - Sha: sha, 46 46 - } 47 47 - if err := AddPipeline(tx, pipeline); err != nil { 48 48 - tx.Rollback() 49 49 - t.Fatalf("AddPipeline: %v", err) 50 50 - } 51 51 - if err := tx.Commit(); err != nil { 52 52 - t.Fatalf("Commit: %v", err) 53 53 - } 54 54 - return pipeline 55 55 - } 56 56 - 57 57 - // seedStatus inserts a pipeline_status row directly. 58 58 - func seedStatus(t *testing.T, d *DB, spindleInstance, rkey, pipelineKnot, pipelineRkey, workflow string) { 59 59 - t.Helper() 60 60 - status := models.PipelineStatus{ 61 61 - Spindle: spindleInstance, 62 62 - Rkey: rkey, 63 63 - PipelineKnot: pipelineKnot, 64 64 - PipelineRkey: pipelineRkey, 65 65 - Workflow: workflow, 66 66 - Status: spindle.StatusKindSuccess, 67 67 - Created: time.Now(), 68 68 - } 69 69 - if err := AddPipelineStatus(context.Background(), d, status); err != nil { 70 70 - t.Fatalf("AddPipelineStatus: %v", err) 71 71 - } 72 72 - } 73 73 - 74 74 - // TestGetPipelineStatuses_SpindleValidation verifies that GetPipelineStatuses 75 75 - // only returns statuses emitted by the spindle registered for the pipeline's 76 76 - // repo, and silently drops statuses from a rogue spindle. 77 77 - func TestGetPipelineStatuses_SpindleValidation(t *testing.T) { 78 78 - d := newTestDB(t) 79 79 - 80 80 - const ( 81 81 - knot = "knot.example.com" 82 82 - correctSpindle = "spindle.example.com" 83 83 - rogueSpindle = "evil.example.com" 84 84 - repoDid = "did:plc:testrepo" 85 85 - pipelineRkey = "pipeline1" 86 86 - ) 87 87 - 88 88 - // seed repo with the correct spindle 89 89 - repo := seedRepo(t, d, "did:plc:owner", knot, "repo", "repo", repoDid) 90 90 - if err := UpdateSpindle(d, repo.RepoDid, &[]string{correctSpindle}[0]); err != nil { 91 91 - t.Fatalf("UpdateSpindle: %v", err) 92 92 - } 93 93 - 94 94 - // seed the pipeline for this repo 95 95 - seedPipeline(t, d, knot, pipelineRkey, repoDid) 96 96 - 97 97 - // insert one status from the correct spindle, one from a rogue spindle 98 98 - seedStatus(t, d, correctSpindle, "status-valid", knot, pipelineRkey, "build") 99 99 - seedStatus(t, d, rogueSpindle, "status-rogue", knot, pipelineRkey, "build") 100 100 - 101 101 - pipelines, err := GetPipelineStatuses(d, 10, orm.FilterEq("p.repo_did", repoDid)) 102 102 - if err != nil { 103 103 - t.Fatalf("GetPipelineStatuses: %v", err) 104 104 - } 105 105 - if len(pipelines) != 1 { 106 106 - t.Fatalf("expected 1 pipeline, got %d", len(pipelines)) 107 107 - } 108 108 - 109 109 - statuses := pipelines[0].Statuses["build"].Data 110 110 - if len(statuses) != 1 { 111 111 - t.Fatalf("expected 1 status (from correct spindle), got %d", len(statuses)) 112 112 - } 113 113 - if statuses[0].Spindle != correctSpindle { 114 114 - t.Errorf("expected spindle %q, got %q", correctSpindle, statuses[0].Spindle) 115 115 - } 116 116 - }

+7 -7

appview/pages/pages.go

Reviewed

··· 891 891 EmailToDid map[string]string 892 892 VerifiedCommits commitverify.VerifiedCommits 893 893 Languages []types.RepoLanguageDetails 894 894 - Pipelines map[string]models.Pipeline 894 894 + Pipelines map[string]*tangled.CiDefs_Pipeline 895 895 NeedsKnotUpgrade bool 896 896 KnotUnreachable bool 897 897 types.RepoIndexResponse ··· 960 960 Active string 961 961 EmailToDid map[string]string 962 962 VerifiedCommits commitverify.VerifiedCommits 963 963 - Pipelines map[string]models.Pipeline 963 963 + Pipelines map[string]*tangled.CiDefs_Pipeline 964 964 965 965 types.RepoLogResponse 966 966 } ··· 975 975 RepoInfo repoinfo.RepoInfo 976 976 Active string 977 977 EmailToDid map[string]string 978 978 - Pipeline *models.Pipeline 978 978 + Pipeline *tangled.CiDefs_Pipeline 979 979 DiffOpts types.DiffOpts 980 980 981 981 // singular because it's always going to be just one ··· 1376 1376 FilterQuery string 1377 1377 BaseFilterQuery string 1378 1378 Stacks []models.Stack 1379 1379 - Pipelines map[string]models.Pipeline 1379 1379 + Pipelines map[string]tangled.CiDefs_Pipeline 1380 1380 LabelDefs map[string]*models.LabelDefinition 1381 1381 Page pagination.Page 1382 1382 PullCount int ··· 1416 1416 BranchDeleteStatus *models.BranchDeleteStatus 1417 1417 MergeCheck types.MergeCheckResponse 1418 1418 ResubmitCheck ResubmitResult 1419 1419 - Pipelines map[string]models.Pipeline 1419 1419 + Pipelines map[string]tangled.CiDefs_Pipeline 1420 1420 Diff types.DiffRenderer 1421 1421 DiffOpts types.DiffOpts 1422 1422 ActiveRound int ··· 1588 1588 type PipelinesParams struct { 1589 1589 BaseParams 1590 1590 RepoInfo repoinfo.RepoInfo 1591 1591 - Pipelines []models.Pipeline 1591 1591 + Pipelines []*tangled.CiDefs_Pipeline 1592 1592 Active string 1593 1593 FilterKind string 1594 1594 Total int64 ··· 1642 1642 type WorkflowParams struct { 1643 1643 BaseParams 1644 1644 RepoInfo repoinfo.RepoInfo 1645 1645 - Pipeline models.Pipeline 1645 1645 + Pipeline *tangled.CiDefs_Pipeline 1646 1646 Workflow string 1647 1647 LogUrl string 1648 1648 Active string

+23 -31

appview/pipelines/logs.go

Reviewed

··· 1 1 package pipelines 2 2 3 3 import ( 4 4 + "errors" 4 5 "html/template" 5 5 - "path" 6 6 "regexp" 7 7 "strings" 8 8 + "time" 8 9 9 10 terminal "github.com/buildkite/terminal-to-html/v3" 10 11 "github.com/gorilla/websocket" 11 12 "tangled.org/core/appview/pages/markup/sanitizer" 12 12 - "tangled.org/core/hostutil" 13 13 ) 14 14 15 15 // matches any ANSI escape sequence: ESC [ <params> m ··· 50 50 return template.HTML(sanitized) 51 51 } 52 52 53 53 - type LogEvent struct { 54 54 - Msg []byte 55 55 - Err error 56 56 - } 57 57 - 58 58 - func (ev *LogEvent) IsCloseError() bool { 59 59 - return websocket.IsCloseError( 60 60 - ev.Err, 61 61 - websocket.CloseNormalClosure, 62 62 - websocket.CloseGoingAway, 63 63 - websocket.CloseAbnormalClosure, 64 64 - ) 65 65 - } 66 66 - 67 67 - func ReadLogs(conn *websocket.Conn, ch chan LogEvent) { 68 68 - defer close(ch) 69 69 - for { 70 70 - if conn == nil { 71 71 - return 72 72 - } 73 73 - _, msg, err := conn.ReadMessage() 74 74 - if err != nil { 75 75 - ch <- LogEvent{Err: err} 76 76 - return 53 53 + // isExpectedClose reports whether err is a clean websocket close (or nil). 54 54 + func isExpectedClose(err error) bool { 55 55 + if err == nil { 56 56 + return true 57 57 + } 58 58 + var ce *websocket.CloseError 59 59 + if errors.As(err, &ce) { 60 60 + switch ce.Code { 61 61 + case websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure: 62 62 + return true 77 63 } 78 78 - ch <- LogEvent{Msg: msg} 79 64 } 65 65 + return false 80 66 } 81 67 82 82 - func SpindleURL(spindle, knot, rkey, workflow string) string { 83 83 - url, err := hostutil.EnsureWsScheme(spindle) 84 84 - if err != nil { 68 68 + func derefStr(s *string) string { 69 69 + if s == nil { 85 70 return "" 86 71 } 72 72 + return *s 73 73 + } 87 74 88 88 - return url + path.Join("/logs", knot, rkey, workflow) 75 75 + func parseRFC3339(s string) time.Time { 76 76 + t, err := time.Parse(time.RFC3339, s) 77 77 + if err != nil { 78 78 + return time.Time{} 79 79 + } 80 80 + return t 89 81 }

-52

appview/pipelines/notifier.go

Reviewed

··· 1 1 - package pipelines 2 2 - 3 3 - import ( 4 4 - "sync" 5 5 - 6 6 - "github.com/bluesky-social/indigo/atproto/syntax" 7 7 - "tangled.org/core/notifier" 8 8 - ) 9 9 - 10 10 - // StatusNotifier is a keyed broadcast notifier for pipeline status changes, keyed by the pipeline's AT URI. 11 11 - // 12 12 - // subscribers are notified whenever a status update arrives for that pipeline 13 13 - type StatusNotifier struct { 14 14 - mu sync.Mutex 15 15 - keys map[syntax.ATURI]*notifier.Notifier 16 16 - } 17 17 - 18 18 - func NewStatusNotifier() *StatusNotifier { 19 19 - return &StatusNotifier{ 20 20 - keys: make(map[syntax.ATURI]*notifier.Notifier), 21 21 - } 22 22 - } 23 23 - 24 24 - func (n *StatusNotifier) Publish(uri syntax.ATURI) { 25 25 - n.mu.Lock() 26 26 - p, ok := n.keys[uri] 27 27 - n.mu.Unlock() 28 28 - if ok { 29 29 - p.NotifyAll() 30 30 - } 31 31 - } 32 32 - 33 33 - func (n *StatusNotifier) Subscribe(uri syntax.ATURI) chan struct{} { 34 34 - n.mu.Lock() 35 35 - p, ok := n.keys[uri] 36 36 - if !ok { 37 37 - nb := notifier.New() 38 38 - p = &nb 39 39 - n.keys[uri] = p 40 40 - } 41 41 - n.mu.Unlock() 42 42 - return p.Subscribe() 43 43 - } 44 44 - 45 45 - func (n *StatusNotifier) Unsubscribe(uri syntax.ATURI, ch chan struct{}) { 46 46 - n.mu.Lock() 47 47 - p, ok := n.keys[uri] 48 48 - n.mu.Unlock() 49 49 - if ok { 50 50 - p.Unsubscribe(ch) 51 51 - } 52 52 - }

+206 -229

appview/pipelines/pipelines.go

Reviewed

··· 3 3 import ( 4 4 "bytes" 5 5 "context" 6 6 - "encoding/json" 7 7 - "fmt" 8 6 "log/slog" 9 7 "net/http" 8 8 + "sync" 10 9 "time" 11 10 12 11 "tangled.org/core/api/tangled" 13 12 "tangled.org/core/appview/config" 14 13 "tangled.org/core/appview/db" 15 14 "tangled.org/core/appview/middleware" 16 16 - "tangled.org/core/appview/models" 17 15 "tangled.org/core/appview/oauth" 18 16 "tangled.org/core/appview/pages" 19 17 "tangled.org/core/appview/reporesolver" 20 20 - "tangled.org/core/eventconsumer" 21 18 "tangled.org/core/hostutil" 22 19 "tangled.org/core/idresolver" 20 20 + "tangled.org/core/lexutil" 23 21 "tangled.org/core/orm" 24 22 "tangled.org/core/rbac" 25 25 - spindlemodel "tangled.org/core/spindle/models" 26 23 24 24 + "github.com/bluesky-social/indigo/atproto/syntax" 25 25 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 27 26 "github.com/go-chi/chi/v5" 28 27 "github.com/gorilla/websocket" 29 28 ) 30 29 31 30 type Pipelines struct { 32 32 - repoResolver *reporesolver.RepoResolver 33 33 - idResolver *idresolver.Resolver 34 34 - config *config.Config 35 35 - oauth *oauth.OAuth 36 36 - pages *pages.Pages 37 37 - spindlestream *eventconsumer.Consumer 38 38 - pipelineNotifier *StatusNotifier 39 39 - db *db.DB 40 40 - enforcer *rbac.Enforcer 41 41 - logger *slog.Logger 31 31 + repoResolver *reporesolver.RepoResolver 32 32 + idResolver *idresolver.Resolver 33 33 + config *config.Config 34 34 + oauth *oauth.OAuth 35 35 + pages *pages.Pages 36 36 + db *db.DB 37 37 + enforcer *rbac.Enforcer 38 38 + logger *slog.Logger 42 39 } 43 40 44 41 func (p *Pipelines) Router(mw *middleware.Middleware) http.Handler { ··· 48 45 r.Get("/{pipeline}/workflow/{workflow}/logs", p.Logs) 49 46 r. 50 47 With(mw.RepoPermissionMiddleware("repo:owner")). 51 51 - Post("/{pipeline}/workflow/{workflow}/cancel", p.Cancel) 48 48 + Post("/{pipeline}/workflow/{workflow}/cancel", p.CancelWorkflow) 52 49 53 50 return r 54 51 } ··· 57 54 oauth *oauth.OAuth, 58 55 repoResolver *reporesolver.RepoResolver, 59 56 pages *pages.Pages, 60 60 - spindlestream *eventconsumer.Consumer, 61 61 - pipelineNotifier *StatusNotifier, 62 57 idResolver *idresolver.Resolver, 63 58 db *db.DB, 64 59 config *config.Config, ··· 66 61 logger *slog.Logger, 67 62 ) *Pipelines { 68 63 return &Pipelines{ 69 69 - oauth: oauth, 70 70 - repoResolver: repoResolver, 71 71 - pages: pages, 72 72 - idResolver: idResolver, 73 73 - config: config, 74 74 - spindlestream: spindlestream, 75 75 - pipelineNotifier: pipelineNotifier, 76 76 - db: db, 77 77 - enforcer: enforcer, 78 78 - logger: logger, 64 64 + oauth: oauth, 65 65 + repoResolver: repoResolver, 66 66 + pages: pages, 67 67 + idResolver: idResolver, 68 68 + config: config, 69 69 + db: db, 70 70 + enforcer: enforcer, 71 71 + logger: logger, 79 72 } 80 73 } 81 74 ··· 103 96 filterKind = "all" 104 97 } 105 98 106 106 - ps, err := db.GetPipelineStatuses( 107 107 - p.db, 108 108 - 30, 109 109 - filters..., 110 110 - ) 99 99 + // sh.tangled.ci.queryPipelines(repo, kind, limit=30) 100 100 + xrpcc := indigoxrpc.Client{Host: f.Spindle} 101 101 + out, err := tangled.CiQueryPipelines(r.Context(), &xrpcc, nil, "", 1, f.RepoDid) 111 102 if err != nil { 112 112 - l.Error("failed to query db", "err", err) 113 113 - return 114 114 - } 115 115 - 116 116 - total, err := db.GetPipelineCount(p.db, filters...) 117 117 - if err != nil { 118 118 - l.Error("failed to query db", "err", err) 119 119 - return 103 103 + l.Error("failed to fetch pipelines", "err", err) 104 104 + panic("unimplemented") // spindle failure, appview should not fail. 120 105 } 121 106 122 107 p.pages.Pipelines(w, pages.PipelinesParams{ 123 108 BaseParams: pages.BaseParamsFromContext(r.Context()), 124 109 RepoInfo: p.repoResolver.GetRepoInfo(r, user), 125 125 - Pipelines: ps, 110 110 + Pipelines: out.Pipelines, 126 111 FilterKind: filterKind, 127 127 - Total: total, 112 112 + Total: out.Total, 128 113 }) 129 114 } 130 115 ··· 135 120 f, err := p.repoResolver.Resolve(r) 136 121 if err != nil { 137 122 l.Error("failed to get repo and knot", "err", err) 123 123 + p.pages.Error404(w) 138 124 return 139 125 } 140 126 141 141 - pipelineId := chi.URLParam(r, "pipeline") 142 142 - if pipelineId == "" { 143 143 - l.Error("empty pipeline ID") 127 127 + pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) 128 128 + if err != nil { 129 129 + l.Debug("invalid pipeline id", "id", pipelineId) 130 130 + p.pages.Error404(w) 144 131 return 145 132 } 146 133 147 147 - workflow := chi.URLParam(r, "workflow") 148 148 - if workflow == "" { 149 149 - l.Error("empty workflow name") 134 134 + workflowName := chi.URLParam(r, "workflow") 135 135 + if workflowName == "" { 136 136 + l.Debug("empty workflow name") 137 137 + p.pages.Error404(w) 150 138 return 151 139 } 152 140 153 153 - ps, err := db.GetPipelineStatuses( 154 154 - p.db, 155 155 - 1, 156 156 - orm.FilterEq("p.repo_did", f.RepoDid), 157 157 - orm.FilterEq("p.id", pipelineId), 158 158 - ) 141 141 + l = l.With("pipeline", pipelineId, "workflow", workflowName) 142 142 + 143 143 + // TODO: change url path to: 144 144 + // /{owner}/{slug}/pipelines/{spindle-did}/{pipeline-id}/workflow/{workflow-id} 145 145 + 146 146 + xrpcc := &indigoxrpc.Client{Host: f.Spindle} 147 147 + out, err := tangled.CiGetPipeline(r.Context(), xrpcc, pipelineId.String()) 159 148 if err != nil { 160 160 - l.Error("failed to query db", "err", err) 149 149 + // TODO(boltless): change behavior based on error 150 150 + l.Debug("failed to get pipeline", "err", err) 151 151 + p.pages.Error404(w) 161 152 return 162 153 } 163 154 164 164 - if len(ps) != 1 { 165 165 - l.Error("invalid number of pipelines", "len", len(ps)) 155 155 + // ensure workflow exists 156 156 + exist := false 157 157 + for _, workflow := range out.Workflows { 158 158 + if workflow.Name == workflowName { 159 159 + exist = true 160 160 + break 161 161 + } 162 162 + } 163 163 + if !exist { 164 164 + l.Debug("workflow doesn't exist in pipeline") 165 165 + p.pages.Error404(w) 166 166 return 167 167 } 168 168 169 169 - singlePipeline := ps[0] 170 170 - 171 169 p.pages.Workflow(w, pages.WorkflowParams{ 172 170 BaseParams: pages.BaseParamsFromContext(r.Context()), 173 171 RepoInfo: p.repoResolver.GetRepoInfo(r, user), 174 174 - Pipeline: singlePipeline, 175 175 - Workflow: workflow, 172 172 + Pipeline: out, 173 173 + Workflow: workflowName, 176 174 }) 177 175 } 178 176 ··· 181 179 WriteBufferSize: 1024, 182 180 } 183 181 182 182 + type webLogScheduler struct { 183 183 + ch chan *tangled.CiPipelineSubscribeLogs_Event 184 184 + } 185 185 + 186 186 + var _ lexutil.Scheduler[tangled.CiPipelineSubscribeLogs_Event] = (*webLogScheduler)(nil) 187 187 + 188 188 + // AddWork implements [lexutil.Scheduler]. 189 189 + func (w *webLogScheduler) AddWork(ctx context.Context, _ string, val *tangled.CiPipelineSubscribeLogs_Event) error { 190 190 + select { 191 191 + case w.ch <- val: 192 192 + return nil 193 193 + case <-ctx.Done(): 194 194 + return ctx.Err() 195 195 + } 196 196 + } 197 197 + 198 198 + // Shutdown implements [lexutil.Scheduler]. 199 199 + func (w *webLogScheduler) Shutdown() { close(w.ch) } 200 200 + 184 201 func (p *Pipelines) Logs(w http.ResponseWriter, r *http.Request) { 185 202 l := p.logger.With("handler", "logs") 186 203 ··· 191 208 return 192 209 } 193 210 194 194 - pipelineId := chi.URLParam(r, "pipeline") 195 195 - workflow := chi.URLParam(r, "workflow") 196 196 - if pipelineId == "" || workflow == "" { 197 197 - http.Error(w, "missing pipeline ID or workflow", http.StatusBadRequest) 211 211 + if f.Spindle == "" { 212 212 + http.Error(w, "invalid repo info", http.StatusBadRequest) 198 213 return 199 214 } 200 215 201 201 - ps, err := db.GetPipelineStatuses( 202 202 - p.db, 203 203 - 1, 204 204 - orm.FilterEq("p.repo_did", f.RepoDid), 205 205 - orm.FilterEq("p.id", pipelineId), 206 206 - ) 207 207 - if err != nil || len(ps) != 1 { 208 208 - l.Error("pipeline query failed", "err", err, "count", len(ps)) 209 209 - http.Error(w, "pipeline not found", http.StatusNotFound) 216 216 + pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) 217 217 + if err != nil { 218 218 + l.Debug("invalid pipeline id", "id", pipelineId) 219 219 + http.Error(w, "invalid pipeline id", http.StatusBadRequest) 210 220 return 211 221 } 212 222 213 213 - singlePipeline := ps[0] 214 214 - spindle := f.Spindle 215 215 - knot := f.Knot 216 216 - rkey := singlePipeline.Rkey 217 217 - 218 218 - statusCh := p.pipelineNotifier.Subscribe(singlePipeline.AtUri()) 219 219 - defer p.pipelineNotifier.Unsubscribe(singlePipeline.AtUri(), statusCh) 220 220 - 221 221 - if spindle == "" || knot == "" || rkey == "" { 222 222 - http.Error(w, "invalid repo info", http.StatusBadRequest) 223 223 + workflowName := chi.URLParam(r, "workflow") 224 224 + if workflowName == "" { 225 225 + l.Debug("empty workflow name") 226 226 + http.Error(w, "invalid workflow name", http.StatusBadRequest) 223 227 return 224 228 } 225 229 226 226 - url := SpindleURL(spindle, knot, rkey, workflow) 227 227 - if url == "" { 228 228 - http.Error(w, "invalid spindle hostname", http.StatusBadRequest) 229 229 - return 230 230 - } 231 231 - l = l.With("url", url) 232 232 - 233 230 clientConn, err := upgrader.Upgrade(w, r, nil) 234 231 if err != nil { 235 232 l.Error("websocket upgrade failed", "err", err) 236 233 return 237 234 } 238 238 - defer func() { 239 239 - _ = clientConn.WriteControl( 240 240 - websocket.CloseMessage, 241 241 - websocket.FormatCloseMessage(websocket.CloseNormalClosure, "log stream complete"), 242 242 - time.Now().Add(time.Second), 243 243 - ) 244 244 - clientConn.Close() 245 245 - }() 235 235 + defer clientConn.Close() 246 236 247 237 ctx, cancel := context.WithCancel(r.Context()) 248 238 defer cancel() 249 239 250 250 - l.Info("logs endpoint hit") 240 240 + evChan := make(chan *tangled.CiPipelineSubscribeLogs_Event, 100) 241 241 + done := make(chan error, 1) 242 242 + sched := &webLogScheduler{ch: evChan} 243 243 + xrpcc := &lexutil.Client{Client: indigoxrpc.Client{Host: f.Spindle}} 244 244 + go func() { 245 245 + done <- tangled.CiPipelineSubscribeLogs(ctx, xrpcc, pipelineId.String(), []string{workflowName}, sched) 246 246 + }() 247 247 + 248 248 + var lastWriteLk sync.Mutex 249 249 + lastWrite := time.Now() 250 250 + 251 251 + // Start a goroutine to ping the client periodically to check if it's still 252 252 + // alive. If the client doesn't respond to a ping within 5 seconds, we'll 253 253 + // close the connection and teardown the consumer. 254 254 + go func() { 255 255 + ticker := time.NewTicker(30 * time.Second) 256 256 + defer ticker.Stop() 257 257 + for { 258 258 + select { 259 259 + case <-ticker.C: 260 260 + lastWriteLk.Lock() 261 261 + lw := lastWrite 262 262 + lastWriteLk.Unlock() 263 263 + if time.Since(lw) < 30*time.Second { 264 264 + continue 265 265 + } 266 266 + if err := clientConn.WriteControl(websocket.PingMessage, nil, time.Now().Add(5*time.Second)); err != nil { 267 267 + l.Warn("failed to ping client", "err", err) 268 268 + cancel() 269 269 + return 270 270 + } 271 271 + case <-ctx.Done(): 272 272 + return 273 273 + } 274 274 + } 275 275 + }() 251 276 252 252 - spindleConn, _, err := websocket.DefaultDialer.Dial(url, nil) 253 253 - if err != nil { 254 254 - l.Error("websocket dial failed", "err", err) 255 255 - return 256 256 - } 257 257 - defer spindleConn.Close() 277 277 + clientConn.SetPingHandler(func(message string) error { 278 278 + err := clientConn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(60*time.Second)) 279 279 + if err == websocket.ErrCloseSent { 280 280 + return nil 281 281 + } 282 282 + return err 283 283 + }) 258 284 259 259 - // create a channel for incoming messages 260 260 - evChan := make(chan LogEvent, 100) 261 261 - // start a goroutine to read from spindle 262 262 - go ReadLogs(spindleConn, evChan) 285 285 + // Start a goroutine to read messages from the client and discard them. 286 286 + go func() { 287 287 + for { 288 288 + if _, _, err := clientConn.ReadMessage(); err != nil { 289 289 + cancel() 290 290 + return 291 291 + } 292 292 + } 293 293 + }() 263 294 295 295 + // Main loop: sole writer of data frames to the client. 264 296 stepStartTimes := make(map[int]time.Time) 265 297 stepAnsi := make(map[int]*ansiState) 266 298 var fragment bytes.Buffer ··· 272 304 273 305 case ev, ok := <-evChan: 274 306 if !ok { 275 275 - continue 276 276 - } 277 277 - 278 278 - if ev.Err != nil && ev.IsCloseError() { 279 279 - l.Debug("graceful shutdown, tail complete", "err", err) 280 280 - return 281 281 - } 282 282 - if ev.Err != nil { 283 283 - l.Error("error reading from spindle", "err", err) 307 307 + // Stream ended: Shutdown closed the upstream channel. 308 308 + if err := <-done; !isExpectedClose(err) { 309 309 + l.Error("spindle stream error", "err", err) 310 310 + } 284 311 return 285 312 } 286 313 287 287 - var logLine spindlemodel.LogLine 288 288 - if err = json.Unmarshal(ev.Msg, &logLine); err != nil { 289 289 - l.Error("failed to parse logline", "err", err) 290 290 - continue 291 291 - } 314 314 + fragment.Reset() 292 315 293 293 - fragment.Reset() 316 316 + switch { 317 317 + case ev.Error != nil: 318 318 + l.Error("spindle error frame", "err", ev.Error.Error, "msg", ev.Error.Message) 319 319 + return 294 320 295 295 - switch logLine.Kind { 296 296 - case spindlemodel.LogKindControl: 297 297 - switch logLine.StepStatus { 298 298 - case spindlemodel.StepStatusStart: 299 299 - stepStartTimes[logLine.StepId] = logLine.Time 300 300 - collapsed := false 301 301 - if logLine.StepKind == spindlemodel.StepKindSystem { 302 302 - collapsed = true 303 303 - } 321 321 + case ev.Control != nil: 322 322 + c := ev.Control 323 323 + step := int(c.Step) 324 324 + switch derefStr(c.Status) { 325 325 + case "start": 326 326 + t := parseRFC3339(c.Time) 327 327 + stepStartTimes[step] = t 328 328 + // "system" steps are injected by the CI runner; collapse them. 329 329 + collapsed := derefStr(c.Kind) == "system" 304 330 err = p.pages.LogBlock(&fragment, pages.LogBlockParams{ 305 305 - Id: logLine.StepId, 306 306 - Name: logLine.Content, 307 307 - Command: logLine.StepCommand, 331 331 + Id: step, 332 332 + Name: c.Content, 333 333 + Command: derefStr(c.Command), 308 334 Collapsed: collapsed, 309 309 - StartTime: logLine.Time, 335 335 + StartTime: t, 310 336 }) 311 311 - case spindlemodel.StepStatusEnd: 312 312 - startTime := stepStartTimes[logLine.StepId] 313 313 - endTime := logLine.Time 337 337 + case "end": 314 338 err = p.pages.LogBlockEnd(&fragment, pages.LogBlockEndParams{ 315 315 - Id: logLine.StepId, 316 316 - StartTime: startTime, 317 317 - EndTime: endTime, 339 339 + Id: step, 340 340 + StartTime: stepStartTimes[step], 341 341 + EndTime: parseRFC3339(c.Time), 318 342 }) 319 343 } 320 344 321 321 - case spindlemodel.LogKindData: 322 322 - ansi, ok := stepAnsi[logLine.StepId] 345 345 + case ev.Data != nil: 346 346 + d := ev.Data 347 347 + step := int(d.Step) 348 348 + ansi, ok := stepAnsi[step] 323 349 if !ok { 324 350 ansi = NewAnsiState() 325 325 - stepAnsi[logLine.StepId] = ansi 351 351 + stepAnsi[step] = ansi 326 352 } 327 353 err = p.pages.LogLine(&fragment, pages.LogLineParams{ 328 328 - Id: logLine.StepId, 329 329 - Content: ansi.Render(logLine.Content), 354 354 + Id: step, 355 355 + Content: ansi.Render(d.Content), 330 356 }) 331 357 } 332 358 if err != nil { ··· 338 364 l.Error("error writing to client", "err", err) 339 365 return 340 366 } 341 341 - 342 342 - case _, ok := <-statusCh: 343 343 - if !ok { 344 344 - continue 345 345 - } 346 346 - fresh, err := db.GetPipelineStatuses( 347 347 - p.db, 348 348 - 1, 349 349 - orm.FilterEq("p.repo_did", f.RepoDid), 350 350 - orm.FilterEq("p.id", pipelineId), 351 351 - ) 352 352 - if err != nil || len(fresh) == 0 { 353 353 - continue 354 354 - } 355 355 - for name, ws := range fresh[0].Statuses { 356 356 - fragment.Reset() 357 357 - if err = p.pages.WorkflowSymbolOOB(&fragment, pages.WorkflowSymbolOOBParams{ 358 358 - Name: name, 359 359 - Statuses: ws, 360 360 - }); err != nil { 361 361 - l.Error("failed to render workflow symbol OOB", "err", err) 362 362 - continue 363 363 - } 364 364 - if err = clientConn.WriteMessage(websocket.TextMessage, fragment.Bytes()); err != nil { 365 365 - l.Error("error writing workflow symbol to client", "err", err) 366 366 - return 367 367 - } 368 368 - } 369 369 - 370 370 - case <-time.After(30 * time.Second): 371 371 - l.Debug("sent keepalive") 372 372 - if err = clientConn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil { 373 373 - l.Error("failed to write control", "err", err) 374 374 - return 375 375 - } 367 367 + lastWriteLk.Lock() 368 368 + lastWrite = time.Now() 369 369 + lastWriteLk.Unlock() 376 370 } 377 371 } 378 372 } 379 373 380 380 - func (p *Pipelines) Cancel(w http.ResponseWriter, r *http.Request) { 381 381 - l := p.logger.With("handler", "Cancel") 382 382 - 383 383 - var ( 384 384 - pipelineId = chi.URLParam(r, "pipeline") 385 385 - workflow = chi.URLParam(r, "workflow") 386 386 - ) 387 387 - if pipelineId == "" || workflow == "" { 388 388 - http.Error(w, "missing pipeline ID or workflow", http.StatusBadRequest) 389 389 - return 390 390 - } 374 374 + func (p *Pipelines) CancelWorkflow(w http.ResponseWriter, r *http.Request) { 375 375 + l := p.logger.With("handler", "CancelWorkflow") 376 376 + errorId := "workflow-error" 391 377 392 378 f, err := p.repoResolver.Resolve(r) 393 379 if err != nil { 394 380 l.Error("failed to get repo and knot", "err", err) 395 395 - http.Error(w, "bad repo/knot", http.StatusBadRequest) 381 381 + p.pages.Notice(w, errorId, "Failed to cancel workflow") 396 382 return 397 383 } 384 384 + l = l.With("repo", f.RepoDid) 398 385 399 399 - pipeline, err := func() (models.Pipeline, error) { 400 400 - ps, err := db.GetPipelineStatuses( 401 401 - p.db, 402 402 - 1, 403 403 - orm.FilterEq("p.repo_did", f.RepoDid), 404 404 - orm.FilterEq("p.id", pipelineId), 405 405 - ) 406 406 - if err != nil { 407 407 - return models.Pipeline{}, err 408 408 - } 409 409 - if len(ps) != 1 { 410 410 - return models.Pipeline{}, fmt.Errorf("wrong pipeline count %d", len(ps)) 411 411 - } 412 412 - return ps[0], nil 413 413 - }() 386 386 + if f.Spindle == "" { 387 387 + l.Debug("spindle is empty") 388 388 + p.pages.Notice(w, errorId, "Failed to cancel workflow") 389 389 + return 390 390 + } 391 391 + 392 392 + pipelineId, err := syntax.ParseTID(chi.URLParam(r, "pipeline")) 414 393 if err != nil { 415 415 - l.Error("pipeline query failed", "err", err) 416 416 - http.Error(w, "pipeline not found", http.StatusNotFound) 394 394 + l.Debug("invalid pipeline id", "id", pipelineId) 395 395 + p.pages.Error404(w) 396 396 + return 417 397 } 418 418 - var ( 419 419 - spindle = f.Spindle 420 420 - knot = f.Knot 421 421 - rkey = pipeline.Rkey 422 422 - ) 423 398 424 424 - if spindle == "" || knot == "" || rkey == "" { 425 425 - http.Error(w, "invalid repo info", http.StatusBadRequest) 399 399 + workflowName := chi.URLParam(r, "workflow") 400 400 + if workflowName == "" { 401 401 + l.Debug("empty workflow name") 402 402 + p.pages.Error404(w) 426 403 return 427 404 } 428 405 429 429 - hostname, noTLS, err := hostutil.ParseHostname(spindle) 406 406 + l = l.With("pipeline", pipelineId, "workflow", workflowName) 407 407 + 408 408 + hostname, noTLS, err := hostutil.ParseHostname(f.Spindle) 430 409 if err != nil { 431 410 http.Error(w, "invalid spindle hostname", http.StatusBadRequest) 432 411 return ··· 440 419 oauth.WithTimeout(time.Second*30), // workflow cleanup usually takes time 441 420 ) 442 421 443 443 - err = tangled.PipelineCancelPipeline( 422 422 + if err := tangled.PipelineCancelPipeline( 444 423 r.Context(), 445 424 spindleClient, 446 425 &tangled.PipelineCancelPipeline_Input{ 447 426 Repo: string(f.RepoAt()), 448 448 - Pipeline: pipeline.AtUri().String(), 449 449 - Workflow: workflow, 427 427 + Pipeline: pipelineId.String(), 428 428 + Workflow: workflowName, 450 429 }, 451 451 - ) 452 452 - errorId := "workflow-error" 453 453 - if err != nil { 430 430 + ); err != nil { 454 431 l.Error("failed to cancel workflow", "err", err) 455 432 p.pages.Notice(w, errorId, "Failed to cancel workflow") 456 433 return 457 434 } 458 458 - l.Debug("canceled pipeline", "uri", pipeline.AtUri()) 435 435 + l.Debug("canceled workflow") 459 436 }

+43

appview/pipelines/ssh/cihelpers.go

Reviewed

··· 1 1 + package ssh 2 2 + 3 3 + import ( 4 4 + "time" 5 5 + 6 6 + "tangled.org/core/api/tangled" 7 7 + ) 8 8 + 9 9 + // helper functions against generated code 10 10 + 11 11 + func workflowElapsed(wf *tangled.CiDefs_Workflow, now time.Time) time.Duration { 12 12 + if wf.StartedAt == nil { 13 13 + return 0 14 14 + } 15 15 + started, err := time.Parse(time.RFC3339, *wf.StartedAt) 16 16 + if err != nil { 17 17 + return 0 18 18 + } 19 19 + if wf.FinishedAt == nil { 20 20 + return now.Sub(started) 21 21 + } 22 22 + finished, err := time.Parse(time.RFC3339, *wf.FinishedAt) 23 23 + if err != nil { 24 24 + return 0 25 25 + } 26 26 + return finished.Sub(started) 27 27 + } 28 28 + 29 29 + var finishedStatuses = map[string]bool{ 30 30 + "failed": true, 31 31 + "timeout": true, 32 32 + "cancelled": true, 33 33 + "success": true, 34 34 + } 35 35 + 36 36 + func pipelineFinished(p *tangled.CiDefs_Pipeline) bool { 37 37 + for _, wf := range p.Workflows { 38 38 + if !finishedStatuses[wf.Status] { 39 39 + return false 40 40 + } 41 41 + } 42 42 + return true 43 43 + }

+21 -21

appview/pipelines/ssh/logstream.go

Reviewed

··· 1 1 package ssh 2 2 3 3 import ( 4 4 + "context" 4 5 "time" 5 6 6 6 - tea "github.com/charmbracelet/bubbletea" 7 7 - "github.com/gorilla/websocket" 8 8 - "tangled.org/core/appview/pipelines" 9 9 - spindlemodel "tangled.org/core/spindle/models" 7 7 + "tangled.org/core/api/tangled" 10 8 ) 11 9 12 10 type step struct { 13 13 - id int 11 11 + id int64 14 12 name string 15 13 command string 16 16 - kind spindlemodel.StepKind 17 14 lines []string 18 15 startTime time.Time 19 16 endTime time.Time ··· 21 18 } 22 19 23 20 type logDoneMsg struct { 24 24 - workflow string 25 25 - err error 21 21 + err error 26 22 } 27 23 28 24 type logEventMsg struct { 29 29 - workflow string 30 30 - ev pipelines.LogEvent 31 31 - conn *websocket.Conn 32 32 - ch chan pipelines.LogEvent 25 25 + ev *tangled.CiPipelineSubscribeLogs_Event 26 26 + events chan *tangled.CiPipelineSubscribeLogs_Event 27 27 + done chan error 33 28 } 34 29 35 35 - func readNextCmd(workflow string, conn *websocket.Conn, ch chan pipelines.LogEvent) tea.Cmd { 36 36 - return func() tea.Msg { 37 37 - return readNextLogEvent(workflow, conn, ch) 38 38 - } 30 30 + type eventScheduler struct { 31 31 + ch chan *tangled.CiPipelineSubscribeLogs_Event 32 32 + } 33 33 + 34 34 + func newEventScheduler() *eventScheduler { 35 35 + return &eventScheduler{ch: make(chan *tangled.CiPipelineSubscribeLogs_Event, 1024)} 39 36 } 40 37 41 41 - func readNextLogEvent(workflow string, conn *websocket.Conn, ch chan pipelines.LogEvent) tea.Msg { 42 42 - ev, ok := <-ch 43 43 - if !ok { 44 44 - return logDoneMsg{workflow: workflow} 38 38 + func (s *eventScheduler) AddWork(ctx context.Context, _ string, v *tangled.CiPipelineSubscribeLogs_Event) error { 39 39 + select { 40 40 + case s.ch <- v: 41 41 + return nil 42 42 + case <-ctx.Done(): 43 43 + return ctx.Err() 45 44 } 46 46 - return logEventMsg{workflow: workflow, ev: ev, conn: conn, ch: ch} 47 45 } 46 46 + 47 47 + func (s *eventScheduler) Shutdown() { close(s.ch) }

+5 -7

appview/pipelines/ssh/server.go

Reviewed

··· 10 10 tea "github.com/charmbracelet/wish/bubbletea" 11 11 "tangled.org/core/appview/config" 12 12 "tangled.org/core/appview/db" 13 13 - "tangled.org/core/appview/pipelines" 14 13 ) 15 14 16 15 type Server struct { 17 17 - db *db.DB 18 18 - config *config.Config 19 19 - pipelineNotifier *pipelines.StatusNotifier 20 20 - logger *slog.Logger 16 16 + db *db.DB 17 17 + config *config.Config 18 18 + logger *slog.Logger 21 19 } 22 20 23 23 - func New(db *db.DB, cfg *config.Config, pn *pipelines.StatusNotifier, logger *slog.Logger) *Server { 24 24 - return &Server{db: db, config: cfg, pipelineNotifier: pn, logger: logger} 21 21 + func New(db *db.DB, cfg *config.Config, logger *slog.Logger) *Server { 22 22 + return &Server{db: db, config: cfg, logger: logger} 25 23 } 26 24 27 25 func (s *Server) ListenAndServe(ctx context.Context) error {

+34 -8

appview/pipelines/ssh/session.go

Reviewed

··· 3 3 import ( 4 4 "fmt" 5 5 6 6 + "github.com/bluesky-social/indigo/atproto/syntax" 7 7 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 6 8 tea "github.com/charmbracelet/bubbletea" 7 9 "github.com/charmbracelet/ssh" 8 10 wishtea "github.com/charmbracelet/wish/bubbletea" 11 11 + "tangled.org/core/api/tangled" 9 12 "tangled.org/core/appview/db" 13 13 + "tangled.org/core/hostutil" 14 14 + extlexutil "tangled.org/core/lexutil" 10 15 "tangled.org/core/orm" 11 16 ) 12 17 ··· 29 34 30 35 l = l.With("repoDID", repoDID, "sha", sha) 31 36 32 32 - pipelines, err := db.GetPipelineStatuses(s.db, 1, 33 33 - orm.FilterEq("p.repo_did", repoDID), 34 34 - orm.FilterEq("p.sha", sha), 35 35 - ) 36 36 - if err != nil || len(pipelines) == 0 { 37 37 + repo, err := db.GetRepo(s.db, orm.FilterEq("repo_did", repoDID)) 38 38 + if err != nil { 39 39 + l.Warn("repo not found", "err", err) 40 40 + return newErrorModel(renderer, fmt.Sprintf("repo %s not found", repoDID)), wishtea.MakeOptions(sess) 41 41 + } 42 42 + if repo.Spindle == "" { 43 43 + l.Warn("no spindle configured") 44 44 + return newErrorModel(renderer, "no spindle configured for this repo"), wishtea.MakeOptions(sess) 45 45 + } 46 46 + 47 47 + l = l.With("spindle", repo.Spindle) 48 48 + 49 49 + host, err := hostutil.EnsureHttpScheme(repo.Spindle) 50 50 + if err != nil { 51 51 + l.Warn("invalid spindlie hostname", "err", err) 52 52 + return newErrorModel(renderer, fmt.Sprintf("invalid spindle host %q", repo.Spindle)), wishtea.MakeOptions(sess) 53 53 + } 54 54 + 55 55 + xrpcc := extlexutil.Client{Client: indigoxrpc.Client{Host: host}} 56 56 + out, err := tangled.CiQueryPipelines(sess.Context(), &xrpcc, []string{sha}, "", 1, repoDID) 57 57 + if err != nil || len(out.Pipelines) == 0 { 37 58 l.Warn("pipeline not found", "err", err) 38 59 return newErrorModel(renderer, fmt.Sprintf("pipeline not found for repo %s @ %s", repoDID, sha)), wishtea.MakeOptions(sess) 39 60 } 40 61 41 41 - pipeline := pipelines[0] 42 42 - l.Info("serving pipeline", "workflows", len(pipeline.Statuses)) 62 62 + pipeline := out.Pipelines[0] 63 63 + if _, err := syntax.ParseTID(pipeline.Id); err != nil { 64 64 + l.Warn("invalid pipeline id", "id", pipeline.Id, "err", err) 65 65 + return newErrorModel(renderer, fmt.Sprintf("invalid pipeline id %q", pipeline.Id)), wishtea.MakeOptions(sess) 66 66 + } 67 67 + 68 68 + l.Info("serving pipeline", "pipeline", pipeline.Id, "workflows", len(pipeline.Workflows)) 43 69 pty, _, _ := sess.Pty() 44 70 opts := append(wishtea.MakeOptions(sess), tea.WithAltScreen()) 45 45 - return newPipelineModel(renderer, s, pipeline, pty.Window.Width, pty.Window.Height), opts 71 71 + return newPipelineModel(renderer, &xrpcc, pipeline, pty.Window.Width, pty.Window.Height), opts 46 72 }

+214 -162

appview/pipelines/ssh/tui.go

Reviewed

··· 1 1 package ssh 2 2 3 3 import ( 4 4 - "encoding/json" 4 4 + "context" 5 5 + "errors" 5 6 "fmt" 6 7 "strings" 7 8 "time" ··· 11 12 tea "github.com/charmbracelet/bubbletea" 12 13 "github.com/charmbracelet/lipgloss" 13 14 "github.com/gorilla/websocket" 14 14 - "tangled.org/core/appview/db" 15 15 - "tangled.org/core/appview/models" 16 16 - "tangled.org/core/appview/pipelines" 17 17 - "tangled.org/core/orm" 18 18 - spindlemodel "tangled.org/core/spindle/models" 15 15 + "tangled.org/core/api/tangled" 16 16 + extlexutil "tangled.org/core/lexutil" 19 17 ) 20 18 21 19 var ( ··· 27 25 type tickMsg time.Time 28 26 29 27 type statusUpdateMsg struct { 30 30 - pipeline models.Pipeline 28 28 + pipeline *tangled.CiDefs_Pipeline 31 29 } 32 30 33 31 type statusUpdateErrMsg struct{ err error } 34 32 35 33 type pipelineModel struct { 36 36 - renderer *lipgloss.Renderer 37 37 - server *Server 38 38 - pipeline models.Pipeline 39 39 - workflows []string 40 40 - selected int 41 41 - logs map[string]*workflowLogs 42 42 - statusCh chan struct{} 43 43 - spinner spinner.Model 44 44 - width int 45 45 - height int 34 34 + renderer *lipgloss.Renderer 35 35 + xrpcc *extlexutil.Client 36 36 + pipeline *tangled.CiDefs_Pipeline 37 37 + selected int 38 38 + logs map[string]*workflowLogs 39 39 + 40 40 + // pipeline log stream: cancel tears down the consumer goroutine on quit. 41 41 + // the event/done channels are threaded through log messages, not stored here. 42 42 + cancel context.CancelFunc 43 43 + streamDone bool 44 44 + streamErr error 45 45 + 46 46 + spinner spinner.Model 47 47 + width int 48 48 + height int 46 49 } 47 50 48 51 type workflowLogs struct { 49 52 steps []step 50 50 - stepIndex map[int]int 53 53 + stepIndex map[int64]int // stepId -> index map 51 54 vp viewport.Model 52 55 ready bool 53 53 - done bool 54 54 - err error 55 56 } 56 57 57 57 - func newPipelineModel(renderer *lipgloss.Renderer, s *Server, pipeline models.Pipeline, width, height int) *pipelineModel { 58 58 - workflows := pipeline.Workflows() 59 59 - logs := make(map[string]*workflowLogs, len(workflows)) 60 60 - for _, wf := range workflows { 61 61 - logs[wf] = &workflowLogs{stepIndex: make(map[int]int)} 58 58 + func newPipelineModel(renderer *lipgloss.Renderer, xrpcc *extlexutil.Client, pipeline *tangled.CiDefs_Pipeline, width, height int) *pipelineModel { 59 59 + logs := make(map[string]*workflowLogs, len(pipeline.Workflows)) 60 60 + for _, wf := range pipeline.Workflows { 61 61 + logs[wf.Name] = &workflowLogs{stepIndex: make(map[int64]int)} 62 62 } 63 63 - statusCh := s.pipelineNotifier.Subscribe(pipeline.AtUri()) 64 63 sp := spinner.New(spinner.WithSpinner(spinner.Line)) 65 64 return &pipelineModel{ 66 66 - renderer: renderer, 67 67 - server: s, 68 68 - pipeline: pipeline, 69 69 - workflows: workflows, 70 70 - logs: logs, 71 71 - statusCh: statusCh, 72 72 - spinner: sp, 73 73 - width: width, 74 74 - height: height, 65 65 + renderer: renderer, 66 66 + xrpcc: xrpcc, 67 67 + pipeline: pipeline, 68 68 + logs: logs, 69 69 + spinner: sp, 70 70 + width: width, 71 71 + height: height, 75 72 } 76 73 } 77 74 78 75 func (m *pipelineModel) Init() tea.Cmd { 79 79 - cmds := []tea.Cmd{tick(), m.spinner.Tick, m.waitForStatusUpdate(m.statusCh)} 80 80 - for _, wf := range m.workflows { 81 81 - cmds = append(cmds, m.connectCmd(wf)) 82 82 - } 83 83 - return tea.Batch(cmds...) 76 76 + return tea.Batch(tick(), m.spinner.Tick, m.subscribeCmd()) 84 77 } 85 78 86 79 func tick() tea.Cmd { 87 80 return tea.Tick(time.Second, func(t time.Time) tea.Msg { return tickMsg(t) }) 88 81 } 89 82 90 90 - // waitForStatusUpdate blocks on the notifier channel, re-fetches pipeline statuses, and returns the result as a tea.Msg. 91 91 - func (m *pipelineModel) waitForStatusUpdate(ch chan struct{}) tea.Cmd { 92 92 - knot := m.pipeline.Knot 93 93 - rkey := m.pipeline.Rkey 83 83 + // subscribeCmd opens the ci.pipeline.subscribeLogs stream for current pipeline. 84 84 + // A consumer goroutine pushes decoded events onto the scheduler channel; the 85 85 + // returned command yields the first event into the bubbletea loop. 86 86 + func (m *pipelineModel) subscribeCmd() tea.Cmd { 87 87 + // cancel existing subscriptions just in case 88 88 + if m.cancel != nil { 89 89 + m.cancel() 90 90 + } 91 91 + sched := newEventScheduler() 92 92 + done := make(chan error, 1) 93 93 + ctx, cancel := context.WithCancel(context.Background()) 94 94 + m.cancel = cancel 95 95 + 96 96 + pipelineId := m.pipeline.Id 97 97 + go func() { 98 98 + err := tangled.CiPipelineSubscribeLogs(ctx, m.xrpcc, pipelineId, nil, sched) 99 99 + done <- err 100 100 + }() 101 101 + 102 102 + return readEventCmd(sched.ch, done) 103 103 + } 104 104 + 105 105 + func readEventCmd(events chan *tangled.CiPipelineSubscribeLogs_Event, done chan error) tea.Cmd { 94 106 return func() tea.Msg { 95 95 - if _, ok := <-ch; !ok { 96 96 - return nil 97 97 - } 98 98 - ps, err := db.GetPipelineStatuses(m.server.db, 1, 99 99 - orm.FilterEq("p.knot", knot), 100 100 - orm.FilterEq("p.rkey", rkey), 101 101 - ) 102 102 - if err != nil || len(ps) == 0 { 103 103 - return statusUpdateErrMsg{err: fmt.Errorf("refreshing pipeline: %w", err)} 107 107 + ev, ok := <-events 108 108 + if !ok { 109 109 + return logDoneMsg{err: <-done} 104 110 } 105 105 - return statusUpdateMsg{pipeline: ps[0]} 111 111 + return logEventMsg{ev: ev, events: events, done: done} 106 112 } 107 113 } 108 114 109 109 - // connectCmd dials the spindle websocket for the given workflow and starts streaming log events. 110 110 - func (m *pipelineModel) connectCmd(workflow string) tea.Cmd { 115 115 + // fetchStatusCmd re-fetches the pipeline 116 116 + func (m *pipelineModel) fetchStatusCmd() tea.Cmd { 117 117 + pipelineId := m.pipeline.Id 111 118 return func() tea.Msg { 112 112 - ws, ok := m.pipeline.Statuses[workflow] 113 113 - if !ok || len(ws.Data) == 0 { 114 114 - return logDoneMsg{workflow: workflow} 115 115 - } 116 116 - url := pipelines.SpindleURL(ws.Data[0].Spindle, m.pipeline.Knot, m.pipeline.Rkey, workflow) 117 117 - conn, _, err := websocket.DefaultDialer.Dial(url, nil) 119 119 + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) 120 120 + defer cancel() 121 121 + out, err := tangled.CiGetPipeline(ctx, m.xrpcc, pipelineId) 118 122 if err != nil { 119 119 - return logDoneMsg{workflow: workflow, err: fmt.Errorf("connecting to spindle: %w", err)} 123 123 + return statusUpdateErrMsg{err: fmt.Errorf("refreshing pipeline: %w", err)} 120 124 } 121 121 - ch := make(chan pipelines.LogEvent, 100) 122 122 - go pipelines.ReadLogs(conn, ch) 123 123 - return readNextLogEvent(workflow, conn, ch) 125 125 + return statusUpdateMsg{pipeline: out} 124 126 } 125 127 } 126 128 ··· 153 155 m.resizeViewports() 154 156 155 157 case tickMsg: 158 158 + // re-render running workflows so elapsed times advance 159 159 + m.refreshRunning() 156 160 return m, tick() 157 161 158 162 case spinner.TickMsg: ··· 163 167 case tea.KeyMsg: 164 168 switch msg.String() { 165 169 case "q", "ctrl+c": 166 166 - m.server.pipelineNotifier.Unsubscribe(m.pipeline.AtUri(), m.statusCh) 170 170 + if m.cancel != nil { 171 171 + m.cancel() 172 172 + } 167 173 return m, tea.Quit 168 174 case "tab", "right", "l": 169 169 - m.selected = (m.selected + 1) % len(m.workflows) 175 175 + m.selected = (m.selected + 1) % len(m.pipeline.Workflows) 170 176 return m, nil 171 177 case "shift+tab", "left", "h": 172 172 - m.selected = (m.selected - 1 + len(m.workflows)) % len(m.workflows) 178 178 + m.selected = (m.selected - 1 + len(m.pipeline.Workflows)) % len(m.pipeline.Workflows) 173 179 return m, nil 174 180 } 175 181 if wl := m.selectedLogs(); wl != nil && wl.ready { ··· 193 199 } 194 200 195 201 case logEventMsg: 196 196 - return m, m.handleLogEvent(msg) 202 202 + m.applyEvent(msg.ev) 203 203 + return m, readEventCmd(msg.events, msg.done) 197 204 198 205 case logDoneMsg: 199 199 - if wl, ok := m.logs[msg.workflow]; ok { 200 200 - wl.done, wl.err = true, msg.err 201 201 - m.initViewport(wl) 202 202 - wl.vp.SetContent(renderLogs(m.renderer, wl, m.width)) 203 203 - wl.vp.GotoBottom() 206 206 + m.streamDone = true 207 207 + if !isExpectedClose(msg.err) { 208 208 + m.streamErr = msg.err 204 209 } 210 210 + m.refreshAll() 211 211 + // resolve final workflow statuses once now that the stream has ended 212 212 + return m, m.fetchStatusCmd() 205 213 206 214 case statusUpdateMsg: 207 207 - // detect any workflows that are new since the last update 208 208 - known := make(map[string]bool, len(m.workflows)) 209 209 - for _, wf := range m.workflows { 210 210 - known[wf] = true 215 215 + m.pipeline = msg.pipeline 216 216 + known := make(map[string]bool, len(m.pipeline.Workflows)) 217 217 + for _, wf := range m.pipeline.Workflows { 218 218 + known[wf.Name] = true 211 219 } 212 212 - m.pipeline = msg.pipeline 213 213 - var newCmds []tea.Cmd 214 214 - for _, wf := range msg.pipeline.Workflows() { 215 215 - if !known[wf] { 216 216 - m.workflows = append(m.workflows, wf) 217 217 - m.logs[wf] = &workflowLogs{stepIndex: make(map[int]int)} 218 218 - newCmds = append(newCmds, m.connectCmd(wf)) 220 220 + for name := range m.logs { 221 221 + if !known[name] { 222 222 + delete(m.logs, name) 219 223 } 220 224 } 221 221 - // re-subscribe for the next update 222 222 - newCmds = append(newCmds, m.waitForStatusUpdate(m.statusCh)) 223 223 - return m, tea.Batch(newCmds...) 225 225 + if m.selected >= len(m.pipeline.Workflows) { 226 226 + m.selected = max(len(m.pipeline.Workflows)-1, 0) 227 227 + } 228 228 + m.refreshAll() 224 229 225 230 case statusUpdateErrMsg: 226 226 - // re-subscribe even on error so we don't stop listening 227 227 - return m, m.waitForStatusUpdate(m.statusCh) 231 231 + // best-effort final status refresh; ignore failures 228 232 } 229 233 230 234 return m, nil 231 235 } 232 236 233 237 func (m *pipelineModel) selectedLogs() *workflowLogs { 234 234 - if len(m.workflows) == 0 { 238 238 + if len(m.pipeline.Workflows) < 1+m.selected { 235 239 return nil 236 240 } 237 237 - return m.logs[m.workflows[m.selected]] 241 241 + return m.logs[m.pipeline.Workflows[m.selected].Name] 238 242 } 239 243 240 244 func (m *pipelineModel) initViewport(wl *workflowLogs) { ··· 245 249 wl.ready = true 246 250 } 247 251 248 248 - // handleLogEvent processes a single log event, updates the step state, and re-renders the viewport. 249 249 - func (m *pipelineModel) handleLogEvent(msg logEventMsg) tea.Cmd { 250 250 - wl, ok := m.logs[msg.workflow] 252 252 + // ensureWorkflow returns the log state for a workflow, lazily creating its routing entry. 253 253 + func (m *pipelineModel) ensureWorkflow(name string) *workflowLogs { 254 254 + wl, ok := m.logs[name] 251 255 if !ok { 252 252 - return nil 256 256 + wl = &workflowLogs{stepIndex: make(map[int64]int)} 257 257 + m.logs[name] = wl 253 258 } 254 254 - if msg.ev.Err != nil { 255 255 - wl.done = true 256 256 - if !msg.ev.IsCloseError() { 257 257 - wl.err = msg.ev.Err 258 258 - } 259 259 - m.initViewport(wl) 260 260 - wl.vp.SetContent(renderLogs(m.renderer, wl, m.width)) 261 261 - return nil 262 262 - } 263 263 - var line spindlemodel.LogLine 264 264 - if err := json.Unmarshal(msg.ev.Msg, &line); err != nil { 265 265 - return readNextCmd(msg.workflow, msg.conn, msg.ch) 266 266 - } 267 267 - applyLogLine(wl, line) 259 259 + return wl 260 260 + } 261 261 + 262 262 + // renderWorkflow re-renders a workflow's viewport, preserving bottom-stickiness. 263 263 + func (m *pipelineModel) renderWorkflow(wl *workflowLogs) { 268 264 m.initViewport(wl) 269 265 atBottom := wl.vp.AtBottom() 270 266 wl.vp.SetContent(renderLogs(m.renderer, wl, m.width)) 271 267 if atBottom { 272 268 wl.vp.GotoBottom() 273 269 } 274 274 - return readNextCmd(msg.workflow, msg.conn, msg.ch) 270 270 + } 271 271 + 272 272 + // refreshAll re-renders every initialized viewport. 273 273 + func (m *pipelineModel) refreshAll() { 274 274 + for _, wl := range m.logs { 275 275 + m.renderWorkflow(wl) 276 276 + } 275 277 } 276 278 277 277 - // applyLogLine mutates wl by appending the log line to the appropriate step. 278 278 - func applyLogLine(wl *workflowLogs, line spindlemodel.LogLine) { 279 279 - switch line.Kind { 280 280 - case spindlemodel.LogKindControl: 281 281 - switch line.StepStatus { 282 282 - case spindlemodel.StepStatusStart: 283 283 - idx := len(wl.steps) 284 284 - wl.stepIndex[line.StepId] = idx 279 279 + // refreshRunning re-renders workflows with unfinished steps so elapsed times advance. 280 280 + func (m *pipelineModel) refreshRunning() { 281 281 + if m.streamDone { 282 282 + return 283 283 + } 284 284 + for _, wl := range m.logs { 285 285 + if !wl.ready { 286 286 + continue 287 287 + } 288 288 + for i := range wl.steps { 289 289 + if !wl.steps[i].finished { 290 290 + m.renderWorkflow(wl) 291 291 + break 292 292 + } 293 293 + } 294 294 + } 295 295 + } 296 296 + 297 297 + // applyEvent routes a decoded subscribeLogs event into the matching workflow. 298 298 + func (m *pipelineModel) applyEvent(ev *tangled.CiPipelineSubscribeLogs_Event) { 299 299 + switch { 300 300 + case ev.Error != nil: 301 301 + if ev.Error.Message != "" { 302 302 + m.streamErr = fmt.Errorf("%s: %s", ev.Error.Error, ev.Error.Message) 303 303 + } else { 304 304 + m.streamErr = fmt.Errorf("%s", ev.Error.Error) 305 305 + } 306 306 + 307 307 + case ev.Control != nil: 308 308 + c := ev.Control 309 309 + wl := m.ensureWorkflow(c.Workflow) 310 310 + switch derefStr(c.Status) { 311 311 + case "start": 312 312 + wl.stepIndex[c.Step] = len(wl.steps) 285 313 wl.steps = append(wl.steps, step{ 286 286 - id: line.StepId, name: line.Content, command: line.StepCommand, 287 287 - kind: line.StepKind, startTime: line.Time, 314 314 + id: c.Step, name: c.Content, command: derefStr(c.Command), startTime: parseRFC3339(c.Time), 288 315 }) 289 289 - case spindlemodel.StepStatusEnd: 290 290 - if idx, ok := wl.stepIndex[line.StepId]; ok { 291 291 - wl.steps[idx].endTime, wl.steps[idx].finished = line.Time, true 316 316 + case "end": 317 317 + if idx, ok := wl.stepIndex[c.Step]; ok { 318 318 + wl.steps[idx].endTime, wl.steps[idx].finished = parseRFC3339(c.Time), true 292 319 } 293 320 } 294 294 - case spindlemodel.LogKindData: 295 295 - if idx, ok := wl.stepIndex[line.StepId]; ok { 296 296 - wl.steps[idx].lines = append(wl.steps[idx].lines, line.Content) 321 321 + m.renderWorkflow(wl) 322 322 + 323 323 + case ev.Data != nil: 324 324 + d := ev.Data 325 325 + wl := m.ensureWorkflow(d.Workflow) 326 326 + if idx, ok := wl.stepIndex[d.Step]; ok { 327 327 + wl.steps[idx].lines = append(wl.steps[idx].lines, d.Content) 297 328 } 329 329 + m.renderWorkflow(wl) 298 330 } 299 331 } 300 332 333 333 + // isExpectedClose reports whether err is a clean websocket close (or nil). 334 334 + func isExpectedClose(err error) bool { 335 335 + if err == nil { 336 336 + return true 337 337 + } 338 338 + var ce *websocket.CloseError 339 339 + if errors.As(err, &ce) { 340 340 + switch ce.Code { 341 341 + case websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure: 342 342 + return true 343 343 + } 344 344 + } 345 345 + return false 346 346 + } 347 347 + 301 348 // renderLogs builds the full log content string for a workflow, used as viewport content. 302 349 func renderLogs(r *lipgloss.Renderer, wl *workflowLogs, width int) string { 303 350 headerStyle := r.NewStyle().Foreground(colorFg).Bold(true) 304 351 cmdStyle := r.NewStyle().Foreground(colorBlue).Width(width) 305 352 dimStyle := r.NewStyle().Faint(true) 306 306 - now := time.Now() 307 353 var sb strings.Builder 308 354 for i := range wl.steps { 309 355 st := &wl.steps[i] ··· 311 357 if st.finished { 312 358 dur = st.endTime.Sub(st.startTime).Round(time.Millisecond).String() 313 359 } else if !st.startTime.IsZero() { 314 314 - dur = now.Sub(st.startTime).Round(time.Second).String() 360 360 + dur = time.Since(st.startTime).Round(time.Second).String() 315 361 } 316 362 // build overlay: "── name ──...── dur ──" 317 363 nameStr := headerStyle.Render(st.name + " ") ··· 330 376 } 331 377 sb.WriteString("\n") 332 378 } 333 333 - if wl.done && wl.err != nil { 334 334 - sb.WriteString("error: " + wl.err.Error() + "\n") 335 335 - } 336 379 return sb.String() 337 380 } 338 381 ··· 340 383 body := "" 341 384 if wl := m.selectedLogs(); wl != nil && wl.ready { 342 385 body = wl.vp.View() 386 386 + } 387 387 + if m.streamErr != nil { 388 388 + body = lipgloss.JoinVertical(lipgloss.Left, body, m.renderer.NewStyle().Foreground(colorBlue).Render("stream error: "+m.streamErr.Error())) 343 389 } 344 390 return lipgloss.JoinVertical(lipgloss.Left, m.topbarView(), "", body) 345 391 } ··· 352 398 now := time.Now() 353 399 354 400 var tabs strings.Builder 355 355 - for i, wf := range m.workflows { 356 356 - status := spindlemodel.StatusKindPending 357 357 - elapsed := "" 358 358 - if ws, ok := m.pipeline.Statuses[wf]; ok { 359 359 - latest := ws.Latest() 360 360 - status = latest.Status 361 361 - if t := ws.TimeTaken(); t > 0 { 362 362 - elapsed = t.Round(time.Second).String() 363 363 - } else { 364 364 - elapsed = now.Sub(latest.Created).Round(time.Second).String() 365 365 - } 366 366 - } 367 367 - dim := r.NewStyle().Faint(true) 368 368 - base := " " + statusIcon(status, m.spinner.View()) + " " + wf 401 401 + for i, wf := range m.pipeline.Workflows { 402 402 + status := wf.Status 403 403 + elapsed := workflowElapsed(wf, now).Round(time.Second).String() 404 404 + base := " " + statusIcon(status, m.spinner.View()) + " " + wf.Name 369 405 if i == m.selected { 370 406 tab := base 371 407 if elapsed != "" { ··· 376 412 } else { 377 413 tabs.WriteString(base) 378 414 if elapsed != "" { 415 415 + dim := r.NewStyle().Faint(true) 379 416 tabs.WriteString(" " + dim.Render(elapsed)) 380 417 } 381 418 tabs.WriteString(" ") ··· 383 420 } 384 421 385 422 tabsStr := tabs.String() 386 386 - infoStr := triggerLine(r, m.pipeline.Trigger, m.pipeline.Sha) + " · " + helpText(r) 423 423 + infoStr := triggerLine(r, m.pipeline.Trigger, m.pipeline.Commit) + " · " + helpText(r) 387 424 388 425 gap := max(m.width-lipgloss.Width(tabsStr)-lipgloss.Width(infoStr), 1) 389 426 ··· 410 447 return sha 411 448 } 412 449 413 413 - func triggerLine(r *lipgloss.Renderer, t *models.Trigger, sha string) string { 450 450 + func triggerLine(r *lipgloss.Renderer, t *tangled.CiDefs_Pipeline_Trigger, sha string) string { 414 451 hash := shortSha(sha) 415 452 dim := r.NewStyle().Faint(true) 416 453 if t == nil { 417 454 return dim.Render(hash) 418 455 } 419 419 - if t.IsPush() { 420 420 - return t.TargetRef() + dim.Render("@"+hash) + dim.Render(" (push)") 456 456 + if t.CiTrigger_Push != nil { 457 457 + return t.CiTrigger_Push.Ref + dim.Render("@"+hash) + dim.Render(" (push)") 421 458 } 422 422 - if t.IsPullRequest() { 459 459 + if t.CiTrigger_PullRequest != nil { 423 460 source := "" 424 424 - if t.PRSourceBranch != nil { 425 425 - source = *t.PRSourceBranch 461 461 + if t.CiTrigger_PullRequest.SourceBranch != nil { 462 462 + source = *t.CiTrigger_PullRequest.SourceBranch 426 463 } 427 427 - return t.TargetRef() + dim.Render(" <- "+source+"@"+hash) + dim.Render(" (pull-request)") 464 464 + return t.CiTrigger_PullRequest.TargetBranch + dim.Render(" <- "+source+"@"+hash) + dim.Render(" (pull-request)") 428 465 } 429 466 return dim.Render(hash) 430 467 } 431 468 432 432 - func statusIcon(status spindlemodel.StatusKind, spinnerFrame string) string { 469 469 + func statusIcon(status string, spinnerFrame string) string { 433 470 switch status { 434 434 - case spindlemodel.StatusKindSuccess: 471 471 + case "success": 435 472 return "✓" 436 436 - case spindlemodel.StatusKindFailed: 473 473 + case "failed": 437 474 return "×" 438 438 - case spindlemodel.StatusKindRunning: 475 475 + case "running": 439 476 return spinnerFrame 440 440 - case spindlemodel.StatusKindPending: 477 477 + case "pending": 441 478 return "·" 442 442 - case spindlemodel.StatusKindTimeout: 479 479 + case "timeout": 443 480 return "⌀" 444 444 - case spindlemodel.StatusKindCancelled: 481 481 + case "cancelled": 445 482 return "-" 446 483 default: 447 484 return "?" 448 485 } 449 486 } 487 487 + 488 488 + func derefStr(s *string) string { 489 489 + if s == nil { 490 490 + return "" 491 491 + } 492 492 + return *s 493 493 + } 494 494 + 495 495 + func parseRFC3339(s string) time.Time { 496 496 + t, err := time.Parse(time.RFC3339, s) 497 497 + if err != nil { 498 498 + return time.Time{} 499 499 + } 500 500 + return t 501 501 + }

+20 -15

appview/pulls/list.go

Reviewed

··· 14 14 "tangled.org/core/orm" 15 15 16 16 "github.com/bluesky-social/indigo/atproto/syntax" 17 17 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 17 18 ) 18 19 19 20 func (s *Pulls) RepoPulls(w http.ResponseWriter, r *http.Request) { ··· 261 262 stacks = append(stacks, stack) 262 263 } 263 264 264 264 - ps, err := db.GetPipelineStatuses( 265 265 - s.db, 266 266 - len(shas), 267 267 - orm.FilterEq("p.repo_did", f.RepoDid), 268 268 - orm.FilterIn("p.sha", shas), 269 269 - ) 270 270 - if err != nil { 271 271 - l.Warn("failed to fetch pipeline statuses", "err", err) 272 272 - // non-fatal 273 273 - } 274 274 - m := make(map[string]models.Pipeline) 275 275 - for _, p := range ps { 276 276 - m[p.Sha] = p 277 277 - } 265 265 + // commitId -> latest pipeline 266 266 + pipelines := func(ctx context.Context, shas []string) map[string]tangled.CiDefs_Pipeline { 267 267 + xrpcc := &indigoxrpc.Client{Host: f.Spindle} 268 268 + out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", 0, f.RepoDid) 269 269 + if err != nil { 270 270 + l.Error("failed to fetch pipelines", "err", err) 271 271 + } 272 272 + 273 273 + m := make(map[string]tangled.CiDefs_Pipeline) 274 274 + 275 275 + for _, pipeline := range out.Pipelines { 276 276 + if pipeline == nil { 277 277 + continue 278 278 + } 279 279 + m[pipeline.Commit] = *pipeline 280 280 + } 281 281 + return m 282 282 + }(r.Context(), shas) 278 283 279 284 labelDefs, err := db.GetLabelDefinitions( 280 285 s.db, ··· 317 322 FilterState: filterState, 318 323 FilterQuery: query.String(), 319 324 Stacks: stacks, 320 320 - Pipelines: m, 325 325 + Pipelines: pipelines, 321 326 Page: page, 322 327 PullCount: totalPulls, 323 328 VouchRelationships: vouchRelationships,

+19 -16

appview/pulls/single.go

Reviewed

··· 1 1 package pulls 2 2 3 3 import ( 4 4 + "context" 4 5 "fmt" 5 6 "net/http" 6 7 "strconv" ··· 150 151 // can be nil if this pull is not stacked 151 152 stack, _ := r.Context().Value("stack").(models.Stack) 152 153 153 153 - m := make(map[string]models.Pipeline) 154 154 - 155 154 var shas []string 156 155 for _, s := range pull.Submissions { 157 156 shas = append(shas, s.SourceRev) ··· 160 159 shas = append(shas, p.LatestSha()) 161 160 } 162 161 163 163 - ps, err := db.GetPipelineStatuses( 164 164 - s.db, 165 165 - len(shas), 166 166 - orm.FilterEq("p.repo_did", f.RepoDid), 167 167 - orm.FilterIn("p.sha", shas), 168 168 - ) 169 169 - if err != nil { 170 170 - l.Error("failed to fetch pipeline statuses", "err", err) 171 171 - // non-fatal 172 172 - } 162 162 + // commitId -> latest pipeline 163 163 + pipelines := func(ctx context.Context) map[string]tangled.CiDefs_Pipeline { 164 164 + xrpcc := &indigoxrpc.Client{Host: f.Spindle} 165 165 + out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", 0, f.RepoDid) 166 166 + if err != nil { 167 167 + l.Error("failed to fetch pipelines", "err", err) 168 168 + } 173 169 174 174 - for _, p := range ps { 175 175 - m[p.Sha] = p 176 176 - } 170 170 + m := make(map[string]tangled.CiDefs_Pipeline) 171 171 + 172 172 + for _, pipeline := range out.Pipelines { 173 173 + if pipeline == nil { 174 174 + continue 175 175 + } 176 176 + m[pipeline.Commit] = *pipeline 177 177 + } 178 178 + return m 179 179 + }(r.Context()) 177 180 178 181 entities := []syntax.ATURI{pull.AtUri()} 179 182 for _, s := range pull.Submissions { ··· 256 259 BranchDeleteStatus: nil, 257 260 MergeCheck: types.MergeCheckResponse{}, 258 261 ResubmitCheck: pages.Unknown, 259 259 - Pipelines: m, 262 262 + Pipelines: pipelines, 260 263 Diff: diff, 261 264 DiffOpts: diffOpts, 262 265 ActiveRound: roundIdInt,

+1 -1

appview/repo/index.go

Reviewed

··· 177 177 for _, c := range commitsTrunc { 178 178 shas = append(shas, c.Hash.String()) 179 179 } 180 180 - pipelines, err := getPipelineStatuses(rp.db, f, shas) 180 180 + pipelines, err := getPipelineStatuses(r.Context(), f, shas) 181 181 if err != nil { 182 182 l.Error("failed to fetch pipeline statuses", "err", err) 183 183 // non-fatal

+4 -5

appview/repo/log.go

Reviewed

··· 11 11 "tangled.org/core/api/tangled" 12 12 "tangled.org/core/appview/commitverify" 13 13 "tangled.org/core/appview/db" 14 14 - "tangled.org/core/appview/models" 15 14 "tangled.org/core/appview/pages" 16 15 xrpcclient "tangled.org/core/appview/xrpcclient" 17 16 "tangled.org/core/types" ··· 178 177 for _, c := range xrpcResp.Commits { 179 178 shas = append(shas, c.Hash.String()) 180 179 } 181 181 - pipelines, err := getPipelineStatuses(rp.db, f, shas) 180 180 + pipelines, err := getPipelineStatuses(r.Context(), f, shas) 182 181 if err != nil { 183 182 l.Error("failed to getPipelineStatuses", "err", err) 184 183 // non-fatal ··· 250 249 } 251 250 252 251 user := rp.oauth.GetMultiAccountUser(r) 253 253 - pipelines, err := getPipelineStatuses(rp.db, f, []string{result.Diff.Commit.This}) 252 252 + pipelines, err := getPipelineStatuses(r.Context(), f, []string{result.Diff.Commit.This}) 254 253 if err != nil { 255 254 l.Error("failed to getPipelineStatuses", "err", err) 256 255 // non-fatal 257 256 } 258 258 - var pipeline *models.Pipeline 257 257 + var pipeline *tangled.CiDefs_Pipeline 259 258 if p, ok := pipelines[result.Diff.Commit.This]; ok { 260 260 - pipeline = &p 259 259 + pipeline = p 261 260 } 262 261 263 262 rp.pages.RepoCommit(w, pages.RepoCommitParams{

+27 -48

appview/repo/repo.go

Reviewed

··· 29 29 "tangled.org/core/appview/sites" 30 30 xrpcclient "tangled.org/core/appview/xrpcclient" 31 31 "tangled.org/core/consts" 32 32 - "tangled.org/core/eventconsumer" 33 32 "tangled.org/core/idresolver" 34 33 "tangled.org/core/ogre" 35 34 "tangled.org/core/orm" ··· 46 45 ) 47 46 48 47 type Repo struct { 49 49 - repoResolver *reporesolver.RepoResolver 50 50 - idResolver *idresolver.Resolver 51 51 - config *config.Config 52 52 - oauth *oauth.OAuth 53 53 - pages *pages.Pages 54 54 - spindlestream *eventconsumer.Consumer 55 55 - db *db.DB 56 56 - enforcer *rbac.Enforcer 57 57 - acl *knotacl.Service 58 58 - notifier notify.Notifier 59 59 - logger *slog.Logger 60 60 - serviceAuth *serviceauth.ServiceAuth 61 61 - cfClient *cloudflare.Client 62 62 - ogreClient *ogre.Client 63 63 - codesearch *codesearch.CodeSearch 48 48 + repoResolver *reporesolver.RepoResolver 49 49 + idResolver *idresolver.Resolver 50 50 + config *config.Config 51 51 + oauth *oauth.OAuth 52 52 + pages *pages.Pages 53 53 + db *db.DB 54 54 + enforcer *rbac.Enforcer 55 55 + acl *knotacl.Service 56 56 + notifier notify.Notifier 57 57 + logger *slog.Logger 58 58 + serviceAuth *serviceauth.ServiceAuth 59 59 + cfClient *cloudflare.Client 60 60 + ogreClient *ogre.Client 61 61 + codesearch *codesearch.CodeSearch 64 62 } 65 63 66 64 func New( 67 65 oauth *oauth.OAuth, 68 66 repoResolver *reporesolver.RepoResolver, 69 67 pages *pages.Pages, 70 70 - spindlestream *eventconsumer.Consumer, 71 68 idResolver *idresolver.Resolver, 72 69 db *db.DB, 73 70 config *config.Config, ··· 79 76 codesearch *codesearch.CodeSearch, 80 77 ) *Repo { 81 78 return &Repo{ 82 82 - oauth: oauth, 83 83 - repoResolver: repoResolver, 84 84 - pages: pages, 85 85 - idResolver: idResolver, 86 86 - config: config, 87 87 - spindlestream: spindlestream, 88 88 - db: db, 89 89 - notifier: notifier, 90 90 - enforcer: enforcer, 91 91 - acl: acl, 92 92 - logger: logger, 93 93 - cfClient: cfClient, 94 94 - ogreClient: ogre.NewClient(config.Ogre.Host), 95 95 - codesearch: codesearch, 79 79 + oauth: oauth, 80 80 + repoResolver: repoResolver, 81 81 + pages: pages, 82 82 + idResolver: idResolver, 83 83 + config: config, 84 84 + db: db, 85 85 + notifier: notifier, 86 86 + enforcer: enforcer, 87 87 + acl: acl, 88 88 + logger: logger, 89 89 + cfClient: cfClient, 90 90 + ogreClient: ogre.NewClient(config.Ogre.Host), 91 91 + codesearch: codesearch, 96 92 } 97 93 } 98 94 ··· 171 167 if err != nil { 172 168 fail("Failed to update spindle, unable to save to PDS.", err) 173 169 return 174 174 - } 175 175 - 176 176 - oldSpindle := f.Spindle 177 177 - if oldSpindle != "" && oldSpindle != newSpindle { 178 178 - remaining, qErr := db.GetRepos(rp.db, orm.FilterEq("spindle", oldSpindle)) 179 179 - if qErr != nil { 180 180 - l.Warn("failed to count repos using old spindle", "err", qErr) 181 181 - } else if len(remaining) == 0 { 182 182 - rp.spindlestream.RemoveSource(eventconsumer.NewSpindleSource(oldSpindle)) 183 183 - } 184 184 - } 185 185 - 186 186 - if !removingSpindle { 187 187 - rp.spindlestream.AddSource( 188 188 - context.Background(), 189 189 - eventconsumer.NewSpindleSource(newSpindle), 190 190 - ) 191 170 } 192 171 193 172 rp.pages.HxRefresh(w)

+10 -13

appview/repo/repo_util.go

Reviewed

··· 1 1 package repo 2 2 3 3 import ( 4 4 + "context" 4 5 "maps" 5 6 "slices" 6 7 "sort" 7 8 "strings" 8 9 9 9 - "tangled.org/core/appview/db" 10 10 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 11 11 + "tangled.org/core/api/tangled" 10 12 "tangled.org/core/appview/models" 11 11 - "tangled.org/core/orm" 12 13 "tangled.org/core/types" 13 14 ) 14 15 ··· 90 91 // 91 92 // golang is so blessed that it requires 35 lines of imperative code for this 92 93 func getPipelineStatuses( 93 93 - d *db.DB, 94 94 + ctx context.Context, 94 95 repo *models.Repo, 95 96 shas []string, 96 96 - ) (map[string]models.Pipeline, error) { 97 97 - m := make(map[string]models.Pipeline) 97 97 + ) (map[string]*tangled.CiDefs_Pipeline, error) { 98 98 + m := make(map[string]*tangled.CiDefs_Pipeline) 98 99 99 100 if len(shas) == 0 { 100 101 return m, nil 101 102 } 102 103 103 103 - ps, err := db.GetPipelineStatuses( 104 104 - d, 105 105 - len(shas), 106 106 - orm.FilterEq("p.repo_did", repo.RepoDid), 107 107 - orm.FilterIn("p.sha", shas), 108 108 - ) 104 104 + xrpcc := &indigoxrpc.Client{Host: repo.Spindle} 105 105 + out, err := tangled.CiQueryPipelines(ctx, xrpcc, shas, "", 0, repo.RepoDid) 109 106 if err != nil { 110 107 return nil, err 111 108 } 112 109 113 113 - for _, p := range ps { 114 114 - m[p.Sha] = p 110 110 + for _, p := range out.Pipelines { 111 111 + m[p.Commit] = p 115 112 } 116 113 117 114 return m, nil

+2 -3

appview/state/router.go

Reviewed

··· 154 154 r := chi.NewRouter() 155 155 r.Use(mw.InjectBaseParams) 156 156 157 157 + // TODO: workflow status update requests (30s polling) 158 158 + 157 159 r.With(mw.ResolveIdent()).Route("/{user}", func(r chi.Router) { 158 160 r.Get("/", s.Profile) 159 161 r.Get("/feed.atom", s.AtomFeedPage) ··· 407 409 s.oauth, 408 410 s.repoResolver, 409 411 s.pages, 410 410 - s.spindlestream, 411 412 s.idResolver, 412 413 s.db, 413 414 s.config, ··· 426 427 s.oauth, 427 428 s.repoResolver, 428 429 s.pages, 429 429 - s.spindlestream, 430 430 - s.pipelineNotifier, 431 430 s.idResolver, 432 431 s.db, 433 432 s.config,

-182

appview/state/spindlestream.go

Reviewed

··· 1 1 - package state 2 2 - 3 3 - import ( 4 4 - "context" 5 5 - "encoding/json" 6 6 - "fmt" 7 7 - "strings" 8 8 - "time" 9 9 - 10 10 - "github.com/bluesky-social/indigo/atproto/syntax" 11 11 - "tangled.org/core/api/tangled" 12 12 - "tangled.org/core/appview/config" 13 13 - "tangled.org/core/appview/db" 14 14 - "tangled.org/core/appview/models" 15 15 - "tangled.org/core/appview/pipelines" 16 16 - ec "tangled.org/core/eventconsumer" 17 17 - "tangled.org/core/eventstream" 18 18 - "tangled.org/core/log" 19 19 - "tangled.org/core/orm" 20 20 - "tangled.org/core/rbac" 21 21 - spindle "tangled.org/core/spindle/models" 22 22 - "tangled.org/core/workflow" 23 23 - ) 24 24 - 25 25 - func Spindlestream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer, pn *pipelines.StatusNotifier) (*ec.Consumer, error) { 26 26 - spindles, err := db.GetSpindles(ctx, d, orm.FilterIsNot("verified", "null")) 27 27 - if err != nil { 28 28 - return nil, err 29 29 - } 30 30 - 31 31 - hosts := make([]string, len(spindles)) 32 32 - for i, s := range spindles { 33 33 - hosts[i] = s.Instance 34 34 - } 35 35 - 36 36 - return bootstrapStream( 37 37 - ctx, "spindlestream", ec.KindSpindle, hosts, c.Redis.Addr, 38 38 - c.Spindlestream, 39 39 - spindleIngester(d, pn), 40 40 - ), nil 41 41 - } 42 42 - 43 43 - func spindleIngester(d *db.DB, pn *pipelines.StatusNotifier) ec.ProcessFunc { 44 44 - return func(ctx context.Context, source ec.Source, msg eventstream.Event) error { 45 45 - switch msg.Nsid { 46 46 - case tangled.PipelineNSID: 47 47 - return ingestPipeline(ctx, d, source, msg) 48 48 - case tangled.PipelineStatusNSID: 49 49 - return ingestPipelineStatus(ctx, d, pn, source, msg) 50 50 - } 51 51 - return nil 52 52 - } 53 53 - } 54 54 - 55 55 - func ingestPipeline(ctx context.Context, d *db.DB, source ec.Source, msg eventstream.Event) error { 56 56 - l := log.FromContext(ctx) 57 57 - 58 58 - var record tangled.Pipeline 59 59 - if err := json.Unmarshal(msg.EventJson, &record); err != nil { 60 60 - return fmt.Errorf("unmarshal pipeline: %w", err) 61 61 - } 62 62 - 63 63 - if record.TriggerMetadata == nil { 64 64 - return fmt.Errorf("empty trigger metadata: nsid %s, rkey %s", msg.Nsid, msg.Rkey) 65 65 - } 66 66 - 67 67 - if record.TriggerMetadata.Repo == nil { 68 68 - return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey) 69 69 - } 70 70 - 71 71 - repoName := "" 72 72 - if record.TriggerMetadata.Repo.Repo != nil { 73 73 - repoName = *record.TriggerMetadata.Repo.Repo 74 74 - } 75 75 - 76 76 - repo, lookupErr := resolveRepo(d, record.TriggerMetadata.Repo.RepoDid, record.TriggerMetadata.Repo.Did, repoName) 77 77 - if lookupErr != nil { 78 78 - return fmt.Errorf("failed to look up repo: %w", lookupErr) 79 79 - } 80 80 - if repo.Spindle == "" { 81 81 - return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey) 82 82 - } 83 83 - 84 84 - // trigger info 85 85 - var trigger models.Trigger 86 86 - var sha string 87 87 - trigger.Kind = workflow.TriggerKind(record.TriggerMetadata.Kind) 88 88 - switch trigger.Kind { 89 89 - case workflow.TriggerKindPush: 90 90 - trigger.PushRef = &record.TriggerMetadata.Push.Ref 91 91 - trigger.PushNewSha = &record.TriggerMetadata.Push.NewSha 92 92 - trigger.PushOldSha = &record.TriggerMetadata.Push.OldSha 93 93 - sha = *trigger.PushNewSha 94 94 - case workflow.TriggerKindPullRequest: 95 95 - trigger.PRSourceBranch = &record.TriggerMetadata.PullRequest.SourceBranch 96 96 - trigger.PRTargetBranch = &record.TriggerMetadata.PullRequest.TargetBranch 97 97 - trigger.PRSourceSha = &record.TriggerMetadata.PullRequest.SourceSha 98 98 - trigger.PRAction = &record.TriggerMetadata.PullRequest.Action 99 99 - sha = *trigger.PRSourceSha 100 100 - } 101 101 - 102 102 - tx, err := d.Begin() 103 103 - if err != nil { 104 104 - return fmt.Errorf("failed to start txn: %w", err) 105 105 - } 106 106 - 107 107 - triggerId, err := db.AddTrigger(tx, trigger) 108 108 - if err != nil { 109 109 - return fmt.Errorf("failed to add trigger entry: %w", err) 110 110 - } 111 111 - 112 112 - // TODO: we shouldn't even use knot to identify pipelines 113 113 - knot := record.TriggerMetadata.Repo.Knot 114 114 - pipeline := models.Pipeline{ 115 115 - Rkey: msg.Rkey, 116 116 - Knot: knot, 117 117 - RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did), 118 118 - RepoName: repoName, 119 119 - RepoDid: repo.RepoDid, 120 120 - TriggerId: int(triggerId), 121 121 - Sha: sha, 122 122 - } 123 123 - 124 124 - err = db.AddPipeline(tx, pipeline) 125 125 - if err != nil { 126 126 - return fmt.Errorf("failed to add pipeline: %w", err) 127 127 - } 128 128 - 129 129 - err = tx.Commit() 130 130 - if err != nil { 131 131 - return fmt.Errorf("failed to commit txn: %w", err) 132 132 - } 133 133 - 134 134 - l.Info("added pipeline", "pipeline", pipeline) 135 135 - 136 136 - return nil 137 137 - } 138 138 - 139 139 - func ingestPipelineStatus(ctx context.Context, d *db.DB, pn *pipelines.StatusNotifier, source ec.Source, msg eventstream.Event) error { 140 140 - var record tangled.PipelineStatus 141 141 - err := json.Unmarshal(msg.EventJson, &record) 142 142 - if err != nil { 143 143 - return err 144 144 - } 145 145 - 146 146 - pipelineUri, err := syntax.ParseATURI(record.Pipeline) 147 147 - if err != nil { 148 148 - return err 149 149 - } 150 150 - 151 151 - exitCode := 0 152 152 - if record.ExitCode != nil { 153 153 - exitCode = int(*record.ExitCode) 154 154 - } 155 155 - 156 156 - // pick the record creation time if possible, or use time.Now 157 157 - created := time.Now() 158 158 - if t, err := time.Parse(time.RFC3339, record.CreatedAt); err == nil && created.After(t) { 159 159 - created = t 160 160 - } 161 161 - 162 162 - status := models.PipelineStatus{ 163 163 - Spindle: source.Host, 164 164 - Rkey: msg.Rkey, 165 165 - PipelineKnot: strings.TrimPrefix(pipelineUri.Authority().String(), "did:web:"), 166 166 - PipelineRkey: pipelineUri.RecordKey().String(), 167 167 - Created: created, 168 168 - Workflow: record.Workflow, 169 169 - Status: spindle.StatusKind(record.Status), 170 170 - Error: record.Error, 171 171 - ExitCode: exitCode, 172 172 - } 173 173 - 174 174 - err = db.AddPipelineStatus(ctx, d, status) 175 175 - if err != nil { 176 176 - return fmt.Errorf("failed to add pipeline status: %w", err) 177 177 - } 178 178 - 179 179 - pn.Publish(pipelineUri) 180 180 - 181 181 - return nil 182 182 - }

-144

appview/state/spindlestream_test.go

Reviewed

··· 1 1 - package state 2 2 - 3 3 - import ( 4 4 - "context" 5 5 - "io" 6 6 - "log/slog" 7 7 - "net/http" 8 8 - "net/http/httptest" 9 9 - "path/filepath" 10 10 - "strings" 11 11 - "testing" 12 12 - "time" 13 13 - 14 14 - "tangled.org/core/appview/db" 15 15 - "tangled.org/core/appview/pipelines" 16 16 - ec "tangled.org/core/eventconsumer" 17 17 - "tangled.org/core/eventconsumer/cursor" 18 18 - "tangled.org/core/eventstream" 19 19 - "tangled.org/core/notifier" 20 20 - spindledb "tangled.org/core/spindle/db" 21 21 - spindlemodels "tangled.org/core/spindle/models" 22 22 - ) 23 23 - 24 24 - func TestColdStart_SpindleEventsRebuildPipelineStatuses(t *testing.T) { 25 25 - ctx := t.Context() 26 26 - 27 27 - spindleDB, err := spindledb.Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) 28 28 - if err != nil { 29 29 - t.Fatalf("spindle Make: %v", err) 30 30 - } 31 31 - t.Cleanup(func() { spindleDB.Close() }) 32 32 - 33 33 - n := notifier.New() 34 34 - workflowId := spindlemodels.WorkflowId{ 35 35 - PipelineId: spindlemodels.PipelineId{Knot: "knot.boltless.example", Rkey: "pipeline-rk1"}, 36 36 - Name: "build", 37 37 - } 38 38 - for _, step := range []func() error{ 39 39 - func() error { return spindleDB.StatusPending(workflowId, &n) }, 40 40 - func() error { return spindleDB.StatusRunning(workflowId, &n) }, 41 41 - func() error { return spindleDB.StatusSuccess(workflowId, &n) }, 42 42 - } { 43 43 - if err := step(); err != nil { 44 44 - t.Fatalf("seed spindle event: %v", err) 45 45 - } 46 46 - } 47 47 - 48 48 - mux := http.NewServeMux() 49 49 - mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) { 50 50 - _ = eventstream.Stream(w, r, eventstream.StreamConfig{ 51 51 - Backend: spindleDB, 52 52 - Notifier: &n, 53 53 - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), 54 54 - }) 55 55 - }) 56 56 - srv := httptest.NewServer(mux) 57 57 - t.Cleanup(srv.Close) 58 58 - source := ec.Source{Kind: "test", Host: strings.TrimPrefix(srv.URL, "http://"), NoTLS: true} 59 59 - 60 60 - appviewDB, err := db.Make(ctx, filepath.Join(t.TempDir(), "appview.db")) 61 61 - if err != nil { 62 62 - t.Fatalf("appview Make: %v", err) 63 63 - } 64 64 - t.Cleanup(func() { appviewDB.Close() }) 65 65 - 66 66 - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) 67 67 - processFunc := spindleIngester(appviewDB, pipelines.NewStatusNotifier()) 68 68 - 69 69 - cfg := ec.ConsumerConfig{ 70 70 - ProcessFunc: processFunc, 71 71 - WorkerCount: 1, 72 72 - QueueSize: 16, 73 73 - ConnectionTimeout: 2 * time.Second, 74 74 - CursorStore: &cursor.MemoryStore{}, 75 75 - Logger: logger, 76 76 - } 77 77 - c := ec.NewConsumer(cfg) 78 78 - 79 79 - consumerCtx, cancel := context.WithCancel(ctx) 80 80 - defer cancel() 81 81 - c.Start(consumerCtx) 82 82 - c.AddSource(consumerCtx, source) 83 83 - 84 84 - deadline := time.Now().Add(3 * time.Second) 85 85 - for time.Now().Before(deadline) { 86 86 - var n int 87 87 - if err := appviewDB.QueryRow(`select count(*) from pipeline_statuses`).Scan(&n); err != nil { 88 88 - t.Fatalf("count: %v", err) 89 89 - } 90 90 - if n >= 3 { 91 91 - break 92 92 - } 93 93 - time.Sleep(20 * time.Millisecond) 94 94 - } 95 95 - 96 96 - rows, err := appviewDB.Query(` 97 97 - select spindle, pipeline_knot, pipeline_rkey, workflow, status 98 98 - from pipeline_statuses 99 99 - order by created asc 100 100 - `) 101 101 - if err != nil { 102 102 - t.Fatalf("query: %v", err) 103 103 - } 104 104 - defer rows.Close() 105 105 - 106 106 - type rec struct { 107 107 - spindle, knot, rkey, workflow, status string 108 108 - } 109 109 - var got []rec 110 110 - for rows.Next() { 111 111 - var r rec 112 112 - if err := rows.Scan(&r.spindle, &r.knot, &r.rkey, &r.workflow, &r.status); err != nil { 113 113 - t.Fatalf("scan: %v", err) 114 114 - } 115 115 - got = append(got, r) 116 116 - } 117 117 - 118 118 - if len(got) != 3 { 119 119 - t.Fatalf("pipeline_statuses rows = %d, want 3: %+v", len(got), got) 120 120 - } 121 121 - 122 122 - wantStatuses := []string{"pending", "running", "success"} 123 123 - gotStatuses := map[string]bool{} 124 124 - for _, r := range got { 125 125 - gotStatuses[r.status] = true 126 126 - if r.spindle != source.Host { 127 127 - t.Errorf("spindle = %q, want %q", r.spindle, source.Host) 128 128 - } 129 129 - if r.knot != workflowId.Knot { 130 130 - t.Errorf("pipeline_knot = %q, want %q", r.knot, workflowId.Knot) 131 131 - } 132 132 - if r.rkey != workflowId.Rkey { 133 133 - t.Errorf("pipeline_rkey = %q, want %q", r.rkey, workflowId.Rkey) 134 134 - } 135 135 - if r.workflow != workflowId.Name { 136 136 - t.Errorf("workflow = %q, want %q", r.workflow, workflowId.Name) 137 137 - } 138 138 - } 139 139 - for _, want := range wantStatuses { 140 140 - if !gotStatuses[want] { 141 141 - t.Errorf("missing status %q in projection", want) 142 142 - } 143 143 - } 144 144 - }

+1 -14

appview/state/state.go

Reviewed

··· 31 31 whnotify "tangled.org/core/appview/notify/webhook" 32 32 "tangled.org/core/appview/oauth" 33 33 "tangled.org/core/appview/pages" 34 34 - "tangled.org/core/appview/pipelines" 35 34 pipelinessh "tangled.org/core/appview/pipelines/ssh" 36 35 "tangled.org/core/appview/reporesolver" 37 36 "tangled.org/core/appview/repoverify" ··· 71 70 repoResolver *reporesolver.RepoResolver 72 71 aclService *knotacl.Service 73 72 knotstream *eventconsumer.Consumer 74 74 - spindlestream *eventconsumer.Consumer 75 75 - pipelineNotifier *pipelines.StatusNotifier 76 73 logger *slog.Logger 77 74 cfClient *cloudflare.Client 78 75 codesearch *codesearch.CodeSearch ··· 220 217 } 221 218 knotstream.Start(ctx) 222 219 223 223 - pipelineNotifier := pipelines.NewStatusNotifier() 224 224 - 225 225 - spindlestream, err := Spindlestream(ctx, config, d, enforcer, pipelineNotifier) 226 226 - if err != nil { 227 227 - return nil, fmt.Errorf("failed to start spindlestream consumer: %w", err) 228 228 - } 229 229 - spindlestream.Start(ctx) 230 230 - 231 220 state := &State{ 232 221 db: d, 233 222 notifier: notifier, ··· 244 233 repoResolver: repoResolver, 245 234 aclService: aclService, 246 235 knotstream: knotstream, 247 247 - spindlestream: spindlestream, 248 248 - pipelineNotifier: pipelineNotifier, 249 236 logger: logger, 250 237 cfClient: cfClient, 251 238 codesearch: &codesearch.CodeSearch{Host: config.CodeSearch.ZoektUrl}, ··· 263 250 } 264 251 265 252 func (s *State) NewSSHServer() *pipelinessh.Server { 266 266 - return pipelinessh.New(s.db, s.config, s.pipelineNotifier, log.SubLogger(s.logger, "pipelinessh")) 253 253 + return pipelinessh.New(s.db, s.config, log.SubLogger(s.logger, "pipelinessh")) 267 254 } 268 255 269 256 func (s *State) SecurityTxt(w http.ResponseWriter, r *http.Request) {