forked from tangled.org/core
Monorepo for Tangled — https://tangled.org
1package db 2 3import ( 4 "context" 5 "database/sql" 6 "fmt" 7 "log/slog" 8 "reflect" 9 "strings" 10 11 _ "github.com/mattn/go-sqlite3" 12 "tangled.org/core/log" 13) 14 15type DB struct { 16 *sql.DB 17 logger *slog.Logger 18} 19 20type Execer interface { 21 Query(query string, args ...any) (*sql.Rows, error) 22 QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) 23 QueryRow(query string, args ...any) *sql.Row 24 QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row 25 Exec(query string, args ...any) (sql.Result, error) 26 ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) 27 Prepare(query string) (*sql.Stmt, error) 28 PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) 29} 30 31func Make(ctx context.Context, dbPath string) (*DB, error) { 32 // https://github.com/mattn/go-sqlite3#connection-string 33 opts := []string{ 34 "_foreign_keys=1", 35 "_journal_mode=WAL", 36 "_synchronous=NORMAL", 37 "_auto_vacuum=incremental", 38 } 39 40 logger := log.FromContext(ctx) 41 logger = log.SubLogger(logger, "db") 42 43 db, err := sql.Open("sqlite3", dbPath+"?"+strings.Join(opts, "&")) 44 if err != nil { 45 return nil, err 46 } 47 48 conn, err := db.Conn(ctx) 49 if err != nil { 50 return nil, err 51 } 52 defer conn.Close() 53 54 _, err = conn.ExecContext(ctx, ` 55 create table if not exists registrations ( 56 id integer primary key autoincrement, 57 domain text not null unique, 58 did text not null, 59 secret text not null, 60 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 61 registered text 62 ); 63 create table if not exists public_keys ( 64 id integer primary key autoincrement, 65 did text not null, 66 name text not null, 67 key text not null, 68 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 69 unique(did, name, key) 70 ); 71 create table if not exists repos ( 72 id integer primary key autoincrement, 73 did text not null, 74 name text not null, 75 knot text not null, 76 rkey text not null, 77 at_uri text not null unique, 78 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 79 unique(did, name, knot, rkey) 80 ); 81 create table if not exists collaborators ( 82 id integer primary key autoincrement, 83 did text not null, 84 repo integer not null, 85 foreign key (repo) references repos(id) on delete cascade 86 ); 87 create table if not exists follows ( 88 user_did text not null, 89 subject_did text not null, 90 rkey text not null, 91 followed_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 92 primary key (user_did, subject_did), 93 check (user_did <> subject_did) 94 ); 95 create table if not exists issues ( 96 id integer primary key autoincrement, 97 owner_did text not null, 98 repo_at text not null, 99 issue_id integer not null, 100 title text not null, 101 body text not null, 102 open integer not null default 1, 103 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 104 issue_at text, 105 unique(repo_at, issue_id), 106 foreign key (repo_at) references repos(at_uri) on delete cascade 107 ); 108 create table if not exists comments ( 109 id integer primary key autoincrement, 110 owner_did text not null, 111 issue_id integer not null, 112 repo_at text not null, 113 comment_id integer not null, 114 comment_at text not null, 115 body text not null, 116 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 117 unique(issue_id, comment_id), 118 foreign key (repo_at, issue_id) references issues(repo_at, issue_id) on delete cascade 119 ); 120 create table if not exists pulls ( 121 -- identifiers 122 id integer primary key autoincrement, 123 pull_id integer not null, 124 125 -- at identifiers 126 repo_at text not null, 127 owner_did text not null, 128 rkey text not null, 129 pull_at text, 130 131 -- content 132 title text not null, 133 body text not null, 134 target_branch text not null, 135 state integer not null default 0 check (state in (0, 1, 2)), -- open, merged, closed 136 137 -- meta 138 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 139 140 -- constraints 141 unique(repo_at, pull_id), 142 foreign key (repo_at) references repos(at_uri) on delete cascade 143 ); 144 145 -- every pull must have atleast 1 submission: the initial submission 146 create table if not exists pull_submissions ( 147 -- identifiers 148 id integer primary key autoincrement, 149 pull_id integer not null, 150 151 -- at identifiers 152 repo_at text not null, 153 154 -- content, these are immutable, and require a resubmission to update 155 round_number integer not null default 0, 156 patch text, 157 158 -- meta 159 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 160 161 -- constraints 162 unique(repo_at, pull_id, round_number), 163 foreign key (repo_at, pull_id) references pulls(repo_at, pull_id) on delete cascade 164 ); 165 166 create table if not exists pull_comments ( 167 -- identifiers 168 id integer primary key autoincrement, 169 pull_id integer not null, 170 submission_id integer not null, 171 172 -- at identifiers 173 repo_at text not null, 174 owner_did text not null, 175 comment_at text not null, 176 177 -- content 178 body text not null, 179 180 -- meta 181 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 182 183 -- constraints 184 foreign key (repo_at, pull_id) references pulls(repo_at, pull_id) on delete cascade, 185 foreign key (submission_id) references pull_submissions(id) on delete cascade 186 ); 187 188 create table if not exists _jetstream ( 189 id integer primary key autoincrement, 190 last_time_us integer not null 191 ); 192 193 create table if not exists repo_issue_seqs ( 194 repo_at text primary key, 195 next_issue_id integer not null default 1 196 ); 197 198 create table if not exists repo_pull_seqs ( 199 repo_at text primary key, 200 next_pull_id integer not null default 1 201 ); 202 203 create table if not exists stars ( 204 id integer primary key autoincrement, 205 starred_by_did text not null, 206 repo_at text not null, 207 rkey text not null, 208 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 209 foreign key (repo_at) references repos(at_uri) on delete cascade, 210 unique(starred_by_did, repo_at) 211 ); 212 213 create table if not exists reactions ( 214 id integer primary key autoincrement, 215 reacted_by_did text not null, 216 thread_at text not null, 217 kind text not null, 218 rkey text not null, 219 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 220 unique(reacted_by_did, thread_at, kind) 221 ); 222 223 create table if not exists emails ( 224 id integer primary key autoincrement, 225 did text not null, 226 email text not null, 227 verified integer not null default 0, 228 verification_code text not null, 229 last_sent text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 230 is_primary integer not null default 0, 231 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 232 unique(did, email) 233 ); 234 235 create table if not exists artifacts ( 236 -- id 237 id integer primary key autoincrement, 238 did text not null, 239 rkey text not null, 240 241 -- meta 242 repo_at text not null, 243 tag binary(20) not null, 244 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 245 246 -- data 247 blob_cid text not null, 248 name text not null, 249 size integer not null default 0, 250 mimetype string not null default "*/*", 251 252 -- constraints 253 unique(did, rkey), -- record must be unique 254 unique(repo_at, tag, name), -- for a given tag object, each file must be unique 255 foreign key (repo_at) references repos(at_uri) on delete cascade 256 ); 257 258 create table if not exists profile ( 259 -- id 260 id integer primary key autoincrement, 261 did text not null, 262 263 -- data 264 description text not null, 265 include_bluesky integer not null default 0, 266 location text, 267 268 -- constraints 269 unique(did) 270 ); 271 create table if not exists profile_links ( 272 -- id 273 id integer primary key autoincrement, 274 did text not null, 275 276 -- data 277 link text not null, 278 279 -- constraints 280 foreign key (did) references profile(did) on delete cascade 281 ); 282 create table if not exists profile_stats ( 283 -- id 284 id integer primary key autoincrement, 285 did text not null, 286 287 -- data 288 kind text not null check (kind in ( 289 "merged-pull-request-count", 290 "closed-pull-request-count", 291 "open-pull-request-count", 292 "open-issue-count", 293 "closed-issue-count", 294 "repository-count" 295 )), 296 297 -- constraints 298 foreign key (did) references profile(did) on delete cascade 299 ); 300 create table if not exists profile_pinned_repositories ( 301 -- id 302 id integer primary key autoincrement, 303 did text not null, 304 305 -- data 306 at_uri text not null, 307 308 -- constraints 309 unique(did, at_uri), 310 foreign key (did) references profile(did) on delete cascade, 311 foreign key (at_uri) references repos(at_uri) on delete cascade 312 ); 313 314 create table if not exists oauth_requests ( 315 id integer primary key autoincrement, 316 auth_server_iss text not null, 317 state text not null, 318 did text not null, 319 handle text not null, 320 pds_url text not null, 321 pkce_verifier text not null, 322 dpop_auth_server_nonce text not null, 323 dpop_private_jwk text not null 324 ); 325 326 create table if not exists oauth_sessions ( 327 id integer primary key autoincrement, 328 did text not null, 329 handle text not null, 330 pds_url text not null, 331 auth_server_iss text not null, 332 access_jwt text not null, 333 refresh_jwt text not null, 334 dpop_pds_nonce text, 335 dpop_auth_server_nonce text not null, 336 dpop_private_jwk text not null, 337 expiry text not null 338 ); 339 340 create table if not exists punchcard ( 341 did text not null, 342 date text not null, -- yyyy-mm-dd 343 count integer, 344 primary key (did, date) 345 ); 346 347 create table if not exists spindles ( 348 id integer primary key autoincrement, 349 owner text not null, 350 instance text not null, 351 verified text, -- time of verification 352 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 353 354 unique(owner, instance) 355 ); 356 357 create table if not exists spindle_members ( 358 -- identifiers for the record 359 id integer primary key autoincrement, 360 did text not null, 361 rkey text not null, 362 363 -- data 364 instance text not null, 365 subject text not null, 366 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 367 368 -- constraints 369 unique (did, instance, subject) 370 ); 371 372 create table if not exists pipelines ( 373 -- identifiers 374 id integer primary key autoincrement, 375 knot text not null, 376 rkey text not null, 377 378 repo_owner text not null, 379 repo_name text not null, 380 381 -- every pipeline must be associated with exactly one commit 382 sha text not null check (length(sha) = 40), 383 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 384 385 -- trigger data 386 trigger_id integer not null, 387 388 unique(knot, rkey), 389 foreign key (trigger_id) references triggers(id) on delete cascade 390 ); 391 392 create table if not exists triggers ( 393 -- primary key 394 id integer primary key autoincrement, 395 396 -- top-level fields 397 kind text not null, 398 399 -- pushTriggerData fields 400 push_ref text, 401 push_new_sha text check (length(push_new_sha) = 40), 402 push_old_sha text check (length(push_old_sha) = 40), 403 404 -- pullRequestTriggerData fields 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 409 ); 410 411 create table if not exists pipeline_statuses ( 412 -- identifiers 413 id integer primary key autoincrement, 414 spindle text not null, 415 rkey text not null, 416 417 -- referenced pipeline. these form the (did, rkey) pair 418 pipeline_knot text not null, 419 pipeline_rkey text not null, 420 421 -- content 422 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 423 workflow text not null, 424 status text not null, 425 error text, 426 exit_code integer not null default 0, 427 428 unique (spindle, rkey), 429 foreign key (pipeline_knot, pipeline_rkey) 430 references pipelines (knot, rkey) 431 on delete cascade 432 ); 433 434 create table if not exists repo_languages ( 435 -- identifiers 436 id integer primary key autoincrement, 437 438 -- repo identifiers 439 repo_at text not null, 440 ref text not null, 441 is_default_ref integer not null default 0, 442 443 -- language breakdown 444 language text not null, 445 bytes integer not null check (bytes >= 0), 446 447 unique(repo_at, ref, language) 448 ); 449 450 create table if not exists signups_inflight ( 451 id integer primary key autoincrement, 452 email text not null unique, 453 invite_code text not null, 454 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) 455 ); 456 457 create table if not exists strings ( 458 -- identifiers 459 did text not null, 460 rkey text not null, 461 462 -- content 463 filename text not null, 464 description text, 465 content text not null, 466 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 467 edited text, 468 469 primary key (did, rkey) 470 ); 471 472 create table if not exists label_definitions ( 473 -- identifiers 474 id integer primary key autoincrement, 475 did text not null, 476 rkey text not null, 477 at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.label.definition' || '/' || rkey) stored, 478 479 -- content 480 name text not null, 481 value_type text not null check (value_type in ( 482 "null", 483 "boolean", 484 "integer", 485 "string" 486 )), 487 value_format text not null default "any", 488 value_enum text, -- comma separated list 489 scope text not null, -- comma separated list of nsid 490 color text, 491 multiple integer not null default 0, 492 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 493 494 -- constraints 495 unique (did, rkey) 496 unique (at_uri) 497 ); 498 499 -- ops are flattened, a record may contain several additions and deletions, but the table will include one row per add/del 500 create table if not exists label_ops ( 501 -- identifiers 502 id integer primary key autoincrement, 503 did text not null, 504 rkey text not null, 505 at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.label.op' || '/' || rkey) stored, 506 507 -- content 508 subject text not null, 509 operation text not null check (operation in ("add", "del")), 510 operand_key text not null, 511 operand_value text not null, 512 -- we need two time values: performed is declared by the user, indexed is calculated by the av 513 performed text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 514 indexed text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 515 516 -- constraints 517 -- traditionally (did, rkey) pair should be unique, but not in this case 518 -- operand_key should reference a label definition 519 foreign key (operand_key) references label_definitions (at_uri) on delete cascade, 520 unique (did, rkey, subject, operand_key, operand_value) 521 ); 522 523 create table if not exists repo_labels ( 524 -- identifiers 525 id integer primary key autoincrement, 526 527 -- repo identifiers 528 repo_at text not null, 529 530 -- label to subscribe to 531 label_at text not null, 532 533 unique (repo_at, label_at) 534 ); 535 536 create table if not exists notifications ( 537 id integer primary key autoincrement, 538 recipient_did text not null, 539 actor_did text not null, 540 type text not null, 541 entity_type text not null, 542 entity_id text not null, 543 read integer not null default 0, 544 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 545 repo_id integer references repos(id), 546 issue_id integer references issues(id), 547 pull_id integer references pulls(id) 548 ); 549 550 create table if not exists notification_preferences ( 551 id integer primary key autoincrement, 552 user_did text not null unique, 553 repo_starred integer not null default 1, 554 issue_created integer not null default 1, 555 issue_commented integer not null default 1, 556 pull_created integer not null default 1, 557 pull_commented integer not null default 1, 558 followed integer not null default 1, 559 pull_merged integer not null default 1, 560 issue_closed integer not null default 1, 561 email_notifications integer not null default 0 562 ); 563 564 create table if not exists migrations ( 565 id integer primary key autoincrement, 566 name text unique 567 ); 568 569 -- indexes for better performance 570 create index if not exists idx_notifications_recipient_created on notifications(recipient_did, created desc); 571 create index if not exists idx_notifications_recipient_read on notifications(recipient_did, read); 572 create index if not exists idx_stars_created on stars(created); 573 create index if not exists idx_stars_repo_at_created on stars(repo_at, created); 574 `) 575 if err != nil { 576 return nil, err 577 } 578 579 // run migrations 580 runMigration(conn, logger, "add-description-to-repos", func(tx *sql.Tx) error { 581 tx.Exec(` 582 alter table repos add column description text check (length(description) <= 200); 583 `) 584 return nil 585 }) 586 587 runMigration(conn, logger, "add-rkey-to-pubkeys", func(tx *sql.Tx) error { 588 // add unconstrained column 589 _, err := tx.Exec(` 590 alter table public_keys 591 add column rkey text; 592 `) 593 if err != nil { 594 return err 595 } 596 597 // backfill 598 _, err = tx.Exec(` 599 update public_keys 600 set rkey = '' 601 where rkey is null; 602 `) 603 if err != nil { 604 return err 605 } 606 607 return nil 608 }) 609 610 runMigration(conn, logger, "add-rkey-to-comments", func(tx *sql.Tx) error { 611 _, err := tx.Exec(` 612 alter table comments drop column comment_at; 613 alter table comments add column rkey text; 614 `) 615 return err 616 }) 617 618 runMigration(conn, logger, "add-deleted-and-edited-to-issue-comments", func(tx *sql.Tx) error { 619 _, err := tx.Exec(` 620 alter table comments add column deleted text; -- timestamp 621 alter table comments add column edited text; -- timestamp 622 `) 623 return err 624 }) 625 626 runMigration(conn, logger, "add-source-info-to-pulls-and-submissions", func(tx *sql.Tx) error { 627 _, err := tx.Exec(` 628 alter table pulls add column source_branch text; 629 alter table pulls add column source_repo_at text; 630 alter table pull_submissions add column source_rev text; 631 `) 632 return err 633 }) 634 635 runMigration(conn, logger, "add-source-to-repos", func(tx *sql.Tx) error { 636 _, err := tx.Exec(` 637 alter table repos add column source text; 638 `) 639 return err 640 }) 641 642 // disable foreign-keys for the next migration 643 // NOTE: this cannot be done in a transaction, so it is run outside [0] 644 // 645 // [0]: https://sqlite.org/pragma.html#pragma_foreign_keys 646 conn.ExecContext(ctx, "pragma foreign_keys = off;") 647 runMigration(conn, logger, "recreate-pulls-column-for-stacking-support", func(tx *sql.Tx) error { 648 _, err := tx.Exec(` 649 create table pulls_new ( 650 -- identifiers 651 id integer primary key autoincrement, 652 pull_id integer not null, 653 654 -- at identifiers 655 repo_at text not null, 656 owner_did text not null, 657 rkey text not null, 658 659 -- content 660 title text not null, 661 body text not null, 662 target_branch text not null, 663 state integer not null default 0 check (state in (0, 1, 2, 3)), -- closed, open, merged, deleted 664 665 -- source info 666 source_branch text, 667 source_repo_at text, 668 669 -- stacking 670 stack_id text, 671 change_id text, 672 parent_change_id text, 673 674 -- meta 675 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 676 677 -- constraints 678 unique(repo_at, pull_id), 679 foreign key (repo_at) references repos(at_uri) on delete cascade 680 ); 681 682 insert into pulls_new ( 683 id, pull_id, 684 repo_at, owner_did, rkey, 685 title, body, target_branch, state, 686 source_branch, source_repo_at, 687 created 688 ) 689 select 690 id, pull_id, 691 repo_at, owner_did, rkey, 692 title, body, target_branch, state, 693 source_branch, source_repo_at, 694 created 695 FROM pulls; 696 697 drop table pulls; 698 alter table pulls_new rename to pulls; 699 `) 700 return err 701 }) 702 conn.ExecContext(ctx, "pragma foreign_keys = on;") 703 704 runMigration(conn, logger, "add-spindle-to-repos", func(tx *sql.Tx) error { 705 tx.Exec(` 706 alter table repos add column spindle text; 707 `) 708 return nil 709 }) 710 711 // drop all knot secrets, add unique constraint to knots 712 // 713 // knots will henceforth use service auth for signed requests 714 runMigration(conn, logger, "no-more-secrets", func(tx *sql.Tx) error { 715 _, err := tx.Exec(` 716 create table registrations_new ( 717 id integer primary key autoincrement, 718 domain text not null, 719 did text not null, 720 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 721 registered text, 722 read_only integer not null default 0, 723 unique(domain, did) 724 ); 725 726 insert into registrations_new (id, domain, did, created, registered, read_only) 727 select id, domain, did, created, registered, 1 from registrations 728 where registered is not null; 729 730 drop table registrations; 731 alter table registrations_new rename to registrations; 732 `) 733 return err 734 }) 735 736 // recreate and add rkey + created columns with default constraint 737 runMigration(conn, logger, "rework-collaborators-table", func(tx *sql.Tx) error { 738 // create new table 739 // - repo_at instead of repo integer 740 // - rkey field 741 // - created field 742 _, err := tx.Exec(` 743 create table collaborators_new ( 744 -- identifiers for the record 745 id integer primary key autoincrement, 746 did text not null, 747 rkey text, 748 749 -- content 750 subject_did text not null, 751 repo_at text not null, 752 753 -- meta 754 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 755 756 -- constraints 757 foreign key (repo_at) references repos(at_uri) on delete cascade 758 ) 759 `) 760 if err != nil { 761 return err 762 } 763 764 // copy data 765 _, err = tx.Exec(` 766 insert into collaborators_new (id, did, rkey, subject_did, repo_at) 767 select 768 c.id, 769 r.did, 770 '', 771 c.did, 772 r.at_uri 773 from collaborators c 774 join repos r on c.repo = r.id 775 `) 776 if err != nil { 777 return err 778 } 779 780 // drop old table 781 _, err = tx.Exec(`drop table collaborators`) 782 if err != nil { 783 return err 784 } 785 786 // rename new table 787 _, err = tx.Exec(`alter table collaborators_new rename to collaborators`) 788 return err 789 }) 790 791 runMigration(conn, logger, "add-rkey-to-issues", func(tx *sql.Tx) error { 792 _, err := tx.Exec(` 793 alter table issues add column rkey text not null default ''; 794 795 -- get last url section from issue_at and save to rkey column 796 update issues 797 set rkey = replace(issue_at, rtrim(issue_at, replace(issue_at, '/', '')), ''); 798 `) 799 return err 800 }) 801 802 // repurpose the read-only column to "needs-upgrade" 803 runMigration(conn, logger, "rename-registrations-read-only-to-needs-upgrade", func(tx *sql.Tx) error { 804 _, err := tx.Exec(` 805 alter table registrations rename column read_only to needs_upgrade; 806 `) 807 return err 808 }) 809 810 // require all knots to upgrade after the release of total xrpc 811 runMigration(conn, logger, "migrate-knots-to-total-xrpc", func(tx *sql.Tx) error { 812 _, err := tx.Exec(` 813 update registrations set needs_upgrade = 1; 814 `) 815 return err 816 }) 817 818 // require all knots to upgrade after the release of total xrpc 819 runMigration(conn, logger, "migrate-spindles-to-xrpc-owner", func(tx *sql.Tx) error { 820 _, err := tx.Exec(` 821 alter table spindles add column needs_upgrade integer not null default 0; 822 `) 823 return err 824 }) 825 826 // remove issue_at from issues and replace with generated column 827 // 828 // this requires a full table recreation because stored columns 829 // cannot be added via alter 830 // 831 // couple other changes: 832 // - columns renamed to be more consistent 833 // - adds edited and deleted fields 834 // 835 // disable foreign-keys for the next migration 836 conn.ExecContext(ctx, "pragma foreign_keys = off;") 837 runMigration(conn, logger, "remove-issue-at-from-issues", func(tx *sql.Tx) error { 838 _, err := tx.Exec(` 839 create table if not exists issues_new ( 840 -- identifiers 841 id integer primary key autoincrement, 842 did text not null, 843 rkey text not null, 844 at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.repo.issue' || '/' || rkey) stored, 845 846 -- at identifiers 847 repo_at text not null, 848 849 -- content 850 issue_id integer not null, 851 title text not null, 852 body text not null, 853 open integer not null default 1, 854 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 855 edited text, -- timestamp 856 deleted text, -- timestamp 857 858 unique(did, rkey), 859 unique(repo_at, issue_id), 860 unique(at_uri), 861 foreign key (repo_at) references repos(at_uri) on delete cascade 862 ); 863 `) 864 if err != nil { 865 return err 866 } 867 868 // transfer data 869 _, err = tx.Exec(` 870 insert into issues_new (id, did, rkey, repo_at, issue_id, title, body, open, created) 871 select 872 i.id, 873 i.owner_did, 874 i.rkey, 875 i.repo_at, 876 i.issue_id, 877 i.title, 878 i.body, 879 i.open, 880 i.created 881 from issues i; 882 `) 883 if err != nil { 884 return err 885 } 886 887 // drop old table 888 _, err = tx.Exec(`drop table issues`) 889 if err != nil { 890 return err 891 } 892 893 // rename new table 894 _, err = tx.Exec(`alter table issues_new rename to issues`) 895 return err 896 }) 897 conn.ExecContext(ctx, "pragma foreign_keys = on;") 898 899 // - renames the comments table to 'issue_comments' 900 // - rework issue comments to update constraints: 901 // * unique(did, rkey) 902 // * remove comment-id and just use the global ID 903 // * foreign key (repo_at, issue_id) 904 // - new columns 905 // * column "reply_to" which can be any other comment 906 // * column "at-uri" which is a generated column 907 runMigration(conn, logger, "rework-issue-comments", func(tx *sql.Tx) error { 908 _, err := tx.Exec(` 909 create table if not exists issue_comments ( 910 -- identifiers 911 id integer primary key autoincrement, 912 did text not null, 913 rkey text, 914 at_uri text generated always as ('at://' || did || '/' || 'sh.tangled.repo.issue.comment' || '/' || rkey) stored, 915 916 -- at identifiers 917 issue_at text not null, 918 reply_to text, -- at_uri of parent comment 919 920 -- content 921 body text not null, 922 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 923 edited text, 924 deleted text, 925 926 -- constraints 927 unique(did, rkey), 928 unique(at_uri), 929 foreign key (issue_at) references issues(at_uri) on delete cascade 930 ); 931 `) 932 if err != nil { 933 return err 934 } 935 936 // transfer data 937 _, err = tx.Exec(` 938 insert into issue_comments (id, did, rkey, issue_at, body, created, edited, deleted) 939 select 940 c.id, 941 c.owner_did, 942 c.rkey, 943 i.at_uri, -- get at_uri from issues table 944 c.body, 945 c.created, 946 c.edited, 947 c.deleted 948 from comments c 949 join issues i on c.repo_at = i.repo_at and c.issue_id = i.issue_id; 950 `) 951 if err != nil { 952 return err 953 } 954 955 // drop old table 956 _, err = tx.Exec(`drop table comments`) 957 return err 958 }) 959 960 // add generated at_uri column to pulls table 961 // 962 // this requires a full table recreation because stored columns 963 // cannot be added via alter 964 // 965 // disable foreign-keys for the next migration 966 conn.ExecContext(ctx, "pragma foreign_keys = off;") 967 runMigration(conn, logger, "add-at-uri-to-pulls", func(tx *sql.Tx) error { 968 _, err := tx.Exec(` 969 create table if not exists pulls_new ( 970 -- identifiers 971 id integer primary key autoincrement, 972 pull_id integer not null, 973 at_uri text generated always as ('at://' || owner_did || '/' || 'sh.tangled.repo.pull' || '/' || rkey) stored, 974 975 -- at identifiers 976 repo_at text not null, 977 owner_did text not null, 978 rkey text not null, 979 980 -- content 981 title text not null, 982 body text not null, 983 target_branch text not null, 984 state integer not null default 0 check (state in (0, 1, 2, 3)), -- closed, open, merged, deleted 985 986 -- source info 987 source_branch text, 988 source_repo_at text, 989 990 -- stacking 991 stack_id text, 992 change_id text, 993 parent_change_id text, 994 995 -- meta 996 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 997 998 -- constraints 999 unique(repo_at, pull_id), 1000 unique(at_uri), 1001 foreign key (repo_at) references repos(at_uri) on delete cascade 1002 ); 1003 `) 1004 if err != nil { 1005 return err 1006 } 1007 1008 // transfer data 1009 _, err = tx.Exec(` 1010 insert into pulls_new ( 1011 id, pull_id, repo_at, owner_did, rkey, 1012 title, body, target_branch, state, 1013 source_branch, source_repo_at, 1014 stack_id, change_id, parent_change_id, 1015 created 1016 ) 1017 select 1018 id, pull_id, repo_at, owner_did, rkey, 1019 title, body, target_branch, state, 1020 source_branch, source_repo_at, 1021 stack_id, change_id, parent_change_id, 1022 created 1023 from pulls; 1024 `) 1025 if err != nil { 1026 return err 1027 } 1028 1029 // drop old table 1030 _, err = tx.Exec(`drop table pulls`) 1031 if err != nil { 1032 return err 1033 } 1034 1035 // rename new table 1036 _, err = tx.Exec(`alter table pulls_new rename to pulls`) 1037 return err 1038 }) 1039 conn.ExecContext(ctx, "pragma foreign_keys = on;") 1040 1041 // remove repo_at and pull_id from pull_submissions and replace with pull_at 1042 // 1043 // this requires a full table recreation because stored columns 1044 // cannot be added via alter 1045 // 1046 // disable foreign-keys for the next migration 1047 conn.ExecContext(ctx, "pragma foreign_keys = off;") 1048 runMigration(conn, logger, "remove-repo-at-pull-id-from-pull-submissions", func(tx *sql.Tx) error { 1049 _, err := tx.Exec(` 1050 create table if not exists pull_submissions_new ( 1051 -- identifiers 1052 id integer primary key autoincrement, 1053 pull_at text not null, 1054 1055 -- content, these are immutable, and require a resubmission to update 1056 round_number integer not null default 0, 1057 patch text, 1058 source_rev text, 1059 1060 -- meta 1061 created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), 1062 1063 -- constraints 1064 unique(pull_at, round_number), 1065 foreign key (pull_at) references pulls(at_uri) on delete cascade 1066 ); 1067 `) 1068 if err != nil { 1069 return err 1070 } 1071 1072 // transfer data, constructing pull_at from pulls table 1073 _, err = tx.Exec(` 1074 insert into pull_submissions_new (id, pull_at, round_number, patch, created) 1075 select 1076 ps.id, 1077 'at://' || p.owner_did || '/sh.tangled.repo.pull/' || p.rkey, 1078 ps.round_number, 1079 ps.patch, 1080 ps.created 1081 from pull_submissions ps 1082 join pulls p on ps.repo_at = p.repo_at and ps.pull_id = p.pull_id; 1083 `) 1084 if err != nil { 1085 return err 1086 } 1087 1088 // drop old table 1089 _, err = tx.Exec(`drop table pull_submissions`) 1090 if err != nil { 1091 return err 1092 } 1093 1094 // rename new table 1095 _, err = tx.Exec(`alter table pull_submissions_new rename to pull_submissions`) 1096 return err 1097 }) 1098 conn.ExecContext(ctx, "pragma foreign_keys = on;") 1099 1100 // knots may report the combined patch for a comparison, we can store that on the appview side 1101 // (but not on the pds record), because calculating the combined patch requires a git index 1102 runMigration(conn, logger, "add-combined-column-submissions", func(tx *sql.Tx) error { 1103 _, err := tx.Exec(` 1104 alter table pull_submissions add column combined text; 1105 `) 1106 return err 1107 }) 1108 1109 runMigration(conn, logger, "add-pronouns-profile", func(tx *sql.Tx) error { 1110 _, err := tx.Exec(` 1111 alter table profile add column pronouns text; 1112 `) 1113 return err 1114 }) 1115 1116 runMigration(conn, logger, "add-meta-column-repos", func(tx *sql.Tx) error { 1117 _, err := tx.Exec(` 1118 alter table repos add column website text; 1119 alter table repos add column topics text; 1120 `) 1121 return err 1122 }) 1123 1124 runMigration(conn, logger, "add-usermentioned-preference", func(tx *sql.Tx) error { 1125 _, err := tx.Exec(` 1126 alter table notification_preferences add column user_mentioned integer not null default 1; 1127 `) 1128 return err 1129 }) 1130 1131 return &DB{ 1132 db, 1133 logger, 1134 }, nil 1135} 1136 1137type migrationFn = func(*sql.Tx) error 1138 1139func runMigration(c *sql.Conn, logger *slog.Logger, name string, migrationFn migrationFn) error { 1140 logger = logger.With("migration", name) 1141 1142 tx, err := c.BeginTx(context.Background(), nil) 1143 if err != nil { 1144 return err 1145 } 1146 defer tx.Rollback() 1147 1148 var exists bool 1149 err = tx.QueryRow("select exists (select 1 from migrations where name = ?)", name).Scan(&exists) 1150 if err != nil { 1151 return err 1152 } 1153 1154 if !exists { 1155 // run migration 1156 err = migrationFn(tx) 1157 if err != nil { 1158 logger.Error("failed to run migration", "err", err) 1159 return err 1160 } 1161 1162 // mark migration as complete 1163 _, err = tx.Exec("insert into migrations (name) values (?)", name) 1164 if err != nil { 1165 logger.Error("failed to mark migration as complete", "err", err) 1166 return err 1167 } 1168 1169 // commit the transaction 1170 if err := tx.Commit(); err != nil { 1171 return err 1172 } 1173 1174 logger.Info("migration applied successfully") 1175 } else { 1176 logger.Warn("skipped migration, already applied") 1177 } 1178 1179 return nil 1180} 1181 1182func (d *DB) Close() error { 1183 return d.DB.Close() 1184} 1185 1186type filter struct { 1187 key string 1188 arg any 1189 cmp string 1190} 1191 1192func newFilter(key, cmp string, arg any) filter { 1193 return filter{ 1194 key: key, 1195 arg: arg, 1196 cmp: cmp, 1197 } 1198} 1199 1200func FilterEq(key string, arg any) filter { return newFilter(key, "=", arg) } 1201func FilterNotEq(key string, arg any) filter { return newFilter(key, "<>", arg) } 1202func FilterGte(key string, arg any) filter { return newFilter(key, ">=", arg) } 1203func FilterLte(key string, arg any) filter { return newFilter(key, "<=", arg) } 1204func FilterIs(key string, arg any) filter { return newFilter(key, "is", arg) } 1205func FilterIsNot(key string, arg any) filter { return newFilter(key, "is not", arg) } 1206func FilterIn(key string, arg any) filter { return newFilter(key, "in", arg) } 1207func FilterLike(key string, arg any) filter { return newFilter(key, "like", arg) } 1208func FilterNotLike(key string, arg any) filter { return newFilter(key, "not like", arg) } 1209func FilterContains(key string, arg any) filter { 1210 return newFilter(key, "like", fmt.Sprintf("%%%v%%", arg)) 1211} 1212 1213func (f filter) Condition() string { 1214 rv := reflect.ValueOf(f.arg) 1215 kind := rv.Kind() 1216 1217 // if we have `FilterIn(k, [1, 2, 3])`, compile it down to `k in (?, ?, ?)` 1218 if (kind == reflect.Slice && rv.Type().Elem().Kind() != reflect.Uint8) || kind == reflect.Array { 1219 if rv.Len() == 0 { 1220 // always false 1221 return "1 = 0" 1222 } 1223 1224 placeholders := make([]string, rv.Len()) 1225 for i := range placeholders { 1226 placeholders[i] = "?" 1227 } 1228 1229 return fmt.Sprintf("%s %s (%s)", f.key, f.cmp, strings.Join(placeholders, ", ")) 1230 } 1231 1232 return fmt.Sprintf("%s %s ?", f.key, f.cmp) 1233} 1234 1235func (f filter) Arg() []any { 1236 rv := reflect.ValueOf(f.arg) 1237 kind := rv.Kind() 1238 if (kind == reflect.Slice && rv.Type().Elem().Kind() != reflect.Uint8) || kind == reflect.Array { 1239 if rv.Len() == 0 { 1240 return nil 1241 } 1242 1243 out := make([]any, rv.Len()) 1244 for i := range rv.Len() { 1245 out[i] = rv.Index(i).Interface() 1246 } 1247 return out 1248 } 1249 1250 return []any{f.arg} 1251}