<?php

namespace auctionforge\inc\sync;

use auctionforge\inc\InstanceDB;
use auctionforge\inc\Debug;

class SyncQueue
{
    const TABLE_NAME = 'af_sync_queue';

    const STATUS_PENDING    = 'pending';
    const STATUS_PROCESSING = 'processing';
    const STATUS_COMPLETED  = 'completed';
    const STATUS_FAILED     = 'failed';
    const STATUS_SKIPPED    = 'skipped';

    const TYPE_CATALOG_SYNC       = 'catalog_sync';
    const TYPE_ITEM_UPDATE        = 'item_update';
    const TYPE_FULL_SYNC          = 'full_sync';
    const TYPE_ARCHIVE_SYNC       = 'archive_sync';
    const TYPE_DAY_ENDED          = 'day_ended';
    const TYPE_FULL_SYNC_UPCOMING = 'full_sync_upcoming';
    const TYPE_FULL_SYNC_PAST     = 'full_sync_past';

    /**
     * Get the full table name with prefix
     * @return string
     */
    public static function getTableName()
    {
        $db = InstanceDB::getInstanceDB();
        return $db->prefix . self::TABLE_NAME;
    }

    /**
     * Enqueue a sync job. Deduplicates by auction_id + job_type if a pending job already exists.
     *
     * @param string $jobType     One of the TYPE_ constants
     * @param string $auctionId   BidSpirit auction ID
     * @param mixed  $payload     JSON-serializable payload data
     * @param int    $priority    1=highest, 5=lowest
     * @param string $source      'webhook', 'admin', 'cron', 'safety'
     * @return int|false  The job ID, or false if deduplicated/failed
     */
    public static function enqueue($jobType, $auctionId, $payload = null, $priority = 3, $source = 'webhook')
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();

        // Deduplication: skip if a pending/processing job exists for same auction + type
        // But auto-reset ghost "processing" rows whose worker process has died
        $existing = $db->get_row($db->prepare(
            "SELECT id, status, started_at, worker_pid FROM `{$table}` WHERE auction_id = %s AND job_type = %s AND status IN ('pending', 'processing') LIMIT 1",
            $auctionId,
            $jobType
        ));

        if ($existing) {
            // If the existing job is 'processing', check if it's actually stuck
            if ($existing->status === 'processing') {
                $staleSeconds = defined('AUCTIONFORGE_SYNC_STALE_JOB') ? AUCTIONFORGE_SYNC_STALE_JOB : 600;
                // Use a shorter threshold for dedup checks — 5 minutes
                $dedupStaleSeconds = min($staleSeconds, 300);
                $startedAt = strtotime($existing->started_at . ' UTC');
                $elapsed = time() - $startedAt;

                // Also check if the worker PID is still alive
                $pidAlive = true;
                if (!empty($existing->worker_pid) && function_exists('posix_kill')) {
                    $pidAlive = @posix_kill((int) $existing->worker_pid, 0);
                }

                if ($elapsed > $dedupStaleSeconds || !$pidAlive) {
                    // Ghost row: worker is dead or job is stale — reset it and allow new job
                    $reason = !$pidAlive ? "PID {$existing->worker_pid} dead" : "stale {$elapsed}s";
                    $db->update($table, [
                        'status'        => self::STATUS_FAILED,
                        'error_message' => "Auto-recovered: {$reason} (replaced by new {$source} job)",
                        'completed_at'  => current_time('mysql', true),
                        'updated_at'    => current_time('mysql', true),
                    ], ['id' => $existing->id], ['%s', '%s', '%s', '%s'], ['%d']);
                    SyncLogger::log('queue', "Auto-recovered stuck job #{$existing->id} ({$reason}) — allowing new {$jobType} for auction {$auctionId}", 'warning', $auctionId, $existing->id);
                } else {
                    SyncLogger::log('queue', "Deduplicated {$jobType} for auction {$auctionId} (existing job #{$existing->id}, {$existing->status} for {$elapsed}s)", 'info', $auctionId);
                    return false;
                }
            } else {
                // Existing job is 'pending' — safe to deduplicate
                SyncLogger::log('queue', "Deduplicated {$jobType} for auction {$auctionId} (existing pending job #{$existing->id})", 'info', $auctionId);
                return false;
            }
        }

        $payloadJson = is_string($payload) ? $payload : wp_json_encode($payload);

        $result = $db->insert($table, [
            'job_type'    => $jobType,
            'auction_id'  => $auctionId,
            'payload'     => $payloadJson,
            'status'      => self::STATUS_PENDING,
            'priority'    => $priority,
            'source'      => $source,
            'attempts'    => 0,
            'created_at'  => current_time('mysql', true),
            'updated_at'  => current_time('mysql', true),
        ], ['%s', '%s', '%s', '%s', '%d', '%s', '%d', '%s', '%s']);

        if ($result === false) {
            Debug::errorLog("SyncQueue::enqueue failed: " . $db->last_error);
            return false;
        }

        $jobId = $db->insert_id;
        SyncLogger::log('queue', "Enqueued {$jobType} for auction {$auctionId} (job #{$jobId}, source: {$source})", 'info', $auctionId, $jobId);

        return $jobId;
    }

    /**
     * Atomically claim the next pending job for processing.
     * Uses row-level locking to prevent race conditions.
     *
     * @param string|null $workerPid  Optional worker process ID for tracking
     * @return object|null  The claimed job row, or null if queue empty
     */
    public static function claimNext($workerPid = null)
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();

        // Atomic claim: UPDATE with WHERE status=pending, ORDER BY priority ASC, created_at ASC
        $now = current_time('mysql', true);
        $pid = $workerPid ?: getmypid();

        // Find the next job
        $job = $db->get_row(
            "SELECT * FROM `{$table}` WHERE status = 'pending' ORDER BY priority ASC, created_at ASC LIMIT 1"
        );

        if (!$job) {
            return null;
        }

        // Atomically claim it (only succeeds if still pending)
        $claimed = $db->query($db->prepare(
            "UPDATE `{$table}` SET status = 'processing', worker_pid = %s, started_at = %s, attempts = attempts + 1, updated_at = %s WHERE id = %d AND status = 'pending'",
            $pid,
            $now,
            $now,
            $job->id
        ));

        if ($claimed === 0) {
            // Another worker grabbed it — try again
            return self::claimNext($workerPid);
        }

        // Re-fetch with updated fields
        $job = $db->get_row($db->prepare("SELECT * FROM `{$table}` WHERE id = %d", $job->id));

        SyncLogger::log('worker', "Claimed job #{$job->id} ({$job->job_type} for auction {$job->auction_id})", 'info', $job->auction_id, $job->id);

        return $job;
    }

    /**
     * Mark a job as completed
     *
     * @param int    $jobId
     * @param string $resultSummary  Optional summary text
     * @return bool
     */
    public static function complete($jobId, $resultSummary = '')
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $now = current_time('mysql', true);

        $result = $db->update($table, [
            'status'         => self::STATUS_COMPLETED,
            'completed_at'   => $now,
            'result_summary' => $resultSummary,
            'updated_at'     => $now,
        ], ['id' => $jobId], ['%s', '%s', '%s', '%s'], ['%d']);

        if ($result !== false) {
            SyncLogger::log('worker', "Job #{$jobId} completed: {$resultSummary}", 'success', null, $jobId);
        }

        return $result !== false;
    }

    /**
     * Mark a job as failed
     *
     * @param int    $jobId
     * @param string $errorMessage
     * @return bool
     */
    public static function fail($jobId, $errorMessage = '')
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $now = current_time('mysql', true);

        $result = $db->update($table, [
            'status'         => self::STATUS_FAILED,
            'completed_at'   => $now,
            'error_message'  => $errorMessage,
            'updated_at'     => $now,
        ], ['id' => $jobId], ['%s', '%s', '%s', '%s'], ['%d']);

        if ($result !== false) {
            SyncLogger::log('worker', "Job #{$jobId} failed: {$errorMessage}", 'error', null, $jobId);
        }

        return $result !== false;
    }

    /**
     * Mark a job as skipped (e.g. delta check shows no changes)
     *
     * @param int    $jobId
     * @param string $reason
     * @return bool
     */
    public static function skip($jobId, $reason = '')
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $now = current_time('mysql', true);

        $result = $db->update($table, [
            'status'         => self::STATUS_SKIPPED,
            'completed_at'   => $now,
            'result_summary' => $reason,
            'updated_at'     => $now,
        ], ['id' => $jobId], ['%s', '%s', '%s', '%s'], ['%d']);

        return $result !== false;
    }

    /**
     * Reset stuck jobs (processing for longer than the stale threshold)
     *
     * @param int $staleSeconds  How long before a processing job is considered stuck (default: 600)
     * @return int  Number of jobs reset
     */
    public static function resetStuckJobs($staleSeconds = null)
    {
        if ($staleSeconds === null) {
            $staleSeconds = defined('AUCTIONFORGE_SYNC_STALE_JOB') ? AUCTIONFORGE_SYNC_STALE_JOB : 600;
        }

        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $now = current_time('mysql', true);
        $cutoff = gmdate('Y-m-d H:i:s', time() - $staleSeconds);

        // Use updated_at (refreshed by heartbeat) to detect truly stuck jobs.
        // Falls back to started_at for jobs that predate the heartbeat feature.
        $count = $db->query($db->prepare(
            "UPDATE `{$table}` SET status = 'failed', worker_pid = NULL, error_message = CONCAT('Auto-reset: stale >', %d, 's'), completed_at = %s, updated_at = %s WHERE status = 'processing' AND COALESCE(updated_at, started_at) < %s",
            $staleSeconds,
            $now,
            $now,
            $cutoff
        ));

        if ($count > 0) {
            SyncLogger::log('safety', "Reset {$count} stuck job(s) (stale > {$staleSeconds}s)", 'warning');
        }

        return (int) $count;
    }

    /**
     * Update the heartbeat timestamp for a processing job.
     * Call this periodically during long-running jobs so that
     * resetStuckJobs() knows the worker is still alive.
     *
     * @param int $jobId
     * @return void
     */
    public static function heartbeat($jobId)
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();

        $db->update($table, [
            'updated_at' => current_time('mysql', true),
        ], ['id' => $jobId, 'status' => self::STATUS_PROCESSING], ['%s'], ['%d', '%s']);
    }

    /**
     * Get counts by status
     *
     * @return array ['pending' => int, 'processing' => int, 'completed' => int, 'failed' => int, 'skipped' => int]
     */
    public static function getStatusCounts()
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();

        $rows = $db->get_results("SELECT status, COUNT(*) as cnt FROM `{$table}` GROUP BY status");

        $counts = [
            'pending'    => 0,
            'processing' => 0,
            'completed'  => 0,
            'failed'     => 0,
            'skipped'    => 0,
        ];

        if ($rows) {
            foreach ($rows as $row) {
                if (isset($counts[$row->status])) {
                    $counts[$row->status] = (int) $row->cnt;
                }
            }
        }

        return $counts;
    }

    /**
     * Get recent jobs (for admin display)
     *
     * @param int $limit
     * @return array
     */
    public static function getRecentJobs($limit = 25)
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();

        return $db->get_results($db->prepare(
            "SELECT * FROM `{$table}` ORDER BY created_at DESC LIMIT %d",
            $limit
        ));
    }

    /**
     * Check if there's an active worker (any processing jobs exist)
     *
     * @return bool
     */
    public static function hasActiveWorker()
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();

        $processingJobs = $db->get_results("SELECT id, worker_pid, started_at FROM `{$table}` WHERE status = 'processing'");

        if (empty($processingJobs)) {
            return false;
        }

        // Verify at least one processing job has a live worker process
        $staleSeconds = defined('AUCTIONFORGE_SYNC_STALE_JOB') ? AUCTIONFORGE_SYNC_STALE_JOB : 600;
        $now = time();

        foreach ($processingJobs as $job) {
            $elapsed = $now - strtotime($job->started_at . ' UTC');

            // Check if the worker PID is still alive (if posix functions available)
            if (!empty($job->worker_pid) && function_exists('posix_kill')) {
                $pidAlive = @posix_kill((int) $job->worker_pid, 0);
                if (!$pidAlive && $elapsed > 60) {
                    // PID is dead and job has been running for >60s — auto-reset
                    $db->update($table, [
                        'status'        => self::STATUS_FAILED,
                        'error_message' => "Auto-recovered: worker PID {$job->worker_pid} no longer running (elapsed {$elapsed}s)",
                        'completed_at'  => current_time('mysql', true),
                        'updated_at'    => current_time('mysql', true),
                    ], ['id' => $job->id], ['%s', '%s', '%s', '%s'], ['%d']);
                    SyncLogger::log('safety', "Auto-recovered job #{$job->id}: PID {$job->worker_pid} dead after {$elapsed}s", 'warning', null, $job->id);
                    continue; // This job is not active
                }
            }

            // If we can't check PID, fall back to stale threshold
            if ($elapsed < $staleSeconds) {
                return true; // This job appears legitimately active
            }
        }

        // All processing jobs are stale/dead
        return false;
    }

    /**
     * Spawn a background worker process
     *
     * @return bool  True if worker was spawned
     */
    
    public static function spawnWorker()
    {
        // Self-heal: reset any stuck processing jobs before spawning.
        // This ensures every webhook/spawn call cleans up ghost rows.
        self::resetStuckJobs();

        $restRoute = 'auctionforge/v1/sync-worker';

        // Strategy 1: HTTP loopback to 127.0.0.1 (fastest, avoids DNS/TLS).
        // Skippable via AUCTIONFORGE_WORKER_SKIP_LOOPBACK on hosts where
        // 127.0.0.1 + Host header lands on the wrong vhost — some
        // LiteSpeed/cPanel setups return 404/403 there, which silently
        // breaks the worker (see status-code handling below).
        $skipLoopback = defined('AUCTIONFORGE_WORKER_SKIP_LOOPBACK') && AUCTIONFORGE_WORKER_SKIP_LOOPBACK;
        if (!$skipLoopback) {
            $url  = 'http://127.0.0.1/' . ltrim(rest_get_url_prefix() . '/' . $restRoute, '/');
            $host = wp_parse_url(home_url(), PHP_URL_HOST);
            $result = wp_remote_post($url, [
                'timeout'   => 5,
                'blocking'  => true,
                'sslverify' => false,
                'headers'   => [
                    'Host'            => $host,
                    'X-AF-Worker-Key' => self::getWorkerKey(),
                ],
            ]);
            if (!is_wp_error($result)) {
                $code = (int) wp_remote_retrieve_response_code($result);
                // Only a 2xx means the worker endpoint actually ran. A 404/403
                // here means the loopback hit the wrong vhost or was blocked —
                // do NOT treat that as success (the old code did, and never
                // fell back). Fall through to the real HTTPS URL instead.
                if ($code >= 200 && $code < 300) {
                    SyncLogger::log('system', "Spawned background worker via HTTP loopback (HTTP {$code})", 'info');
                    return true;
                }
                SyncLogger::log('system', "Loopback returned HTTP {$code} (worker did not start) — trying HTTPS fallback...", 'warning');
            } else {
                SyncLogger::log('system', "Loopback failed ({$result->get_error_message()}) — trying HTTPS fallback...", 'warning');
            }
        }

        // Strategy 2: HTTPS via the site's real REST URL (no Host spoofing).
        $result = wp_remote_post(rest_url($restRoute), [
            'timeout'   => 5,
            'blocking'  => true,
            'sslverify' => false,
            'headers'   => [
                'X-AF-Worker-Key' => self::getWorkerKey(),
            ],
        ]);
        if (is_wp_error($result)) {
            SyncLogger::log('system', "Worker spawn HTTPS failed: " . $result->get_error_message(), 'error');
            return false;
        }
        $code = (int) wp_remote_retrieve_response_code($result);
        if ($code >= 200 && $code < 300) {
            SyncLogger::log('system', "Spawned background worker via HTTPS fallback (HTTP {$code})", 'info');
            return true;
        }
        SyncLogger::log('system', "Worker spawn failed — HTTPS returned HTTP {$code}", 'error');
        return false;
    }
    

    /**
     * Trigger the staging processor via a separate HTTP request.
     * InsertLotsBS doesn't work after fastcgi_finish_request(), so we
     * fire a fresh request that processes staging → WP posts + bp_meta_data.
     *
     * @param bool $blocking  When false, fire-and-forget: the processor runs
     *   in its own request (logging progress to af_sync_log) and we return
     *   immediately. Use this from cron / admin where a 300s blocking wait
     *   would exceed max_execution_time or a gateway timeout.
     * @return array ['total_inserted' => int|null, 'passes' => int|null]
     */
    public static function triggerStagingProcessor($blocking = true)
    {
        $url = rest_url('auctionforge/v1/sync-worker');

        $args = [
            'timeout'   => $blocking ? 300 : 1,
            'blocking'  => $blocking,
            'sslverify' => false,
            'headers'   => [
                'X-AF-Worker-Key' => self::getWorkerKey(),
                'X-AF-Action'     => 'process-staging',
            ],
        ];

        $response = wp_remote_post($url, $args);

        if (!$blocking) {
            // The processor runs server-side regardless of the early cutoff;
            // the (expected) WP_Error from the short timeout is ignored.
            return ['total_inserted' => null, 'passes' => null, 'async' => true];
        }

        if (is_wp_error($response)) {
            SyncLogger::log('system', "Staging processor HTTP failed: " . $response->get_error_message(), 'warning');
            return ['total_inserted' => 0, 'passes' => 0];
        }

        $body = json_decode(wp_remote_retrieve_body($response), true);
        return $body ?: ['total_inserted' => 0, 'passes' => 0];
    }

    /**
     * Count rows staged in import_history that have not been converted to WP
     * posts yet (syncFinished = 0). Drives the safety cron's staging pump and
     * the admin "Process Staging" action.
     *
     * @return int
     */
    public static function stagingBacklogCount()
    {
        $db = InstanceDB::getInstanceDB();
        $table = $db->prefix . 'import_history';
        if ($db->get_var("SHOW TABLES LIKE '{$table}'") !== $table) {
            return 0;
        }
        return (int) $db->get_var("SELECT COUNT(*) FROM `{$table}` WHERE syncFinished = 0");
    }

    /**
     * Generate a simple auth key for the worker endpoint
     * Uses wp_salt() so it's unique per site and not guessable.
     *
     * @return string
     */
    public static function getWorkerKey()
    {
        return substr(md5('af_sync_worker_' . wp_salt('auth')), 0, 32);
    }

    /**
     * Cancel a single job (set status to 'skipped' if pending, or force-fail if processing)
     *
     * @param int $jobId
     * @return bool
     */
    public static function cancelJob($jobId)
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $now = gmdate('Y-m-d H:i:s');

        $job = $db->get_row($db->prepare("SELECT * FROM `{$table}` WHERE id = %d", $jobId));
        if (!$job) return false;

        if ($job->status === 'pending') {
            $db->update($table, [
                'status' => 'skipped',
                'result_summary' => 'Cancelled by admin',
                'updated_at' => $now,
                'completed_at' => $now,
            ], ['id' => $jobId]);
            SyncLogger::log('queue', "Job #{$jobId} cancelled (was pending)", 'warning', $job->auction_id, $jobId);
            return true;
        }

        if ($job->status === 'processing') {
            $db->update($table, [
                'status' => 'failed',
                'error_message' => 'Force-killed by admin',
                'updated_at' => $now,
                'completed_at' => $now,
            ], ['id' => $jobId]);
            SyncLogger::log('queue', "Job #{$jobId} force-killed (was processing, PID {$job->worker_pid})", 'warning', $job->auction_id, $jobId);
            return true;
        }

        return false;
    }

    /**
     * Cancel all pending jobs and force-fail all processing jobs
     *
     * @return array ['cancelled' => int, 'killed' => int]
     */
    public static function cancelAll()
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $now = gmdate('Y-m-d H:i:s');

        $cancelled = (int) $db->query($db->prepare(
            "UPDATE `{$table}` SET status = 'skipped', result_summary = 'Bulk cancelled by admin', updated_at = %s, completed_at = %s WHERE status = 'pending'",
            $now, $now
        ));

        $killed = (int) $db->query($db->prepare(
            "UPDATE `{$table}` SET status = 'failed', error_message = 'Bulk force-killed by admin', updated_at = %s, completed_at = %s WHERE status = 'processing'",
            $now, $now
        ));

        if ($cancelled > 0 || $killed > 0) {
            SyncLogger::log('queue', "Bulk cancel: {$cancelled} pending cancelled, {$killed} processing killed", 'warning');
        }

        return ['cancelled' => $cancelled, 'killed' => $killed];
    }

    /**
     * Purge old completed/failed/skipped jobs older than $days
     *
     * @param int $days
     * @return int  Number of rows deleted
     */
    public static function purge($days = 30)
    {
        $db = InstanceDB::getInstanceDB();
        $table = self::getTableName();
        $cutoff = gmdate('Y-m-d H:i:s', time() - ($days * 86400));

        $count = $db->query($db->prepare(
            "DELETE FROM `{$table}` WHERE status IN ('completed', 'failed', 'skipped') AND created_at < %s",
            $cutoff
        ));

        return (int) $count;
    }
}
