<?php

declare(strict_types=1);

namespace PhpWA\Storage;

use InvalidArgumentException;
use PDO;
use PDOException;
use PhpWA\Exception\StorageConflictException;
use PhpWA\Exception\StorageException;

/** @internal Created only by PdoAccountStore inside a database transaction. */
final class PdoAccountTransaction implements AccountTransactionInterface
{
    /** @var PDO */ private $pdo;
    /** @var SecretCipherInterface */ private $cipher;
    /** @var string */ private $indexKey;
    /** @var string */ private $prefix;
    /** @var string */ private $driver;
    /** @var string */ private $accountId;
    /** @var string */ private $accountHash;
    /** @var bool */ private $dirty = false;
    /** @var bool */ private $active = true;

    public function __construct(PDO $pdo, SecretCipherInterface $cipher, string $indexKey, string $prefix, string $driver, string $accountId, string $accountHash)
    {
        $this->pdo = $pdo;
        $this->cipher = $cipher;
        $this->indexKey = $indexKey;
        $this->prefix = $prefix;
        $this->driver = $driver;
        $this->accountId = $accountId;
        $this->accountHash = $accountHash;
    }

    public function get(string $collection, string $key): ?StoredValue
    {
        $this->assertActive();
        self::assertCollectionAndKey($collection, $key);
        $hash = $this->stateKeyHash($collection, $key);
        $statement = $this->pdo->prepare("SELECT encrypted_value, revision, state_key FROM {$this->prefix}state WHERE account_hash = ? AND collection = ? AND state_key_hash = ?");
        $statement->execute([$this->accountHash, $collection, $hash]);
        $row = $statement->fetch(PDO::FETCH_ASSOC);
        if ($row === false) { return null; }
        $storedKey = $this->decryptStateKey($collection, $hash, $row['state_key'] ?? null);
        if (!hash_equals($key, $storedKey)) {
            throw new StorageException('PDO account state key hash collision detected.');
        }
        return new StoredValue($this->decryptState($collection, $key, $row['encrypted_value']), (int) $row['revision']);
    }

    public function list(string $collection, int $limit = 1000): array
    {
        $this->assertActive();
        self::assertCollectionAndKey($collection, 'probe');
        if ($limit < 1 || $limit > 10000) {
            throw new InvalidArgumentException('Account state list limit is invalid.');
        }
        $statement = $this->pdo->prepare("SELECT state_key_hash, state_key, encrypted_value, revision FROM {$this->prefix}state WHERE account_hash = ? AND collection = ? ORDER BY state_key_hash LIMIT " . $limit);
        $statement->execute([$this->accountHash, $collection]);
        $result = [];
        while (($row = $statement->fetch(PDO::FETCH_ASSOC)) !== false) {
            $hash = $row['state_key_hash'] ?? null;
            if (!is_string($hash) || preg_match('/\A[a-f0-9]{64}\z/D', $hash) !== 1) {
                throw new StorageException('PDO account state key hash is malformed.');
            }
            $key = $this->decryptStateKey($collection, $hash, $row['state_key'] ?? null);
            self::assertCollectionAndKey($collection, $key);
            if (!hash_equals($hash, $this->stateKeyHash($collection, $key))) {
                throw new StorageException('PDO account state key hash collision detected.');
            }
            $result[$key] = new StoredValue($this->decryptState($collection, $key, $row['encrypted_value']), (int) $row['revision']);
        }
        return $result;
    }

    public function put(string $collection, string $key, string $value, ?int $expectedRevision = null): StoredValue
    {
        $this->assertActive();
        self::assertCollectionAndKey($collection, $key);
        self::assertValueAndExpectedRevision($value, $expectedRevision);
        $current = $this->getForUpdate($collection, $key);
        $currentRevision = $current === null ? 0 : (int) $current['revision'];
        if ($expectedRevision !== null && $expectedRevision !== $currentRevision) {
            throw new StorageConflictException('Account state was changed by another operation.');
        }
        if ($currentRevision === PHP_INT_MAX) { throw new StorageException('Account state revision is exhausted.'); }
        $revision = $currentRevision + 1;
        $encrypted = $this->cipher->encrypt($this->stateContext($collection, $key), $value);
        if ($current === null) {
            $statement = $this->pdo->prepare("INSERT INTO {$this->prefix}state (account_hash, collection, state_key_hash, state_key, encrypted_value, revision, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)");
            $hash = $this->stateKeyHash($collection, $key);
            $statement->execute([$this->accountHash, $collection, $hash, $this->encryptStateKey($collection, $hash, $key), $encrypted, $revision, time()]);
        } else {
            $statement = $this->pdo->prepare("UPDATE {$this->prefix}state SET encrypted_value = ?, revision = ?, updated_at = ? WHERE account_hash = ? AND collection = ? AND state_key_hash = ? AND revision = ?");
            $statement->execute([$encrypted, $revision, time(), $this->accountHash, $collection, $this->stateKeyHash($collection, $key), $currentRevision]);
            if ($statement->rowCount() !== 1) { throw new StorageConflictException('Account state was changed by another operation.'); }
        }
        $this->dirty = true;
        return new StoredValue($value, $revision);
    }

    public function delete(string $collection, string $key, ?int $expectedRevision = null): void
    {
        $this->assertActive();
        self::assertCollectionAndKey($collection, $key);
        if ($expectedRevision !== null && $expectedRevision < 0) { throw new InvalidArgumentException('Expected account state revision is invalid.'); }
        $current = $this->getForUpdate($collection, $key);
        $revision = $current === null ? 0 : (int) $current['revision'];
        if ($expectedRevision !== null && $expectedRevision !== $revision) {
            throw new StorageConflictException('Account state was changed by another operation.');
        }
        if ($current !== null) {
            $statement = $this->pdo->prepare("DELETE FROM {$this->prefix}state WHERE account_hash = ? AND collection = ? AND state_key_hash = ? AND revision = ?");
            $statement->execute([$this->accountHash, $collection, $this->stateKeyHash($collection, $key), $revision]);
            if ($statement->rowCount() !== 1) { throw new StorageConflictException('Account state was changed by another operation.'); }
            $this->dirty = true;
        }
    }

    public function hasIncoming(string $messageId): bool
    {
        $this->assertActive();
        $statement = $this->pdo->prepare("SELECT 1 FROM {$this->prefix}dedupe WHERE account_hash = ? AND message_hash = ?");
        $statement->execute([$this->accountHash, $this->messageHash($messageId)]);
        return $statement->fetchColumn() !== false;
    }

    public function stageIncoming(string $messageId, string $eventType, string $jsonPayload, int $receivedAt): bool
    {
        $this->assertActive();
        self::assertMessageId($messageId);
        OutboxEvent::assertType($eventType);
        OutboxEvent::assertPayload($jsonPayload);
        if ($receivedAt < 0) { throw new InvalidArgumentException('Incoming message timestamp is invalid.'); }
        if ($this->hasIncoming($messageId)) { return false; }
        $messageHash = $this->messageHash($messageId);
        $eventId = OutboxEvent::incomingId(hash('sha256', $this->accountId), $messageId, $eventType);
        $insert = $this->pdo->prepare("INSERT INTO {$this->prefix}dedupe (account_hash, message_hash, received_at) VALUES (?, ?, ?)");
        $insert->execute([$this->accountHash, $messageHash, $receivedAt]);
        $payload = $this->cipher->encrypt($this->outboxContext($eventId), $jsonPayload);
        $insert = $this->pdo->prepare("INSERT INTO {$this->prefix}outbox (event_id, account_hash, event_type, encrypted_payload, occurred_at, attempts, available_at, lease_owner, lease_until) VALUES (?, ?, ?, ?, ?, 0, ?, NULL, 0)");
        $insert->execute([$eventId, $this->accountHash, $eventType, $payload, $receivedAt, $receivedAt]);
        $delete = $this->pdo->prepare("DELETE FROM {$this->prefix}retries WHERE account_hash = ? AND message_hash = ?");
        $delete->execute([$this->accountHash, $messageHash]);
        $this->dirty = true;
        return true;
    }

    public function claimOutbox(string $leaseOwner, int $limit, int $now, int $leaseSeconds): array
    {
        $this->assertActive();
        self::assertLeaseOwner($leaseOwner);
        if ($limit < 1 || $limit > 100 || $now < 0 || $leaseSeconds < 5 || $leaseSeconds > 3600 || $now > PHP_INT_MAX - $leaseSeconds) {
            throw new InvalidArgumentException('Outbox claim bounds are invalid.');
        }
        $suffix = $this->driver === 'mysql' ? ' FOR UPDATE' : '';
        $statement = $this->pdo->prepare("SELECT event_id, event_type, encrypted_payload, occurred_at, attempts FROM {$this->prefix}outbox WHERE account_hash = ? AND available_at <= ? AND (lease_owner IS NULL OR lease_until <= ?) ORDER BY available_at, occurred_at, event_id LIMIT " . $limit . $suffix);
        $statement->execute([$this->accountHash, $now, $now]);
        $rows = $statement->fetchAll(PDO::FETCH_ASSOC);
        $events = [];
        foreach ($rows as $row) {
            $attempts = (int) $row['attempts'];
            if ($attempts === PHP_INT_MAX) { throw new StorageException('Outbox attempt counter is exhausted.'); }
            $attempts++;
            $update = $this->pdo->prepare("UPDATE {$this->prefix}outbox SET attempts = ?, lease_owner = ?, lease_until = ? WHERE event_id = ? AND account_hash = ?");
            $update->execute([$attempts, $leaseOwner, $now + $leaseSeconds, $row['event_id'], $this->accountHash]);
            $payload = $this->cipher->decrypt($this->outboxContext((string) $row['event_id']), (string) $row['encrypted_payload']);
            $events[] = new OutboxEvent((string) $row['event_id'], (string) $row['event_type'], $payload, (int) $row['occurred_at'], $attempts, $leaseOwner);
            $this->dirty = true;
        }
        return $events;
    }

    public function completeOutbox(string $eventId, string $leaseOwner): void
    {
        $this->assertActive();
        self::assertEventId($eventId);
        self::assertLeaseOwner($leaseOwner);
        $this->assertLeaseOwnership($eventId, $leaseOwner);
        $statement = $this->pdo->prepare("DELETE FROM {$this->prefix}outbox WHERE event_id = ? AND account_hash = ? AND lease_owner = ?");
        $statement->execute([$eventId, $this->accountHash, $leaseOwner]);
        if ($statement->rowCount() !== 1) { throw new StorageConflictException('Outbox event lease changed before completion.'); }
        $this->dirty = true;
    }

    public function releaseOutbox(string $eventId, string $leaseOwner, int $availableAt): void
    {
        $this->assertActive();
        self::assertEventId($eventId);
        self::assertLeaseOwner($leaseOwner);
        if ($availableAt < 0) { throw new InvalidArgumentException('Outbox availability time is invalid.'); }
        $this->assertLeaseOwnership($eventId, $leaseOwner);
        $statement = $this->pdo->prepare("UPDATE {$this->prefix}outbox SET lease_owner = NULL, lease_until = 0, available_at = ? WHERE event_id = ? AND account_hash = ? AND lease_owner = ?");
        $statement->execute([$availableAt, $eventId, $this->accountHash, $leaseOwner]);
        if ($statement->rowCount() !== 1) { throw new StorageConflictException('Outbox event lease changed before release.'); }
        $this->dirty = true;
    }

    public function hasPendingOutbox(): bool
    {
        $this->assertActive();
        $statement = $this->pdo->prepare("SELECT 1 FROM {$this->prefix}outbox WHERE account_hash = ? LIMIT 1");
        $statement->execute([$this->accountHash]);
        return $statement->fetchColumn() !== false;
    }

    public function registerRetry(string $messageId, int $now, int $maximumAttempts = 5, int $baseDelaySeconds = 2): RetryDecision
    {
        $this->assertActive();
        self::assertMessageId($messageId);
        if ($now < 0 || $maximumAttempts < 1 || $maximumAttempts > 10 || $baseDelaySeconds < 1 || $baseDelaySeconds > 60) {
            throw new InvalidArgumentException('Retry bounds are invalid.');
        }
        $hash = $this->messageHash($messageId);
        $suffix = $this->driver === 'mysql' ? ' FOR UPDATE' : '';
        $statement = $this->pdo->prepare("SELECT attempts, next_allowed_at FROM {$this->prefix}retries WHERE account_hash = ? AND message_hash = ?" . $suffix);
        $statement->execute([$this->accountHash, $hash]);
        $row = $statement->fetch(PDO::FETCH_ASSOC);
        $attempts = $row === false ? 0 : (int) $row['attempts'];
        $nextAllowed = $row === false ? 0 : (int) $row['next_allowed_at'];
        if ($attempts >= $maximumAttempts) { return new RetryDecision($attempts, false, true, $nextAllowed); }
        if ($nextAllowed > $now) { return new RetryDecision($attempts, false, false, $nextAllowed); }
        $attempts++;
        $delay = min(300, $baseDelaySeconds * (1 << min(7, $attempts - 1)));
        $nextAllowed = $now > PHP_INT_MAX - $delay ? PHP_INT_MAX : $now + $delay;
        if ($row === false) {
            $statement = $this->pdo->prepare("INSERT INTO {$this->prefix}retries (account_hash, message_hash, attempts, next_allowed_at, updated_at) VALUES (?, ?, ?, ?, ?)");
            $statement->execute([$this->accountHash, $hash, $attempts, $nextAllowed, $now]);
        } else {
            $statement = $this->pdo->prepare("UPDATE {$this->prefix}retries SET attempts = ?, next_allowed_at = ?, updated_at = ? WHERE account_hash = ? AND message_hash = ?");
            $statement->execute([$attempts, $nextAllowed, $now, $this->accountHash, $hash]);
        }
        $this->dirty = true;
        return new RetryDecision($attempts, true, false, $nextAllowed);
    }

    public function clearRetry(string $messageId): void
    {
        $this->assertActive();
        $statement = $this->pdo->prepare("DELETE FROM {$this->prefix}retries WHERE account_hash = ? AND message_hash = ?");
        $statement->execute([$this->accountHash, $this->messageHash($messageId)]);
        if ($statement->rowCount() > 0) { $this->dirty = true; }
    }

    public function prune(int $dedupeBefore, int $retryBefore): void
    {
        $this->assertActive();
        if ($dedupeBefore < 0 || $retryBefore < 0) { throw new InvalidArgumentException('Prune timestamps are invalid.'); }
        // Retain all dedupe rows while this account has an undelivered outbox
        // event. This conservative boundary prevents a redelivery from
        // colliding with or replacing the stable event ID after early pruning.
        $dedupe = $this->pdo->prepare("DELETE FROM {$this->prefix}dedupe WHERE account_hash = ? AND received_at < ? AND NOT EXISTS (SELECT 1 FROM {$this->prefix}outbox WHERE account_hash = ?)");
        $dedupe->execute([$this->accountHash, $dedupeBefore, $this->accountHash]);
        $retries = $this->pdo->prepare("DELETE FROM {$this->prefix}retries WHERE account_hash = ? AND updated_at < ?");
        $retries->execute([$this->accountHash, $retryBefore]);
        if ($dedupe->rowCount() > 0 || $retries->rowCount() > 0) { $this->dirty = true; }
    }

    public function isDirty(): bool { return $this->dirty; }
    public function close(): void { $this->active = false; }

    /** @return array<string,mixed>|null */
    private function getForUpdate(string $collection, string $key): ?array
    {
        $suffix = $this->driver === 'mysql' ? ' FOR UPDATE' : '';
        $hash = $this->stateKeyHash($collection, $key);
        $statement = $this->pdo->prepare("SELECT state_key, revision FROM {$this->prefix}state WHERE account_hash = ? AND collection = ? AND state_key_hash = ?" . $suffix);
        $statement->execute([$this->accountHash, $collection, $hash]);
        $row = $statement->fetch(PDO::FETCH_ASSOC);
        if ($row === false) { return null; }
        if (!hash_equals($key, $this->decryptStateKey($collection, $hash, $row['state_key'] ?? null))) {
            throw new StorageException('PDO account state key hash collision detected.');
        }
        return $row;
    }

    private function assertLeaseOwnership(string $eventId, string $leaseOwner): void
    {
        $suffix = $this->driver === 'mysql' ? ' FOR UPDATE' : '';
        $statement = $this->pdo->prepare("SELECT lease_owner FROM {$this->prefix}outbox WHERE event_id = ? AND account_hash = ?" . $suffix);
        $statement->execute([$eventId, $this->accountHash]);
        $owner = $statement->fetchColumn();
        if (!is_string($owner) || !hash_equals($owner, $leaseOwner)) {
            throw new StorageConflictException('Outbox event is not leased by this worker.');
        }
    }

    private function decryptState(string $collection, string $key, $envelope): string
    {
        if (!is_string($envelope)) { throw new StorageException('PDO account state payload is malformed.'); }
        return $this->cipher->decrypt($this->stateContext($collection, $key), $envelope);
    }

    private function encryptStateKey(string $collection, string $hash, string $key): string
    {
        return $this->cipher->encrypt($this->stateKeyContext($collection, $hash), $key);
    }

    /** @param mixed $envelope */
    private function decryptStateKey(string $collection, string $hash, $envelope): string
    {
        if (!is_string($envelope)) {
            throw new StorageException('PDO account state key is malformed.');
        }
        return $this->cipher->decrypt($this->stateKeyContext($collection, $hash), $envelope);
    }

    private function stateKeyContext(string $collection, string $hash): string
    {
        return "pdo-state-key\0" . $this->accountId . "\0" . $collection . "\0" . $hash;
    }

    private function stateContext(string $collection, string $key): string
    {
        return "pdo-state\0" . $this->accountId . "\0" . $collection . "\0" . $key;
    }

    private function outboxContext(string $eventId): string
    {
        return "pdo-outbox\0" . $this->accountId . "\0" . $eventId;
    }

    private function stateKeyHash(string $collection, string $key): string
    {
        return hash_hmac('sha256', "state\0" . $collection . "\0" . $key, $this->indexKey);
    }

    private function messageHash(string $messageId): string
    {
        self::assertMessageId($messageId);
        return hash_hmac('sha256', "message\0" . $messageId, $this->indexKey);
    }

    private function assertActive(): void
    {
        if (!$this->active || !$this->pdo->inTransaction()) {
            throw new StorageException('PDO account transaction is already closed.');
        }
    }

    private static function assertCollectionAndKey(string $collection, string $key): void
    {
        if (preg_match('/\A[a-z][a-z0-9_]{0,63}\z/D', $collection) !== 1 || $key === '' || strlen($key) > 1024 || preg_match('//u', $key) !== 1 || preg_match('/[\x00-\x1f\x7f]/', $key) === 1) {
            throw new InvalidArgumentException('Account state collection or key is invalid.');
        }
    }

    private static function assertValueAndExpectedRevision(string $value, ?int $revision): void
    {
        if (strlen($value) > 16777216 || ($revision !== null && $revision < 0)) {
            throw new InvalidArgumentException('Account state value or expected revision is invalid.');
        }
    }

    private static function assertMessageId(string $messageId): void
    {
        if ($messageId === '' || strlen($messageId) > 512 || preg_match('/[\x00-\x1f\x7f]/', $messageId) === 1) {
            throw new InvalidArgumentException('WhatsApp message ID is invalid.');
        }
    }

    private static function assertLeaseOwner(string $owner): void
    {
        if (preg_match('/\A[A-Za-z0-9._:-]{1,128}\z/D', $owner) !== 1) { throw new InvalidArgumentException('Outbox lease owner is invalid.'); }
    }

    private static function assertEventId(string $eventId): void
    {
        if (preg_match('/\A[a-f0-9]{64}\z/D', $eventId) !== 1) { throw new InvalidArgumentException('Outbox event ID is invalid.'); }
    }
}
