<?php
/**
 * سودانا — QueueEngine v2
 * Retry with exponential backoff. MySQL 5.7+ compatible.
 * Unique event_id prevents duplicate task insertion.
 */
class QueueEngine {

    private const MAX_ATTEMPTS = 3;

    public static function enqueue(string $type, array $payload, ?string $runAt = null, ?string $eventId = null): int {
        try {
            $stmt = db()->prepare(
                "INSERT INTO tasks (type, payload, status, run_at, event_id, created_at)
                 VALUES (?, ?, 'pending', ?, ?, NOW())
                 ON DUPLICATE KEY UPDATE id=id"
            );
            $stmt->execute([
                $type,
                json_encode($payload, JSON_UNESCAPED_UNICODE),
                $runAt ?? date('Y-m-d H:i:s'),
                $eventId,
            ]);
            return (int)db()->lastInsertId();
        } catch (Throwable $e) {
            error_log("[QueueEngine::enqueue] " . $e->getMessage());
            throw $e;
        }
    }

    public static function fetch(int $limit = 5): array {
        // FIX 5: atomic claim — UPDATE first, then SELECT claimed rows
        // Eliminates race condition between SELECT and UPDATE
        try {
            $claimKey = uniqid('w_', true);
            // Step 1: Atomically mark up to $limit pending tasks as claimed
            $claimed = db()->prepare(
                "UPDATE tasks
                 SET status = 'processing', last_error = ?
                 WHERE status = 'pending'
                   AND run_at <= NOW()
                   AND attempts < " . self::MAX_ATTEMPTS . "
                 LIMIT $limit"
            );
            $claimed->execute([$claimKey]);
            if ($claimed->rowCount() === 0) return [];
            // Step 2: Fetch only the rows we just claimed
            $rows = db()->prepare(
                "SELECT * FROM tasks WHERE status='processing' AND last_error=? ORDER BY run_at ASC"
            );
            $rows->execute([$claimKey]);
            $tasks = $rows->fetchAll();
            // Clear the claim key so last_error shows real errors later
            if ($tasks) {
                $in = implode(',', array_fill(0, count($tasks), '?'));
                $ids = array_column($tasks, 'id');
                db()->prepare("UPDATE tasks SET last_error=NULL WHERE id IN ($in)")->execute($ids);
            }
            return $tasks;
        } catch (Throwable $e) {
            error_log("[QueueEngine::fetch] " . $e->getMessage());
            return [];
        }
    }

    public static function done(int $id): void {
        db()->prepare("UPDATE tasks SET status='done', updated_at=NOW() WHERE id=?")->execute([$id]);
    }

    public static function fail(int $id, string $error = ''): void {
        try {
            $row = db()->prepare("SELECT attempts FROM tasks WHERE id=? LIMIT 1");
            $row->execute([$id]);
            $attempts = (int)($row->fetchColumn() ?? 0);
            if ($attempts + 1 < self::MAX_ATTEMPTS) {
                $delay = pow(2, $attempts);
                $runAt = date('Y-m-d H:i:s', strtotime("+{$delay} minutes"));
                db()->prepare(
                    "UPDATE tasks SET status='pending', attempts=attempts+1, last_error=?, run_at=?, updated_at=NOW() WHERE id=?"
                )->execute([$error ?: null, $runAt, $id]);
                error_log("[QueueEngine] Task $id retry in {$delay}min");
            } else {
                db()->prepare(
                    "UPDATE tasks SET status='failed', attempts=attempts+1, last_error=?, failed_at=NOW(), updated_at=NOW() WHERE id=?"
                )->execute([$error ?: 'max_attempts_reached', $id]);
                try {
                    db()->prepare(
                        "INSERT INTO audit_logs (action, entity_type, entity_id, data, created_at) VALUES ('task.dead','task',?,?,NOW())"
                    )->execute([$id, json_encode(['error'=>$error], JSON_UNESCAPED_UNICODE)]);
                } catch (Throwable $ignored) {}
                error_log("[QueueEngine] Task $id DEAD: $error");
            }
        } catch (Throwable $e) {
            error_log("[QueueEngine::fail] " . $e->getMessage());
        }
    }

    public static function schedule(string $type, array $payload, string $runAt, ?string $eventId = null): int {
        return self::enqueue($type, $payload, $runAt, $eventId);
    }

    public static function depth(): int {
        try {
            return (int)db()->query("SELECT COUNT(*) FROM tasks WHERE status IN ('pending','processing')")->fetchColumn();
        } catch (Throwable $e) { return -1; }
    }

    /** Reset tasks stuck in 'processing' > 10 min (worker crash recovery) */
    public static function recoverStuck(): void {
        try {
            db()->exec(
                "UPDATE tasks SET status='pending', updated_at=NOW()
                 WHERE status='processing'
                   AND updated_at < DATE_SUB(NOW(), INTERVAL 10 MINUTE)
                   AND attempts < " . self::MAX_ATTEMPTS
            );
        } catch (Throwable $e) {
            error_log("[QueueEngine::recoverStuck] " . $e->getMessage());
        }
    }
}
