diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index 5c58f0a27..f61d38b45 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -176,27 +176,31 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o return m.runReactions(ctx, id, prContent(o)) } -// writePR upserts the scalar facts, records each check run, and replaces the -// comment set. PR-table CDC is emitted by the DB triggers. +// writePR persists the observation's scalar facts, check runs, and comment set +// in one atomic store call. PR-table CDC is emitted by the DB triggers. func (m *Manager) writePR(ctx context.Context, id domain.SessionID, o ports.PRObservation) error { now := m.clock() - if err := m.pr.UpsertPR(ctx, ports.PRRow{ + row := ports.PRRow{ URL: o.URL, SessionID: string(id), Number: o.Number, Draft: o.Draft, Merged: o.Merged, Closed: o.Closed, CI: o.CI, Review: o.Review, Mergeability: o.Mergeability, UpdatedAt: now, - }); err != nil { - return err } - for _, c := range o.Checks { + checks := make([]ports.PRCheckRow, len(o.Checks)) + for i, c := range o.Checks { c.PRURL = o.URL if c.CreatedAt.IsZero() { c.CreatedAt = now } - if err := m.pr.RecordCheck(ctx, c); err != nil { - return err - } + checks[i] = c } - return m.pr.ReplacePRComments(ctx, o.URL, o.Comments) + comments := make([]ports.PRComment, len(o.Comments)) + for i, c := range o.Comments { + if c.CreatedAt.IsZero() { + c.CreatedAt = now + } + comments[i] = c + } + return m.pr.WritePR(ctx, row, checks, comments) } // ---- mutation commands from the Session Manager ---- diff --git a/backend/internal/lifecycle/manager_test.go b/backend/internal/lifecycle/manager_test.go index 7843f8af2..4ae9aaafd 100644 --- a/backend/internal/lifecycle/manager_test.go +++ b/backend/internal/lifecycle/manager_test.go @@ -82,12 +82,10 @@ func (f *fakeStore) PRFactsForSession(_ context.Context, id domain.SessionID) (d } return facts, nil } -func (f *fakeStore) UpsertPR(_ context.Context, r ports.PRRow) error { - f.pr[domain.SessionID(r.SessionID)] = r - return nil -} -func (f *fakeStore) RecordCheck(_ context.Context, r ports.PRCheckRow) error { - f.checks = append(f.checks, r) +func (f *fakeStore) WritePR(_ context.Context, pr ports.PRRow, checks []ports.PRCheckRow, comments []ports.PRComment) error { + f.pr[domain.SessionID(pr.SessionID)] = pr + f.checks = append(f.checks, checks...) + f.comments[pr.URL] = comments return nil } func (f *fakeStore) RecentCheckStatuses(_ context.Context, url, name string, limit int) ([]string, error) { @@ -99,10 +97,6 @@ func (f *fakeStore) RecentCheckStatuses(_ context.Context, url, name string, lim } return out, nil } -func (f *fakeStore) ReplacePRComments(_ context.Context, url string, cs []ports.PRComment) error { - f.comments[url] = cs - return nil -} type fakeNotifier struct{ events []ports.Event } diff --git a/backend/internal/ports/outbound.go b/backend/internal/ports/outbound.go index d180f5385..75a24bf07 100644 --- a/backend/internal/ports/outbound.go +++ b/backend/internal/ports/outbound.go @@ -24,10 +24,12 @@ type SessionStore interface { // PRWriter records the PR facts a PR observation carries. The pr table's own DB // triggers emit the CDC; this just writes the rows. type PRWriter interface { - UpsertPR(ctx context.Context, r PRRow) error - RecordCheck(ctx context.Context, r PRCheckRow) error + // WritePR persists a full PR observation — scalar facts, check runs, and the + // replacement comment set — in one transaction, so the rows and the CDC + // events they emit are all-or-nothing. + WritePR(ctx context.Context, pr PRRow, checks []PRCheckRow, comments []PRComment) error + // RecentCheckStatuses reads the last `limit` runs of a check (the CI brake). RecentCheckStatuses(ctx context.Context, prURL, name string, limit int) ([]string, error) - ReplacePRComments(ctx context.Context, prURL string, comments []PRComment) error } // Notifier delivers an event to the human (desktop/Slack later). Push, never poll. diff --git a/backend/internal/storage/sqlite/db.go b/backend/internal/storage/sqlite/db.go index 63f6b7dd5..8b001d119 100644 --- a/backend/internal/storage/sqlite/db.go +++ b/backend/internal/storage/sqlite/db.go @@ -11,6 +11,9 @@ import ( "path/filepath" "github.com/pressly/goose/v3" + // modernc.org/sqlite is the pure-Go (CGO-free) SQLite driver — chosen so the + // daemon cross-compiles and ships as a static binary with no libsqlite/CGO + // toolchain dependency, at the cost of some raw throughput vs a C-backed driver. _ "modernc.org/sqlite" ) diff --git a/backend/internal/storage/sqlite/migrations/0001_init.sql b/backend/internal/storage/sqlite/migrations/0001_init.sql index 6534816d7..9d5a6a22d 100644 --- a/backend/internal/storage/sqlite/migrations/0001_init.sql +++ b/backend/internal/storage/sqlite/migrations/0001_init.sql @@ -202,6 +202,25 @@ BEGIN END; -- +goose StatementEnd +-- A re-polled check can change status on the same commit (in_progress -> failed) +-- via UpsertPRCheck's ON CONFLICT DO UPDATE. Without this trigger that status +-- transition would update the row silently, so CDC consumers would never see it. +-- Guarded on the status so a no-op re-poll emits nothing. +-- +goose StatementBegin +CREATE TRIGGER pr_checks_cdc_update +AFTER UPDATE ON pr_checks +WHEN OLD.status <> NEW.status +BEGIN + INSERT INTO change_log (project_id, session_id, event_type, payload, created_at) + VALUES ( + (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url), + (SELECT session_id FROM pr WHERE url = NEW.pr_url), + 'pr_check_recorded', + json_object('pr', NEW.pr_url, 'name', NEW.name, 'commit', NEW.commit_hash, 'status', NEW.status), + NEW.created_at); +END; +-- +goose StatementEnd + -- +goose Down -- +goose StatementBegin DROP TABLE change_log; diff --git a/backend/internal/storage/sqlite/pr_cdc_test.go b/backend/internal/storage/sqlite/pr_cdc_test.go new file mode 100644 index 000000000..8c8f7ea2e --- /dev/null +++ b/backend/internal/storage/sqlite/pr_cdc_test.go @@ -0,0 +1,86 @@ +package sqlite + +import ( + "context" + "strings" + "testing" + "time" +) + +// A check can change status on the same commit (in_progress -> failed) via +// UpsertPRCheck's ON CONFLICT DO UPDATE. CDC must emit on that transition, not +// only on the first insert — otherwise live clients never see the status change. +func TestPRChecksCDC_EmitsOnInsertAndStatusUpdate(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + rec, err := s.CreateSession(ctx, sampleRecord("mer")) + if err != nil { + t.Fatal(err) + } + url := "https://example/pr/1" + if err := s.UpsertPR(ctx, PRRow{URL: url, SessionID: string(rec.ID), Number: 1}); err != nil { + t.Fatal(err) + } + + now := time.Now() + mustCheck := func(status string) { + if err := s.RecordCheck(ctx, PRCheckRow{PRURL: url, Name: "build", CommitHash: "c1", Status: status, CreatedAt: now}); err != nil { + t.Fatal(err) + } + } + mustCheck("in_progress") // insert -> event + mustCheck("failed") // status change on same commit (update) -> event + mustCheck("failed") // no-op re-poll (status unchanged) -> NO event + + rows, err := s.ReadChangeLogAfter(ctx, 0, 100) + if err != nil { + t.Fatal(err) + } + var checkEvents []ChangeLogRow + for _, r := range rows { + if r.EventType == "pr_check_recorded" { + checkEvents = append(checkEvents, r) + } + } + if len(checkEvents) != 2 { + t.Fatalf("want 2 check CDC events (insert + status change, no-op suppressed), got %d", len(checkEvents)) + } + if !strings.Contains(checkEvents[1].Payload, `"status":"failed"`) { + t.Fatalf("the update event should carry the new status, got %q", checkEvents[1].Payload) + } +} + +// WritePRObservation persists scalar facts, checks, and comments in one tx; all +// three should be queryable afterward. +func TestWritePRObservation_PersistsScalarsChecksAndComments(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + seedProject(t, s, "mer") + rec, err := s.CreateSession(ctx, sampleRecord("mer")) + if err != nil { + t.Fatal(err) + } + url := "https://example/pr/7" + now := time.Now() + + err = s.WritePRObservation(ctx, + PRRow{URL: url, SessionID: string(rec.ID), Number: 7, CIState: "failing", UpdatedAt: now}, + []PRCheckRow{{PRURL: url, Name: "build", CommitHash: "c1", Status: "failed", CreatedAt: now}}, + []PRCommentRow{{PRURL: url, CommentID: "1", Author: "reviewer", Body: "use a const", CreatedAt: now}}, + ) + if err != nil { + t.Fatal(err) + } + + pr, ok, err := s.GetPR(ctx, url) + if err != nil || !ok || pr.CIState != "failing" { + t.Fatalf("scalar facts not persisted: ok=%v ci=%q err=%v", ok, pr.CIState, err) + } + if checks, _ := s.ListChecks(ctx, url); len(checks) != 1 || checks[0].Status != "failed" { + t.Fatalf("check not persisted: %+v", checks) + } + if comments, _ := s.ListPRComments(ctx, url); len(comments) != 1 || comments[0].Body != "use a const" { + t.Fatalf("comment not persisted: %+v", comments) + } +} diff --git a/backend/internal/storage/sqlite/pr_store.go b/backend/internal/storage/sqlite/pr_store.go index 4170da4d9..8b41396c5 100644 --- a/backend/internal/storage/sqlite/pr_store.go +++ b/backend/internal/storage/sqlite/pr_store.go @@ -27,18 +27,7 @@ type PRRow struct { // fields default to their "nothing known yet" value so a partial row is valid // against the CHECK constraints (matches the domain zero values none/unknown). func (s *Store) UpsertPR(ctx context.Context, r PRRow) error { - if r.State == "" { - r.State = "open" - } - if r.ReviewDecision == "" { - r.ReviewDecision = "none" - } - if r.CIState == "" { - r.CIState = "unknown" - } - if r.Mergeability == "" { - r.Mergeability = "unknown" - } + r = r.withDefaults() s.writeMu.Lock() defer s.writeMu.Unlock() return s.qw.UpsertPR(ctx, gen.UpsertPRParams{ @@ -53,6 +42,67 @@ func (s *Store) UpsertPR(ctx context.Context, r PRRow) error { }) } +// WritePRObservation persists a full PR observation — scalar facts, check runs, +// and the replacement comment set — in one write transaction, so the rows and +// the change_log events their triggers emit are committed all-or-nothing. The +// scalar PR upsert runs first so the checks'/comments' CDC triggers can resolve +// the session id from the pr row within the same transaction. +func (s *Store) WritePRObservation(ctx context.Context, pr PRRow, checks []PRCheckRow, comments []PRCommentRow) error { + pr = pr.withDefaults() + s.writeMu.Lock() + defer s.writeMu.Unlock() + return s.inTx(ctx, "write pr observation", func(q *gen.Queries) error { + if err := q.UpsertPR(ctx, gen.UpsertPRParams{ + Url: pr.URL, SessionID: pr.SessionID, Number: pr.Number, + PrState: pr.State, ReviewDecision: pr.ReviewDecision, + CiState: pr.CIState, Mergeability: pr.Mergeability, UpdatedAt: pr.UpdatedAt, + }); err != nil { + return err + } + for _, c := range checks { + if c.Status == "" { + c.Status = "unknown" + } + if err := q.UpsertPRCheck(ctx, gen.UpsertPRCheckParams{ + PrUrl: c.PRURL, Name: c.Name, CommitHash: c.CommitHash, + Status: c.Status, Url: c.URL, LogTail: c.LogTail, CreatedAt: c.CreatedAt, + }); err != nil { + return err + } + } + if err := q.DeletePRComments(ctx, pr.URL); err != nil { + return err + } + for _, cm := range comments { + if err := q.UpsertPRComment(ctx, gen.UpsertPRCommentParams{ + PrUrl: pr.URL, CommentID: cm.CommentID, Author: cm.Author, File: cm.File, + Line: cm.Line, Body: cm.Body, Resolved: boolToInt(cm.Resolved), CreatedAt: cm.CreatedAt, + }); err != nil { + return fmt.Errorf("comment %q: %w", cm.CommentID, err) + } + } + return nil + }) +} + +// withDefaults fills empty enum fields with their "nothing known yet" value so a +// partial row satisfies the CHECK constraints (matches UpsertPR). +func (r PRRow) withDefaults() PRRow { + if r.State == "" { + r.State = "open" + } + if r.ReviewDecision == "" { + r.ReviewDecision = "none" + } + if r.CIState == "" { + r.CIState = "unknown" + } + if r.Mergeability == "" { + r.Mergeability = "unknown" + } + return r +} + // GetPR returns the PR facts for a URL, or ok=false if absent. func (s *Store) GetPR(ctx context.Context, url string) (PRRow, bool, error) { p, err := s.qr.GetPR(ctx, url) diff --git a/backend/lifecycle_wiring.go b/backend/lifecycle_wiring.go index f69b1ce46..d736d6536 100644 --- a/backend/lifecycle_wiring.go +++ b/backend/lifecycle_wiring.go @@ -87,33 +87,30 @@ func (a storeAdapter) PRFactsForSession(ctx context.Context, id domain.SessionID return facts, nil } -func (a storeAdapter) UpsertPR(ctx context.Context, r ports.PRRow) error { - return a.Store.UpsertPR(ctx, sqlite.PRRow{ - URL: r.URL, SessionID: r.SessionID, Number: int64(r.Number), - State: prState(r), - ReviewDecision: string(r.Review), - CIState: string(r.CI), - Mergeability: string(r.Mergeability), - UpdatedAt: r.UpdatedAt, - }) -} - -func (a storeAdapter) RecordCheck(ctx context.Context, r ports.PRCheckRow) error { - return a.Store.RecordCheck(ctx, sqlite.PRCheckRow{ - PRURL: r.PRURL, Name: r.Name, CommitHash: r.CommitHash, - Status: r.Status, URL: r.URL, LogTail: r.LogTail, CreatedAt: r.CreatedAt, - }) -} - -func (a storeAdapter) ReplacePRComments(ctx context.Context, prURL string, comments []ports.PRComment) error { - rows := make([]sqlite.PRCommentRow, len(comments)) +func (a storeAdapter) WritePR(ctx context.Context, pr ports.PRRow, checks []ports.PRCheckRow, comments []ports.PRComment) error { + row := sqlite.PRRow{ + URL: pr.URL, SessionID: pr.SessionID, Number: int64(pr.Number), + State: prState(pr), + ReviewDecision: string(pr.Review), + CIState: string(pr.CI), + Mergeability: string(pr.Mergeability), + UpdatedAt: pr.UpdatedAt, + } + checkRows := make([]sqlite.PRCheckRow, len(checks)) + for i, c := range checks { + checkRows[i] = sqlite.PRCheckRow{ + PRURL: c.PRURL, Name: c.Name, CommitHash: c.CommitHash, + Status: c.Status, URL: c.URL, LogTail: c.LogTail, CreatedAt: c.CreatedAt, + } + } + commentRows := make([]sqlite.PRCommentRow, len(comments)) for i, c := range comments { - rows[i] = sqlite.PRCommentRow{ - PRURL: prURL, CommentID: c.ID, Author: c.Author, File: c.File, + commentRows[i] = sqlite.PRCommentRow{ + PRURL: pr.URL, CommentID: c.ID, Author: c.Author, File: c.File, Line: int64(c.Line), Body: c.Body, Resolved: c.Resolved, CreatedAt: c.CreatedAt, } } - return a.Store.ReplacePRComments(ctx, prURL, rows) + return a.Store.WritePRObservation(ctx, row, checkRows, commentRows) } // prState collapses the PR's bools into the single pr.state column value.