forked from tangled.org/core
Monorepo for Tangled — https://tangled.org
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 repo_languages ( 425 -- identifiers 426 id integer primary key autoincrement, 427 428 -- repo identifiers 429 repo_at text not null, 430 ref text not null, 431 is_default_ref integer not null default 0, 432 433 -- language breakdown 434 language text not null, 435 bytes integer not null check (bytes >= 0), 436 437 unique(repo_at, ref, language) 438 ); 439 440 create table if not exists migrations ( 441 id integer primary key autoincrement, 442 name text unique 443 ); 444 `) 445 if err != nil { 446 return nil, err 447 } 448 449 // run migrations 450 runMigration(db, "add-description-to-repos", func(tx *sql.Tx) error { 451 tx.Exec(` 452 alter table repos add column description text check (length(description) <= 200); 453 `) 454 return nil 455 }) 456 457 runMigration(db, "add-rkey-to-pubkeys", func(tx *sql.Tx) error { 458 // add unconstrained column 459 _, err := tx.Exec(` 460 alter table public_keys 461 add column rkey text; 462 `) 463 if err != nil { 464 return err 465 } 466 467 // backfill 468 _, err = tx.Exec(` 469 update public_keys 470 set rkey = '' 471 where rkey is null; 472 `) 473 if err != nil { 474 return err 475 } 476 477 return nil 478 }) 479 480 runMigration(db, "add-rkey-to-comments", func(tx *sql.Tx) error { 481 _, err := tx.Exec(` 482 alter table comments drop column comment_at; 483 alter table comments add column rkey text; 484 `) 485 return err 486 }) 487 488 runMigration(db, "add-deleted-and-edited-to-issue-comments", func(tx *sql.Tx) error { 489 _, err := tx.Exec(` 490 alter table comments add column deleted text; -- timestamp 491 alter table comments add column edited text; -- timestamp 492 `) 493 return err 494 }) 495 496 runMigration(db, "add-source-info-to-pulls-and-submissions", func(tx *sql.Tx) error { 497 _, err := tx.Exec(` 498 alter table pulls add column source_branch text; 499 alter table pulls add column source_repo_at text; 500 alter table pull_submissions add column source_rev text; 501 `) 502 return err 503 }) 504 505 runMigration(db, "add-source-to-repos", func(tx *sql.Tx) error { 506 _, err := tx.Exec(` 507 alter table repos add column source text; 508 `) 509 return err 510 }) 511 512 // disable foreign-keys for the next migration 513 // NOTE: this cannot be done in a transaction, so it is run outside [0] 514 // 515 // [0]: https://sqlite.org/pragma.html#pragma_foreign_keys 516 db.Exec("pragma foreign_keys = off;") 517 runMigration(db, "recreate-pulls-column-for-stacking-support", func(tx *sql.Tx) error { 518 _, err := tx.Exec(` 519 create table pulls_new ( 520 -- identifiers 521 id integer primary key autoincrement, 522 pull_id integer not null, 523 524 -- at identifiers 525 repo_at text not null, 526 owner_did text not null, 527 rkey text not null, 528 529 -- content 530 title text not null, 531 body text not null, 532 target_branch text not null, 533 state integer not null default 0 check (state in (0, 1, 2, 3)), -- closed, open, merged, deleted 534 535 -- source info 536 source_branch text, 537 source_repo_at text, 538 539 -- stacking 540 stack_id text, 541 change_id text, 542 parent_change_id text, 543 544 -- meta 545 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 546 547 -- constraints 548 unique(repo_at, pull_id), 549 foreign key (repo_at) references repos(at_uri) on delete cascade 550 ); 551 552 insert into pulls_new ( 553 id, pull_id, 554 repo_at, owner_did, rkey, 555 title, body, target_branch, state, 556 source_branch, source_repo_at, 557 created 558 ) 559 select 560 id, pull_id, 561 repo_at, owner_did, rkey, 562 title, body, target_branch, state, 563 source_branch, source_repo_at, 564 created 565 FROM pulls; 566 567 drop table pulls; 568 alter table pulls_new rename to pulls; 569 `) 570 return err 571 }) 572 db.Exec("pragma foreign_keys = on;") 573 574 // run migrations 575 runMigration(db, "add-spindle-to-repos", func(tx *sql.Tx) error { 576 tx.Exec(` 577 alter table repos add column spindle text; 578 `) 579 return nil 580 }) 581 582 return &DB{db}, nil 583} 584 585type migrationFn = func(*sql.Tx) error 586 587func runMigration(d *sql.DB, name string, migrationFn migrationFn) error { 588 tx, err := d.Begin() 589 if err != nil { 590 return err 591 } 592 defer tx.Rollback() 593 594 var exists bool 595 err = tx.QueryRow("select exists (select 1 from migrations where name = ?)", name).Scan(&exists) 596 if err != nil { 597 return err 598 } 599 600 if !exists { 601 // run migration 602 err = migrationFn(tx) 603 if err != nil { 604 log.Printf("Failed to run migration %s: %v", name, err) 605 return err 606 } 607 608 // mark migration as complete 609 _, err = tx.Exec("insert into migrations (name) values (?)", name) 610 if err != nil { 611 log.Printf("Failed to mark migration %s as complete: %v", name, err) 612 return err 613 } 614 615 // commit the transaction 616 if err := tx.Commit(); err != nil { 617 return err 618 } 619 620 log.Printf("migration %s applied successfully", name) 621 } else { 622 log.Printf("skipped migration %s, already applied", name) 623 } 624 625 return nil 626} 627 628type filter struct { 629 key string 630 arg any 631 cmp string 632} 633 634func newFilter(key, cmp string, arg any) filter { 635 return filter{ 636 key: key, 637 arg: arg, 638 cmp: cmp, 639 } 640} 641 642func FilterEq(key string, arg any) filter { return newFilter(key, "=", arg) } 643func FilterNotEq(key string, arg any) filter { return newFilter(key, "<>", arg) } 644func FilterGte(key string, arg any) filter { return newFilter(key, ">=", arg) } 645func FilterLte(key string, arg any) filter { return newFilter(key, "<=", arg) } 646func FilterIs(key string, arg any) filter { return newFilter(key, "is", arg) } 647func FilterIsNot(key string, arg any) filter { return newFilter(key, "is not", arg) } 648func FilterIn(key string, arg any) filter { return newFilter(key, "in", arg) } 649 650func (f filter) Condition() string { 651 rv := reflect.ValueOf(f.arg) 652 kind := rv.Kind() 653 654 // if we have `FilterIn(k, [1, 2, 3])`, compile it down to `k in (?, ?, ?)` 655 if kind == reflect.Slice || kind == reflect.Array { 656 if rv.Len() == 0 { 657 panic(fmt.Sprintf("empty slice passed to %q filter on %s", f.cmp, f.key)) 658 } 659 660 placeholders := make([]string, rv.Len()) 661 for i := range placeholders { 662 placeholders[i] = "?" 663 } 664 665 return fmt.Sprintf("%s %s (%s)", f.key, f.cmp, strings.Join(placeholders, ", ")) 666 } 667 668 return fmt.Sprintf("%s %s ?", f.key, f.cmp) 669} 670 671func (f filter) Arg() []any { 672 rv := reflect.ValueOf(f.arg) 673 kind := rv.Kind() 674 if kind == reflect.Slice || kind == reflect.Array { 675 if rv.Len() == 0 { 676 panic(fmt.Sprintf("empty slice passed to %q filter on %s", f.cmp, f.key)) 677 } 678 679 out := make([]any, rv.Len()) 680 for i := range rv.Len() { 681 out[i] = rv.Index(i).Interface() 682 } 683 return out 684 } 685 686 return []any{f.arg} 687}