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