<?php
// SPDX-License-Identifier: GPL-3.0-or-later
declare(strict_types=1);

namespace Aiwend\Usage;

use Aiwend\Db\Connection;
use Aiwend\Provider\UsageDetail;
use Aiwend\Support\Clock;
use Aiwend\Support\OptionStore;

/**
 * Atomic spend controls (B04, ADR-07).
 *
 * Units are integers (token units until pricing is configured). Lifecycle:
 *
 *   reserved ──dispatch──▶ dispatched ──settle──▶ settled
 *      │                        │
 *      └─release─▶ released     └─timeout/unknown─▶ uncertain ──reconcile──▶ settled
 *
 * A reservation is keyed by (site, request ID); repeating a request ID never
 * reserves twice. Released funds return only from "reserved" (pre-dispatch).
 * Uncertain reservations stay held until reconciled; nothing re-dispatches them.
 *
 * Every settled amount records its basis (Reservation::BASIS_*): provider-reported,
 * administrator-entered from provider records, a conservative estimate (actual unknown),
 * simulated, or not dispatched. Only the first two describe actual usage. An estimate can
 * be replaced once by entered usage. Units are tokens, not money; nothing here is matched
 * against provider invoices automatically.
 */
final class Ledger
{
    public const KILL_SWITCH_OPTION = 'aiwend_spend_halted';

    public function __construct(
        private Connection $db,
        private OptionStore $options,
        private Clock $clock,
    ) {
    }

    public function setLimit(int $siteId, string $bucket, int $limitUnits): void
    {
        $this->assertBucket($bucket);
        if ($limitUnits < 0) {
            throw new \InvalidArgumentException('Limit must not be negative.');
        }
        $now = $this->clock->now();
        $updated = $this->db->execute(
            'UPDATE {budgets} SET limit_units = ?, updated_at = ? WHERE site_id = ? AND bucket = ?',
            [$limitUnits, $now, $siteId, $bucket]
        );
        if ($updated === 0 && !$this->bucketExists($siteId, $bucket)) {
            try {
                $this->db->insert('budgets', [
                    'site_id' => $siteId, 'bucket' => $bucket, 'limit_units' => $limitUnits,
                    'reserved_units' => 0, 'spent_units' => 0, 'updated_at' => $now,
                ]);
            } catch (\Throwable $e) {
                if (!$this->db->isUniqueViolation($e)) {
                    throw $e;
                }
                // A concurrent writer created it; apply our limit.
                $this->db->execute('UPDATE {budgets} SET limit_units = ? WHERE site_id = ? AND bucket = ?', [$limitUnits, $siteId, $bucket]);
            }
        }
    }

    /**
     * Creates the bucket with $defaultLimit only if it does not exist yet. A closure is called (once) only when the bucket has to
     * be created, so the common case of an existing bucket costs a single indexed lookup.
     *
     * @param int|\Closure(): int $defaultLimit
     */
    public function ensureBucket(int $siteId, string $bucket, int|\Closure $defaultLimit): void
    {
        if (!$this->bucketExists($siteId, $bucket)) {
            $defaultLimit = $defaultLimit instanceof \Closure ? (int) $defaultLimit() : $defaultLimit;
            if ($defaultLimit < 0) {
                throw new \InvalidArgumentException('Limit must not be negative.');
            }
            $this->assertBucket($bucket);
            // Phase 86: create-if-absent in ONE statement. The earlier UPDATE-then-INSERT (setLimit) took a gap lock first, and two
            // concurrent first messages in different sessions deadlocked on the budgets index (InnoDB error 1213 -> HTTP 500).
            $this->db->insertIfAbsent('budgets', [
                'site_id' => $siteId, 'bucket' => $bucket, 'limit_units' => $defaultLimit,
                'reserved_units' => 0, 'spent_units' => 0, 'updated_at' => $this->clock->now(),
            ], ['site_id', 'bucket']);
        }
    }

    /** @return array{limit:int, reserved:int, spent:int, available:int}|null */
    public function balance(int $siteId, string $bucket): ?array
    {
        $row = $this->db->selectOne('SELECT limit_units, reserved_units, spent_units FROM {budgets} WHERE site_id = ? AND bucket = ?', [$siteId, $bucket]);
        if ($row === null) {
            return null;
        }
        $limit = (int) $row['limit_units'];
        $reserved = (int) $row['reserved_units'];
        $spent = (int) $row['spent_units'];
        return ['limit' => $limit, 'reserved' => $reserved, 'spent' => $spent, 'available' => $limit - $reserved - $spent];
    }

    public function halt(bool $halted): void
    {
        $this->options->set(self::KILL_SWITCH_OPTION, $halted);
    }

    public function isHalted(): bool
    {
        return (bool) $this->options->get(self::KILL_SWITCH_OPTION, false);
    }

    /**
     * Atomically reserves $units against every bucket, or none.
     *
     * @param list<string> $buckets
     * @throws SpendingHaltedException|BudgetExceededException
     */
    public function reserve(int $siteId, string $requestId, array $buckets, int $units): Reservation
    {
        if ($this->isHalted()) {
            throw new SpendingHaltedException('New paid requests are paused by the site administrator.');
        }
        if ($units <= 0) {
            throw new \InvalidArgumentException('Reservation must be positive.');
        }
        if ($buckets === []) {
            throw new \InvalidArgumentException('At least one budget bucket is required.');
        }
        foreach ($buckets as $b) {
            $this->assertBucket($b);
        }
        $buckets = self::lockOrder($buckets);
        $existing = $this->find($siteId, $requestId);
        if ($existing !== null) {
            return $existing->asDuplicate();
        }

        $now = $this->clock->now();
        try {
            $this->db->transaction(function () use ($siteId, $requestId, $buckets, $units, $now): void {
                // Phase 89: the reservation row first, then the bucket rows (client, conversation, site). settle() and release() also
                // take the reservation row before the bucket rows, so every transaction locks in one order (reservation, then
                // buckets in lockOrder()) and a reserve can no longer wait on a settle that waits on it. A repeated request ID now fails
                // on the unique key before any bucket is touched. Everything is still one transaction: a refusal rolls the row back.
                $this->db->insert('reservations', [
                    'site_id' => $siteId,
                    'request_id' => $requestId,
                    'buckets' => json_encode(array_values($buckets), JSON_THROW_ON_ERROR),
                    'reserved_units' => $units,
                    'settled_units' => null,
                    'state' => Reservation::RESERVED,
                    'provider_ref' => null,
                    'note' => null,
                    'created_at' => $now,
                    'updated_at' => $now,
                ]);
                foreach ($buckets as $bucket) {
                    $ok = $this->db->execute(
                        'UPDATE {budgets} SET reserved_units = reserved_units + ?, updated_at = ?
                         WHERE site_id = ? AND bucket = ? AND limit_units - reserved_units - spent_units >= ?',
                        [$units, $now, $siteId, $bucket, $units]
                    );
                    if ($ok !== 1) {
                        // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
                        throw new BudgetExceededException($bucket);
                    }
                }
            });
        } catch (BudgetExceededException $e) {
            throw $e;
        } catch (\Throwable $e) {
            // A concurrent request with the same ID won; our bucket increments were rolled back.
            if ($this->db->isUniqueViolation($e)) {
                $winner = $this->find($siteId, $requestId);
                if ($winner !== null) {
                    return $winner->asDuplicate();
                }
            }
            throw $e;
        }
        return $this->find($siteId, $requestId) ?? throw new \LogicException('Reservation vanished.');
    }

    /**
     * The one order in which every reservation locks its bucket rows: client, then conversation, then anything else, the shared
     * site bucket last. A single global order means two concurrent reservations can never wait on each other in a cycle (no
     * deadlock between buckets), and the hot site row is locked for the shortest time. Equal ranks keep the caller's order.
     *
     * @param list<string> $buckets
     * @return list<string>
     */
    public static function lockOrder(array $buckets): array
    {
        $rank = static fn(string $b): int => match (true) {
            str_starts_with($b, 'client:') => 0,
            str_starts_with($b, 'session:') => 1,
            $b === 'site' => 9,
            default => 5,
        };
        $buckets = array_values(array_unique($buckets));
        usort($buckets, static fn(string $a, string $b): int => $rank($a) <=> $rank($b));
        return $buckets;
    }

    /**
     * Retention for per-client daily buckets: deletes buckets that nothing holds (no reserved units) and that were last touched
     * before $before, at most $limit rows per call. A bucket with a reservation in flight is never removed. Returns the number
     * deleted. Never touches site or conversation buckets.
     */
    public function purgeClientBuckets(int $siteId, int $before, int $limit = 500): int
    {
        $ids = array_map(
            static fn(array $r): int => (int) $r['id'],
            $this->db->select(
                "SELECT id FROM {budgets} WHERE site_id = ? AND bucket LIKE 'client:%' AND reserved_units = 0 AND updated_at < ? ORDER BY id LIMIT ?",
                [$siteId, $before, max(1, $limit)]
            )
        );
        if ($ids === []) {
            return 0;
        }
        $in = implode(', ', array_fill(0, count($ids), '?'));
        return $this->db->execute("DELETE FROM {budgets} WHERE site_id = ? AND bucket LIKE 'client:%' AND reserved_units = 0 AND id IN ($in)", array_merge([$siteId], $ids));
    }

    /** Marks the point after which the provider may have done (billable) work. */
    public function markDispatched(int $siteId, string $requestId): void
    {
        $this->transition($siteId, $requestId, [Reservation::RESERVED], Reservation::DISPATCHED)
            || throw new \LogicException('Only a reserved request can be dispatched.');
    }

    /** Returns funds for a request that was never dispatched. */
    public function release(int $siteId, string $requestId, string $note = ''): bool
    {
        return $this->db->transaction(function () use ($siteId, $requestId, $note): bool {
            $r = $this->find($siteId, $requestId);
            if ($r === null || !$this->transition($siteId, $requestId, [Reservation::RESERVED], Reservation::RELEASED, $note)) {
                return false;
            }
            $this->adjustBuckets($siteId, $r, 0);
            return true;
        });
    }

    /** Provider outcome is unknown after dispatch: keep funds held, never auto-retry. */
    public function markUncertain(int $siteId, string $requestId, string $note): bool
    {
        return $this->transition($siteId, $requestId, [Reservation::DISPATCHED], Reservation::UNCERTAIN, $note);
    }

    /**
     * Charges usage exactly once and records where the amount comes from ($basis, see
     * Reservation::BASIS_*). Returns false if already settled or released.
     */
    /**
     * Settles a request with the usage it consumed (internal units). The provider's own input and output token counts are recorded
     * only for provider-reported usage, only when both are known and non-negative (they are a measurement, independent of the settled units); in every other case
     * they stay NULL (not measured). A later settlement or reconcile never changes a stored measurement (exactly once, by state).
     */
    public function settle(int $siteId, string $requestId, int $actualUnits, ?string $providerRef = null, string $basis = Reservation::BASIS_REPORTED, ?int $providerInputTokens = null, ?int $providerOutputTokens = null, ?UsageDetail $detail = null): bool
    {
        if ($actualUnits < 0) {
            throw new \InvalidArgumentException('Usage must not be negative.');
        }
        // The split is the provider's own measurement, independent of the amount settled: stored only for provider-reported usage when
        // both counts were given and are non-negative (never invented, never from an estimate). It never feeds the limits.
        $measured = $basis === Reservation::BASIS_REPORTED && $providerInputTokens !== null && $providerOutputTokens !== null
            && $providerInputTokens >= 0 && $providerOutputTokens >= 0;
        $in = $measured ? $providerInputTokens : null;
        $out = $measured ? $providerOutputTokens : null;
        // The price details belong to the split they describe: kept only with it, and a count that cannot be right is dropped, not stored.
        $d = $measured && $detail !== null && in_array($detail->provider, ['openai', 'anthropic', 'gemini', 'deepseek', 'xai', 'openrouter', 'groq'], true)
            ? UsageDetail::from($detail->provider, $detail->servedModel, (int) $in, (int) $out, $detail->cachedInputTokens, $detail->reasoningTokens, $detail->otherTokens, $detail->processingTier)
            : null;
        return $this->db->transaction(function () use ($siteId, $requestId, $actualUnits, $providerRef, $basis, $in, $out, $d): bool {
            $r = $this->find($siteId, $requestId);
            if ($r === null) {
                return false;
            }
            $changed = $this->db->execute(
                'UPDATE {reservations} SET state = ?, settled_units = ?, usage_basis = ?, provider_ref = ?, provider_input_tokens = ?, provider_output_tokens = ?, provider = COALESCE(?, provider), served_model = ?, provider_cached_input_tokens = ?, provider_reasoning_tokens = ?, provider_other_tokens = ?, processing_tier = ?, updated_at = ?
                 WHERE site_id = ? AND request_id = ? AND state IN (?, ?, ?)',
                [Reservation::SETTLED, $actualUnits, $basis, $providerRef, $in, $out, $d?->provider, $d?->servedModel, $d?->cachedInputTokens, $d?->reasoningTokens, $d?->otherTokens, $d?->processingTier, $this->clock->now(), $siteId, $requestId,
                 Reservation::RESERVED, Reservation::DISPATCHED, Reservation::UNCERTAIN]
            );
            if ($changed !== 1) {
                return false;
            }
            $this->adjustBuckets($siteId, $r, $actualUnits);
            return true;
        });
    }

    /**
     * Resolves a request whose usage is not confirmed. Supported recovery (there is no
     * automatic matching against provider invoices):
     *   - uncertain, $actualUnits null: charge the held reservation as a conservative ESTIMATE
     *     (basis estimated_unconfirmed); it counts against the limit but is not confirmed usage;
     *   - uncertain, $actualUnits given: the administrator entered usage from the provider's
     *     records (basis administrator_entered);
     *   - settled as an estimate, $actualUnits given: replace the estimate ONCE; budgets change
     *     by the difference only, and the original estimate stays in estimate_units.
     * Anything else (already confirmed, released, in progress) returns false and changes nothing.
     */
    public function reconcile(int $siteId, string $requestId, ?int $actualUnits, ?string $providerRef = null): bool
    {
        if ($actualUnits !== null && $actualUnits < 0) {
            throw new \InvalidArgumentException('Usage must not be negative.');
        }
        $r = $this->find($siteId, $requestId);
        if ($r === null) {
            return false;
        }
        if ($r->state === Reservation::UNCERTAIN) {
            if ($actualUnits !== null) {
                return $this->settle($siteId, $requestId, $actualUnits, $providerRef, Reservation::BASIS_ADMIN_ENTERED);
            }
            $ok = $this->settle($siteId, $requestId, $r->reservedUnits, $providerRef, Reservation::BASIS_ESTIMATED);
            if ($ok) {
                $this->db->execute('UPDATE {reservations} SET note = ?, estimate_units = reserved_units WHERE site_id = ? AND request_id = ?', ['charged-reserved-estimate', $siteId, $requestId]);
            }
            return $ok;
        }
        if ($r->state !== Reservation::SETTLED || $r->usageBasis !== Reservation::BASIS_ESTIMATED || $actualUnits === null) {
            return false;
        }
        return $this->db->transaction(function () use ($siteId, $requestId, $actualUnits, $providerRef, $r): bool {
            // Conditional on the estimate still being in place: a repeated or concurrent adjustment matches 0 rows.
            $changed = $this->db->execute(
                'UPDATE {reservations} SET settled_units = ?, usage_basis = ?, provider_ref = COALESCE(?, provider_ref), note = ?, updated_at = ?
                 WHERE site_id = ? AND request_id = ? AND state = ? AND usage_basis = ? AND settled_units = ?',
                [$actualUnits, Reservation::BASIS_ADMIN_ENTERED, $providerRef, 'adjusted-from-estimate', $this->clock->now(),
                 $siteId, $requestId, Reservation::SETTLED, Reservation::BASIS_ESTIMATED, (int) $r->settledUnits]
            );
            if ($changed !== 1) {
                return false;
            }
            foreach (self::lockOrder($r->buckets) as $bucket) {
                $this->db->execute(
                    'UPDATE {budgets} SET spent_units = spent_units + ?, updated_at = ? WHERE site_id = ? AND bucket = ?',
                    [$actualUnits - (int) $r->settledUnits, $this->clock->now(), $siteId, $bucket]
                );
            }
            return true;
        });
    }

    /**
     * Chat/site usage by state and basis (indexing reservations on the separate embedding bucket are summarised by EmbeddingUsage), so an estimate is never shown as confirmed usage.
     * Keys: in_progress (reserved), uncertain (held, pending reconciliation), provider_reported,
     * administrator_entered, estimated_unconfirmed, simulated, not_dispatched, provider_rejected, legacy,
     * released (units returned), adjusted (estimates later replaced; units = net change).
     *
     * @return array<string, array{requests:int, units:int}>
     */
    public function usageSummary(int $siteId): array
    {
        $keys = ['in_progress', 'uncertain', Reservation::BASIS_REPORTED, Reservation::BASIS_ADMIN_ENTERED, Reservation::BASIS_ESTIMATED,
            Reservation::BASIS_SIMULATED, Reservation::BASIS_NOT_DISPATCHED, Reservation::BASIS_REJECTED, Reservation::BASIS_LEGACY, 'released', 'adjusted'];
        $out = array_fill_keys($keys, ['requests' => 0, 'units' => 0]);
        $rows = $this->db->select(
            'SELECT state, usage_basis, COUNT(*) AS n, SUM(reserved_units) AS reserved, SUM(COALESCE(settled_units, 0)) AS settled
             FROM {reservations} WHERE site_id = ? AND buckets <> ? GROUP BY state, usage_basis',
            [$siteId, EmbeddingUsage::BUCKETS_JSON]
        );
        foreach ($rows as $row) {
            $basis = (string) ($row['usage_basis'] ?? '');
            [$key, $units] = match ((string) $row['state']) {
                Reservation::RESERVED, Reservation::DISPATCHED => ['in_progress', (int) $row['reserved']],
                Reservation::UNCERTAIN => ['uncertain', (int) $row['reserved']],
                Reservation::RELEASED => ['released', (int) $row['reserved']],
                default => [isset($out[$basis]) ? $basis : Reservation::BASIS_LEGACY, (int) $row['settled']],
            };
            $out[$key]['requests'] += (int) $row['n'];
            $out[$key]['units'] += $units;
        }
        $adj = $this->db->selectOne(
            "SELECT COUNT(*) AS n, SUM(settled_units - estimate_units) AS delta FROM {reservations} WHERE site_id = ? AND note = 'adjusted-from-estimate'",
            [$siteId]
        );
        $out['adjusted'] = ['requests' => (int) ($adj['n'] ?? 0), 'units' => (int) ($adj['delta'] ?? 0)];
        return $out;
    }

    /** The model identifier as the request carried it, or null when it is not a plain identifier (stored as unknown, never altered). */
    private static function cleanModel(?string $model): ?string
    {
        return $model !== null && preg_match('/^[A-Za-z0-9][A-Za-z0-9._:\/@+-]{0,190}$/D', $model) === 1 ? $model : null;
    }

    /** Records where a reserved request came from (assistant, connection, simulated or real). */
    public function attribute(int $siteId, string $requestId, string $origin, ?int $botId, ?int $connectionId, ?string $requestModel = null): bool
    {
        if (!in_array($origin, [Reservation::ORIGIN_SIMULATED, Reservation::ORIGIN_PROVIDER, Reservation::ORIGIN_TEST, Reservation::ORIGIN_CONTENT_STUDIO, Reservation::ORIGIN_WOOCOMMERCE_AI, Reservation::ORIGIN_AI_FORMS], true)) {
            throw new \InvalidArgumentException('Invalid usage origin.');
        }
        // The provider is known from the connection the request was sent through, so a request that later fails or stays uncertain is still attributed
        // (F7). Only the provider name of an existing connection of this site is read; the served model stays unknown until a provider answer returns it.
        // One statement, as before: the provider name comes from a sub-select on this site's connection (NULL for a simulated request or an unknown connection).
        $lookup = $origin === Reservation::ORIGIN_SIMULATED ? null : $connectionId;
        return $this->db->execute(
            'UPDATE {reservations} SET origin = ?, bot_id = ?, connection_id = ?, request_model = ?,
                provider = (SELECT c.provider FROM {connections} c WHERE c.id = ? AND c.site_id = ? AND c.provider IN (\'openai\', \'anthropic\', \'gemini\'))
             WHERE site_id = ? AND request_id = ?',
            [$origin, $botId, $connectionId, self::cleanModel($requestModel), $lookup, $siteId, $siteId, $requestId]
        ) === 1;
    }

    /**
     * Usage grouped by origin, connection and assistant since $since (0 = all time). Units are the
     * settled amount, or the held amount for requests still open or uncertain; they are Aiwend's
     * accounting units (provider-reported tokens where available), never a currency.
     *
     * @return list<array{origin:?string, connection_id:?int, bot_id:?int, state:string, basis:?string, requests:int, units:int}>
     */
    public function breakdown(int $siteId, int $since = 0): array
    {
        $rows = $this->db->select(
            'SELECT origin, connection_id, bot_id, state, usage_basis, COUNT(*) AS n, SUM(reserved_units) AS reserved, SUM(COALESCE(settled_units, 0)) AS settled,
                    COUNT(provider_input_tokens) AS split_n, SUM(provider_input_tokens) AS tok_in, SUM(provider_output_tokens) AS tok_out
             FROM {reservations} WHERE site_id = ? AND created_at >= ? AND buckets <> ? GROUP BY origin, connection_id, bot_id, state, usage_basis',
            [$siteId, $since, EmbeddingUsage::BUCKETS_JSON]
        );
        return array_map(static fn(array $r): array => [
            'origin' => $r['origin'] === null ? null : (string) $r['origin'],
            'connection_id' => $r['connection_id'] === null ? null : (int) $r['connection_id'],
            'bot_id' => $r['bot_id'] === null ? null : (int) $r['bot_id'],
            'state' => (string) $r['state'],
            'basis' => $r['usage_basis'] === null ? null : (string) $r['usage_basis'],
            'requests' => (int) $r['n'],
            'units' => in_array($r['state'], [Reservation::SETTLED], true) ? (int) $r['settled'] : (int) $r['reserved'],
            'with_breakdown' => (int) $r['split_n'],
            'provider_input' => (int) $r['split_n'] > 0 ? (int) $r['tok_in'] : null,
            'provider_output' => (int) $r['split_n'] > 0 ? (int) $r['tok_out'] : null,
        ], $rows);
    }

    /**
     * Usage grouped by the model identifier the requests carried (null = not recorded), since $since. Units are the settled amount
     * (or the held amount for requests still open or uncertain); provider tokens only from requests that carry a breakdown.
     *
     * @return list<array{model:?string, requests:int, units:int, with_breakdown:int, provider_input:?int, provider_output:?int}>
     */
    public function modelBreakdown(int $siteId, int $since = 0): array
    {
        $rows = $this->db->select(
            'SELECT request_model, state, COUNT(*) AS n, SUM(reserved_units) AS reserved, SUM(COALESCE(settled_units, 0)) AS settled,
                    COUNT(provider_input_tokens) AS split_n, SUM(provider_input_tokens) AS tok_in, SUM(provider_output_tokens) AS tok_out
             FROM {reservations} WHERE site_id = ? AND created_at >= ? AND buckets <> ? GROUP BY request_model, state',
            [$siteId, $since, EmbeddingUsage::BUCKETS_JSON]
        );
        $out = [];
        foreach ($rows as $r) {
            $key = $r['request_model'] === null ? '' : (string) $r['request_model'];
            $out[$key] ??= ['model' => $r['request_model'] === null ? null : (string) $r['request_model'], 'requests' => 0, 'units' => 0, 'with_breakdown' => 0, 'provider_input' => null, 'provider_output' => null];
            $out[$key]['requests'] += (int) $r['n'];
            $out[$key]['units'] += $r['state'] === Reservation::SETTLED ? (int) $r['settled'] : (int) $r['reserved'];
            if ((int) $r['split_n'] > 0) {
                $out[$key]['with_breakdown'] += (int) $r['split_n'];
                $out[$key]['provider_input'] = (int) ($out[$key]['provider_input'] ?? 0) + (int) $r['tok_in'];
                $out[$key]['provider_output'] = (int) ($out[$key]['provider_output'] ?? 0) + (int) $r['tok_out'];
            }
        }
        uasort($out, static fn(array $a, array $b): int => $b['units'] <=> $a['units']);
        return array_values($out);
    }

    /**
     * Totals for consecutive periods (newest first, as UsageSeries builds them) in ONE bounded statement: the periods are decided by
     * the caller from the site's calendar, so a day or a month is never assumed to be a fixed number of seconds. Embedding work is
     * not included (it has its own accounting). Provider tokens are null when no request of the period carries a breakdown.
     *
     * @param list<array{start:int, end:int}> $periods
     * @return list<array{requests:int, units:int, units_by_basis:array<string,int>, held_units:int, unresolved_requests:int, in_progress_requests:int, reported_requests:int, reported_with_breakdown:int, provider_input_tokens:?int, provider_output_tokens:?int}>
     */
    public function periodTotals(int $siteId, array $periods): array
    {
        $blank = static fn(): array => ['requests' => 0, 'units' => 0, 'units_by_basis' => [], 'held_units' => 0, 'unresolved_requests' => 0, 'in_progress_requests' => 0,
            'reported_requests' => 0, 'reported_with_breakdown' => 0, 'provider_input_tokens' => null, 'provider_output_tokens' => null];
        $totals = array_map($blank, $periods);
        if ($periods === []) {
            return [];
        }
        $case = 'CASE';
        $params = [];
        foreach ($periods as $i => $p) {
            $case .= ' WHEN created_at >= ? THEN ' . (int) $i;
            $params[] = (int) $p['start'];
        }
        $case .= ' END';
        $newest = $periods[0];
        $oldest = $periods[count($periods) - 1];
        $rows = $this->db->select(
            "SELECT $case AS period, state, usage_basis, COUNT(*) AS n, SUM(reserved_units) AS reserved, SUM(COALESCE(settled_units, 0)) AS settled,
                    COUNT(provider_input_tokens) AS split_n, SUM(provider_input_tokens) AS tok_in, SUM(provider_output_tokens) AS tok_out
             FROM {reservations} WHERE site_id = ? AND buckets <> ? AND created_at >= ? AND created_at < ? GROUP BY period, state, usage_basis",
            array_merge($params, [$siteId, EmbeddingUsage::BUCKETS_JSON, (int) $oldest['start'], (int) $newest['end']])
        );
        foreach ($rows as $r) {
            $i = (int) $r['period'];
            if (!isset($totals[$i])) {
                continue;
            }
            $t = &$totals[$i];
            $n = (int) $r['n'];
            $t['requests'] += $n;
            switch ((string) $r['state']) {
                case Reservation::SETTLED:
                    $basis = (string) ($r['usage_basis'] ?? '') !== '' ? (string) $r['usage_basis'] : Reservation::BASIS_LEGACY;
                    $units = (int) $r['settled'];
                    $t['units'] += $units;
                    $t['units_by_basis'][$basis] = ($t['units_by_basis'][$basis] ?? 0) + $units;
                    if ($basis === Reservation::BASIS_REPORTED) {
                        $t['reported_requests'] += $n;
                        if ((int) $r['split_n'] > 0) {
                            $t['reported_with_breakdown'] += (int) $r['split_n'];
                            $t['provider_input_tokens'] = (int) ($t['provider_input_tokens'] ?? 0) + (int) $r['tok_in'];
                            $t['provider_output_tokens'] = (int) ($t['provider_output_tokens'] ?? 0) + (int) $r['tok_out'];
                        }
                    }
                    break;
                case Reservation::UNCERTAIN:
                    $t['held_units'] += (int) $r['reserved'];
                    $t['unresolved_requests'] += $n;
                    break;
                case Reservation::RESERVED:
                case Reservation::DISPATCHED:
                    $t['held_units'] += (int) $r['reserved'];
                    $t['in_progress_requests'] += $n;
                    break;
                default:
                    break; // released: nothing was used
            }
            unset($t);
        }
        return $totals;
    }

    /**
     * What recent requests looked like, from real reservation rows since $since: how many, and the average size of the amount held for
     * one answer (input estimate plus the longest allowed answer). Requests that never reached a provider are not counted.
     *
     * @return array{requests:int, avg_reserved:?float}
     */
    public function recentProfile(int $siteId, int $since): array
    {
        $row = $this->db->selectOne(
            'SELECT COUNT(*) AS n, AVG(reserved_units) AS avg_reserved FROM {reservations}
             WHERE site_id = ? AND created_at >= ? AND buckets <> ? AND state = ? AND usage_basis NOT IN (?, ?)',
            [$siteId, $since, EmbeddingUsage::BUCKETS_JSON, Reservation::SETTLED, Reservation::BASIS_NOT_DISPATCHED, Reservation::BASIS_REJECTED]
        );
        $n = (int) ($row['n'] ?? 0);
        return ['requests' => $n, 'avg_reserved' => $n > 0 ? (float) $row['avg_reserved'] : null];
    }

    /** @return list<Reservation> settled as unconfirmed estimates (each can still be replaced once) */
    public function estimated(int $siteId, int $limit = 100): array
    {
        $rows = $this->db->select('SELECT * FROM {reservations} WHERE site_id = ? AND state = ? AND usage_basis = ? ORDER BY id LIMIT ?', [$siteId, Reservation::SETTLED, Reservation::BASIS_ESTIMATED, $limit]);
        return array_map([Reservation::class, 'fromRow'], $rows);
    }

    public function find(int $siteId, string $requestId): ?Reservation
    {
        $row = $this->db->selectOne('SELECT * FROM {reservations} WHERE site_id = ? AND request_id = ?', [$siteId, $requestId]);
        return $row === null ? null : Reservation::fromRow($row);
    }

    /** @return list<Reservation> */
    public function uncertain(int $siteId, int $limit = 100): array
    {
        $rows = $this->db->select('SELECT * FROM {reservations} WHERE site_id = ? AND state = ? ORDER BY id LIMIT ?', [$siteId, Reservation::UNCERTAIN, $limit]);
        return array_map([Reservation::class, 'fromRow'], $rows);
    }

    /** Seconds after which a reservation still in `reserved` or `dispatched` is reported as pending longer than expected (observation only, see holdSummary()). */
    public const PENDING_AFTER = 900;
    /** Most rows one summary reads; a larger number is reported as "at least". */
    public const HOLD_SCAN_CAP = 200;

    /**
     * Read-only observation of outstanding holds (Phase 1 visibility; changes nothing and calls no provider).
     *
     * Holding units against the budgets are the states `reserved`, `dispatched` and `uncertain`; `settled` and `released` hold nothing.
     *  - pending: `reserved` or `dispatched` and not updated for $pendingAfter seconds. The longest provider call is 120 s, so 15 minutes is far
     *    beyond a normal request. Age alone does not mean the request failed: it may be a slow request, a process that stopped, or
     *    a request that never left. Nothing here is expired, released or charged.
     *  - uncertain: always unresolved by definition, shown with its age.
     * Units are the reservation's own `reserved_units`, counted once per reservation (a reservation holds the same amount on each of its
     * buckets, so summing per bucket would count it several times). They are Aiwend's local usage units, never provider billing.
     * Age is measured from `updated_at` (the last state change), falling back to `created_at`; a row with neither is counted but has no age.
     * Bounded: one query on the site and state (the `state` index), at most $cap + 1 rows, ordered by id.
     *
     * @return array{pending:array{count:int, units:int, oldest_age:?int, capped:bool}, uncertain:array{count:int, units:int, oldest_age:?int, capped:bool}}
     */
    public function holdSummary(int $siteId, int $pendingAfter = self::PENDING_AFTER, int $cap = self::HOLD_SCAN_CAP): array
    {
        $now = $this->clock->now();
        $cap = max(1, $cap);
        $out = ['pending' => ['count' => 0, 'units' => 0, 'oldest_age' => null, 'capped' => false], 'uncertain' => ['count' => 0, 'units' => 0, 'oldest_age' => null, 'capped' => false]];
        foreach (['pending' => [[Reservation::RESERVED, Reservation::DISPATCHED], true], 'uncertain' => [[Reservation::UNCERTAIN], false]] as $group => [$states, $aged]) {
            $in = implode(', ', array_fill(0, count($states), '?'));
            $rows = $this->db->select(
                "SELECT reserved_units, created_at, updated_at FROM {reservations} WHERE site_id = ? AND state IN ($in)" . ($aged ? ' AND updated_at <= ?' : '') . ' ORDER BY id LIMIT ?',
                array_merge([$siteId], $states, $aged ? [$now - max(0, $pendingAfter)] : [], [$cap + 1])
            );
            $out[$group]['capped'] = count($rows) > $cap;
            foreach (array_slice($rows, 0, $cap) as $row) {
                $out[$group]['count']++;
                $out[$group]['units'] += max(0, (int) ($row['reserved_units'] ?? 0));
                $at = (int) ($row['updated_at'] ?? 0) > 0 ? (int) $row['updated_at'] : (int) ($row['created_at'] ?? 0);
                if ($at > 0) {
                    $age = max(0, $now - $at);
                    $out[$group]['oldest_age'] = max($out[$group]['oldest_age'] ?? 0, $age);
                }
            }
        }
        return $out;
    }

    /**
     * Stale holds an administrator may act on (Phase 137): `reserved` or `dispatched`, not updated for $pendingAfter seconds. Read-only, bounded
     * (at most $limit rows, state index), oldest first. `internal_id` is the reservation row id, the safe correlation value for audit entries.
     *
     * @return list<array{internal_id:int, request_id:string, state:string, units:int, age:?int, origin:?string}>
     */
    public function staleHolds(int $siteId, int $limit = 100, int $pendingAfter = self::PENDING_AFTER): array
    {
        $now = $this->clock->now();
        $rows = $this->db->select(
            'SELECT id, request_id, state, reserved_units, created_at, updated_at, origin FROM {reservations} WHERE site_id = ? AND state IN (?, ?) AND updated_at <= ? ORDER BY id LIMIT ?',
            [$siteId, Reservation::RESERVED, Reservation::DISPATCHED, $now - max(0, $pendingAfter), max(1, $limit)]
        );
        return array_map(static function (array $r) use ($now): array {
            $at = (int) ($r['updated_at'] ?? 0) > 0 ? (int) $r['updated_at'] : (int) ($r['created_at'] ?? 0);
            return ['internal_id' => (int) $r['id'], 'request_id' => (string) $r['request_id'], 'state' => (string) $r['state'], 'units' => max(0, (int) $r['reserved_units']),
                'age' => $at > 0 ? max(0, $now - $at) : null, 'origin' => isset($r['origin']) ? (string) $r['origin'] : null];
        }, $rows);
    }

    public const RECOVER_DONE = 'done';
    public const RECOVER_NOT_FOUND = 'not_found';
    /** The record is younger than the threshold at execution time (age is rechecked inside the conditional update). */
    public const RECOVER_NOT_STALE = 'not_stale';
    /** The record is no longer in the state the action needs: settled, released, uncertain, or changed by someone else a moment ago. */
    public const RECOVER_CONFLICT = 'conflict';

    /**
     * Administrator reclassification: a stale `dispatched` reservation becomes `uncertain`. This is a LABEL, not a settlement: the held units, the
     * buckets and the amount spent do not change, nothing is charged or released, and the provider is not contacted. Only an explicit call does it
     * (no background process). One conditional UPDATE decides: it matches only `dispatched` rows not updated for $pendingAfter seconds, so a
     * request settled, released, already reclassified or touched again in the meantime matches nothing and a repeat or a race changes nothing.
     * A worker that is in fact still running is not harmed: settle() accepts `uncertain` and records the provider's usage exactly once.
     *
     * @return array{result:string, from:?string, units:int, age:?int, internal_id:?int}
     */
    public function reclassifyStale(int $siteId, string $requestId, int $pendingAfter = self::PENDING_AFTER): array
    {
        return $this->recover($siteId, $requestId, Reservation::DISPATCHED, Reservation::UNCERTAIN, 'stale-after-dispatch', $pendingAfter, false);
    }

    /**
     * Administrator release of a stale `reserved` reservation. State `reserved` is proof that nothing was sent: in every module the
     * `reserved` to `dispatched` transition (markDispatched) happens BEFORE the request to the provider is made and a request that cannot make
     * the transition never makes the call (pinned by a source-order test), so a row still `reserved` never reached the provider, whatever its age. The age
     * threshold only keeps this away from requests still being prepared. The units go back to the buckets; nothing is added to spent.
     * Same conditional-update guarantees as reclassifyStale().
     *
     * @return array{result:string, from:?string, units:int, age:?int, internal_id:?int}
     */
    public function releaseStaleReserved(int $siteId, string $requestId, int $pendingAfter = self::PENDING_AFTER): array
    {
        return $this->recover($siteId, $requestId, Reservation::RESERVED, Reservation::RELEASED, 'stale-never-dispatched', $pendingAfter, true);
    }

    /** @return array{result:string, from:?string, units:int, age:?int, internal_id:?int} */
    private function recover(int $siteId, string $requestId, string $from, string $to, string $note, int $pendingAfter, bool $releaseUnits): array
    {
        return $this->db->transaction(function () use ($siteId, $requestId, $from, $to, $note, $pendingAfter, $releaseUnits): array {
            $now = $this->clock->now();
            $row = $this->db->selectOne('SELECT id, state, reserved_units, created_at, updated_at FROM {reservations} WHERE site_id = ? AND request_id = ?', [$siteId, $requestId]);
            if ($row === null) {
                return ['result' => self::RECOVER_NOT_FOUND, 'from' => null, 'units' => 0, 'age' => null, 'internal_id' => null];
            }
            $at = (int) $row['updated_at'] > 0 ? (int) $row['updated_at'] : (int) $row['created_at'];
            $info = ['from' => (string) $row['state'], 'units' => max(0, (int) $row['reserved_units']), 'age' => $at > 0 ? max(0, $now - $at) : null, 'internal_id' => (int) $row['id']];
            if ($row['state'] !== $from) {
                return ['result' => self::RECOVER_CONFLICT] + $info;
            }
            $changed = $this->db->execute(
                'UPDATE {reservations} SET state = ?, note = ?, updated_at = ? WHERE site_id = ? AND request_id = ? AND state = ? AND updated_at <= ?',
                [$to, $note, $now, $siteId, $requestId, $from, $now - max(0, $pendingAfter)]
            );
            if ($changed !== 1) {
                // Nothing matched: either it is younger than the threshold (the common case for a reserved or dispatched row) or someone changed it.
                $again = $this->db->selectOne('SELECT state FROM {reservations} WHERE site_id = ? AND request_id = ?', [$siteId, $requestId]);
                return ['result' => ($again['state'] ?? null) === $from ? self::RECOVER_NOT_STALE : self::RECOVER_CONFLICT] + $info;
            }
            if ($releaseUnits) {
                $r = $this->find($siteId, $requestId);
                if ($r !== null) {
                    $this->adjustBuckets($siteId, $r, 0);
                }
            }
            return ['result' => self::RECOVER_DONE] + $info;
        });
    }

    /** @param list<string> $from */
    private function transition(int $siteId, string $requestId, array $from, string $to, string $note = ''): bool
    {
        $in = implode(', ', array_fill(0, count($from), '?'));
        return $this->db->execute(
            "UPDATE {reservations} SET state = ?, note = ?, updated_at = ? WHERE site_id = ? AND request_id = ? AND state IN ($in)",
            array_merge([$to, $note === '' ? null : mb_substr($note, 0, 191), $this->clock->now(), $siteId, $requestId], $from)
        ) === 1;
    }

    private function adjustBuckets(int $siteId, Reservation $r, int $spent): void
    {
        foreach (self::lockOrder($r->buckets) as $bucket) {
            $this->db->execute(
                'UPDATE {budgets} SET reserved_units = reserved_units - ?, spent_units = spent_units + ?, updated_at = ? WHERE site_id = ? AND bucket = ?',
                [$r->reservedUnits, $spent, $this->clock->now(), $siteId, $bucket]
            );
        }
    }

    private function bucketExists(int $siteId, string $bucket): bool
    {
        return $this->db->selectOne('SELECT id FROM {budgets} WHERE site_id = ? AND bucket = ?', [$siteId, $bucket]) !== null;
    }

    private function assertBucket(string $bucket): void
    {
        if (!preg_match('/^[a-z]+(:[A-Za-z0-9_-]{1,64})?$/', $bucket)) {
            throw new \InvalidArgumentException('Invalid budget bucket.');
        }
    }
}