<?php

/**
 * Reliable broadcast / forward-all queue (batch send + self-chaining worker).
 */
class BroadcastQueueService
{
    private const BATCH_SIZE = 25;
    private const MAX_SECONDS_PER_RUN = 12;
    private const USLEEP_MICROS = 50000;
    private const LOCK_NAME = 'wizwiz_broadcast_queue';

    public static function ensureSchema(mysqli $connection): void
    {
        $res = $connection->query("SHOW COLUMNS FROM `send_list` LIKE 'updated_at'");
        if ($res && $res->num_rows === 0) {
            $connection->query('ALTER TABLE `send_list` ADD COLUMN `updated_at` INT NOT NULL DEFAULT 0');
        }
    }

    /** @return array<string, mixed>|null */
    public static function getActiveJob(mysqli $connection): ?array
    {
        self::ensureSchema($connection);
        $stmt = $connection->prepare('SELECT * FROM `send_list` WHERE `state` = 1 ORDER BY `id` ASC LIMIT 1');
        $stmt->execute();
        $row = $stmt->get_result()->fetch_assoc();
        $stmt->close();
        return $row ?: null;
    }

    public static function countUsers(mysqli $connection): int
    {
        $stmt = $connection->prepare('SELECT COUNT(*) AS c FROM `users`');
        $stmt->execute();
        $count = (int) ($stmt->get_result()->fetch_assoc()['c'] ?? 0);
        $stmt->close();
        return $count;
    }

    /** @return array{active:bool,type?:string,offset:int,total:int,remaining:int,stale:bool} */
    public static function getStatus(mysqli $connection): array
    {
        $job = self::getActiveJob($connection);
        if (!$job) {
            return ['active' => false, 'offset' => 0, 'total' => 0, 'remaining' => 0, 'stale' => false];
        }
        $total = self::countUsers($connection);
        $offset = (int) ($job['offset'] ?? 0);
        $updatedAt = (int) ($job['updated_at'] ?? 0);
        $stale = $offset === 0 && $updatedAt > 0 && (time() - $updatedAt) > 90;
        if ($offset > 0 && $updatedAt > 0 && (time() - $updatedAt) > 600) {
            $stale = true;
        }
        return [
            'active' => true,
            'type' => (string) ($job['type'] ?? ''),
            'offset' => $offset,
            'total' => $total,
            'remaining' => max(0, $total - $offset),
            'stale' => $stale,
            'id' => (int) ($job['id'] ?? 0),
        ];
    }

    public static function formatStatusMessage(array $status): string
    {
        if (!$status['active']) {
            return '';
        }
        $type = ($status['type'] ?? '') === 'forwardall' ? 'فروارد' : 'پیام';
        $sentLabel = ($status['type'] ?? '') === 'forwardall' ? 'فروارد شده' : 'ارسال شده';
        $lines = [
            "❗️ یک {$type} همگانی در صف انتشار است.",
            '',
            "🔰 تعداد کاربران: {$status['total']}",
            "☑️ {$sentLabel}: {$status['offset']}",
            "📣 باقیمانده: {$status['remaining']}",
        ];
        if (($status['offset'] ?? 0) === 0) {
            $lines[] = '';
            $lines[] = '💡 اگر ارسال شروع نشد، «🔄 ادامه ارسال» را بزنید.';
        }
        if (!empty($status['stale'])) {
            $lines[] = '';
            $lines[] = '⚠️ صف گیر کرده به نظر می‌رسد. «لغو صف» بزنید و دوباره ارسال کنید.';
        }
        return implode("\n", $lines);
    }

    public static function cancelActive(mysqli $connection): bool
    {
        $stmt = $connection->prepare('DELETE FROM `send_list` WHERE `state` = 1');
        $stmt->execute();
        $deleted = $stmt->affected_rows > 0;
        $stmt->close();
        return $deleted;
    }

    public static function cancelById(mysqli $connection, int $id): void
    {
        $stmt = $connection->prepare('DELETE FROM `send_list` WHERE `id` = ?');
        $stmt->bind_param('i', $id);
        $stmt->execute();
        $stmt->close();
    }

    public static function activate(mysqli $connection, int $id): void
    {
        self::ensureSchema($connection);
        $now = time();
        $stmt = $connection->prepare('UPDATE `send_list` SET `state` = 1, `offset` = 0, `updated_at` = ? WHERE `id` = ?');
        $stmt->bind_param('ii', $now, $id);
        $stmt->execute();
        $stmt->close();
    }

    public static function triggerWorker(string $botUrl): void
    {
        self::triggerWorkerHttp($botUrl);
        self::triggerWorkerCli();
    }

    public static function resumeProcessing(string $botUrl, mysqli $connection): void
    {
        self::triggerWorker($botUrl);
        if (function_exists('fastcgi_finish_request')) {
            @fastcgi_finish_request();
        }
        @ignore_user_abort(true);
        @set_time_limit(max(30, self::MAX_SECONDS_PER_RUN + 18));
        self::runWorker($connection);
    }

    private static function workerUrl(string $botUrl): string
    {
        return rtrim($botUrl, '/') . '/settings/messagewizwiz.php';
    }

    private static function triggerWorkerHttp(string $botUrl): void
    {
        if ($botUrl === '') {
            return;
        }
        $url = self::workerUrl($botUrl);
        if (!function_exists('curl_init')) {
            @file_get_contents($url, false, stream_context_create([
                'http' => ['timeout' => 1, 'ignore_errors' => true],
            ]));
            return;
        }
        $ch = curl_init($url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_FOLLOWLOCATION, true);
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false);
        curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, 0);
        curl_setopt($ch, CURLOPT_NOSIGNAL, 1);
        if (defined('CURLOPT_TIMEOUT_MS')) {
            curl_setopt($ch, CURLOPT_TIMEOUT_MS, 1500);
            curl_setopt($ch, CURLOPT_CONNECTTIMEOUT_MS, 1500);
        } else {
            curl_setopt($ch, CURLOPT_TIMEOUT, 2);
            curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 2);
        }
        @curl_exec($ch);
        curl_close($ch);
    }

    private static function triggerWorkerCli(): void
    {
        $script = dirname(__DIR__) . '/settings/messagewizwiz.php';
        if (!is_file($script)) {
            return;
        }
        $php = PHP_BINARY;
        if (!is_string($php) || $php === '') {
            $php = 'php';
        }
        if (strtoupper(substr(PHP_OS, 0, 3)) === 'WIN') {
            if (function_exists('popen')) {
                pclose(popen('start /B "" ' . escapeshellarg($php) . ' ' . escapeshellarg($script), 'r'));
            }
            return;
        }
        if (function_exists('exec')) {
            exec(escapeshellarg($php) . ' ' . escapeshellarg($script) . ' > /dev/null 2>&1 &');
        }
    }

    /** @return array{processed:int,continues:bool,finished:bool} */
    public static function runWorker(mysqli $connection): array
    {
        global $admin, $buttonValues, $botUrl;

        self::ensureSchema($connection);
        if (!self::acquireLock($connection)) {
            return ['processed' => 0, 'continues' => false, 'finished' => false];
        }

        $processed = 0;
        $continues = false;
        $finished = false;
        $deadline = time() + self::MAX_SECONDS_PER_RUN;

        try {
            while (time() < $deadline) {
                $batch = self::processBatch($connection);
                $processed += $batch['sent'];
                if ($batch['finished']) {
                    $finished = true;
                    $continues = false;
                    break;
                }
                if (!$batch['had_work']) {
                    break;
                }
                $continues = true;
            }
            if (!$finished && self::getActiveJob($connection) !== null) {
                $continues = true;
            }
        } finally {
            self::releaseLock($connection);
        }

        if ($continues && !empty($botUrl)) {
            self::triggerWorker($botUrl);
        }

        return ['processed' => $processed, 'continues' => $continues, 'finished' => $finished];
    }

    private static function acquireLock(mysqli $connection): bool
    {
        $name = self::LOCK_NAME;
        $res = $connection->query("SELECT GET_LOCK('{$name}', 0) AS l");
        $row = $res ? $res->fetch_assoc() : null;
        if (isset($row['l']) && (int) $row['l'] === 1) {
            return true;
        }

        $lockFile = dirname(__DIR__) . '/settings/broadcast.worker.lock';
        $fp = @fopen($lockFile, 'c+');
        if ($fp !== false && @flock($fp, LOCK_EX | LOCK_NB)) {
            $GLOBALS['_wizwiz_broadcast_lock_fp'] = $fp;
            return true;
        }
        if (is_resource($fp)) {
            fclose($fp);
        }

        return false;
    }

    private static function releaseLock(mysqli $connection): void
    {
        $name = self::LOCK_NAME;
        $connection->query("SELECT RELEASE_LOCK('{$name}')");
        if (!empty($GLOBALS['_wizwiz_broadcast_lock_fp']) && is_resource($GLOBALS['_wizwiz_broadcast_lock_fp'])) {
            @flock($GLOBALS['_wizwiz_broadcast_lock_fp'], LOCK_UN);
            @fclose($GLOBALS['_wizwiz_broadcast_lock_fp']);
            unset($GLOBALS['_wizwiz_broadcast_lock_fp']);
        }
    }

    /** @return array{sent:int,had_work:bool,finished:bool} */
    private static function processBatch(mysqli $connection): array
    {
        global $admin, $buttonValues;

        $job = self::getActiveJob($connection);
        if (!$job) {
            return ['sent' => 0, 'had_work' => false, 'finished' => true];
        }

        $sendId = (int) $job['id'];
        $offset = (int) ($job['offset'] ?? 0);
        $type = (string) ($job['type'] ?? '');
        $fileId = (string) ($job['file_id'] ?? '');
        $chatId = (string) ($job['chat_id'] ?? '');
        $text = (string) ($job['text'] ?? '');
        $fwdMessageId = (int) ($job['message_id'] ?? 0);
        $now = time();

        if ($offset === 0) {
            $startMsg = $type === 'forwardall'
                ? 'عملیات هدایت همگانی شروع شد'
                : 'عملیات ارسال پیام همگانی شروع شد';
            if (!empty($admin)) {
                bot('sendMessage', ['chat_id' => $admin, 'text' => $startMsg]);
            }
        }

        $stmt = $connection->prepare(
            'SELECT `userid` FROM `users` ORDER BY `id` ASC LIMIT '
            . self::BATCH_SIZE
            . ' OFFSET '
            . max(0, $offset)
        );
        $stmt->execute();
        $usersList = $stmt->get_result();
        $stmt->close();

        if ($usersList->num_rows === 0) {
            $doneMsg = $type === 'forwardall'
                ? 'عملیات هدایت همگانی با موفقیت انجام شد'
                : 'عملیات ارسال پیام همگانی با موفقیت انجام شد';
            if (!empty($admin)) {
                bot('sendMessage', [
                    'chat_id' => $admin,
                    'text' => $doneMsg . "\nبه " . $offset . ' نفر پیام ارسال شد',
                ]);
            }
            $stmt = $connection->prepare('DELETE FROM `send_list` WHERE `id` = ?');
            $stmt->bind_param('i', $sendId);
            $stmt->execute();
            $stmt->close();
            return ['sent' => 0, 'had_work' => true, 'finished' => true];
        }

        $keys = json_encode([
            'inline_keyboard' => [
                [['text' => $buttonValues['start_bot'] ?? 'شروع', 'callback_data' => 'mainMenu']],
            ],
        ]);

        $sent = 0;
        while ($user = $usersList->fetch_assoc()) {
            $userId = (int) $user['userid'];
            if ($userId <= 0) {
                $offset++;
                continue;
            }
            self::deliverToUser($type, $userId, $text, $fileId, $chatId, $fwdMessageId, $keys);
            $offset++;
            $sent++;
            usleep(self::USLEEP_MICROS);
        }

        $stmt = $connection->prepare('UPDATE `send_list` SET `offset` = ?, `updated_at` = ? WHERE `id` = ?');
        $stmt->bind_param('iii', $offset, $now, $sendId);
        $stmt->execute();
        $stmt->close();

        return ['sent' => $sent, 'had_work' => true, 'finished' => false];
    }

    private static function deliverToUser(
        string $type,
        int $userId,
        string $text,
        string $fileId,
        string $chatId,
        int $fwdMessageId,
        string $keys
    ): void {
        switch ($type) {
            case 'text':
                sendMessage($text, $keys, null, $userId);
                break;
            case 'music':
                bot('sendAudio', [
                    'chat_id' => $userId,
                    'audio' => $fileId,
                    'caption' => $text,
                    'reply_markup' => $keys,
                ]);
                break;
            case 'video':
                bot('sendVideo', [
                    'chat_id' => $userId,
                    'video' => $fileId,
                    'caption' => $text,
                    'reply_markup' => $keys,
                ]);
                break;
            case 'voice':
                bot('sendVoice', [
                    'chat_id' => $userId,
                    'voice' => $fileId,
                    'caption' => $text,
                    'reply_markup' => $keys,
                ]);
                break;
            case 'photo':
                bot('sendPhoto', [
                    'chat_id' => $userId,
                    'photo' => $fileId,
                    'caption' => $text,
                    'reply_markup' => $keys,
                ]);
                break;
            case 'forwardall':
                forwardmessage($userId, $chatId, $fwdMessageId);
                break;
            default:
                bot('sendDocument', [
                    'chat_id' => $userId,
                    'document' => $fileId,
                    'caption' => $text,
                    'reply_markup' => $keys,
                ]);
                break;
        }
    }
}
