<?php

declare(strict_types=1);

namespace PhpWA\Runtime;

use InvalidArgumentException;
use PhpWA\AppState\AppStateName;
use PhpWA\AppState\AppStateSynchronizerInterface;
use PhpWA\Exception\ProtocolException;
use PhpWA\Integration\EventType;
use PhpWA\Storage\AccountStoreInterface;
use PhpWA\Storage\AccountTransactionInterface;

 
final class AppStateSyncJobWorker
{
    private const JOBS = 'app_state_sync_job';
    private const DEAD = 'app_state_sync_dead_letter';
    private const ERROR_CODE = 'appstate_sync_failed';

      private $accountId;
      private $accounts;
      private $synchronizer;
      private $leaseOwner;
      private $clock;
      private $leaseSeconds;
      private $maximumAttempts;
      private $baseDelaySeconds;

    public function __construct(
        string $accountId,
        AccountStoreInterface $accounts,
        AppStateSynchronizerInterface $synchronizer,
        ?string $leaseOwner = null,
        ?callable $clock = null,
        int $leaseSeconds = 120,
        int $maximumAttempts = 10,
        int $baseDelaySeconds = 5
    ) {
        self::assertAccountId($accountId);
        if ($leaseSeconds < 5 || $leaseSeconds > 3600 || $maximumAttempts < 1 || $maximumAttempts > 100
            || $baseDelaySeconds < 1 || $baseDelaySeconds > 300) {
            throw new InvalidArgumentException('App-state job worker bounds are invalid.');
        }
        if ($leaseOwner === null) { $leaseOwner = 'appstate-' . bin2hex(random_bytes(16)); }
        if (preg_match('/\A[A-Za-z0-9._:-]{1,128}\z/D', $leaseOwner) !== 1) {
            throw new InvalidArgumentException('App-state lease owner is invalid.');
        }
        $this->accountId = $accountId; $this->accounts = $accounts; $this->synchronizer = $synchronizer;
        $this->leaseOwner = $leaseOwner; $this->clock = $clock ?? static function (): int { return time(); };
        $this->leaseSeconds = $leaseSeconds; $this->maximumAttempts = $maximumAttempts;
        $this->baseDelaySeconds = $baseDelaySeconds;
    }

    public function run(Deadline $deadline, int $maximumJobs = 20, int $maximumPagesPerJob = 20): AppStateJobRunResult
    {
        if ($maximumJobs < 1 || $maximumJobs > 100 || $maximumPagesPerJob < 1 || $maximumPagesPerJob > 100) {
            throw new InvalidArgumentException('App-state job run bounds are invalid.');
        }
        $jobs = 0; $pages = 0; $mutations = 0; $failed = 0; $dead = 0; $lastError = null;
        while ($jobs < $maximumJobs && !$deadline->isExpired()) {
            $job = $this->claim($this->now());
            if ($job === null) { break; }
            $jobs++;
            try {
                $last = null;
                for ($page = 0; $page < $maximumPagesPerJob && !$deadline->isExpired(); $page++) {
                    $last = $this->synchronizer->sync($job['name'], $deadline);
                    if ($last->name() !== $job['name']) { throw new ProtocolException('App-state synchronizer changed the requested collection.'); }
                    $pages++; $mutations += $last->mutations();
                    if (!$last->hasMore()) { break; }
                }
                if ($last === null) {
                    $this->releasePartial($job['name'], $this->now());
                    break;
                }
                if (!$last->hasMore() && $last->version() < $job['target_version']) {
                    throw new ProtocolException('App-state server ended the sync below the notified target version.');
                }
                if ($last->hasMore()) {
                    $this->releasePartial($job['name'], $this->now());
                    break;
                }
                $this->completeOrContinue($job['name'], $last->version(), $this->now());
            } catch (\Throwable $error) {
                 
                 
                $failed++; $lastError = self::ERROR_CODE;
                if ($this->fail($job['name'], $lastError, $this->now())) { $dead++; }
                 
                 
                break;
            }
        }
        return new AppStateJobRunResult($jobs, $pages, $mutations, $failed, $dead, $lastError);
    }

     
    private function claim(int $now): ?array
    {
        return $this->accounts->transaction(
            $this->accountId,
            function (AccountTransactionInterface $tx) use ($now): ?array {
                $candidates = [];
                foreach ($tx->list(self::JOBS, 1000) as $name => $stored) {
                    AppStateName::assert($name); $job = $this->decode($stored->value());
                    if ($job['available_at'] > $now || ($job['lease_owner'] !== null && $job['lease_until'] > $now)) { continue; }
                    $candidates[] = ['name' => $name, 'job' => $job, 'revision' => $stored->revision()];
                }
                if ($candidates === []) { return null; }
                usort($candidates, static function (array $left, array $right): int {
                    return [$left['job']['requested_at'], $left['name']] <=> [$right['job']['requested_at'], $right['name']];
                });
                $chosen = $candidates[0]; $job = $chosen['job'];
                $job['lease_owner'] = $this->leaseOwner; $job['lease_until'] = $this->add($now, $this->leaseSeconds);
                $tx->put(self::JOBS, $chosen['name'], $this->encode($job), $chosen['revision']);
                return ['name' => $chosen['name'], 'target_version' => $job['target_version']];
            }
        );
    }

    private function releasePartial(string $name, int $now): void
    {
        $this->accounts->transaction($this->accountId, function (AccountTransactionInterface $tx) use ($name, $now): void {
            $stored = $this->owned($tx, $name); $job = $this->decode($stored->value());
            $job['available_at'] = $now; $job['lease_owner'] = null; $job['lease_until'] = 0;
            $tx->put(self::JOBS, $name, $this->encode($job), $stored->revision());
        });
    }

    private function completeOrContinue(string $name, int $version, int $now): void
    {
        $this->accounts->transaction($this->accountId, function (AccountTransactionInterface $tx) use ($name, $version, $now): void {
            $stored = $this->owned($tx, $name); $job = $this->decode($stored->value());
            if ($version >= $job['target_version']) { $tx->delete(self::JOBS, $name, $stored->revision()); return; }
            $job['available_at'] = $now; $job['lease_owner'] = null; $job['lease_until'] = 0;
            $tx->put(self::JOBS, $name, $this->encode($job), $stored->revision());
        });
    }

     
    private function fail(string $name, string $error, int $now): bool
    {
        return $this->accounts->transaction($this->accountId, function (AccountTransactionInterface $tx) use ($name, $error, $now): bool {
            $stored = $this->owned($tx, $name); $job = $this->decode($stored->value()); $job['attempts']++;
            if ($job['attempts'] >= $this->maximumAttempts) {
                $letter = $this->encode([
                    'v' => 2, 'target_version' => $job['target_version'], 'requested_at' => $job['requested_at'],
                    'attempts' => $job['attempts'], 'available_at' => 0, 'lease_owner' => null, 'lease_until' => 0,
                    'failed_at' => $now, 'error_code' => $error,
                ], true);
                $known = $tx->get(self::DEAD, $name);
                $tx->put(self::DEAD, $name, $letter, $known === null ? 0 : $known->revision());
                $event = $this->json(['schema_version' => 2, 'collection' => $name, 'target_version' => $job['target_version'], 'attempts' => $job['attempts'], 'error_code' => $error]);
                $tx->stageIncoming('appstate-failed:' . $name . ':' . $job['target_version'], EventType::APPSTATE_SYNC_FAILED, $event, $now);
                $tx->delete(self::JOBS, $name, $stored->revision());
                return true;
            }
            $job['available_at'] = $this->add($now, min(3600, $this->baseDelaySeconds * (1 << min(10, $job['attempts'] - 1))));
            $job['lease_owner'] = null; $job['lease_until'] = 0;
            $tx->put(self::JOBS, $name, $this->encode($job), $stored->revision());
            return false;
        });
    }

    private function owned(AccountTransactionInterface $tx, string $name): \PhpWA\Storage\StoredValue
    {
        $stored = $tx->get(self::JOBS, $name);
        if ($stored === null) { throw new ProtocolException('Claimed app-state job disappeared before completion.'); }
        $job = $this->decode($stored->value());
        if ($job['lease_owner'] !== $this->leaseOwner) { throw new ProtocolException('App-state job lease ownership was lost.'); }
        return $stored;
    }

     
    private function decode(string $json): array
    {
        try { $value = json_decode($json, true, 16, JSON_THROW_ON_ERROR); }
        catch (\Throwable $error) { throw new ProtocolException('Stored app-state sync job is malformed.', 0, $error); }
        if (!is_array($value) || ($value['v'] ?? null) !== 1 || !is_int($value['target_version'] ?? null) || $value['target_version'] < 0
            || !is_int($value['requested_at'] ?? null) || $value['requested_at'] < 0) {
            throw new ProtocolException('Stored app-state sync job has an invalid shape.');
        }
        $attempts = $value['attempts'] ?? 0; $available = $value['available_at'] ?? 0;
        $owner = $value['lease_owner'] ?? null; $until = $value['lease_until'] ?? 0;
        if (!is_int($attempts) || $attempts < 0 || $attempts >= PHP_INT_MAX || !is_int($available) || $available < 0
            || ($owner !== null && (!is_string($owner) || preg_match('/\A[A-Za-z0-9._:-]{1,128}\z/D', $owner) !== 1))
            || !is_int($until) || $until < 0) {
            throw new ProtocolException('Stored app-state sync job lease is invalid.');
        }
        return ['v' => 1, 'target_version' => $value['target_version'], 'requested_at' => $value['requested_at'],
            'attempts' => $attempts, 'available_at' => $available, 'lease_owner' => $owner, 'lease_until' => $until];
    }

     
    private function encode(array $job, bool $dead = false): string
    {
        if (!$dead) {
            $job = ['v' => 1, 'target_version' => $job['target_version'], 'requested_at' => $job['requested_at'],
                'attempts' => $job['attempts'], 'available_at' => $job['available_at'],
                'lease_owner' => $job['lease_owner'], 'lease_until' => $job['lease_until']];
        }
        return $this->json($job);
    }

     
    private function json(array $value): string
    {
        try { $json = json_encode($value, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR); }
        catch (\Throwable $error) { throw new ProtocolException('Unable to serialize app-state job state.', 0, $error); }
        if (!is_string($json) || $json === '' || strlen($json) > 4194304) { throw new ProtocolException('App-state job state exceeds its limit.'); }
        return $json;
    }

    private function now(): int
    {
        $now = ($this->clock)();
        if (!is_int($now) || $now < 0) { throw new ProtocolException('App-state job clock returned an invalid timestamp.'); }
        return $now;
    }
    private function add(int $value, int $increment): int { return $value > PHP_INT_MAX - $increment ? PHP_INT_MAX : $value + $increment; }
    private static function assertAccountId(string $accountId): void
    {
        if ($accountId === '' || strlen($accountId) > 512 || preg_match('//u', $accountId) !== 1
            || preg_match('/[\x00-\x1f\x7f]/', $accountId) === 1) { throw new InvalidArgumentException('App-state worker account ID is invalid.'); }
    }
}
