<?php
declare(strict_types=1);

// This endpoint is deployed as /worker_api.php on the AI server. It only
// exposes queue metadata; account credentials and mail bodies stay server-side.
const WORKER_SECRET_FILE = '/etc/yntak2.com/delivery-worker.secret';
const WORKER_BASELINE_FILE = '/etc/yntak2.com/delivery-worker-baseline.json';
const MAIL_CONFIG_FILE = '/var/www/yntak2.com/mail/config.php';
const MAX_CLOCK_SKEW = 300;

require_once MAIL_CONFIG_FILE;

function worker_response(int $status, array $payload): never
{
    http_response_code($status);
    header('Content-Type: application/json; charset=utf-8');
    header('Cache-Control: no-store');
    echo json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
    exit;
}

function worker_secret(): string
{
    $secret = @file_get_contents(WORKER_SECRET_FILE);
    if ($secret === false || trim($secret) === '') {
        worker_response(503, ['ok' => false, 'error' => 'worker secret is not configured']);
    }
    return trim($secret);
}

function verify_worker_request(string $body): void
{
    if (($_SERVER['REQUEST_METHOD'] ?? '') !== 'POST') {
        worker_response(405, ['ok' => false, 'error' => 'POST required']);
    }

    $timestamp = trim((string)($_SERVER['HTTP_X_WORKER_TIMESTAMP'] ?? ''));
    $signature = trim((string)($_SERVER['HTTP_X_WORKER_SIGNATURE'] ?? ''));
    if (!preg_match('/^\d{10}$/', $timestamp)) {
        worker_response(401, ['ok' => false, 'error' => 'invalid request timestamp']);
    }
    if (abs(time() - (int)$timestamp) > MAX_CLOCK_SKEW) {
        worker_response(401, ['ok' => false, 'error' => 'request timestamp expired']);
    }

    $expected = hash_hmac('sha256', $timestamp . '.' . $body, worker_secret());
    if ($signature === '' || !hash_equals($expected, $signature)) {
        worker_response(401, ['ok' => false, 'error' => 'invalid request signature']);
    }
}

function request_payload(): array
{
    $body = file_get_contents('php://input');
    if ($body === false) {
        worker_response(400, ['ok' => false, 'error' => 'request body is missing']);
    }
    verify_worker_request($body);
    try {
        $payload = json_decode($body, true, 512, JSON_THROW_ON_ERROR);
    } catch (Throwable $error) {
        worker_response(400, ['ok' => false, 'error' => 'invalid JSON body']);
    }
    return is_array($payload) ? $payload : [];
}

function ensure_worker_schema(): void
{
    db()->exec(<<<'SQL'
        CREATE TABLE IF NOT EXISTS automation_dispatch_jobs (
            job_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
            kind VARCHAR(16) NOT NULL,
            message_id BIGINT UNSIGNED NOT NULL,
            chat_url VARCHAR(512) NOT NULL,
            quantity INT UNSIGNED NULL,
            state VARCHAR(16) NOT NULL DEFAULT 'claimed',
            attempts TINYINT UNSIGNED NOT NULL DEFAULT 1,
            last_error TEXT NULL,
            next_attempt_at DATETIME NULL,
            delivery_id BIGINT UNSIGNED NULL,
            created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
            completed_at DATETIME NULL,
            PRIMARY KEY (job_id),
            UNIQUE KEY uq_worker_message (kind, message_id),
            UNIQUE KEY uq_worker_chat (kind, chat_url),
            KEY idx_worker_state (state, next_attempt_at)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    SQL);
}

function ensure_listing_schema(): void
{
    db()->exec(<<<'SQL'
        CREATE TABLE IF NOT EXISTS listing_control_commands (
            command_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
            action VARCHAR(32) NOT NULL,
            payload_json MEDIUMTEXT NOT NULL,
            state VARCHAR(16) NOT NULL DEFAULT 'queued',
            attempts TINYINT UNSIGNED NOT NULL DEFAULT 0,
            last_error TEXT NULL,
            created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
            completed_at DATETIME NULL,
            PRIMARY KEY (command_id),
            KEY idx_listing_command_state (state, created_at)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    SQL);
    db()->exec(<<<'SQL'
        CREATE TABLE IF NOT EXISTS listing_control_state (
            state_id TINYINT UNSIGNED NOT NULL,
            state_json MEDIUMTEXT NOT NULL,
            updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
            PRIMARY KEY (state_id)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    SQL);
}

function listing_command_payload(string $action, mixed $value): array
{
    $allowed = ['set_enabled', 'save_config', 'run_manual'];
    if (!in_array($action, $allowed, true) || !is_array($value)) {
        worker_response(400, ['ok' => false, 'error' => 'invalid listing command']);
    }

    if ($action === 'set_enabled') {
        return ['enabled' => !empty($value['enabled'])];
    }
    if ($action === 'run_manual') {
        $site = (string)($value['site'] ?? '');
        $url = trim((string)($value['url'] ?? ''));
        $host = strtolower((string)(parse_url($url, PHP_URL_HOST) ?: ''));
        if (!in_array($site, ['maclub', 'gameclub'], true) || $url === '') {
            worker_response(400, ['ok' => false, 'error' => 'invalid manual listing command']);
        }
        $expectedHost = $site === 'maclub' ? 'maclub.jp' : 'gameclub.jp';
        if ($host !== $expectedHost) {
            worker_response(400, ['ok' => false, 'error' => 'listing URL host does not match site']);
        }
        return ['site' => $site, 'url' => mb_substr($url, 0, 1000)];
    }

    $normaliseUrls = static function (mixed $urls): array {
        if (!is_array($urls)) {
            return [];
        }
        $result = [];
        foreach ($urls as $url) {
            $url = trim((string)$url);
            if ($url !== '' && filter_var($url, FILTER_VALIDATE_URL)) {
                $result[] = mb_substr($url, 0, 1000);
            }
        }
        return array_values(array_unique($result));
    };
    $min = max(1, min(1440, (int)($value['min_interval'] ?? 60)));
    $max = max($min, min(1440, (int)($value['max_interval'] ?? $min)));
    return [
        'maclub_urls' => $normaliseUrls($value['maclub_urls'] ?? []),
        'gameclub_urls' => $normaliseUrls($value['gameclub_urls'] ?? []),
        'min_interval' => $min,
        'max_interval' => $max,
    ];
}

function enqueue_listing_command(array $payload): void
{
    ensure_listing_schema();
    $action = trim((string)($payload['action'] ?? ''));
    $command = listing_command_payload($action, $payload['payload'] ?? null);
    $stmt = db()->prepare(
        'INSERT INTO listing_control_commands (action, payload_json, state)
         VALUES (?, ?, \'queued\')'
    );
    $stmt->execute([
        $action,
        json_encode($command, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR),
    ]);
    worker_response(200, ['ok' => true, 'command_id' => (int)db()->lastInsertId()]);
}

function pull_listing_command(): void
{
    ensure_listing_schema();
    $pdo = db();
    $pdo->beginTransaction();
    try {
        $stmt = $pdo->query(
            "SELECT * FROM listing_control_commands
             WHERE state = 'queued'
             ORDER BY created_at ASC, command_id ASC
             LIMIT 1 FOR UPDATE"
        );
        $command = $stmt->fetch(PDO::FETCH_ASSOC);
        if (!$command) {
            $pdo->commit();
            worker_response(200, ['ok' => true, 'job' => null]);
        }
        $update = $pdo->prepare(
            "UPDATE listing_control_commands
             SET state = 'claimed', attempts = attempts + 1, updated_at = NOW()
             WHERE command_id = ?"
        );
        $update->execute([(int)$command['command_id']]);
        $pdo->commit();
        $decoded = json_decode((string)$command['payload_json'], true);
        worker_response(200, [
            'ok' => true,
            'job' => [
                'command_id' => (int)$command['command_id'],
                'action' => (string)$command['action'],
                'payload' => is_array($decoded) ? $decoded : [],
                'attempts' => (int)$command['attempts'] + 1,
            ],
        ]);
    } catch (Throwable $error) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        worker_response(500, ['ok' => false, 'error' => 'listing command pull failed']);
    }
}

function acknowledge_listing_command(array $payload): void
{
    ensure_listing_schema();
    $commandId = filter_var($payload['command_id'] ?? null, FILTER_VALIDATE_INT);
    $state = (string)($payload['state'] ?? '');
    if (!$commandId || !in_array($state, ['succeeded', 'failed'], true)) {
        worker_response(400, ['ok' => false, 'error' => 'invalid listing acknowledgement']);
    }
    $error = mb_substr(trim((string)($payload['error'] ?? '')), 0, 2000);
    $stmt = db()->prepare(
        'UPDATE listing_control_commands
         SET state = ?, last_error = ?, updated_at = NOW(), completed_at = ?
         WHERE command_id = ? AND state = \'claimed\''
    );
    $completedAt = $state === 'succeeded' ? date('Y-m-d H:i:s') : null;
    $stmt->execute([$state, $error ?: null, $completedAt, (int)$commandId]);
    if ($stmt->rowCount() === 0) {
        worker_response(404, ['ok' => false, 'error' => 'listing command not found or already acknowledged']);
    }
    worker_response(200, ['ok' => true]);
}

function update_listing_state(array $payload): void
{
    ensure_listing_schema();
    $state = $payload['state'] ?? null;
    if (!is_array($state)) {
        worker_response(400, ['ok' => false, 'error' => 'invalid listing state']);
    }
    $json = json_encode($state, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR);
    if (strlen($json) > 100000) {
        worker_response(413, ['ok' => false, 'error' => 'listing state is too large']);
    }
    $stmt = db()->prepare(
        'INSERT INTO listing_control_state (state_id, state_json)
         VALUES (1, ?)
         ON DUPLICATE KEY UPDATE state_json = VALUES(state_json), updated_at = NOW()'
    );
    $stmt->execute([$json]);
    worker_response(200, ['ok' => true]);
}

function listing_status(): void
{
    ensure_listing_schema();
    $stmt = db()->query('SELECT state_json, updated_at FROM listing_control_state WHERE state_id = 1');
    $row = $stmt->fetch(PDO::FETCH_ASSOC);
    $state = $row ? json_decode((string)$row['state_json'], true) : null;
    worker_response(200, [
        'ok' => true,
        'state' => is_array($state) ? $state : null,
        'updated_at' => $row['updated_at'] ?? null,
    ]);
}

function automation_logs(): void
{
    ensure_worker_schema();
    $stmt = db()->query(
        'SELECT kind, message_id, chat_url, quantity, state, attempts,
                last_error, delivery_id, created_at, updated_at, completed_at
         FROM automation_dispatch_jobs
         ORDER BY updated_at DESC, job_id DESC
         LIMIT 100'
    );
    $logs = [];
    foreach ($stmt->fetchAll(PDO::FETCH_ASSOC) as $row) {
        $logs[] = [
            'kind' => (string)$row['kind'],
            'message_id' => (int)$row['message_id'],
            'chat_url' => (string)$row['chat_url'],
            'quantity' => $row['quantity'] !== null ? (int)$row['quantity'] : null,
            'state' => (string)$row['state'],
            'attempts' => (int)$row['attempts'],
            'has_detail' => trim((string)($row['last_error'] ?? '')) !== '',
            'delivery_id' => $row['delivery_id'] !== null ? (int)$row['delivery_id'] : null,
            'management_no' => $row['delivery_id'] !== null
                ? sprintf('DL-%06d', (int)$row['delivery_id'])
                : null,
            'created_at' => (string)$row['created_at'],
            'updated_at' => (string)$row['updated_at'],
            'completed_at' => $row['completed_at'] !== null ? (string)$row['completed_at'] : null,
        ];
    }
    worker_response(200, ['ok' => true, 'logs' => $logs]);
}

function worker_baseline(): array
{
    $baseline = ['review' => 128, 'delivery' => 134];
    $raw = @file_get_contents(WORKER_BASELINE_FILE);
    if ($raw !== false) {
        try {
            $value = json_decode($raw, true, 512, JSON_THROW_ON_ERROR);
            foreach (['review', 'delivery'] as $key) {
                if (isset($value[$key]) && is_numeric($value[$key])) {
                    $baseline[$key] = max(0, (int)$value[$key]);
                }
            }
        } catch (Throwable $error) {
            // Keep the conservative built-in baseline when the file is invalid.
        }
    }
    return $baseline;
}

function pull_job(): void
{
    ensure_worker_schema();
    $baseline = worker_baseline();
    $pdo = db();
    $pdo->beginTransaction();
    try {
        $stmt = $pdo->prepare(<<<'SQL'
            SELECT m.id, m.received_at, m.platform, m.chat_url, m.quantity,
                   CASE WHEN m.category IN ('review_received', 'review_required')
                        THEN 'review' ELSE 'delivery' END AS kind
            FROM delivery_inbox_messages AS m
            LEFT JOIN automation_dispatch_jobs AS j
              ON j.kind = CASE WHEN m.category IN ('review_received', 'review_required')
                               THEN 'review' ELSE 'delivery' END
             AND (j.message_id = m.id OR j.chat_url = m.chat_url)
            WHERE (
                    (m.category IN ('review_received', 'review_required')
                     AND m.id > :review_baseline)
                 OR (m.category = 'delivery_candidate' AND m.id > :delivery_baseline
                     AND m.payment_confirmed = 1 AND COALESCE(m.quantity, 0) > 0)
                  )
              AND m.source = 'delivery-inbox'
              AND COALESCE(m.chat_url, '') <> ''
              AND (
                    j.job_id IS NULL
                 OR (j.state = 'failed' AND j.kind = 'review'
                     AND j.attempts < 3 AND j.next_attempt_at <= NOW())
                  )
            ORDER BY m.received_at ASC, m.id ASC
            LIMIT 1
            FOR UPDATE
        SQL);
        $stmt->execute([
            'review_baseline' => $baseline['review'],
            'delivery_baseline' => $baseline['delivery'],
        ]);
        $candidate = $stmt->fetch(PDO::FETCH_ASSOC);
        if (!$candidate) {
            $pdo->commit();
            worker_response(200, ['ok' => true, 'job' => null]);
        }

        $existing = $pdo->prepare(
            'SELECT * FROM automation_dispatch_jobs
             WHERE kind = ? AND (message_id = ? OR chat_url = ?) FOR UPDATE'
        );
        $existing->execute([$candidate['kind'], $candidate['id'], $candidate['chat_url']]);
        $job = $existing->fetch(PDO::FETCH_ASSOC);
        if ($job) {
            $update = $pdo->prepare(
                "UPDATE automation_dispatch_jobs
                 SET state = 'claimed', attempts = attempts + 1, last_error = NULL,
                     next_attempt_at = NULL, updated_at = NOW()
                 WHERE job_id = ?"
            );
            $update->execute([(int)$job['job_id']]);
            $job['state'] = 'claimed';
            $job['attempts'] = (int)$job['attempts'] + 1;
            $job['last_error'] = null;
        } else {
            $insert = $pdo->prepare(
                'INSERT INTO automation_dispatch_jobs
                    (kind, message_id, chat_url, quantity, state)
                 VALUES (?, ?, ?, ?, \'claimed\')'
            );
            $insert->execute([
                $candidate['kind'],
                (int)$candidate['id'],
                $candidate['chat_url'],
                $candidate['quantity'] !== null ? (int)$candidate['quantity'] : null,
            ]);
            $job = [
                'job_id' => (int)$pdo->lastInsertId(),
                'kind' => $candidate['kind'],
                'message_id' => (int)$candidate['id'],
                'chat_url' => $candidate['chat_url'],
                'quantity' => $candidate['quantity'] !== null ? (int)$candidate['quantity'] : null,
                'attempts' => 1,
            ];
        }
        $pdo->commit();
        worker_response(200, [
            'ok' => true,
            'job' => [
                'job_id' => (int)$job['job_id'],
                'kind' => (string)$job['kind'],
                'message_id' => (int)$job['message_id'],
                'chat_url' => (string)$job['chat_url'],
                'quantity' => $job['quantity'] !== null ? (int)$job['quantity'] : null,
                'attempts' => (int)$job['attempts'],
            ],
        ]);
    } catch (Throwable $error) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        worker_response(500, ['ok' => false, 'error' => 'queue pull failed']);
    }
}

function acknowledge_job(array $payload): void
{
    ensure_worker_schema();
    $jobId = filter_var($payload['job_id'] ?? null, FILTER_VALIDATE_INT);
    $state = (string)($payload['state'] ?? '');
    if (!$jobId || !in_array($state, ['succeeded', 'failed', 'already_done'], true)) {
        worker_response(400, ['ok' => false, 'error' => 'invalid acknowledgement']);
    }
    $error = mb_substr(trim((string)($payload['error'] ?? '')), 0, 2000);
    $deliveryId = filter_var($payload['delivery_id'] ?? null, FILTER_VALIDATE_INT);
    $pdo = db();
    $stmt = $pdo->prepare('SELECT kind, attempts FROM automation_dispatch_jobs WHERE job_id = ?');
    $stmt->execute([(int)$jobId]);
    $job = $stmt->fetch(PDO::FETCH_ASSOC);
    if (!$job) {
        worker_response(404, ['ok' => false, 'error' => 'job not found']);
    }

    if ($state === 'failed' && $job['kind'] === 'review' && (int)$job['attempts'] < 3) {
        $update = $pdo->prepare(
            "UPDATE automation_dispatch_jobs
             SET state = 'failed', last_error = ?, next_attempt_at = DATE_ADD(NOW(), INTERVAL 5 MINUTE),
                 updated_at = NOW()
             WHERE job_id = ?"
        );
        $update->execute([$error ?: null, (int)$jobId]);
    } else {
        $update = $pdo->prepare(
            'UPDATE automation_dispatch_jobs
             SET state = ?, last_error = ?, delivery_id = ?, next_attempt_at = NULL,
                 updated_at = NOW(), completed_at = ?
             WHERE job_id = ?'
        );
        $completedAt = $state === 'failed' ? null : date('Y-m-d H:i:s');
        $update->execute([$state, $error ?: null, $deliveryId ?: null, $completedAt, (int)$jobId]);
    }
    worker_response(200, ['ok' => true]);
}

$payload = request_payload();
$operation = (string)($payload['operation'] ?? '');
if ($operation === 'pull') {
    pull_job();
}
if ($operation === 'ack') {
    acknowledge_job($payload);
}
if ($operation === 'pull_listing') {
    pull_listing_command();
}
if ($operation === 'ack_listing') {
    acknowledge_listing_command($payload);
}
if ($operation === 'enqueue_listing') {
    enqueue_listing_command($payload);
}
if ($operation === 'listing_heartbeat') {
    update_listing_state($payload);
}
if ($operation === 'listing_status') {
    listing_status();
}
if ($operation === 'automation_logs') {
    automation_logs();
}
worker_response(400, ['ok' => false, 'error' => 'unknown operation']);