package meta

import (
	"context"
	"encoding/json"
	"errors"
	"testing"
	"time"
)

func TestBillingStorageSnapshotRoundTrip(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()

	base := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)
	snaps := []BillingStorageSnapshot{
		{TenantID: "t1", OrgID: "org-1", DBBytes: 100, S3Bytes: 10, CapturedAt: base},
		{TenantID: "t1", OrgID: "org-1", DBBytes: 200, S3Bytes: 20, CapturedAt: base.Add(5 * time.Minute)},
		{TenantID: "t2", OrgID: "org-2", DBBytes: 7, S3Bytes: 0, CapturedAt: base.Add(time.Minute)},
	}
	if err := s.InsertBillingStorageSnapshots(ctx, snaps); err != nil {
		t.Fatalf("insert snapshots: %v", err)
	}
	if err := s.InsertBillingStorageSnapshots(ctx, []BillingStorageSnapshot{{TenantID: "t1", OrgID: ""}}); err == nil {
		t.Fatal("insert with empty org id should fail")
	}

	left, err := s.LatestBillingStorageSnapshotAtOrBefore(ctx, "t1", base.Add(2*time.Minute))
	if err != nil {
		t.Fatalf("latest at or before: %v", err)
	}
	if left.DBBytes != 100 || left.S3Bytes != 10 {
		t.Fatalf("left boundary = %+v, want 100/10", left)
	}
	left, err = s.LatestBillingStorageSnapshotAtOrBefore(ctx, "t1", base.Add(time.Hour))
	if err != nil {
		t.Fatalf("latest at or before hour end: %v", err)
	}
	if left.DBBytes != 200 {
		t.Fatalf("left boundary = %+v, want db=200", left)
	}
	if _, err := s.LatestBillingStorageSnapshotAtOrBefore(ctx, "t1", base.Add(-time.Minute)); !errors.Is(err, ErrNotFound) {
		t.Fatalf("expected ErrNotFound, got %v", err)
	}

	tenants, err := s.ListBillingSnapshotTenants(ctx, base.Add(time.Hour))
	if err != nil {
		t.Fatalf("list snapshot tenants: %v", err)
	}
	if len(tenants) != 2 || tenants[0].TenantID != "t1" || tenants[1].TenantID != "t2" || tenants[1].OrgID != "org-2" {
		t.Fatalf("snapshot tenants = %+v", tenants)
	}
	// Capacity comes from each tenant's newest snapshot at/before the cutoff.
	if tenants[0].DBBytes != 200 || tenants[0].S3Bytes != 20 {
		t.Fatalf("t1 snapshot = %+v, want 200/20", tenants[0])
	}
	tenants, err = s.ListBillingSnapshotTenants(ctx, base.Add(30*time.Second))
	if err != nil {
		t.Fatalf("list snapshot tenants before: %v", err)
	}
	if len(tenants) != 1 || tenants[0].TenantID != "t1" {
		t.Fatalf("snapshot tenants before = %+v, want [t1]", tenants)
	}
}

func TestBillingUsageLedgerUpsertIdempotent(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	start := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)

	row := BillingUsageLedgerRow{
		CostID:      "cost-1",
		TenantID:    "t1",
		OrgID:       "42",
		Unit:        BillingUnitReadOps,
		PeriodStart: start,
		Quantity:    100,
		Labels:      map[string]string{"service": "drive9", "traffic_path": "db9"},
		Status:      BillingLedgerStatusComplete,
	}
	outboxRow := BillingOutboxRow{CostID: "cost-1", Payload: []byte(`{"cost":100}`)}
	for i := 0; i < 2; i++ {
		if err := s.WriteBillingUsageWithOutbox(ctx, []BillingUsageLedgerRow{row}, []BillingOutboxRow{outboxRow}); err != nil {
			t.Fatalf("write %d: %v", i, err)
		}
	}

	var quantity int64
	var status string
	if err := s.DB().QueryRowContext(ctx, `SELECT quantity, status FROM billing_usage_ledger
		WHERE org_id = '42' AND unit = ? AND period_start = ?`, BillingUnitReadOps, start).Scan(&quantity, &status); err != nil {
		t.Fatalf("read ledger: %v", err)
	}
	if quantity != 100 || status != BillingLedgerStatusComplete {
		t.Fatalf("ledger = %d/%s", quantity, status)
	}
	var outboxCount int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_sqs_outbox WHERE cost_id = 'cost-1'`).Scan(&outboxCount); err != nil {
		t.Fatalf("count outbox: %v", err)
	}
	if outboxCount != 1 {
		t.Fatalf("outbox rows = %d, want 1 (INSERT IGNORE must dedupe)", outboxCount)
	}

	// Ledger upsert updates quantity on re-collect; outbox row is untouched.
	row.Quantity = 250
	if err := s.WriteBillingUsageWithOutbox(ctx, []BillingUsageLedgerRow{row}, []BillingOutboxRow{outboxRow}); err != nil {
		t.Fatalf("rewrite: %v", err)
	}
	if err := s.DB().QueryRowContext(ctx, `SELECT quantity FROM billing_usage_ledger WHERE cost_id = 'cost-1'`).Scan(&quantity); err != nil {
		t.Fatal(err)
	}
	if quantity != 250 {
		t.Fatalf("quantity = %d, want 250", quantity)
	}
}

func TestBillingUsageLedgerMultiDimensionCoexist(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	start := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)

	// Same org/unit/hour split by traffic_path must coexist as two rows:
	// the ops collector emits one row per traffic_path dimension.
	rows := []BillingUsageLedgerRow{
		{
			CostID:      "cost-db9",
			TenantID:    "t1",
			OrgID:       "42",
			Unit:        BillingUnitReadOps,
			PeriodStart: start,
			Quantity:    100,
			Labels:      map[string]string{"traffic_path": "db9"},
			Status:      BillingLedgerStatusComplete,
		},
		{
			CostID:      "cost-s3",
			TenantID:    "t1",
			OrgID:       "42",
			Unit:        BillingUnitReadOps,
			PeriodStart: start,
			Quantity:    7,
			Labels:      map[string]string{"traffic_path": "s3"},
			Status:      BillingLedgerStatusComplete,
		},
	}
	if err := s.WriteBillingUsageWithOutbox(ctx, rows, nil); err != nil {
		t.Fatalf("write multi-dimension rows: %v", err)
	}
	var count int
	var total int64
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*), COALESCE(SUM(quantity), 0) FROM billing_usage_ledger
		WHERE org_id = '42' AND unit = ? AND period_start = ?`, BillingUnitReadOps, start).Scan(&count, &total); err != nil {
		t.Fatalf("read ledger: %v", err)
	}
	if count != 2 || total != 107 {
		t.Fatalf("ledger rows = %d total = %d, want 2 rows summing to 107", count, total)
	}
}

func TestBillingOutboxClaimSendAckLifecycle(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	row := BillingOutboxRow{CostID: "cost-claim", Payload: []byte(`{"cost":1}`)}
	if err := s.WriteBillingUsageWithOutbox(ctx, nil, []BillingOutboxRow{row}); err != nil {
		t.Fatalf("write outbox: %v", err)
	}

	claimed, err := s.ClaimBillingOutboxBatch(ctx, "owner-a", time.Minute, 10)
	if err != nil {
		t.Fatalf("claim: %v", err)
	}
	if len(claimed) != 1 || claimed[0].CostID != "cost-claim" {
		t.Fatalf("claimed = %+v", claimed)
	}
	// The JSON column normalizes payload formatting; compare semantically.
	var payload map[string]any
	if err := json.Unmarshal(claimed[0].Payload, &payload); err != nil {
		t.Fatalf("payload not JSON: %v", err)
	}
	if payload["cost"] != float64(1) {
		t.Fatalf("payload = %s", claimed[0].Payload)
	}
	// Second claim by another owner must not see the leased row.
	again, err := s.ClaimBillingOutboxBatch(ctx, "owner-b", time.Minute, 10)
	if err != nil {
		t.Fatalf("re-claim: %v", err)
	}
	if len(again) != 0 {
		t.Fatalf("owner-b claimed %+v while leased", again)
	}

	// A stale owner must not be able to ack.
	if err := s.MarkBillingOutboxSent(ctx, "cost-claim", "owner-b"); err != nil {
		t.Fatalf("stale ack: %v", err)
	}
	stats, err := s.BillingOutboxStats(ctx)
	if err != nil {
		t.Fatalf("stats: %v", err)
	}
	if stats.Sending != 1 || stats.Sent != 0 {
		t.Fatalf("stats after stale ack = %+v", stats)
	}

	if err := s.MarkBillingOutboxSent(ctx, "cost-claim", "owner-a"); err != nil {
		t.Fatalf("ack: %v", err)
	}
	stats, err = s.BillingOutboxStats(ctx)
	if err != nil {
		t.Fatalf("stats: %v", err)
	}
	if stats.Sent != 1 || stats.Sending != 0 || stats.Pending != 0 {
		t.Fatalf("stats after ack = %+v", stats)
	}
}

func TestBillingOutboxFailureRetryAndPermanent(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	row := BillingOutboxRow{CostID: "cost-fail", Payload: []byte(`{}`)}
	if err := s.WriteBillingUsageWithOutbox(ctx, nil, []BillingOutboxRow{row}); err != nil {
		t.Fatalf("write outbox: %v", err)
	}
	claimed, err := s.ClaimBillingOutboxBatch(ctx, "owner-a", time.Minute, 10)
	if err != nil || len(claimed) != 1 {
		t.Fatalf("claim: %v %+v", err, claimed)
	}

	// Retryable failure returns the row to pending with a backoff, attempts
	// incremented.
	nextAttempt := time.Now().Add(time.Minute)
	if err := s.MarkBillingOutboxFailed(ctx, "cost-fail", "owner-a", nextAttempt, errors.New("timeout")); err != nil {
		t.Fatalf("mark failed: %v", err)
	}
	stats, err := s.BillingOutboxStats(ctx)
	if err != nil {
		t.Fatalf("stats: %v", err)
	}
	if stats.Pending != 1 {
		t.Fatalf("stats = %+v, want pending=1", stats)
	}
	// The backoff means the row is not claimable until next_attempt_at.
	early, err := s.ClaimBillingOutboxBatch(ctx, "owner-a", time.Minute, 10)
	if err != nil {
		t.Fatalf("early claim: %v", err)
	}
	if len(early) != 0 {
		t.Fatalf("claimed before backoff elapsed: %+v", early)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE billing_sqs_outbox SET next_attempt_at = ? WHERE cost_id = 'cost-fail'`, time.Now().Add(-time.Minute)); err != nil {
		t.Fatalf("force due: %v", err)
	}

	// Re-claim and fail permanently (zero next_attempt_at).
	claimed, err = s.ClaimBillingOutboxBatch(ctx, "owner-a", time.Minute, 10)
	if err != nil || len(claimed) != 1 {
		t.Fatalf("re-claim: %v %+v", err, claimed)
	}
	if claimed[0].Attempts != 1 {
		t.Fatalf("attempts = %d, want 1", claimed[0].Attempts)
	}
	if err := s.MarkBillingOutboxFailed(ctx, "cost-fail", "owner-a", time.Time{}, errors.New("bad message")); err != nil {
		t.Fatalf("mark permanent: %v", err)
	}
	stats, err = s.BillingOutboxStats(ctx)
	if err != nil {
		t.Fatalf("stats: %v", err)
	}
	if stats.PermanentFailed != 1 || stats.Pending != 0 {
		t.Fatalf("stats = %+v, want permanent_failed=1", stats)
	}
	// Permanently failed rows are never claimed again.
	claimed, err = s.ClaimBillingOutboxBatch(ctx, "owner-a", time.Minute, 10)
	if err != nil {
		t.Fatalf("claim after permanent: %v", err)
	}
	if len(claimed) != 0 {
		t.Fatalf("claimed permanently failed row: %+v", claimed)
	}
}

func TestBillingCollectCheckpoint(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()

	if _, ok, err := s.BillingCollectCheckpoint(ctx); err != nil || ok {
		t.Fatalf("initial checkpoint ok=%v err=%v, want false/nil", ok, err)
	}
	hour := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)
	if err := s.AdvanceBillingCollectCheckpoint(ctx, hour); err != nil {
		t.Fatalf("advance: %v", err)
	}
	last, ok, err := s.BillingCollectCheckpoint(ctx)
	if err != nil || !ok {
		t.Fatalf("checkpoint ok=%v err=%v", ok, err)
	}
	if !last.Equal(hour) {
		t.Fatalf("checkpoint = %v, want %v", last, hour)
	}
	later := hour.Add(time.Hour)
	if err := s.AdvanceBillingCollectCheckpoint(ctx, later); err != nil {
		t.Fatalf("advance again: %v", err)
	}
	last, _, err = s.BillingCollectCheckpoint(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if !last.Equal(later) {
		t.Fatalf("checkpoint = %v, want %v", last, later)
	}
}

func TestListTenantStorageTierUsageFoldsUntieredIntoDB(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()

	// A native (non-shared) tenant with a completed tier scan.
	insertNativeTenantForOrgTest(t, s, "tenant-tiered", "42")
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_quota_usage
		(tenant_id, storage_bytes, db_storage_bytes, s3_storage_bytes) VALUES ('tenant-tiered', 300, 100, 200)`); err != nil {
		t.Fatalf("insert tiered usage: %v", err)
	}
	// A tenant whose tier scan never completed: db/s3 are NULL, the whole
	// storage_bytes folds into DB.
	insertNativeTenantForOrgTest(t, s, "tenant-untiered", "42")
	if _, err := s.DB().ExecContext(ctx, `INSERT INTO tenant_quota_usage
		(tenant_id, storage_bytes, db_storage_bytes, s3_storage_bytes) VALUES ('tenant-untiered', 50, NULL, NULL)`); err != nil {
		t.Fatalf("insert untiered usage: %v", err)
	}

	usages, err := s.ListTenantStorageTierUsage(ctx)
	if err != nil {
		t.Fatalf("list tier usage: %v", err)
	}
	if len(usages) != 2 {
		t.Fatalf("usages = %+v", usages)
	}
	byTenant := map[string]TenantStorageTierUsage{}
	for _, u := range usages {
		byTenant[u.TenantID] = u
	}
	if got := byTenant["tenant-tiered"]; got.TiDBCloudOrgID != "42" || got.DBBytes != 100 || got.S3Bytes != 200 {
		t.Fatalf("tiered usage = %+v, want org=42 db=100 s3=200", got)
	}
	// Untiered tenant: the whole storage_bytes folds into DB.
	if got := byTenant["tenant-untiered"]; got.TiDBCloudOrgID != "42" || got.DBBytes != 50 || got.S3Bytes != 0 {
		t.Fatalf("untiered usage = %+v, want org=42 db=50 s3=0", got)
	}
}

// insertNativeTenantForOrgTest inserts a minimal native tenant row plus its
// organization binding so billing queries can attribute usage.
func insertNativeTenantForOrgTest(t *testing.T, s *Store, tenantID, organizationID string) {
	t.Helper()
	if _, err := s.DB().ExecContext(context.Background(), `INSERT INTO tenants
		(id, db_host, db_port, db_user, db_password, db_name, provider, status)
		VALUES (?, 'h', 4000, 'u', '', 'd', ?, ?)`, tenantID, "tidb_cloud_native", TenantActive); err != nil {
		t.Fatalf("insert tenant: %v", err)
	}
	if err := s.UpsertTenantTiDBCloudOrgBinding(context.Background(), &TenantTiDBCloudOrgBinding{
		TenantID:       tenantID,
		OrganizationID: organizationID,
		ClusterID:      "cluster-" + tenantID,
	}); err != nil {
		t.Fatalf("insert org binding: %v", err)
	}
}

func TestListIncompleteBillingStorageHours(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	hour := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)

	rows := []BillingUsageLedgerRow{
		{CostID: "c-inc-1", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePerf, PeriodStart: hour, Status: BillingLedgerStatusIncomplete},
		{CostID: "c-inc-2", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePooled, PeriodStart: hour, Status: BillingLedgerStatusIncomplete},
		{CostID: "c-ok", TenantID: "t1", OrgID: "42", Unit: BillingUnitReadOps, PeriodStart: hour, Quantity: 5, Status: BillingLedgerStatusComplete},
		{CostID: "c-old", TenantID: "t2", OrgID: "7", Unit: BillingUnitStoragePerf, PeriodStart: hour.Add(-72 * time.Hour), Status: BillingLedgerStatusIncomplete},
	}
	if err := s.WriteBillingUsageWithOutbox(ctx, rows, nil); err != nil {
		t.Fatalf("write ledger: %v", err)
	}

	hours, err := s.ListIncompleteBillingStorageHours(ctx, hour.Add(-24*time.Hour))
	if err != nil {
		t.Fatalf("list incomplete: %v", err)
	}
	// Distinct (tenant, hour): the two rows for t1 collapse into one entry;
	// the 72h-old row is outside the retry window.
	if len(hours) != 1 || hours[0].TenantID != "t1" || hours[0].OrgID != "42" || !hours[0].PeriodStart.Equal(hour) {
		t.Fatalf("incomplete hours = %+v", hours)
	}

	// Completing the hour (upsert with status=complete) removes it from the
	// retry set.
	for i := range rows[:2] {
		rows[i].Status = BillingLedgerStatusComplete
		rows[i].Quantity = 3600
	}
	if err := s.WriteBillingUsageWithOutbox(ctx, rows[:2], nil); err != nil {
		t.Fatalf("complete ledger: %v", err)
	}
	hours, err = s.ListIncompleteBillingStorageHours(ctx, hour.Add(-24*time.Hour))
	if err != nil {
		t.Fatalf("list incomplete after completion: %v", err)
	}
	if len(hours) != 0 {
		t.Fatalf("incomplete hours after completion = %+v, want none", hours)
	}
}

func TestBillingSnapshotPruneKeepsBoundaryRow(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	base := time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)
	snaps := []BillingStorageSnapshot{
		{TenantID: "t1", OrgID: "42", DBBytes: 1, CapturedAt: base.Add(-3 * time.Hour)},
		{TenantID: "t1", OrgID: "42", DBBytes: 2, CapturedAt: base.Add(-2 * time.Hour)},
		{TenantID: "t1", OrgID: "42", DBBytes: 3, CapturedAt: base.Add(-time.Hour)},
		{TenantID: "t2", OrgID: "7", DBBytes: 9, CapturedAt: base.Add(-2 * time.Hour)},
	}
	if err := s.InsertBillingStorageSnapshots(ctx, snaps); err != nil {
		t.Fatal(err)
	}
	// Prune everything older than base-90m: t1 keeps its newest-below-cutoff
	// row (db=2), the newer one survives anyway, t2 keeps its only row.
	deleted, err := s.DeleteBillingStorageSnapshotsBefore(ctx, base.Add(-90*time.Minute))
	if err != nil {
		t.Fatal(err)
	}
	if deleted != 1 {
		t.Fatalf("deleted = %d, want 1 (t1's oldest; newest-below-cutoff is boundary-kept)", deleted)
	}
	tenants, err := s.ListBillingSnapshotTenants(ctx, base)
	if err != nil {
		t.Fatal(err)
	}
	if len(tenants) != 2 {
		t.Fatalf("tenants after prune = %+v, want both", tenants)
	}
}

func TestTerminalBillingOutboxPrune(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	rows := []BillingOutboxRow{
		{CostID: "c-sent", Payload: []byte(`{}`)},
		{CostID: "c-fail", Payload: []byte(`{}`)},
		{CostID: "c-pending", Payload: []byte(`{}`)},
	}
	if err := s.WriteBillingUsageWithOutbox(ctx, nil, rows); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE billing_sqs_outbox SET status = ? WHERE cost_id = 'c-sent'`, BillingOutboxSent); err != nil {
		t.Fatal(err)
	}
	if _, err := s.DB().ExecContext(ctx, `UPDATE billing_sqs_outbox SET status = ? WHERE cost_id = 'c-fail'`, BillingOutboxPermanentFailed); err != nil {
		t.Fatal(err)
	}
	deleted, err := s.DeleteTerminalBillingOutboxBefore(ctx, time.Now().Add(time.Minute))
	if err != nil {
		t.Fatal(err)
	}
	if deleted != 2 {
		t.Fatalf("deleted = %d, want 2 (sent + permanent_failed)", deleted)
	}
	var remaining int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_sqs_outbox`).Scan(&remaining); err != nil {
		t.Fatal(err)
	}
	if remaining != 1 {
		t.Fatalf("remaining = %d, want 1 (pending row kept)", remaining)
	}
}

func TestWriteBillingHourWithOutboxAdvancesCheckpointAtomically(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	hour := time.Date(2026, 9, 4, 8, 0, 0, 0, time.UTC)

	rows := []BillingUsageLedgerRow{{
		CostID: "cost-atomic", TenantID: "t1", OrgID: "42",
		Unit: BillingUnitReadOps, PeriodStart: hour, Quantity: 5,
		Labels: map[string]string{"traffic_path": "db9"}, Status: BillingLedgerStatusComplete,
	}}
	outbox := []BillingOutboxRow{{CostID: "cost-atomic", Payload: []byte(`{}`)}}
	if err := s.WriteBillingHourWithOutbox(ctx, rows, outbox, hour); err != nil {
		t.Fatalf("write hour: %v", err)
	}
	last, ok, err := s.BillingCollectCheckpoint(ctx)
	if err != nil || !ok || !last.Equal(hour) {
		t.Fatalf("checkpoint = %v/%v/%v, want %v", last, ok, err, hour)
	}
	var quantity int64
	if err := s.DB().QueryRowContext(ctx, `SELECT quantity FROM billing_usage_ledger WHERE cost_id = 'cost-atomic'`).Scan(&quantity); err != nil {
		t.Fatal(err)
	}
	if quantity != 5 {
		t.Fatalf("quantity = %d, want 5", quantity)
	}
	claimed, err := s.ClaimBillingOutboxBatch(ctx, "owner", time.Minute, 10)
	if err != nil || len(claimed) != 1 {
		t.Fatalf("claim after atomic close: %v %v", err, claimed)
	}
}

func TestWriteBillingHourWithOutboxAdvancesCheckpointOnEmptyHour(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	hour := time.Date(2026, 9, 4, 9, 0, 0, 0, time.UTC)

	// An empty hour (no usage rows) must still advance the checkpoint —
	// otherwise the collector would re-walk it forever.
	if err := s.WriteBillingHourWithOutbox(ctx, nil, nil, hour); err != nil {
		t.Fatalf("write empty hour: %v", err)
	}
	last, ok, err := s.BillingCollectCheckpoint(ctx)
	if err != nil || !ok || !last.Equal(hour) {
		t.Fatalf("checkpoint = %v/%v/%v, want %v", last, ok, err, hour)
	}
}

func TestPruneBillingLedgerRows(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	old := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
	recent := time.Date(2026, 9, 4, 0, 0, 0, 0, time.UTC)
	rows := []BillingUsageLedgerRow{
		{CostID: "c-zero-old", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePerf, PeriodStart: old, Quantity: 0, Status: BillingLedgerStatusComplete},
		{CostID: "c-inc-old", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePerf, PeriodStart: old, Quantity: 0, Status: BillingLedgerStatusIncomplete},
		{CostID: "c-nonzero-old", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePerf, PeriodStart: old, Quantity: 5, Status: BillingLedgerStatusComplete},
		{CostID: "c-zero-recent", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePerf, PeriodStart: recent, Quantity: 0, Status: BillingLedgerStatusComplete},
		// A recent incomplete row is still recoverable by the retry pass and
		// must survive pruning.
		{CostID: "c-inc-recent", TenantID: "t1", OrgID: "42", Unit: BillingUnitStoragePerf, PeriodStart: recent, Quantity: 0, Status: BillingLedgerStatusIncomplete},
	}
	if err := s.WriteBillingUsageWithOutbox(ctx, rows, nil); err != nil {
		t.Fatal(err)
	}
	deleted, err := s.PruneBillingLedgerRows(ctx, time.Date(2026, 9, 3, 0, 0, 0, 0, time.UTC))
	if err != nil {
		t.Fatal(err)
	}
	if deleted != 2 {
		t.Fatalf("deleted = %d, want 2 (zero-old + incomplete-old)", deleted)
	}
	var remaining int
	if err := s.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM billing_usage_ledger`).Scan(&remaining); err != nil {
		t.Fatal(err)
	}
	if remaining != 3 {
		t.Fatalf("remaining = %d, want 3 (nonzero-old + zero-recent + inc-recent kept)", remaining)
	}
	// Assert by identity, not just count: the recent recoverable incomplete
	// row must be one of the survivors.
	kept := map[string]bool{}
	rows2, err := s.DB().QueryContext(ctx, `SELECT cost_id FROM billing_usage_ledger`)
	if err != nil {
		t.Fatal(err)
	}
	defer func() { _ = rows2.Close() }()
	for rows2.Next() {
		var id string
		_ = rows2.Scan(&id)
		kept[id] = true
	}
	for _, want := range []string{"c-nonzero-old", "c-zero-recent", "c-inc-recent"} {
		if !kept[want] {
			t.Fatalf("survivor %s missing (kept=%v)", want, kept)
		}
	}
}