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

namespace Aiwend\Queue;

use Aiwend\Db\Connection;
use Aiwend\Secrets\Redactor;
use Aiwend\Support\Clock;

/**
 * Persistent job queue (B07, ADR-04).
 *
 * - Deduplication: (site, type, idempotency key) is unique; enqueueing again returns the existing job.
 * - Leases: a claimed job carries an owner and expiry. An expired lease (crashed worker)
 *   makes the job claimable again; completion by a stale owner is rejected.
 * - Retries: exponential backoff up to max_attempts, then "dead" (dead-letter).
 * - Module control: jobs of disabled modules are suspended and cannot be enqueued.
 */
final class Queue
{
    public const QUEUED = 'queued';
    public const RUNNING = 'running';
    public const DONE = 'done';
    public const DEAD = 'dead';
    public const SUSPENDED = 'suspended';
    public const CANCELLED = 'cancelled';

    /** @var callable(string): bool */
    private $typeGuard;

    /** @param callable(string): bool|null $typeGuard */
    public function __construct(
        private Connection $db,
        private Clock $clock,
        private Redactor $redactor,
        ?callable $typeGuard = null,
        private int $baseBackoff = 30,
        private int $maxBackoff = 3600,
    ) {
        $this->typeGuard = $typeGuard ?? static fn(string $t): bool => true;
    }

    /** @param array<string, mixed> $payload */
    public function enqueue(int $siteId, string $type, array $payload, string $idempotencyKey, int $delay = 0, int $maxAttempts = 5): int
    {
        if (!preg_match('/^[a-z][a-z0-9_.]{2,99}$/', $type)) {
            throw new \InvalidArgumentException('Invalid job type.');
        }
        if (!($this->typeGuard)($type)) {
            // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- exception messages go to logs and callers, never echoed; public errors are fixed codes
            throw new JobRejectedException("Job type \"$type\" belongs to a disabled module.");
        }
        $key = substr($idempotencyKey, 0, 191);
        $existing = $this->db->selectOne('SELECT id FROM {jobs} WHERE site_id = ? AND type = ? AND idempotency_key = ?', [$siteId, $type, $key]);
        if ($existing !== null) {
            return (int) $existing['id'];
        }
        $now = $this->clock->now();
        try {
            return $this->db->insert('jobs', [
                'site_id' => $siteId,
                'type' => $type,
                'idempotency_key' => $key,
                'payload' => json_encode($payload, JSON_THROW_ON_ERROR),
                'state' => self::QUEUED,
                'attempts' => 0,
                'max_attempts' => max(1, $maxAttempts),
                'lease_owner' => null,
                'lease_expires' => null,
                'next_run_at' => $now + max(0, $delay),
                'last_error' => null,
                'created_at' => $now,
                'updated_at' => $now,
            ]);
        } catch (\Throwable $e) {
            if ($this->db->isUniqueViolation($e)) {
                $row = $this->db->selectOne('SELECT id FROM {jobs} WHERE site_id = ? AND type = ? AND idempotency_key = ?', [$siteId, $type, $key]);
                if ($row !== null) {
                    return (int) $row['id'];
                }
            }
            throw $e;
        }
    }

    /** How many jobs of this type are ready to start (state queued and due; a job waiting out a retry delay is not counted) for this site. Bounded by the (state, next_run_at) index. */
    public function waiting(int $siteId, string $type): int
    {
        return (int) ($this->db->selectOne('SELECT COUNT(*) AS n FROM {jobs} WHERE state = ? AND next_run_at <= ? AND site_id = ? AND type = ?', [self::QUEUED, $this->clock->now(), $siteId, $type])['n'] ?? 0);
    }

    /** The oldest job of this type that has not started yet (state queued), or null. Bounded by the (state, next_run_at) index: finished jobs are not scanned. */
    public function queued(int $siteId, string $type): ?int
    {
        $row = $this->db->selectOne('SELECT id FROM {jobs} WHERE state = ? AND site_id = ? AND type = ? ORDER BY id LIMIT 1', [self::QUEUED, $siteId, $type]);
        return $row === null ? null : (int) $row['id'];
    }

    /**
     * Claims one ready job for $worker. Ready means queued and due, or running
     * with an expired lease (recovery after a crash).
     *
     * @param list<string>|null $types
     */
    public function claim(string $worker, int $leaseSeconds = 120, ?array $types = null): ?Job
    {
        $now = $this->clock->now();
        for ($attempt = 0; $attempt < 5; $attempt++) {
            $params = [self::QUEUED, $now, self::RUNNING, $now];
            $typeSql = '';
            if ($types !== null && $types !== []) {
                $typeSql = ' AND type IN (' . implode(', ', array_fill(0, count($types), '?')) . ')';
                $params = array_merge($params, $types);
            }
            $candidate = $this->db->selectOne(
                "SELECT id, type, state, lease_owner FROM {jobs}
                 WHERE ((state = ? AND next_run_at <= ?) OR (state = ? AND lease_expires < ?))$typeSql
                 ORDER BY next_run_at, id LIMIT 1",
                $params
            );
            if ($candidate === null) {
                return null;
            }
            if (!($this->typeGuard)((string) $candidate['type'])) {
                $this->db->execute('UPDATE {jobs} SET state = ?, lease_owner = NULL, lease_expires = NULL WHERE id = ?', [self::SUSPENDED, $candidate['id']]);
                continue;
            }
            // Compare-and-set on the exact state observed; a racing worker makes this affect 0 rows.
            $claimed = $this->db->execute(
                'UPDATE {jobs} SET state = ?, lease_owner = ?, lease_expires = ?, attempts = attempts + 1, updated_at = ?
                 WHERE id = ? AND state = ? AND (lease_owner IS NULL OR lease_owner = ?) AND ((state = ? AND next_run_at <= ?) OR (state = ? AND lease_expires < ?))',
                [self::RUNNING, $worker, $now + $leaseSeconds, $now, $candidate['id'], $candidate['state'], $candidate['lease_owner'] ?? '',
                 self::QUEUED, $now, self::RUNNING, $now]
            );
            if ($claimed === 1) {
                return $this->get((int) $candidate['id']);
            }
        }
        return null;
    }

    public function complete(Job $job, string $worker): bool
    {
        return $this->db->execute(
            'UPDATE {jobs} SET state = ?, lease_owner = NULL, lease_expires = NULL, updated_at = ? WHERE id = ? AND state = ? AND lease_owner = ?',
            [self::DONE, $this->clock->now(), $job->id, self::RUNNING, $worker]
        ) === 1;
    }

    /** Records a failure. Retryable failures back off; others (or exhausted attempts) go to dead-letter. */
    public function fail(Job $job, string $worker, string $error, bool $retryable = true): bool
    {
        $row = $this->db->selectOne('SELECT attempts, max_attempts FROM {jobs} WHERE id = ?', [$job->id]);
        if ($row === null) {
            return false;
        }
        $attempts = (int) $row['attempts'];
        $dead = !$retryable || $attempts >= (int) $row['max_attempts'];
        $now = $this->clock->now();
        $delay = min($this->maxBackoff, $this->baseBackoff * (2 ** max(0, $attempts - 1)));
        return $this->db->execute(
            'UPDATE {jobs} SET state = ?, lease_owner = NULL, lease_expires = NULL, next_run_at = ?, last_error = ?, updated_at = ?
             WHERE id = ? AND state = ? AND lease_owner = ?',
            [$dead ? self::DEAD : self::QUEUED, $now + ($dead ? 0 : $delay), $this->redactor->summary($error), $now, $job->id, self::RUNNING, $worker]
        ) === 1;
    }

    /** Extends a lease for long-running work. Fails if the lease was lost. */
    public function heartbeat(Job $job, string $worker, int $leaseSeconds = 120): bool
    {
        $now = $this->clock->now();
        return $this->db->execute(
            'UPDATE {jobs} SET lease_expires = ?, updated_at = ? WHERE id = ? AND state = ? AND lease_owner = ? AND lease_expires >= ?',
            [$now + $leaseSeconds, $now, $job->id, self::RUNNING, $worker, $now]
        ) === 1;
    }

    /** @param list<string> $types */
    public function suspendTypes(array $types): int
    {
        if ($types === []) {
            return 0;
        }
        $in = implode(', ', array_fill(0, count($types), '?'));
        return $this->db->execute("UPDATE {jobs} SET state = ?, updated_at = ? WHERE state = ? AND type IN ($in)", array_merge([self::SUSPENDED, $this->clock->now(), self::QUEUED], $types));
    }

    /** @param list<string> $types */
    public function resumeTypes(array $types): int
    {
        if ($types === []) {
            return 0;
        }
        $in = implode(', ', array_fill(0, count($types), '?'));
        return $this->db->execute("UPDATE {jobs} SET state = ?, updated_at = ? WHERE state = ? AND type IN ($in)", array_merge([self::QUEUED, $this->clock->now(), self::SUSPENDED], $types));
    }

    public function cancel(int $siteId, int $jobId): bool
    {
        return $this->db->execute('UPDATE {jobs} SET state = ?, updated_at = ? WHERE id = ? AND site_id = ? AND state IN (?, ?)', [self::CANCELLED, $this->clock->now(), $jobId, $siteId, self::QUEUED, self::SUSPENDED]) === 1;
    }

    /** Moves a dead job back to the queue after an operator fixes the cause. */
    public function retryDead(int $siteId, int $jobId): bool
    {
        return $this->db->execute('UPDATE {jobs} SET state = ?, attempts = 0, next_run_at = ?, updated_at = ? WHERE id = ? AND site_id = ? AND state = ?', [self::QUEUED, $this->clock->now(), $this->clock->now(), $jobId, $siteId, self::DEAD]) === 1;
    }

    /**
     * Moves one dead job back to the queue, but only if it belongs to this site, is dead right now and is of one of the given types: all
     * three conditions are part of the one UPDATE, so a stale, repeated or forged request changes nothing. Attempts start again from zero
     * (normal backoff applies). Nothing is dispatched here; a worker runs it later.
     *
     * @param list<string> $types
     */
    public function retryDeadOfTypes(int $siteId, int $jobId, array $types): bool
    {
        if ($types === []) {
            return false;
        }
        $now = $this->clock->now();
        $marks = implode(', ', array_fill(0, count($types), '?'));
        return $this->db->execute(
            "UPDATE {jobs} SET state = ?, attempts = 0, lease_owner = NULL, lease_expires = NULL, next_run_at = ?, updated_at = ? WHERE id = ? AND site_id = ? AND state = ? AND type IN ($marks)",
            array_merge([self::QUEUED, $now, $now, $jobId, $siteId, self::DEAD], array_values($types))
        ) === 1;
    }

    /**
     * The newest dead jobs of this site (bounded). The error text is the redacted one the queue stored when the job failed.
     *
     * @return list<array{id:int, type:string, attempts:int, max_attempts:int, error:string, updated_at:int}>
     */
    public function deadJobs(int $siteId, int $limit = 20): array
    {
        $rows = $this->db->select('SELECT id, type, attempts, max_attempts, last_error, updated_at FROM {jobs} WHERE site_id = ? AND state = ? ORDER BY updated_at DESC, id DESC LIMIT ?', [$siteId, self::DEAD, max(1, min(100, $limit))]);
        return array_map(static fn(array $r): array => ['id' => (int) $r['id'], 'type' => (string) $r['type'], 'attempts' => (int) $r['attempts'], 'max_attempts' => (int) $r['max_attempts'], 'error' => (string) ($r['last_error'] ?? ''), 'updated_at' => (int) $r['updated_at']], $rows);
    }

    public function get(int $id): ?Job
    {
        $row = $this->db->selectOne('SELECT * FROM {jobs} WHERE id = ?', [$id]);
        return $row === null ? null : Job::fromRow($row);
    }

    /** @return array{counts: array<string,int>, oldest_ready_age: int|null, expired_leases: int} */
    public function health(int $siteId): array
    {
        $now = $this->clock->now();
        $counts = [];
        foreach ($this->db->select('SELECT state, COUNT(*) AS n FROM {jobs} WHERE site_id = ? GROUP BY state', [$siteId]) as $r) {
            $counts[(string) $r['state']] = (int) $r['n'];
        }
        $oldest = $this->db->selectOne('SELECT MIN(next_run_at) AS t FROM {jobs} WHERE site_id = ? AND state = ? AND next_run_at <= ?', [$siteId, self::QUEUED, $now]);
        $expired = $this->db->selectOne('SELECT COUNT(*) AS n FROM {jobs} WHERE site_id = ? AND state = ? AND lease_expires < ?', [$siteId, self::RUNNING, $now]);
        return [
            'counts' => $counts,
            'oldest_ready_age' => ($oldest !== null && $oldest['t'] !== null) ? $now - (int) $oldest['t'] : null,
            'expired_leases' => (int) ($expired['n'] ?? 0),
        ];
    }
}