package meta

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/pingcap/failpoint"
)

// Billing metered-unit SKU stems. The emitted BillingMeasure.unit composes
// them as fs-{sku}:{provider}:{region} (e.g. fs-read-ops:aws:us-west-2);
// the ledger stores the same composite string. Do not rename them once
// messages have been emitted. Pooled = direct-to-S3 traffic (traffic_path
// s3); non-pooled = through the drive9 HTTP data plane (traffic_path db9).
// Storage - Performance is the DB/cache tier, Storage - Pooled the S3 tier.
const (
	BillingUnitReadOps            = "read-ops"
	BillingUnitWriteOps           = "write-ops"
	BillingUnitPooledReadOps      = "pooled-read-ops"
	BillingUnitPooledWriteOps     = "pooled-write-ops"
	BillingUnitStoragePerf        = "storage-performance"
	BillingUnitStoragePooled      = "storage-pooled"
	BillingLedgerStatusComplete   = "complete"
	BillingLedgerStatusIncomplete = "incomplete"
)

// Billing outbox row states.
const (
	BillingOutboxPending         = "pending"
	BillingOutboxSending         = "sending"
	BillingOutboxSent            = "sent"
	BillingOutboxPermanentFailed = "permanent_failed"
)

// BillingStorageSnapshot is one point-in-time per-tenant confirmed storage
// reading. Tenants whose storage tier scan has not completed fold their total
// storage_bytes into DBBytes (see ListTenantStorageTierUsage).
type BillingStorageSnapshot struct {
	TenantID   string
	OrgID      string
	DBBytes    int64
	S3Bytes    int64
	CapturedAt time.Time
}

// BillingUsageLedgerRow is one closed (or incomplete) metering bucket for a
// tenant, unit, and UTC hour.
type BillingUsageLedgerRow struct {
	CostID      string
	TenantID    string
	OrgID       string
	Unit        string
	PeriodStart time.Time
	Quantity    int64
	Labels      map[string]string
	Status      string
}

// BillingOutboxRow is one SQS message awaiting delivery.
type BillingOutboxRow struct {
	CostID   string
	Payload  []byte
	Status   string
	Attempts int
}

// BillingOutboxStats aggregates outbox state for worker metrics.
type BillingOutboxStats struct {
	Pending             int64
	Sending             int64
	Sent                int64
	PermanentFailed     int64
	OldestPendingAgeSec float64
}

func billingSchemaStatements() []string {
	return []string{
		`CREATE TABLE IF NOT EXISTS billing_storage_snapshot (
			id          BIGINT AUTO_INCREMENT PRIMARY KEY,
			tenant_id   VARCHAR(64) NOT NULL,
			org_id      VARCHAR(64) NOT NULL,
			db_bytes    BIGINT NOT NULL DEFAULT 0,
			s3_bytes    BIGINT NOT NULL DEFAULT 0,
			captured_at DATETIME(3) NOT NULL,
			created_at  DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			INDEX idx_billing_snapshot_tenant_time (tenant_id, captured_at),
			INDEX idx_billing_snapshot_org_time (org_id, captured_at)
		)`,
		`CREATE TABLE IF NOT EXISTS billing_usage_ledger (
			cost_id      VARCHAR(36) PRIMARY KEY,
			tenant_id    VARCHAR(64) NOT NULL DEFAULT '',
			org_id       VARCHAR(64) NOT NULL,
			unit         VARCHAR(64) NOT NULL,
			period_start DATETIME(3) NOT NULL,
			quantity     BIGINT NOT NULL DEFAULT 0,
			labels       JSON NULL,
			status       VARCHAR(16) NOT NULL DEFAULT 'complete',
			created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			INDEX idx_billing_ledger_period (org_id, unit, period_start),
			INDEX idx_billing_ledger_incomplete (status, period_start)
		)`,
		`CREATE TABLE IF NOT EXISTS billing_sqs_outbox (
			cost_id         VARCHAR(36) PRIMARY KEY,
			payload         JSON NOT NULL,
			status          VARCHAR(20) NOT NULL DEFAULT 'pending',
			lease_owner     VARCHAR(128) NOT NULL DEFAULT '',
			lease_until     DATETIME(3) NULL,
			attempts        INT UNSIGNED NOT NULL DEFAULT 0,
			next_attempt_at DATETIME(3) NOT NULL DEFAULT '1970-01-01 00:00:00',
			last_error      TEXT NULL,
			created_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
			updated_at      DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
			sent_at         DATETIME(3) NULL,
			INDEX idx_billing_outbox_claim (status, next_attempt_at, lease_until)
		)`,
		`CREATE TABLE IF NOT EXISTS billing_collect_checkpoint (
			id                  TINYINT PRIMARY KEY,
			last_collected_hour DATETIME(3) NOT NULL,
			updated_at          DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)
		)`,
	}
}

// TenantStorageTierUsage is the per-tenant confirmed storage split used for
// billing snapshots. Tenants whose storage tier scan has not completed
// (db_storage_bytes IS NULL) fold their whole storage_bytes into DBBytes; the
// S3 share is only known once the tier counters are populated.
type TenantStorageTierUsage struct {
	TenantID       string
	TiDBCloudOrgID string
	DBBytes        int64
	S3Bytes        int64
}

// listTenantStorageTierUsageSQL mirrors listOrganizationStorageUsageSQL but
// returns per-tenant rows with the db9/s3 confirmed split. COALESCE implements
// the fallback rule for tenants without tier data: all bytes count as DB
// storage.
const listTenantStorageTierUsageSQL = `SELECT tenant_id, organization_id,
       COALESCE(db_storage_bytes, storage_bytes),
       COALESCE(s3_storage_bytes, 0)
FROM (
	SELECT q.tenant_id, b.organization_id, q.storage_bytes, q.db_storage_bytes, q.s3_storage_bytes
	FROM tenant_quota_usage q
	STRAIGHT_JOIN tenants t ON t.id = q.tenant_id
	STRAIGHT_JOIN tenant_tidbcloud_org_bindings b ON b.tenant_id = q.tenant_id
	WHERE t.provider <> ? AND t.status <> ? AND b.organization_id <> ''
	UNION ALL
	SELECT q.tenant_id, d.org_id AS organization_id, q.storage_bytes, q.db_storage_bytes, q.s3_storage_bytes
	FROM tenant_quota_usage q
	STRAIGHT_JOIN tenants t ON t.id = q.tenant_id
	STRAIGHT_JOIN fs_registry f ON f.tenant_id = q.tenant_id
	STRAIGHT_JOIN tenant_placements p ON p.fs_id = f.fs_id
	STRAIGHT_JOIN db_pool d ON d.db_id = p.db_id
	WHERE t.provider = ? AND t.status <> ? AND d.org_id <> ''
) tenant_tier_usage
ORDER BY tenant_id`

// ListTenantStorageTierUsage returns per-tenant confirmed storage split into
// DB and S3 tiers for billing snapshots.
func (s *Store) ListTenantStorageTierUsage(ctx context.Context) (out []TenantStorageTierUsage, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_tenant_storage_tier_usage", start, &err)
	rows, err := s.db.QueryContext(ctx, listTenantStorageTierUsageSQL,
		tidbCloudNativeSharedProvider, TenantDeleted,
		tidbCloudNativeSharedProvider, TenantDeleted,
	)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var usage TenantStorageTierUsage
		if err = rows.Scan(&usage.TenantID, &usage.TiDBCloudOrgID, &usage.DBBytes, &usage.S3Bytes); err != nil {
			return out, err
		}
		out = append(out, usage)
	}
	return out, rows.Err()
}

// InsertBillingStorageSnapshots inserts one snapshot row per tenant in a
// single transaction. capturedAt is truncated to milliseconds to match the
// DATETIME(3) column.
func (s *Store) InsertBillingStorageSnapshots(ctx context.Context, snapshots []BillingStorageSnapshot) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_billing_storage_snapshots", start, &err)
	if len(snapshots) == 0 {
		return nil
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	stmt, err := tx.PrepareContext(ctx, `INSERT INTO billing_storage_snapshot (tenant_id, org_id, db_bytes, s3_bytes, captured_at) VALUES (?, ?, ?, ?, ?)`)
	if err != nil {
		return err
	}
	defer func() { _ = stmt.Close() }()
	for _, snap := range snapshots {
		if snap.TenantID == "" || snap.OrgID == "" {
			return fmt.Errorf("billing snapshot with empty tenant/org id")
		}
		if _, err = stmt.ExecContext(ctx, snap.TenantID, snap.OrgID, snap.DBBytes, snap.S3Bytes,
			snap.CapturedAt.UTC().Truncate(time.Millisecond)); err != nil {
			return err
		}
	}
	return tx.Commit()
}

// LatestBillingStorageSnapshotAtOrBefore returns the newest snapshot for the
// tenant captured at or before ts. Returns ErrNotFound when none exists.
func (s *Store) LatestBillingStorageSnapshotAtOrBefore(ctx context.Context, tenantID string, ts time.Time) (snap BillingStorageSnapshot, err error) {
	start := time.Now()
	defer observeMeta(ctx, "latest_billing_storage_snapshot", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT tenant_id, org_id, db_bytes, s3_bytes, captured_at
		FROM billing_storage_snapshot
		WHERE tenant_id = ? AND captured_at <= ?
		ORDER BY captured_at DESC, id DESC
		LIMIT 1`, tenantID, ts.UTC().Truncate(time.Millisecond))
	if err = row.Scan(&snap.TenantID, &snap.OrgID, &snap.DBBytes, &snap.S3Bytes, &snap.CapturedAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return snap, ErrNotFound
		}
		return snap, err
	}
	return snap, nil
}

// DeleteBillingStorageSnapshotsBefore removes snapshots older than the given
// time, but always keeps each tenant's newest snapshot below the cutoff: that
// row is the boundary evidence the collector needs to mark lagged hours as
// incomplete instead of silently closing them with no storage rows. Returns
// the number of deleted rows.
func (s *Store) DeleteBillingStorageSnapshotsBefore(ctx context.Context, before time.Time) (deleted int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_billing_storage_snapshots", start, &err)
	res, err := s.db.ExecContext(ctx, `DELETE FROM billing_storage_snapshot
		WHERE captured_at < ? AND id NOT IN (
			SELECT id FROM (
				SELECT MAX(id) AS id FROM billing_storage_snapshot
				WHERE captured_at < ? GROUP BY tenant_id
			) keep
		)`, before.UTC().Truncate(time.Millisecond), before.UTC().Truncate(time.Millisecond))
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// BillingSnapshotTenant is a tenant's latest storage snapshot at or before a
// given time — the capacity row the billing worker bills for that hour.
type BillingSnapshotTenant struct {
	TenantID   string
	OrgID      string
	DBBytes    int64
	S3Bytes    int64
	CapturedAt time.Time
}

// ListBillingSnapshotTenants returns the newest snapshot per tenant captured
// at or before the given time. Used by the billing worker when closing an
// hour; the hour's billed capacity is this point-in-time reading.
func (s *Store) ListBillingSnapshotTenants(ctx context.Context, before time.Time) (out []BillingSnapshotTenant, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_billing_snapshot_tenants", start, &err)
	rows, err := s.db.QueryContext(ctx, `SELECT tenant_id, org_id, db_bytes, s3_bytes, captured_at FROM (
			SELECT tenant_id, org_id, db_bytes, s3_bytes, captured_at,
			       ROW_NUMBER() OVER (PARTITION BY tenant_id ORDER BY captured_at DESC, id DESC) AS rn
			FROM billing_storage_snapshot
			WHERE captured_at <= ?
		) latest WHERE rn = 1
		ORDER BY tenant_id`, before.UTC().Truncate(time.Millisecond))
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var tenant BillingSnapshotTenant
		if err = rows.Scan(&tenant.TenantID, &tenant.OrgID, &tenant.DBBytes, &tenant.S3Bytes, &tenant.CapturedAt); err != nil {
			return out, err
		}
		out = append(out, tenant)
	}
	return out, rows.Err()
}

// BillingIncompleteHour identifies one tenant's incomplete metering bucket,
// used by the billing worker's bounded retry pass.
type BillingIncompleteHour struct {
	TenantID    string
	OrgID       string
	PeriodStart time.Time
}

// ListIncompleteBillingStorageHours returns distinct (tenant_id, period_start)
// pairs that have incomplete ledger rows with period_start at or after since.
func (s *Store) ListIncompleteBillingStorageHours(ctx context.Context, since time.Time) (out []BillingIncompleteHour, err error) {
	start := time.Now()
	defer observeMeta(ctx, "list_incomplete_billing_hours", start, &err)
	rows, err := s.db.QueryContext(ctx, `SELECT DISTINCT tenant_id, org_id, period_start FROM billing_usage_ledger
		WHERE status = ? AND period_start >= ?
		ORDER BY period_start, tenant_id`, BillingLedgerStatusIncomplete, since.UTC().Truncate(time.Millisecond))
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var hour BillingIncompleteHour
		if err = rows.Scan(&hour.TenantID, &hour.OrgID, &hour.PeriodStart); err != nil {
			return out, err
		}
		out = append(out, hour)
	}
	return out, rows.Err()
}

// BillingCollectCheckpoint reads the singleton collection checkpoint. The
// second return value is false when no hour has been collected yet.
func (s *Store) BillingCollectCheckpoint(ctx context.Context) (lastHour time.Time, ok bool, err error) {
	start := time.Now()
	defer observeMeta(ctx, "billing_collect_checkpoint_get", start, &err)
	row := s.db.QueryRowContext(ctx, `SELECT last_collected_hour FROM billing_collect_checkpoint WHERE id = 1`)
	if err = row.Scan(&lastHour); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return time.Time{}, false, nil
		}
		return time.Time{}, false, err
	}
	return lastHour, true, nil
}

// AdvanceBillingCollectCheckpoint upserts the singleton checkpoint. It is
// only used for first-run initialization; steady-state hour closes advance
// the checkpoint inside WriteBillingHourWithOutbox's transaction.
func (s *Store) AdvanceBillingCollectCheckpoint(ctx context.Context, lastHour time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "billing_collect_checkpoint_advance", start, &err)
	return advanceBillingCollectCheckpoint(ctx, s.db, lastHour)
}

// advanceBillingCollectCheckpoint is the single source of truth for the
// checkpoint upsert, shared by first-run init and the atomic hour close.
type execer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

func advanceBillingCollectCheckpoint(ctx context.Context, ex execer, lastHour time.Time) error {
	_, err := ex.ExecContext(ctx, `INSERT INTO billing_collect_checkpoint (id, last_collected_hour) VALUES (1, ?)
		ON DUPLICATE KEY UPDATE last_collected_hour = VALUES(last_collected_hour)`,
		lastHour.UTC().Truncate(time.Millisecond))
	return err
}

// writeBillingLedgerAndOutboxTx is the single source of truth for the billing
// write contract: ledger upserts (idempotent per cost_id) plus outbox inserts
// (INSERT IGNORE, payload is a pure function of the cost_id inputs) on an
// existing transaction.
func writeBillingLedgerAndOutboxTx(ctx context.Context, tx *sql.Tx, ledger []BillingUsageLedgerRow, outbox []BillingOutboxRow) error {
	ledgerStmt, err := tx.PrepareContext(ctx, `INSERT INTO billing_usage_ledger
		(cost_id, tenant_id, org_id, unit, period_start, quantity, labels, status)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		ON DUPLICATE KEY UPDATE quantity = VALUES(quantity), labels = VALUES(labels), status = VALUES(status)`)
	if err != nil {
		return err
	}
	defer func() { _ = ledgerStmt.Close() }()
	for _, row := range ledger {
		labels, lerr := json.Marshal(row.Labels)
		if lerr != nil {
			return fmt.Errorf("marshal ledger labels for %s: %w", row.CostID, lerr)
		}
		if _, err = ledgerStmt.ExecContext(ctx, row.CostID, row.TenantID, row.OrgID, row.Unit,
			row.PeriodStart.UTC().Truncate(time.Millisecond), row.Quantity, string(labels), row.Status); err != nil {
			return err
		}
	}

	outboxStmt, err := tx.PrepareContext(ctx, `INSERT IGNORE INTO billing_sqs_outbox (cost_id, payload) VALUES (?, ?)`)
	if err != nil {
		return err
	}
	defer func() { _ = outboxStmt.Close() }()
	for _, row := range outbox {
		if _, err = outboxStmt.ExecContext(ctx, row.CostID, string(row.Payload)); err != nil {
			return err
		}
	}
	return nil
}

// WriteBillingHourWithOutbox upserts one hour's ledger rows, inserts the
// matching SQS outbox rows, and advances the collection checkpoint — all in
// one transaction. Closing the hour and recording that it closed are atomic,
// so a crash between them can never re-collect the hour under a different
// (e.g. post-upgrade) CostID scheme.
func (s *Store) WriteBillingHourWithOutbox(ctx context.Context, ledger []BillingUsageLedgerRow, outbox []BillingOutboxRow, checkpointHour time.Time) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "write_billing_hour_with_outbox", start, &err)
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	if err = writeBillingLedgerAndOutboxTx(ctx, tx, ledger, outbox); err != nil {
		return err
	}

	if err = advanceBillingCollectCheckpoint(ctx, tx, checkpointHour); err != nil {
		return err
	}
	// Failpoint for the atomicity regression test: abort after all writes,
	// before commit (nothing must be visible afterwards).
	var injected error
	failpoint.InjectCall("billingHourBeforeCommit", &injected)
	if injected != nil {
		return injected
	}
	return tx.Commit()
}

// WriteBillingUsageWithOutbox upserts ledger rows and inserts the matching
// SQS outbox rows in one transaction, without touching the checkpoint. Ledger
// rows are idempotent per cost_id (which encodes tenant, org, unit, period,
// and billing dimensions such as traffic_path, so multiple dimensions per
// tenant/unit/hour coexist); outbox rows are keyed by cost_id and never
// rewritten once present (the payload is a pure function of cost_id inputs).
//
// Caller contract: pass an outbox row only for complete ledger rows with
// quantity > 0 — this function inserts whatever it is given.
func (s *Store) WriteBillingUsageWithOutbox(ctx context.Context, ledger []BillingUsageLedgerRow, outbox []BillingOutboxRow) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "write_billing_usage_with_outbox", start, &err)
	if len(ledger) == 0 && len(outbox) == 0 {
		return nil
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if err = writeBillingLedgerAndOutboxTx(ctx, tx, ledger, outbox); err != nil {
		return err
	}
	return tx.Commit()
}

// ClaimBillingOutboxBatch claims up to limit due pending (next_attempt_at
// reached) or expired in-flight outbox rows for owner, transitioning them to
// sending. SendMessage happens outside the transaction; the lease lets a
// crashed sender's rows be reclaimed after leaseUntil.
func (s *Store) ClaimBillingOutboxBatch(ctx context.Context, owner string, lease time.Duration, limit int) (out []BillingOutboxRow, err error) {
	start := time.Now()
	defer observeMeta(ctx, "claim_billing_outbox", start, &err)
	if owner == "" || limit <= 0 {
		return nil, nil
	}
	now := time.Now().UTC().Truncate(time.Millisecond)
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()
	rows, err := tx.QueryContext(ctx, `SELECT cost_id, payload, attempts
		FROM billing_sqs_outbox
		WHERE (status = ? AND next_attempt_at <= ?) OR (status = ? AND lease_until IS NOT NULL AND lease_until <= ?)
		ORDER BY created_at
		LIMIT ?
		FOR UPDATE`, BillingOutboxPending, now, BillingOutboxSending, now, limit)
	if err != nil {
		return nil, err
	}
	var ids []string
	payloads := map[string][]byte{}
	attempts := map[string]int{}
	for rows.Next() {
		var id string
		var payload string
		var att int
		if err = rows.Scan(&id, &payload, &att); err != nil {
			_ = rows.Close()
			return nil, err
		}
		ids = append(ids, id)
		payloads[id] = []byte(payload)
		attempts[id] = att
	}
	if err = rows.Err(); err != nil {
		_ = rows.Close()
		return nil, err
	}
	_ = rows.Close()
	if len(ids) == 0 {
		return nil, tx.Commit()
	}
	leaseUntil := now.Add(lease).Truncate(time.Millisecond)
	stmt, err := tx.PrepareContext(ctx, `UPDATE billing_sqs_outbox
		SET status = ?, lease_owner = ?, lease_until = ? WHERE cost_id = ?`)
	if err != nil {
		return nil, err
	}
	defer func() { _ = stmt.Close() }()
	for _, id := range ids {
		if _, err = stmt.ExecContext(ctx, BillingOutboxSending, owner, leaseUntil, id); err != nil {
			return nil, err
		}
		out = append(out, BillingOutboxRow{CostID: id, Payload: payloads[id], Status: BillingOutboxSending, Attempts: attempts[id]})
	}
	return out, tx.Commit()
}

// MarkBillingOutboxSent marks a claimed row sent. The lease_owner guard
// fences a stale sender whose lease expired and was reclaimed.
func (s *Store) MarkBillingOutboxSent(ctx context.Context, costID, owner string) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_billing_outbox_sent", start, &err)
	_, err = s.db.ExecContext(ctx, `UPDATE billing_sqs_outbox
		SET status = ?, sent_at = ?, lease_owner = '', lease_until = NULL
		WHERE cost_id = ? AND status = ? AND lease_owner = ?`,
		BillingOutboxSent, time.Now().UTC().Truncate(time.Millisecond), costID, BillingOutboxSending, owner)
	return err
}

// MarkBillingOutboxFailed records a failed send attempt. Retryable failures
// return the row to pending with a next_attempt_at backoff (computed by the
// caller); permanent failures (or exhausted attempts) move it to
// permanent_failed for manual requeue.
func (s *Store) MarkBillingOutboxFailed(ctx context.Context, costID, owner string, nextAttemptAt time.Time, sendErr error) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "mark_billing_outbox_failed", start, &err)
	errText := ""
	if sendErr != nil {
		errText = sendErr.Error()
		if len(errText) > 1024 {
			errText = errText[:1024]
		}
	}
	status := BillingOutboxPermanentFailed
	if !nextAttemptAt.IsZero() {
		status = BillingOutboxPending
	}
	if nextAttemptAt.IsZero() {
		// Permanent failures park far in the future instead of a zero time
		// (TiDB rejects 0000-00-00 in strict mode); they are never claimable
		// because their status is not pending.
		nextAttemptAt = time.Date(9999, 12, 31, 23, 59, 59, 0, time.UTC)
	}
	_, err = s.db.ExecContext(ctx, `UPDATE billing_sqs_outbox
		SET status = ?, lease_owner = '', lease_until = NULL, attempts = attempts + 1,
		    next_attempt_at = ?, last_error = ?
		WHERE cost_id = ? AND status = ? AND lease_owner = ?`,
		status, nextAttemptAt.UTC().Truncate(time.Millisecond), errText, costID, BillingOutboxSending, owner)
	return err
}

// DeleteTerminalBillingOutboxBefore removes sent and permanent_failed rows
// created before the given time, bounding the outbox table. Pending/sending
// rows are never pruned here.
func (s *Store) DeleteTerminalBillingOutboxBefore(ctx context.Context, before time.Time) (deleted int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_terminal_billing_outbox", start, &err)
	res, err := s.db.ExecContext(ctx, `DELETE FROM billing_sqs_outbox
		WHERE status IN (?, ?) AND created_at < ?`,
		BillingOutboxSent, BillingOutboxPermanentFailed, before.UTC().Truncate(time.Millisecond))
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}

// PruneBillingLedgerRows deletes ledger rows that carry no billing
// information: zero-quantity rows (written so retried hours converge, but
// they never produce a message) and incomplete rows that can no longer be
// recovered (older than the caller's cutoff, which the worker sets beyond
// the incomplete-retry window). Returns the number of deleted rows.
//
// Two batched DELETEs instead of one OR: `quantity` is not indexed, so a
// single disjunct scan would grow with the ledger; batching also bounds a
// first-run backlog so one statement cannot trip TiDB's txn entry limit.
func (s *Store) PruneBillingLedgerRows(ctx context.Context, before time.Time) (deleted int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "prune_billing_ledger_rows", start, &err)
	at := before.UTC().Truncate(time.Millisecond)
	for _, q := range []struct {
		sql    string
		status string
	}{
		{`DELETE FROM billing_usage_ledger WHERE status = ? AND period_start < ? LIMIT ?`, BillingLedgerStatusIncomplete},
		{`DELETE FROM billing_usage_ledger WHERE status = ? AND quantity = 0 AND period_start < ? LIMIT ?`, BillingLedgerStatusComplete},
	} {
		for {
			res, eerr := s.db.ExecContext(ctx, q.sql, q.status, at, billingLedgerPruneBatchSize)
			if eerr != nil {
				return deleted, eerr
			}
			n, eerr := res.RowsAffected()
			if eerr != nil {
				return deleted, eerr
			}
			deleted += n
			if n < billingLedgerPruneBatchSize {
				break
			}
		}
	}
	return deleted, nil
}

// billingLedgerPruneBatchSize bounds rows deleted per statement during ledger
// pruning, mirroring the mutation-log purge worker's batching.
const billingLedgerPruneBatchSize = 1000

// BillingOutboxStats aggregates outbox counts by status and the age of the
// oldest pending row for worker metrics.
func (s *Store) BillingOutboxStats(ctx context.Context) (stats BillingOutboxStats, err error) {
	start := time.Now()
	defer observeMeta(ctx, "billing_outbox_stats", start, &err)
	rows, err := s.db.QueryContext(ctx, `SELECT status, COUNT(*) FROM billing_sqs_outbox GROUP BY status`)
	if err != nil {
		return stats, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var status string
		var count int64
		if err = rows.Scan(&status, &count); err != nil {
			return stats, err
		}
		switch status {
		case BillingOutboxPending:
			stats.Pending = count
		case BillingOutboxSending:
			stats.Sending = count
		case BillingOutboxSent:
			stats.Sent = count
		case BillingOutboxPermanentFailed:
			stats.PermanentFailed = count
		}
	}
	if err = rows.Err(); err != nil {
		return stats, err
	}
	row := s.db.QueryRowContext(ctx, `SELECT MIN(created_at) FROM billing_sqs_outbox WHERE status = ?`, BillingOutboxPending)
	var oldest sql.NullTime
	if err = row.Scan(&oldest); err != nil {
		return stats, err
	}
	if oldest.Valid {
		stats.OldestPendingAgeSec = time.Since(oldest.Time).Seconds()
	}
	return stats, nil
}