<?php
// Queue-based public broadcast/forward sender for Telegram.
// Sends in controlled batches to avoid Telegram request limits.
// Telegram allows higher throughput than Bale, but shared hosting is safer with 20 requests per cron run.
if(!defined('BOT_BROADCAST_PER_RUN')) define('BOT_BROADCAST_PER_RUN', 20);
if(!defined('BOT_BROADCAST_MAX_PER_RUN')) define('BOT_BROADCAST_MAX_PER_RUN', 20);
if(!defined('BOT_TELEGRAM_BROADCAST_SAFE_RPS')) define('BOT_TELEGRAM_BROADCAST_SAFE_RPS', 20);
if(!function_exists('bot_scale_get_user_ids_fast') && is_file(__DIR__ . '/../app/Services/ScaleOptimizerService.php')){ require_once __DIR__ . '/../app/Services/ScaleOptimizerService.php'; if(function_exists('bot_scale_bootstrap')) bot_scale_bootstrap(); }

if(!function_exists('bot_broadcast_cache_dir')){
function bot_broadcast_cache_dir(){
    $dir = dirname(__DIR__) . '/cache';
    if(!is_dir($dir)) @mkdir($dir, 0775, true);
    return $dir;
}
function bot_broadcast_queue_file(){ return bot_broadcast_cache_dir() . '/broadcast_queue.json'; }
function bot_broadcast_lock_file(){ return bot_broadcast_cache_dir() . '/broadcast_queue.lock'; }
function bot_broadcast_read_queue(){
    $file = bot_broadcast_queue_file();
    if(!is_file($file)) return ['jobs'=>[]];
    $data = json_decode((string)@file_get_contents($file), true);
    if(!is_array($data)) $data = ['jobs'=>[]];
    if(!isset($data['jobs']) || !is_array($data['jobs'])) $data['jobs'] = [];
    return $data;
}
function bot_broadcast_write_queue($queue){
    if(!is_array($queue)) $queue = ['jobs'=>[]];
    if(!isset($queue['jobs']) || !is_array($queue['jobs'])) $queue['jobs'] = [];
    $file = bot_broadcast_queue_file();
    $tmp = $file . '.tmp';
    @file_put_contents($tmp, json_encode($queue, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT), LOCK_EX);
    @rename($tmp, $file);
    return true;
}
function bot_broadcast_user_ids($useCache = true){
    // 100k-user mode: use the shared scale cache and SQL SELECT instead of scanning/decoding files.
    if(function_exists('bot_scale_get_user_ids_fast')){
        return bot_scale_get_user_ids_fast($useCache ? 300 : 0);
    }
    $cache = bot_broadcast_cache_dir() . '/broadcast_user_ids.json';
    if($useCache && is_file($cache)){
        $data = json_decode((string)@file_get_contents($cache), true);
        if(is_array($data) && isset($data['created_at'], $data['ids']) && is_array($data['ids']) && time() - (int)$data['created_at'] < 300){
            return $data['ids'];
        }
    }
    $ids = [];
    if(function_exists('bot_data_glob')) $files = bot_data_glob('melat/*.json');
    else $files = glob(dirname(__DIR__).'/melat/*.json') ?: [];
    foreach($files as $file){
        $id = basename($file, '.json');
        if(is_numeric($id) && $id !== '0') $ids[] = (string)$id;
    }
    $ids = array_values(array_unique($ids));
    @file_put_contents($cache, json_encode(['created_at'=>time(), 'ids'=>$ids], JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT), LOCK_EX);
    return $ids;
}
function bot_broadcast_enqueue($type, $adminId, $fromChatId, $messageId){
    $type = $type === 'forward' ? 'forward' : 'copy';
    $ids = bot_broadcast_user_ids(false);
    $queue = bot_broadcast_read_queue();
    $job = [
        'id' => 'bc_' . date('Ymd_His') . '_' . substr(sha1((string)$adminId.'|'.(string)$messageId.'|'.microtime(true)),0,6),
        'type' => $type,
        'admin_id' => (string)$adminId,
        'from_chat_id' => (string)$fromChatId,
        'message_id' => (int)$messageId,
        'status' => 'queued',
        'created_at' => time(),
        'started_at' => 0,
        'finished_at' => 0,
        'offset' => 0,
        'total' => count($ids),
        'success' => 0,
        'failed' => 0,
        'last_error' => '',
        'last_run_at' => 0,
    ];
    $queue['jobs'][] = $job;
    // Keep the queue file compact; preserve last 30 jobs.
    if(count($queue['jobs']) > 30) $queue['jobs'] = array_slice($queue['jobs'], -30);
    bot_broadcast_write_queue($queue);
    return $job;
}

function bot_broadcast_enqueue_text($adminId, $text, $target = 'all'){
    $ids = bot_broadcast_user_ids(false);
    $queue = bot_broadcast_read_queue();
    $job = [
        'id' => 'bc_txt_' . date('Ymd_His') . '_' . substr(sha1((string)$adminId.'|'.(string)$text.'|'.microtime(true)),0,6),
        'type' => 'text',
        'target' => (string)$target,
        'admin_id' => (string)$adminId,
        'text' => (string)$text,
        'status' => 'queued',
        'created_at' => time(),
        'started_at' => 0,
        'finished_at' => 0,
        'offset' => 0,
        'total' => count($ids),
        'success' => 0,
        'failed' => 0,
        'last_error' => '',
        'last_run_at' => 0,
    ];
    $queue['jobs'][] = $job;
    if(count($queue['jobs']) > 30) $queue['jobs'] = array_slice($queue['jobs'], -30);
    bot_broadcast_write_queue($queue);
    return $job;
}
function bot_broadcast_enqueue_coin($adminId, $amount, $notifyText = ''){
    $amount = (int)$amount;
    if($amount <= 0) return ['ok'=>false, 'error'=>'invalid_amount'];
    $ids = bot_broadcast_user_ids(false);
    $queue = bot_broadcast_read_queue();
    $job = [
        'id' => 'coin_' . date('Ymd_His') . '_' . substr(sha1((string)$adminId.'|'.$amount.'|'.microtime(true)),0,6),
        'type' => 'coin',
        'admin_id' => (string)$adminId,
        'amount' => $amount,
        'notify_text' => (string)$notifyText,
        'status' => 'queued',
        'created_at' => time(),
        'started_at' => 0,
        'finished_at' => 0,
        'offset' => 0,
        'total' => count($ids),
        'success' => 0,
        'failed' => 0,
        'notify_failed' => 0,
        'last_error' => '',
        'last_run_at' => 0,
    ];
    $queue['jobs'][] = $job;
    if(count($queue['jobs']) > 30) $queue['jobs'] = array_slice($queue['jobs'], -30);
    bot_broadcast_write_queue($queue);
    return $job;
}

function bot_broadcast_coin_notify_text($job, $balanceAfter = null){
    global $money;
    $asset = isset($money) && trim((string)$money) !== '' ? (string)$money : 'الماس';
    $amount = (int)($job['amount'] ?? 0);
    $text = trim((string)($job['notify_text'] ?? ''));
    if($text === ''){
        $text = "🎁 هدیه همگانی مدیریت\n\n+{$amount} {$asset} به حساب شما اضافه شد.";
        if($balanceAfter !== null) $text .= "\n💰 موجودی فعلی: {$balanceAfter} {$asset}";
    }else{
        $text = str_replace(['AMOUNT','BALANCE','MONEY'], [(string)$amount, (string)($balanceAfter ?? ''), $asset], $text);
    }
    return $text;
}
function bot_broadcast_current_job(&$queue = null, &$index = null){
    $queue = bot_broadcast_read_queue();
    foreach($queue['jobs'] as $i=>$job){
        $status = (string)($job['status'] ?? 'queued');
        if($status === 'queued' || $status === 'running'){
            $index = $i;
            return $job;
        }
    }
    $index = null;
    return null;
}
function bot_broadcast_cancel_current_job($adminId = ''){
    $queue = null; $idx = null; $job = bot_broadcast_current_job($queue, $idx);
    if(!$job || $idx === null){
        return ['ok'=>false, 'error'=>'empty'];
    }
    $job['status'] = 'cancelled';
    $job['cancelled_at'] = time();
    $job['cancelled_by'] = (string)$adminId;
    $job['last_error'] = 'cancelled_by_admin';
    $queue['jobs'][$idx] = $job;
    bot_broadcast_write_queue($queue);
    return ['ok'=>true, 'job_id'=>(string)($job['id'] ?? ''), 'offset'=>(int)($job['offset'] ?? 0), 'total'=>(int)($job['total'] ?? 0), 'type'=>(string)($job['type'] ?? '')];
}

function bot_broadcast_status_text(){
    $queue = bot_broadcast_read_queue();
    $jobs = $queue['jobs'];
    if(!$jobs) return "📭 **صف ارسال همگانی خالی است.**";
    $text = "📊 **وضعیت ارسال همگانی**\n━━━━━━━━━━━━━━\n";
    $shown = 0;
    for($i=count($jobs)-1; $i>=0 && $shown<5; $i--, $shown++){
        $j = $jobs[$i];
        $status = (string)($j['status'] ?? 'queued');
        $icon = $status === 'done' ? '✅' : ($status === 'running' ? '▶️' : ($status === 'cancelled' ? '⛔️' : ($status === 'failed' ? '❌' : '⏳')));
        $total = max(0, (int)($j['total'] ?? 0));
        $offset = max(0, (int)($j['offset'] ?? 0));
        $pct = $total > 0 ? (int)floor(min(100, ($offset/$total)*100)) : 0;
        $type = (($j['type'] ?? 'copy') === 'forward') ? 'فوروارد' : ((($j['type'] ?? 'copy') === 'text') ? 'متنی' : ((($j['type'] ?? 'copy') === 'coin') ? 'سکه/الماس همگانی +' . (int)($j['amount'] ?? 0) : 'پیام'));
        $text .= "\n{$icon} <code>".htmlspecialchars((string)$j['id'], ENT_QUOTES, 'UTF-8')."</code> | {$type}\n".
                 "📌 وضعیت: **".htmlspecialchars($status, ENT_QUOTES, 'UTF-8')."** | {$offset}/{$total} ({$pct}%)\n".
                 "✅ موفق: **".(int)($j['success'] ?? 0)."** | ❌ ناموفق: **".(int)($j['failed'] ?? 0)."**\n";
        if(!empty($j['last_error'])) $text .= "⚠️ آخرین خطا: <code>".htmlspecialchars((string)$j['last_error'], ENT_QUOTES, 'UTF-8')."</code>\n";
    }
    if(function_exists('bot_stats_summary')){
        $st = bot_stats_summary();
        $text .= "
📌 وضعیت کاربران: 👥 کل: <b>".number_format((int)$st['total'])."</b> | ✅ فعال: <b>".number_format((int)$st['active'])."</b> | ☑️ غیرفعال قطعی: <b>".number_format((int)$st['inactive'])."</b>
";
    }
    $text .= "
⏱ روش ارسال: کرون جاب هر ۱ دقیقه، هر بار تا ۲۰ درخواست با مکث داخلی و رعایت خطای 429.";
    return $text;
}
function bot_broadcast_status_keyboard(){
    return json_encode(['inline_keyboard'=>[
        [['text'=>'🔄 بروزرسانی وضعیت', 'callback_data'=>'broadcast_status']],
        [['text'=>'▶️ اجرای فوری ۲۰ ارسال', 'callback_data'=>'broadcast_run_now']],
        [['text'=>'⛔️ لغو صف فعال', 'callback_data'=>'broadcast_cancel_current']],
    ]], JSON_UNESCAPED_UNICODE);
}
function bot_broadcast_send_admin($adminId, $text){
    if(!$adminId || !function_exists('bot')) return false;
    $r = bot('sendMessage', ['chat_id'=>(string)$adminId, 'text'=>$text, 'parse_mode'=>'html']);
    return isset($r->ok) && $r->ok;
}
function bot_broadcast_retry_after($res){
    if(!is_object($res)) return 0;
    if(isset($res->parameters) && is_object($res->parameters) && isset($res->parameters->retry_after)){
        return max(1, min(3600, (int)$res->parameters->retry_after));
    }
    $desc = isset($res->description) ? (string)$res->description : '';
    if(preg_match('/retry after\s+(\d+)/i', $desc, $m)) return max(1, min(3600, (int)$m[1]));
    return 0;
}
function bot_broadcast_note_delivery($userId, $res, $source='broadcast'){
    if(!function_exists('bot_stats_mark_active') && is_file(dirname(__DIR__) . '/app/Services/AdminStatsService.php')){
        include_once dirname(__DIR__) . '/app/Services/AdminStatsService.php';
    }
    if(is_object($res) && isset($res->ok) && $res->ok){
        if(function_exists('bot_stats_mark_active')) bot_stats_mark_active($userId, $source.'_ok');
        return 'active';
    }
    if(function_exists('bot_stats_is_permanent_delivery_failure') && bot_stats_is_permanent_delivery_failure($res)){
        if(function_exists('bot_stats_mark_inactive')) bot_stats_mark_inactive($userId, isset($res->description) ? (string)$res->description : 'permanent_delivery_failure');
        return 'inactive';
    }
    return 'unknown';
}
function bot_broadcast_effective_limit($limit){
    $max = defined('BOT_BROADCAST_MAX_PER_RUN') ? (int)BOT_BROADCAST_MAX_PER_RUN : 20;
    if($max < 1) $max = 20;
    return max(1, min($max, (int)$limit));
}
function bot_broadcast_effective_rps($rps = 0){
    $default = defined('BOT_TELEGRAM_BROADCAST_SAFE_RPS') ? (int)BOT_TELEGRAM_BROADCAST_SAFE_RPS : 20;
    $rps = (int)$rps > 0 ? (int)$rps : $default;
    return max(1, min(30, $rps));
}
function bot_broadcast_sleep_for_rps($rps){
    $rps = bot_broadcast_effective_rps($rps);
    // Slightly slower than the theoretical maximum to avoid shared-hosting spikes.
    $sleep = (int)ceil(1000000 / $rps);
    if($sleep > 0) @usleep($sleep);
}
function bot_broadcast_process($limit = BOT_BROADCAST_PER_RUN, $maxRuntime = 25, $safeRps = 0){
    $limit = bot_broadcast_effective_limit($limit);
    $safeRps = bot_broadcast_effective_rps($safeRps);
    $maxRuntime = max(5, min(55, (int)$maxRuntime));
    $started = time();
    $lockFile = bot_broadcast_lock_file();
    $lock = @fopen($lockFile, 'c+');
    if($lock && !@flock($lock, LOCK_EX | LOCK_NB)){
        return ['ok'=>true, 'status'=>'locked', 'processed'=>0, 'success'=>0, 'failed'=>0, 'message'=>'previous run is still active'];
    }
    $queue = null; $idx = null; $job = bot_broadcast_current_job($queue, $idx);
    if(!$job){
        if($lock){ @flock($lock, LOCK_UN); @fclose($lock); }
        return ['ok'=>true, 'status'=>'empty', 'processed'=>0, 'success'=>0, 'failed'=>0];
    }
    $ids = bot_broadcast_user_ids(true);
    if((int)($job['total'] ?? 0) !== count($ids)) $job['total'] = count($ids);
    if(empty($job['started_at'])) $job['started_at'] = time();
    if(!empty($job['cooldown_until']) && (int)$job['cooldown_until'] > time()){
        if($lock){ @flock($lock, LOCK_UN); @fclose($lock); }
        return ['ok'=>true, 'status'=>'cooldown', 'job_id'=>$job['id'], 'processed'=>0, 'success'=>0, 'failed'=>0, 'cooldown_until'=>(int)$job['cooldown_until'], 'retry_after_seconds'=>((int)$job['cooldown_until'] - time()), 'limit_per_run'=>$limit, 'safe_rps'=>$safeRps];
    }
    $job['status'] = 'running';
    $job['last_run_at'] = time();
    $offset = max(0, (int)($job['offset'] ?? 0));
    $processed = $success = $failed = 0;
    while($offset < count($ids) && $processed < $limit && (time() - $started) < $maxRuntime){
        $target = (string)$ids[$offset];
        if (($job['type'] ?? 'copy') === 'coin') {
            $amount = (int)($job['amount'] ?? 0);
            $wallet = ['ok'=>false, 'error'=>'wallet_missing'];
            if($amount > 0 && function_exists('bot_wallet_add')){
                $wallet = bot_wallet_add($target, $amount, 'mass_coin_admin', ['admin_id'=>(string)($job['admin_id'] ?? ''), 'job_id'=>(string)($job['id'] ?? '')]);
            } elseif($amount > 0) {
                $u = function_exists('bot_data_get_json') ? bot_data_get_json('melat/'.$target.'.json', []) : [];
                if(!is_array($u)) $u=[];
                $before=(int)($u['Points'] ?? 0); $after=$before+$amount; $u['Points']=$after;
                if(function_exists('bot_data_put_json')) bot_data_put_json('melat/'.$target.'.json',$u);
                elseif(function_exists('saveJson')) saveJson('melat/'.$target.'.json',$u);
                $wallet=['ok'=>true,'before'=>$before,'after'=>$after];
            }
            $processed++;
            if(!empty($wallet['ok'])){
                $success++; $job['success']=(int)($job['success'] ?? 0)+1;
                $res = bot('sendMessage', [
                    'chat_id'=>$target,
                    'text'=>bot_broadcast_coin_notify_text($job, $wallet['after'] ?? null),
                    'parse_mode'=>'MarkDown',
                ]);
                bot_broadcast_note_delivery($target, $res, 'mass_coin_notify');
                $retryAfter = bot_broadcast_retry_after($res);
                $offset++;
                $job['offset']=$offset;
                if(!is_object($res) || empty($res->ok)){
                    $job['notify_failed']=(int)($job['notify_failed'] ?? 0)+1;
                    $job['last_error']=isset($res->description) ? (string)$res->description : 'notify_failed';
                }
                $queue['jobs'][$idx] = $job;
                bot_broadcast_write_queue($queue);
                if($retryAfter > 0){
                    $job['cooldown_until'] = time() + $retryAfter;
                    $job['last_error'] = 'Telegram 429 retry_after=' . $retryAfter;
                    $queue['jobs'][$idx] = $job;
                    bot_broadcast_write_queue($queue);
                    if($lock){ @flock($lock, LOCK_UN); @fclose($lock); }
                    return ['ok'=>true, 'status'=>'cooldown', 'job_id'=>$job['id'], 'processed'=>$processed, 'success'=>$success, 'failed'=>$failed, 'offset'=>$job['offset'], 'total'=>$job['total'], 'retry_after_seconds'=>$retryAfter, 'limit_per_run'=>$limit, 'safe_rps'=>$safeRps, 'elapsed_seconds'=>time()-$started];
                }
            }else{
                $failed++; $job['failed']=(int)($job['failed'] ?? 0)+1;
                $job['last_error']=(string)($wallet['error'] ?? 'wallet_failed');
                $offset++; $job['offset']=$offset;
                $queue['jobs'][$idx] = $job;
                bot_broadcast_write_queue($queue);
            }
        } elseif (($job['type'] ?? 'copy') === 'text') {
            $res = bot('sendMessage', [
                'chat_id' => $target,
                'text' => (string)($job['text'] ?? ''),
                'parse_mode' => 'MarkDown',
            ]);
            $processed++;
            $retryAfter = bot_broadcast_retry_after($res);
            if($retryAfter > 0){
                $job['cooldown_until'] = time() + $retryAfter;
                $job['last_error'] = 'Telegram 429 retry_after=' . $retryAfter;
                $queue['jobs'][$idx] = $job;
                bot_broadcast_write_queue($queue);
                if($lock){ @flock($lock, LOCK_UN); @fclose($lock); }
                return ['ok'=>true, 'status'=>'cooldown', 'job_id'=>$job['id'], 'processed'=>$processed, 'success'=>$success, 'failed'=>$failed, 'offset'=>$job['offset'], 'total'=>$job['total'], 'retry_after_seconds'=>$retryAfter, 'limit_per_run'=>$limit, 'safe_rps'=>$safeRps, 'elapsed_seconds'=>time()-$started];
            }
            if(isset($res->ok) && $res->ok){
                $success++; $job['success'] = (int)($job['success'] ?? 0) + 1;
                bot_broadcast_note_delivery($target, $res, 'broadcast_text');
            }else{
                $failed++; $job['failed'] = (int)($job['failed'] ?? 0) + 1;
                $job['last_error'] = isset($res->description) ? (string)$res->description : 'unknown_error';
                bot_broadcast_note_delivery($target, $res, 'broadcast_text');
            }
            $offset++;
            $job['offset'] = $offset;
        } else {
            $method = (($job['type'] ?? 'copy') === 'forward') ? 'forwardMessage' : 'copyMessage';
            $res = bot($method, [
                'chat_id' => $target,
                'from_chat_id' => (string)$job['from_chat_id'],
                'message_id' => (int)$job['message_id'],
            ]);
            $processed++;
            $retryAfter = bot_broadcast_retry_after($res);
            if($retryAfter > 0){
                $job['cooldown_until'] = time() + $retryAfter;
                $job['last_error'] = 'Telegram 429 retry_after=' . $retryAfter;
                $queue['jobs'][$idx] = $job;
                bot_broadcast_write_queue($queue);
                if($lock){ @flock($lock, LOCK_UN); @fclose($lock); }
                return ['ok'=>true, 'status'=>'cooldown', 'job_id'=>$job['id'], 'processed'=>$processed, 'success'=>$success, 'failed'=>$failed, 'offset'=>$job['offset'], 'total'=>$job['total'], 'retry_after_seconds'=>$retryAfter, 'limit_per_run'=>$limit, 'safe_rps'=>$safeRps, 'elapsed_seconds'=>time()-$started];
            }
            if(isset($res->ok) && $res->ok){
                $success++; $job['success'] = (int)($job['success'] ?? 0) + 1;
                bot_broadcast_note_delivery($target, $res, 'broadcast_message');
            }else{
                $failed++; $job['failed'] = (int)($job['failed'] ?? 0) + 1;
                $job['last_error'] = isset($res->description) ? (string)$res->description : 'unknown_error';
                bot_broadcast_note_delivery($target, $res, 'broadcast_message');
            }
            $offset++;
            $job['offset'] = $offset;
        }
        if($offset < count($ids) && $processed < $limit && (time() - $started) < $maxRuntime){
            bot_broadcast_sleep_for_rps($safeRps);
        }
    }
    if($offset >= count($ids)){
        $job['status'] = 'done';
        $job['finished_at'] = time();
        // Do not overwrite active/inactive statistics with the success count of the last broadcast.
        // Accurate status is stored in lib/kodam/user_status.json only when Telegram returns a permanent delivery result.
        if($processed < $limit){
            $statsLine = '';
            if(function_exists('bot_stats_summary')){
                $st = bot_stats_summary();
                $statsLine = "

👥 کل کاربران: <b>".number_format((int)$st['total'])."</b>
✅ فعال: <b>".number_format((int)$st['active'])."</b>
☑️ غیرفعال قطعی: <b>".number_format((int)$st['inactive'])."</b>";
            }
            bot_broadcast_send_admin((string)($job['admin_id'] ?? ''), "✅ **ارسال همگانی تمام شد**

🆔 صف: <code>".htmlspecialchars((string)$job['id'], ENT_QUOTES, 'UTF-8')."</code>
✅ موفق: **".(int)$job['success']."**
❌ ناموفق: **".(int)$job['failed']."**".$statsLine);
        }
    }
    $queue['jobs'][$idx] = $job;
    bot_broadcast_write_queue($queue);
    if($lock){ @flock($lock, LOCK_UN); @fclose($lock); }
    return ['ok'=>true, 'status'=>$job['status'], 'job_id'=>$job['id'], 'processed'=>$processed, 'success'=>$success, 'failed'=>$failed, 'offset'=>$job['offset'], 'total'=>$job['total'], 'limit_per_run'=>$limit, 'safe_rps'=>$safeRps, 'elapsed_seconds'=>time()-$started];
}
}
