<?php

class GalleryImportQueue
{
    public const STALLED_AFTER = 1800;
    public const MAX_WORKERS = 3;

    public static function maxWorkers(): int
    {
        return max(1, min(self::MAX_WORKERS, (int) Setting::get('catalogue_max_workers', '3')));
    }

    public static function lockKey(int $slot): string
    {
        return 'gallery_import_worker_' . $slot;
    }

    public static function anyWorkerAlive(): bool
    {
        for ($slot = 0; $slot < self::maxWorkers(); $slot++) {
            $row = Database::fetchOne('SELECT IS_USED_LOCK(?) AS c', [self::lockKey($slot)]);
            if (!empty($row['c'])) {
                return true;
            }
        }
        return false;
    }

    public static function awaitWorker(int $timeoutMs = 600): bool
    {
        $deadline = microtime(true) + ($timeoutMs / 1000);
        do {
            if (self::anyWorkerAlive()) {
                return true;
            }
            usleep(60000);
        } while (microtime(true) < $deadline);

        return false;
    }

    public static function drainSlot(int $slot, int $maxSeconds = 0): string
    {
        if ($slot < 0 || $slot >= self::maxWorkers()) {
            return 'slot';
        }

        $lockKey = self::lockKey($slot);
        $lk = Database::fetchOne('SELECT GET_LOCK(?, 0) AS l', [$lockKey]);
        if (!$lk || (int) $lk['l'] !== 1) {
            return 'busy';
        }

        $started = time();
        $reason = 'empty';

        try {
            self::recoverStalled();
            $sid = MediaStorage::activeDefaultId() ?: null;

            while (($row = self::claim()) !== null) {
                self::importClaimedRow($row, $sid);

                if ($maxSeconds > 0 && (time() - $started) >= $maxSeconds) {
                    $reason = 'timeout';
                    break;
                }
            }
        } catch (\Throwable $e) {
            error_log('GalleryImportQueue::drainSlot: ' . $e->getMessage());
            $reason = 'error';
        } finally {
            Database::query('SELECT RELEASE_LOCK(?)', [$lockKey]);
        }

        return $reason;
    }

    private static function importClaimedRow(array $row, ?int $sid): void
    {
        require_once APP_ROOT . '/src/Admin/CatalogueController.php';

        $gid = (int) $row['gallery_id'];
        $importSettings = json_decode((string) ($row['import_settings'] ?? ''), true);
        $importSettings = is_array($importSettings) ? $importSettings : [];

        $ctx = [
            'inc' => [
                'description' => !array_key_exists('description', $importSettings) || !empty($importSettings['description']),
                'categories'  => !array_key_exists('categories', $importSettings)  || !empty($importSettings['categories']),
                'tags'        => !array_key_exists('tags', $importSettings)        || !empty($importSettings['tags']),
                'performers'  => !array_key_exists('performers', $importSettings)  || !empty($importSettings['performers']),
                'channels'    => !array_key_exists('channels', $importSettings)    || !empty($importSettings['channels']),
                'views'       => !empty($importSettings['views']),
            ],
            'sid'          => $sid,
            'default_type' => (string) ($row['video_type'] ?? ''),
            'max_images'   => 300,
            'user_id'      => (int) ($row['user_id'] ?? 0),
        ];

        try {
            $result = AdminCatalogueController::importOneGallery($gid, $ctx);
        } catch (\Throwable $e) {
            error_log('gallery-import-worker: gallery ' . $gid . ' threw: ' . $e->getMessage());
            $result = ['status' => 'error', 'gallery_id' => $gid, 'error' => $e->getMessage()];
        }

        self::finish((int) $row['id'], $result);

        $albumId = (int) ($result['album_id'] ?? 0);
        $ai = json_decode((string) ($row['ai_settings'] ?? ''), true);
        if ($result['status'] === 'imported' && $albumId > 0 && is_array($ai) && !empty($ai['enabled'])) {
            self::setAiStatus($albumId, 'processing');
            try {
                $out = CatalogueProcessor::applyAlbumAi($albumId, $ai);
                if (is_array($out) && isset($out['error'])) {
                    self::setAiStatus($albumId, 'failed', (string) $out['error']);
                } else {
                    self::setAiStatus($albumId, 'completed');
                }
            } catch (\Throwable $e) {
                error_log('gallery-import-worker: AI on album ' . $albumId . ' failed: ' . $e->getMessage());
                self::setAiStatus($albumId, 'failed', $e->getMessage());
            }
        }
    }

    private static function loopbackToken(): string
    {
        $tok = (string) Setting::get('gallery_worker_token', '');
        if ($tok === '') {
            $tok = bin2hex(random_bytes(20));
            Setting::set('gallery_worker_token', $tok);
        }
        return $tok;
    }

    private static function slotLease(int $slot): string
    {
        return APP_ROOT . '/storage/tmp/gallery-loop-' . $slot . '.lease';
    }

    private static function leaseFresh(int $slot, int $maxAge = 90): bool
    {
        $f = self::slotLease($slot);
        clearstatcache(false, $f);
        return is_file($f) && (time() - (int) @filemtime($f)) < $maxAge;
    }

    public static function fireLoopbackWorkers(int $desired = 0): bool
    {
        $base = rtrim((string) Setting::get('site_url', ''), '/');
        if ($base === '') {
            return false;
        }

        $stats = self::stats();
        $active = (int) ($stats['active'] ?? 0);
        if ($active < 1) {
            return false;
        }

        $maxWorkers = self::maxWorkers();
        $desired = $desired > 0 ? min($maxWorkers, $desired) : min($maxWorkers, $active);
        $desired = max(1, $desired);
        $token = self::loopbackToken();

        $mh = curl_multi_init();
        $handles = [];
        for ($i = 0; $i < $desired; $i++) {
            if (self::leaseFresh($i)) {
                continue;
            }
            $ch = curl_init($base . '/cron/gallery-worker?slot=' . $i . '&token=' . urlencode($token));
            curl_setopt_array($ch, [
                CURLOPT_RETURNTRANSFER    => true,
                CURLOPT_TIMEOUT_MS        => 1200,
                CURLOPT_CONNECTTIMEOUT_MS => 1000,
                CURLOPT_SSL_VERIFYPEER    => true,
                CURLOPT_HTTPHEADER        => ['Accept: application/json'],
                CURLOPT_USERAGENT         => 'TubePress-QueueWorker/1.0',
            ]);
            curl_multi_add_handle($mh, $ch);
            $handles[] = $ch;
        }

        if (!$handles) {
            curl_multi_close($mh);
            return true;
        }

        $deadline = microtime(true) + 1.3;
        do {
            curl_multi_exec($mh, $running);
            if ($running) {
                curl_multi_select($mh, 0.1);
            }
        } while ($running && microtime(true) < $deadline);

        foreach ($handles as $ch) {
            curl_multi_remove_handle($mh, $ch);
            curl_close($ch);
        }
        curl_multi_close($mh);

        return true;
    }

    public static function runLoopbackSlot(int $slot, string $token, bool $chain): void
    {
        $expected = (string) Setting::get('gallery_worker_token', '');
        if ($expected === '' || !hash_equals($expected, $token)) {
            http_response_code(404);
            echo 'Not found';
            exit;
        }
        if ($slot < 0 || $slot >= self::maxWorkers()) {
            http_response_code(204);
            exit;
        }
        if (!$chain && self::leaseFresh($slot)) {
            http_response_code(204);
            exit;
        }

        $lease = self::slotLease($slot);
        $dir = dirname($lease);
        if (!is_dir($dir)) {
            @mkdir($dir, 0755, true);
        }
        @touch($lease);

        ignore_user_abort(true);
        @set_time_limit(0);
        http_response_code(204);
        header('X-Robots-Tag: noindex');
        while (ob_get_level() > 0) {
            @ob_end_clean();
        }
        if (function_exists('fastcgi_finish_request')) {
            fastcgi_finish_request();
        } else {
            flush();
        }
        if (function_exists('session_write_close')) {
            @session_write_close();
        }

        $reason = self::drainSlot($slot, 45);

        $stats = self::stats();
        if ($reason === 'timeout' && (int) ($stats['active'] ?? 0) > 0) {
            @touch($lease);
            self::fireLoopbackChain($slot);
        } else {
            @unlink($lease);
        }
        exit;
    }

    private static function fireLoopbackChain(int $slot): void
    {
        $base = rtrim((string) Setting::get('site_url', ''), '/');
        if ($base === '') {
            @unlink(self::slotLease($slot));
            return;
        }
        $ch = curl_init($base . '/cron/gallery-worker?slot=' . $slot . '&token=' . urlencode(self::loopbackToken()) . '&chain=1');
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER    => true,
            CURLOPT_TIMEOUT_MS        => 1200,
            CURLOPT_CONNECTTIMEOUT_MS => 1000,
            CURLOPT_SSL_VERIFYPEER    => true,
            CURLOPT_HTTPHEADER        => ['Accept: application/json'],
            CURLOPT_USERAGENT         => 'TubePress-QueueWorker/1.0',
        ]);
        @curl_exec($ch);
        curl_close($ch);
    }

    // Schema upgrades run at the top of the front controller, but a request that
    // loses the migration lock can reach here with the new code and the old schema.
    // Money is taken before anything is queued, so this must not be the thing that
    // fails. The migration file is replayed rather than a copy of it kept here.
    private static function ensureTable(): void
    {
        static $checked = false;
        if ($checked) {
            return;
        }
        $checked = true;
        try {
            Database::fetchOne('SELECT 1 FROM catalogue_gallery_imports LIMIT 1');
        } catch (\PDOException) {
            foreach (['125_gallery_import_queue.sql', '126_gallery_import_queue_media.sql'] as $file) {
                $sql = @file_get_contents(APP_ROOT . '/sql/' . $file);
                if (!is_string($sql) || trim($sql) === '') {
                    continue;
                }
                foreach (array_filter(array_map('trim', explode(';', $sql))) as $statement) {
                    try {
                        Database::get()->exec($statement);
                    } catch (\PDOException $e) {
                        // a column or table already there is the expected case
                    }
                }
            }
        }
    }

    public static function paidGalleryIds(array $galleryIds): array
    {
        self::ensureTable();
        $ids = array_values(array_unique(array_filter(array_map('intval', $galleryIds), static fn($i) => $i > 0)));
        if (empty($ids)) {
            return [];
        }

        $ph = implode(',', array_fill(0, count($ids), '?'));
        $rows = Database::fetchAll(
            "SELECT gallery_id FROM catalogue_gallery_imports WHERE gallery_id IN ({$ph})",
            $ids
        );

        $paid = [];
        foreach ($rows as $row) {
            $paid[(int) $row['gallery_id']] = true;
        }
        return $paid;
    }

    public static function enqueue(array $galleryIds, array $opts): int
    {
        self::ensureTable();
        $ids = array_values(array_unique(array_filter(array_map('intval', $galleryIds), static fn($i) => $i > 0)));
        if (empty($ids)) {
            return 0;
        }

        $batchId = (string) ($opts['batch_id'] ?? '');
        $userId = (int) ($opts['user_id'] ?? 0);
        $videoType = (string) ($opts['video_type'] ?? '');
        $importSettings = json_encode((array) ($opts['import_settings'] ?? []), JSON_UNESCAPED_UNICODE);
        $aiSettings = isset($opts['ai_settings']) && is_array($opts['ai_settings'])
            ? json_encode($opts['ai_settings'], JSON_UNESCAPED_UNICODE)
            : null;
        $titles = (array) ($opts['titles'] ?? []);
        $thumbs = (array) ($opts['thumbnails'] ?? []);

        $queued = 0;
        foreach ($ids as $gid) {
            // A gallery already sitting in the queue is re-armed rather than duplicated,
            // so retrying a selection that partly failed picks up where it stopped.
            Database::query(
                'INSERT INTO catalogue_gallery_imports
                    (gallery_id, title, thumbnail_url, batch_id, user_id, status, video_type, import_settings, ai_settings)
                 VALUES (?, ?, ?, ?, ?, \'pending\', ?, ?, ?)
                 ON DUPLICATE KEY UPDATE
                    thumbnail_url = COALESCE(VALUES(thumbnail_url), thumbnail_url),
                    batch_id = VALUES(batch_id),
                    user_id = VALUES(user_id),
                    video_type = VALUES(video_type),
                    import_settings = VALUES(import_settings),
                    ai_settings = VALUES(ai_settings),
                    error_message = NULL,
                    attempt_count = IF(status IN (\'completed\', \'skipped\', \'importing\'), attempt_count, 0),
                    status = IF(status IN (\'completed\', \'skipped\', \'importing\'), status, \'pending\')',
                [$gid, mb_substr((string) ($titles[$gid] ?? ''), 0, 500),
                 (($thumbs[$gid] ?? '') !== '') ? mb_substr((string) $thumbs[$gid], 0, 1000) : null, $batchId, $userId ?: null, $videoType, $importSettings, $aiSettings]
            );
            $queued++;
        }
        return $queued;
    }

    public static function claim(string $batchId = ''): ?array
    {
        for ($attempt = 0; $attempt < 5; $attempt++) {
            $sql = 'SELECT * FROM catalogue_gallery_imports WHERE status = \'pending\'';
            $params = [];
            if ($batchId !== '') {
                $sql .= ' AND batch_id = ?';
                $params[] = $batchId;
            }
            $sql .= ' ORDER BY id LIMIT 1';
            $row = Database::fetchOne($sql, $params);
            if (!$row) {
                return null;
            }
            $stmt = Database::query(
                'UPDATE catalogue_gallery_imports
                    SET status = \'importing\', started_at = NOW(), attempt_count = attempt_count + 1
                  WHERE id = ? AND status = \'pending\'',
                [(int) $row['id']]
            );
            if ($stmt->rowCount() === 1) {
                $row['attempt_count'] = (int) $row['attempt_count'] + 1;
                $row['status'] = 'importing';
                return $row;
            }
        }
        return null;
    }

    public static function finish(int $id, array $result): void
    {
        $status = (string) ($result['status'] ?? 'error');
        if ($status === 'imported') {
            Database::query(
                'UPDATE catalogue_gallery_imports
                    SET status = \'completed\', album_id = ?, image_count = ?, title = IF(? = \'\', title, ?),
                        error_message = NULL, completed_at = NOW()
                  WHERE id = ?',
                [(int) ($result['album_id'] ?? 0), (int) ($result['images'] ?? 0),
                 (string) ($result['title'] ?? ''), mb_substr((string) ($result['title'] ?? ''), 0, 500), $id]
            );
            return;
        }
        if ($status === 'skipped') {
            Database::query(
                'UPDATE catalogue_gallery_imports
                    SET status = \'skipped\', album_id = ?, error_message = NULL, completed_at = NOW()
                  WHERE id = ?',
                [(int) ($result['album_id'] ?? 0) ?: null, $id]
            );
            return;
        }

        // Retry a failure until max_attempts, then leave it visible as failed.
        Database::query(
            'UPDATE catalogue_gallery_imports
                SET status = IF(attempt_count >= max_attempts, \'failed\', \'pending\'),
                    error_message = ?,
                    completed_at = IF(attempt_count >= max_attempts, NOW(), NULL)
              WHERE id = ?',
            [mb_substr((string) ($result['error'] ?? 'import failed'), 0, 1000), $id]
        );
    }

    public static function recoverStalled(int $olderThanSeconds = self::STALLED_AFTER): int
    {
        $stmt = Database::query(
            'UPDATE catalogue_gallery_imports
                SET status = IF(attempt_count >= max_attempts, \'failed\', \'pending\'),
                    error_message = \'interrupted, retrying\'
              WHERE status = \'importing\' AND started_at < (NOW() - INTERVAL ? SECOND)',
            [$olderThanSeconds]
        );
        return $stmt->rowCount();
    }

    public static function stats(): array
    {
        self::ensureTable();
        $out = ['pending' => 0, 'importing' => 0, 'completed' => 0, 'failed' => 0, 'skipped' => 0, 'cancelled' => 0];
        foreach (Database::fetchAll('SELECT status, COUNT(*) AS c FROM catalogue_gallery_imports GROUP BY status') as $r) {
            $out[(string) $r['status']] = (int) $r['c'];
        }
        $out['total'] = array_sum($out);
        $out['active'] = $out['pending'] + $out['importing'];
        return $out;
    }

    public static function rows(int $limit = 100, string $status = ''): array
    {
        self::ensureTable();
        $limit = max(1, min(500, $limit));
        $sql = 'SELECT id, gallery_id, title, thumbnail_url, batch_id, status, album_id, image_count,
                       ai_status, ai_error, attempt_count, max_attempts,
                       error_message, created_at, started_at, completed_at
                  FROM catalogue_gallery_imports';
        $params = [];
        if (in_array($status, ['pending', 'importing', 'completed', 'failed', 'skipped', 'cancelled'], true)) {
            $sql .= ' WHERE status = ?';
            $params[] = $status;
        }
        $sql .= ' ORDER BY FIELD(status, \'importing\', \'pending\', \'failed\', \'completed\', \'skipped\', \'cancelled\'), id DESC LIMIT ' . $limit;
        return Database::fetchAll($sql, $params);
    }

    public static function retryFailed(): int
    {
        $stmt = Database::query(
            'UPDATE catalogue_gallery_imports
                SET status = \'pending\', attempt_count = 0, error_message = NULL, completed_at = NULL
              WHERE status IN (\'failed\', \'cancelled\')'
        );
        return $stmt->rowCount();
    }

    public static function cancelPending(): int
    {
        $stmt = Database::query('UPDATE catalogue_gallery_imports SET status = \'cancelled\' WHERE status = \'pending\'');
        return $stmt->rowCount();
    }

    public static function clearFinished(): int
    {
        $stmt = Database::query('DELETE FROM catalogue_gallery_imports WHERE status IN (\'completed\', \'skipped\', \'cancelled\')');
        return $stmt->rowCount();
    }

    public static function markAiQueued(array $albumIds): void
    {
        $albumIds = array_values(array_filter(array_map('intval', $albumIds), static fn($i) => $i > 0));
        if (empty($albumIds)) {
            return;
        }
        $in = implode(',', array_fill(0, count($albumIds), '?'));
        Database::query('UPDATE catalogue_gallery_imports SET ai_status = \'pending\' WHERE album_id IN (' . $in . ')', $albumIds);
    }

    public static function setAiStatus(int $albumId, string $status, string $error = ''): void
    {
        if ($albumId < 1) {
            return;
        }
        Database::query(
            'UPDATE catalogue_gallery_imports SET ai_status = ?, ai_error = ? WHERE album_id = ?',
            [$status, $error !== '' ? mb_substr($error, 0, 500) : null, $albumId]
        );
    }

    public static function retryOne(int $id): bool
    {
        $stmt = Database::query(
            'UPDATE catalogue_gallery_imports
                SET status = \'pending\', attempt_count = 0, error_message = NULL, completed_at = NULL
              WHERE id = ? AND status IN (\'failed\', \'cancelled\')',
            [$id]
        );
        return $stmt->rowCount() === 1;
    }

    public static function cancelOne(int $id): bool
    {
        $stmt = Database::query(
            'UPDATE catalogue_gallery_imports SET status = \'cancelled\', completed_at = NOW()
              WHERE id = ? AND status IN (\'pending\', \'importing\')',
            [$id]
        );
        return $stmt->rowCount() === 1;
    }

    public static function deleteOne(int $id): bool
    {
        $stmt = Database::query(
            'DELETE FROM catalogue_gallery_imports WHERE id = ? AND status IN (\'completed\', \'skipped\', \'failed\', \'cancelled\')',
            [$id]
        );
        return $stmt->rowCount() === 1;
    }
}
