+116 -2

appview/db/db.go

··· 377 378 repo_owner text not null, 379 repo_name text not null, 0 380 381 -- every pipeline must be associated with exactly one commit 382 sha text not null check (length(sha) = 40), ··· 386 trigger_id integer not null, 387 388 unique(knot, rkey), 389 - foreign key (trigger_id) references triggers(id) on delete cascade 0 390 ); 391 392 create table if not exists triggers ( ··· 395 396 -- top-level fields 397 kind text not null, 0 398 399 -- pushTriggerData fields 400 push_ref text, ··· 405 pr_source_branch text, 406 pr_target_branch text, 407 pr_source_sha text check (length(pr_source_sha) = 40), 408 - pr_action text 0 0 409 ); 410 411 create table if not exists pipeline_statuses ( ··· 1204 `) 1205 return err 1206 }) 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 1207 1208 return &DB{ 1209 db,

··· 377 378 repo_owner text not null, 379 repo_name text not null, 380 + repo_at text not null, 381 382 -- every pipeline must be associated with exactly one commit 383 sha text not null check (length(sha) = 40), ··· 387 trigger_id integer not null, 388 389 unique(knot, rkey), 390 + foreign key (trigger_id) references triggers(id) on delete cascade, 391 + foreign key (repo_at) references repos(at_uri) on delete cascade 392 ); 393 394 create table if not exists triggers ( ··· 397 398 -- top-level fields 399 kind text not null, 400 + repo_at text not null, 401 402 -- pushTriggerData fields 403 push_ref text, ··· 408 pr_source_branch text, 409 pr_target_branch text, 410 pr_source_sha text check (length(pr_source_sha) = 40), 411 + pr_action text, 412 + 413 + foreign key (repo_at) references repos(at_uri) on delete cascade 414 ); 415 416 create table if not exists pipeline_statuses ( ··· 1209 `) 1210 return err 1211 }) 1212 + 1213 + conn.ExecContext(ctx, "pragma foreign_keys = off;") 1214 + orm.RunMigration(conn, logger, "add-repo-at-fk-to-pipelines", func(tx *sql.Tx) error { 1215 + _, err := tx.Exec(` 1216 + create table if not exists pipelines_new ( 1217 + id integer primary key autoincrement, 1218 + knot text not null, 1219 + rkey text not null, 1220 + 1221 + repo_owner text not null, 1222 + repo_name text not null, 1223 + repo_at text not null, 1224 + 1225 + sha text not null check (length(sha) = 40), 1226 + created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 1227 + 1228 + trigger_id integer not null, 1229 + 1230 + unique(knot, rkey), 1231 + foreign key (trigger_id) references triggers(id) on delete cascade, 1232 + foreign key (repo_at) references repos(at_uri) on delete cascade 1233 + ); 1234 + `) 1235 + if err != nil { 1236 + return err 1237 + } 1238 + 1239 + _, err = tx.Exec(` 1240 + -- backfill repo_at from repos; orphaned rows are dropped 1241 + insert into pipelines_new (id, knot, rkey, repo_owner, repo_name, repo_at, sha, created, trigger_id) 1242 + select 1243 + p.id, 1244 + p.knot, 1245 + p.rkey, 1246 + p.repo_owner, 1247 + p.repo_name, 1248 + r.at_uri, 1249 + p.sha, 1250 + p.created, 1251 + p.trigger_id 1252 + from pipelines p 1253 + join repos r on p.repo_owner = r.did and p.repo_name = r.name; 1254 + `) 1255 + if err != nil { 1256 + return err 1257 + } 1258 + 1259 + _, err = tx.Exec(`drop table pipelines`) 1260 + if err != nil { 1261 + return err 1262 + } 1263 + 1264 + _, err = tx.Exec(`alter table pipelines_new rename to pipelines`) 1265 + return err 1266 + }) 1267 + 1268 + orm.RunMigration(conn, logger, "add-repo-at-fk-to-triggers", func(tx *sql.Tx) error { 1269 + _, err := tx.Exec(` 1270 + create table if not exists triggers_new ( 1271 + id integer primary key autoincrement, 1272 + kind text not null, 1273 + repo_at text not null, 1274 + 1275 + push_ref text, 1276 + push_new_sha text check (length(push_new_sha) = 40), 1277 + push_old_sha text check (length(push_old_sha) = 40), 1278 + 1279 + pr_source_branch text, 1280 + pr_target_branch text, 1281 + pr_source_sha text check (length(pr_source_sha) = 40), 1282 + pr_action text, 1283 + 1284 + foreign key (repo_at) references repos(at_uri) on delete cascade 1285 + ); 1286 + `) 1287 + if err != nil { 1288 + return err 1289 + } 1290 + 1291 + _, err = tx.Exec(` 1292 + -- backfill repo_at from pipelines; orphaned rows are dropped 1293 + insert into triggers_new (id, kind, repo_at, push_ref, push_new_sha, push_old_sha, pr_source_branch, pr_target_branch, pr_source_sha, pr_action) 1294 + select 1295 + t.id, 1296 + t.kind, 1297 + p.repo_at, 1298 + t.push_ref, 1299 + t.push_new_sha, 1300 + t.push_old_sha, 1301 + t.pr_source_branch, 1302 + t.pr_target_branch, 1303 + t.pr_source_sha, 1304 + t.pr_action 1305 + from triggers t 1306 + join pipelines p on t.id = p.trigger_id; 1307 + `) 1308 + if err != nil { 1309 + return err 1310 + } 1311 + 1312 + _, err = tx.Exec(`drop table triggers`) 1313 + if err != nil { 1314 + return err 1315 + } 1316 + 1317 + _, err = tx.Exec(`alter table triggers_new rename to triggers`) 1318 + return err 1319 + }) 1320 + conn.ExecContext(ctx, "pragma foreign_keys = on;") 1321 1322 return &DB{ 1323 db,

+8 -1

appview/db/pipeline.go

··· 26 whereClause = " where " + strings.Join(conditions, " and ") 27 } 28 29 - query := fmt.Sprintf(`select id, rkey, knot, repo_owner, repo_name, sha, created from pipelines %s`, whereClause) 30 31 rows, err := e.Query(query, args...) 32 ··· 44 &pipeline.Knot, 45 &pipeline.RepoOwner, 46 &pipeline.RepoName, 0 47 &pipeline.Sha, 48 &createdAt, 49 ) ··· 71 pipeline.Knot, 72 pipeline.RepoOwner, 73 pipeline.RepoName, 0 74 pipeline.TriggerId, 75 pipeline.Sha, 76 } ··· 86 knot, 87 repo_owner, 88 repo_name, 0 89 trigger_id, 90 sha 91 ) values (%s) ··· 99 func AddTrigger(e Execer, trigger models.Trigger) (int64, error) { 100 args := []any{ 101 trigger.Kind, 0 102 trigger.PushRef, 103 trigger.PushNewSha, 104 trigger.PushOldSha, ··· 115 116 query := fmt.Sprintf(`insert or ignore into triggers ( 117 kind, 0 118 push_ref, 119 push_new_sha, 120 push_old_sha, ··· 193 p.rkey, 194 p.repo_owner, 195 p.repo_name, 0 196 p.sha, 197 p.created, 198 t.id, ··· 231 &p.Rkey, 232 &p.RepoOwner, 233 &p.RepoName, 0 234 &p.Sha, 235 &created, 236 &p.TriggerId,

··· 26 whereClause = " where " + strings.Join(conditions, " and ") 27 } 28 29 + query := fmt.Sprintf(`select id, rkey, knot, repo_owner, repo_name, repo_at, sha, created from pipelines %s`, whereClause) 30 31 rows, err := e.Query(query, args...) 32 ··· 44 &pipeline.Knot, 45 &pipeline.RepoOwner, 46 &pipeline.RepoName, 47 + &pipeline.RepoAt, 48 &pipeline.Sha, 49 &createdAt, 50 ) ··· 72 pipeline.Knot, 73 pipeline.RepoOwner, 74 pipeline.RepoName, 75 + pipeline.RepoAt, 76 pipeline.TriggerId, 77 pipeline.Sha, 78 } ··· 88 knot, 89 repo_owner, 90 repo_name, 91 + repo_at, 92 trigger_id, 93 sha 94 ) values (%s) ··· 102 func AddTrigger(e Execer, trigger models.Trigger) (int64, error) { 103 args := []any{ 104 trigger.Kind, 105 + trigger.RepoAt, 106 trigger.PushRef, 107 trigger.PushNewSha, 108 trigger.PushOldSha, ··· 119 120 query := fmt.Sprintf(`insert or ignore into triggers ( 121 kind, 122 + repo_at, 123 push_ref, 124 push_new_sha, 125 push_old_sha, ··· 198 p.rkey, 199 p.repo_owner, 200 p.repo_name, 201 + p.repo_at, 202 p.sha, 203 p.created, 204 t.id, ··· 237 &p.Rkey, 238 &p.RepoOwner, 239 &p.RepoName, 240 + &p.RepoAt, 241 &p.Sha, 242 &created, 243 &p.TriggerId,

+4 -2

appview/models/pipeline.go

··· 19 Knot string 20 RepoOwner syntax.DID 21 RepoName string 0 22 TriggerId int 23 Sha string 24 Created time.Time ··· 127 } 128 129 type Trigger struct { 130 - Id int 131 - Kind workflow.TriggerKind 0 132 133 // push trigger fields 134 PushRef *string

··· 19 Knot string 20 RepoOwner syntax.DID 21 RepoName string 22 + RepoAt syntax.ATURI 23 TriggerId int 24 Sha string 25 Created time.Time ··· 128 } 129 130 type Trigger struct { 131 + Id int 132 + Kind workflow.TriggerKind 133 + RepoAt syntax.ATURI 134 135 // push trigger fields 136 PushRef *string

+4 -8

appview/pipelines/pipelines.go

··· 88 89 filterKind := r.URL.Query().Get("trigger") 90 filters := []orm.Filter{ 91 - orm.FilterEq("p.repo_owner", f.Did), 92 - orm.FilterEq("p.repo_name", f.Name), 93 orm.FilterEq("p.knot", f.Knot), 94 } 95 switch filterKind { ··· 152 ps, err := db.GetPipelineStatuses( 153 p.db, 154 1, 155 - orm.FilterEq("p.repo_owner", f.Did), 156 - orm.FilterEq("p.repo_name", f.Name), 157 orm.FilterEq("p.knot", f.Knot), 158 orm.FilterEq("p.id", pipelineId), 159 ) ··· 219 ps, err := db.GetPipelineStatuses( 220 p.db, 221 1, 222 - orm.FilterEq("p.repo_owner", f.Did), 223 - orm.FilterEq("p.repo_name", f.Name), 224 orm.FilterEq("p.knot", f.Knot), 225 orm.FilterEq("p.id", pipelineId), 226 ) ··· 368 ps, err := db.GetPipelineStatuses( 369 p.db, 370 1, 371 - orm.FilterEq("p.repo_owner", f.Did), 372 - orm.FilterEq("p.repo_name", f.Name), 373 orm.FilterEq("p.knot", f.Knot), 374 orm.FilterEq("p.id", pipelineId), 375 )

··· 88 89 filterKind := r.URL.Query().Get("trigger") 90 filters := []orm.Filter{ 91 + orm.FilterEq("p.repo_at", f.RepoAt()), 0 92 orm.FilterEq("p.knot", f.Knot), 93 } 94 switch filterKind { ··· 151 ps, err := db.GetPipelineStatuses( 152 p.db, 153 1, 154 + orm.FilterEq("p.repo_at", f.RepoAt()), 0 155 orm.FilterEq("p.knot", f.Knot), 156 orm.FilterEq("p.id", pipelineId), 157 ) ··· 217 ps, err := db.GetPipelineStatuses( 218 p.db, 219 1, 220 + orm.FilterEq("p.repo_at", f.RepoAt()), 0 221 orm.FilterEq("p.knot", f.Knot), 222 orm.FilterEq("p.id", pipelineId), 223 ) ··· 365 ps, err := db.GetPipelineStatuses( 366 p.db, 367 1, 368 + orm.FilterEq("p.repo_at", f.RepoAt()), 0 369 orm.FilterEq("p.knot", f.Knot), 370 orm.FilterEq("p.id", pipelineId), 371 )

+2 -4

appview/pulls/pulls.go

··· 214 ps, err := db.GetPipelineStatuses( 215 s.db, 216 len(shas), 217 - orm.FilterEq("p.repo_owner", f.Did), 218 - orm.FilterEq("p.repo_name", f.Name), 219 orm.FilterEq("p.knot", f.Knot), 220 orm.FilterIn("p.sha", shas), 221 ) ··· 636 ps, err := db.GetPipelineStatuses( 637 s.db, 638 len(shas), 639 - orm.FilterEq("p.repo_owner", f.Did), 640 - orm.FilterEq("p.repo_name", f.Name), 641 orm.FilterEq("p.knot", f.Knot), 642 orm.FilterIn("p.sha", shas), 643 )

··· 214 ps, err := db.GetPipelineStatuses( 215 s.db, 216 len(shas), 217 + orm.FilterEq("p.repo_at", f.RepoAt()), 0 218 orm.FilterEq("p.knot", f.Knot), 219 orm.FilterIn("p.sha", shas), 220 ) ··· 635 ps, err := db.GetPipelineStatuses( 636 s.db, 637 len(shas), 638 + orm.FilterEq("p.repo_at", f.RepoAt()), 0 639 orm.FilterEq("p.knot", f.Knot), 640 orm.FilterIn("p.sha", shas), 641 )

+1 -2

appview/repo/repo_util.go

··· 103 ps, err := db.GetPipelineStatuses( 104 d, 105 len(shas), 106 - orm.FilterEq("p.repo_owner", repo.Did), 107 - orm.FilterEq("p.repo_name", repo.Name), 108 orm.FilterEq("p.knot", repo.Knot), 109 orm.FilterIn("p.sha", shas), 110 )

··· 103 ps, err := db.GetPipelineStatuses( 104 d, 105 len(shas), 106 + orm.FilterEq("p.repo_at", repo.RepoAt()), 0 107 orm.FilterEq("p.knot", repo.Knot), 108 orm.FilterIn("p.sha", shas), 109 )

+2

appview/state/knotstream.go

··· 246 return fmt.Errorf("failed to start txn: %w", err) 247 } 248 0 249 triggerId, err := db.AddTrigger(tx, trigger) 250 if err != nil { 251 return fmt.Errorf("failed to add trigger entry: %w", err) ··· 256 Knot: source.Key(), 257 RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did), 258 RepoName: record.TriggerMetadata.Repo.Repo, 0 259 TriggerId: int(triggerId), 260 Sha: sha, 261 }

··· 246 return fmt.Errorf("failed to start txn: %w", err) 247 } 248 249 + trigger.RepoAt = repos[0].RepoAt() 250 triggerId, err := db.AddTrigger(tx, trigger) 251 if err != nil { 252 return fmt.Errorf("failed to add trigger entry: %w", err) ··· 257 Knot: source.Key(), 258 RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did), 259 RepoName: record.TriggerMetadata.Repo.Repo, 260 + RepoAt: repos[0].RepoAt(), 261 TriggerId: int(triggerId), 262 Sha: sha, 263 }

+13 -10

cmd/populatepipelines/populate_pipelines.go

··· 68 log.Fatalf("Invalid repo format: %s (expected: did:plc:xyz/reponame)", *repo) 69 } 70 0 0 0 71 db, err := sql.Open("sqlite3", *dbPath) 72 if err != nil { 73 log.Fatalf("Failed to open database: %v", err) ··· 87 88 var triggerId int64 89 if isPush { 90 - triggerId, err = createPushTrigger(db, branches) 91 } else { 92 - triggerId, err = createPRTrigger(db, branches) 93 } 94 if err != nil { 95 log.Fatalf("Failed to create trigger: %v", err) ··· 147 return "", "", false 148 } 149 150 - func createPushTrigger(db *sql.DB, branches []string) (int64, error) { 151 branch := branches[rand.Intn(len(branches))] 152 oldSha := generateRandomSha() 153 newSha := generateRandomSha() 154 155 result, err := db.Exec(` 156 - INSERT INTO triggers (kind, push_ref, push_new_sha, push_old_sha) 157 - VALUES (?, ?, ?, ?) 158 - `, "push", "refs/heads/"+branch, newSha, oldSha) 159 160 if err != nil { 161 return 0, err ··· 164 return result.LastInsertId() 165 } 166 167 - func createPRTrigger(db *sql.DB, branches []string) (int64, error) { 168 targetBranch := branches[0] // Usually main 169 sourceBranch := branches[rand.Intn(len(branches)-1)+1] 170 sourceSha := generateRandomSha() ··· 172 action := actions[rand.Intn(len(actions))] 173 174 result, err := db.Exec(` 175 - INSERT INTO triggers (kind, pr_source_branch, pr_target_branch, pr_source_sha, pr_action) 176 - VALUES (?, ?, ?, ?, ?) 177 - `, "pull_request", sourceBranch, targetBranch, sourceSha, action) 178 179 if err != nil { 180 return 0, err

··· 68 log.Fatalf("Invalid repo format: %s (expected: did:plc:xyz/reponame)", *repo) 69 } 70 71 + // Construct the AT URI for repo_at lookups 72 + repoAt := fmt.Sprintf("at://%s/sh.tangled.repo/%s", did, repoName) 73 + 74 db, err := sql.Open("sqlite3", *dbPath) 75 if err != nil { 76 log.Fatalf("Failed to open database: %v", err) ··· 90 91 var triggerId int64 92 if isPush { 93 + triggerId, err = createPushTrigger(db, repoAt, branches) 94 } else { 95 + triggerId, err = createPRTrigger(db, repoAt, branches) 96 } 97 if err != nil { 98 log.Fatalf("Failed to create trigger: %v", err) ··· 150 return "", "", false 151 } 152 153 + func createPushTrigger(db *sql.DB, repoAt string, branches []string) (int64, error) { 154 branch := branches[rand.Intn(len(branches))] 155 oldSha := generateRandomSha() 156 newSha := generateRandomSha() 157 158 result, err := db.Exec(` 159 + INSERT INTO triggers (kind, repo_at, push_ref, push_new_sha, push_old_sha) 160 + VALUES (?, ?, ?, ?, ?) 161 + `, "push", repoAt, "refs/heads/"+branch, newSha, oldSha) 162 163 if err != nil { 164 return 0, err ··· 167 return result.LastInsertId() 168 } 169 170 + func createPRTrigger(db *sql.DB, repoAt string, branches []string) (int64, error) { 171 targetBranch := branches[0] // Usually main 172 sourceBranch := branches[rand.Intn(len(branches)-1)+1] 173 sourceSha := generateRandomSha() ··· 175 action := actions[rand.Intn(len(actions))] 176 177 result, err := db.Exec(` 178 + INSERT INTO triggers (kind, repo_at, pr_source_branch, pr_target_branch, pr_source_sha, pr_action) 179 + VALUES (?, ?, ?, ?, ?, ?) 180 + `, "pull_request", repoAt, sourceBranch, targetBranch, sourceSha, action) 181 182 if err != nil { 183 return 0, err