forked from tangled.org/core
this repo has no description
1package db 2 3import ( 4 "context" 5 "database/sql" 6 "fmt" 7 "log" 8 "reflect" 9 "strings" 10 11 _ "github.com/mattn/go-sqlite3" 12) 13 14type DB struct { 15 *sql.DB 16} 17 18type Execer interface { 19 Query(query string, args ...any) (*sql.Rows, error) 20 QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) 21 QueryRow(query string, args ...any) *sql.Row 22 QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row 23 Exec(query string, args ...any) (sql.Result, error) 24 ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) 25 Prepare(query string) (*sql.Stmt, error) 26 PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) 27} 28 29func Make(dbPath string) (*DB, error) { 30 db, err := sql.Open("sqlite3", dbPath) 31 if err != nil { 32 return nil, err 33 } 34 _, err = db.Exec(` 35 pragma journal_mode = WAL; 36 pragma synchronous = normal; 37 pragma foreign_keys = on; 38 pragma temp_store = memory; 39 pragma mmap_size = 30000000000; 40 pragma page_size = 32768; 41 pragma auto_vacuum = incremental; 42 pragma busy_timeout = 5000; 43 44 create table if not exists registrations ( 45 id integer primary key autoincrement, 46 domain text not null unique, 47 did text not null, 48 secret text not null, 49 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 50 registered text 51 ); 52 create table if not exists public_keys ( 53 id integer primary key autoincrement, 54 did text not null, 55 name text not null, 56 key text not null, 57 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 58 unique(did, name, key) 59 ); 60 create table if not exists repos ( 61 id integer primary key autoincrement, 62 did text not null, 63 name text not null, 64 knot text not null, 65 rkey text not null, 66 at_uri text not null unique, 67 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 68 unique(did, name, knot, rkey) 69 ); 70 create table if not exists collaborators ( 71 id integer primary key autoincrement, 72 did text not null, 73 repo integer not null, 74 foreign key (repo) references repos(id) on delete cascade 75 ); 76 create table if not exists follows ( 77 user_did text not null, 78 subject_did text not null, 79 rkey text not null, 80 followed_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 81 primary key (user_did, subject_did), 82 check (user_did <> subject_did) 83 ); 84 create table if not exists issues ( 85 id integer primary key autoincrement, 86 owner_did text not null, 87 repo_at text not null, 88 issue_id integer not null, 89 title text not null, 90 body text not null, 91 open integer not null default 1, 92 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 93 issue_at text, 94 unique(repo_at, issue_id), 95 foreign key (repo_at) references repos(at_uri) on delete cascade 96 ); 97 create table if not exists comments ( 98 id integer primary key autoincrement, 99 owner_did text not null, 100 issue_id integer not null, 101 repo_at text not null, 102 comment_id integer not null, 103 comment_at text not null, 104 body text not null, 105 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 106 unique(issue_id, comment_id), 107 foreign key (repo_at, issue_id) references issues(repo_at, issue_id) on delete cascade 108 ); 109 create table if not exists pulls ( 110 -- identifiers 111 id integer primary key autoincrement, 112 pull_id integer not null, 113 114 -- at identifiers 115 repo_at text not null, 116 owner_did text not null, 117 rkey text not null, 118 pull_at text, 119 120 -- content 121 title text not null, 122 body text not null, 123 target_branch text not null, 124 state integer not null default 0 check (state in (0, 1, 2)), -- open, merged, closed 125 126 -- meta 127 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 128 129 -- constraints 130 unique(repo_at, pull_id), 131 foreign key (repo_at) references repos(at_uri) on delete cascade 132 ); 133 134 -- every pull must have atleast 1 submission: the initial submission 135 create table if not exists pull_submissions ( 136 -- identifiers 137 id integer primary key autoincrement, 138 pull_id integer not null, 139 140 -- at identifiers 141 repo_at text not null, 142 143 -- content, these are immutable, and require a resubmission to update 144 round_number integer not null default 0, 145 patch text, 146 147 -- meta 148 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 149 150 -- constraints 151 unique(repo_at, pull_id, round_number), 152 foreign key (repo_at, pull_id) references pulls(repo_at, pull_id) on delete cascade 153 ); 154 155 create table if not exists pull_comments ( 156 -- identifiers 157 id integer primary key autoincrement, 158 pull_id integer not null, 159 submission_id integer not null, 160 161 -- at identifiers 162 repo_at text not null, 163 owner_did text not null, 164 comment_at text not null, 165 166 -- content 167 body text not null, 168 169 -- meta 170 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 171 172 -- constraints 173 foreign key (repo_at, pull_id) references pulls(repo_at, pull_id) on delete cascade, 174 foreign key (submission_id) references pull_submissions(id) on delete cascade 175 ); 176 177 create table if not exists _jetstream ( 178 id integer primary key autoincrement, 179 last_time_us integer not null 180 ); 181 182 create table if not exists repo_issue_seqs ( 183 repo_at text primary key, 184 next_issue_id integer not null default 1 185 ); 186 187 create table if not exists repo_pull_seqs ( 188 repo_at text primary key, 189 next_pull_id integer not null default 1 190 ); 191 192 create table if not exists stars ( 193 id integer primary key autoincrement, 194 starred_by_did text not null, 195 repo_at text not null, 196 rkey text not null, 197 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 198 foreign key (repo_at) references repos(at_uri) on delete cascade, 199 unique(starred_by_did, repo_at) 200 ); 201 202 create table if not exists reactions ( 203 id integer primary key autoincrement, 204 reacted_by_did text not null, 205 thread_at text not null, 206 kind text not null, 207 rkey text not null, 208 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 209 unique(reacted_by_did, thread_at, kind) 210 ); 211 212 create table if not exists emails ( 213 id integer primary key autoincrement, 214 did text not null, 215 email text not null, 216 verified integer not null default 0, 217 verification_code text not null, 218 last_sent text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 219 is_primary integer not null default 0, 220 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 221 unique(did, email) 222 ); 223 224 create table if not exists artifacts ( 225 -- id 226 id integer primary key autoincrement, 227 did text not null, 228 rkey text not null, 229 230 -- meta 231 repo_at text not null, 232 tag binary(20) not null, 233 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 234 235 -- data 236 blob_cid text not null, 237 name text not null, 238 size integer not null default 0, 239 mimetype string not null default "*/*", 240 241 -- constraints 242 unique(did, rkey), -- record must be unique 243 unique(repo_at, tag, name), -- for a given tag object, each file must be unique 244 foreign key (repo_at) references repos(at_uri) on delete cascade 245 ); 246 247 create table if not exists profile ( 248 -- id 249 id integer primary key autoincrement, 250 did text not null, 251 252 -- data 253 description text not null, 254 include_bluesky integer not null default 0, 255 location text, 256 257 -- constraints 258 unique(did) 259 ); 260 create table if not exists profile_links ( 261 -- id 262 id integer primary key autoincrement, 263 did text not null, 264 265 -- data 266 link text not null, 267 268 -- constraints 269 foreign key (did) references profile(did) on delete cascade 270 ); 271 create table if not exists profile_stats ( 272 -- id 273 id integer primary key autoincrement, 274 did text not null, 275 276 -- data 277 kind text not null check (kind in ( 278 "merged-pull-request-count", 279 "closed-pull-request-count", 280 "open-pull-request-count", 281 "open-issue-count", 282 "closed-issue-count", 283 "repository-count" 284 )), 285 286 -- constraints 287 foreign key (did) references profile(did) on delete cascade 288 ); 289 create table if not exists profile_pinned_repositories ( 290 -- id 291 id integer primary key autoincrement, 292 did text not null, 293 294 -- data 295 at_uri text not null, 296 297 -- constraints 298 unique(did, at_uri), 299 foreign key (did) references profile(did) on delete cascade, 300 foreign key (at_uri) references repos(at_uri) on delete cascade 301 ); 302 303 create table if not exists oauth_requests ( 304 id integer primary key autoincrement, 305 auth_server_iss text not null, 306 state text not null, 307 did text not null, 308 handle text not null, 309 pds_url text not null, 310 pkce_verifier text not null, 311 dpop_auth_server_nonce text not null, 312 dpop_private_jwk text not null 313 ); 314 315 create table if not exists oauth_sessions ( 316 id integer primary key autoincrement, 317 did text not null, 318 handle text not null, 319 pds_url text not null, 320 auth_server_iss text not null, 321 access_jwt text not null, 322 refresh_jwt text not null, 323 dpop_pds_nonce text, 324 dpop_auth_server_nonce text not null, 325 dpop_private_jwk text not null, 326 expiry text not null 327 ); 328 329 create table if not exists punchcard ( 330 did text not null, 331 date text not null, -- yyyy-mm-dd 332 count integer, 333 primary key (did, date) 334 ); 335 336 create table if not exists spindles ( 337 id integer primary key autoincrement, 338 owner text not null, 339 instance text not null, 340 verified text, -- time of verification 341 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 342 343 unique(owner, instance) 344 ); 345 346 create table if not exists spindle_members ( 347 -- identifiers for the record 348 id integer primary key autoincrement, 349 did text not null, 350 rkey text not null, 351 352 -- data 353 instance text not null, 354 subject text not null, 355 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 356 357 -- constraints 358 foreign key (did, instance) references spindles(owner, instance) on delete cascade, 359 unique (did, instance, subject) 360 ); 361 362 create table if not exists pipelines ( 363 -- identifiers 364 id integer primary key autoincrement, 365 knot text not null, 366 rkey text not null, 367 368 repo_owner text not null, 369 repo_name text not null, 370 371 -- every pipeline must be associated with exactly one commit 372 sha text not null check (length(sha) = 40), 373 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 374 375 -- trigger data 376 trigger_id integer not null, 377 378 unique(knot, rkey), 379 foreign key (trigger_id) references triggers(id) on delete cascade 380 ); 381 382 create table if not exists triggers ( 383 -- primary key 384 id integer primary key autoincrement, 385 386 -- top-level fields 387 kind text not null, 388 389 -- pushTriggerData fields 390 push_ref text, 391 push_new_sha text check (length(push_new_sha) = 40), 392 push_old_sha text check (length(push_old_sha) = 40), 393 394 -- pullRequestTriggerData fields 395 pr_source_branch text, 396 pr_target_branch text, 397 pr_source_sha text check (length(pr_source_sha) = 40), 398 pr_action text 399 ); 400 401 create table if not exists pipeline_statuses ( 402 -- identifiers 403 id integer primary key autoincrement, 404 spindle text not null, 405 rkey text not null, 406 407 -- referenced pipeline. these form the (did, rkey) pair 408 pipeline_knot text not null, 409 pipeline_rkey text not null, 410 411 -- content 412 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 413 workflow text not null, 414 status text not null, 415 error text, 416 exit_code integer not null default 0, 417 418 unique (spindle, rkey), 419 foreign key (pipeline_knot, pipeline_rkey) 420 references pipelines (knot, rkey) 421 on delete cascade 422 ); 423 424 create table if not exists migrations ( 425 id integer primary key autoincrement, 426 name text unique 427 ); 428 `) 429 if err != nil { 430 return nil, err 431 } 432 433 // run migrations 434 runMigration(db, "add-description-to-repos", func(tx *sql.Tx) error { 435 tx.Exec(` 436 alter table repos add column description text check (length(description) <= 200); 437 `) 438 return nil 439 }) 440 441 runMigration(db, "add-rkey-to-pubkeys", func(tx *sql.Tx) error { 442 // add unconstrained column 443 _, err := tx.Exec(` 444 alter table public_keys 445 add column rkey text; 446 `) 447 if err != nil { 448 return err 449 } 450 451 // backfill 452 _, err = tx.Exec(` 453 update public_keys 454 set rkey = '' 455 where rkey is null; 456 `) 457 if err != nil { 458 return err 459 } 460 461 return nil 462 }) 463 464 runMigration(db, "add-rkey-to-comments", func(tx *sql.Tx) error { 465 _, err := tx.Exec(` 466 alter table comments drop column comment_at; 467 alter table comments add column rkey text; 468 `) 469 return err 470 }) 471 472 runMigration(db, "add-deleted-and-edited-to-issue-comments", func(tx *sql.Tx) error { 473 _, err := tx.Exec(` 474 alter table comments add column deleted text; -- timestamp 475 alter table comments add column edited text; -- timestamp 476 `) 477 return err 478 }) 479 480 runMigration(db, "add-source-info-to-pulls-and-submissions", func(tx *sql.Tx) error { 481 _, err := tx.Exec(` 482 alter table pulls add column source_branch text; 483 alter table pulls add column source_repo_at text; 484 alter table pull_submissions add column source_rev text; 485 `) 486 return err 487 }) 488 489 runMigration(db, "add-source-to-repos", func(tx *sql.Tx) error { 490 _, err := tx.Exec(` 491 alter table repos add column source text; 492 `) 493 return err 494 }) 495 496 // disable foreign-keys for the next migration 497 // NOTE: this cannot be done in a transaction, so it is run outside [0] 498 // 499 // [0]: https://sqlite.org/pragma.html#pragma_foreign_keys 500 db.Exec("pragma foreign_keys = off;") 501 runMigration(db, "recreate-pulls-column-for-stacking-support", func(tx *sql.Tx) error { 502 _, err := tx.Exec(` 503 create table pulls_new ( 504 -- identifiers 505 id integer primary key autoincrement, 506 pull_id integer not null, 507 508 -- at identifiers 509 repo_at text not null, 510 owner_did text not null, 511 rkey text not null, 512 513 -- content 514 title text not null, 515 body text not null, 516 target_branch text not null, 517 state integer not null default 0 check (state in (0, 1, 2, 3)), -- closed, open, merged, deleted 518 519 -- source info 520 source_branch text, 521 source_repo_at text, 522 523 -- stacking 524 stack_id text, 525 change_id text, 526 parent_change_id text, 527 528 -- meta 529 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 530 531 -- constraints 532 unique(repo_at, pull_id), 533 foreign key (repo_at) references repos(at_uri) on delete cascade 534 ); 535 536 insert into pulls_new ( 537 id, pull_id, 538 repo_at, owner_did, rkey, 539 title, body, target_branch, state, 540 source_branch, source_repo_at, 541 created 542 ) 543 select 544 id, pull_id, 545 repo_at, owner_did, rkey, 546 title, body, target_branch, state, 547 source_branch, source_repo_at, 548 created 549 FROM pulls; 550 551 drop table pulls; 552 alter table pulls_new rename to pulls; 553 `) 554 return err 555 }) 556 db.Exec("pragma foreign_keys = on;") 557 558 // run migrations 559 runMigration(db, "add-spindle-to-repos", func(tx *sql.Tx) error { 560 tx.Exec(` 561 alter table repos add column spindle text; 562 `) 563 return nil 564 }) 565 566 return &DB{db}, nil 567} 568 569type migrationFn = func(*sql.Tx) error 570 571func runMigration(d *sql.DB, name string, migrationFn migrationFn) error { 572 tx, err := d.Begin() 573 if err != nil { 574 return err 575 } 576 defer tx.Rollback() 577 578 var exists bool 579 err = tx.QueryRow("select exists (select 1 from migrations where name = ?)", name).Scan(&exists) 580 if err != nil { 581 return err 582 } 583 584 if !exists { 585 // run migration 586 err = migrationFn(tx) 587 if err != nil { 588 log.Printf("Failed to run migration %s: %v", name, err) 589 return err 590 } 591 592 // mark migration as complete 593 _, err = tx.Exec("insert into migrations (name) values (?)", name) 594 if err != nil { 595 log.Printf("Failed to mark migration %s as complete: %v", name, err) 596 return err 597 } 598 599 // commit the transaction 600 if err := tx.Commit(); err != nil { 601 return err 602 } 603 604 log.Printf("migration %s applied successfully", name) 605 } else { 606 log.Printf("skipped migration %s, already applied", name) 607 } 608 609 return nil 610} 611 612type filter struct { 613 key string 614 arg any 615 cmp string 616} 617 618func newFilter(key, cmp string, arg any) filter { 619 return filter{ 620 key: key, 621 arg: arg, 622 cmp: cmp, 623 } 624} 625 626func FilterEq(key string, arg any) filter { return newFilter(key, "=", arg) } 627func FilterNotEq(key string, arg any) filter { return newFilter(key, "<>", arg) } 628func FilterGte(key string, arg any) filter { return newFilter(key, ">=", arg) } 629func FilterLte(key string, arg any) filter { return newFilter(key, "<=", arg) } 630func FilterIs(key string, arg any) filter { return newFilter(key, "is", arg) } 631func FilterIsNot(key string, arg any) filter { return newFilter(key, "is not", arg) } 632func FilterIn(key string, arg any) filter { return newFilter(key, "in", arg) } 633 634func (f filter) Condition() string { 635 rv := reflect.ValueOf(f.arg) 636 kind := rv.Kind() 637 638 // if we have `FilterIn(k, [1, 2, 3])`, compile it down to `k in (?, ?, ?)` 639 if kind == reflect.Slice || kind == reflect.Array { 640 if rv.Len() == 0 { 641 panic(fmt.Sprintf("empty slice passed to %q filter on %s", f.cmp, f.key)) 642 } 643 644 placeholders := make([]string, rv.Len()) 645 for i := range placeholders { 646 placeholders[i] = "?" 647 } 648 649 return fmt.Sprintf("%s %s (%s)", f.key, f.cmp, strings.Join(placeholders, ", ")) 650 } 651 652 return fmt.Sprintf("%s %s ?", f.key, f.cmp) 653} 654 655func (f filter) Arg() []any { 656 rv := reflect.ValueOf(f.arg) 657 kind := rv.Kind() 658 if kind == reflect.Slice || kind == reflect.Array { 659 if rv.Len() == 0 { 660 panic(fmt.Sprintf("empty slice passed to %q filter on %s", f.cmp, f.key)) 661 } 662 663 out := make([]any, rv.Len()) 664 for i := range rv.Len() { 665 out[i] = rv.Index(i).Interface() 666 } 667 return out 668 } 669 670 return []any{f.arg} 671}