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

namespace Aiwend\Usage;

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

/**
 * Fail-safe circuit breaker for repeated AMBIGUOUS provider failures (Phase 91a hardening; source-aware since Phase 93).
 *
 * A timeout, reset, 5xx or truncated stream leaves its reservation uncertain (held, never guessed harmless). During a provider outage
 * every visitor message would add one more held reservation. The breaker stops NEW requests to the affected connection before they are
 * dispatched; it never releases, settles or deletes anything, so the accounting of what already happened is untouched.
 *
 * It derives its state from the reservation rows, which are written atomically anyway:
 *  - failures: this connection's reservations that are `uncertain` with a `provider-*` note (not `provider-cancelled`, which is a visitor
 *    pressing Stop) updated within WINDOW seconds. Definitive refusals (authentication, quota, validation: settled at zero) and owner
 *    test messages never appear here.
 *  - in flight: this connection's `reserved`/`dispatched` reservations updated within STALE seconds (the caller's own included).
 * check() runs AFTER the caller inserted its own reservation row, so concurrent requests see each other: a burst can never all pass a
 * count that only some of them could pass (it errs towards refusing, never towards admitting). Cost: two small queries on the `state`
 * index per provider request, at most ROW_CAP rows each, plus one option read; no new table, no lock.
 *
 * Whose failure is it? A shared connection serves every visitor, so one visitor must not be able to stop the others. Every reservation
 * records its buckets: the visitor's daily client bucket is the SOURCE (one address or account, see Rest\ChatController::clientAllowance),
 * the conversation is the finer unit.
 *  - conversation paused: CONVERSATION_THRESHOLD ambiguous failures in WINDOW pause that conversation for COOLDOWN.
 *  - client paused:       CLIENT_THRESHOLD failures pause that client (all its conversations). It is higher because several people can
 *                         share one address; one bad conversation never pauses its neighbours.
 *  - source busy:         a client with more than SOURCE_IN_FLIGHT running requests is refused. Only that client.
 *  - shared breaker:      counts DISTINCT clients and only failures NEWER than the connection's latest proven success. THRESHOLD
 *                         different clients must each have failed since the connection last answered. While honest traffic succeeds,
 *                         failures are disproved and the breaker cannot open; a real outage stops successes and opens it as before. A
 *                         success (any visitor) closes it at once. Each client counts at most SOURCE_IN_FLIGHT towards MAX_IN_FLIGHT.
 *
 * Bounded, documented limits: (1) visitors who cannot be identified apart share one client (declared-proxy misconfiguration) and so
 * one pause; (2) an attacker with THRESHOLD or more distinct identities (addresses, /64 networks, accounts if registration is open)
 * can open the breaker while NO request is succeeding (a quiet site), indistinguishable from an outage, at the cost of one failing
 * request per identity per cooldown; (3) enough identities can fill the in-flight slots, which exist to protect PHP workers; those
 * clear within seconds; (4) intermittent partial outages with interleaved successes do not open the breaker (every client is still
 * bounded by the pauses above).
 *
 *  - closed:    fewer than THRESHOLD failing clients since the last success, in flight at most MAX_IN_FLIGHT. Requests pass.
 *  - open:      THRESHOLD or more and the latest failure less than COOLDOWN seconds ago. Requests are refused.
 *  - half-open: the cooldown elapsed but the failures are still inside WINDOW. One request at a time may pass as a probe; its success
 *               closes the breaker, its failure restarts the cooldown.
 *  - busy:      more than MAX_IN_FLIGHT (counted per client as above) requests running (a hanging provider accumulates in-flight
 *               requests for up to the 60 s timeout before any failure is recorded).
 * A refused request is released by the caller at once and costs the visitor and the site nothing.
 */
final class ProviderBreaker
{
    public const THRESHOLD = 4;
    public const WINDOW = 900;
    public const COOLDOWN = 300;
    public const MAX_IN_FLIGHT = 12;
    public const STALE = 150;
    /** Ambiguous failures in one conversation after which only that conversation is paused. */
    public const CONVERSATION_THRESHOLD = 3;
    /** Ambiguous failures of one client (address or account) after which only that client is paused. */
    public const CLIENT_THRESHOLD = 6;
    /** Requests one client may have in flight; no client counts for more than this towards MAX_IN_FLIGHT. */
    public const SOURCE_IN_FLIGHT = 3;
    /** The proof-of-success stamp is rewritten at most this often (one small option write per connection). */
    private const STAMP_INTERVAL = 5;
    /** Rows read per query. Reaching it is already answer enough. */
    private const ROW_CAP = 400;

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

    /** Called when the provider returned a usable answer for this connection. Throttled; harmless if lost or stale by a few seconds. */
    public function recordSuccess(int $siteId, int $connectionId): void
    {
        if ($this->options === null) {
            return;
        }
        $now = $this->clock->now();
        if ($now - $this->lastSuccess($siteId, $connectionId) >= self::STAMP_INTERVAL) {
            $this->options->set(self::stampKey($siteId, $connectionId), $now);
        }
    }

    /**
     * $source is the caller's client bucket and $conversation its conversation bucket (see sourceOf()). Without them only the
     * connection-wide rules apply.
     *
     * @return array{reason:string, retry_after:int}|null null = the request may be dispatched
     */
    public function check(int $siteId, int $connectionId, ?string $source = null, ?string $conversation = null): ?array
    {
        $now = $this->clock->now();
        $running = $this->db->select(
            "SELECT request_id, buckets FROM {reservations} WHERE site_id = ? AND connection_id = ? AND state IN ('reserved', 'dispatched') AND updated_at >= ? LIMIT " . self::ROW_CAP,
            [$siteId, $connectionId, $now - self::STALE]
        );
        $perSource = [];
        foreach ($running as $r) {
            $key = self::sourceOf((string) $r['buckets'], (string) $r['request_id']);
            $perSource[$key] = ($perSource[$key] ?? 0) + 1;
        }
        // 1. One client cannot hold more than its share of the connection: it is refused, the others are not.
        if ($source !== null && ($perSource[$source] ?? 0) > self::SOURCE_IN_FLIGHT) {
            return ['reason' => 'busy', 'retry_after' => 10];
        }
        // 2. Shared capacity. No client counts for more than SOURCE_IN_FLIGHT.
        $effective = 0;
        foreach ($perSource as $n) {
            $effective += min($n, self::SOURCE_IN_FLIGHT);
        }
        if ($effective > self::MAX_IN_FLIGHT || count($running) >= self::ROW_CAP) {
            return ['reason' => 'busy', 'retry_after' => 10];
        }
        [$clients, $conversations] = $this->failureGroups($siteId, $connectionId, $now);
        // 3. Repeated failures pause the conversation, or (more of them) the client, alone. These use ALL failures in the window:
        //    a success elsewhere does not excuse a visitor who keeps failing.
        foreach ([[$conversation, $conversations, self::CONVERSATION_THRESHOLD], [$source, $clients, self::CLIENT_THRESHOLD]] as [$who, $groups, $limit]) {
            if ($who !== null && isset($groups[$who]) && $groups[$who]['n'] >= $limit && $now - $groups[$who]['last'] < self::COOLDOWN) {
                return ['reason' => 'source', 'retry_after' => max(1, self::COOLDOWN - ($now - $groups[$who]['last']))];
            }
        }
        // 4. The shared breaker: distinct clients whose failure is newer than the connection's latest success.
        $ok = $this->lastSuccess($siteId, $connectionId);
        $count = 0;
        $last = 0;
        foreach ($clients as $c) {
            if ($c['last'] > $ok) {
                $count++;
                $last = max($last, $c['last']);
            }
        }
        if ($count < self::THRESHOLD) {
            return null;
        }
        $elapsed = $now - $last;
        if ($elapsed < self::COOLDOWN) {
            return ['reason' => 'open', 'retry_after' => max(1, self::COOLDOWN - $elapsed)];
        }
        // Half-open: a single probe. The running list includes the caller's own row, so more than one means another request is running.
        return count($running) > 1 ? ['reason' => 'probing', 'retry_after' => 10] : null;
    }

    /**
     * Connections whose breaker is currently open or half-open, for the administrator notice.
     *
     * @return list<array{connection_id:int, failures:int, last:int, open:bool, resumes_at:int}>
     */
    public function tripped(int $siteId): array
    {
        $now = $this->clock->now();
        $rows = $this->db->select(
            "SELECT connection_id, request_id, buckets, updated_at FROM {reservations}
             WHERE site_id = ? AND connection_id IS NOT NULL AND state = 'uncertain' AND note LIKE 'provider-%' AND note <> 'provider-cancelled' AND updated_at >= ?
             ORDER BY updated_at DESC LIMIT " . (self::ROW_CAP * 4),
            [$siteId, $now - self::WINDOW]
        );
        $by = [];
        foreach ($rows as $r) {
            $c = (int) $r['connection_id'];
            $key = self::sourceOf((string) $r['buckets'], (string) $r['request_id']);
            $by[$c]['clients'][$key] = max($by[$c]['clients'][$key] ?? 0, (int) $r['updated_at']);
            $by[$c]['n'] = ($by[$c]['n'] ?? 0) + 1;
        }
        $out = [];
        foreach ($by as $connection => $v) {
            $ok = $this->lastSuccess($siteId, $connection);
            $fresh = array_filter($v['clients'], static fn(int $t): bool => $t > $ok);
            if (count($fresh) >= self::THRESHOLD) {
                $last = max($fresh);
                $out[] = ['connection_id' => $connection, 'failures' => $v['n'], 'last' => $last, 'open' => $now - $last < self::COOLDOWN, 'resumes_at' => $last + self::COOLDOWN];
            }
        }
        return $out;
    }

    /**
     * Ambiguous failures in the window, grouped by client and by conversation.
     *
     * @return array{0:array<string, array{n:int, last:int}>, 1:array<string, array{n:int, last:int}>}
     */
    private function failureGroups(int $siteId, int $connectionId, int $now): array
    {
        $rows = $this->db->select(
            "SELECT request_id, buckets, updated_at FROM {reservations}
             WHERE site_id = ? AND connection_id = ? AND state = 'uncertain' AND note LIKE 'provider-%' AND note <> 'provider-cancelled' AND updated_at >= ?
             ORDER BY updated_at DESC LIMIT " . self::ROW_CAP,
            [$siteId, $connectionId, $now - self::WINDOW]
        );
        $clients = [];
        $conversations = [];
        foreach ($rows as $r) {
            $t = (int) $r['updated_at'];
            $key = self::sourceOf((string) $r['buckets'], (string) $r['request_id']);
            $clients[$key]['n'] = ($clients[$key]['n'] ?? 0) + 1;
            $clients[$key]['last'] = max($clients[$key]['last'] ?? 0, $t);
            $conv = self::conversationOf((string) $r['buckets']);
            if ($conv !== null) {
                $conversations[$conv]['n'] = ($conversations[$conv]['n'] ?? 0) + 1;
                $conversations[$conv]['last'] = max($conversations[$conv]['last'] ?? 0, $t);
            }
        }
        return [$clients, $conversations];
    }

    /** When this connection last returned a usable answer (0 = never recorded). Throttled by recordSuccess, so it can be a few seconds old. */
    public function lastSuccessAt(int $siteId, int $connectionId): int
    {
        return $this->lastSuccess($siteId, $connectionId);
    }

    private function lastSuccess(int $siteId, int $connectionId): int
    {
        return $this->options === null ? 0 : max(0, (int) $this->options->get(self::stampKey($siteId, $connectionId), 0));
    }

    private static function stampKey(int $siteId, int $connectionId): string
    {
        return 'aiwend_breaker_ok_' . $siteId . '_' . $connectionId;
    }

    /**
     * The client a reservation is attributed to: its client bucket, else its conversation, else the request itself (older rows and rows
     * without either bucket count as one client each, so they can only make the breaker more cautious).
     */
    public static function sourceOf(string $bucketsJson, string $requestId): string
    {
        $buckets = json_decode($bucketsJson, true);
        $session = null;
        if (is_array($buckets)) {
            foreach ($buckets as $b) {
                if (is_string($b) && str_starts_with($b, 'client:')) {
                    return $b;
                }
                if ($session === null && is_string($b) && str_starts_with($b, 'session:')) {
                    $session = $b;
                }
            }
        }
        return $session ?? 'request:' . $requestId;
    }

    private static function conversationOf(string $bucketsJson): ?string
    {
        $buckets = json_decode($bucketsJson, true);
        if (is_array($buckets)) {
            foreach ($buckets as $b) {
                if (is_string($b) && str_starts_with($b, 'session:')) {
                    return $b;
                }
            }
        }
        return null;
    }
}