package billing

import (
	"context"
	"encoding/json"
	"errors"
	"sort"
	"sync"
	"testing"
	"time"

	"github.com/mem9-ai/drive9/pkg/meta"
)

// fakeMetaStore implements MetaStore in memory for collector/sender tests.
type fakeMetaStore struct {
	mu            sync.Mutex
	snapshots     []meta.BillingStorageSnapshot
	opsSums       []meta.BillingOpsSum
	opsErr        error
	ledger        []meta.BillingUsageLedgerRow
	outbox        map[string]*fakeOutboxRow
	nextAttemptAt map[string]time.Time
	checkpoint    time.Time
	hasCkpt       bool
}

type fakeOutboxRow struct {
	row      meta.BillingOutboxRow
	status   string
	owner    string
	attempts int
}

func newFakeMetaStore() *fakeMetaStore {
	return &fakeMetaStore{outbox: map[string]*fakeOutboxRow{}, nextAttemptAt: map[string]time.Time{}}
}

func (f *fakeMetaStore) ListTenantStorageTierUsage(_ context.Context) ([]meta.TenantStorageTierUsage, error) {
	return nil, nil
}

func (f *fakeMetaStore) InsertBillingStorageSnapshots(_ context.Context, snapshots []meta.BillingStorageSnapshot) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.snapshots = append(f.snapshots, snapshots...)
	return nil
}

func (f *fakeMetaStore) LatestBillingStorageSnapshotAtOrBefore(_ context.Context, tenantID string, ts time.Time) (meta.BillingStorageSnapshot, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var best meta.BillingStorageSnapshot
	for _, s := range f.snapshots {
		if s.TenantID != tenantID || s.CapturedAt.After(ts) {
			continue
		}
		if best.CapturedAt.IsZero() || s.CapturedAt.After(best.CapturedAt) {
			best = s
		}
	}
	if best.CapturedAt.IsZero() {
		return best, meta.ErrNotFound
	}
	return best, nil
}

func (f *fakeMetaStore) ListBillingSnapshotTenants(_ context.Context, before time.Time) ([]meta.BillingSnapshotTenant, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	latest := map[string]meta.BillingStorageSnapshot{}
	for _, s := range f.snapshots {
		if s.CapturedAt.After(before) {
			continue
		}
		cur, ok := latest[s.TenantID]
		if !ok || s.CapturedAt.After(cur.CapturedAt) {
			latest[s.TenantID] = s
		}
	}
	var out []meta.BillingSnapshotTenant
	for tenantID, s := range latest {
		out = append(out, meta.BillingSnapshotTenant{
			TenantID: tenantID, OrgID: s.OrgID,
			DBBytes: s.DBBytes, S3Bytes: s.S3Bytes, CapturedAt: s.CapturedAt,
		})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].TenantID < out[j].TenantID })
	return out, nil
}

func (f *fakeMetaStore) SumBillingOpsDeltas(_ context.Context, _, _ time.Time) ([]meta.BillingOpsSum, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]meta.BillingOpsSum(nil), f.opsSums...), f.opsErr
}

func (f *fakeMetaStore) DeleteBillingOpsDeltasBefore(_ context.Context, _ time.Time) (int64, error) {
	return 0, nil
}

func (f *fakeMetaStore) DeleteBillingStorageSnapshotsBefore(_ context.Context, before time.Time) (int64, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	// Mirror the boundary-preserving delete: keep each tenant's newest
	// snapshot below the cutoff.
	newestBelow := map[string]time.Time{}
	for _, s := range f.snapshots {
		if s.CapturedAt.Before(before) && s.CapturedAt.After(newestBelow[s.TenantID]) {
			newestBelow[s.TenantID] = s.CapturedAt
		}
	}
	kept := f.snapshots[:0]
	var deleted int64
	for _, s := range f.snapshots {
		if s.CapturedAt.Before(before) && !s.CapturedAt.Equal(newestBelow[s.TenantID]) {
			deleted++
			continue
		}
		kept = append(kept, s)
	}
	f.snapshots = kept
	return deleted, nil
}

func (f *fakeMetaStore) BillingCollectCheckpoint(_ context.Context) (time.Time, bool, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.checkpoint, f.hasCkpt, nil
}

func (f *fakeMetaStore) AdvanceBillingCollectCheckpoint(_ context.Context, lastHour time.Time) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.checkpoint, f.hasCkpt = lastHour, true
	return nil
}

func (f *fakeMetaStore) WriteBillingUsageWithOutbox(_ context.Context, ledger []meta.BillingUsageLedgerRow, outbox []meta.BillingOutboxRow) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	// Mirror the DB's upsert semantics: one row per cost_id.
	existing := map[string]int{}
	for i, row := range f.ledger {
		existing[row.CostID] = i
	}
	for _, row := range ledger {
		if i, ok := existing[row.CostID]; ok {
			f.ledger[i] = row
			continue
		}
		existing[row.CostID] = len(f.ledger)
		f.ledger = append(f.ledger, row)
	}
	for _, row := range outbox {
		if _, exists := f.outbox[row.CostID]; !exists {
			f.outbox[row.CostID] = &fakeOutboxRow{row: row, status: meta.BillingOutboxPending}
		}
	}
	return nil
}

func (f *fakeMetaStore) ClaimBillingOutboxBatch(_ context.Context, owner string, _ time.Duration, limit int) ([]meta.BillingOutboxRow, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []meta.BillingOutboxRow
	for _, row := range f.outbox {
		if row.status != meta.BillingOutboxPending {
			continue
		}
		row.status = meta.BillingOutboxSending
		row.owner = owner
		row.row.Attempts = row.attempts
		out = append(out, row.row)
		if len(out) >= limit {
			break
		}
	}
	return out, nil
}

func (f *fakeMetaStore) MarkBillingOutboxSent(_ context.Context, costID, owner string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	row := f.outbox[costID]
	if row == nil || row.owner != owner {
		return nil
	}
	row.status = meta.BillingOutboxSent
	row.owner = ""
	return nil
}

func (f *fakeMetaStore) MarkBillingOutboxFailed(_ context.Context, costID, owner string, nextAttemptAt time.Time, _ error) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	row := f.outbox[costID]
	if row == nil || row.owner != owner {
		return nil
	}
	row.attempts++
	row.owner = ""
	f.nextAttemptAt[costID] = nextAttemptAt
	if !nextAttemptAt.IsZero() {
		row.status = meta.BillingOutboxPending
	} else {
		row.status = meta.BillingOutboxPermanentFailed
	}
	return nil
}

func (f *fakeMetaStore) DeleteTerminalBillingOutboxBefore(_ context.Context, _ time.Time) (int64, error) {
	return 0, nil
}

func (f *fakeMetaStore) PruneBillingLedgerRows(_ context.Context, before time.Time) (int64, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	kept := f.ledger[:0]
	var deleted int64
	for _, row := range f.ledger {
		if row.PeriodStart.Before(before) && (row.Quantity == 0 || row.Status == meta.BillingLedgerStatusIncomplete) {
			deleted++
			continue
		}
		kept = append(kept, row)
	}
	f.ledger = kept
	return deleted, nil
}

func (f *fakeMetaStore) BillingOutboxStats(_ context.Context) (meta.BillingOutboxStats, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var stats meta.BillingOutboxStats
	for _, row := range f.outbox {
		switch row.status {
		case meta.BillingOutboxPending:
			stats.Pending++
		case meta.BillingOutboxSending:
			stats.Sending++
		case meta.BillingOutboxSent:
			stats.Sent++
		case meta.BillingOutboxPermanentFailed:
			stats.PermanentFailed++
		}
	}
	return stats, nil
}

func (f *fakeMetaStore) ListIncompleteBillingStorageHours(_ context.Context, since time.Time) ([]meta.BillingIncompleteHour, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	type key struct {
		tenant string
		start  time.Time
	}
	seen := map[key]bool{}
	var out []meta.BillingIncompleteHour
	for _, row := range f.ledger {
		if row.Status != meta.BillingLedgerStatusIncomplete || row.PeriodStart.Before(since) {
			continue
		}
		k := key{tenant: row.TenantID, start: row.PeriodStart}
		if !seen[k] {
			seen[k] = true
			out = append(out, meta.BillingIncompleteHour{TenantID: row.TenantID, OrgID: row.OrgID, PeriodStart: row.PeriodStart})
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].PeriodStart.Before(out[j].PeriodStart) })
	return out, nil
}

func (f *fakeMetaStore) WriteBillingHourWithOutbox(ctx context.Context, ledger []meta.BillingUsageLedgerRow, outbox []meta.BillingOutboxRow, checkpointHour time.Time) error {
	if err := f.WriteBillingUsageWithOutbox(ctx, ledger, outbox); err != nil {
		return err
	}
	return f.AdvanceBillingCollectCheckpoint(ctx, checkpointHour)
}

func (f *fakeMetaStore) outboxStatus(costID string) string {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.outbox[costID].status
}

type fakeSQS struct {
	mu   sync.Mutex
	sent []string
	err  error
}

func (f *fakeSQS) SendMessage(_ context.Context, _, body string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.err != nil {
		return f.err
	}
	f.sent = append(f.sent, body)
	return nil
}

func (f *fakeSQS) sentCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.sent)
}

func testConfig() Config {
	cfg := Config{
		Provider: "aws",
		Region:   "us-west-2",
	}
	return cfg.withDefaults()
}

var testHour = time.Date(2026, 9, 2, 10, 0, 0, 0, time.UTC)

func TestCostIDStableAndDimensionSensitive(t *testing.T) {
	a := CostID(42, meta.BillingUnitReadOps, testHour, "aws", "us-west-2", "db9")
	b := CostID(42, meta.BillingUnitReadOps, testHour, "aws", "us-west-2", "db9")
	if a != b {
		t.Fatalf("CostID not deterministic: %s != %s", a, b)
	}
	for _, diff := range []string{
		CostID(43, meta.BillingUnitReadOps, testHour, "aws", "us-west-2", "db9"),
		CostID(42, meta.BillingUnitWriteOps, testHour, "aws", "us-west-2", "db9"),
		CostID(42, meta.BillingUnitReadOps, testHour.Add(time.Hour), "aws", "us-west-2", "db9"),
		CostID(42, meta.BillingUnitReadOps, testHour, "aws", "us-west-2", "s3"),
		CostID(42, meta.BillingUnitReadOps, testHour, "aws", "us-east-1", "db9"),
	} {
		if diff == a {
			t.Fatalf("CostID collision for differing dimensions: %s", a)
		}
	}
}

func TestBuildOpsRecords(t *testing.T) {
	c := newCollector(testConfig(), newFakeMetaStore(), nil)
	sums := []meta.BillingOpsSum{
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 1500},
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "s3", Ops: 30},
		{TenantID: "fs-a", OrgID: "42", Direction: "write", TrafficPath: "s3", Ops: 7},
		{TenantID: "fs-b", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 20},
		{TenantID: "local", OrgID: "guest", Direction: "read", TrafficPath: "db9", Ops: 100},
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 0},
	}
	ledger, outbox, unmapped, err := c.buildOpsRecords(testHour, sums)
	if err != nil {
		t.Fatalf("buildOpsRecords: %v", err)
	}
	if unmapped != 1 {
		t.Fatalf("unmapped = %d, want 1", unmapped)
	}
	// Per-tenant, per-traffic-path push: fs-a read db9, fs-a read s3, fs-a
	// write s3, fs-b read db9; the zero-quantity row is skipped.
	if len(ledger) != 4 || len(outbox) != 4 {
		t.Fatalf("ledger=%d outbox=%d, want 4/4 (per-tenant per-path, zero skipped)", len(ledger), len(outbox))
	}
	if ledger[0].Quantity != 1500 || ledger[0].Unit != "fs-read-ops:aws:us-west-2" || ledger[0].TenantID != "fs-a" ||
		ledger[0].Labels["traffic_path"] != "db9" {
		t.Fatalf("ledger[0] = %+v", ledger[0])
	}
	if ledger[1].Quantity != 30 || ledger[1].Labels["traffic_path"] != "s3" ||
		ledger[1].Unit != "fs-pooled-read-ops:aws:us-west-2" {
		t.Fatalf("ledger[1] = %+v", ledger[1])
	}
	// db9 and s3 are different SKUs with different units and prices.
	if ledger[0].Unit == ledger[1].Unit {
		t.Fatalf("db9/s3 must map to different units, both %q", ledger[0].Unit)
	}
	if ledger[2].Quantity != 7 || ledger[2].Unit != "fs-pooled-write-ops:aws:us-west-2" ||
		ledger[2].Labels["resource_id"] != "fs-a" {
		t.Fatalf("ledger[2] = %+v", ledger[2])
	}
	// CostIDs differ across traffic paths for the same tenant/unit/hour.
	if ledger[0].CostID == ledger[1].CostID {
		t.Fatalf("db9/s3 CostIDs must differ: %s", ledger[0].CostID)
	}
	if ledger[3].TenantID != "fs-b" || ledger[3].Quantity != 20 {
		t.Fatalf("ledger[3] = %+v", ledger[3])
	}
	var measure BillingMeasure
	if err := json.Unmarshal(outbox[0].Payload, &measure); err != nil {
		t.Fatalf("unmarshal payload: %v", err)
	}
	if measure.TenantID != 42 || measure.Cost != 1500 || measure.CostID != ledger[0].CostID ||
		measure.Unit != "fs-read-ops:aws:us-west-2" || measure.CostTimestamp != testHour.Unix() ||
		measure.Labels["provider"] != "aws" || measure.Labels["region"] != "us-west-2" ||
		measure.Labels["resource_id"] != "fs-a" || measure.Labels["resource_type"] != "fs" ||
		measure.Labels["traffic_path"] != "db9" {
		t.Fatalf("payload = %+v", measure)
	}
}

func TestCollectDueHoursEndToEnd(t *testing.T) {
	store := newFakeMetaStore()
	// One tenant with complete snapshot coverage of testHour.
	left := meta.BillingStorageSnapshot{TenantID: "fs-a", OrgID: "42", DBBytes: 10 << 20, S3Bytes: 2 << 20, CapturedAt: testHour.Add(-time.Minute)}
	tail := meta.BillingStorageSnapshot{TenantID: "fs-a", OrgID: "42", DBBytes: 10 << 20, S3Bytes: 2 << 20, CapturedAt: testHour.Add(59 * time.Minute)}
	if err := store.InsertBillingStorageSnapshots(context.Background(), []meta.BillingStorageSnapshot{left, tail}); err != nil {
		t.Fatal(err)
	}

	store.opsSums = []meta.BillingOpsSum{
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 3000},
	}
	now := testHour.Add(time.Hour + 11*time.Minute)
	c := newCollector(testConfig(), store, func() time.Time { return now })

	// First call initializes the checkpoint without collecting.
	if err := c.CollectDueHours(context.Background()); err != nil {
		t.Fatalf("init: %v", err)
	}
	if len(store.ledger) != 0 {
		t.Fatalf("init collected %d rows, want 0", len(store.ledger))
	}
	if err := c.CollectDueHours(context.Background()); err != nil {
		t.Fatalf("collect: %v", err)
	}
	// 1 ops row + 2 storage rows (db + s3), all complete.
	if len(store.ledger) != 3 {
		t.Fatalf("ledger rows = %d, want 3: %+v", len(store.ledger), store.ledger)
	}
	var opsRow, dbRow, s3Row meta.BillingUsageLedgerRow
	for _, row := range store.ledger {
		switch row.Unit {
		case "fs-read-ops:aws:us-west-2":
			opsRow = row
		case "fs-storage-performance:aws:us-west-2":
			dbRow = row
		case "fs-storage-pooled:aws:us-west-2":
			s3Row = row
		}
	}
	if opsRow.Quantity != 3000 || opsRow.Status != meta.BillingLedgerStatusComplete {
		t.Fatalf("ops row = %+v", opsRow)
	}
	// quantities are average MiB capacity over the hour: 10 MiB (2 MiB) held for one hour
	if dbRow.Quantity != 10 || s3Row.Quantity != 2 {
		t.Fatalf("storage rows = %d/%d", dbRow.Quantity, s3Row.Quantity)
	}
	if len(store.outbox) != 3 {
		t.Fatalf("outbox rows = %d, want 3", len(store.outbox))
	}

	// Idempotent: re-closing the same hour via a manual collect must not
	// duplicate outbox rows.
	if err := c.collectHour(context.Background(), testHour); err != nil {
		t.Fatalf("re-collect: %v", err)
	}
	if len(store.outbox) != 3 {
		t.Fatalf("outbox rows after re-collect = %d, want 3", len(store.outbox))
	}

	// Checkpoint consumed the hour; no further work.
	if err := c.CollectDueHours(context.Background()); err != nil {
		t.Fatalf("collect again: %v", err)
	}
	if len(store.ledger) != 3 {
		t.Fatalf("ledger rows after re-collect = %d, want 3", len(store.ledger))
	}
}

func TestCollectDueHoursRetriesOnOpsReadError(t *testing.T) {
	store := newFakeMetaStore()
	store.hasCkpt = true
	store.checkpoint = testHour.Add(-time.Hour)
	store.opsErr = errors.New("meta down")
	c := newCollector(testConfig(), store, func() time.Time {
		return testHour.Add(time.Hour + 11*time.Minute)
	})
	if err := c.CollectDueHours(context.Background()); err == nil {
		t.Fatal("expected ops read error")
	}
	if !store.checkpoint.Equal(testHour.Add(-time.Hour)) {
		t.Fatalf("checkpoint advanced past failed hour: %v", store.checkpoint)
	}
}

func TestStorageHourWithoutSnapshotsIsIncomplete(t *testing.T) {
	store := newFakeMetaStore()
	// Tenant fs-a has snapshots, tenant fs-b has none covering the hour.
	if err := store.InsertBillingStorageSnapshots(context.Background(), []meta.BillingStorageSnapshot{
		{TenantID: "fs-a", OrgID: "42", DBBytes: 1 << 20, S3Bytes: 1 << 20, CapturedAt: testHour.Add(-time.Minute)},
		{TenantID: "fs-a", OrgID: "42", DBBytes: 1 << 20, S3Bytes: 1 << 20, CapturedAt: testHour.Add(59 * time.Minute)},
		{TenantID: "fs-b", OrgID: "7", DBBytes: 5 << 20, S3Bytes: 5 << 20, CapturedAt: testHour.Add(-48 * time.Hour)},
	}); err != nil {
		t.Fatal(err)
	}
	c := newCollector(testConfig(), store, func() time.Time {
		return testHour.Add(time.Hour + 11*time.Minute)
	})
	ledger, outbox, err := c.buildStorageRecords(context.Background(), testHour, testHour.Add(time.Hour))
	if err != nil {
		t.Fatalf("buildStorageRecords: %v", err)
	}
	// fs-a: 2 complete rows; fs-b: 2 incomplete rows (ledger-only).
	var complete, incomplete int
	for _, row := range ledger {
		switch row.Status {
		case meta.BillingLedgerStatusComplete:
			complete++
		case meta.BillingLedgerStatusIncomplete:
			incomplete++
			if row.TenantID != "fs-b" || row.Quantity != 0 {
				t.Fatalf("incomplete row = %+v", row)
			}
		}
	}
	if complete != 2 || incomplete != 2 {
		t.Fatalf("complete=%d incomplete=%d, want 2/2", complete, incomplete)
	}
	if len(outbox) != 2 {
		t.Fatalf("outbox = %d, want 2 (incomplete hours produce no message)", len(outbox))
	}
}

// TestRetryIncompleteHoursBackfillsLateCoverage simulates the recoverable gap
// shape: an hour closes while an organization's snapshots have not committed
// yet (incomplete), the snapshots land shortly after, and the bounded retry
// pass completes the hour and emits the storage outbox rows.
func TestRetryIncompleteHoursBackfillsLateCoverage(t *testing.T) {
	store := newFakeMetaStore()
	// Tenant fs-b has only an ancient snapshot: hour testHour has no usable
	// coverage.
	if err := store.InsertBillingStorageSnapshots(context.Background(), []meta.BillingStorageSnapshot{
		{TenantID: "fs-b", OrgID: "7", DBBytes: 5 << 20, S3Bytes: 5 << 20, CapturedAt: testHour.Add(-48 * time.Hour)},
	}); err != nil {
		t.Fatal(err)
	}
	now := testHour.Add(time.Hour + 11*time.Minute)
	c := newCollector(testConfig(), store, func() time.Time { return now })
	ledger, outbox, err := c.buildStorageRecords(context.Background(), testHour, testHour.Add(time.Hour))
	if err != nil {
		t.Fatal(err)
	}
	if err := store.WriteBillingUsageWithOutbox(context.Background(), ledger, outbox); err != nil {
		t.Fatal(err)
	}
	if len(store.outbox) != 0 {
		t.Fatalf("outbox = %d, want 0 for incomplete hour", len(store.outbox))
	}
	if rows := countLedgerStatus(store, meta.BillingLedgerStatusIncomplete); rows != 2 {
		t.Fatalf("incomplete rows = %d, want 2", rows)
	}

	// Snapshots captured inside the hour commit late (commit/boundary race).
	if err := store.InsertBillingStorageSnapshots(context.Background(), []meta.BillingStorageSnapshot{
		{TenantID: "fs-b", OrgID: "7", DBBytes: 5 << 20, S3Bytes: 5 << 20, CapturedAt: testHour.Add(-time.Minute)},
		{TenantID: "fs-b", OrgID: "7", DBBytes: 5 << 20, S3Bytes: 5 << 20, CapturedAt: testHour.Add(59 * time.Minute)},
	}); err != nil {
		t.Fatal(err)
	}
	if err := c.RetryIncompleteHours(context.Background()); err != nil {
		t.Fatalf("retry: %v", err)
	}
	if rows := countLedgerStatus(store, meta.BillingLedgerStatusIncomplete); rows != 0 {
		t.Fatalf("incomplete rows after retry = %d, want 0", rows)
	}
	if rows := countLedgerStatus(store, meta.BillingLedgerStatusComplete); rows != 2 {
		t.Fatalf("complete rows after retry = %d, want 2", rows)
	}
	if len(store.outbox) != 2 {
		t.Fatalf("outbox after retry = %d, want 2 storage messages", len(store.outbox))
	}
	for _, row := range store.ledger {
		// 5 MiB held for the full hour reports an average capacity of 5 MiB.
		if row.Quantity != 5 {
			t.Fatalf("row = %+v, want 5 MiB average capacity", row)
		}
	}

	// Idempotent: a second retry pass changes nothing.
	if err := c.RetryIncompleteHours(context.Background()); err != nil {
		t.Fatal(err)
	}
	if len(store.outbox) != 2 {
		t.Fatalf("outbox after second retry = %d, want 2", len(store.outbox))
	}
}

// TestRetryIncompleteHoursLeavesPermanentGaps pins that a downtime gap (no
// snapshots ever captured in the hour) stays incomplete and unbilled.
func TestRetryIncompleteHoursLeavesPermanentGaps(t *testing.T) {
	store := newFakeMetaStore()
	c := newCollector(testConfig(), store, func() time.Time {
		return testHour.Add(time.Hour + 11*time.Minute)
	})
	// Seed the incomplete rows directly to model a past close with a gap.
	tenant := meta.BillingSnapshotTenant{TenantID: "fs-b", OrgID: "7"}
	if err := store.WriteBillingUsageWithOutbox(context.Background(), c.incompleteStorageRows(tenant, testHour), nil); err != nil {
		t.Fatal(err)
	}
	if err := c.RetryIncompleteHours(context.Background()); err != nil {
		t.Fatal(err)
	}
	if rows := countLedgerStatus(store, meta.BillingLedgerStatusIncomplete); rows != 2 {
		t.Fatalf("incomplete rows = %d, want 2 (permanent gap stays incomplete)", rows)
	}
	if len(store.outbox) != 0 {
		t.Fatalf("outbox = %d, want 0 (permanent gap is never billed)", len(store.outbox))
	}
}

func countLedgerStatus(store *fakeMetaStore, status string) int {
	store.mu.Lock()
	defer store.mu.Unlock()
	n := 0
	for _, row := range store.ledger {
		if row.Status == status {
			n++
		}
	}
	return n
}

// TestPruneNeverOutrunsCheckpoint pins the blocking-review invariant: prune
// is fenced by the collection checkpoint, so a lagging collector never has
// its unprocessed hours' data deleted. Old data behind the checkpoint window
// is pruned, but each tenant's newest boundary snapshot below the cutoff is
// kept, so a catch-up close marks stale hours incomplete instead of silently
// closing them with no rows.
func TestPruneNeverOutrunsCheckpoint(t *testing.T) {
	store := newFakeMetaStore()
	// Checkpoint 50h behind; snapshots and ops deltas exist for the lagged
	// hours (servers kept flushing while the worker was down).
	old := testHour.Add(-50 * time.Hour)
	store.hasCkpt = true
	store.checkpoint = old
	if err := store.InsertBillingStorageSnapshots(context.Background(), []meta.BillingStorageSnapshot{
		{TenantID: "fs-a", OrgID: "42", DBBytes: 1 << 20, S3Bytes: 1 << 20, CapturedAt: old.Add(-25 * time.Hour)},
		{TenantID: "fs-a", OrgID: "42", DBBytes: 2 << 20, S3Bytes: 2 << 20, CapturedAt: old.Add(-26 * time.Hour)},
	}); err != nil {
		t.Fatal(err)
	}
	store.opsSums = []meta.BillingOpsSum{{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 5}}

	c := newCollector(testConfig(), store, func() time.Time { return testHour.Add(time.Hour + 11*time.Minute) })
	// Ledger rows straddling the terminal cutoff (now - 48h): rows older
	// than it are dropped; the recent incomplete row survives (inside the 2x
	// prune margin, though already past the 24h retry window).
	cutoff := testHour.Add(time.Hour + 11*time.Minute).Add(-48 * time.Hour)
	if err := store.WriteBillingUsageWithOutbox(context.Background(), []meta.BillingUsageLedgerRow{
		{CostID: "old-zero", TenantID: "fs-a", OrgID: "42", Unit: "fs-storage-performance:aws:us-west-2", PeriodStart: cutoff.Add(-time.Hour), Quantity: 0, Status: meta.BillingLedgerStatusComplete},
		{CostID: "old-inc", TenantID: "fs-a", OrgID: "42", Unit: "fs-storage-performance:aws:us-west-2", PeriodStart: cutoff.Add(-2 * time.Hour), Quantity: 0, Status: meta.BillingLedgerStatusIncomplete},
		{CostID: "recent-inc", TenantID: "fs-a", OrgID: "42", Unit: "fs-storage-performance:aws:us-west-2", PeriodStart: cutoff.Add(time.Hour), Quantity: 0, Status: meta.BillingLedgerStatusIncomplete},
	}, nil); err != nil {
		t.Fatal(err)
	}
	if err := c.PruneOnce(context.Background()); err != nil {
		t.Fatalf("prune: %v", err)
	}
	if rows := countLedgerStatus(store, meta.BillingLedgerStatusIncomplete); rows != 1 {
		t.Fatalf("incomplete rows after prune = %d, want 1 (recent-inc survives)", rows)
	}
	var oldZero bool
	for _, row := range store.ledger {
		if row.CostID == "old-zero" {
			oldZero = true
		}
	}
	if oldZero {
		t.Fatal("old-zero should have been pruned")
	}
	// The boundary snapshot below the cutoff survives; its older sibling is
	// pruned. Cutoff = checkpoint - 24h; both test snapshots are below it, so
	// the newest one below the cutoff is kept.
	if len(store.snapshots) != 1 {
		t.Fatalf("snapshots after prune = %+v, want 1 boundary row", store.snapshots)
	}
	if store.opsSums == nil {
		t.Fatal("ops deltas must not be pruned when they postdate the cutoff")
	}

	// Catch-up closes the lagged hour: boundary snapshot is stale →
	// incomplete rows, NOT a silent empty close.
	ledger, outbox, err := c.buildStorageRecords(context.Background(), old.Add(time.Hour), old.Add(2*time.Hour))
	if err != nil {
		t.Fatal(err)
	}
	if len(ledger) != 2 || len(outbox) != 0 {
		t.Fatalf("ledger=%+v outbox=%+v, want 2 incomplete rows and no outbox", ledger, outbox)
	}
	for _, row := range ledger {
		if row.Status != meta.BillingLedgerStatusIncomplete {
			t.Fatalf("row = %+v, want incomplete", row)
		}
	}
}

// TestRetryConvergesZeroSiblingSKU pins that a retry with a zero S3 tier
// marks BOTH storage SKUs complete: no outbox message for the zero tier, and
// no stale incomplete marker left behind.
func TestRetryConvergesZeroSiblingSKU(t *testing.T) {
	store := newFakeMetaStore()
	c := newCollector(testConfig(), store, func() time.Time {
		return testHour.Add(time.Hour + 11*time.Minute)
	})
	tenant := meta.BillingSnapshotTenant{TenantID: "fs-b", OrgID: "7"}
	if err := store.WriteBillingUsageWithOutbox(context.Background(), c.incompleteStorageRows(tenant, testHour), nil); err != nil {
		t.Fatal(err)
	}
	// Late coverage arrives with zero S3 bytes (untiered fold: all in DB).
	if err := store.InsertBillingStorageSnapshots(context.Background(), []meta.BillingStorageSnapshot{
		{TenantID: "fs-b", OrgID: "7", DBBytes: 5 << 20, S3Bytes: 0, CapturedAt: testHour.Add(59 * time.Minute)},
	}); err != nil {
		t.Fatal(err)
	}
	if err := c.RetryIncompleteHours(context.Background()); err != nil {
		t.Fatal(err)
	}
	if rows := countLedgerStatus(store, meta.BillingLedgerStatusIncomplete); rows != 0 {
		t.Fatalf("incomplete rows after retry = %d, want 0", rows)
	}
	var dbRow, s3Row *meta.BillingUsageLedgerRow
	for i := range store.ledger {
		row := &store.ledger[i]
		switch row.Unit {
		case "fs-storage-performance:aws:us-west-2":
			dbRow = row
		case "fs-storage-pooled:aws:us-west-2":
			s3Row = row
		}
	}
	if dbRow == nil || dbRow.Status != meta.BillingLedgerStatusComplete || dbRow.Quantity != 5 {
		t.Fatalf("db row = %+v", dbRow)
	}
	if s3Row == nil || s3Row.Status != meta.BillingLedgerStatusComplete || s3Row.Quantity != 0 {
		t.Fatalf("s3 row = %+v, want complete zero-quantity row", s3Row)
	}
	if len(store.outbox) != 1 {
		t.Fatalf("outbox = %d, want 1 (db only; zero s3 sends nothing)", len(store.outbox))
	}
}

// TestCollectSkipsUnknownOpsDirection pins that an unknown direction label
// skips that series with a metric instead of failing the whole hour.
func TestCollectSkipsUnknownOpsDirection(t *testing.T) {
	store := newFakeMetaStore()
	store.opsSums = []meta.BillingOpsSum{
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 100},
		{TenantID: "fs-a", OrgID: "42", Direction: "teleport", TrafficPath: "db9", Ops: 50},
	}
	c := newCollector(testConfig(), store, nil)
	ledger, _, _, err := c.buildOpsRecords(testHour, store.opsSums)
	if err != nil {
		t.Fatalf("unknown direction must not fail the hour: %v", err)
	}
	if len(ledger) != 1 || ledger[0].Quantity != 100 {
		t.Fatalf("ledger = %+v, want only the read row", ledger)
	}
}

// TestCollectSkipsUnknownTrafficPath pins that an unknown traffic_path never
// gets guessed into a SKU (billing is the last boundary before charging).
func TestCollectSkipsUnknownTrafficPath(t *testing.T) {
	store := newFakeMetaStore()
	store.opsSums = []meta.BillingOpsSum{
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "db9", Ops: 100},
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "s3", Ops: 10},
		{TenantID: "fs-a", OrgID: "42", Direction: "read", TrafficPath: "tproxy", Ops: 999},
	}
	c := newCollector(testConfig(), store, nil)
	ledger, _, _, err := c.buildOpsRecords(testHour, store.opsSums)
	if err != nil {
		t.Fatalf("unknown traffic_path must not fail the hour: %v", err)
	}
	if len(ledger) != 2 {
		t.Fatalf("ledger = %+v, want db9+s3 rows only", ledger)
	}
	for _, row := range ledger {
		if row.Labels["traffic_path"] == "tproxy" {
			t.Fatalf("unknown path billed: %+v", row)
		}
	}
}

// TestStorageRecordsCeilSubMiBCapacity pins the headline ceil behavior at
// the call site: sub-MiB capacities bill at least 1 MiB per SKU per hour,
// and outbox messages are only emitted for positive quantities.
func TestStorageRecordsCeilSubMiBCapacity(t *testing.T) {
	start := testHour
	fresh := start.Add(59 * time.Minute)
	for _, tc := range []struct {
		name    string
		db, s3  int64
		wantDB  int64
		wantS3  int64
		wantMsg int
	}{
		{"one byte", 1, 1, 1, 1, 2},
		{"exact one MiB", 1 << 20, 1 << 20, 1, 1, 2},
		{"one MiB plus one", (1 << 20) + 1, (1 << 20) + 1, 2, 2, 2},
		{"db only", 100, 0, 1, 0, 1},
		{"true zero", 0, 0, 0, 0, 0},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newFakeMetaStore()
			c := newCollector(testConfig(), store, nil)
			tenant := meta.BillingSnapshotTenant{TenantID: "fs-a", OrgID: "42", DBBytes: tc.db, S3Bytes: tc.s3, CapturedAt: fresh}
			ledger, outbox, err := c.buildStorageRecordsForTenant(context.Background(), tenant, start, start.Add(time.Hour))
			if err != nil {
				t.Fatal(err)
			}
			var gotDB, gotS3 = int64(-1), int64(-1)
			for _, row := range ledger {
				if row.Status != meta.BillingLedgerStatusComplete {
					t.Fatalf("row = %+v, want complete", row)
				}
				switch row.Unit {
				case "fs-storage-performance:aws:us-west-2":
					gotDB = row.Quantity
				case "fs-storage-pooled:aws:us-west-2":
					gotS3 = row.Quantity
				}
			}
			if gotDB != tc.wantDB || gotS3 != tc.wantS3 {
				t.Fatalf("quantities = %d/%d, want %d/%d", gotDB, gotS3, tc.wantDB, tc.wantS3)
			}
			if len(outbox) != tc.wantMsg {
				t.Fatalf("outbox = %d, want %d (messages only for positive quantities)", len(outbox), tc.wantMsg)
			}
		})
	}
}