From 1572fb3b9ffe6c21c912c9b931701edfd13a8019 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 15:33:31 -0700 Subject: [PATCH 1/2] feat(store): preserve child tombstones --- CHANGELOG.md | 2 + docs/maintainer-archive.md | 2 + internal/cli/app_test.go | 6 +- internal/store/comments.go | 147 ++++++++++- internal/store/comments_test.go | 101 ++++++++ internal/store/coverage_test.go | 37 ++- internal/store/migration_observation_test.go | 142 ++++++++++ internal/store/portable.go | 117 ++++++++- internal/store/pull_requests.go | 99 ++++--- internal/store/pull_requests_test.go | 35 ++- internal/store/review_threads.go | 116 ++++++++- internal/store/review_threads_test.go | 95 ++++++- internal/store/schema.go | 64 +++++ internal/store/schema_convergence.go | 4 + internal/store/sqlc/queries.sql | 50 ++-- internal/store/sqlc/schema.sql | 62 +++++ internal/store/store.go | 5 +- internal/store/storedb/models.go | 103 ++++++-- internal/store/storedb/queries.sql.go | 182 +++++++------ internal/store/storedb/queries_test.go | 8 +- internal/store/tombstone_schema.go | 257 +++++++++++++++++++ internal/syncer/pull_details.go | 20 +- internal/syncer/syncer.go | 28 +- internal/syncer/syncer_test.go | 82 ++++-- 24 files changed, 1544 insertions(+), 220 deletions(-) create mode 100644 internal/store/tombstone_schema.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f7a3117d..3ca83a0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## 0.8.3 - Unreleased +- Preserve comments, pull-request reviews, review threads, and commit references as explicit tombstones; comments and review threads retain edit history, and child hydration now merges by identity so only a sourced deletion event deletes an item while a later live observation restores it. + ## 0.8.2 - 2026-07-18 ### Highlights diff --git a/docs/maintainer-archive.md b/docs/maintainer-archive.md index ee5368b0..4dc2d306 100644 --- a/docs/maintainer-archive.md +++ b/docs/maintainer-archive.md @@ -40,6 +40,8 @@ The first command mirrors open issues and pull requests into SQLite. The targete Use `--include-comments` when issue discussion matters. Use `--with pr-details` for pull requests that need files, commits, checks, workflow runs, and review-thread state. Those detail rows are useful for review work, but they cost more GitHub API calls than metadata-only sync. +Child hydration is lossless by default. Gitcrawl merges comments, reviews, review threads, and pull-request commit references by their stable GitHub identity; an item missing from one response is not treated as deleted. Deletion therefore requires an explicit sourced record rather than inference from GitHub list responses. Explicit deletion records remain as tombstones with their reason, edited comments and review threads retain revisions, and a later live observation restores their current row without discarding that revision history. Commit references are immutable identities, so they retain tombstone state but do not have an edit-revision table. + ## Search with bounded staleness Use local search for discovery and add a staleness bound when an agent or script should refresh before answering: diff --git a/internal/cli/app_test.go b/internal/cli/app_test.go index b26e32b8..42f0168b 100644 --- a/internal/cli/app_test.go +++ b/internal/cli/app_test.go @@ -3697,10 +3697,10 @@ func TestDoctorJSONReportsCurrentSchemaDiagnosticsWithoutMutation(t *testing.T) if got := schema["state"]; got != "current" { t.Fatalf("db_schema.state = %#v, payload=%#v", got, schema) } - if got := schema["current_version"]; got != float64(12) { + if got := schema["current_version"]; got != float64(13) { t.Fatalf("db_schema.current_version = %#v, payload=%#v", got, schema) } - if got := schema["supported_version"]; got != float64(12) { + if got := schema["supported_version"]; got != float64(13) { t.Fatalf("db_schema.supported_version = %#v, payload=%#v", got, schema) } if got := schema["child_observation_reservations"]; got != true { @@ -3962,7 +3962,7 @@ func TestDoctorJSONReportsLegacyPendingSchemaWithoutMutation(t *testing.T) { t.Fatalf("pr_details.duplicate_path_files_supported = %#v, payload=%#v", got, prDetails) } pending := doctorStringList(t, schema, "pending_migrations") - if !doctorListContains(pending, "schema_version_3_to_12") || + if !doctorListContains(pending, "schema_version_3_to_13") || !doctorListContains(pending, "pull_request_files_position_key") || !doctorListContains(pending, "thread_child_observation_reservations_table") { t.Fatalf("pending_migrations = %#v", pending) diff --git a/internal/store/comments.go b/internal/store/comments.go index d0e1153b..0f9bdf2f 100644 --- a/internal/store/comments.go +++ b/internal/store/comments.go @@ -2,8 +2,11 @@ package store import ( "context" + "database/sql" "encoding/json" "fmt" + "strings" + "time" "github.com/openclaw/gitcrawl/internal/store/storedb" ) @@ -21,30 +24,128 @@ type Comment struct { RawJSON string `json:"-"` CreatedAtGitHub string `json:"created_at_gh,omitempty"` UpdatedAtGitHub string `json:"updated_at_gh,omitempty"` + DeletedAt string `json:"deleted_at,omitempty"` + DeletionReason string `json:"deletion_reason,omitempty"` } func (s *Store) UpsertComment(ctx context.Context, comment Comment) (int64, error) { + if err := validateTombstone(comment.DeletedAt, comment.DeletionReason); err != nil { + return 0, fmt.Errorf("upsert comment: %w", err) + } + if s.queries == nil { + var id int64 + err := s.WithTx(ctx, func(tx *Store) error { + var err error + id, err = tx.upsertComment(ctx, comment) + return err + }) + return id, err + } + return s.upsertComment(ctx, comment) +} + +func (s *Store) upsertComment(ctx context.Context, comment Comment) (int64, error) { + if comment.DeletedAt != "" { + id, applied, err := s.tombstoneComment(ctx, comment.ThreadID, comment.CommentType, comment.GitHubID, comment.DeletedAt, comment.DeletionReason) + if err != nil { + return 0, err + } + if applied { + return id, nil + } + } id, err := s.qsql().UpsertComment(ctx, storedb.UpsertCommentParams{ - ThreadID: comment.ThreadID, - GithubID: comment.GitHubID, - CommentType: comment.CommentType, - AuthorLogin: nullString(comment.AuthorLogin), - AuthorType: nullString(comment.AuthorType), - Body: comment.Body, - IsBot: int64(boolInt(comment.IsBot)), - RawJson: comment.RawJSON, - CreatedAtGh: nullString(comment.CreatedAtGitHub), - UpdatedAtGh: nullString(comment.UpdatedAtGitHub), + ThreadID: comment.ThreadID, + GithubID: comment.GitHubID, + CommentType: comment.CommentType, + AuthorLogin: nullString(comment.AuthorLogin), + AuthorType: nullString(comment.AuthorType), + Body: comment.Body, + IsBot: int64(boolInt(comment.IsBot)), + RawJson: comment.RawJSON, + CreatedAtGh: nullString(comment.CreatedAtGitHub), + UpdatedAtGh: nullString(comment.UpdatedAtGitHub), + DeletedAt: nullString(comment.DeletedAt), + DeletionReason: nullString(comment.DeletionReason), }) if err != nil { return 0, fmt.Errorf("upsert comment: %w", err) } + if err := s.recordCommentRevision(ctx, id, time.Now().UTC().Format(timeLayout)); err != nil { + return 0, err + } return id, nil } -func (s *Store) DeleteCommentsForThread(ctx context.Context, threadID int64) error { - if _, err := s.q().ExecContext(ctx, `delete from comments where thread_id = ?`, threadID); err != nil { - return fmt.Errorf("delete comments for thread: %w", err) +func (s *Store) TombstoneComment(ctx context.Context, threadID int64, commentType, githubID, deletedAt, reason string) (bool, error) { + if err := validateTombstone(deletedAt, reason); err != nil { + return false, fmt.Errorf("tombstone comment: %w", err) + } + if deletedAt == "" { + return false, fmt.Errorf("tombstone comment: deleted_at is required") + } + if s.queries == nil { + var applied bool + err := s.WithTx(ctx, func(tx *Store) error { + var err error + _, applied, err = tx.tombstoneComment(ctx, threadID, commentType, githubID, deletedAt, reason) + return err + }) + return applied, err + } + _, applied, err := s.tombstoneComment(ctx, threadID, commentType, githubID, deletedAt, reason) + return applied, err +} + +func (s *Store) tombstoneComment(ctx context.Context, threadID int64, commentType, githubID, deletedAt, reason string) (int64, bool, error) { + var id int64 + err := s.q().QueryRowContext(ctx, ` + update comments + set deleted_at = ?, deletion_reason = ? + where thread_id = ? and comment_type = ? and github_id = ? + returning id + `, deletedAt, reason, threadID, commentType, githubID).Scan(&id) + if err == sql.ErrNoRows { + return 0, false, nil + } + if err != nil { + return 0, false, fmt.Errorf("tombstone comment: %w", err) + } + if err := s.recordCommentRevision(ctx, id, time.Now().UTC().Format(timeLayout)); err != nil { + return 0, false, err + } + return id, true, nil +} + +func (s *Store) recordCommentRevision(ctx context.Context, commentID int64, recordedAt string) error { + if _, err := s.q().ExecContext(ctx, ` + insert into comment_revisions( + comment_id, author_login, author_type, body, is_bot, raw_json, + created_at_gh, updated_at_gh, deleted_at, deletion_reason, recorded_at + ) + select c.id, c.author_login, c.author_type, c.body, c.is_bot, c.raw_json, + c.created_at_gh, c.updated_at_gh, c.deleted_at, c.deletion_reason, ? + from comments c + where c.id = ? + and not exists ( + select 1 + from comment_revisions r + where r.id = ( + select max(latest.id) from comment_revisions latest + where latest.comment_id = c.id + ) + and r.author_login is c.author_login + and r.author_type is c.author_type + and r.body is c.body + and r.is_bot is c.is_bot + and r.raw_json is c.raw_json + and r.created_at_gh is c.created_at_gh + and r.updated_at_gh is c.updated_at_gh + and r.deleted_at is c.deleted_at + and r.deletion_reason is c.deletion_reason + ) + `, recordedAt, commentID); err != nil { + return fmt.Errorf("record comment revision: %w", err) } return nil } @@ -72,11 +173,31 @@ func (s *Store) ListComments(ctx context.Context, threadID int64) ([]Comment, er RawJSON: row.RawJson, CreatedAtGitHub: stringValue(row.CreatedAtGh), UpdatedAtGitHub: stringValue(row.UpdatedAtGh), + DeletedAt: stringValue(row.DeletedAt), + DeletionReason: stringValue(row.DeletionReason), }) } return comments, nil } +func validateTombstone(deletedAt, reason string) error { + normalizedDeletedAt := strings.TrimSpace(deletedAt) + normalizedReason := strings.TrimSpace(reason) + if deletedAt != normalizedDeletedAt || reason != normalizedReason { + return fmt.Errorf("deleted_at and deletion_reason must not have surrounding whitespace") + } + if (normalizedDeletedAt == "") != (normalizedReason == "") { + return fmt.Errorf("deleted_at and deletion_reason must be set together") + } + if normalizedDeletedAt == "" { + return nil + } + if _, err := time.Parse(timeLayout, normalizedDeletedAt); err != nil { + return fmt.Errorf("deleted_at must be RFC3339: %w", err) + } + return nil +} + func reviewStateFromRawJSON(raw string) string { var value struct { State string `json:"state"` diff --git a/internal/store/comments_test.go b/internal/store/comments_test.go index d9a6025d..1630406b 100644 --- a/internal/store/comments_test.go +++ b/internal/store/comments_test.go @@ -3,6 +3,7 @@ package store import ( "context" "path/filepath" + "strings" "testing" ) @@ -50,4 +51,104 @@ func TestUpsertComment(t *testing.T) { if len(comments) != 2 || comments[0].GitHubID != "c0" || !comments[0].IsBot || comments[1].GitHubID != "c1" { t.Fatalf("comments = %+v", comments) } + + edited := comments[1] + edited.Body = "same edited bug here" + edited.UpdatedAtGitHub = "2026-04-26T00:02:00Z" + if _, err := st.UpsertComment(ctx, edited); err != nil { + t.Fatalf("edit comment: %v", err) + } + deletedAt := "2026-04-26T00:03:00Z" + if _, err := st.UpsertComment(ctx, Comment{ + ThreadID: threadID, GitHubID: "c1", CommentType: "issue_comment", + DeletedAt: deletedAt, DeletionReason: "explicit-source-delete", + }); err != nil { + t.Fatalf("import sparse comment tombstone: %v", err) + } + comments, err = st.ListComments(ctx, threadID) + if err != nil { + t.Fatalf("list comments after tombstone: %v", err) + } + if len(comments) != 1 || comments[0].GitHubID != "c0" { + t.Fatalf("tombstoned comment remained visible: %+v", comments) + } + var revisionCount int + if err := st.DB().QueryRowContext(ctx, ` + select count(*) from comment_revisions where comment_id = ? + `, id).Scan(&revisionCount); err != nil { + t.Fatalf("comment revisions: %v", err) + } + if revisionCount != 3 { + t.Fatalf("comment revisions = %d, want create/edit/delete", revisionCount) + } + var retainedBody string + if err := st.DB().QueryRowContext(ctx, `select body from comments where id = ?`, id).Scan(&retainedBody); err != nil { + t.Fatalf("retained tombstone body: %v", err) + } + if retainedBody != edited.Body { + t.Fatalf("sparse tombstone replaced body with %q", retainedBody) + } + edited.DeletedAt = "" + edited.DeletionReason = "" + edited.UpdatedAtGitHub = "2026-04-26T00:04:00Z" + if _, err := st.UpsertComment(ctx, edited); err != nil { + t.Fatalf("restore comment: %v", err) + } + comments, err = st.ListComments(ctx, threadID) + if err != nil { + t.Fatalf("list restored comments: %v", err) + } + if len(comments) != 2 || comments[1].GitHubID != "c1" || comments[1].Body != edited.Body { + t.Fatalf("restored comments = %+v", comments) + } + if err := st.DB().QueryRowContext(ctx, ` + select count(*) from comment_revisions where comment_id = ? + `, id).Scan(&revisionCount); err != nil { + t.Fatalf("restored comment revisions: %v", err) + } + if revisionCount != 4 { + t.Fatalf("comment revisions after restore = %d, want 4", revisionCount) + } +} + +func TestTombstoneFieldsRejectSurroundingWhitespace(t *testing.T) { + ctx := context.Background() + st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + defer st.Close() + + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "gitcrawl", FullName: "openclaw/gitcrawl", RawJSON: "{}", UpdatedAt: "2026-05-01T00:00:00Z"}) + if err != nil { + t.Fatalf("repo: %v", err) + } + threadID, err := st.UpsertThread(ctx, Thread{ + RepoID: repoID, GitHubID: "2", Number: 2, Kind: "issue", State: "open", + Title: "whitespace", HTMLURL: "https://github.com/openclaw/gitcrawl/issues/2", + LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: "hash", UpdatedAt: "2026-05-01T00:00:00Z", + }) + if err != nil { + t.Fatalf("thread: %v", err) + } + + for _, test := range []struct { + name string + deletedAt string + reason string + }{ + {name: "whitespace-only pair", deletedAt: " ", reason: " "}, + {name: "timestamp padding", deletedAt: " 2026-05-01T00:01:00Z", reason: "explicit-source-delete"}, + {name: "reason padding", deletedAt: "2026-05-01T00:01:00Z", reason: "explicit-source-delete "}, + } { + t.Run(test.name, func(t *testing.T) { + _, err := st.UpsertComment(ctx, Comment{ + ThreadID: threadID, GitHubID: "space", CommentType: "issue_comment", + Body: "body", RawJSON: `{}`, DeletedAt: test.deletedAt, DeletionReason: test.reason, + }) + if err == nil || !strings.Contains(err.Error(), "surrounding whitespace") { + t.Fatalf("upsert comment error = %v, want surrounding whitespace rejection", err) + } + }) + } } diff --git a/internal/store/coverage_test.go b/internal/store/coverage_test.go index 494f89bc..03a9058a 100644 --- a/internal/store/coverage_test.go +++ b/internal/store/coverage_test.go @@ -214,6 +214,21 @@ func TestPortablePruneCanonicalizesSchemaAndMetadata(t *testing.T) { if _, err := st.UpsertComment(ctx, Comment{ThreadID: threadIDs[0], GitHubID: "c1", CommentType: "issue_comment", AuthorLogin: "alice", Body: "portable comment body", RawJSON: `{"body":"portable comment body"}`, CreatedAtGitHub: "2026-04-30T00:00:00Z", UpdatedAtGitHub: "2026-04-30T00:00:00Z"}); err != nil { t.Fatalf("upsert comment: %v", err) } + reviewThread := PullRequestReviewThread{ + ReviewThreadID: "portable-review", FirstCommentBody: "historical review body secret", + CommentsJSON: `[{"body":"historical nested review body secret"}]`, RawJSON: `{"raw":"historical review body secret"}`, + FetchedAt: "2026-04-30T00:00:00Z", + } + if err := st.UpsertPullRequestReviewThreads(ctx, threadIDs[1], reviewThread.FetchedAt, []PullRequestReviewThread{reviewThread}); err != nil { + t.Fatalf("upsert historical review thread: %v", err) + } + reviewThread.FirstCommentBody = "current review body secret" + reviewThread.CommentsJSON = `[{"body":"current nested review body secret"}]` + reviewThread.RawJSON = `{"raw":"current review body secret"}` + reviewThread.FetchedAt = "2026-04-30T00:01:00Z" + if err := st.UpsertPullRequestReviewThreads(ctx, threadIDs[1], reviewThread.FetchedAt, []PullRequestReviewThread{reviewThread}); err != nil { + t.Fatalf("upsert current review thread: %v", err) + } if _, err := st.DB().ExecContext(ctx, `insert into sync_runs(repo_id, scope, status, started_at, finished_at, stats_json) values(?, 'open', 'success', '2026-04-30T00:00:00Z', '2026-04-30T00:01:00Z', '{}')`, repoID); err != nil { t.Fatalf("seed sync run: %v", err) } @@ -230,6 +245,23 @@ func TestPortablePruneCanonicalizesSchemaAndMetadata(t *testing.T) { if !st.tableExists(ctx, "comments") { t.Fatalf("comments should remain in portable v2") } + if !st.tableExists(ctx, "comment_revisions") || !st.tableExists(ctx, "pull_request_review_thread_revisions") { + t.Fatalf("family revision history should remain in portable v2") + } + var compactReviewBodies string + if err := st.DB().QueryRowContext(ctx, ` + select group_concat(first_comment_body || comments_json || raw_json, '|') + from ( + select first_comment_body, comments_json, raw_json from pull_request_review_threads + union all + select first_comment_body, comments_json, raw_json from pull_request_review_thread_revisions + ) + `).Scan(&compactReviewBodies); err != nil { + t.Fatalf("read compact review history: %v", err) + } + if strings.Contains(compactReviewBodies, "review body secret") || strings.Contains(compactReviewBodies, "nested review") || strings.Contains(compactReviewBodies, `"raw":`) { + t.Fatalf("portable review history retained full text: %q", compactReviewBodies) + } var schema, includes, excluded string if err := st.DB().QueryRowContext(ctx, `select value from portable_metadata where key = 'schema'`).Scan(&schema); err != nil { t.Fatalf("schema metadata: %v", err) @@ -240,7 +272,10 @@ func TestPortablePruneCanonicalizesSchemaAndMetadata(t *testing.T) { if err := st.DB().QueryRowContext(ctx, `select value from portable_metadata where key = 'excluded'`).Scan(&excluded); err != nil { t.Fatalf("excluded metadata: %v", err) } - if schema != "gitcrawl-portable-sync-v2" || !strings.Contains(includes, "comments") || strings.Contains(excluded, "comments") { + if schema != "gitcrawl-portable-sync-v2" || + !strings.Contains(includes, "comment_revisions") || + !strings.Contains(includes, "pull_request_review_thread_revisions") || + strings.Contains(excluded, "comments") { t.Fatalf("portable metadata schema=%q includes=%q excluded=%q", schema, includes, excluded) } var version int diff --git a/internal/store/migration_observation_test.go b/internal/store/migration_observation_test.go index 76d2a86f..15c6fde2 100644 --- a/internal/store/migration_observation_test.go +++ b/internal/store/migration_observation_test.go @@ -15,6 +15,148 @@ import ( "time" ) +func TestOpenMigratesFamilyTombstonesLosslessly(t *testing.T) { + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "gitcrawl.db") + st, err := Open(ctx, dbPath) + if err != nil { + t.Fatalf("open seed store: %v", err) + } + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "gitcrawl", FullName: "openclaw/gitcrawl", RawJSON: "{}", UpdatedAt: "2026-07-18T00:00:00Z"}) + if err != nil { + t.Fatalf("seed repository: %v", err) + } + threadID, err := st.UpsertThread(ctx, Thread{ + RepoID: repoID, GitHubID: "17", Number: 17, Kind: "pull_request", State: "open", + Title: "legacy family rows", HTMLURL: "https://github.com/openclaw/gitcrawl/pull/17", + LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: "thread-hash", UpdatedAt: "2026-07-18T00:00:00Z", + }) + if err != nil { + t.Fatalf("seed thread: %v", err) + } + if _, err := st.UpsertComment(ctx, Comment{ + ThreadID: threadID, GitHubID: "comment-17", CommentType: "pull_review", + AuthorLogin: "alice", Body: "legacy review", RawJSON: `{"body":"legacy review"}`, + CreatedAtGitHub: "2026-07-18T00:01:00Z", UpdatedAtGitHub: "2026-07-18T00:02:00Z", + }); err != nil { + t.Fatalf("seed comment: %v", err) + } + detail := PullRequestDetail{ThreadID: threadID, RepoID: repoID, Number: 17, RawJSON: "{}", FetchedAt: "2026-07-18T00:03:00Z", UpdatedAt: "2026-07-18T00:03:00Z"} + if err := st.UpsertPullRequestCache(ctx, detail, nil, []PullRequestCommit{{ + SHA: "abc17", Message: "legacy commit", RawJSON: `{"sha":"abc17"}`, FetchedAt: "2026-07-18T00:03:00Z", + }}, nil, nil); err != nil { + t.Fatalf("seed commit: %v", err) + } + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, "2026-07-18T00:04:00Z", []PullRequestReviewThread{{ + ReviewThreadID: "rt-17", Path: "legacy.go", Line: 17, CommentsJSON: `[{"body":"legacy thread"}]`, + RawJSON: `{"id":"rt-17"}`, FetchedAt: "2026-07-18T00:04:00Z", + }}); err != nil { + t.Fatalf("seed review thread: %v", err) + } + if err := st.Close(); err != nil { + t.Fatalf("close seed store: %v", err) + } + + raw := openRawMigrationDB(t, dbPath) + statements := []string{ + `pragma foreign_keys = off`, + `drop table comment_revisions`, + `drop table pull_request_review_thread_revisions`, + `create table comments_legacy ( + id integer primary key, thread_id integer not null references threads(id) on delete cascade, + github_id text not null, comment_type text not null, author_login text, author_type text, + body text not null, is_bot integer not null default 0, raw_json text not null, + raw_json_blob_id integer references blobs(id) on delete set null, + created_at_gh text, updated_at_gh text, unique(thread_id, comment_type, github_id))`, + `insert into comments_legacy select id, thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, raw_json_blob_id, created_at_gh, updated_at_gh from comments`, + `drop table comments`, + `alter table comments_legacy rename to comments`, + `create table pull_request_commits_legacy ( + thread_id integer not null references threads(id) on delete cascade, sha text not null, + message text, author_login text, author_name text, committed_at text, html_url text, + raw_json text not null, fetched_at text not null, primary key(thread_id, sha))`, + `insert into pull_request_commits_legacy select thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at from pull_request_commits`, + `drop table pull_request_commits`, + `alter table pull_request_commits_legacy rename to pull_request_commits`, + `create table pull_request_review_threads_legacy ( + thread_id integer not null references threads(id) on delete cascade, review_thread_id text not null, + path text, line integer not null default 0, start_line integer not null default 0, + is_resolved integer not null default 0, is_outdated integer not null default 0, + viewer_can_resolve integer not null default 0, viewer_can_unresolve integer not null default 0, + viewer_can_reply integer not null default 0, first_author_login text, first_author_type text, + first_comment_body text, first_comment_url text, first_comment_created_at text, + first_comment_updated_at text, comments_json text not null, raw_json text not null, + fetched_at text not null, primary key(thread_id, review_thread_id))`, + `insert into pull_request_review_threads_legacy select thread_id, review_thread_id, path, line, start_line, is_resolved, is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, first_author_login, first_author_type, first_comment_body, first_comment_url, first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at from pull_request_review_threads`, + `drop table pull_request_review_threads`, + `alter table pull_request_review_threads_legacy rename to pull_request_review_threads`, + `pragma user_version = 12`, + `pragma foreign_keys = on`, + } + for _, statement := range statements { + if _, err := raw.ExecContext(ctx, statement); err != nil { + _ = raw.Close() + t.Fatalf("prepare legacy family schema with %q: %v", statement, err) + } + } + if err := raw.Close(); err != nil { + t.Fatalf("close legacy store: %v", err) + } + + st, err = Open(ctx, dbPath) + if err != nil { + t.Fatalf("migrate legacy family schema: %v", err) + } + defer st.Close() + comments, err := st.ListComments(ctx, threadID) + if err != nil || len(comments) != 1 || comments[0].Body != "legacy review" { + t.Fatalf("migrated comments = %+v err=%v", comments, err) + } + commits, err := st.PullRequestCommits(ctx, threadID) + if err != nil || len(commits) != 1 || commits[0].Message != "legacy commit" { + t.Fatalf("migrated commits = %+v err=%v", commits, err) + } + reviewThreads, err := st.PullRequestReviewThreads(ctx, threadID) + if err != nil || len(reviewThreads) != 1 || reviewThreads[0].Path != "legacy.go" { + t.Fatalf("migrated review threads = %+v err=%v", reviewThreads, err) + } + for _, check := range []struct { + table string + want int + }{ + {table: "comment_revisions", want: 1}, + {table: "pull_request_review_thread_revisions", want: 1}, + } { + var count int + if err := st.DB().QueryRowContext(ctx, `select count(*) from `+check.table).Scan(&count); err != nil { + t.Fatalf("count %s: %v", check.table, err) + } + if count != check.want { + t.Fatalf("%s rows = %d, want %d", check.table, count, check.want) + } + } + if !st.familyTombstoneSchemaHasCurrentShape(ctx) { + t.Fatal("migrated family tombstone schema did not converge to the fresh schema") + } + for _, invalidUpdate := range []string{ + `update comments set deleted_at = '2026-07-18T00:05:00Z', deletion_reason = null where github_id = 'comment-17'`, + `update pull_request_commits set deleted_at = '2026-07-18T00:05:00Z', deletion_reason = null where sha = 'abc17'`, + `update pull_request_review_threads set deleted_at = '2026-07-18T00:05:00Z', deletion_reason = null where review_thread_id = 'rt-17'`, + } { + if _, err := st.DB().ExecContext(ctx, invalidUpdate); err == nil || !strings.Contains(err.Error(), "CHECK constraint failed") { + t.Fatalf("migrated tombstone constraint error = %v for %q", err, invalidUpdate) + } + } + rows, err := st.DB().QueryContext(ctx, `pragma foreign_key_check`) + if err != nil { + t.Fatalf("foreign key check: %v", err) + } + defer rows.Close() + if rows.Next() { + t.Fatal("migration left a foreign key violation") + } +} + func TestOpenRepairsMinimumThreadObservationSequence(t *testing.T) { ctx := context.Background() dbPath, _, threadID := seedMigrationPullRequest(t, "minimum-sequence-head", 11) diff --git a/internal/store/portable.go b/internal/store/portable.go index 89ba6f4a..ad5d128f 100644 --- a/internal/store/portable.go +++ b/internal/store/portable.go @@ -109,6 +109,17 @@ func (s *Store) PrunePortablePayloads(ctx context.Context, options PortablePrune stats.CommentsPruned = rowsAffected(result) } } + if s.tableExists(ctx, "comment_revisions") { + if _, err := s.db.ExecContext(ctx, ` + update comment_revisions + set body = case when length(body) > ? then substr(body, 1, ?) else body end + `, options.BodyChars, options.BodyChars); err != nil { + return stats, fmt.Errorf("prune comment revision bodies: %w", err) + } + } + if err := s.compactPortableReviewThreadBodies(ctx, options.BodyChars); err != nil { + return stats, err + } if labels, assignees, err := s.compactPortableThreadMetadata(ctx); err != nil { return stats, err } else { @@ -358,8 +369,8 @@ func (s *Store) canonicalizePortableSchema(ctx context.Context, bodyChars int, i if err := s.ensurePortableMetadata(ctx); err != nil { return err } - capabilities := "body_excerpts,comment_excerpts,author_association,thread_revisions,thread_fingerprints,thread_key_summaries,pr_details,pr_files,pr_commits,pr_checks,pr_review_threads,workflow_runs,raw_json_stripped" - includes := "repositories,threads,comments,thread_revisions,thread_fingerprints,thread_key_summaries,pull_request_details,pull_request_files,pull_request_commits,pull_request_checks,pull_request_review_threads,pull_request_review_thread_syncs,github_workflow_runs" + capabilities := "body_excerpts,comment_excerpts,author_association,thread_revisions,thread_fingerprints,thread_key_summaries,pr_details,pr_files,pr_commits,pr_checks,pr_review_threads,workflow_runs,family_tombstones,comment_revisions,pr_review_thread_revisions,raw_json_stripped" + includes := "repositories,threads,comments,comment_revisions,thread_revisions,thread_fingerprints,thread_key_summaries,pull_request_details,pull_request_files,pull_request_commits,pull_request_checks,pull_request_review_threads,pull_request_review_thread_revisions,pull_request_review_thread_syncs,github_workflow_runs" excluded := "raw_json,documents,fts,vectors,code_snapshots,code_documents,cluster_events,run_history,similarity_edges,blobs,sync_attempt_failures" if stats.SyncFailuresIncluded { capabilities += ",sync_failure_ledger_redacted" @@ -459,6 +470,106 @@ func (s *Store) compactPortableThreadMetadata(ctx context.Context) (int64, int64 return labelsCompacted, assigneesCompacted, nil } +func (s *Store) compactPortableReviewThreadBodies(ctx context.Context, bodyChars int) error { + for _, table := range []string{"pull_request_review_threads", "pull_request_review_thread_revisions"} { + if !s.hasColumns(ctx, table, "first_comment_body", "comments_json") { + continue + } + rows, err := s.db.QueryContext(ctx, ` + select rowid, first_comment_body, comments_json + from `+sqliteIdentifier(table)+` + order by rowid + `) + if err != nil { + return fmt.Errorf("read portable review bodies from %s: %w", table, err) + } + type update struct { + rowID int64 + firstBody sql.NullString + commentsJSON string + } + var updates []update + for rows.Next() { + var rowID int64 + var firstBody sql.NullString + var commentsJSON string + if err := rows.Scan(&rowID, &firstBody, &commentsJSON); err != nil { + _ = rows.Close() + return fmt.Errorf("scan portable review bodies from %s: %w", table, err) + } + nextFirst := firstBody + if nextFirst.Valid { + nextFirst.String = truncatePortableText(nextFirst.String, bodyChars) + } + nextComments := compactPortableReviewComments(commentsJSON, bodyChars) + if nextFirst == firstBody && nextComments == commentsJSON { + continue + } + updates = append(updates, update{rowID: rowID, firstBody: nextFirst, commentsJSON: nextComments}) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return fmt.Errorf("read portable review bodies from %s: %w", table, err) + } + if err := rows.Close(); err != nil { + return fmt.Errorf("close portable review bodies from %s: %w", table, err) + } + for _, update := range updates { + if _, err := s.db.ExecContext(ctx, ` + update `+sqliteIdentifier(table)+` + set first_comment_body = ?, comments_json = ? + where rowid = ? + `, update.firstBody, update.commentsJSON, update.rowID); err != nil { + return fmt.Errorf("compact portable review bodies in %s: %w", table, err) + } + } + } + return nil +} + +func compactPortableReviewComments(raw string, bodyChars int) string { + var value any + if err := json.Unmarshal([]byte(raw), &value); err != nil { + return "[]" + } + truncatePortableReviewBodies(value, bodyChars) + compact, err := json.Marshal(value) + if err != nil { + return "[]" + } + return string(compact) +} + +func truncatePortableReviewBodies(value any, bodyChars int) { + switch typed := value.(type) { + case []any: + for _, item := range typed { + truncatePortableReviewBodies(item, bodyChars) + } + case map[string]any: + for key, item := range typed { + if key == "body" || key == "bodyText" { + if text, ok := item.(string); ok { + typed[key] = truncatePortableText(text, bodyChars) + } + continue + } + truncatePortableReviewBodies(item, bodyChars) + } + } +} + +func truncatePortableText(value string, limit int) string { + if limit <= 0 { + return "" + } + runes := []rune(value) + if len(runes) <= limit { + return value + } + return string(runes[:limit]) +} + func compactPortableNameList(raw, field string) string { var values []any if err := json.Unmarshal([]byte(raw), &values); err != nil { @@ -508,11 +619,13 @@ func (s *Store) clearPortableRawJSON(ctx context.Context) (int64, error) { name string }{ {table: "comments", name: "raw_json"}, + {table: "comment_revisions", name: "raw_json"}, {table: "pull_request_details", name: "raw_json"}, {table: "pull_request_files", name: "raw_json"}, {table: "pull_request_commits", name: "raw_json"}, {table: "pull_request_checks", name: "raw_json"}, {table: "pull_request_review_threads", name: "raw_json"}, + {table: "pull_request_review_thread_revisions", name: "raw_json"}, {table: "github_workflow_runs", name: "raw_json"}, } { if !s.hasColumn(ctx, column.table, column.name) { diff --git a/internal/store/pull_requests.go b/internal/store/pull_requests.go index 2f78d360..ad64e4e3 100644 --- a/internal/store/pull_requests.go +++ b/internal/store/pull_requests.go @@ -40,15 +40,17 @@ type PullRequestFile struct { } type PullRequestCommit struct { - ThreadID int64 `json:"thread_id"` - SHA string `json:"sha"` - Message string `json:"message,omitempty"` - AuthorLogin string `json:"author_login,omitempty"` - AuthorName string `json:"author_name,omitempty"` - CommittedAt string `json:"committed_at,omitempty"` - HTMLURL string `json:"html_url,omitempty"` - RawJSON string `json:"raw_json,omitempty"` - FetchedAt string `json:"fetched_at"` + ThreadID int64 `json:"thread_id"` + SHA string `json:"sha"` + Message string `json:"message,omitempty"` + AuthorLogin string `json:"author_login,omitempty"` + AuthorName string `json:"author_name,omitempty"` + CommittedAt string `json:"committed_at,omitempty"` + HTMLURL string `json:"html_url,omitempty"` + RawJSON string `json:"raw_json,omitempty"` + FetchedAt string `json:"fetched_at"` + DeletedAt string `json:"deleted_at,omitempty"` + DeletionReason string `json:"deletion_reason,omitempty"` } type PullRequestCheck struct { @@ -254,20 +256,31 @@ func (s *Store) upsertPullRequestCacheFamilies( } } if families.Commits { - if err := s.qsql().DeletePullRequestCommits(ctx, detail.ThreadID); err != nil { - return fmt.Errorf("clear pull request commits: %w", err) - } for _, commit := range commits { - if err := s.qsql().InsertPullRequestCommit(ctx, storedb.InsertPullRequestCommitParams{ - ThreadID: detail.ThreadID, - Sha: commit.SHA, - Message: nullString(commit.Message), - AuthorLogin: nullString(commit.AuthorLogin), - AuthorName: nullString(commit.AuthorName), - CommittedAt: nullString(commit.CommittedAt), - HtmlUrl: nullString(commit.HTMLURL), - RawJson: commit.RawJSON, - FetchedAt: commit.FetchedAt, + if err := validateTombstone(commit.DeletedAt, commit.DeletionReason); err != nil { + return fmt.Errorf("upsert pull request commit %q: %w", commit.SHA, err) + } + if commit.DeletedAt != "" { + applied, err := s.tombstonePullRequestCommit(ctx, detail.ThreadID, commit.SHA, commit.DeletedAt, commit.DeletionReason) + if err != nil { + return err + } + if applied { + continue + } + } + if err := s.qsql().UpsertPullRequestCommit(ctx, storedb.UpsertPullRequestCommitParams{ + ThreadID: detail.ThreadID, + Sha: commit.SHA, + Message: nullString(commit.Message), + AuthorLogin: nullString(commit.AuthorLogin), + AuthorName: nullString(commit.AuthorName), + CommittedAt: nullString(commit.CommittedAt), + HtmlUrl: nullString(commit.HTMLURL), + RawJson: commit.RawJSON, + FetchedAt: commit.FetchedAt, + DeletedAt: nullString(commit.DeletedAt), + DeletionReason: nullString(commit.DeletionReason), }); err != nil { return fmt.Errorf("upsert pull request commit: %w", err) } @@ -460,20 +473,44 @@ func (s *Store) PullRequestCommits(ctx context.Context, threadID int64) ([]PullR out := make([]PullRequestCommit, 0, len(rows)) for _, row := range rows { out = append(out, PullRequestCommit{ - ThreadID: row.ThreadID, - SHA: row.Sha, - Message: stringValue(row.Message), - AuthorLogin: stringValue(row.AuthorLogin), - AuthorName: stringValue(row.AuthorName), - CommittedAt: stringValue(row.CommittedAt), - HTMLURL: stringValue(row.HtmlUrl), - RawJSON: row.RawJson, - FetchedAt: row.FetchedAt, + ThreadID: row.ThreadID, + SHA: row.Sha, + Message: stringValue(row.Message), + AuthorLogin: stringValue(row.AuthorLogin), + AuthorName: stringValue(row.AuthorName), + CommittedAt: stringValue(row.CommittedAt), + HTMLURL: stringValue(row.HtmlUrl), + RawJSON: row.RawJson, + FetchedAt: row.FetchedAt, + DeletedAt: stringValue(row.DeletedAt), + DeletionReason: stringValue(row.DeletionReason), }) } return out, nil } +func (s *Store) TombstonePullRequestCommit(ctx context.Context, threadID int64, sha, deletedAt, reason string) (bool, error) { + if err := validateTombstone(deletedAt, reason); err != nil { + return false, fmt.Errorf("tombstone pull request commit: %w", err) + } + if deletedAt == "" { + return false, fmt.Errorf("tombstone pull request commit: deleted_at is required") + } + return s.tombstonePullRequestCommit(ctx, threadID, sha, deletedAt, reason) +} + +func (s *Store) tombstonePullRequestCommit(ctx context.Context, threadID int64, sha, deletedAt, reason string) (bool, error) { + result, err := s.q().ExecContext(ctx, ` + update pull_request_commits + set deleted_at = ?, deletion_reason = ? + where thread_id = ? and sha = ? + `, deletedAt, reason, threadID, sha) + if err != nil { + return false, fmt.Errorf("tombstone pull request commit: %w", err) + } + return rowsAffected(result) != 0, nil +} + func (s *Store) PullRequestChecks(ctx context.Context, threadID int64) ([]PullRequestCheck, error) { rows, err := s.qsql().PullRequestChecks(ctx, threadID) if err != nil { diff --git a/internal/store/pull_requests_test.go b/internal/store/pull_requests_test.go index 14e6a450..ffec4456 100644 --- a/internal/store/pull_requests_test.go +++ b/internal/store/pull_requests_test.go @@ -92,9 +92,42 @@ func TestPullRequestCacheRoundTripAndWorkflowFilters(t *testing.T) { if err != nil { t.Fatalf("updated pull request cache: %v", err) } - if cache.Detail.HeadSHA != "head-v2" || len(cache.Files) != 1 || len(cache.Commits) != 0 || len(cache.Checks) != 0 { + if cache.Detail.HeadSHA != "head-v2" || len(cache.Files) != 1 || len(cache.Commits) != 1 || cache.Commits[0].SHA != "abc" || len(cache.Checks) != 0 { t.Fatalf("updated cache = %+v", cache) } + if err := st.UpsertPullRequestCacheFamilies(ctx, detail, nil, []PullRequestCommit{{ + SHA: "abc", DeletedAt: "2026-05-05T10:01:00Z", DeletionReason: "explicit-source-delete", + }}, nil, nil, PullRequestHydrationFamilies{Commits: true}); err != nil { + t.Fatalf("import sparse pull request commit tombstone: %v", err) + } + commitsAfterDelete, err := st.PullRequestCommits(ctx, threadID) + if err != nil || len(commitsAfterDelete) != 0 { + t.Fatalf("tombstoned commits = %+v err=%v", commitsAfterDelete, err) + } + if err := st.UpsertPullRequestCacheFamilies(ctx, detail, nil, nil, nil, nil, PullRequestHydrationFamilies{Commits: true}); err != nil { + t.Fatalf("empty commit merge: %v", err) + } + var deletedAt, deletionReason string + if err := st.DB().QueryRowContext(ctx, `select deleted_at, deletion_reason from pull_request_commits where thread_id = ? and sha = 'abc'`, threadID).Scan(&deletedAt, &deletionReason); err != nil { + t.Fatalf("read preserved commit tombstone: %v", err) + } + if deletedAt == "" || deletionReason != "explicit-source-delete" { + t.Fatalf("commit tombstone = %q/%q", deletedAt, deletionReason) + } + var retainedMessage string + if err := st.DB().QueryRowContext(ctx, `select message from pull_request_commits where thread_id = ? and sha = 'abc'`, threadID).Scan(&retainedMessage); err != nil { + t.Fatalf("read retained commit message: %v", err) + } + if retainedMessage != "feat: cache" { + t.Fatalf("sparse commit tombstone replaced message with %q", retainedMessage) + } + if err := st.UpsertPullRequestCacheFamilies(ctx, detail, nil, commits, nil, nil, PullRequestHydrationFamilies{Commits: true}); err != nil { + t.Fatalf("restore pull request commit: %v", err) + } + restoredCommits, err := st.PullRequestCommits(ctx, threadID) + if err != nil || len(restoredCommits) != 1 || restoredCommits[0].SHA != "abc" { + t.Fatalf("restored commits = %+v err=%v", restoredCommits, err) + } } func TestMissingPullRequestDetailNumbersSelectsOnlyUnfilledPRs(t *testing.T) { diff --git a/internal/store/review_threads.go b/internal/store/review_threads.go index 9c2ba143..baf3d1c6 100644 --- a/internal/store/review_threads.go +++ b/internal/store/review_threads.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "time" "github.com/openclaw/gitcrawl/internal/store/storedb" ) @@ -28,9 +29,16 @@ type PullRequestReviewThread struct { CommentsJSON string `json:"comments_json"` RawJSON string `json:"-"` FetchedAt string `json:"fetched_at"` + DeletedAt string `json:"deleted_at,omitempty"` + DeletionReason string `json:"deletion_reason,omitempty"` } func (s *Store) UpsertPullRequestReviewThreads(ctx context.Context, threadID int64, fetchedAt string, threads []PullRequestReviewThread) error { + for _, thread := range threads { + if err := validateTombstone(thread.DeletedAt, thread.DeletionReason); err != nil { + return fmt.Errorf("upsert pull request review thread %q: %w", thread.ReviewThreadID, err) + } + } if s.queries == nil { return s.WithTx(ctx, func(tx *Store) error { return tx.upsertPullRequestReviewThreads(ctx, threadID, fetchedAt, threads) @@ -40,9 +48,6 @@ func (s *Store) UpsertPullRequestReviewThreads(ctx context.Context, threadID int } func (s *Store) upsertPullRequestReviewThreads(ctx context.Context, threadID int64, fetchedAt string, threads []PullRequestReviewThread) error { - if err := s.qsql().DeletePullRequestReviewThreads(ctx, threadID); err != nil { - return fmt.Errorf("clear pull request review threads: %w", err) - } if err := s.qsql().UpsertPullRequestReviewThreadSync(ctx, storedb.UpsertPullRequestReviewThreadSyncParams{ ThreadID: threadID, FetchedAt: fetchedAt, @@ -53,6 +58,15 @@ func (s *Store) upsertPullRequestReviewThreads(ctx context.Context, threadID int if thread.ReviewThreadID == "" { continue } + if thread.DeletedAt != "" { + applied, err := s.tombstonePullRequestReviewThread(ctx, threadID, thread.ReviewThreadID, thread.DeletedAt, thread.DeletionReason) + if err != nil { + return err + } + if applied { + continue + } + } if err := s.qsql().UpsertPullRequestReviewThread(ctx, storedb.UpsertPullRequestReviewThreadParams{ ThreadID: threadID, ReviewThreadID: thread.ReviewThreadID, @@ -73,9 +87,103 @@ func (s *Store) upsertPullRequestReviewThreads(ctx context.Context, threadID int CommentsJson: thread.CommentsJSON, RawJson: thread.RawJSON, FetchedAt: thread.FetchedAt, + DeletedAt: nullString(thread.DeletedAt), + DeletionReason: nullString(thread.DeletionReason), }); err != nil { return fmt.Errorf("upsert pull request review thread: %w", err) } + if err := s.recordPullRequestReviewThreadRevision(ctx, threadID, thread.ReviewThreadID, time.Now().UTC().Format(timeLayout)); err != nil { + return err + } + } + return nil +} + +func (s *Store) TombstonePullRequestReviewThread(ctx context.Context, threadID int64, reviewThreadID, deletedAt, reason string) (bool, error) { + if err := validateTombstone(deletedAt, reason); err != nil { + return false, fmt.Errorf("tombstone pull request review thread: %w", err) + } + if deletedAt == "" { + return false, fmt.Errorf("tombstone pull request review thread: deleted_at is required") + } + if s.queries == nil { + var applied bool + err := s.WithTx(ctx, func(tx *Store) error { + var err error + applied, err = tx.tombstonePullRequestReviewThread(ctx, threadID, reviewThreadID, deletedAt, reason) + return err + }) + return applied, err + } + return s.tombstonePullRequestReviewThread(ctx, threadID, reviewThreadID, deletedAt, reason) +} + +func (s *Store) tombstonePullRequestReviewThread(ctx context.Context, threadID int64, reviewThreadID, deletedAt, reason string) (bool, error) { + result, err := s.q().ExecContext(ctx, ` + update pull_request_review_threads + set deleted_at = ?, deletion_reason = ? + where thread_id = ? and review_thread_id = ? + `, deletedAt, reason, threadID, reviewThreadID) + if err != nil { + return false, fmt.Errorf("tombstone pull request review thread: %w", err) + } + if rowsAffected(result) == 0 { + return false, nil + } + if err := s.recordPullRequestReviewThreadRevision(ctx, threadID, reviewThreadID, time.Now().UTC().Format(timeLayout)); err != nil { + return false, err + } + return true, nil +} + +func (s *Store) recordPullRequestReviewThreadRevision(ctx context.Context, threadID int64, reviewThreadID, recordedAt string) error { + if _, err := s.q().ExecContext(ctx, ` + insert into pull_request_review_thread_revisions( + thread_id, review_thread_id, path, line, start_line, is_resolved, + is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, + first_author_login, first_author_type, first_comment_body, + first_comment_url, first_comment_created_at, first_comment_updated_at, + comments_json, raw_json, fetched_at, deleted_at, deletion_reason, recorded_at + ) + select rt.thread_id, rt.review_thread_id, rt.path, rt.line, rt.start_line, + rt.is_resolved, rt.is_outdated, rt.viewer_can_resolve, + rt.viewer_can_unresolve, rt.viewer_can_reply, rt.first_author_login, + rt.first_author_type, rt.first_comment_body, rt.first_comment_url, + rt.first_comment_created_at, rt.first_comment_updated_at, + rt.comments_json, rt.raw_json, rt.fetched_at, rt.deleted_at, + rt.deletion_reason, ? + from pull_request_review_threads rt + where rt.thread_id = ? and rt.review_thread_id = ? + and not exists ( + select 1 + from pull_request_review_thread_revisions r + where r.id = ( + select max(latest.id) + from pull_request_review_thread_revisions latest + where latest.thread_id = rt.thread_id + and latest.review_thread_id = rt.review_thread_id + ) + and r.path is rt.path + and r.line is rt.line + and r.start_line is rt.start_line + and r.is_resolved is rt.is_resolved + and r.is_outdated is rt.is_outdated + and r.viewer_can_resolve is rt.viewer_can_resolve + and r.viewer_can_unresolve is rt.viewer_can_unresolve + and r.viewer_can_reply is rt.viewer_can_reply + and r.first_author_login is rt.first_author_login + and r.first_author_type is rt.first_author_type + and r.first_comment_body is rt.first_comment_body + and r.first_comment_url is rt.first_comment_url + and r.first_comment_created_at is rt.first_comment_created_at + and r.first_comment_updated_at is rt.first_comment_updated_at + and r.comments_json is rt.comments_json + and r.raw_json is rt.raw_json + and r.deleted_at is rt.deleted_at + and r.deletion_reason is rt.deletion_reason + ) + `, recordedAt, threadID, reviewThreadID); err != nil { + return fmt.Errorf("record pull request review thread revision: %w", err) } return nil } @@ -110,6 +218,8 @@ func (s *Store) PullRequestReviewThreads(ctx context.Context, threadID int64) ([ CommentsJSON: row.CommentsJson, RawJSON: row.RawJson, FetchedAt: row.FetchedAt, + DeletedAt: stringValue(row.DeletedAt), + DeletionReason: stringValue(row.DeletionReason), }) } return threads, nil diff --git a/internal/store/review_threads_test.go b/internal/store/review_threads_test.go index ecaf4d21..b8f71d84 100644 --- a/internal/store/review_threads_test.go +++ b/internal/store/review_threads_test.go @@ -7,7 +7,7 @@ import ( "testing" ) -func TestUpsertPullRequestReviewThreadsRollsBackReplacementOnInsertError(t *testing.T) { +func TestUpsertPullRequestReviewThreadsRollsBackMergeOnInsertError(t *testing.T) { ctx := context.Background() st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) if err != nil { @@ -55,7 +55,7 @@ func TestUpsertPullRequestReviewThreadsRollsBackReplacementOnInsertError(t *test t.Fatalf("list review threads: %v", err) } if len(threads) != 1 || threads[0].ReviewThreadID != "old" { - t.Fatalf("review thread replacement should roll back, got %+v", threads) + t.Fatalf("review thread merge should roll back, got %+v", threads) } fetchedAt, err := st.PullRequestReviewThreadsFetchedAt(ctx, threadID) if err != nil { @@ -65,3 +65,94 @@ func TestUpsertPullRequestReviewThreadsRollsBackReplacementOnInsertError(t *test t.Fatalf("fetched marker = %q, want %q", fetchedAt, oldFetchedAt) } } + +func TestPullRequestReviewThreadMergeTombstoneRevisionAndRestore(t *testing.T) { + ctx := context.Background() + st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + defer st.Close() + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "gitcrawl", FullName: "openclaw/gitcrawl", RawJSON: "{}", UpdatedAt: "2026-05-15T00:00:00Z"}) + if err != nil { + t.Fatalf("upsert repo: %v", err) + } + threadID, err := st.UpsertThread(ctx, Thread{ + RepoID: repoID, GitHubID: "2", Number: 2, Kind: "pull_request", State: "open", + Title: "Review thread history", HTMLURL: "https://github.com/openclaw/gitcrawl/pull/2", + LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: "hash", UpdatedAt: "2026-05-15T00:00:00Z", + }) + if err != nil { + t.Fatalf("upsert thread: %v", err) + } + first := PullRequestReviewThread{ + ReviewThreadID: "rt-1", Path: "a.go", Line: 10, CommentsJSON: `[{"body":"first"}]`, + RawJSON: `{"id":"rt-1","version":1}`, FetchedAt: "2026-05-15T00:01:00Z", + } + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, first.FetchedAt, []PullRequestReviewThread{first}); err != nil { + t.Fatalf("seed review thread: %v", err) + } + refetched := first + refetched.FetchedAt = "2026-05-15T00:01:30Z" + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, refetched.FetchedAt, []PullRequestReviewThread{refetched}); err != nil { + t.Fatalf("refetch unchanged review thread: %v", err) + } + var revisions int + if err := st.DB().QueryRowContext(ctx, `select count(*) from pull_request_review_thread_revisions where thread_id = ? and review_thread_id = 'rt-1'`, threadID).Scan(&revisions); err != nil { + t.Fatalf("unchanged review thread revisions: %v", err) + } + if revisions != 1 { + t.Fatalf("unchanged refetch created %d review thread revisions", revisions) + } + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, "2026-05-15T00:02:00Z", nil); err != nil { + t.Fatalf("empty merge: %v", err) + } + threads, err := st.PullRequestReviewThreads(ctx, threadID) + if err != nil || len(threads) != 1 || threads[0].ReviewThreadID != "rt-1" { + t.Fatalf("not-seen review thread was removed: threads=%+v err=%v", threads, err) + } + first.IsResolved = true + first.RawJSON = `{"id":"rt-1","version":2}` + first.FetchedAt = "2026-05-15T00:03:00Z" + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, first.FetchedAt, []PullRequestReviewThread{first}); err != nil { + t.Fatalf("edit review thread: %v", err) + } + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, "2026-05-15T00:04:00Z", []PullRequestReviewThread{{ + ReviewThreadID: "rt-1", DeletedAt: "2026-05-15T00:04:00Z", DeletionReason: "explicit-source-delete", + }}); err != nil { + t.Fatalf("import sparse review thread tombstone: %v", err) + } + threads, err = st.PullRequestReviewThreads(ctx, threadID) + if err != nil || len(threads) != 0 { + t.Fatalf("tombstoned review thread remained visible: threads=%+v err=%v", threads, err) + } + if err := st.DB().QueryRowContext(ctx, `select count(*) from pull_request_review_thread_revisions where thread_id = ? and review_thread_id = 'rt-1'`, threadID).Scan(&revisions); err != nil { + t.Fatalf("review thread revisions: %v", err) + } + if revisions != 3 { + t.Fatalf("review thread revisions = %d, want create/edit/delete", revisions) + } + var retainedPath string + if err := st.DB().QueryRowContext(ctx, `select path from pull_request_review_threads where thread_id = ? and review_thread_id = 'rt-1'`, threadID).Scan(&retainedPath); err != nil { + t.Fatalf("retained review thread path: %v", err) + } + if retainedPath != "a.go" { + t.Fatalf("sparse review thread tombstone replaced path with %q", retainedPath) + } + first.DeletedAt = "" + first.DeletionReason = "" + first.FetchedAt = "2026-05-15T00:05:00Z" + if err := st.UpsertPullRequestReviewThreads(ctx, threadID, first.FetchedAt, []PullRequestReviewThread{first}); err != nil { + t.Fatalf("restore review thread: %v", err) + } + threads, err = st.PullRequestReviewThreads(ctx, threadID) + if err != nil || len(threads) != 1 || !threads[0].IsResolved { + t.Fatalf("restored review thread = %+v err=%v", threads, err) + } + if err := st.DB().QueryRowContext(ctx, `select count(*) from pull_request_review_thread_revisions where thread_id = ? and review_thread_id = 'rt-1'`, threadID).Scan(&revisions); err != nil { + t.Fatalf("restored review thread revisions: %v", err) + } + if revisions != 4 { + t.Fatalf("review thread revisions after restore = %d, want 4", revisions) + } +} diff --git a/internal/store/schema.go b/internal/store/schema.go index 19f10ed8..576e9673 100644 --- a/internal/store/schema.go +++ b/internal/store/schema.go @@ -66,9 +66,30 @@ create table if not exists comments ( raw_json_blob_id integer references blobs(id) on delete set null, created_at_gh text, updated_at_gh text, + deleted_at text, + deletion_reason text, + check ( + (deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '') + ), unique(thread_id, comment_type, github_id) ); +create table if not exists comment_revisions ( + id integer primary key, + comment_id integer not null references comments(id) on delete cascade, + author_login text, + author_type text, + body text not null, + is_bot integer not null default 0, + raw_json text not null, + created_at_gh text, + updated_at_gh text, + deleted_at text, + deletion_reason text, + recorded_at text not null +); + create table if not exists blobs ( id integer primary key, sha256 text not null unique, @@ -211,6 +232,12 @@ create table if not exists pull_request_commits ( html_url text, raw_json text not null, fetched_at text not null, + deleted_at text, + deletion_reason text, + check ( + (deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '') + ), primary key(thread_id, sha) ); @@ -249,9 +276,44 @@ create table if not exists pull_request_review_threads ( comments_json text not null, raw_json text not null, fetched_at text not null, + deleted_at text, + deletion_reason text, + check ( + (deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '') + ), primary key(thread_id, review_thread_id) ); +create table if not exists pull_request_review_thread_revisions ( + id integer primary key, + thread_id integer not null, + review_thread_id text not null, + path text, + line integer not null default 0, + start_line integer not null default 0, + is_resolved integer not null default 0, + is_outdated integer not null default 0, + viewer_can_resolve integer not null default 0, + viewer_can_unresolve integer not null default 0, + viewer_can_reply integer not null default 0, + first_author_login text, + first_author_type text, + first_comment_body text, + first_comment_url text, + first_comment_created_at text, + first_comment_updated_at text, + comments_json text not null, + raw_json text not null, + fetched_at text not null, + deleted_at text, + deletion_reason text, + recorded_at text not null, + foreign key(thread_id, review_thread_id) + references pull_request_review_threads(thread_id, review_thread_id) + on delete cascade +); + create table if not exists pull_request_review_thread_syncs ( thread_id integer primary key references threads(id) on delete cascade, fetched_at text not null @@ -607,6 +669,7 @@ create index if not exists idx_threads_repo_number on threads(repo_id, number); create index if not exists idx_threads_repo_state_closed on threads(repo_id, state, closed_at_local); create index if not exists idx_threads_repo_updated on threads(repo_id, updated_at); create index if not exists idx_comments_thread_type on comments(thread_id, comment_type); +create index if not exists idx_comment_revisions_comment on comment_revisions(comment_id, id); create index if not exists idx_thread_revisions_thread_created on thread_revisions(thread_id, created_at); create index if not exists idx_thread_changed_files_path on thread_changed_files(path); create index if not exists idx_pull_request_details_repo_number on pull_request_details(repo_id, number); @@ -614,6 +677,7 @@ create index if not exists idx_pull_request_files_path on pull_request_files(pat create index if not exists idx_pull_request_files_thread_path on pull_request_files(thread_id, path); create index if not exists idx_pull_request_checks_thread_status on pull_request_checks(thread_id, status, conclusion); create index if not exists idx_pull_request_review_threads_thread_resolved on pull_request_review_threads(thread_id, is_resolved); +create index if not exists idx_pull_request_review_thread_revisions_thread on pull_request_review_thread_revisions(thread_id, review_thread_id, id); create index if not exists idx_pull_request_review_thread_syncs_fetched on pull_request_review_thread_syncs(fetched_at); create index if not exists idx_github_workflow_runs_repo_branch on github_workflow_runs(repo_id, head_branch, run_id); create index if not exists idx_github_workflow_runs_repo_sha on github_workflow_runs(repo_id, head_sha, run_id); diff --git a/internal/store/schema_convergence.go b/internal/store/schema_convergence.go index 7a070540..59698397 100644 --- a/internal/store/schema_convergence.go +++ b/internal/store/schema_convergence.go @@ -21,6 +21,7 @@ const ( migrationWorkflowRunReservationsShape = "workflow_run_observation_reservations_shape" migrationWorkflowRunReservationsBackfill = "workflow_run_observation_reservations_backfill" migrationObservationSchemaConvergence = "observation_schema_convergence" + migrationFamilyTombstoneSchema = "family_tombstone_schema" ) func inspectCompatibilityMigrations( @@ -101,6 +102,9 @@ func inspectCompatibilityMigrationsMode( add(migrationThreadRevisionsCanonicalSchema) } } + if current > 0 && !st.familyTombstoneSchemaHasCurrentShape(ctx) { + add(migrationFamilyTombstoneSchema) + } allocatorCurrent := false if current > 0 && !st.hasTable(ctx, "thread_observation_sequence") { diff --git a/internal/store/sqlc/queries.sql b/internal/store/sqlc/queries.sql index c4a19141..9234304d 100644 --- a/internal/store/sqlc/queries.sql +++ b/internal/store/sqlc/queries.sql @@ -168,8 +168,8 @@ set closed_at_local = null, close_reason_local = null, updated_at = sqlc.arg(upd where repo_id = sqlc.arg(repo_id) and number = sqlc.arg(number); -- name: UpsertComment :one -insert into comments(thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh) -values(sqlc.arg(thread_id), sqlc.arg(github_id), sqlc.arg(comment_type), sqlc.narg(author_login), sqlc.narg(author_type), sqlc.arg(body), sqlc.arg(is_bot), sqlc.arg(raw_json), sqlc.narg(created_at_gh), sqlc.narg(updated_at_gh)) +insert into comments(thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh, deleted_at, deletion_reason) +values(sqlc.arg(thread_id), sqlc.arg(github_id), sqlc.arg(comment_type), sqlc.narg(author_login), sqlc.narg(author_type), sqlc.arg(body), sqlc.arg(is_bot), sqlc.arg(raw_json), sqlc.narg(created_at_gh), sqlc.narg(updated_at_gh), sqlc.narg(deleted_at), sqlc.narg(deletion_reason)) on conflict(thread_id, comment_type, github_id) do update set author_login=excluded.author_login, author_type=excluded.author_type, @@ -177,13 +177,16 @@ on conflict(thread_id, comment_type, github_id) do update set is_bot=excluded.is_bot, raw_json=excluded.raw_json, created_at_gh=excluded.created_at_gh, - updated_at_gh=excluded.updated_at_gh + updated_at_gh=excluded.updated_at_gh, + deleted_at=excluded.deleted_at, + deletion_reason=excluded.deletion_reason returning id; -- name: ListComments :many -select id, thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh +select id, thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh, deleted_at, deletion_reason from comments where thread_id = sqlc.arg(thread_id) + and deleted_at is null order by created_at_gh, id; -- name: UpsertDocument :one @@ -290,12 +293,19 @@ delete from pull_request_files where thread_id = sqlc.arg(thread_id); insert into pull_request_files(thread_id, position, path, status, additions, deletions, changes, previous_path, patch, raw_json, fetched_at) values(sqlc.arg(thread_id), sqlc.arg(position), sqlc.arg(path), sqlc.narg(status), sqlc.arg(additions), sqlc.arg(deletions), sqlc.arg(changes), sqlc.narg(previous_path), sqlc.narg(patch), sqlc.arg(raw_json), sqlc.arg(fetched_at)); --- name: DeletePullRequestCommits :exec -delete from pull_request_commits where thread_id = sqlc.arg(thread_id); - --- name: InsertPullRequestCommit :exec -insert into pull_request_commits(thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at) -values(sqlc.arg(thread_id), sqlc.arg(sha), sqlc.narg(message), sqlc.narg(author_login), sqlc.narg(author_name), sqlc.narg(committed_at), sqlc.narg(html_url), sqlc.arg(raw_json), sqlc.arg(fetched_at)); +-- name: UpsertPullRequestCommit :exec +insert into pull_request_commits(thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at, deleted_at, deletion_reason) +values(sqlc.arg(thread_id), sqlc.arg(sha), sqlc.narg(message), sqlc.narg(author_login), sqlc.narg(author_name), sqlc.narg(committed_at), sqlc.narg(html_url), sqlc.arg(raw_json), sqlc.arg(fetched_at), sqlc.narg(deleted_at), sqlc.narg(deletion_reason)) +on conflict(thread_id, sha) do update set + message=excluded.message, + author_login=excluded.author_login, + author_name=excluded.author_name, + committed_at=excluded.committed_at, + html_url=excluded.html_url, + raw_json=excluded.raw_json, + fetched_at=excluded.fetched_at, + deleted_at=excluded.deleted_at, + deletion_reason=excluded.deletion_reason; -- name: DeletePullRequestChecks :exec delete from pull_request_checks where thread_id = sqlc.arg(thread_id); @@ -333,9 +343,10 @@ where thread_id = sqlc.arg(thread_id) order by path, position; -- name: PullRequestCommits :many -select thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at +select thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at, deleted_at, deletion_reason from pull_request_commits where thread_id = sqlc.arg(thread_id) + and deleted_at is null order by rowid; -- name: PullRequestChecks :many @@ -353,9 +364,6 @@ where repo_id = sqlc.arg(repo_id) order by updated_at_gh desc, run_id desc limit sqlc.arg(row_limit); --- name: DeletePullRequestReviewThreads :exec -delete from pull_request_review_threads where thread_id = sqlc.arg(thread_id); - -- name: UpsertPullRequestReviewThreadSync :exec insert into pull_request_review_thread_syncs(thread_id, fetched_at) values(sqlc.arg(thread_id), sqlc.arg(fetched_at)) @@ -366,13 +374,15 @@ insert into pull_request_review_threads( thread_id, review_thread_id, path, line, start_line, is_resolved, is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, first_author_login, first_author_type, first_comment_body, first_comment_url, - first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at + first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at, + deleted_at, deletion_reason ) values( sqlc.arg(thread_id), sqlc.arg(review_thread_id), sqlc.narg(path), sqlc.arg(line), sqlc.arg(start_line), sqlc.arg(is_resolved), sqlc.arg(is_outdated), sqlc.arg(viewer_can_resolve), sqlc.arg(viewer_can_unresolve), sqlc.arg(viewer_can_reply), sqlc.narg(first_author_login), sqlc.narg(first_author_type), sqlc.narg(first_comment_body), sqlc.narg(first_comment_url), - sqlc.narg(first_comment_created_at), sqlc.narg(first_comment_updated_at), sqlc.arg(comments_json), sqlc.arg(raw_json), sqlc.arg(fetched_at) + sqlc.narg(first_comment_created_at), sqlc.narg(first_comment_updated_at), sqlc.arg(comments_json), sqlc.arg(raw_json), sqlc.arg(fetched_at), + sqlc.narg(deleted_at), sqlc.narg(deletion_reason) ) on conflict(thread_id, review_thread_id) do update set path=excluded.path, @@ -391,15 +401,19 @@ on conflict(thread_id, review_thread_id) do update set first_comment_updated_at=excluded.first_comment_updated_at, comments_json=excluded.comments_json, raw_json=excluded.raw_json, - fetched_at=excluded.fetched_at; + fetched_at=excluded.fetched_at, + deleted_at=excluded.deleted_at, + deletion_reason=excluded.deletion_reason; -- name: PullRequestReviewThreads :many select thread_id, review_thread_id, path, line, start_line, is_resolved, is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, first_author_login, first_author_type, first_comment_body, first_comment_url, - first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at + first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at, + deleted_at, deletion_reason from pull_request_review_threads where thread_id = sqlc.arg(thread_id) + and deleted_at is null order by is_resolved, path, line, review_thread_id; -- name: PullRequestReviewThreadsFetchedAt :one diff --git a/internal/store/sqlc/schema.sql b/internal/store/sqlc/schema.sql index 4f1def1d..801abc83 100644 --- a/internal/store/sqlc/schema.sql +++ b/internal/store/sqlc/schema.sql @@ -59,9 +59,30 @@ create table comments ( raw_json_blob_id integer references blobs(id) on delete set null, created_at_gh text, updated_at_gh text, + deleted_at text, + deletion_reason text, + check ( + (deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '') + ), unique(thread_id, comment_type, github_id) ); +create table comment_revisions ( + id integer primary key, + comment_id integer not null references comments(id) on delete cascade, + author_login text, + author_type text, + body text not null, + is_bot integer not null default 0, + raw_json text not null, + created_at_gh text, + updated_at_gh text, + deleted_at text, + deletion_reason text, + recorded_at text not null +); + create table blobs ( id integer primary key, sha256 text not null unique, @@ -167,6 +188,12 @@ create table pull_request_commits ( html_url text, raw_json text not null, fetched_at text not null, + deleted_at text, + deletion_reason text, + check ( + (deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '') + ), primary key(thread_id, sha) ); @@ -205,9 +232,44 @@ create table pull_request_review_threads ( comments_json text not null, raw_json text not null, fetched_at text not null, + deleted_at text, + deletion_reason text, + check ( + (deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '') + ), primary key(thread_id, review_thread_id) ); +create table pull_request_review_thread_revisions ( + id integer primary key, + thread_id integer not null, + review_thread_id text not null, + path text, + line integer not null default 0, + start_line integer not null default 0, + is_resolved integer not null default 0, + is_outdated integer not null default 0, + viewer_can_resolve integer not null default 0, + viewer_can_unresolve integer not null default 0, + viewer_can_reply integer not null default 0, + first_author_login text, + first_author_type text, + first_comment_body text, + first_comment_url text, + first_comment_created_at text, + first_comment_updated_at text, + comments_json text not null, + raw_json text not null, + fetched_at text not null, + deleted_at text, + deletion_reason text, + recorded_at text not null, + foreign key(thread_id, review_thread_id) + references pull_request_review_threads(thread_id, review_thread_id) + on delete cascade +); + create table pull_request_review_thread_syncs ( thread_id integer primary key references threads(id) on delete cascade, fetched_at text not null diff --git a/internal/store/store.go b/internal/store/store.go index b1ed9b4b..57a2a3be 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -13,7 +13,7 @@ import ( ) const ( - schemaVersion = 12 + schemaVersion = 13 timeLayout = time.RFC3339Nano ) @@ -273,6 +273,9 @@ func (s *Store) migrate(ctx context.Context) error { if err := s.ensureLegacyPortableColumns(ctx); err != nil { return err } + if err := s.ensureFamilyTombstoneSchema(ctx); err != nil { + return err + } if err := s.ensureCanonicalObservationTables(ctx); err != nil { return err } diff --git a/internal/store/storedb/models.go b/internal/store/storedb/models.go index 30d17b26..ee284208 100644 --- a/internal/store/storedb/models.go +++ b/internal/store/storedb/models.go @@ -69,18 +69,35 @@ type CodeSnapshot struct { } type Comment struct { - ID int64 `json:"id"` - ThreadID int64 `json:"thread_id"` - GithubID string `json:"github_id"` - CommentType string `json:"comment_type"` - AuthorLogin sql.NullString `json:"author_login"` - AuthorType sql.NullString `json:"author_type"` - Body string `json:"body"` - IsBot int64 `json:"is_bot"` - RawJson string `json:"raw_json"` - RawJsonBlobID sql.NullInt64 `json:"raw_json_blob_id"` - CreatedAtGh sql.NullString `json:"created_at_gh"` - UpdatedAtGh sql.NullString `json:"updated_at_gh"` + ID int64 `json:"id"` + ThreadID int64 `json:"thread_id"` + GithubID string `json:"github_id"` + CommentType string `json:"comment_type"` + AuthorLogin sql.NullString `json:"author_login"` + AuthorType sql.NullString `json:"author_type"` + Body string `json:"body"` + IsBot int64 `json:"is_bot"` + RawJson string `json:"raw_json"` + RawJsonBlobID sql.NullInt64 `json:"raw_json_blob_id"` + CreatedAtGh sql.NullString `json:"created_at_gh"` + UpdatedAtGh sql.NullString `json:"updated_at_gh"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` +} + +type CommentRevision struct { + ID int64 `json:"id"` + CommentID int64 `json:"comment_id"` + AuthorLogin sql.NullString `json:"author_login"` + AuthorType sql.NullString `json:"author_type"` + Body string `json:"body"` + IsBot int64 `json:"is_bot"` + RawJson string `json:"raw_json"` + CreatedAtGh sql.NullString `json:"created_at_gh"` + UpdatedAtGh sql.NullString `json:"updated_at_gh"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` + RecordedAt string `json:"recorded_at"` } type Document struct { @@ -141,15 +158,17 @@ type PullRequestCheck struct { } type PullRequestCommit struct { - ThreadID int64 `json:"thread_id"` - Sha string `json:"sha"` - Message sql.NullString `json:"message"` - AuthorLogin sql.NullString `json:"author_login"` - AuthorName sql.NullString `json:"author_name"` - CommittedAt sql.NullString `json:"committed_at"` - HtmlUrl sql.NullString `json:"html_url"` - RawJson string `json:"raw_json"` - FetchedAt string `json:"fetched_at"` + ThreadID int64 `json:"thread_id"` + Sha string `json:"sha"` + Message sql.NullString `json:"message"` + AuthorLogin sql.NullString `json:"author_login"` + AuthorName sql.NullString `json:"author_name"` + CommittedAt sql.NullString `json:"committed_at"` + HtmlUrl sql.NullString `json:"html_url"` + RawJson string `json:"raw_json"` + FetchedAt string `json:"fetched_at"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` } type PullRequestDetail struct { @@ -203,6 +222,34 @@ type PullRequestReviewThread struct { CommentsJson string `json:"comments_json"` RawJson string `json:"raw_json"` FetchedAt string `json:"fetched_at"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` +} + +type PullRequestReviewThreadRevision struct { + ID int64 `json:"id"` + ThreadID int64 `json:"thread_id"` + ReviewThreadID string `json:"review_thread_id"` + Path sql.NullString `json:"path"` + Line int64 `json:"line"` + StartLine int64 `json:"start_line"` + IsResolved int64 `json:"is_resolved"` + IsOutdated int64 `json:"is_outdated"` + ViewerCanResolve int64 `json:"viewer_can_resolve"` + ViewerCanUnresolve int64 `json:"viewer_can_unresolve"` + ViewerCanReply int64 `json:"viewer_can_reply"` + FirstAuthorLogin sql.NullString `json:"first_author_login"` + FirstAuthorType sql.NullString `json:"first_author_type"` + FirstCommentBody sql.NullString `json:"first_comment_body"` + FirstCommentUrl sql.NullString `json:"first_comment_url"` + FirstCommentCreatedAt sql.NullString `json:"first_comment_created_at"` + FirstCommentUpdatedAt sql.NullString `json:"first_comment_updated_at"` + CommentsJson string `json:"comments_json"` + RawJson string `json:"raw_json"` + FetchedAt string `json:"fetched_at"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` + RecordedAt string `json:"recorded_at"` } type PullRequestReviewThreadSync struct { @@ -240,6 +287,20 @@ type SummaryRun struct { ErrorText sql.NullString `json:"error_text"` } +type SyncAttemptFailure struct { + ID int64 `json:"id"` + RepoID int64 `json:"repo_id"` + ThreadID sql.NullInt64 `json:"thread_id"` + Number int64 `json:"number"` + Operation string `json:"operation"` + ErrorClass string `json:"error_class"` + ErrorMessage string `json:"error_message"` + FirstSeenAt string `json:"first_seen_at"` + LastSeenAt string `json:"last_seen_at"` + RetryCount int64 `json:"retry_count"` + ResolvedAt sql.NullString `json:"resolved_at"` +} + type SyncRun struct { ID int64 `json:"id"` RepoID int64 `json:"repo_id"` diff --git a/internal/store/storedb/queries.sql.go b/internal/store/storedb/queries.sql.go index 55b5c042..edad7950 100644 --- a/internal/store/storedb/queries.sql.go +++ b/internal/store/storedb/queries.sql.go @@ -89,15 +89,6 @@ func (q *Queries) DeletePullRequestChecks(ctx context.Context, threadID int64) e return err } -const deletePullRequestCommits = `-- name: DeletePullRequestCommits :exec -delete from pull_request_commits where thread_id = ?1 -` - -func (q *Queries) DeletePullRequestCommits(ctx context.Context, threadID int64) error { - _, err := q.db.ExecContext(ctx, deletePullRequestCommits, threadID) - return err -} - const deletePullRequestFiles = `-- name: DeletePullRequestFiles :exec delete from pull_request_files where thread_id = ?1 ` @@ -107,15 +98,6 @@ func (q *Queries) DeletePullRequestFiles(ctx context.Context, threadID int64) er return err } -const deletePullRequestReviewThreads = `-- name: DeletePullRequestReviewThreads :exec -delete from pull_request_review_threads where thread_id = ?1 -` - -func (q *Queries) DeletePullRequestReviewThreads(ctx context.Context, threadID int64) error { - _, err := q.db.ExecContext(ctx, deletePullRequestReviewThreads, threadID) - return err -} - const insertPullRequestCheck = `-- name: InsertPullRequestCheck :exec insert into pull_request_checks(thread_id, name, status, conclusion, details_url, workflow_name, started_at, completed_at, raw_json, fetched_at) values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) @@ -150,38 +132,6 @@ func (q *Queries) InsertPullRequestCheck(ctx context.Context, arg InsertPullRequ return err } -const insertPullRequestCommit = `-- name: InsertPullRequestCommit :exec -insert into pull_request_commits(thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at) -values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) -` - -type InsertPullRequestCommitParams struct { - ThreadID int64 `json:"thread_id"` - Sha string `json:"sha"` - Message sql.NullString `json:"message"` - AuthorLogin sql.NullString `json:"author_login"` - AuthorName sql.NullString `json:"author_name"` - CommittedAt sql.NullString `json:"committed_at"` - HtmlUrl sql.NullString `json:"html_url"` - RawJson string `json:"raw_json"` - FetchedAt string `json:"fetched_at"` -} - -func (q *Queries) InsertPullRequestCommit(ctx context.Context, arg InsertPullRequestCommitParams) error { - _, err := q.db.ExecContext(ctx, insertPullRequestCommit, - arg.ThreadID, - arg.Sha, - arg.Message, - arg.AuthorLogin, - arg.AuthorName, - arg.CommittedAt, - arg.HtmlUrl, - arg.RawJson, - arg.FetchedAt, - ) - return err -} - const insertPullRequestFile = `-- name: InsertPullRequestFile :exec insert into pull_request_files(thread_id, position, path, status, additions, deletions, changes, previous_path, patch, raw_json, fetched_at) values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) @@ -301,24 +251,27 @@ func (q *Queries) ListClusterRuns(ctx context.Context, arg ListClusterRunsParams } const listComments = `-- name: ListComments :many -select id, thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh +select id, thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh, deleted_at, deletion_reason from comments where thread_id = ?1 + and deleted_at is null order by created_at_gh, id ` type ListCommentsRow struct { - ID int64 `json:"id"` - ThreadID int64 `json:"thread_id"` - GithubID string `json:"github_id"` - CommentType string `json:"comment_type"` - AuthorLogin sql.NullString `json:"author_login"` - AuthorType sql.NullString `json:"author_type"` - Body string `json:"body"` - IsBot int64 `json:"is_bot"` - RawJson string `json:"raw_json"` - CreatedAtGh sql.NullString `json:"created_at_gh"` - UpdatedAtGh sql.NullString `json:"updated_at_gh"` + ID int64 `json:"id"` + ThreadID int64 `json:"thread_id"` + GithubID string `json:"github_id"` + CommentType string `json:"comment_type"` + AuthorLogin sql.NullString `json:"author_login"` + AuthorType sql.NullString `json:"author_type"` + Body string `json:"body"` + IsBot int64 `json:"is_bot"` + RawJson string `json:"raw_json"` + CreatedAtGh sql.NullString `json:"created_at_gh"` + UpdatedAtGh sql.NullString `json:"updated_at_gh"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` } func (q *Queries) ListComments(ctx context.Context, threadID int64) ([]ListCommentsRow, error) { @@ -342,6 +295,8 @@ func (q *Queries) ListComments(ctx context.Context, threadID int64) ([]ListComme &i.RawJson, &i.CreatedAtGh, &i.UpdatedAtGh, + &i.DeletedAt, + &i.DeletionReason, ); err != nil { return nil, err } @@ -830,9 +785,10 @@ func (q *Queries) PullRequestChecks(ctx context.Context, threadID int64) ([]Pull } const pullRequestCommits = `-- name: PullRequestCommits :many -select thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at +select thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at, deleted_at, deletion_reason from pull_request_commits where thread_id = ?1 + and deleted_at is null order by rowid ` @@ -855,6 +811,8 @@ func (q *Queries) PullRequestCommits(ctx context.Context, threadID int64) ([]Pul &i.HtmlUrl, &i.RawJson, &i.FetchedAt, + &i.DeletedAt, + &i.DeletionReason, ); err != nil { return nil, err } @@ -948,9 +906,11 @@ const pullRequestReviewThreads = `-- name: PullRequestReviewThreads :many select thread_id, review_thread_id, path, line, start_line, is_resolved, is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, first_author_login, first_author_type, first_comment_body, first_comment_url, - first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at + first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at, + deleted_at, deletion_reason from pull_request_review_threads where thread_id = ?1 + and deleted_at is null order by is_resolved, path, line, review_thread_id ` @@ -983,6 +943,8 @@ func (q *Queries) PullRequestReviewThreads(ctx context.Context, threadID int64) &i.CommentsJson, &i.RawJson, &i.FetchedAt, + &i.DeletedAt, + &i.DeletionReason, ); err != nil { return nil, err } @@ -1192,8 +1154,8 @@ func (q *Queries) RepositoryByFullName(ctx context.Context, fullName string) (Re } const upsertComment = `-- name: UpsertComment :one -insert into comments(thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh) -values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) +insert into comments(thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, created_at_gh, updated_at_gh, deleted_at, deletion_reason) +values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) on conflict(thread_id, comment_type, github_id) do update set author_login=excluded.author_login, author_type=excluded.author_type, @@ -1201,21 +1163,25 @@ on conflict(thread_id, comment_type, github_id) do update set is_bot=excluded.is_bot, raw_json=excluded.raw_json, created_at_gh=excluded.created_at_gh, - updated_at_gh=excluded.updated_at_gh + updated_at_gh=excluded.updated_at_gh, + deleted_at=excluded.deleted_at, + deletion_reason=excluded.deletion_reason returning id ` type UpsertCommentParams struct { - ThreadID int64 `json:"thread_id"` - GithubID string `json:"github_id"` - CommentType string `json:"comment_type"` - AuthorLogin sql.NullString `json:"author_login"` - AuthorType sql.NullString `json:"author_type"` - Body string `json:"body"` - IsBot int64 `json:"is_bot"` - RawJson string `json:"raw_json"` - CreatedAtGh sql.NullString `json:"created_at_gh"` - UpdatedAtGh sql.NullString `json:"updated_at_gh"` + ThreadID int64 `json:"thread_id"` + GithubID string `json:"github_id"` + CommentType string `json:"comment_type"` + AuthorLogin sql.NullString `json:"author_login"` + AuthorType sql.NullString `json:"author_type"` + Body string `json:"body"` + IsBot int64 `json:"is_bot"` + RawJson string `json:"raw_json"` + CreatedAtGh sql.NullString `json:"created_at_gh"` + UpdatedAtGh sql.NullString `json:"updated_at_gh"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` } func (q *Queries) UpsertComment(ctx context.Context, arg UpsertCommentParams) (int64, error) { @@ -1230,6 +1196,8 @@ func (q *Queries) UpsertComment(ctx context.Context, arg UpsertCommentParams) (i arg.RawJson, arg.CreatedAtGh, arg.UpdatedAtGh, + arg.DeletedAt, + arg.DeletionReason, ) var id int64 err := row.Scan(&id) @@ -1275,6 +1243,52 @@ func (q *Queries) UpsertDocument(ctx context.Context, arg UpsertDocumentParams) return id, err } +const upsertPullRequestCommit = `-- name: UpsertPullRequestCommit :exec +insert into pull_request_commits(thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at, deleted_at, deletion_reason) +values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) +on conflict(thread_id, sha) do update set + message=excluded.message, + author_login=excluded.author_login, + author_name=excluded.author_name, + committed_at=excluded.committed_at, + html_url=excluded.html_url, + raw_json=excluded.raw_json, + fetched_at=excluded.fetched_at, + deleted_at=excluded.deleted_at, + deletion_reason=excluded.deletion_reason +` + +type UpsertPullRequestCommitParams struct { + ThreadID int64 `json:"thread_id"` + Sha string `json:"sha"` + Message sql.NullString `json:"message"` + AuthorLogin sql.NullString `json:"author_login"` + AuthorName sql.NullString `json:"author_name"` + CommittedAt sql.NullString `json:"committed_at"` + HtmlUrl sql.NullString `json:"html_url"` + RawJson string `json:"raw_json"` + FetchedAt string `json:"fetched_at"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` +} + +func (q *Queries) UpsertPullRequestCommit(ctx context.Context, arg UpsertPullRequestCommitParams) error { + _, err := q.db.ExecContext(ctx, upsertPullRequestCommit, + arg.ThreadID, + arg.Sha, + arg.Message, + arg.AuthorLogin, + arg.AuthorName, + arg.CommittedAt, + arg.HtmlUrl, + arg.RawJson, + arg.FetchedAt, + arg.DeletedAt, + arg.DeletionReason, + ) + return err +} + const upsertPullRequestDetail = `-- name: UpsertPullRequestDetail :exec insert into pull_request_details(thread_id, repo_id, number, base_sha, head_sha, head_ref, head_repo_full_name, mergeable_state, additions, deletions, changed_files, raw_json, fetched_at, updated_at) values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) @@ -1336,13 +1350,15 @@ insert into pull_request_review_threads( thread_id, review_thread_id, path, line, start_line, is_resolved, is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, first_author_login, first_author_type, first_comment_body, first_comment_url, - first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at + first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at, + deleted_at, deletion_reason ) values( ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, - ?15, ?16, ?17, ?18, ?19 + ?15, ?16, ?17, ?18, ?19, + ?20, ?21 ) on conflict(thread_id, review_thread_id) do update set path=excluded.path, @@ -1361,7 +1377,9 @@ on conflict(thread_id, review_thread_id) do update set first_comment_updated_at=excluded.first_comment_updated_at, comments_json=excluded.comments_json, raw_json=excluded.raw_json, - fetched_at=excluded.fetched_at + fetched_at=excluded.fetched_at, + deleted_at=excluded.deleted_at, + deletion_reason=excluded.deletion_reason ` type UpsertPullRequestReviewThreadParams struct { @@ -1384,6 +1402,8 @@ type UpsertPullRequestReviewThreadParams struct { CommentsJson string `json:"comments_json"` RawJson string `json:"raw_json"` FetchedAt string `json:"fetched_at"` + DeletedAt sql.NullString `json:"deleted_at"` + DeletionReason sql.NullString `json:"deletion_reason"` } func (q *Queries) UpsertPullRequestReviewThread(ctx context.Context, arg UpsertPullRequestReviewThreadParams) error { @@ -1407,6 +1427,8 @@ func (q *Queries) UpsertPullRequestReviewThread(ctx context.Context, arg UpsertP arg.CommentsJson, arg.RawJson, arg.FetchedAt, + arg.DeletedAt, + arg.DeletionReason, ) return err } diff --git a/internal/store/storedb/queries_test.go b/internal/store/storedb/queries_test.go index 40f22bc5..16a8a126 100644 --- a/internal/store/storedb/queries_test.go +++ b/internal/store/storedb/queries_test.go @@ -151,7 +151,7 @@ func TestGeneratedQueriesRoundTrip(t *testing.T) { if rows, err := q.PullRequestFiles(ctx, threadID); err != nil || len(rows) != 1 { t.Fatalf("pull request files len = %d, %v", len(rows), err) } - if err := q.InsertPullRequestCommit(ctx, storedb.InsertPullRequestCommitParams{ + if err := q.UpsertPullRequestCommit(ctx, storedb.UpsertPullRequestCommitParams{ ThreadID: threadID, Sha: "abc123", Message: ns("commit"), AuthorLogin: ns("alice"), AuthorName: ns("Alice"), CommittedAt: ns(now), HtmlUrl: ns("https://github.com/openclaw/gitcrawl/commit/abc123"), RawJson: `{"commit":true}`, FetchedAt: now, }); err != nil { @@ -233,13 +233,7 @@ func TestGeneratedQueriesRoundTrip(t *testing.T) { if err := q.DeletePullRequestChecks(ctx, threadID); err != nil { t.Fatalf("delete pr checks: %v", err) } - if err := q.DeletePullRequestCommits(ctx, threadID); err != nil { - t.Fatalf("delete pr commits: %v", err) - } if err := q.DeletePullRequestFiles(ctx, threadID); err != nil { t.Fatalf("delete pr files: %v", err) } - if err := q.DeletePullRequestReviewThreads(ctx, threadID); err != nil { - t.Fatalf("delete review threads: %v", err) - } } diff --git a/internal/store/tombstone_schema.go b/internal/store/tombstone_schema.go new file mode 100644 index 00000000..c0a2f7fb --- /dev/null +++ b/internal/store/tombstone_schema.go @@ -0,0 +1,257 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "strings" +) + +func (s *Store) familyTombstoneSchemaHasCurrentShape(ctx context.Context) bool { + for _, table := range []string{"comments", "pull_request_commits", "pull_request_review_threads"} { + if !s.hasColumns(ctx, table, "deleted_at", "deletion_reason") || + !s.tableHasFamilyTombstoneConstraint(ctx, table) { + return false + } + } + return s.hasTable(ctx, "comment_revisions") && + s.hasTable(ctx, "pull_request_review_thread_revisions") +} + +func (s *Store) ensureFamilyTombstoneSchema(ctx context.Context) error { + for _, table := range []string{"comments", "pull_request_commits", "pull_request_review_threads"} { + if err := s.ensureColumn(ctx, table, "deleted_at", "text"); err != nil { + return err + } + if err := s.ensureColumn(ctx, table, "deletion_reason", "text"); err != nil { + return err + } + } + if err := s.ensureFamilyTombstoneConstraints(ctx); err != nil { + return err + } + if _, err := s.db.ExecContext(ctx, ` + create table if not exists comment_revisions ( + id integer primary key, + comment_id integer not null references comments(id) on delete cascade, + author_login text, + author_type text, + body text not null, + is_bot integer not null default 0, + raw_json text not null, + created_at_gh text, + updated_at_gh text, + deleted_at text, + deletion_reason text, + recorded_at text not null + ); + create index if not exists idx_comment_revisions_comment + on comment_revisions(comment_id, id); + create table if not exists pull_request_review_thread_revisions ( + id integer primary key, + thread_id integer not null, + review_thread_id text not null, + path text, + line integer not null default 0, + start_line integer not null default 0, + is_resolved integer not null default 0, + is_outdated integer not null default 0, + viewer_can_resolve integer not null default 0, + viewer_can_unresolve integer not null default 0, + viewer_can_reply integer not null default 0, + first_author_login text, + first_author_type text, + first_comment_body text, + first_comment_url text, + first_comment_created_at text, + first_comment_updated_at text, + comments_json text not null, + raw_json text not null, + fetched_at text not null, + deleted_at text, + deletion_reason text, + recorded_at text not null, + foreign key(thread_id, review_thread_id) + references pull_request_review_threads(thread_id, review_thread_id) + on delete cascade + ); + create index if not exists idx_pull_request_review_thread_revisions_thread + on pull_request_review_thread_revisions(thread_id, review_thread_id, id) + `); err != nil { + return fmt.Errorf("ensure family revision tables: %w", err) + } + if _, err := s.db.ExecContext(ctx, ` + insert into comment_revisions( + comment_id, author_login, author_type, body, is_bot, raw_json, + created_at_gh, updated_at_gh, deleted_at, deletion_reason, recorded_at + ) + select c.id, c.author_login, c.author_type, c.body, c.is_bot, c.raw_json, + c.created_at_gh, c.updated_at_gh, c.deleted_at, c.deletion_reason, + coalesce(nullif(c.updated_at_gh, ''), nullif(c.created_at_gh, ''), '1970-01-01T00:00:00Z') + from comments c + where not exists ( + select 1 from comment_revisions r where r.comment_id = c.id + ) + `); err != nil { + return fmt.Errorf("backfill comment revisions: %w", err) + } + if _, err := s.db.ExecContext(ctx, ` + insert into pull_request_review_thread_revisions( + thread_id, review_thread_id, path, line, start_line, is_resolved, + is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, + first_author_login, first_author_type, first_comment_body, + first_comment_url, first_comment_created_at, first_comment_updated_at, + comments_json, raw_json, fetched_at, deleted_at, deletion_reason, recorded_at + ) + select rt.thread_id, rt.review_thread_id, rt.path, rt.line, rt.start_line, + rt.is_resolved, rt.is_outdated, rt.viewer_can_resolve, + rt.viewer_can_unresolve, rt.viewer_can_reply, rt.first_author_login, + rt.first_author_type, rt.first_comment_body, rt.first_comment_url, + rt.first_comment_created_at, rt.first_comment_updated_at, + rt.comments_json, rt.raw_json, rt.fetched_at, rt.deleted_at, + rt.deletion_reason, rt.fetched_at + from pull_request_review_threads rt + where not exists ( + select 1 from pull_request_review_thread_revisions r + where r.thread_id = rt.thread_id + and r.review_thread_id = rt.review_thread_id + ) + `); err != nil { + return fmt.Errorf("backfill pull request review thread revisions: %w", err) + } + return nil +} + +func (s *Store) tableHasFamilyTombstoneConstraint(ctx context.Context, table string) bool { + var definition string + if err := s.db.QueryRowContext(ctx, ` + select sql from sqlite_master where type = 'table' and name = ? + `, table).Scan(&definition); err != nil { + return false + } + normalized := strings.Join(strings.Fields(strings.ToLower(definition)), " ") + return strings.Contains(normalized, "(deleted_at is null and deletion_reason is null)") && + strings.Contains(normalized, "(deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '')") +} + +func (s *Store) ensureFamilyTombstoneConstraints(ctx context.Context) error { + tables := []string{"comments", "pull_request_commits", "pull_request_review_threads"} + rebuild := make(map[string]bool, len(tables)) + for _, table := range tables { + if !s.tableHasFamilyTombstoneConstraint(ctx, table) { + rebuild[table] = true + } + } + if len(rebuild) == 0 { + return nil + } + return s.withForeignKeysDisabled(ctx, "family tombstone constraints", func(tx *sql.Tx) error { + for _, migration := range []struct { + table string + definition string + columns string + indexes []string + }{ + { + table: "comments", + definition: `( + id integer primary key, + thread_id integer not null references threads(id) on delete cascade, + github_id text not null, + comment_type text not null, + author_login text, + author_type text, + body text not null, + is_bot integer not null default 0, + raw_json text not null, + raw_json_blob_id integer references blobs(id) on delete set null, + created_at_gh text, + updated_at_gh text, + deleted_at text, + deletion_reason text, + check ((deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '')), + unique(thread_id, comment_type, github_id) + )`, + columns: "id, thread_id, github_id, comment_type, author_login, author_type, body, is_bot, raw_json, raw_json_blob_id, created_at_gh, updated_at_gh, deleted_at, deletion_reason", + indexes: []string{ + `create index if not exists idx_comments_thread_type on comments(thread_id, comment_type)`, + }, + }, + { + table: "pull_request_commits", + definition: `( + thread_id integer not null references threads(id) on delete cascade, + sha text not null, + message text, + author_login text, + author_name text, + committed_at text, + html_url text, + raw_json text not null, + fetched_at text not null, + deleted_at text, + deletion_reason text, + check ((deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '')), + primary key(thread_id, sha) + )`, + columns: "thread_id, sha, message, author_login, author_name, committed_at, html_url, raw_json, fetched_at, deleted_at, deletion_reason", + }, + { + table: "pull_request_review_threads", + definition: `( + thread_id integer not null references threads(id) on delete cascade, + review_thread_id text not null, + path text, + line integer not null default 0, + start_line integer not null default 0, + is_resolved integer not null default 0, + is_outdated integer not null default 0, + viewer_can_resolve integer not null default 0, + viewer_can_unresolve integer not null default 0, + viewer_can_reply integer not null default 0, + first_author_login text, + first_author_type text, + first_comment_body text, + first_comment_url text, + first_comment_created_at text, + first_comment_updated_at text, + comments_json text not null, + raw_json text not null, + fetched_at text not null, + deleted_at text, + deletion_reason text, + check ((deleted_at is null and deletion_reason is null) + or (deleted_at is not null and deletion_reason is not null and trim(deletion_reason) <> '')), + primary key(thread_id, review_thread_id) + )`, + columns: "thread_id, review_thread_id, path, line, start_line, is_resolved, is_outdated, viewer_can_resolve, viewer_can_unresolve, viewer_can_reply, first_author_login, first_author_type, first_comment_body, first_comment_url, first_comment_created_at, first_comment_updated_at, comments_json, raw_json, fetched_at, deleted_at, deletion_reason", + indexes: []string{ + `create index if not exists idx_pull_request_review_threads_thread_resolved on pull_request_review_threads(thread_id, is_resolved)`, + }, + }, + } { + if !rebuild[migration.table] { + continue + } + newTable := migration.table + "_tombstone_new" + for _, statement := range []string{ + fmt.Sprintf("create table %s %s", newTable, migration.definition), + fmt.Sprintf("insert into %s(%s) select %s from %s", newTable, migration.columns, migration.columns, migration.table), + fmt.Sprintf("drop table %s", migration.table), + fmt.Sprintf("alter table %s rename to %s", newTable, migration.table), + } { + if _, err := tx.ExecContext(ctx, statement); err != nil { + return fmt.Errorf("rebuild %s: %w", migration.table, err) + } + } + for _, statement := range migration.indexes { + if _, err := tx.ExecContext(ctx, statement); err != nil { + return fmt.Errorf("recreate %s index: %w", migration.table, err) + } + } + } + return nil + }) +} diff --git a/internal/syncer/pull_details.go b/internal/syncer/pull_details.go index 4f29154d..c2b152cc 100644 --- a/internal/syncer/pull_details.go +++ b/internal/syncer/pull_details.go @@ -695,15 +695,17 @@ func mapPullCommits(threadID int64, rows []map[string]any, fetchedAt string) []s continue } out = append(out, store.PullRequestCommit{ - ThreadID: threadID, - SHA: sha, - Message: nestedString(row, "commit", "message"), - AuthorLogin: nestedString(row, "author", "login"), - AuthorName: nestedString(row, "commit", "author", "name"), - CommittedAt: nestedString(row, "commit", "author", "date"), - HTMLURL: stringValue(row["html_url"]), - RawJSON: mustJSON(row), - FetchedAt: fetchedAt, + ThreadID: threadID, + SHA: sha, + Message: nestedString(row, "commit", "message"), + AuthorLogin: nestedString(row, "author", "login"), + AuthorName: nestedString(row, "commit", "author", "name"), + CommittedAt: nestedString(row, "commit", "author", "date"), + HTMLURL: stringValue(row["html_url"]), + RawJSON: mustJSON(row), + FetchedAt: fetchedAt, + DeletedAt: stringValue(row["deleted_at"]), + DeletionReason: stringValue(row["deletion_reason"]), }) } return out diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index b39d5af9..88de407a 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -345,11 +345,12 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { } var comments []store.Comment if options.IncludeComments && childReservations[store.ThreadChildComments] { - comments, err = persistComments(ctx, st, thread, payload.commentRows) + var synced int + comments, synced, err = persistComments(ctx, st, thread, payload.commentRows) if err != nil { return err } - attempt.CommentsSynced += len(comments) + attempt.CommentsSynced += synced } else { var err error comments, err = st.ListComments(ctx, thread.ID) @@ -793,22 +794,23 @@ func (s *Syncer) fetchCommentRows(ctx context.Context, options Options, threadKi return rows, nil } -func persistComments(ctx context.Context, st *store.Store, thread store.Thread, rows []commentRow) ([]store.Comment, error) { - if err := st.DeleteCommentsForThread(ctx, thread.ID); err != nil { - return nil, err - } - comments := make([]store.Comment, 0, len(rows)) +func persistComments(ctx context.Context, st *store.Store, thread store.Thread, rows []commentRow) ([]store.Comment, int, error) { + synced := 0 for _, row := range rows { comment := mapComment(thread.ID, row.kind, row.raw) - if comment.Body == "" && row.kind != "pull_review" { + if comment.Body == "" && row.kind != "pull_review" && comment.DeletedAt == "" { continue } if _, err := st.UpsertComment(ctx, comment); err != nil { - return nil, err + return nil, 0, err } - comments = append(comments, comment) + synced++ + } + comments, err := st.ListComments(ctx, thread.ID) + if err != nil { + return nil, 0, err } - return comments, nil + return comments, synced, nil } func (s *Syncer) fetchPullReviewThreadRows(ctx context.Context, options Options, number int) ([]map[string]any, string, error) { @@ -865,6 +867,8 @@ func mapPullReviewThread(threadID int64, row map[string]any, fetchedAt string) s CommentsJSON: mustJSON(comments), RawJSON: mustJSON(row), FetchedAt: fetchedAt, + DeletedAt: stringValue(row["deleted_at"]), + DeletionReason: stringValue(row["deletion_reason"]), } } @@ -888,6 +892,8 @@ func mapComment(threadID int64, kind string, row map[string]any) store.Comment { RawJSON: mustJSON(row), CreatedAtGitHub: stringValue(row["created_at"]), UpdatedAtGitHub: stringValue(row["updated_at"]), + DeletedAt: stringValue(row["deleted_at"]), + DeletionReason: stringValue(row["deletion_reason"]), } } diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index b417b4f9..9c552f36 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -1781,12 +1781,8 @@ func TestSyncPartialPRCommentsUseParentObservationGeneration(t *testing.T) { if err != nil { t.Fatalf("comments: %v", err) } - wantBody := "comment-v2" - if test.secondPersistsFirst { - wantBody = "comment-v1" - } - if len(comments) != 1 || comments[0].Body != wantBody { - t.Fatalf("comments = %+v, want latest completed snapshot %s", comments, wantBody) + if len(comments) != 2 || comments[0].Body == comments[1].Body { + t.Fatalf("comments = %+v, want both independently observed rows", comments) } assertChildReservation( t, @@ -3010,8 +3006,13 @@ func assertVersionedPRHydration( if err != nil { t.Fatalf("commits: %v", err) } - if len(commits) != 1 || commits[0].SHA != fmt.Sprintf("commit-v%d", version) { - t.Fatalf("commits = %+v, want version %d", commits, version) + wantCommit := fmt.Sprintf("commit-v%d", version) + foundCommit := false + for _, commit := range commits { + foundCommit = foundCommit || commit.SHA == wantCommit + } + if !foundCommit { + t.Fatalf("commits = %+v, want merged row %s", commits, wantCommit) } checks, err := st.PullRequestChecks(ctx, thread.ID) if err != nil { @@ -3024,9 +3025,13 @@ func assertVersionedPRHydration( if err != nil { t.Fatalf("review threads: %v", err) } - if len(reviewThreads) != 1 || - reviewThreads[0].ReviewThreadID != fmt.Sprintf("review-thread-v%d", version) { - t.Fatalf("review threads = %+v, want version %d", reviewThreads, version) + wantReviewThread := fmt.Sprintf("review-thread-v%d", version) + foundReviewThread := false + for _, reviewThread := range reviewThreads { + foundReviewThread = foundReviewThread || reviewThread.ReviewThreadID == wantReviewThread + } + if !foundReviewThread { + t.Fatalf("review threads = %+v, want merged row %s", reviewThreads, wantReviewThread) } runs, err := st.ListWorkflowRuns(ctx, repo.ID, store.WorkflowRunListOptions{ HeadSHA: detail.HeadSHA, @@ -3321,7 +3326,7 @@ func assertStoredPullDraft(t *testing.T, ctx context.Context, st *store.Store, w t.Fatal("pull request was not stored") } -func TestCommentHydrationReplacesDeletedCommentsWithEmptySnapshot(t *testing.T) { +func TestCommentHydrationDoesNotTreatNotSeenAsDeleted(t *testing.T) { ctx := context.Background() st, err := store.Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) if err != nil { @@ -3343,11 +3348,11 @@ func TestCommentHydrationReplacesDeletedCommentsWithEmptySnapshot(t *testing.T) if err != nil { t.Fatalf("empty comment sync: %v", err) } - if stats.CommentsSynced != 0 || stats.RevisionsCreated != 1 { + if stats.CommentsSynced != 0 || stats.RevisionsCreated != 0 { t.Fatalf("empty comment sync stats = %#v", stats) } - assertTableRowCount(t, st, "comments", 0) - assertDocumentFTSCount(t, st, "same", 0) + assertTableRowCount(t, st, "comments", 1) + assertDocumentFTSCount(t, st, "same", 1) coverage, err := st.ArchiveCoverage(ctx, store.ArchiveCoverageOptions{}) if err != nil { t.Fatalf("archive coverage after empty snapshot: %v", err) @@ -3361,8 +3366,51 @@ func TestCommentHydrationReplacesDeletedCommentsWithEmptySnapshot(t *testing.T) if _, err := s.Sync(ctx, Options{Owner: "openclaw", Repo: "gitcrawl", Numbers: []int{7}}); err != nil { t.Fatalf("metadata sync after empty snapshot: %v", err) } - assertTableRowCount(t, st, "comments", 0) - assertDocumentFTSCount(t, st, "same", 0) + assertTableRowCount(t, st, "comments", 1) + assertDocumentFTSCount(t, st, "same", 1) +} + +func TestPersistCommentsAppliesSparseExplicitTombstone(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + defer st.Close() + repoID, err := st.UpsertRepository(ctx, store.Repository{Owner: "openclaw", Name: "gitcrawl", FullName: "openclaw/gitcrawl", RawJSON: "{}", UpdatedAt: "2026-07-18T00:00:00Z"}) + if err != nil { + t.Fatalf("seed repo: %v", err) + } + thread := store.Thread{ + RepoID: repoID, GitHubID: "18", Number: 18, Kind: "issue", State: "open", + Title: "sparse tombstone", HTMLURL: "https://github.com/openclaw/gitcrawl/issues/18", + LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: "hash", UpdatedAt: "2026-07-18T00:00:00Z", + } + thread.ID, err = st.UpsertThread(ctx, thread) + if err != nil { + t.Fatalf("seed thread: %v", err) + } + if _, _, err := persistComments(ctx, st, thread, []commentRow{{kind: "issue_comment", raw: map[string]any{ + "id": 1801, "body": "retain this body", "created_at": "2026-07-18T00:01:00Z", + }}}); err != nil { + t.Fatalf("seed comment: %v", err) + } + comments, synced, err := persistComments(ctx, st, thread, []commentRow{{kind: "issue_comment", raw: map[string]any{ + "id": 1801, "deleted_at": "2026-07-18T00:02:00Z", "deletion_reason": "explicit-source-delete", + }}}) + if err != nil { + t.Fatalf("persist sparse tombstone: %v", err) + } + if synced != 1 || len(comments) != 0 { + t.Fatalf("sparse tombstone result: synced=%d comments=%+v", synced, comments) + } + var body, deletedAt, reason string + if err := st.DB().QueryRowContext(ctx, `select body, deleted_at, deletion_reason from comments where thread_id = ? and github_id = '1801'`, thread.ID).Scan(&body, &deletedAt, &reason); err != nil { + t.Fatalf("read sparse tombstone: %v", err) + } + if body != "retain this body" || deletedAt == "" || reason != "explicit-source-delete" { + t.Fatalf("sparse tombstone row = body %q deleted_at %q reason %q", body, deletedAt, reason) + } } func TestSyncHydratesPullReviewComments(t *testing.T) { From 3ff1b9e55613c6fd5925aee47269ad5fe113ad93 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 15:40:27 -0700 Subject: [PATCH 2/2] test(store): cover tombstone APIs --- internal/store/comments_test.go | 17 +++++++++++++++++ internal/store/pull_requests_test.go | 11 +++++++++++ internal/store/review_threads_test.go | 11 +++++++++++ 3 files changed, 39 insertions(+) diff --git a/internal/store/comments_test.go b/internal/store/comments_test.go index 1630406b..ea7b6bbb 100644 --- a/internal/store/comments_test.go +++ b/internal/store/comments_test.go @@ -151,4 +151,21 @@ func TestTombstoneFieldsRejectSurroundingWhitespace(t *testing.T) { } }) } + if _, err := st.UpsertComment(ctx, Comment{ + ThreadID: threadID, GitHubID: "direct", CommentType: "issue_comment", + Body: "direct tombstone", RawJSON: `{}`, + }); err != nil { + t.Fatalf("seed direct tombstone comment: %v", err) + } + applied, err := st.TombstoneComment(ctx, threadID, "issue_comment", "direct", "2026-05-01T00:02:00Z", "explicit-source-delete") + if err != nil || !applied { + t.Fatalf("direct comment tombstone = %t, %v", applied, err) + } + applied, err = st.TombstoneComment(ctx, threadID, "issue_comment", "missing", "2026-05-01T00:02:00Z", "explicit-source-delete") + if err != nil || applied { + t.Fatalf("missing comment tombstone = %t, %v", applied, err) + } + if _, err := st.TombstoneComment(ctx, threadID, "issue_comment", "direct", "", ""); err == nil || !strings.Contains(err.Error(), "deleted_at is required") { + t.Fatalf("empty comment tombstone error = %v", err) + } } diff --git a/internal/store/pull_requests_test.go b/internal/store/pull_requests_test.go index ffec4456..2aa756de 100644 --- a/internal/store/pull_requests_test.go +++ b/internal/store/pull_requests_test.go @@ -128,6 +128,17 @@ func TestPullRequestCacheRoundTripAndWorkflowFilters(t *testing.T) { if err != nil || len(restoredCommits) != 1 || restoredCommits[0].SHA != "abc" { t.Fatalf("restored commits = %+v err=%v", restoredCommits, err) } + applied, err := st.TombstonePullRequestCommit(ctx, threadID, "abc", "2026-05-05T10:02:00Z", "explicit-source-delete") + if err != nil || !applied { + t.Fatalf("direct commit tombstone = %t, %v", applied, err) + } + applied, err = st.TombstonePullRequestCommit(ctx, threadID, "missing", "2026-05-05T10:02:00Z", "explicit-source-delete") + if err != nil || applied { + t.Fatalf("missing commit tombstone = %t, %v", applied, err) + } + if _, err := st.TombstonePullRequestCommit(ctx, threadID, "abc", "", ""); err == nil || !strings.Contains(err.Error(), "deleted_at is required") { + t.Fatalf("empty commit tombstone error = %v", err) + } } func TestMissingPullRequestDetailNumbersSelectsOnlyUnfilledPRs(t *testing.T) { diff --git a/internal/store/review_threads_test.go b/internal/store/review_threads_test.go index b8f71d84..08b3f871 100644 --- a/internal/store/review_threads_test.go +++ b/internal/store/review_threads_test.go @@ -155,4 +155,15 @@ func TestPullRequestReviewThreadMergeTombstoneRevisionAndRestore(t *testing.T) { if revisions != 4 { t.Fatalf("review thread revisions after restore = %d, want 4", revisions) } + applied, err := st.TombstonePullRequestReviewThread(ctx, threadID, "rt-1", "2026-05-15T00:06:00Z", "explicit-source-delete") + if err != nil || !applied { + t.Fatalf("direct review thread tombstone = %t, %v", applied, err) + } + applied, err = st.TombstonePullRequestReviewThread(ctx, threadID, "missing", "2026-05-15T00:06:00Z", "explicit-source-delete") + if err != nil || applied { + t.Fatalf("missing review thread tombstone = %t, %v", applied, err) + } + if _, err := st.TombstonePullRequestReviewThread(ctx, threadID, "rt-1", "", ""); err == nil || !strings.Contains(err.Error(), "deleted_at is required") { + t.Fatalf("empty review thread tombstone error = %v", err) + } }