package billing

import (
	"bufio"
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"aigw/internal/domain"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/stripe/stripe-go/v86"
)

const microsPerUnit = int64(1_000_000)

type Service struct {
	db                          *pgxpool.Pool
	currency                    string
	defaultMaxOutputTokens      int64
	minTopUpMinor               int64
	maxTopUpMinor               int64
	stripeEnabled               bool
	stripeWebhookSecret         string
	stripeSuccessURL            string
	stripeCancelURL             string
	stripePortalReturnURL       string
	stripeAutomaticTax          bool
	stripeProductTaxCode        string
	integrationIdentifier       string
	createStripeCheckout        stripeCheckoutCreator
	createStripePortalSession   stripePortalSessionCreator
	createStripeCustomer        stripeCustomerCreator
	updateStripeCustomer        stripeCustomerUpdater
	retrieveStripeSetupIntent   stripeSetupIntentRetriever
	createStripePaymentIntent   stripePaymentIntentCreator
	retrieveStripePaymentIntent stripePaymentIntentRetriever
	stripeClient                *stripe.Client
	settlementSpoolPath         string
	metrics                     OperationalMetrics
	spoolMu                     sync.Mutex
}

func New(ctx context.Context, options Options) (*Service, error) {
	db, err := pgxpool.New(ctx, options.DatabaseURL)
	if err != nil {
		return nil, fmt.Errorf("configure billing PostgreSQL: %w", err)
	}
	if err := db.Ping(ctx); err != nil {
		db.Close()
		return nil, fmt.Errorf("connect billing PostgreSQL: %w", err)
	}
	service := &Service{
		db: db, currency: options.Currency, defaultMaxOutputTokens: options.DefaultMaxOutputTokens,
		minTopUpMinor: options.MinTopUpMinor, maxTopUpMinor: options.MaxTopUpMinor,
		stripeEnabled: options.StripeEnabled, stripeWebhookSecret: options.StripeWebhookSecret,
		stripeSuccessURL: options.StripeSuccessURL, stripeCancelURL: options.StripeCancelURL,
		stripePortalReturnURL: options.StripePortalReturnURL, stripeAutomaticTax: options.StripeAutomaticTax,
		stripeProductTaxCode:  options.StripeProductTaxCode,
		integrationIdentifier: "aigw_balance_" + randomLetters(8),
		settlementSpoolPath:   strings.TrimSpace(options.SettlementSpoolPath),
		metrics:               options.Metrics,
	}
	if options.StripeEnabled {
		service.stripeClient = stripe.NewClient(options.StripeAPIKey)
		service.createStripeCheckout = service.stripeClient.V1CheckoutSessions.Create
		service.createStripePortalSession = service.stripeClient.V1BillingPortalSessions.Create
		service.createStripeCustomer = service.stripeClient.V1Customers.Create
		service.updateStripeCustomer = service.stripeClient.V1Customers.Update
		service.retrieveStripeSetupIntent = service.stripeClient.V1SetupIntents.Retrieve
		service.createStripePaymentIntent = service.stripeClient.V1PaymentIntents.Create
		service.retrieveStripePaymentIntent = service.stripeClient.V1PaymentIntents.Retrieve
	}
	return service, nil
}

func (s *Service) Close() {
	s.db.Close()
}

func (s *Service) StripeEnabled() bool {
	return s.stripeEnabled
}

func (s *Service) Currency() string {
	return s.currency
}

func (s *Service) Ping(ctx context.Context) error { return s.db.Ping(ctx) }

func (s *Service) Authorize(ctx context.Context, input Authorization) error {
	if input.RequestID == "" || input.Principal.TenantID == "" || input.Principal.ProjectID == "" || input.Principal.KeyID == "" {
		return errors.New("billing authorization identity is incomplete")
	}
	if input.Model.PriceCurrency != "" && input.Model.PriceCurrency != s.currency {
		return fmt.Errorf("model price currency %s does not match wallet currency %s", input.Model.PriceCurrency, s.currency)
	}
	reserved, err := reservationCost(input.Model, input.Body, s.defaultMaxOutputTokens, input.Protocol)
	if err != nil {
		return err
	}
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return fmt.Errorf("begin billing authorization: %w", err)
	}
	defer tx.Rollback(ctx)
	if _, err := tx.Exec(ctx, `
		INSERT INTO tenant_wallets (tenant_id, currency) VALUES ($1, $2)
		ON CONFLICT (tenant_id) DO NOTHING`, input.Principal.TenantID, s.currency); err != nil {
		return fmt.Errorf("ensure tenant wallet: %w", err)
	}
	var currency string
	var balance, held int64
	if err := tx.QueryRow(ctx, `
		SELECT currency, balance_micros, reserved_micros FROM tenant_wallets
		WHERE tenant_id = $1 FOR UPDATE`, input.Principal.TenantID).Scan(&currency, &balance, &held); err != nil {
		return fmt.Errorf("lock tenant wallet: %w", err)
	}
	if currency != s.currency {
		return fmt.Errorf("tenant wallet currency %s does not match billing currency %s", currency, s.currency)
	}
	if input.Policy.MonthlySpendMicros > 0 {
		period := time.Date(time.Now().UTC().Year(), time.Now().UTC().Month(), 1, 0, 0, 0, 0, time.UTC)
		nextPeriod := period.AddDate(0, 1, 0)
		var used, pending int64
		if err := tx.QueryRow(ctx, `SELECT
			COALESCE((SELECT cost_micros FROM usage_monthly_rollups WHERE project_id=$1 AND period_start=$2),0),
			COALESCE((SELECT sum(reserved_micros) FROM billing_reservations WHERE project_id=$1 AND status IN ('pending','metering_failed') AND created_at >= $2 AND created_at < $3),0)`,
			input.Principal.ProjectID, period, nextPeriod).Scan(&used, &pending); err != nil {
			return fmt.Errorf("read monthly spend quota: %w", err)
		}
		if reserved > input.Policy.MonthlySpendMicros || used > input.Policy.MonthlySpendMicros-reserved || pending > input.Policy.MonthlySpendMicros-used-reserved {
			return ErrQuotaExceeded
		}
	}
	if input.Principal.MonthlySpendMicros > 0 {
		period := time.Date(time.Now().UTC().Year(), time.Now().UTC().Month(), 1, 0, 0, 0, 0, time.UTC)
		nextPeriod := period.AddDate(0, 1, 0)
		var used, pending int64
		if err := tx.QueryRow(ctx, `SELECT
			COALESCE((SELECT sum(cost_micros) FROM usage_events WHERE key_id=$1 AND started_at >= $2 AND started_at < $3),0),
			COALESCE((SELECT sum(reserved_micros) FROM billing_reservations WHERE key_id=$1 AND status IN ('pending','metering_failed') AND created_at >= $2 AND created_at < $3),0)`,
			input.Principal.KeyID, period, nextPeriod).Scan(&used, &pending); err != nil {
			return fmt.Errorf("read API key monthly spend quota: %w", err)
		}
		limit := input.Principal.MonthlySpendMicros
		if reserved > limit || used > limit-reserved || pending > limit-used-reserved {
			return ErrQuotaExceeded
		}
	}
	if input.Principal.DailySpendMicros > 0 {
		now := time.Now().UTC()
		period := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC)
		nextPeriod := period.AddDate(0, 0, 1)
		var used, pending int64
		if err := tx.QueryRow(ctx, `SELECT
			COALESCE((SELECT sum(cost_micros) FROM usage_events WHERE key_id=$1 AND started_at >= $2 AND started_at < $3),0),
			COALESCE((SELECT sum(reserved_micros) FROM billing_reservations WHERE key_id=$1 AND status IN ('pending','metering_failed') AND created_at >= $2 AND created_at < $3),0)`,
			input.Principal.KeyID, period, nextPeriod).Scan(&used, &pending); err != nil {
			return fmt.Errorf("read API key daily spend quota: %w", err)
		}
		limit := input.Principal.DailySpendMicros
		if reserved > limit || used > limit-reserved || pending > limit-used-reserved {
			return ErrDailyQuotaExceeded
		}
	}
	if balance-held < reserved {
		return ErrInsufficientBalance
	}
	if _, err := tx.Exec(ctx, `
		INSERT INTO billing_reservations (
			request_id, tenant_id, project_id, key_id, public_model, currency, reserved_micros,
			price_version_id,
			input_price_micros_per_million, output_price_micros_per_million,
			cache_read_price_micros_per_million, cache_write_price_micros_per_million)
		VALUES ($1,$2,$3,$4,$5,$6,$7,NULLIF($8,'')::uuid,$9,$10,$11,$12)`,
		input.RequestID, input.Principal.TenantID, input.Principal.ProjectID, input.Principal.KeyID,
		input.Model.ID, s.currency, reserved, input.Model.PriceVersionID, input.Model.InputPriceMicrosPerMillion,
		input.Model.OutputPriceMicrosPerMillion, input.Model.CacheReadPriceMicrosPerMillion,
		input.Model.CacheWritePriceMicrosPerMillion); err != nil {
		return fmt.Errorf("create billing reservation: %w", err)
	}
	if _, err := tx.Exec(ctx, `INSERT INTO billing_settlement_jobs (request_id) VALUES ($1)`, input.RequestID); err != nil {
		return fmt.Errorf("create settlement job: %w", err)
	}
	if _, err := tx.Exec(ctx, `
		UPDATE tenant_wallets SET reserved_micros = reserved_micros + $2, updated_at = now()
		WHERE tenant_id = $1`, input.Principal.TenantID, reserved); err != nil {
		return fmt.Errorf("reserve tenant balance: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("commit billing authorization: %w", err)
	}
	return nil
}

func (s *Service) EnqueueSettlement(ctx context.Context, event domain.UsageEvent) error {
	payload, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("encode settlement event: %w", err)
	}
	_, err = s.db.Exec(ctx, `
		INSERT INTO billing_settlement_jobs (request_id, event, status, available_at, updated_at)
		VALUES ($1,$2,'pending',now(),now())
		ON CONFLICT (request_id) DO UPDATE SET event=EXCLUDED.event,
			status=CASE WHEN billing_settlement_jobs.status='done' THEN 'done' ELSE 'pending' END,
			available_at=now(), locked_at=NULL, last_error='', updated_at=now()`, event.RequestID, payload)
	if err == nil {
		return nil
	}
	if spoolErr := s.appendSettlementSpool(payload); spoolErr != nil {
		return fmt.Errorf("enqueue settlement in PostgreSQL: %v; append durable spool: %w", err, spoolErr)
	}
	return nil
}

func (s *Service) settle(ctx context.Context, event domain.UsageEvent) error {
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return fmt.Errorf("begin usage settlement: %w", err)
	}
	defer tx.Rollback(ctx)

	var tenantID, projectID, keyID, modelID, currency, status string
	var reserved, inputPrice, outputPrice, cacheReadPrice, cacheWritePrice int64
	err = tx.QueryRow(ctx, `
		SELECT tenant_id::text, project_id::text, key_id::text, public_model, currency, reserved_micros, status,
		       input_price_micros_per_million, output_price_micros_per_million,
		       cache_read_price_micros_per_million, cache_write_price_micros_per_million
		FROM billing_reservations WHERE request_id = $1 FOR UPDATE`, event.RequestID,
	).Scan(&tenantID, &projectID, &keyID, &modelID, &currency, &reserved, &status,
		&inputPrice, &outputPrice, &cacheReadPrice, &cacheWritePrice)
	if err != nil {
		return fmt.Errorf("load billing reservation: %w", err)
	}
	if _, err := tx.Exec(ctx, `UPDATE api_keys
		SET last_used_at = GREATEST(COALESCE(last_used_at, $2), $2)
		WHERE id = $1`, keyID, event.StartedAt); err != nil {
		return fmt.Errorf("update API key last used time: %w", err)
	}
	if status != "pending" {
		return tx.Commit(ctx)
	}

	actualCost := int64(0)
	billableSuccess := event.StatusCode >= 200 && event.StatusCode < 300 && event.Success
	if billableSuccess && !event.UsageReported {
		// Fail closed: keep the authorization hold in place and make the request
		// visible to reconciliation. Releasing it would turn an unmetered success
		// into a free request; guessing tokens here could overcharge the customer.
		if _, err := tx.Exec(ctx, `UPDATE billing_reservations SET status='metering_failed', settled_at=now()
			WHERE request_id=$1`, event.RequestID); err != nil {
			return fmt.Errorf("mark unmetered reservation: %w", err)
		}
		var usageAlreadyRecorded bool
		if err := tx.QueryRow(ctx, `SELECT true FROM usage_events WHERE request_id=$1 FOR UPDATE`, event.RequestID).Scan(&usageAlreadyRecorded); err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return fmt.Errorf("lock unmetered usage event: %w", err)
		}
		if _, err := tx.Exec(ctx, `INSERT INTO usage_events (
			request_id,tenant_id,project_id,key_id,public_model,provider_id,upstream_model,protocol,stream,
			status_code,success,error_type,attempts,started_at,duration_ms,ttft_ms,input_tokens,output_tokens,total_tokens,
			cache_creation_input_tokens,cache_read_input_tokens,cost_micros,charged_micros,uncollected_micros,
			usage_reported,metering_status)
			VALUES ($1,$2,$3,$4,$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,'usage_not_reported',$12,$13,$14,
				$15,$16,$17,$18,$19,$20,0,0,0,false,'missing')
			ON CONFLICT (request_id) DO UPDATE SET error_type='usage_not_reported',ttft_ms=EXCLUDED.ttft_ms,usage_reported=false,metering_status='missing'`,
			event.RequestID, tenantID, projectID, keyID, modelID, event.ProviderID, event.UpstreamModel,
			string(event.Protocol), event.Stream, event.StatusCode, event.Success, event.Attempts, event.StartedAt,
			event.DurationMS, event.TTFTMS, event.Usage.InputTokens, event.Usage.OutputTokens, event.Usage.TotalTokens,
			event.Usage.CacheCreationInputTokens, event.Usage.CacheReadInputTokens); err != nil {
			return fmt.Errorf("persist unmetered usage event: %w", err)
		}
		if !usageAlreadyRecorded {
			period := time.Date(event.StartedAt.UTC().Year(), event.StartedAt.UTC().Month(), 1, 0, 0, 0, 0, time.UTC)
			if _, err := tx.Exec(ctx, `INSERT INTO usage_monthly_rollups
			(period_start,tenant_id,project_id,request_count,successful_requests,input_tokens,output_tokens,total_tokens)
			VALUES ($1,$2,$3,1,1,$4,$5,$6)
			ON CONFLICT (project_id,period_start) DO UPDATE SET
				request_count=usage_monthly_rollups.request_count+1,
				successful_requests=usage_monthly_rollups.successful_requests+1,
				input_tokens=usage_monthly_rollups.input_tokens+EXCLUDED.input_tokens,
				output_tokens=usage_monthly_rollups.output_tokens+EXCLUDED.output_tokens,
				total_tokens=usage_monthly_rollups.total_tokens+EXCLUDED.total_tokens,updated_at=now()`,
				period, tenantID, projectID, event.Usage.InputTokens, event.Usage.OutputTokens, event.Usage.TotalTokens); err != nil {
				return fmt.Errorf("roll up unmetered usage event: %w", err)
			}
		}
		return tx.Commit(ctx)
	}
	if billableSuccess {
		actualCost, err = usageCost(event.Usage, inputPrice, outputPrice, cacheReadPrice, cacheWritePrice)
		if err != nil {
			return err
		}
	}
	var balance, held int64
	if err := tx.QueryRow(ctx, `SELECT balance_micros, reserved_micros FROM tenant_wallets WHERE tenant_id = $1 FOR UPDATE`, tenantID).Scan(&balance, &held); err != nil {
		return fmt.Errorf("lock wallet for settlement: %w", err)
	}
	charged, err := collectibleCharge(actualCost, balance, held, reserved)
	if err != nil {
		return err
	}
	uncollected := actualCost - charged
	newBalance := balance - charged
	if _, err := tx.Exec(ctx, `
		UPDATE tenant_wallets SET balance_micros = $2, reserved_micros = reserved_micros - $3, updated_at = now()
		WHERE tenant_id = $1`, tenantID, newBalance, reserved); err != nil {
		return fmt.Errorf("settle tenant wallet: %w", err)
	}
	reservationStatus := "released"
	if actualCost > 0 {
		reservationStatus = "settled"
	}
	if _, err := tx.Exec(ctx, `
		UPDATE billing_reservations
		SET status = $2, actual_cost_micros = $3, charged_micros = $4, uncollected_micros = $5, settled_at = now()
		WHERE request_id = $1`, event.RequestID, reservationStatus, actualCost, charged, uncollected); err != nil {
		return fmt.Errorf("update billing reservation: %w", err)
	}
	usageAlreadyRecorded := false
	if err := tx.QueryRow(ctx, `SELECT true FROM usage_events WHERE request_id=$1 FOR UPDATE`, event.RequestID).Scan(&usageAlreadyRecorded); err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return fmt.Errorf("lock existing usage event: %w", err)
	}
	if usageAlreadyRecorded {
		if _, err := tx.Exec(ctx, `UPDATE usage_events SET cost_micros=$2, charged_micros=$3, uncollected_micros=$4,
				ttft_ms=GREATEST(ttft_ms,$5),usage_reported=$6,metering_status=$7 WHERE request_id=$1`,
			event.RequestID, actualCost, charged, uncollected, event.TTFTMS, event.UsageReported, meteringStatus(event)); err != nil {
			return fmt.Errorf("apply usage charge: %w", err)
		}
	} else if _, err := tx.Exec(ctx, `
		INSERT INTO usage_events (
			request_id, tenant_id, project_id, key_id, public_model, provider_id, upstream_model,
			protocol, stream, status_code, success, error_type, attempts, started_at, duration_ms,
			ttft_ms, input_tokens, output_tokens, total_tokens, cache_creation_input_tokens, cache_read_input_tokens,
				cost_micros, charged_micros, uncollected_micros, usage_reported, metering_status)
			VALUES ($1,$2,$3,$4,$5,NULLIF($6,''),NULLIF($7,''),$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26)
		ON CONFLICT (request_id) DO NOTHING`,
		event.RequestID, tenantID, projectID, keyID, modelID, event.ProviderID, event.UpstreamModel,
		string(event.Protocol), event.Stream, event.StatusCode, event.Success, event.ErrorType, event.Attempts,
		event.StartedAt, event.DurationMS, event.TTFTMS, event.Usage.InputTokens, event.Usage.OutputTokens,
		event.Usage.TotalTokens, event.Usage.CacheCreationInputTokens, event.Usage.CacheReadInputTokens,
		actualCost, charged, uncollected, event.UsageReported, meteringStatus(event)); err != nil {
		return fmt.Errorf("persist usage event: %w", err)
	}
	if charged > 0 {
		if _, err := tx.Exec(ctx, `
			INSERT INTO billing_ledger (tenant_id, project_id, currency, amount_micros, balance_after_micros, kind, source_type, source_id, description)
			VALUES ($1,$2,$3,$4,$5,'usage','request',$6,$7)
			ON CONFLICT (source_type, source_id) DO NOTHING`, tenantID, projectID, currency, -charged, newBalance, event.RequestID, modelID); err != nil {
			return fmt.Errorf("write usage ledger entry: %w", err)
		}
	}
	period := time.Date(event.StartedAt.UTC().Year(), event.StartedAt.UTC().Month(), 1, 0, 0, 0, 0, time.UTC)
	if usageAlreadyRecorded {
		if _, err := tx.Exec(ctx, `UPDATE usage_monthly_rollups SET cost_micros=cost_micros+$3,
			charged_micros=charged_micros+$4, uncollected_micros=uncollected_micros+$5, updated_at=now()
			WHERE project_id=$1 AND period_start=$2`, projectID, period, actualCost, charged, uncollected); err != nil {
			return fmt.Errorf("apply usage rollup charge: %w", err)
		}
	} else if _, err := tx.Exec(ctx, `
		INSERT INTO usage_monthly_rollups
		(period_start, tenant_id, project_id, request_count, successful_requests, input_tokens, output_tokens, total_tokens, cost_micros, charged_micros, uncollected_micros)
		VALUES ($1,$2,$3,1,$4,$5,$6,$7,$8,$9,$10)
		ON CONFLICT (project_id, period_start) DO UPDATE SET request_count=usage_monthly_rollups.request_count+1,
			successful_requests=usage_monthly_rollups.successful_requests+EXCLUDED.successful_requests,
			input_tokens=usage_monthly_rollups.input_tokens+EXCLUDED.input_tokens,
			output_tokens=usage_monthly_rollups.output_tokens+EXCLUDED.output_tokens,
			total_tokens=usage_monthly_rollups.total_tokens+EXCLUDED.total_tokens,
			cost_micros=usage_monthly_rollups.cost_micros+EXCLUDED.cost_micros,
			charged_micros=usage_monthly_rollups.charged_micros+EXCLUDED.charged_micros,
			uncollected_micros=usage_monthly_rollups.uncollected_micros+EXCLUDED.uncollected_micros,
			updated_at=now()`, period, tenantID, projectID, boolToInt(event.Success), event.Usage.InputTokens,
		event.Usage.OutputTokens, event.Usage.TotalTokens, actualCost, charged, uncollected); err != nil {
		return fmt.Errorf("update usage monthly rollup: %w", err)
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("commit usage settlement: %w", err)
	}
	return nil
}

// RunSettlementWorker processes jobs using PostgreSQL row locks so any gateway
// instance can resume work left by another instance after a crash.
func (s *Service) RunSettlementWorker(ctx context.Context) {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		s.drainSettlementSpool(ctx)
		s.recoverStaleSettlements(ctx)
		for i := 0; i < 32; i++ {
			processed, err := s.processSettlementJob(ctx)
			if err != nil || !processed {
				break
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (s *Service) recoverStaleSettlements(ctx context.Context) {
	// A process may die after authorization but before it can attach a response
	// event. Releasing after an hour prevents permanent holds; such synthetic
	// events stay visible in Usage for reconciliation.
	rows, err := s.db.Query(ctx, `SELECT request_id,tenant_id::text,project_id::text,key_id::text,public_model,created_at
		FROM billing_reservations WHERE status='pending' AND created_at<now()-interval '1 hour'
		AND EXISTS(SELECT 1 FROM billing_settlement_jobs j WHERE j.request_id=billing_reservations.request_id AND j.status='awaiting_event') LIMIT 100`)
	if err != nil {
		return
	}
	defer rows.Close()
	for rows.Next() {
		var event domain.UsageEvent
		if rows.Scan(&event.RequestID, &event.TenantID, &event.ProjectID, &event.KeyID, &event.PublicModel, &event.StartedAt) != nil {
			continue
		}
		event.StatusCode = 500
		event.Success = false
		event.ErrorType = "gateway_interrupted_before_settlement"
		event.DurationMS = time.Since(event.StartedAt).Milliseconds()
		_ = s.EnqueueSettlement(ctx, event)
	}
	_, _ = s.db.Exec(ctx, `UPDATE billing_settlement_jobs SET status='retry',locked_at=NULL,available_at=now(),updated_at=now(),last_error='recovered stale processing lease'
		WHERE status='processing' AND locked_at<now()-interval '5 minutes'`)
}

func (s *Service) processSettlementJob(ctx context.Context) (bool, error) {
	tx, err := s.db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return false, err
	}
	defer tx.Rollback(ctx)
	var requestID string
	var payload []byte
	err = tx.QueryRow(ctx, `
		WITH selected AS (
			SELECT request_id FROM billing_settlement_jobs
			WHERE status IN ('pending','retry') AND available_at <= now()
			ORDER BY available_at, created_at FOR UPDATE SKIP LOCKED LIMIT 1
		)
		UPDATE billing_settlement_jobs j SET status='processing', attempts=attempts+1,
			locked_at=now(), updated_at=now()
		FROM selected WHERE j.request_id=selected.request_id
		RETURNING j.request_id, j.event`,
	).Scan(&requestID, &payload)
	if errors.Is(err, pgx.ErrNoRows) {
		return false, tx.Commit(ctx)
	}
	if err != nil {
		return false, err
	}
	if err := tx.Commit(ctx); err != nil {
		return false, err
	}
	var event domain.UsageEvent
	if err := json.Unmarshal(payload, &event); err != nil {
		s.retrySettlement(ctx, requestID, fmt.Errorf("decode settlement event: %w", err))
		return true, err
	}
	jobCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
	err = s.settle(jobCtx, event)
	cancel()
	if err != nil {
		s.retrySettlement(ctx, requestID, err)
		return true, err
	}
	_, err = s.db.Exec(ctx, `UPDATE billing_settlement_jobs SET status='done', locked_at=NULL,
		last_error='', updated_at=now(), completed_at=now() WHERE request_id=$1`, requestID)
	return true, err
}

func (s *Service) retrySettlement(ctx context.Context, requestID string, cause error) {
	message := cause.Error()
	if len(message) > 1000 {
		message = message[:1000]
	}
	_, _ = s.db.Exec(ctx, `UPDATE billing_settlement_jobs SET status='retry', locked_at=NULL,
		last_error=$2, available_at=now() + make_interval(secs => LEAST(300, power(2, LEAST(attempts, 8))::int)),
		updated_at=now() WHERE request_id=$1`, requestID, message)
}

func (s *Service) appendSettlementSpool(payload []byte) error {
	if s.settlementSpoolPath == "" {
		return errors.New("settlement spool path is not configured")
	}
	s.spoolMu.Lock()
	defer s.spoolMu.Unlock()
	if err := os.MkdirAll(filepath.Dir(s.settlementSpoolPath), 0o700); err != nil {
		return err
	}
	file, err := os.OpenFile(s.settlementSpoolPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
	if err != nil {
		return err
	}
	defer file.Close()
	if _, err := file.Write(append(payload, '\n')); err != nil {
		return err
	}
	return file.Sync()
}

func (s *Service) drainSettlementSpool(ctx context.Context) {
	if s.settlementSpoolPath == "" {
		return
	}
	s.spoolMu.Lock()
	defer s.spoolMu.Unlock()
	file, err := os.Open(s.settlementSpoolPath)
	if errors.Is(err, os.ErrNotExist) {
		return
	}
	if err != nil {
		return
	}
	var pending [][]byte
	scanner := bufio.NewScanner(file)
	scanner.Buffer(make([]byte, 64*1024), 2<<20)
	for scanner.Scan() {
		line := append([]byte(nil), scanner.Bytes()...)
		var event domain.UsageEvent
		if json.Unmarshal(line, &event) != nil || event.RequestID == "" {
			pending = append(pending, line)
			continue
		}
		if _, err := s.db.Exec(ctx, `UPDATE billing_settlement_jobs SET event=$2, status=CASE WHEN status='done' THEN 'done' ELSE 'pending' END,
			available_at=now(), locked_at=NULL, updated_at=now() WHERE request_id=$1`, event.RequestID, line); err != nil {
			pending = append(pending, line)
		}
	}
	_ = file.Close()
	if scanner.Err() != nil {
		return
	}
	temporary := s.settlementSpoolPath + ".tmp"
	out, err := os.OpenFile(temporary, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
	if err != nil {
		return
	}
	for _, line := range pending {
		_, _ = out.Write(append(line, '\n'))
	}
	_ = out.Sync()
	_ = out.Close()
	_ = os.Rename(temporary, s.settlementSpoolPath)
}

func (s *Service) SettlementQueueStatus(ctx context.Context) (SettlementQueueStatus, error) {
	var result SettlementQueueStatus
	err := s.db.QueryRow(ctx, `SELECT
		count(*) FILTER (WHERE status='awaiting_event'), count(*) FILTER (WHERE status='pending'),
		count(*) FILTER (WHERE status='processing'), count(*) FILTER (WHERE status='retry'),
		min(created_at) FILTER (WHERE status IN ('awaiting_event','pending','processing','retry'))
		FROM billing_settlement_jobs`).Scan(&result.AwaitingEvent, &result.Pending, &result.Processing, &result.Retrying, &result.OldestPending)
	if err != nil {
		return result, err
	}
	if s.settlementSpoolPath != "" {
		s.spoolMu.Lock()
		file, openErr := os.Open(s.settlementSpoolPath)
		if openErr == nil {
			scanner := bufio.NewScanner(file)
			for scanner.Scan() {
				result.SpoolRecords++
			}
			_ = file.Close()
		}
		s.spoolMu.Unlock()
	}
	return result, nil
}

func boolToInt(value bool) int {
	if value {
		return 1
	}
	return 0
}

func meteringStatus(event domain.UsageEvent) string {
	if event.StatusCode < 200 || event.StatusCode >= 300 || !event.Success {
		return "upstream_failed"
	}
	if event.UsageReported {
		return "reported"
	}
	return "missing"
}

func reservationCost(model domain.Model, body []byte, defaultMaxOutput int64, protocols ...domain.Protocol) (int64, error) {
	protocol := domain.ProtocolOpenAI
	if len(protocols) > 0 && protocols[0] != "" {
		protocol = protocols[0]
	}
	maxOutput := int64(0)
	var limits struct {
		MaxTokens           int64 `json:"max_tokens"`
		MaxCompletionTokens int64 `json:"max_completion_tokens"`
		MaxOutputTokens     int64 `json:"max_output_tokens"`
	}
	if protocol != domain.ProtocolOpenAIEmbeddings {
		maxOutput = defaultMaxOutput
		if model.MaxOutputTokens > 0 && (maxOutput == 0 || model.MaxOutputTokens < maxOutput) {
			maxOutput = model.MaxOutputTokens
		}
		if json.Unmarshal(body, &limits) == nil {
			explicitMax := int64(0)
			for _, value := range []int64{limits.MaxTokens, limits.MaxCompletionTokens, limits.MaxOutputTokens} {
				if value > explicitMax {
					explicitMax = value
				}
			}
			if explicitMax > 0 {
				maxOutput = explicitMax
			}
		}
	}
	cacheReservePrice := model.CacheReadPriceMicrosPerMillion
	if model.CacheWritePriceMicrosPerMillion > cacheReservePrice {
		cacheReservePrice = model.CacheWritePriceMicrosPerMillion
	}
	return calculateCost(int64(len(body)), maxOutput, 0, int64(len(body)),
		model.InputPriceMicrosPerMillion, model.OutputPriceMicrosPerMillion,
		0, cacheReservePrice)
}

func usageCost(usage domain.Usage, inputPrice, outputPrice, cacheReadPrice, cacheWritePrice int64) (int64, error) {
	return calculateCost(usage.InputTokens, usage.OutputTokens, usage.CacheReadInputTokens,
		usage.CacheCreationInputTokens, inputPrice, outputPrice, cacheReadPrice, cacheWritePrice)
}

func calculateCost(input, output, cacheRead, cacheWrite, inputPrice, outputPrice, cacheReadPrice, cacheWritePrice int64) (int64, error) {
	return calculateMeteredCost([]meteredCharge{
		{Unit: domain.MeteringUnitToken, Quantity: input, PriceMicros: inputPrice, PerQuantity: microsPerUnit},
		{Unit: domain.MeteringUnitToken, Quantity: output, PriceMicros: outputPrice, PerQuantity: microsPerUnit},
		{Unit: domain.MeteringUnitToken, Quantity: cacheRead, PriceMicros: cacheReadPrice, PerQuantity: microsPerUnit},
		{Unit: domain.MeteringUnitToken, Quantity: cacheWrite, PriceMicros: cacheWritePrice, PerQuantity: microsPerUnit},
	})
}

type meteredCharge struct {
	Unit        domain.MeteringUnit
	Quantity    int64
	PriceMicros int64
	PerQuantity int64
}

// calculateMeteredCost is the common fixed-point primitive for token, image,
// and duration pricing. Token rates use PerQuantity=1_000_000; image and second
// rates can use PerQuantity=1 without changing wallet or ledger arithmetic.
func calculateMeteredCost(charges []meteredCharge) (int64, error) {
	byScale := make(map[int64]*big.Int)
	for _, charge := range charges {
		if charge.Unit != domain.MeteringUnitToken && charge.Unit != domain.MeteringUnitImage && charge.Unit != domain.MeteringUnitSecond {
			return 0, fmt.Errorf("unsupported metering unit %q", charge.Unit)
		}
		if charge.Quantity < 0 || charge.PriceMicros < 0 || charge.PerQuantity <= 0 {
			return 0, errors.New("metering quantity, price, or scale is invalid")
		}
		if charge.Quantity == 0 || charge.PriceMicros == 0 {
			continue
		}
		component := new(big.Int).Mul(big.NewInt(charge.Quantity), big.NewInt(charge.PriceMicros))
		if byScale[charge.PerQuantity] == nil {
			byScale[charge.PerQuantity] = new(big.Int)
		}
		byScale[charge.PerQuantity].Add(byScale[charge.PerQuantity], component)
	}
	total := new(big.Int)
	for scale, numerator := range byScale {
		numerator.Add(numerator, big.NewInt(scale-1))
		numerator.Div(numerator, big.NewInt(scale))
		total.Add(total, numerator)
	}
	if total.Sign() == 0 {
		return 0, nil
	}
	if !total.IsInt64() {
		return 0, errors.New("calculated charge exceeds supported range")
	}
	return total.Int64(), nil
}

func collectibleCharge(actualCost, balance, held, reservation int64) (int64, error) {
	if actualCost < 0 || balance < 0 || held < 0 || reservation < 0 || held > balance || reservation > held {
		return 0, errors.New("wallet reservation invariant violated")
	}
	spendable := balance - (held - reservation)
	if actualCost > spendable {
		return spendable, nil
	}
	return actualCost, nil
}

func randomHex(bytes int) string {
	buffer := make([]byte, bytes)
	if _, err := rand.Read(buffer); err != nil {
		return fmt.Sprintf("%08x", time.Now().UnixNano())[:bytes*2]
	}
	return hex.EncodeToString(buffer)
}

func randomLetters(length int) string {
	const letters = "abcdefghijklmnopqrstuvwxyz"
	buffer := make([]byte, length)
	if _, err := rand.Read(buffer); err != nil {
		return strings.Repeat("a", length)
	}
	for i := range buffer {
		buffer[i] = letters[int(buffer[i])%len(letters)]
	}
	return string(buffer)
}

func normalizeDescription(value string) string {
	value = strings.TrimSpace(value)
	if len(value) > 240 {
		value = value[:240]
	}
	return value
}