dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

fix(office): keep budget alerts honest when a policy changes mid-check (#3532)

Browse Source 
Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-14 00:45:53 +08:00
committed by GitHub
parent d62daa9d70
commit 4e4b29b29e
27 changed files with 2363 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/office/costs/budget_idempotency_test.go
+153 -50
View File
@@ -8,6 +8,10 @@ import (
"testing"
"time"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"go.uber.org/zap/zaptest/observer"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/office/costs"
"github.com/kandev/kandev/internal/office/models"
@@ -54,13 +58,14 @@ func newIdempotencyTestPolicy(scopeID string, limit int64) *models.BudgetPolicy
}
}
// monthlyPeriodKey mirrors the RFC3339-UTC month-start rendering the
// monthlyPeriodKeyAt mirrors the RFC3339-UTC month-start rendering the
// service computes internally, for tests that assert on stored claim rows.
func monthlyPeriodKey() string {
now := time.Now().UTC()
func monthlyPeriodKeyAt(now time.Time) string {
return time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC).Format(time.RFC3339)
}
func monthlyPeriodKey() string { return monthlyPeriodKeyAt(time.Now().UTC()) }
func TestCheckBudget_EvaluateTwice_AlertEmitsOnce(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
@@ -144,6 +149,10 @@ func TestCheckBudget_ExceededClaimsCompanionAlert(t *testing.T) {
// own by a rollup query, only inferred via the companion claim.
insertBudgetTestCostEvent(t, execSQL, "agent-companion", "task-1", 600)
// Captured before CheckBudget so this assertion can't compute a
// different month boundary than the evaluation did if the two calls
// straddle a UTC month rollover.
periodKey := monthlyPeriodKey()
results, err := svc.CheckBudget(ctx, "ws-1", "agent-companion", "proj-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
@@ -158,7 +167,7 @@ func TestCheckBudget_ExceededClaimsCompanionAlert(t *testing.T) {
var count int
if err := queryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ? AND period_key = ? AND level = 'alert'`,
policy.ID, monthlyPeriodKey(),
policy.ID, periodKey,
).Scan(&count); err != nil {
t.Fatalf("count companion claim: %v", err)
}
@@ -167,13 +176,58 @@ func TestCheckBudget_ExceededClaimsCompanionAlert(t *testing.T) {
}
}
// TestCheckBudget_ExceededClaimHeld_AlertBandEvaluationSuppressed covers
// AC-OFFICE-COSTS-003.12 directly: while an exceeded-level claim is held
// for a policy, period and revision, an evaluation whose spend reaches only
// the alert level must not emit budget.alert for that period and revision.
// The exceeded claim's companion already holds the alert-level slot, so
// this follows from the ordinary primary-key conflict on a second
// alert-level insert, with no separate read of the exceeded claim.
func TestCheckBudget_ExceededClaimHeld_AlertBandEvaluationSuppressed(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-exceeded-then-alert")
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
svc := costs.NewCostService(repo, logger.Default(), spy, agents, agents)
policy := newIdempotencyTestPolicy("agent-exceeded-then-alert", 1000)
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
periodKey := monthlyPeriodKey()
if claimed, err := repo.ClaimExceeded(ctx, policy.ID, periodKey, policy.Revision); err != nil || !claimed {
t.Fatalf("seed exceeded+companion claim: claimed=%v err=%v", claimed, err)
}
// Spend in the alert band only: >= 80% of the 1000 limit, below it.
insertBudgetTestCostEvent(t, execSQL, "agent-exceeded-then-alert", "task-1", 900)
results, err := svc.CheckBudget(ctx, "ws-1", "agent-exceeded-then-alert", "proj-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if !results[0].AlertFired {
t.Fatal("expected spend to reach the alert level")
}
if results[0].AlertSubmitted {
t.Fatal("budget.alert must not be submitted while an exceeded-level claim is held for this policy, period and revision")
}
if got := spy.count("budget.alert"); got != 0 {
t.Fatalf("budget.alert emissions = %d, want 0", got)
}
}
// TestCheckBudget_CompanionAlertClaimFault_StillSubmitsExceeded covers the
// AC-OFFICE-COSTS-002.5 fault-injection carve-out: when the companion
// alert-level claim errors, the evaluation still submits budget.exceeded
// (already earned by the exceeded-level claim) and the failure is logged
// and counted like any other claim-store error. The test deliberately does
// NOT assert the companion claim was recorded, per the spec's own
// instruction — that is the failure this test injects.
// AC-OFFICE-COSTS-003.10 fault-injection case: when the companion
// alert-level insert errors, the whole atomic pair rolls back — no exceeded
// claim row survives either — but the evaluation still submits
// budget.exceeded (already earned by the exceeded-level insert winning
// before the rollback) and the pair failure is logged and counted exactly
// once, not once per row. The test deliberately does NOT assert the
// companion claim was recorded, per the spec's own instruction — that is
// the failure this test injects.
func TestCheckBudget_CompanionAlertClaimFault_StillSubmitsExceeded(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
@@ -191,6 +245,7 @@ func TestCheckBudget_CompanionAlertClaimFault_StillSubmitsExceeded(t *testing.T)
}
insertBudgetTestCostEvent(t, execSQL, "agent-companion-fault", "task-1", 600)
before := claimFailureCount(t)
results, err := svc.CheckBudget(ctx, "ws-1", "agent-companion-fault", "proj-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
@@ -201,6 +256,9 @@ func TestCheckBudget_CompanionAlertClaimFault_StillSubmitsExceeded(t *testing.T)
if got := spy.count("budget.exceeded"); got != 1 {
t.Fatalf("budget.exceeded submissions = %d, want 1", got)
}
if delta := claimFailureCount(t) - before; delta != 1 {
t.Fatalf("budget_claim_failures_total delta = %d, want 1 (the pair is one claim attempt)", delta)
}
}
// TestCheckBudget_ClaimStoreFault_EmitsAndReportsSubmittedTrue covers
@@ -246,12 +304,14 @@ func TestCheckBudget_ClaimStoreFault_EmitsAndReportsSubmittedTrue(t *testing.T)
}
}
// TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsTwicePerEval
// TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsOncePerEval
// extends TestCheckBudget_ClaimStoreFault_EmitsAndReportsSubmittedTrue to the
// exceeded level: claimAndEmit's afterClaim hook claims the alert-level
// companion right after the exceeded-level claim, so a broken store fails
// both claim attempts once per evaluation (AC-OFFICE-COSTS-002.14).
func TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsTwicePerEval(t *testing.T) {
// exceeded level: AC-OFFICE-COSTS-003.10 makes the exceeded claim and its
// alert-level companion one atomic attempt, so a broken store fails the
// whole pair exactly once per evaluation — one log line, one counter
// increment (AC-OFFICE-COSTS-003.14 amends AC-OFFICE-COSTS-002.14 to say
// so), not once per row as it would if the two claims were independent.
func TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsOncePerEval(t *testing.T) {
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
createBudgetTestAgent(t, repo, "ws-1", "agent-fault-exc")
@@ -283,9 +343,71 @@ func TestCheckBudget_ClaimStoreFault_ExceededPath_EmitsAndCountsTwicePerEval(t *
if got := spy.count("budget.exceeded"); got != 2 {
t.Fatalf("budget.exceeded = %d, want 2 (duplicates permitted under broken store)", got)
}
// exceeded claim + companion alert claim both fail => 2 increments per evaluation.
if delta := claimFailureCount(t) - before; delta != 4 {
t.Fatalf("budget_claim_failures_total delta = %d, want 4 (2 per exceeded eval)", delta)
if delta := claimFailureCount(t) - before; delta != 2 {
t.Fatalf("budget_claim_failures_total delta = %d, want 2 (1 per exceeded eval, the pair is one claim attempt)", delta)
}
}
// TestCheckBudget_ClaimStoreFault_LogsLevelFieldMatchingEmission covers
// AC-OFFICE-COSTS-003.14: the claim-store failure log's level field names
// the level whose emission the failed claim governs — "alert" for a
// standalone alert-band claim, "exceeded" for the atomic pair — never a
// sentinel shared by both.
func TestCheckBudget_ClaimStoreFault_LogsLevelFieldMatchingEmission(t *testing.T) {
core, logs := observer.New(zapcore.ErrorLevel)
log, err := logger.NewFromZap(zap.New(core))
if err != nil {
t.Fatalf("create observer logger: %v", err)
}
injected := errors.New("injected claim store failure")
repo, _, execSQL := newBudgetTestRepo(t)
ctx := context.Background()
agents := &repoAgents{repo: repo}
spy := &budgetActivitySpy{}
createBudgetTestAgent(t, repo, "ws-1", "agent-log-alert")
alertFaulty := &claimFaultRepo{Repository: repo, failLevels: map[string]error{"alert": injected}}
alertSvc := costs.NewCostService(alertFaulty, log, spy, agents, agents)
alertPolicy := newIdempotencyTestPolicy("agent-log-alert", 1000)
if err := alertSvc.CreateBudgetPolicy(ctx, alertPolicy); err != nil {
t.Fatalf("create alert policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-log-alert", "task-1", 850)
if _, err := alertSvc.CheckBudget(ctx, "ws-1", "agent-log-alert", "proj-1"); err != nil {
t.Fatalf("CheckBudget (alert): %v", err)
}
createBudgetTestAgent(t, repo, "ws-1", "agent-log-exceeded")
exceededFaulty := &claimFaultRepo{Repository: repo, failLevels: map[string]error{"exceeded": injected}}
exceededSvc := costs.NewCostService(exceededFaulty, log, spy, agents, agents)
exceededPolicy := newIdempotencyTestPolicy("agent-log-exceeded", 500)
if err := exceededSvc.CreateBudgetPolicy(ctx, exceededPolicy); err != nil {
t.Fatalf("create exceeded policy: %v", err)
}
insertBudgetTestCostEvent(t, execSQL, "agent-log-exceeded", "task-1", 600)
if _, err := exceededSvc.CheckBudget(ctx, "ws-1", "agent-log-exceeded", "proj-1"); err != nil {
t.Fatalf("CheckBudget (exceeded): %v", err)
}
entries := logs.FilterMessage("budget claim store failure").All()
if len(entries) != 2 {
t.Fatalf("claim-store-failure log entries = %d, want 2", len(entries))
}
var sawAlert, sawExceeded bool
for _, e := range entries {
switch e.ContextMap()["level"] {
case "alert":
sawAlert = true
case "exceeded":
sawExceeded = true
}
}
if !sawAlert {
t.Fatal(`expected a log entry with level="alert" for the standalone alert-band claim failure`)
}
if !sawExceeded {
t.Fatal(`expected a log entry with level="exceeded" for the atomic pair's claim failure`)
}
}
@@ -446,34 +568,6 @@ func TestCheckBudget_ConcurrentEvaluation_EmitsOnce(t *testing.T) {
}
}
// TestCheckBudgetAndEvaluateProjectBudget_ShareAlertClaim covers the two
// callers that can evaluate a project policy. They must use the same durable
// claim so a cost event followed by task reassignment emits one alert.
func TestCheckBudgetAndEvaluateProjectBudget_ShareAlertClaim(t *testing.T) {
spy := &budgetActivitySpy{}
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
ctx := context.Background()
policy := newIdempotencyTestPolicy("proj-shared-claim", 1000)
policy.ScopeType = "project"
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestTask(t, execSQL, "task-shared-claim", "ws-1", "proj-shared-claim")
insertBudgetTestCostEvent(t, execSQL, "agent-shared-claim", "task-shared-claim", 850)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-shared-claim", "proj-shared-claim"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-shared-claim"); err != nil {
t.Fatalf("EvaluateProjectBudget: %v", err)
}
if got := spy.count("budget.alert"); got != 1 {
t.Fatalf("budget.alert submissions across callers = %d, want 1", got)
}
}
// TestCheckBudget_PeriodKeyMatchesSpendBoundary covers AC-OFFICE-COSTS-002.6a:
// the claim recorded for a monthly policy is keyed to the same window
// boundary the spend rollup used, rendered as RFC3339 UTC.
@@ -490,9 +584,14 @@ func TestCheckBudget_PeriodKeyMatchesSpendBoundary(t *testing.T) {
}
insertBudgetTestCostEvent(t, execSQL, "agent-period", "task-1", 850)
// The evaluation reads the clock once. Accept the month containing either
// endpoint because a test can straddle a UTC month rollover while the
// evaluation runs.
before := time.Now().UTC()
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-period", "proj-1"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
after := time.Now().UTC()
var gotKey string
if err := queryRow(
@@ -500,8 +599,12 @@ func TestCheckBudget_PeriodKeyMatchesSpendBoundary(t *testing.T) {
).Scan(&gotKey); err != nil {
t.Fatalf("query claim period_key: %v", err)
}
if gotKey != monthlyPeriodKey() {
t.Fatalf("claim period_key = %q, want %q (must match the spend rollup's boundary)", gotKey, monthlyPeriodKey())
wantKeys := map[string]struct{}{
monthlyPeriodKeyAt(before): {},
monthlyPeriodKeyAt(after): {},
}
if _, ok := wantKeys[gotKey]; !ok {
t.Fatalf("claim period_key = %q, want one of %v (must match the spend rollup's boundary)", gotKey, wantKeys)
}
}
@@ -554,7 +657,7 @@ func TestCheckBudget_SpendBelowLevelKeepsExistingClaim(t *testing.T) {
}
periodKey := monthlyPeriodKey()
if claimed, err := repo.Claim(ctx, policy.ID, periodKey, "alert"); err != nil || !claimed {
if claimed, err := repo.Claim(ctx, policy.ID, periodKey, "alert", policy.Revision); err != nil || !claimed {
t.Fatalf("seed claim: claimed=%v err=%v", claimed, err)
}
@@ -564,7 +667,7 @@ func TestCheckBudget_SpendBelowLevelKeepsExistingClaim(t *testing.T) {
t.Fatalf("CheckBudget: %v", err)
}
claimed, err := repo.Claim(ctx, policy.ID, periodKey, "alert")
claimed, err := repo.Claim(ctx, policy.ID, periodKey, "alert", policy.Revision)
if err != nil {
t.Fatalf("re-claim: %v", err)
}
apps/backend/internal/office/costs/budgets.go
+28 -30
View File
@@ -138,43 +138,27 @@ func (s *CostService) evaluatePolicy(
if policy.ActionOnExceed == budgetActionPauseAgent && policy.ScopeType == scopeAgent {
result.AgentPaused = s.pauseAgentForBudget(ctx, policy.ScopeID)
}
result.ExceededSubmitted = s.claimAndEmit(ctx, workspaceID, policy, spent, periodKey, claimLevelExceeded,
// A spend that jumps straight from below the alert level to
// above the limit must also hold the alert-level claim before
// budget.exceeded is emitted, so a concurrent evaluation
// landing in the alert band cannot win that claim in the gap
// and emit a budget.alert describing a reduction in spend. The
// outcome is discarded (claimed or already-claimed are equally
// fine here); a claim-store error is still logged and counted
// like any other.
func() { s.claimCompanionAlert(ctx, policy, periodKey) })
result.ExceededSubmitted = s.claimExceededAndEmit(ctx, workspaceID, policy, spent, periodKey)
case spent >= threshold:
result.AlertFired = true
result.AlertSubmitted = s.claimAndEmit(ctx, workspaceID, policy, spent, periodKey, claimLevelAlert, nil)
result.AlertSubmitted = s.claimAndEmit(ctx, workspaceID, policy, spent, periodKey, claimLevelAlert)
}
return result, nil
}
// claimAndEmit resolves the three outcomes of claiming (policy, periodKey,
// level) and emits that level's activity row when the claim allows it.
// Reports whether this evaluation submitted the row. afterClaim, when
// non-nil, runs immediately after the claim and before the emission
// decision, regardless of the claim's outcome — the seam the exceeded
// level's alert-level companion claim uses so both claims land before
// either emission is decided.
// level) at the policy's current revision, and emits that level's activity
// row when the claim allows it. Reports whether this evaluation submitted
// the row.
func (s *CostService) claimAndEmit(
ctx context.Context,
workspaceID string,
policy *BudgetPolicy,
spent int64,
periodKey, level string,
afterClaim func(),
) bool {
claimed, err := s.repo.Claim(ctx, policy.ID, periodKey, level)
if afterClaim != nil {
afterClaim()
}
claimed, err := s.repo.Claim(ctx, policy.ID, periodKey, level, policy.Revision)
if err != nil {
s.recordClaimFailure(policy.ID, level, err)
s.emitLevel(ctx, workspaceID, policy, spent, level)
@@ -187,15 +171,29 @@ func (s *CostService) claimAndEmit(
return true
}
// claimCompanionAlert records the alert-level claim alongside an
// exceeded-level emission, so a spend that jumps straight past the limit
// still de-escalates correctly. Its outcome never changes the
// budget.exceeded emission; only a claim-store error is reported, exactly
// like any other claim attempt.
func (s *CostService) claimCompanionAlert(ctx context.Context, policy *BudgetPolicy, periodKey string) {
if _, err := s.repo.Claim(ctx, policy.ID, periodKey, claimLevelAlert); err != nil {
s.recordClaimFailure(policy.ID, claimLevelAlert, err)
// claimExceededAndEmit resolves the atomic exceeded-plus-companion claim
// pair at the policy's current revision and emits budget.exceeded when the
// exceeded-level row was won. A claim-store error anywhere in the pair —
// including on the companion insert — fails the whole pair open: the error
// is logged and counted once, and budget.exceeded still emits.
func (s *CostService) claimExceededAndEmit(
ctx context.Context,
workspaceID string,
policy *BudgetPolicy,
spent int64,
periodKey string,
) bool {
claimed, err := s.repo.ClaimExceeded(ctx, policy.ID, periodKey, policy.Revision)
if err != nil {
s.recordClaimFailure(policy.ID, claimLevelExceeded, err)
s.emitLevel(ctx, workspaceID, policy, spent, claimLevelExceeded)
return true
}
if !claimed {
return false
}
s.emitLevel(ctx, workspaceID, policy, spent, claimLevelExceeded)
return true
}
func (s *CostService) emitLevel(
apps/backend/internal/office/costs/budgets_test.go
+73 -14
View File
@@ -74,14 +74,34 @@ type claimFaultRepo struct {
missLevels map[string]bool
}
func (r *claimFaultRepo) Claim(ctx context.Context, policyID, periodKey, level string) (bool, error) {
func (r *claimFaultRepo) Claim(ctx context.Context, policyID, periodKey, level string, revision int64) (bool, error) {
if err, ok := r.failLevels[level]; ok {
return false, err
}
if r.missLevels[level] {
return false, nil
}
return r.Repository.Claim(ctx, policyID, periodKey, level)
return r.Repository.Claim(ctx, policyID, periodKey, level, revision)
}
// ClaimExceeded overrides the atomic pair the same way Claim does, keyed by
// the level whose outcome is being faulted: "exceeded" fails or misses the
// pair before either insert is attempted (mirroring a fenced exceeded
// insert refused or store-faulted); "alert" fails or misses only the
// companion half, after a real exceeded insert has already run for real,
// mirroring AC-OFFICE-COSTS-003.10's atomic rollback of the whole pair on a
// companion-side error.
func (r *claimFaultRepo) ClaimExceeded(ctx context.Context, policyID, periodKey string, revision int64) (bool, error) {
if err, ok := r.failLevels["exceeded"]; ok {
return false, err
}
if r.missLevels["exceeded"] {
return false, nil
}
if err, ok := r.failLevels["alert"]; ok {
return false, err
}
return r.Repository.ClaimExceeded(ctx, policyID, periodKey, revision)
}
// callOrderRecorder records events in the order they occur, for tests that
@@ -113,12 +133,24 @@ type orderRecordingRepo struct {
order *callOrderRecorder
}
func (r *orderRecordingRepo) Claim(ctx context.Context, policyID, periodKey, level string) (bool, error) {
claimed, err := r.Repository.Claim(ctx, policyID, periodKey, level)
func (r *orderRecordingRepo) Claim(ctx context.Context, policyID, periodKey, level string, revision int64) (bool, error) {
claimed, err := r.Repository.Claim(ctx, policyID, periodKey, level, revision)
r.order.record("claim:" + level)
return claimed, err
}
// ClaimExceeded records both of the atomic pair's levels immediately after
// the (single) real call returns, preserving the ordering property the test
// cares about: both claim events precede the emission decision. It cannot
// observe the two inserts individually from outside the transaction, but it
// doesn't need to — only their position relative to the emission matters.
func (r *orderRecordingRepo) ClaimExceeded(ctx context.Context, policyID, periodKey string, revision int64) (bool, error) {
claimed, err := r.Repository.ClaimExceeded(ctx, policyID, periodKey, revision)
r.order.record("claim:exceeded")
r.order.record("claim:alert")
return claimed, err
}
// orderRecordingActivity implements shared.ActivityLogger and records each
// submitted action into the same callOrderRecorder as orderRecordingRepo,
// so a test can assert the interleaving of claims and emissions.
@@ -260,7 +292,7 @@ func insertBudgetTestCostEvent(t *testing.T, execSQL func(string, ...interface{}
now := time.Now().UTC().Format(time.RFC3339)
execSQL(
`INSERT INTO office_cost_events (id, agent_profile_id, task_id, cost_subcents, occurred_at, created_at)
VALUES (?, ?, ?, ?, ?, ?)`,
VALUES (?, ?, ?, ?, ?, ?)`,
uuid.NewString(), agentID, taskID, costSubcents, now, now,
)
}
@@ -283,13 +315,12 @@ func insertBudgetTestCostEventAt(
func TestCheckBudget_PeriodWindowsExcludeOlderSpend(t *testing.T) {
cases := []struct {
name string
period models.BudgetPeriod
olderBy time.Duration
wantLimit bool
name string
period models.BudgetPeriod
older time.Duration
}{
{name: "daily", period: models.BudgetPeriodDaily, olderBy: 48 * time.Hour},
{name: "yearly", period: models.BudgetPeriodYearly, olderBy: 400 * 24 * time.Hour},
{name: "daily", period: models.BudgetPeriodDaily, older: 48 * time.Hour},
{name: "yearly", period: models.BudgetPeriodYearly, older: 400 * 24 * time.Hour},
}
for _, tc := range cases {
@@ -311,7 +342,7 @@ func TestCheckBudget_PeriodWindowsExcludeOlderSpend(t *testing.T) {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestCostEventAt(t, execSQL, "agent-1", "task-1", 600, time.Now().UTC().Add(-tc.olderBy))
insertBudgetTestCostEventAt(t, execSQL, "agent-1", "task-1", 600, time.Now().UTC().Add(-tc.older))
results, err := svc.CheckBudget(ctx, "ws-1", "agent-1", "project-1")
if err != nil {
t.Fatalf("CheckBudget: %v", err)
@@ -319,8 +350,8 @@ func TestCheckBudget_PeriodWindowsExcludeOlderSpend(t *testing.T) {
if len(results) != 1 {
t.Fatalf("results = %d, want 1", len(results))
}
if results[0].LimitExceed != tc.wantLimit {
t.Errorf("LimitExceed = %t, want %t", results[0].LimitExceed, tc.wantLimit)
if results[0].LimitExceed {
t.Fatalf("%s spend outside the window must not exceed the limit", tc.name)
}
})
}
@@ -593,6 +624,34 @@ func TestEvaluateProjectBudget_AlertAtThreshold(t *testing.T) {
}
}
// TestCheckBudgetAndEvaluateProjectBudget_ShareAlertClaim covers the two
// callers that can evaluate a project policy. They must use the same durable
// claim so a cost event followed by task reassignment emits one alert.
func TestCheckBudgetAndEvaluateProjectBudget_ShareAlertClaim(t *testing.T) {
spy := &budgetActivitySpy{}
svc, _, execSQL := newBudgetTestServiceWithActivity(t, spy)
ctx := context.Background()
policy := newIdempotencyTestPolicy("proj-shared-claim", 1000)
policy.ScopeType = models.BudgetScopeProject
if err := svc.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
insertBudgetTestTask(t, execSQL, "task-shared-claim", "ws-1", "proj-shared-claim")
insertBudgetTestCostEvent(t, execSQL, "agent-shared-claim", "task-shared-claim", 850)
if _, err := svc.CheckBudget(ctx, "ws-1", "agent-shared-claim", "proj-shared-claim"); err != nil {
t.Fatalf("CheckBudget: %v", err)
}
if err := svc.EvaluateProjectBudget(ctx, "ws-1", "proj-shared-claim"); err != nil {
t.Fatalf("EvaluateProjectBudget: %v", err)
}
if got := spy.count("budget.alert"); got != 1 {
t.Fatalf("budget.alert submissions across callers = %d, want 1", got)
}
}
// TestEvaluateProjectBudget_ExceededAtLimit covers the limit-exceeded branch.
func TestEvaluateProjectBudget_ExceededAtLimit(t *testing.T) {
spy := &budgetActivitySpy{}
apps/backend/internal/office/costs/handler_budget_revision_test.go
+121
View File
@@ -0,0 +1,121 @@
package costs_test
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/office/costs"
)
// newBudgetTestRouter mounts the budget routes over a fresh CostService, for
// tests that need to observe the HTTP-level wire contract rather than the
// service API directly.
func newBudgetTestRouter(t *testing.T) *gin.Engine {
t.Helper()
svc, _, _ := newBudgetTestService(t)
gin.SetMode(gin.TestMode)
router := gin.New()
costs.NewHandler(svc).RegisterRoutes(router.Group("/api/v1"))
return router
}
func doBudgetRequest(t *testing.T, router *gin.Engine, method, path, body string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(method, path, strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
return rec
}
// TestCreateBudget_ResponseCarriesServerAssignedRevision covers
// AC-OFFICE-COSTS-003.2 and .4 on the wire: the 201 response's budget
// carries "revision": 1, never the column default leaking through as 0,
// and a client-presented revision in the create body has no effect (the
// request DTO has no such field, so this is enforced structurally, but the
// response must still report what the row actually holds).
func TestCreateBudget_ResponseCarriesServerAssignedRevision(t *testing.T) {
router := newBudgetTestRouter(t)
rec := doBudgetRequest(t, router, http.MethodPost, "/api/v1/workspaces/ws-1/budgets", `{
"scope_type": "workspace",
"limit_subcents": 1000,
"period": "monthly",
"alert_threshold_pct": 80,
"action_on_exceed": "notify_only",
"revision": 999
}`)
if rec.Code != http.StatusCreated {
t.Fatalf("status = %d, want 201 (body: %s)", rec.Code, rec.Body.String())
}
var body struct {
Budget struct {
ID string `json:"id"`
Revision int64 `json:"revision"`
} `json:"budget"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
t.Fatalf("decode body %q: %v", rec.Body.String(), err)
}
if body.Budget.Revision != 1 {
t.Fatalf("create response revision = %d, want 1 (a presented 999 must not survive)", body.Budget.Revision)
}
if !strings.Contains(rec.Body.String(), `"revision"`) {
t.Fatalf("response body must serialize revision as a visible field, got %s", rec.Body.String())
}
}
// TestUpdateBudget_ClientRevisionFieldHasNoEffect covers
// AC-OFFICE-COSTS-003.7: the update request body has no revision field, so
// a client-supplied "revision" in the PATCH JSON is inert, and the response
// reports the server's own bump (2, after the create's 1) regardless of
// what the caller sent.
func TestUpdateBudget_ClientRevisionFieldHasNoEffect(t *testing.T) {
router := newBudgetTestRouter(t)
createRec := doBudgetRequest(t, router, http.MethodPost, "/api/v1/workspaces/ws-1/budgets", `{
"scope_type": "workspace",
"limit_subcents": 1000,
"period": "monthly",
"alert_threshold_pct": 80,
"action_on_exceed": "notify_only"
}`)
var created struct {
Budget struct {
ID string `json:"id"`
} `json:"budget"`
}
if err := json.Unmarshal(createRec.Body.Bytes(), &created); err != nil {
t.Fatalf("decode create body %q: %v", createRec.Body.String(), err)
}
updateRec := doBudgetRequest(t, router, http.MethodPatch, "/api/v1/budgets/"+created.Budget.ID, `{
"limit_subcents": 2000,
"revision": 999999
}`)
if updateRec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200 (body: %s)", updateRec.Code, updateRec.Body.String())
}
var updated struct {
Budget struct {
Revision int64 `json:"revision"`
LimitSubcents int64 `json:"limit_subcents"`
} `json:"budget"`
}
if err := json.Unmarshal(updateRec.Body.Bytes(), &updated); err != nil {
t.Fatalf("decode update body %q: %v", updateRec.Body.String(), err)
}
if updated.Budget.Revision != 2 {
t.Fatalf("updated revision = %d, want 2 (a presented 999999 must not survive)", updated.Budget.Revision)
}
if updated.Budget.LimitSubcents != 2000 {
t.Fatalf("limit_subcents = %d, want 2000", updated.Budget.LimitSubcents)
}
}
apps/backend/internal/office/costs/service.go
+7 -2
View File
@@ -44,8 +44,13 @@ type Repository interface {
GetWorkspaceBudgetDefault(ctx context.Context, workspaceID string) (limitSubcents int64, found bool, err error)
SetWorkspaceBudgetDefault(ctx context.Context, workspaceID string, limitSubcents int64) error
// Claim atomically records that (policyID, periodKey, level) may emit its
// budget notification. See docs/specs/office/requirements/costs.md.
Claim(ctx context.Context, policyID, periodKey, level string) (bool, error)
// budget notification, fenced to revision. See
// docs/specs/budget-claim-revision-fencing/spec.md.
Claim(ctx context.Context, policyID, periodKey, level string, revision int64) (bool, error)
// ClaimExceeded atomically records the exceeded-level claim and, only
// when this call wins it, the alert-level companion claim, both fenced
// to revision. See docs/specs/budget-claim-revision-fencing/spec.md.
ClaimExceeded(ctx context.Context, policyID, periodKey string, revision int64) (bool, error)
}
// CostService handles cost recording, summaries, and budget evaluation.
apps/backend/internal/office/models/models.go
+4
View File
@@ -375,6 +375,10 @@ type BudgetPolicy struct {
ActionOnExceed BudgetActionOnExceed `json:"action_on_exceed" db:"action_on_exceed"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
UpdatedAt time.Time `json:"updated_at" db:"updated_at"`
// Revision identifies which immutable set of the fields above a budget
// claim was evaluated against (REQ-OFFICE-COSTS-003). Server-assigned:
// starts at 1 and increases by exactly 1 on each successful update.
Revision int64 `json:"revision" db:"revision"`
}
// SpendWindow reports one scope's priced spend and degradation state for a
apps/backend/internal/office/repository/sqlite/base.go
+46 -13
View File
@@ -76,6 +76,24 @@ type Repository struct {
// before the policy row update, so a test can prove the two are
// transactional without a fault-injecting driver.
failBudgetPolicyUpdateErr error
// failBudgetClaimsRecreateErr is a test-only failpoint: when set,
// recreateBudgetClaimsForRevision returns it instead of touching the
// database, so a test can prove initSchema surfaces the error rather
// than swallowing it.
failBudgetClaimsRecreateErr error
// failBudgetClaimsRecreateAfterDropErr is a test-only failpoint: when
// set, recreateBudgetClaimsForRevision returns it after DROP TABLE has
// run but before CREATE TABLE, inside the same transaction, so a test
// can prove the two are atomic rather than two independent statements.
failBudgetClaimsRecreateAfterDropErr error
// failBudgetExceededCompanionErr is a test-only failpoint: when set,
// ClaimExceeded returns it after the exceeded-level insert has run but
// before the companion alert-level insert, so a test can prove the
// pair rolls back together.
failBudgetExceededCompanionErr error
}
// NewWithDB creates a new office repository with existing database connections.
@@ -126,6 +144,12 @@ func (r *Repository) initSchema() error {
if err := r.createCoreTables(); err != nil {
return err
}
// Runs after createCoreTables (which creates office_budget_claims) so
// it probes the real table rather than racing its creation. Its error
// is returned, not swallowed: see recreateBudgetClaimsForRevision.
if err := r.recreateBudgetClaimsForRevision(); err != nil {
return fmt.Errorf("recreate office_budget_claims: %w", err)
}
if err := r.createExtensionTables(); err != nil {
return err
}
@@ -338,7 +362,8 @@ func (r *Repository) createCostTables() error {
alert_threshold_pct INTEGER DEFAULT 80,
action_on_exceed TEXT DEFAULT 'notify_only',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
updated_at TIMESTAMP NOT NULL,
revision INTEGER NOT NULL DEFAULT 1
);
-- Built-in default spend ceiling (REQ-OFFICE-BUDGET-003): a stable
@@ -349,26 +374,34 @@ func (r *Repository) createCostTables() error {
limit_subcents INTEGER NOT NULL,
updated_at TIMESTAMP NOT NULL
);
` + budgetClaimsDDL)
return err
}
-- office_budget_claims must be created after office_budget_policies:
-- its foreign key references that table, and PostgreSQL rejects a
-- forward reference at CREATE TABLE time even though SQLite tolerates
-- it. One row per (policy, evaluation period, level) that has already
-- produced its budget.alert / budget.exceeded notification; the
-- primary key is the concurrency control for a claim attempt.
-- ON DELETE CASCADE removes a policy's claims on every deletion path
-- without a matching code change at any of them.
// budgetClaimsDDL is the final office_budget_claims shape: one row per
// (policy, evaluation period, level, policy revision) that has already
// produced its budget.alert / budget.exceeded notification. revision is
// part of the primary key, not just a fence input, so a claim written
// against a revision a concurrent update has already superseded can never
// match a later evaluation's fenced insert. Used both for fresh-database
// creation here and for the recreate in recreateBudgetClaimsForRevision, so
// a fresh database and a migrated one converge. office_budget_claims must
// be created after office_budget_policies: its foreign key references that
// table, and PostgreSQL rejects a forward reference at CREATE TABLE time
// even though SQLite tolerates it. ON DELETE CASCADE removes a policy's
// claims on every deletion path without a matching code change at any of
// them.
const budgetClaimsDDL = `
CREATE TABLE IF NOT EXISTS office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
revision INTEGER NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
PRIMARY KEY (policy_id, period_key, level, revision),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
`)
return err
}
`
func (r *Repository) createRunTables() error {
_, err := r.db.Exec(`
apps/backend/internal/office/repository/sqlite/base_migrations.go
+12
View File
@@ -63,6 +63,7 @@ func (r *Repository) runMigrations() error {
r.migrateWakeWaveColumns()
r.migrate.Apply("task_workspace_groups.ownership_generation",
`ALTER TABLE task_workspace_groups ADD COLUMN ownership_generation INTEGER NOT NULL DEFAULT 1`)
r.migrateBudgetPolicyRevision()
r.migrateWorkspacePauseSkipAttribution()
r.migrateLoopLivenessCausationID()
if err := r.migrate.Err(); err != nil {
@@ -86,6 +87,17 @@ func (r *Repository) backfillRoutineTriggerTimezones() error {
return nil
}
// migrateBudgetPolicyRevision adds office_budget_policies.revision for
// databases created before REQ-OFFICE-COSTS-003. The DEFAULT 1 backfills
// every existing row exactly like createCostTables' inline column, so a
// fresh database and a migrated one converge. A boot replay is a no-op:
// db.IsDuplicateColumnError is the only local classifier (ADR 0027), reused
// via MigrateLogger.Apply rather than adding a new one.
func (r *Repository) migrateBudgetPolicyRevision() {
_ = r.migrate.Apply("office_budget_policies.revision",
`ALTER TABLE office_budget_policies ADD COLUMN revision INTEGER NOT NULL DEFAULT 1`)
}
// migrateWorkspacePauseSkipAttribution adds the columns a blocked routine
// fire uses to record why it was skipped and which pause blocked it. The
// partial unique index runs after both ADD COLUMNs so it never executes
apps/backend/internal/office/repository/sqlite/budget_atomicity_internal_test.go
+6 -2
View File
@@ -16,7 +16,8 @@ import (
// discard and the policy row update are one transaction
// (AC-OFFICE-COSTS-002.8): when the row update fails after the discard has
// already run, the whole transaction rolls back, so the claim survives
// alongside the un-updated policy. failBudgetPolicyUpdateErr is a
// alongside the un-updated policy, and the policy's revision does not
// advance either (AC-OFFICE-COSTS-003.6). failBudgetPolicyUpdateErr is a
// test-only failpoint (see base.go) standing in for a fault-injecting
// driver.
func TestUpdateBudgetPolicy_FailedUpdateRollsBackDiscard(t *testing.T) {
@@ -53,7 +54,7 @@ func TestUpdateBudgetPolicy_FailedUpdateRollsBackDiscard(t *testing.T) {
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
if _, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert"); err != nil {
if _, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision); err != nil {
t.Fatalf("claim: %v", err)
}
@@ -80,4 +81,7 @@ func TestUpdateBudgetPolicy_FailedUpdateRollsBackDiscard(t *testing.T) {
if stored.LimitSubcents != 1000 {
t.Fatalf("limit_subcents after rolled-back update = %d, want unchanged 1000", stored.LimitSubcents)
}
if stored.Revision != 1 {
t.Fatalf("revision after rolled-back update = %d, want unchanged 1 (a rolled-back update must not consume a revision number)", stored.Revision)
}
}
apps/backend/internal/office/repository/sqlite/budget_claims_postgres_test.go
+192 -5
View File
@@ -56,7 +56,7 @@ func TestPostgresBudgetClaims(t *testing.T) {
t.Fatalf("create budget policy: %v", err)
}
claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision)
if err != nil {
t.Fatalf("first claim: %v", err)
}
@@ -64,7 +64,7 @@ func TestPostgresBudgetClaims(t *testing.T) {
t.Fatal("first claim should win")
}
claimed, err = repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
claimed, err = repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision)
if err != nil {
t.Fatalf("second claim: %v", err)
}
@@ -72,9 +72,9 @@ func TestPostgresBudgetClaims(t *testing.T) {
t.Fatal("second claim on the same (policy, period, level) must not win")
}
// A foreign-key violation (no such policy) must classify as an ordinary
// miss, not a store error — db.IsForeignKeyViolation's PostgreSQL branch.
claimed, err = repo.Claim(ctx, uuid.NewString(), "lifetime", "alert")
// A claim for a nonexistent policy must be an ordinary fenced miss, not a
// store error. The INSERT ... SELECT fence prevents an FK violation here.
claimed, err = repo.Claim(ctx, uuid.NewString(), "lifetime", "alert", 1)
if err != nil {
t.Fatalf("claim against a nonexistent policy must not be a store error, got: %v", err)
}
@@ -95,3 +95,190 @@ func TestPostgresBudgetClaims(t *testing.T) {
t.Fatalf("claim count after policy delete = %d, want 0 (FK ON DELETE CASCADE)", claimCount)
}
}
// TestPostgresBudgetClaims_RevisionFencingAndFourColumnKey is the PostgreSQL
// twin of the revision-fencing coverage in budget_revision_fencing_test.go:
// under READ COMMITTED a claim's WHERE EXISTS fence check and the row it
// inserts both need proving on a real PostgreSQL backend, not just SQLite's
// single-writer connection. It covers both halves called out by the fence
// design: the fence itself refuses an insert against a superseded revision
// (a stale evaluation cannot emit an obsolete claim), and revision being
// part of the primary key means a claim row already recorded for an old
// revision does not collide with — and so cannot suppress — a fresh
// evaluation's claim at the new revision. Skips unless
// KANDEV_TEST_POSTGRES_DSN is set.
func TestPostgresBudgetClaims_RevisionFencingAndFourColumnKey(t *testing.T) {
db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t))
ctx := context.Background()
if _, err := taskrepo.NewWithDB(db, db, nil); err != nil {
t.Fatalf("init task repo: %v", err)
}
repo, err := sqlite.NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("boot: %v", err)
}
policy := &models.BudgetPolicy{
WorkspaceID: "pg-budget-fencing-ws",
ScopeType: "workspace",
ScopeID: "pg-budget-fencing-ws",
LimitSubcents: 1000,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create budget policy: %v", err)
}
staleRevision := policy.Revision
// A claim recorded at the current revision, before any update.
claimed, err := repo.Claim(ctx, policy.ID, "lifetime", "alert", staleRevision)
if err != nil {
t.Fatalf("claim at revision %d: %v", staleRevision, err)
}
if !claimed {
t.Fatal("claim at the current revision should win")
}
if err := repo.UpdateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("update policy: %v", err)
}
if policy.Revision != staleRevision+1 {
t.Fatalf("revision after update = %d, want %d", policy.Revision, staleRevision+1)
}
// The fence: a claim presented with the now-superseded revision must be
// refused, even though it targets a different period so it collides
// with nothing already in the table.
claimed, err = repo.Claim(ctx, policy.ID, "stale-period", "alert", staleRevision)
if err != nil {
t.Fatalf("stale-revision claim must be an ordinary miss, got error: %v", err)
}
if claimed {
t.Fatal("a claim presented with a superseded revision must not win")
}
// A surviving old-revision row must not suppress a fresh claim. The
// direct insert models a claim that was already committed by an older
// evaluation while the policy update moved the current revision.
if _, err := db.ExecContext(ctx, `
INSERT INTO office_budget_claims (policy_id, period_key, level, revision, claimed_at)
VALUES ($1, 'lifetime', 'alert', $2, NOW())
`, policy.ID, staleRevision); err != nil {
t.Fatalf("seed surviving old-revision claim: %v", err)
}
// The four-column key lets the current revision claim the same period and
// level without colliding with that old-revision row.
claimed, err = repo.Claim(ctx, policy.ID, "lifetime", "alert", policy.Revision)
if err != nil {
t.Fatalf("claim at the new revision: %v", err)
}
if !claimed {
t.Fatal("claim at the current (post-update) revision should win")
}
var claimCount int
if err := db.GetContext(ctx, &claimCount,
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = $1`, policy.ID,
); err != nil {
t.Fatalf("count claims: %v", err)
}
if claimCount != 2 {
t.Fatalf("claim rows for policy = %d, want 2 (old and current revisions must coexist)", claimCount)
}
}
// TestPostgresBudgetClaims_LegacyTableRecreated verifies the destructive
// legacy-table cutover on a real PostgreSQL backend. The old three-column
// table is replaced in one transaction, and a later boot must replay without
// replacing the new table again. Skips unless KANDEV_TEST_POSTGRES_DSN is set.
func TestPostgresBudgetClaims_LegacyTableRecreated(t *testing.T) {
db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t))
ctx := context.Background()
if _, err := taskrepo.NewWithDB(db, db, nil); err != nil {
t.Fatalf("init task repo: %v", err)
}
repo, err := sqlite.NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("initial boot: %v", err)
}
policy := &models.BudgetPolicy{
WorkspaceID: "pg-budget-legacy-ws",
ScopeType: "workspace",
ScopeID: "pg-budget-legacy-ws",
LimitSubcents: 1000,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create budget policy: %v", err)
}
// Replace the current table with the pre-revision three-column shape and
// leave a row in it. The next office boot must take the recreate path.
if _, err := db.ExecContext(ctx, `DROP TABLE office_budget_claims`); err != nil {
t.Fatalf("drop current claims table: %v", err)
}
if _, err := db.ExecContext(ctx, `
CREATE TABLE office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
)
`); err != nil {
t.Fatalf("create legacy claims table: %v", err)
}
if _, err := db.ExecContext(ctx, `
INSERT INTO office_budget_claims (policy_id, period_key, level, claimed_at)
VALUES ($1, 'legacy', 'alert', NOW())
`, policy.ID); err != nil {
t.Fatalf("seed legacy claim: %v", err)
}
if _, err := sqlite.NewWithDB(db, db, nil); err != nil {
t.Fatalf("boot that recreates legacy claims table: %v", err)
}
var revisionColumns int
if err := db.GetContext(ctx, &revisionColumns, `
SELECT COUNT(*) FROM information_schema.columns
WHERE table_schema = current_schema()
AND table_name = 'office_budget_claims'
AND column_name = 'revision'
`); err != nil {
t.Fatalf("probe recreated claims table: %v", err)
}
if revisionColumns != 1 {
t.Fatalf("office_budget_claims.revision columns = %d, want 1", revisionColumns)
}
claimed, err := repo.Claim(ctx, policy.ID, "replayed", "alert", policy.Revision)
if err != nil {
t.Fatalf("claim after legacy recreate: %v", err)
}
if !claimed {
t.Fatal("claim after legacy recreate should win")
}
if _, err := sqlite.NewWithDB(db, db, nil); err != nil {
t.Fatalf("replay boot after recreate: %v", err)
}
var claimCount int
if err := db.GetContext(ctx, &claimCount,
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = $1`, policy.ID,
); err != nil {
t.Fatalf("count claims after replay: %v", err)
}
if claimCount != 1 {
t.Fatalf("claim rows after replay = %d, want 1", claimCount)
}
}
apps/backend/internal/office/repository/sqlite/budget_claims_recreate_internal_test.go
+156
View File
@@ -0,0 +1,156 @@
package sqlite
import (
"errors"
"testing"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
"github.com/kandev/kandev/internal/db"
runssqlite "github.com/kandev/kandev/internal/runs/repository/sqlite"
)
// TestRecreateBudgetClaimsForRevision_FailureIsVisible covers the other
// half of AC-OFFICE-COSTS-003.3a: a failed recreate must reach the caller
// that initializes the schema, not be logged and swallowed the way the
// neighbouring runMigrations()/activateRunOutcome() calls are.
// failBudgetClaimsRecreateErr is a test-only failpoint (see base.go)
// standing in for a fault-injecting driver; the repo is built by hand
// (mirroring NewWithDB) so the failpoint can be set before initSchema runs,
// while the database still carries the pre-.3 three-column claims table the
// probe needs to find revision absent in.
func TestRecreateBudgetClaimsForRevision_FailureIsVisible(t *testing.T) {
sqlDB, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
sqlDB.SetMaxOpenConns(1)
t.Cleanup(func() { _ = sqlDB.Close() })
if _, err := sqlDB.Exec(`
CREATE TABLE office_budget_policies (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL,
scope_type TEXT NOT NULL,
scope_id TEXT NOT NULL,
limit_subcents INTEGER NOT NULL,
period TEXT NOT NULL,
alert_threshold_pct INTEGER DEFAULT 80,
action_on_exceed TEXT DEFAULT 'notify_only',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE TABLE office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
`); err != nil {
t.Fatalf("seed legacy office_budget_claims: %v", err)
}
repo := &Repository{
Repository: runssqlite.NewWithDB(sqlDB, sqlDB),
db: sqlDB,
ro: sqlDB,
migrate: db.NewRequiredMigrateLogger(sqlDB, nil),
failBudgetClaimsRecreateErr: errors.New("injected recreate failure"),
}
if err := repo.initSchema(); err == nil {
t.Fatal("expected initSchema to surface the injected recreate failure, got nil error")
}
}
// budgetClaimsHasRevisionColumn reports whether office_budget_claims
// currently declares a revision column, distinguishing the old three-column
// shape from the final four-column one.
func budgetClaimsHasRevisionColumn(t *testing.T, sqlDB *sqlx.DB) bool {
t.Helper()
var count int
if err := sqlDB.Get(&count,
`SELECT COUNT(*) FROM pragma_table_info('office_budget_claims') WHERE name = 'revision'`,
); err != nil {
t.Fatalf("probe office_budget_claims columns: %v", err)
}
return count == 1
}
// TestRecreateBudgetClaimsForRevision_AtomicAcrossDropAndCreate covers the
// crash-safety half of AC-OFFICE-COSTS-003.3a: DROP and CREATE must commit
// or fail together, not as two independent statements. A boot interrupted
// between them must not leave office_budget_claims absent, because a
// missing table and an old-shape table both probe as "revision absent" and
// an unconditional DROP TABLE (no IF EXISTS) on an already-missing table
// would error forever on every later boot. failBudgetClaimsRecreateAfterDropErr
// (see base.go) fires after DROP has run but before CREATE, inside the same
// transaction, standing in for a process crash at that point.
func TestRecreateBudgetClaimsForRevision_AtomicAcrossDropAndCreate(t *testing.T) {
sqlDB, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
sqlDB.SetMaxOpenConns(1)
t.Cleanup(func() { _ = sqlDB.Close() })
if _, err := sqlDB.Exec(`
CREATE TABLE office_budget_policies (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL,
scope_type TEXT NOT NULL,
scope_id TEXT NOT NULL,
limit_subcents INTEGER NOT NULL,
period TEXT NOT NULL,
alert_threshold_pct INTEGER DEFAULT 80,
action_on_exceed TEXT DEFAULT 'notify_only',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE TABLE office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
`); err != nil {
t.Fatalf("seed legacy office_budget_claims: %v", err)
}
repo := &Repository{
Repository: runssqlite.NewWithDB(sqlDB, sqlDB),
db: sqlDB,
ro: sqlDB,
migrate: db.NewRequiredMigrateLogger(sqlDB, nil),
}
repo.failBudgetClaimsRecreateAfterDropErr = errors.New("injected crash between drop and create")
if err := repo.initSchema(); err == nil {
t.Fatal("expected initSchema to surface the injected failure, got nil error")
}
if budgetClaimsHasRevisionColumn(t, sqlDB) {
t.Fatal("a failure between DROP and CREATE must not leave the new shape partially applied")
}
var count int
if err := sqlDB.Get(&count,
`SELECT COUNT(*) FROM pragma_table_info('office_budget_claims')`,
); err != nil {
t.Fatalf("count office_budget_claims columns: %v", err)
}
if count == 0 {
t.Fatal("a failure between DROP and CREATE must not leave office_budget_claims absent; the DROP must roll back too")
}
repo.failBudgetClaimsRecreateAfterDropErr = nil
if err := repo.initSchema(); err != nil {
t.Fatalf("retry boot after the injected failure must self-heal, got: %v", err)
}
if !budgetClaimsHasRevisionColumn(t, sqlDB) {
t.Fatal("retry boot must complete the recreate and leave the revision column present")
}
}
apps/backend/internal/office/repository/sqlite/budget_exceeded_atomicity_internal_test.go
+71
View File
@@ -0,0 +1,71 @@
package sqlite
import (
"context"
"errors"
"testing"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
settingsstore "github.com/kandev/kandev/internal/agent/settings/store"
"github.com/kandev/kandev/internal/office/models"
)
// TestClaimExceeded_CompanionFailure_RollsBackWholePair proves the
// exceeded-level claim and its alert-level companion are one transaction
// (AC-OFFICE-COSTS-003.10): when the companion insert fails after the
// exceeded insert has already run, the whole transaction rolls back, so
// neither row survives. failBudgetExceededCompanionErr is a test-only
// failpoint (see base.go) standing in for a fault-injecting driver.
func TestClaimExceeded_CompanionFailure_RollsBackWholePair(t *testing.T) {
db, err := sqlx.Open("sqlite3", ":memory:?_foreign_keys=on")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
db.SetMaxOpenConns(1)
t.Cleanup(func() { _ = db.Close() })
if _, err := db.Exec(`PRAGMA foreign_keys = ON`); err != nil {
t.Fatalf("pragma fk: %v", err)
}
if _, _, err := settingsstore.Provide(db, db, nil); err != nil {
t.Fatalf("settings store init: %v", err)
}
repo, err := NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("new repo: %v", err)
}
ctx := context.Background()
policy := &models.BudgetPolicy{
WorkspaceID: "ws-companion-rollback",
ScopeType: "workspace",
ScopeID: "ws-companion-rollback",
LimitSubcents: 1000,
Period: "monthly",
AlertThresholdPct: 80,
ActionOnExceed: "notify_only",
}
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
repo.failBudgetExceededCompanionErr = errors.New("injected companion failure")
claimed, err := repo.ClaimExceeded(ctx, policy.ID, "lifetime", policy.Revision)
if err == nil {
t.Fatal("expected the injected companion failure to surface")
}
if claimed {
t.Fatal("a failed pair must report claimed=false")
}
var count int
if err := db.QueryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ?`, policy.ID,
).Scan(&count); err != nil {
t.Fatalf("count claims: %v", err)
}
if count != 0 {
t.Fatalf("claim rows after a rolled-back pair = %d, want 0 (neither row must survive)", count)
}
}
apps/backend/internal/office/repository/sqlite/budget_revision_fencing_test.go
+424
View File
@@ -0,0 +1,424 @@
package sqlite_test
import (
"context"
"sync"
"testing"
"github.com/jmoiron/sqlx"
_ "github.com/mattn/go-sqlite3"
"github.com/kandev/kandev/internal/office/repository/sqlite"
)
// TestBudgetPolicyRevisionMigration_LegacyRowsBackfillToOne covers
// AC-OFFICE-COSTS-003.1: a database created before the revision column
// exists gets it added as NOT NULL DEFAULT 1, an existing row is backfilled
// to 1, and a boot replay is a no-op (no new local error classifier is
// exercised beyond db.IsDuplicateColumnError, ADR 0027).
func TestBudgetPolicyRevisionMigration_LegacyRowsBackfillToOne(t *testing.T) {
db, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
db.SetMaxOpenConns(1)
t.Cleanup(func() { _ = db.Close() })
// Seed the pre-.3 office_budget_policies shape (no revision column) and
// a legacy row, before the repo (and its migration) ever runs.
if _, err := db.Exec(`
CREATE TABLE office_budget_policies (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL,
scope_type TEXT NOT NULL,
scope_id TEXT NOT NULL,
limit_subcents INTEGER NOT NULL,
period TEXT NOT NULL,
alert_threshold_pct INTEGER DEFAULT 80,
action_on_exceed TEXT DEFAULT 'notify_only',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
)`); err != nil {
t.Fatalf("seed legacy office_budget_policies: %v", err)
}
if _, err := db.Exec(`
INSERT INTO office_budget_policies (
id, workspace_id, scope_type, scope_id, limit_subcents, period, created_at, updated_at
) VALUES ('legacy-1', 'ws-legacy', 'workspace', 'ws-legacy', 1000, 'monthly', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
`); err != nil {
t.Fatalf("seed legacy row: %v", err)
}
if _, err := sqlite.NewWithDB(db, db, nil); err != nil {
t.Fatalf("first boot: %v", err)
}
var revision int64
if err := db.QueryRow(`SELECT revision FROM office_budget_policies WHERE id = 'legacy-1'`).Scan(&revision); err != nil {
t.Fatalf("select revision on legacy row: %v", err)
}
if revision != 1 {
t.Fatalf("legacy row revision = %d, want 1", revision)
}
// Boot replay must be a no-op, not an error.
if _, err := sqlite.NewWithDB(db, db, nil); err != nil {
t.Fatalf("second boot (replay): %v", err)
}
}
// legacyBudgetClaimsFixture builds an in-memory database still carrying
// PR #3287's three-column office_budget_claims shape (no revision, no FK
// dependency needed for these assertions), so the AC-OFFICE-COSTS-003.3a
// recreate actually has to run rather than being short-circuited by a fresh
// database's inline final-shape DDL.
func legacyBudgetClaimsFixture(t *testing.T) *sqlx.DB {
t.Helper()
db, err := sqlx.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
db.SetMaxOpenConns(1)
t.Cleanup(func() { _ = db.Close() })
if _, err := db.Exec(`
CREATE TABLE office_budget_policies (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL,
scope_type TEXT NOT NULL,
scope_id TEXT NOT NULL,
limit_subcents INTEGER NOT NULL,
period TEXT NOT NULL,
alert_threshold_pct INTEGER DEFAULT 80,
action_on_exceed TEXT DEFAULT 'notify_only',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE TABLE office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
`); err != nil {
t.Fatalf("seed legacy office_budget_claims: %v", err)
}
return db
}
// TestRecreateBudgetClaims_OldShapeDatabase_GuardedAcrossBoots covers
// AC-OFFICE-COSTS-003.3a: booting against a database still carrying the
// pre-.3 three-column claim table runs the recreate (destroying whatever
// claims existed before that boot, which AC-OFFICE-COSTS-003.3 expressly
// permits); a claim written *after* that boot survives a second boot, which
// must probe, find revision present, and do nothing.
func TestRecreateBudgetClaims_OldShapeDatabase_GuardedAcrossBoots(t *testing.T) {
db := legacyBudgetClaimsFixture(t)
if _, err := db.Exec(`
INSERT INTO office_budget_policies (
id, workspace_id, scope_type, scope_id, limit_subcents, period, created_at, updated_at
) VALUES ('p1', 'ws-1', 'workspace', 'ws-1', 1000, 'monthly', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
`); err != nil {
t.Fatalf("seed policy: %v", err)
}
// A claim held before the cutover boot: the recreate is permitted to
// destroy it (AC-OFFICE-COSTS-003.3's accepted first-run behavior).
if _, err := db.Exec(`
INSERT INTO office_budget_claims (policy_id, period_key, level, claimed_at)
VALUES ('p1', 'lifetime', 'alert', CURRENT_TIMESTAMP)
`); err != nil {
t.Fatalf("seed pre-cutover claim: %v", err)
}
repo, err := sqlite.NewWithDB(db, db, nil)
if err != nil {
t.Fatalf("first boot (cutover): %v", err)
}
var count int
if err := db.QueryRow(`SELECT COUNT(*) FROM office_budget_claims`).Scan(&count); err != nil {
t.Fatalf("count claims after cutover: %v", err)
}
if count != 0 {
t.Fatalf("claims after cutover = %d, want 0 (pre-cutover claims are destroyed by the recreate)", count)
}
var revisionColExists int
if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('office_budget_claims') WHERE name = 'revision'`).
Scan(&revisionColExists); err != nil {
t.Fatalf("probe revision column: %v", err)
}
if revisionColExists != 1 {
t.Fatal("office_budget_claims.revision must exist after the cutover boot")
}
// A claim written after the cutover boot must survive a second boot.
ctx := context.Background()
if claimed, err := repo.Claim(ctx, "p1", "lifetime", "alert", 1); err != nil || !claimed {
t.Fatalf("post-cutover claim: claimed=%v err=%v", claimed, err)
}
if _, err := sqlite.NewWithDB(db, db, nil); err != nil {
t.Fatalf("second boot (replay): %v", err)
}
if err := db.QueryRow(`SELECT COUNT(*) FROM office_budget_claims`).Scan(&count); err != nil {
t.Fatalf("count claims after replay: %v", err)
}
if count != 1 {
t.Fatalf("claims after replay boot = %d, want 1 (the probe must no-op on a second boot)", count)
}
}
// TestCreateBudgetPolicy_AssignsRevisionOne_IgnoringPresentedValue covers
// AC-OFFICE-COSTS-003.4: revision is always 1 on create, regardless of any
// value already set on the policy struct passed in, and the assigned value
// is written back so the caller sees what the row actually holds.
func TestCreateBudgetPolicy_AssignsRevisionOne_IgnoringPresentedValue(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-revision-create")
policy.Revision = 777 // presented before a hypothetical re-create; must not survive.
policy.ID = "" // force a fresh insert
if err := repo.CreateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("create policy: %v", err)
}
if policy.Revision != 1 {
t.Fatalf("revision returned to caller = %d, want 1", policy.Revision)
}
var stored int64
if err := db.QueryRow(`SELECT revision FROM office_budget_policies WHERE id = ?`, policy.ID).Scan(&stored); err != nil {
t.Fatalf("select stored revision: %v", err)
}
if stored != 1 {
t.Fatalf("stored revision = %d, want 1", stored)
}
}
// TestUpdateBudgetPolicy_NonexistentPolicy_ReturnsErrorNoCommit covers the
// AC-OFFICE-COSTS-003.5 tightening: updating a policy id that no longer
// exists rolls the transaction back and reports an error, a behavior change
// from #3287 where the update ignored its affected-row count and reported
// success.
func TestUpdateBudgetPolicy_NonexistentPolicy_ReturnsErrorNoCommit(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-revision-missing")
if err := repo.DeleteBudgetPolicy(ctx, policy.ID); err != nil {
t.Fatalf("delete policy: %v", err)
}
policy.LimitSubcents = 9999
if err := repo.UpdateBudgetPolicy(ctx, policy); err == nil {
t.Fatal("expected an error updating a policy id that no longer exists")
}
}
// TestUpdateBudgetPolicy_ConcurrentUpdates_RevisionAdvancesByTwo covers
// concurrency case 6 of the forced-to-invent pass: two concurrent updates
// each bump by 1, and the final revision is r+2 with no increment lost —
// serialized by SQLite's single-writer connection, exactly as production
// runs (internal/db/pool.go).
func TestUpdateBudgetPolicy_ConcurrentUpdates_RevisionAdvancesByTwo(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-revision-concurrent")
var wg sync.WaitGroup
wg.Add(2)
for i := 0; i < 2; i++ {
go func(limit int64) {
defer wg.Done()
p, err := repo.GetBudgetPolicy(ctx, policy.ID)
if err != nil {
t.Errorf("get policy: %v", err)
return
}
p.LimitSubcents = limit
if err := repo.UpdateBudgetPolicy(ctx, p); err != nil {
t.Errorf("update policy: %v", err)
}
}(int64(2000 + i))
}
wg.Wait()
var revision int64
if err := db.QueryRow(`SELECT revision FROM office_budget_policies WHERE id = ?`, policy.ID).Scan(&revision); err != nil {
t.Fatalf("select final revision: %v", err)
}
if revision != 3 {
t.Fatalf("final revision = %d, want 3 (started at 1, two updates each +1)", revision)
}
}
// TestClaim_RaceA_StaleRevisionRefused_FreshRevisionSucceeds covers
// AC-OFFICE-COSTS-003.8/.9/.15 (the verification.md "Race A regression"
// bullet): a claim presenting a superseded revision is refused as an
// ordinary miss (no error), and the slot for the new revision stays open so
// a claim at the current revision still wins.
func TestClaim_RaceA_StaleRevisionRefused_FreshRevisionSucceeds(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-race-a")
staleRevision := policy.Revision // 1
// Concurrent UpdateBudgetPolicy commits, bumping the stored revision to 2.
policy.LimitSubcents = 2000
if err := repo.UpdateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("update policy: %v", err)
}
if policy.Revision != staleRevision+1 {
t.Fatalf("revision after update = %d, want %d", policy.Revision, staleRevision+1)
}
// The stale evaluation's claim, fenced to the superseded revision, must
// be refused with no error.
claimed, err := repo.Claim(ctx, policy.ID, "lifetime", "alert", staleRevision)
if err != nil {
t.Fatalf("stale claim must not be a store error, got: %v", err)
}
if claimed {
t.Fatal("a claim fenced to a superseded revision must not win")
}
// The slot for the new revision must still be open.
claimed, err = repo.Claim(ctx, policy.ID, "lifetime", "alert", policy.Revision)
if err != nil {
t.Fatalf("fresh-revision claim: %v", err)
}
if !claimed {
t.Fatal("a claim fenced to the current revision must win even after a stale claim was refused")
}
}
// TestClaim_InvalidInput_FailsOpenNotSilently covers AC-OFFICE-COSTS-003.13:
// each of the four primary-key components, when invalid, is a claim-store
// error (never the silent miss AC-OFFICE-COSTS-003.9 reserves for a refused
// fence), so the guard must run before the fenced statement rather than as
// part of it.
func TestClaim_InvalidInput_FailsOpenNotSilently(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-invalid-input")
cases := []struct {
name string
policyID, period string
level string
revision int64
}{
{"zero revision", policy.ID, "lifetime", "alert", 0},
{"negative revision", policy.ID, "lifetime", "alert", -1},
{"empty policy id", "", "lifetime", "alert", policy.Revision},
{"empty period key", policy.ID, "", "alert", policy.Revision},
{"empty level", policy.ID, "lifetime", "", policy.Revision},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
claimed, err := repo.Claim(ctx, tc.policyID, tc.period, tc.level, tc.revision)
if err == nil {
t.Fatalf("expected a claim-store error for %s, got claimed=%v err=nil", tc.name, claimed)
}
if claimed {
t.Fatalf("an errored claim attempt must report claimed=false, got true for %s", tc.name)
}
})
}
}
// TestClaimExceeded_HappyPath_BothRowsExistAtSameRevision covers
// AC-OFFICE-COSTS-003.10/.11 on a healthy store: a successful ClaimExceeded
// call commits both the exceeded-level and alert-level companion rows at
// the same policy, period and revision.
func TestClaimExceeded_HappyPath_BothRowsExistAtSameRevision(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-exceeded-happy")
claimed, err := repo.ClaimExceeded(ctx, policy.ID, "lifetime", policy.Revision)
if err != nil {
t.Fatalf("claim exceeded: %v", err)
}
if !claimed {
t.Fatal("first exceeded claim should win")
}
for _, level := range []string{"exceeded", "alert"} {
var count int
if err := db.QueryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ? AND period_key = 'lifetime' AND level = ? AND revision = ?`,
policy.ID, level, policy.Revision,
).Scan(&count); err != nil {
t.Fatalf("count %s claim: %v", level, err)
}
if count != 1 {
t.Fatalf("%s claim rows = %d, want 1", level, count)
}
}
}
// TestClaimExceeded_CompanionConflictIsPairSuccess covers the first
// verification.md "pair outcomes that are successes, not failures" case:
// with an alert-level claim already held, an over-limit evaluation's
// companion insert conflicts, but the pair still commits and the exceeded
// claim is won.
func TestClaimExceeded_CompanionConflictIsPairSuccess(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-companion-conflict")
if claimed, err := repo.Claim(ctx, policy.ID, "lifetime", "alert", policy.Revision); err != nil || !claimed {
t.Fatalf("seed alert claim: claimed=%v err=%v", claimed, err)
}
claimed, err := repo.ClaimExceeded(ctx, policy.ID, "lifetime", policy.Revision)
if err != nil {
t.Fatalf("claim exceeded: %v", err)
}
if !claimed {
t.Fatal("exceeded claim must still win despite the companion insert conflicting")
}
}
// TestClaimExceeded_AlreadyHeld_NoCompanionAttempted covers the second
// verification.md "pair outcomes" case: with an exceeded-level claim
// already held, a second over-limit evaluation's exceeded insert writes
// nothing, and no companion insert is attempted.
func TestClaimExceeded_AlreadyHeld_NoCompanionAttempted(t *testing.T) {
repo, db := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-exceeded-already-held")
if claimed, err := repo.ClaimExceeded(ctx, policy.ID, "lifetime", policy.Revision); err != nil || !claimed {
t.Fatalf("first exceeded claim: claimed=%v err=%v", claimed, err)
}
// Remove the companion so a second, incorrectly-attempted companion
// insert would be observable.
if _, err := db.Exec(
`DELETE FROM office_budget_claims WHERE policy_id = ? AND level = 'alert'`, policy.ID,
); err != nil {
t.Fatalf("remove companion: %v", err)
}
claimed, err := repo.ClaimExceeded(ctx, policy.ID, "lifetime", policy.Revision)
if err != nil {
t.Fatalf("second exceeded claim: %v", err)
}
if claimed {
t.Fatal("a second exceeded claim at an already-held (policy, period, revision) must not win")
}
var companionCount int
if err := db.QueryRow(
`SELECT COUNT(*) FROM office_budget_claims WHERE policy_id = ? AND level = 'alert'`, policy.ID,
).Scan(&companionCount); err != nil {
t.Fatalf("count companion rows: %v", err)
}
if companionCount != 0 {
t.Fatal("no companion insert must be attempted when the exceeded insert itself writes no row")
}
}
apps/backend/internal/office/repository/sqlite/budgets.go
+219 -19
View File
@@ -13,7 +13,11 @@ import (
"github.com/kandev/kandev/internal/office/models"
)
// CreateBudgetPolicy creates a new budget policy.
// CreateBudgetPolicy creates a new budget policy. Its revision is always
// assigned 1, regardless of any value already set on policy — including a
// nonzero one presented for creation. The assigned value is written back
// onto policy so the caller (the create response) reports what the row
// actually holds rather than leaning on the column's DEFAULT 1.
func (r *Repository) CreateBudgetPolicy(ctx context.Context, policy *models.BudgetPolicy) error {
if policy.ID == "" {
policy.ID = uuid.New().String()
@@ -21,15 +25,16 @@ func (r *Repository) CreateBudgetPolicy(ctx context.Context, policy *models.Budg
now := time.Now().UTC()
policy.CreatedAt = now
policy.UpdatedAt = now
policy.Revision = 1
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
INSERT INTO office_budget_policies (
id, workspace_id, scope_type, scope_id, limit_subcents, period,
alert_threshold_pct, action_on_exceed, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
alert_threshold_pct, action_on_exceed, created_at, updated_at, revision
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`), policy.ID, policy.WorkspaceID, policy.ScopeType, policy.ScopeID,
policy.LimitSubcents, policy.Period, policy.AlertThresholdPct,
policy.ActionOnExceed, policy.CreatedAt, policy.UpdatedAt)
policy.ActionOnExceed, policy.CreatedAt, policy.UpdatedAt, policy.Revision)
return err
}
@@ -78,6 +83,15 @@ func (r *Repository) UpdateBudgetPolicy(ctx context.Context, policy *models.Budg
return tx.Commit()
}
// updateBudgetPolicyTx discards the policy's claims, bumps its revision by
// exactly 1 in-SQL, and reads the new revision back onto policy, all in the
// caller's transaction. The bump is expressed as revision = revision + 1
// rather than a value computed from a prior read, so two concurrent updates
// can never write the same revision. If no row matches id — the policy was
// deleted between the caller's read and this write — the RETURNING
// read-back yields sql.ErrNoRows, which propagates as an error so the
// caller rolls the whole transaction back rather than committing an empty
// update.
func (r *Repository) updateBudgetPolicyTx(ctx context.Context, tx *sqlx.Tx, policy *models.BudgetPolicy) error {
if _, err := tx.ExecContext(ctx, tx.Rebind(
`DELETE FROM office_budget_claims WHERE policy_id = ?`), policy.ID); err != nil {
@@ -86,14 +100,20 @@ func (r *Repository) updateBudgetPolicyTx(ctx context.Context, tx *sqlx.Tx, poli
if r.failBudgetPolicyUpdateErr != nil {
return r.failBudgetPolicyUpdateErr
}
_, err := tx.ExecContext(ctx, tx.Rebind(`
err := tx.QueryRowContext(ctx, tx.Rebind(`
UPDATE office_budget_policies SET
scope_type = ?, scope_id = ?, limit_subcents = ?, period = ?,
alert_threshold_pct = ?, action_on_exceed = ?, updated_at = ?
alert_threshold_pct = ?, action_on_exceed = ?, updated_at = ?,
revision = revision + 1
WHERE id = ?
RETURNING revision
`), policy.ScopeType, policy.ScopeID, policy.LimitSubcents, policy.Period,
policy.AlertThresholdPct, policy.ActionOnExceed, policy.UpdatedAt, policy.ID)
policy.AlertThresholdPct, policy.ActionOnExceed, policy.UpdatedAt, policy.ID,
).Scan(&policy.Revision)
if err != nil {
if err == sql.ErrNoRows {
return fmt.Errorf("update policy: not found: %s", policy.ID)
}
return fmt.Errorf("update policy: %w", err)
}
return nil
@@ -107,27 +127,207 @@ func (r *Repository) DeleteBudgetPolicy(ctx context.Context, id string) error {
return err
}
// Claim atomically records that (policyID, periodKey, level) may emit its
// budget notification. claimed=true means this call won and the caller
// should emit; claimed=false with a nil error means an earlier evaluation
// already holds the claim, or the referenced policy no longer exists — a
// foreign-key violation is deliberately reported as an ordinary miss, not a
// store failure.
func (r *Repository) Claim(ctx context.Context, policyID, periodKey, level string) (bool, error) {
res, err := r.db.ExecContext(ctx, r.db.Rebind(`
INSERT INTO office_budget_claims (policy_id, period_key, level, claimed_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(policy_id, period_key, level) DO NOTHING
`), policyID, periodKey, level, time.Now().UTC())
// budgetClaimLevelExceeded and budgetClaimLevelAlert are the claim-level
// literals ClaimExceeded writes for its atomic pair. Level is otherwise
// caller-supplied data, not a package concept, but ClaimExceeded's exported
// signature does not take a level parameter, so its two writes need names
// of their own.
const (
budgetClaimLevelExceeded = "exceeded"
budgetClaimLevelAlert = "alert"
)
// execRebinder is the subset of *sqlx.DB / *sqlx.Tx a fenced claim insert
// needs, so the same statement can run standalone or inside ClaimExceeded's
// transaction.
type execRebinder interface {
ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
Rebind(string) string
}
// validateClaimInput reports whether a claim attempt's primary-key
// components are all well-formed. Every failure here is a programming
// error, not a superseded snapshot, so the caller treats it as a
// claim-store error and fails open rather than as the silent miss a
// refused fence produces.
func validateClaimInput(policyID, periodKey, level string, revision int64) error {
if revision <= 0 {
return fmt.Errorf("claim: revision must be positive, got %d", revision)
}
if policyID == "" {
return fmt.Errorf("claim: policy id must not be empty")
}
if periodKey == "" {
return fmt.Errorf("claim: period key must not be empty")
}
if level == "" {
return fmt.Errorf("claim: level must not be empty")
}
return nil
}
// claimInsertFenced inserts a claim row only if policyID still holds
// exactly revision at the moment the statement runs, checked as part of the
// same statement so no evaluation can observe a matching revision and then
// insert against a superseded one. A policy that no longer exists at all
// fails the same EXISTS check, which is how a deleted policy's claim
// attempt is refused without ever reaching the foreign-key constraint. The
// FK violation classifier is kept as a defensive fallback only: dropping it
// would let a genuine foreign-key error read as an ordinary miss.
func claimInsertFenced(ctx context.Context, ex execRebinder, policyID, periodKey, level string, revision int64) (bool, error) {
res, err := ex.ExecContext(ctx, ex.Rebind(`
INSERT INTO office_budget_claims (policy_id, period_key, level, revision, claimed_at)
SELECT ?, ?, ?, ?, ?
WHERE EXISTS (SELECT 1 FROM office_budget_policies WHERE id = ? AND revision = ?)
ON CONFLICT(policy_id, period_key, level, revision) DO NOTHING
`), policyID, periodKey, level, revision, time.Now().UTC(), policyID, revision)
if err != nil {
if db.IsForeignKeyViolation(err) {
return false, nil
}
return false, err
}
return rowsAffectedOne(res)
}
// claimInsertCompanion inserts the alert-level companion row for an
// already-won exceeded claim, keyed to the same revision with no fence of
// its own — its key already names the revision, so a row written against a
// revision that has since moved is inert rather than wrong. Any error,
// including a foreign-key violation from a policy deleted mid-transaction,
// is returned unmodified so ClaimExceeded rolls the whole pair back rather
// than committing a lone exceeded claim.
func claimInsertCompanion(ctx context.Context, tx *sqlx.Tx, policyID, periodKey string, revision int64) (bool, error) {
res, err := tx.ExecContext(ctx, tx.Rebind(`
INSERT INTO office_budget_claims (policy_id, period_key, level, revision, claimed_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(policy_id, period_key, level, revision) DO NOTHING
`), policyID, periodKey, budgetClaimLevelAlert, revision, time.Now().UTC())
if err != nil {
return false, err
}
return rowsAffectedOne(res)
}
func rowsAffectedOne(res sql.Result) (bool, error) {
affected, err := res.RowsAffected()
if err != nil {
return false, err
}
return affected == 1, nil
}
// Claim atomically records that (policyID, periodKey, level) may emit its
// budget notification, fenced to revision: the claim is recorded only while
// revision is still the policy's stored revision. claimed=true means this
// call won and the caller should emit; claimed=false with a nil error means
// an earlier evaluation already holds the claim, the referenced policy no
// longer exists, or the presented revision has been superseded — all three
// are the same ordinary miss.
func (r *Repository) Claim(ctx context.Context, policyID, periodKey, level string, revision int64) (bool, error) {
if err := validateClaimInput(policyID, periodKey, level, revision); err != nil {
return false, err
}
return claimInsertFenced(ctx, r.db, policyID, periodKey, level, revision)
}
// ClaimExceeded atomically records the exceeded-level claim and, only when
// that claim is won by this call, its alert-level companion claim, in one
// transaction: once it commits, either both rows exist for that policy,
// period and revision, or this call wrote neither. The fence is evaluated
// once, on the exceeded insert; the companion insert carries no fence of
// its own. Returns whether the exceeded-level row was inserted by this
// call — the companion's own outcome (won, or an ordinary conflict with an
// alert-level claim an earlier evaluation already holds) never changes it.
func (r *Repository) ClaimExceeded(ctx context.Context, policyID, periodKey string, revision int64) (bool, error) {
if err := validateClaimInput(policyID, periodKey, budgetClaimLevelExceeded, revision); err != nil {
return false, err
}
tx, err := r.db.BeginTxx(ctx, nil)
if err != nil {
return false, err
}
claimed, err := r.claimExceededTx(ctx, tx, policyID, periodKey, revision)
if err != nil {
if rollbackErr := tx.Rollback(); rollbackErr != nil {
return false, fmt.Errorf("%w; rollback: %v", err, rollbackErr)
}
return false, err
}
if !claimed {
return false, tx.Commit()
}
if err := tx.Commit(); err != nil {
return false, err
}
return true, nil
}
func (r *Repository) claimExceededTx(
ctx context.Context, tx *sqlx.Tx, policyID, periodKey string, revision int64,
) (bool, error) {
claimed, err := claimInsertFenced(ctx, tx, policyID, periodKey, budgetClaimLevelExceeded, revision)
if err != nil {
return false, err
}
if !claimed {
// Refused by the fence, or conflicting with an exceeded claim
// already held: no emission to protect, so the companion insert
// must not be attempted.
return false, nil
}
if r.failBudgetExceededCompanionErr != nil {
return false, r.failBudgetExceededCompanionErr
}
if _, err := claimInsertCompanion(ctx, tx, policyID, periodKey, revision); err != nil {
return false, err
}
return true, nil
}
// recreateBudgetClaimsForRevision takes effect at most once per database:
// it probes whether office_budget_claims already declares revision and does
// nothing when it does, which is every boot after the first on a fresh
// database (createCostTables already declares the final shape there) and
// every boot after the first successful recreate on an upgraded one. On a
// database still carrying the pre-revision three-column table, it drops and
// recreates office_budget_claims with the four-column primary key the
// current schema requires, as one transaction: DROP and CREATE either both
// commit or neither does, so a boot interrupted between them leaves the
// table exactly as it was rather than absent, and a retried boot always
// finds the table in a state the probe can classify. It is not expressed as
// an r.migrate.Apply statement: that runner re-executes its SQL on every
// boot and is idempotent only because an already-exists error is
// swallowed, which would destroy claim state on every boot here. It
// returns its error to its caller rather than swallowing it, so a failed
// recreate is visible rather than silently leaving the old three-column
// key in place.
func (r *Repository) recreateBudgetClaimsForRevision() error {
exists, err := db.ColumnExists(r.db, "office_budget_claims", "revision")
if err != nil {
return fmt.Errorf("probe office_budget_claims.revision: %w", err)
}
if exists {
return nil
}
if r.failBudgetClaimsRecreateErr != nil {
return r.failBudgetClaimsRecreateErr
}
tx, err := r.db.Beginx()
if err != nil {
return fmt.Errorf("begin office_budget_claims recreate: %w", err)
}
if _, err := tx.Exec(`DROP TABLE IF EXISTS office_budget_claims`); err != nil {
_ = tx.Rollback()
return fmt.Errorf("drop office_budget_claims: %w", err)
}
if r.failBudgetClaimsRecreateAfterDropErr != nil {
_ = tx.Rollback()
return r.failBudgetClaimsRecreateAfterDropErr
}
if _, err := tx.Exec(budgetClaimsDDL); err != nil {
_ = tx.Rollback()
return fmt.Errorf("recreate office_budget_claims: %w", err)
}
return tx.Commit()
}
apps/backend/internal/office/repository/sqlite/budgets_test.go
+24 -12
View File
@@ -37,10 +37,16 @@ func newBudgetClaimsRepoWithFK(t *testing.T) (*sqlite.Repository, *sqlx.DB) {
}
// office_task_tree_holds and friends carry an FK to the task-package
// owned "tasks" table; with foreign_keys=ON that table must exist (even
// empty) for a workspace-wide DELETE to validate the constraint.
// empty) for a workspace-wide DELETE to validate the constraint. title/
// description/identifier are included so migrateTaskFTS's backfill (which
// selects them unconditionally once a "tasks" table exists) doesn't fail
// schema init on this minimal fixture.
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL DEFAULT ''
workspace_id TEXT NOT NULL DEFAULT '',
title TEXT DEFAULT '',
description TEXT DEFAULT '',
identifier TEXT DEFAULT ''
)`); err != nil {
t.Fatalf("create tasks: %v", err)
}
@@ -84,7 +90,7 @@ func TestClaim_FirstCallWinsSecondCallDoesNotReclaim(t *testing.T) {
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-claim")
claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision)
if err != nil {
t.Fatalf("first claim: %v", err)
}
@@ -92,7 +98,7 @@ func TestClaim_FirstCallWinsSecondCallDoesNotReclaim(t *testing.T) {
t.Fatal("first claim should win")
}
claimed, err = repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert")
claimed, err = repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision)
if err != nil {
t.Fatalf("second claim: %v", err)
}
@@ -106,15 +112,15 @@ func TestClaim_DifferentPeriodOrLevelIsIndependent(t *testing.T) {
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-claim-2")
if claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert"); err != nil || !claimed {
if claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision); err != nil || !claimed {
t.Fatalf("claim september alert: claimed=%v err=%v", claimed, err)
}
// A different period (new window) is unclaimed (AC-OFFICE-COSTS-002.7).
if claimed, err := repo.Claim(ctx, policy.ID, "2026-10-01T00:00:00Z", "alert"); err != nil || !claimed {
if claimed, err := repo.Claim(ctx, policy.ID, "2026-10-01T00:00:00Z", "alert", policy.Revision); err != nil || !claimed {
t.Fatalf("claim october alert: claimed=%v err=%v", claimed, err)
}
// A different level in the same period is unclaimed.
if claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "exceeded"); err != nil || !claimed {
if claimed, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "exceeded", policy.Revision); err != nil || !claimed {
t.Fatalf("claim september exceeded: claimed=%v err=%v", claimed, err)
}
}
@@ -123,7 +129,7 @@ func TestClaim_ForeignKeyViolationReturnsFalseWithNoError(t *testing.T) {
repo, _ := newBudgetClaimsRepoWithFK(t)
ctx := context.Background()
claimed, err := repo.Claim(ctx, uuid.NewString(), "lifetime", "alert")
claimed, err := repo.Claim(ctx, uuid.NewString(), "lifetime", "alert", 1)
if err != nil {
t.Fatalf("claim against a nonexistent policy must not be a store error, got: %v", err)
}
@@ -137,7 +143,7 @@ func TestDeleteBudgetPolicy_CascadesClaims(t *testing.T) {
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-cascade-1")
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert"); err != nil {
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert", policy.Revision); err != nil {
t.Fatalf("claim: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 1 {
@@ -157,7 +163,7 @@ func TestDeleteWorkspaceData_CascadesBudgetClaims(t *testing.T) {
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-cascade-2")
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert"); err != nil {
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert", policy.Revision); err != nil {
t.Fatalf("claim: %v", err)
}
@@ -174,7 +180,7 @@ func TestDeleteBudgetPoliciesForRemovedScopes_CascadesClaims(t *testing.T) {
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-cascade-3")
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert"); err != nil {
if _, err := repo.Claim(ctx, policy.ID, "lifetime", "alert", policy.Revision); err != nil {
t.Fatalf("claim: %v", err)
}
@@ -193,7 +199,7 @@ func TestUpdateBudgetPolicy_DiscardsClaims(t *testing.T) {
ctx := context.Background()
policy := createTestBudgetPolicy(t, repo, "ws-update-1")
if _, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert"); err != nil {
if _, err := repo.Claim(ctx, policy.ID, "2026-09-01T00:00:00Z", "alert", policy.Revision); err != nil {
t.Fatalf("claim: %v", err)
}
if got := countBudgetClaims(t, db, policy.ID); got != 1 {
@@ -204,6 +210,9 @@ func TestUpdateBudgetPolicy_DiscardsClaims(t *testing.T) {
if err := repo.UpdateBudgetPolicy(ctx, policy); err != nil {
t.Fatalf("update budget policy: %v", err)
}
if policy.Revision != 2 {
t.Fatalf("revision returned to caller = %d, want 2", policy.Revision)
}
if got := countBudgetClaims(t, db, policy.ID); got != 0 {
t.Fatalf("claim count after update = %d, want 0 (AC-OFFICE-COSTS-002.8)", got)
}
@@ -215,6 +224,9 @@ func TestUpdateBudgetPolicy_DiscardsClaims(t *testing.T) {
if updated.LimitSubcents != 2000 {
t.Fatalf("limit_subcents = %d, want 2000", updated.LimitSubcents)
}
if updated.Revision != 2 {
t.Fatalf("stored revision = %d, want 2", updated.Revision)
}
}
func TestListBudgetPolicies_OrdersByCreatedAtThenID(t *testing.T) {
apps/backend/internal/office/repository/sqlite/loop_liveness_activation.go
+2 -1
View File
@@ -5,6 +5,7 @@ import (
"go.uber.org/zap"
"github.com/kandev/kandev/internal/db"
"github.com/kandev/kandev/internal/persistence"
)
@@ -37,7 +38,7 @@ func (r *Repository) activateLoopLiveness() {
{"agent_wakeup_requests", causationIDColumn},
{"runs", causationIDColumn},
} {
exists, err := columnExists(r.db, probe.table, probe.column)
exists, err := db.ColumnExists(r.db, probe.table, probe.column)
if err != nil {
r.logActivationWarn("loop liveness schema probe failed", err)
return
apps/backend/internal/office/repository/sqlite/run_outcome_activation.go
+2 -45
View File
@@ -1,13 +1,11 @@
package sqlite
import (
"database/sql"
"fmt"
"time"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/db/dialect"
"github.com/kandev/kandev/internal/db"
"github.com/kandev/kandev/internal/persistence"
)
@@ -35,7 +33,7 @@ func (r *Repository) activateRunOutcome() {
r.logActivationWarn("ensure kandev_meta table failed", err)
return
}
exists, err := columnExists(r.db, "runs", "outcome")
exists, err := db.ColumnExists(r.db, "runs", "outcome")
if err != nil {
r.logActivationWarn("runs.outcome schema probe failed", err)
return
@@ -55,44 +53,3 @@ func (r *Repository) logActivationWarn(msg string, err error) {
r.log.Warn(msg, zap.Error(err))
}
}
// columnExists reports whether table declares column, on either dialect.
// SQLite has no information_schema, so it falls back to PRAGMA table_info;
// Postgres is queried through the standard catalog view scoped to the
// current schema so callers never see another schema's same-named table.
func columnExists(db interface {
DriverName() string
Query(string, ...interface{}) (*sql.Rows, error)
QueryRow(string, ...interface{}) *sql.Row
Rebind(string) string
}, table, column string) (bool, error) {
if dialect.IsPostgres(db.DriverName()) {
var exists bool
err := db.QueryRow(db.Rebind(`
SELECT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_schema = current_schema()
AND table_name = ?
AND column_name = ?
)
`), table, column).Scan(&exists)
return exists, err
}
rows, err := db.Query(fmt.Sprintf("PRAGMA table_info(%s)", table))
if err != nil {
return false, err
}
defer func() { _ = rows.Close() }()
for rows.Next() {
var cid, notnull, pk int
var name, ctype string
var dflt sql.NullString
if err := rows.Scan(&cid, &name, &ctype, &notnull, &dflt, &pk); err != nil {
return false, err
}
if name == column {
return true, nil
}
}
return false, rows.Err()
}
apps/backend/internal/persistence/storeconformance/testdata/upgrades/v0.93.0/manifest.json
Vendored
+2 -2
View File
@@ -5,7 +5,7 @@
{
"engine": "sqlite3",
"file": "sqlite.sql",
"sha256": "e1b490607af347314fce11bb3352bf91599eab4c651a1521a60142e73a3b2d5d",
"sha256": "b7c763f6c3f1f6000e8a8f90a123ee3b114582bdd3a0aabf162eedae18cf57a3",
"sentinels": {
"fixture.sentinel": "v0.93.0-sqlite-sentinel"
},
@@ -37,7 +37,7 @@
{
"engine": "pgx",
"file": "postgres.sql",
"sha256": "f4f26d8d67610ef94a2d5cc308392be7a353efdd3e36e8469fb5ed11d774d165",
"sha256": "8dd108b73d1daaf9cd35ee2255f06f2800f90a7101ba3337c9ee4b0951a51de6",
"sentinels": {
"fixture.sentinel": "v0.93.0-postgres-sentinel"
},
apps/backend/internal/persistence/storeconformance/testdata/upgrades/v0.93.0/postgres.sql
Vendored
+12 -2
View File
@@ -455,8 +455,6 @@ CREATE TABLE office_budget_policies (
updated_at timestamp without time zone NOT NULL
);
CREATE TABLE office_channels (
id text NOT NULL,
workspace_id text NOT NULL,
@@ -2010,6 +2008,18 @@ ALTER TABLE ONLY office_budget_policies
CREATE TABLE office_budget_claims (
policy_id text NOT NULL,
period_key text NOT NULL,
level text NOT NULL,
claimed_at timestamp without time zone NOT NULL,
CONSTRAINT office_budget_claims_pkey PRIMARY KEY (policy_id, period_key, level),
CONSTRAINT office_budget_claims_policy_id_fkey FOREIGN KEY (policy_id)
REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
ALTER TABLE ONLY office_channels
ADD CONSTRAINT office_channels_pkey PRIMARY KEY (id);
apps/backend/internal/persistence/storeconformance/testdata/upgrades/v0.93.0/sqlite.sql
Vendored
+8
View File
@@ -367,6 +367,14 @@ CREATE TABLE office_budget_policies (
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
CREATE TABLE office_budget_claims (
policy_id TEXT NOT NULL,
period_key TEXT NOT NULL,
level TEXT NOT NULL,
claimed_at TIMESTAMP NOT NULL,
PRIMARY KEY (policy_id, period_key, level),
FOREIGN KEY (policy_id) REFERENCES office_budget_policies(id) ON DELETE CASCADE
);
CREATE TABLE office_channels (
id TEXT PRIMARY KEY,
workspace_id TEXT NOT NULL,