coverage — ~bigbes/sr-ht-spec — 64cae3af — db/proposal.go

Coverage
77.6% 97/125 statements
Δ
+0.0
Blob
03b9284
Uncovered L92-L93L98-L99L142-L143L158-L159L188-L198L208-L209L215-L216L220-L221L255-L256L264-L265L268-L269L281-L282L285-L286L292L313-L314L317-L318L330-L331L334-L335L337-L340L354-L355L361-L362
1 package db
2
3 import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "time"
9
10 "sourcecraft.dev/bigbes/sr-ht-spec/core"
11 )
12
13 // BranchPrefix is the ref namespace agents may write, under this package's
14 // name. It is core's constant: the prefix the INSERT below builds a branch from
15 // and the prefix gitx enforces on a push are one value, because a row and a ref
16 // that disagree about a proposal's branch name is a break that surfaces as a
17 // proposal nobody can find.
18 const BranchPrefix = core.ProposalPrefix
19
20 // ProposalBranch is the branch name for a proposal id: "proposals/42". The row
21 // stores it verbatim (proposal.branch) because the row, not this function, is
22 // what the reconciler compares against the refs it finds.
23 //
24 // The derivation is core.ProposalBranch, so this and gitx.ProposalBranch cannot
25 // disagree — and it inherits core's refusal of a non-positive id, which here
26 // means an unwritten row rather than a proposal.
27 8 func ProposalBranch(id int) (string, error) { return core.ProposalBranch(int64(id)) }
28
29 // Proposal is a bundle of document edits awaiting review: a branch under
30 // BranchPrefix plus this row.
31 //
32 // BaseRev is the If-Match value the agent sent when the proposal was opened —
33 // the space's approved-head sha at the time it read the document — and it does
34 // not move as the proposal accumulates edits. Agent and AgentSession are
35 // mandatory provenance: one shared token still yields a full audit trail,
36 // because the identity strings, not the credential, are what say who did what.
37 //
38 // Approval and MergedRev are empty until the proposal merges; Resolved is nil
39 // until it leaves the open state.
40 type Proposal struct {
41 ID int
42 SpaceID int
43 Title string
44 Rationale string
45 BaseRev string
46 Branch string
47 State core.ProposalState
48 Approval core.Approval
49 MergedRev string
50 Agent string
51 AgentSession string
52 Created time.Time
53 Resolved *time.Time
54 }
55
56 // Merge is everything one merge writes to Postgres: the proposal's transition
57 // to merged, and the new location of every document the merge touched. The two
58 // are one invariant — a merged proposal whose documents are still registered at
59 // their old paths would break link resolution and the next staleness check —
60 // so MergeProposal writes them in a single transaction.
61 type Merge struct {
62 ProposalID int
63 SpaceID int
64 Approval core.Approval
65 MergedRev string
66 // Docs is the (id, path) set of the documents the merge landed, at their
67 // paths in the merge commit. Empty is legal but unusual: it means the
68 // proposal touched nothing the registry tracks.
69 Docs []DocRef
70 }
71
72 const proposalSelect = `
73 SELECT id, space_id, title, COALESCE(rationale, ''), base_rev, branch, state,
74 COALESCE(approval, ''), COALESCE(merged_rev, ''), agent, agent_session,
75 created, resolved
76 FROM proposal`
77
78 16 func scanProposal(sc rowScanner) (*Proposal, error) {
79 16 var (
80 16 p Proposal
81 16 state string
82 16 approval string
83 16 resolved sql.NullTime
84 16 )
85 16 if err := sc.Scan(&p.ID, &p.SpaceID, &p.Title, &p.Rationale, &p.BaseRev,
86 16 &p.Branch, &state, &approval, &p.MergedRev, &p.Agent, &p.AgentSession,
87 16 &p.Created, &resolved); err != nil {
88 2 return nil, err
89 2 }
90 14 parsedState, err := core.ParseProposalState(state)
91 14 if err != nil {
92 0 return nil, fmt.Errorf("proposal %d: %w", p.ID, err)
93 0 }
94 14 p.State = parsedState
95 14 if approval != "" {
96 4 parsedApproval, err := core.ParseApproval(approval)
97 4 if err != nil {
98 0 return nil, fmt.Errorf("proposal %d: %w", p.ID, err)
99 0 }
100 4 p.Approval = parsedApproval
101 }
102 14 if resolved.Valid {
103 6 t := resolved.Time
104 6 p.Resolved = &t
105 6 }
106 14 return &p, nil
107 }
108
109 // OpenProposal inserts a new proposal in the open state and returns it with its
110 // id, branch and creation time filled in. p.SpaceID, p.Title, p.BaseRev,
111 // p.Agent and p.AgentSession must be set; State, Approval, MergedRev and
112 // Resolved are ignored on input — a proposal is always born open.
113 //
114 // The branch name derives from the generated id ("proposals/42"), so id and
115 // branch are allocated in one statement: taking the id in a first round trip
116 // and writing the branch in a second would leave a window where a crash yields
117 // a row whose branch names nothing.
118 22 func (s *Store) OpenProposal(ctx context.Context, p *Proposal) (*Proposal, error) {
119 22 if p.Agent == "" || p.AgentSession == "" {
120 1 return nil, fmt.Errorf("open proposal: agent identity and session are required provenance")
121 1 }
122 21 if p.BaseRev == "" {
123 1 return nil, fmt.Errorf("open proposal: base rev is required")
124 1 }
125 20 const q = `
126 20 WITH next AS (SELECT nextval(pg_get_serial_sequence('proposal', 'id')) AS id)
127 20 INSERT INTO proposal (id, space_id, title, rationale, base_rev, branch, state,
128 20 agent, agent_session, created)
129 20 SELECT next.id, $1, $2, $3, $4, $5::text || next.id::text, $6, $7, $8, $9
130 20 FROM next
131 20 RETURNING id, branch, created`
132 20 out := *p
133 20 out.State = core.StateOpen
134 20 out.Approval = ""
135 20 out.MergedRev = ""
136 20 out.Resolved = nil
137 20 err := s.q.QueryRowContext(ctx, q,
138 20 p.SpaceID, p.Title, nullable(p.Rationale), p.BaseRev, BranchPrefix,
139 20 string(core.StateOpen), p.Agent, p.AgentSession, time.Now().UTC(),
140 20 ).Scan(&out.ID, &out.Branch, &out.Created)
141 20 if err != nil {
142 0 return nil, fmt.Errorf("open proposal: %w", err)
143 0 }
144 20 return &out, nil
145 }
146
147 // GetProposal resolves a proposal by id. Returns ErrNotFound if it does not
148 // exist. Proposal URLs are stable and shareable — a link still resolves after
149 // merge or rejection, showing the outcome — so this is the same lookup whatever
150 // the state.
151 10 func (s *Store) GetProposal(ctx context.Context, id int) (*Proposal, error) {
152 10 q := proposalSelect + ` WHERE id = $1`
153 10 p, err := scanProposal(s.q.QueryRowContext(ctx, q, id))
154 10 if errors.Is(err, sql.ErrNoRows) {
155 2 return nil, ErrNotFound
156 2 }
157 8 if err != nil {
158 0 return nil, fmt.Errorf("get proposal %d: %w", id, err)
159 0 }
160 8 return p, nil
161 }
162
163 // ListProposalsByState lists proposals in one state, newest first — the inbox
164 // query ("N proposals waiting on you"), served by ix_proposal_state_created.
165 // limit <= 0 means no limit.
166 5 func (s *Store) ListProposalsByState(ctx context.Context, state core.ProposalState, limit int) ([]*Proposal, error) {
167 5 if _, err := core.ParseProposalState(string(state)); err != nil {
168 1 return nil, err
169 1 }
170 4 q := proposalSelect + ` WHERE state = $1 ORDER BY created DESC, id DESC`
171 4 args := []any{string(state)}
172 4 if limit > 0 {
173 1 q += ` LIMIT $2`
174 1 args = append(args, limit)
175 1 }
176 4 return s.queryProposals(ctx, q, args...)
177 }
178
179 // ListProposalsBySpace lists one space's proposals in one state, newest first —
180 // the per-space proposal listing the read surfaces serve ("the open proposals
181 // on ~bigbes/rfcs"). ListProposalsByState is the instance-wide inbox; this is
182 // its space-scoped counterpart, which is the shape every surface above service/
183 // asks for, because a proposal is only ever meaningful inside its space.
184 //
185 // The (state, created DESC) index still serves this: the extra space_id
186 // predicate is a filter over a set that is already tiny at this instance's
187 // volume. limit <= 0 means no limit.
188 0 func (s *Store) ListProposalsBySpace(ctx context.Context, spaceID int, state core.ProposalState, limit int) ([]*Proposal, error) {
189 0 if _, err := core.ParseProposalState(string(state)); err != nil {
190 0 return nil, err
191 0 }
192 0 q := proposalSelect + ` WHERE space_id = $1 AND state = $2 ORDER BY created DESC, id DESC`
193 0 args := []any{spaceID, string(state)}
194 0 if limit > 0 {
195 0 q += ` LIMIT $3`
196 0 args = append(args, limit)
197 0 }
198 0 return s.queryProposals(ctx, q, args...)
199 }
200
201 // queryProposals runs a proposalSelect query and scans every row. The two
202 // listings above differ only in their WHERE and args, so the row loop — the
203 // part that is easy to get subtly wrong (a missing rows.Err, a leaked cursor) —
204 // lives in one place.
205 4 func (s *Store) queryProposals(ctx context.Context, q string, args ...any) ([]*Proposal, error) {
206 4 rows, err := s.q.QueryContext(ctx, q, args...)
207 4 if err != nil {
208 0 return nil, fmt.Errorf("list proposals: %w", err)
209 0 }
210 4 defer rows.Close()
211 4 var out []*Proposal
212 6 for rows.Next() {
213 6 p, err := scanProposal(rows)
214 6 if err != nil {
215 0 return nil, fmt.Errorf("scan proposal: %w", err)
216 0 }
217 6 out = append(out, p)
218 }
219 4 if err := rows.Err(); err != nil {
220 0 return nil, fmt.Errorf("iterate proposals: %w", err)
221 0 }
222 4 return out, nil
223 }
224
225 // MarkProposalMerged transitions a proposal to merged, recording how it was
226 // authorized (human or policy) and the merge commit. It writes only the row;
227 // use MergeProposal to update the document registry in the same transaction.
228 //
229 // The legality of the transition is enforced in SQL by `WHERE state = 'open'`,
230 // not by reading the row first: a check-then-write would let two concurrent
231 // resolutions both pass the check. When the guard bites, the current state is
232 // read back only to name it in the error.
233 9 func (s *Store) MarkProposalMerged(ctx context.Context, id int, approval core.Approval, mergedRev string) error {
234 9 if _, err := core.ParseApproval(string(approval)); err != nil {
235 1 return err
236 1 }
237 8 if mergedRev == "" {
238 1 return fmt.Errorf("merge proposal %d: merged rev is required", id)
239 1 }
240 7 return s.resolveProposal(ctx, id, core.StateMerged, string(approval), mergedRev)
241 }
242
243 // RejectProposal transitions a proposal to rejected. There is no
244 // request-changes cycle: with one reviewer, a proposal you dislike is rejected
245 // and the agent proposes again.
246 6 func (s *Store) RejectProposal(ctx context.Context, id int) error {
247 6 return s.resolveProposal(ctx, id, core.StateRejected, "", "")
248 6 }
249
250 // resolveProposal is the shared open->terminal update. next must be a legal
251 // destination from open; approval and mergedRev are stored as SQL NULL when
252 // empty, which the ck_proposal_merged constraints require for a rejection.
253 13 func (s *Store) resolveProposal(ctx context.Context, id int, next core.ProposalState, approval, mergedRev string) error {
254 13 if err := core.StateOpen.CanTransitionTo(next); err != nil {
255 0 return err
256 0 }
257 13 const q = `
258 13 UPDATE proposal
259 13 SET state = $2, approval = $3, merged_rev = $4, resolved = $5
260 13 WHERE id = $1 AND state = $6`
261 13 res, err := s.q.ExecContext(ctx, q, id, string(next), nullable(approval),
262 13 nullable(mergedRev), time.Now().UTC(), string(core.StateOpen))
263 13 if err != nil {
264 0 return fmt.Errorf("resolve proposal %d as %s: %w", id, next, err)
265 0 }
266 13 n, err := res.RowsAffected()
267 13 if err != nil {
268 0 return fmt.Errorf("resolve proposal %d: rows affected: %w", id, err)
269 0 }
270 13 if n == 1 {
271 7 return nil
272 7 }
273
274 // Nothing moved: either the proposal is gone, or it is no longer open.
275 6 var current string
276 6 err = s.q.QueryRowContext(ctx, `SELECT state FROM proposal WHERE id = $1`, id).Scan(&current)
277 6 if errors.Is(err, sql.ErrNoRows) {
278 2 return ErrNotFound
279 2 }
280 4 if err != nil {
281 0 return fmt.Errorf("resolve proposal %d: read current state: %w", id, err)
282 0 }
283 4 from, err := core.ParseProposalState(current)
284 4 if err != nil {
285 0 return fmt.Errorf("proposal %d: %w", id, err)
286 0 }
287 4 if err := from.CanTransitionTo(next); err != nil {
288 4 return fmt.Errorf("proposal %d: %w", id, err)
289 4 }
290 // The row is open and the transition is legal, yet the guarded UPDATE
291 // matched nothing. That cannot happen; refuse rather than report success.
292 0 return fmt.Errorf("proposal %d: guarded update matched no row while state is %s", id, from)
293 }
294
295 // DeleteOpenProposal removes an open proposal row outright.
296 //
297 // It is the only delete in this package and it is deliberately narrow: the
298 // reconciler's repair for a row whose branch never appeared, where the daemon
299 // died between the row insert and the branch write. Such a row holds no content
300 // — the agent still has the document it wanted to write and re-proposes — so
301 // deleting it loses nothing. A resolved proposal is history and is never
302 // deleted, which is why this is not a general-purpose delete.
303 //
304 // The `state = 'open'` guard is in the statement, not in Go: a check-then-write
305 // would let a merge land in between and delete the row of a proposal that had
306 // just succeeded. When the guard bites, the current state is read back only to
307 // name it in the error — ErrNotFound when the row is gone, ErrProposalNotOpen
308 // when it has been resolved — exactly as resolveProposal does.
309 5 func (s *Store) DeleteOpenProposal(ctx context.Context, id int) error {
310 5 const q = `DELETE FROM proposal WHERE id = $1 AND state = $2`
311 5 res, err := s.q.ExecContext(ctx, q, id, string(core.StateOpen))
312 5 if err != nil {
313 0 return fmt.Errorf("delete proposal %d: %w", id, err)
314 0 }
315 5 n, err := res.RowsAffected()
316 5 if err != nil {
317 0 return fmt.Errorf("delete proposal %d: rows affected: %w", id, err)
318 0 }
319 5 if n == 1 {
320 1 return nil
321 1 }
322
323 // Nothing was deleted: either the proposal is gone, or it is no longer open.
324 4 var current string
325 4 err = s.q.QueryRowContext(ctx, `SELECT state FROM proposal WHERE id = $1`, id).Scan(&current)
326 4 if errors.Is(err, sql.ErrNoRows) {
327 2 return ErrNotFound
328 2 }
329 2 if err != nil {
330 0 return fmt.Errorf("delete proposal %d: read current state: %w", id, err)
331 0 }
332 2 from, err := core.ParseProposalState(current)
333 2 if err != nil {
334 0 return fmt.Errorf("proposal %d: %w", id, err)
335 0 }
336 2 if from == core.StateOpen {
337 0 // The row is open, yet the guarded DELETE matched nothing. That cannot
338 0 // happen; refuse rather than report success.
339 0 return fmt.Errorf("proposal %d: guarded delete matched no row while state is %s", id, from)
340 0 }
341 2 return fmt.Errorf("%w: proposal %d is %s", ErrProposalNotOpen, id, from)
342 }
343
344 // MergeProposal records a merge: the proposal's transition to merged and the
345 // new registry location of every document it landed, atomically.
346 //
347 // Git refs remain the source of truth for whether the merge happened — this row
348 // is what the reconciler repairs from the ref when a crash lands between the
349 // two. What the transaction buys is that Postgres never holds the half-state
350 // where the proposal reads merged but its documents are still registered at
351 // their pre-merge paths.
352 3 func (s *Store) MergeProposal(ctx context.Context, m Merge) error {
353 3 if _, err := core.ParseApproval(string(m.Approval)); err != nil {
354 0 return err
355 0 }
356 3 if m.MergedRev == "" {
357 1 return fmt.Errorf("merge proposal %d: merged rev is required", m.ProposalID)
358 1 }
359 2 return s.InTx(ctx, func(tx *Store) error {
360 2 if err := tx.MarkProposalMerged(ctx, m.ProposalID, m.Approval, m.MergedRev); err != nil {
361 0 return err
362 0 }
363 2 return tx.UpsertDocIDs(ctx, m.SpaceID, m.Docs, m.MergedRev)
364 })
365 }