<?php
// Professional queue layer for subscribers actions.
// Keeps heavy MikroTik operations outside subscribers.php request.
if (!defined('ALLOW_MANUAL_PPP_CUT_EXECUTION')) define('ALLOW_MANUAL_PPP_CUT_EXECUTION', true);

require_once __DIR__ . '/db_bootstrap.php';
require_once __DIR__ . '/mikrotik_lib.php';
require_once __DIR__ . '/subscriber_history_lib.php';
require_once __DIR__ . '/ppp_active_cache_lib.php';
require_once __DIR__ . '/subscribers_bulk_action_lib.php';

if (!function_exists('sq_json')) {
  function sq_json($v): string { return json_encode($v, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES); }
}
if (!function_exists('sq_log')) {
  function sq_log(string $msg): void {
    $dir = __DIR__ . '/logs';
    if (!is_dir($dir)) @mkdir($dir, 0777, true);
    @file_put_contents($dir.'/subscribers_queue.log', '['.date('Y-m-d H:i:s').'] '.$msg.PHP_EOL, FILE_APPEND);
  }
}

if (!function_exists('sq_purge_ppp_state_caches')) {
  function sq_purge_ppp_state_caches(): void {
    $dir = __DIR__ . '/cache/runtime';
    foreach (['ppp_disabled_state_map.json','ppp_secret_reality_map.json','ppp_active.json'] as $name) {
      $f = $dir . '/' . $name;
      if (is_file($f)) { @unlink($f); }
    }
  }
}
if (!function_exists('sq_table_exists')) {
  function sq_table_exists(mysqli $conn, string $table): bool {
    static $cache=[]; if(isset($cache[$table])) return $cache[$table];
    $t=$conn->real_escape_string($table); $rs=@$conn->query("SHOW TABLES LIKE '$t'");
    return $cache[$table]=($rs && $rs->num_rows>0);
  }
}
if (!function_exists('sq_col_exists')) {
  function sq_col_exists(mysqli $conn, string $table, string $col): bool {
    static $cache=[]; $k=$table.'|'.$col; if(isset($cache[$k])) return $cache[$k];
    $t=$conn->real_escape_string($table); $c=$conn->real_escape_string($col); $rs=@$conn->query("SHOW COLUMNS FROM `$t` LIKE '$c'");
    return $cache[$k]=($rs && $rs->num_rows>0);
  }
}
if (!function_exists('sq_ensure_schema')) {
  function sq_ensure_schema(mysqli $conn): void {
    @$conn->query("CREATE TABLE IF NOT EXISTS subscribers_action_jobs (
      id INT UNSIGNED NOT NULL AUTO_INCREMENT,
      action VARCHAR(32) NOT NULL,
      status ENUM('queued','running','done','partial','failed','cancelled') NOT NULL DEFAULT 'queued',
      total INT UNSIGNED NOT NULL DEFAULT 0,
      processed INT UNSIGNED NOT NULL DEFAULT 0,
      succeeded INT UNSIGNED NOT NULL DEFAULT 0,
      failed INT UNSIGNED NOT NULL DEFAULT 0,
      payload_json MEDIUMTEXT NULL,
      result_json MEDIUMTEXT NULL,
      message VARCHAR(500) NULL,
      requested_by VARCHAR(120) NULL,
      locked_by VARCHAR(120) NULL,
      locked_at DATETIME NULL,
      started_at DATETIME NULL,
      finished_at DATETIME NULL,
      created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
      updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
      PRIMARY KEY (id),
      KEY idx_status_created (status, created_at),
      KEY idx_locked (locked_at)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci");
  }
}
if (!function_exists('sq_request_user')) {
  function sq_request_user(): string {
    if (session_status() === PHP_SESSION_NONE) @session_start();
    foreach (['admin_username','username','user_name','name','admin_name'] as $k) {
      $v = trim((string)($_SESSION[$k] ?? ''));
      if ($v !== '') return mb_substr($v, 0, 110, 'UTF-8');
    }
    return 'admin';
  }
}
if (!function_exists('sq_action_label')) {
  function sq_action_label(string $action): string {
    return ['cut'=>'فصل','restore'=>'استعادة','cancel'=>'إلغاء اشتراك','open_speed'=>'سرعة مفتوحة'][$action] ?? $action;
  }
}
if (!function_exists('sq_enqueue_job')) {
  function sq_enqueue_job(mysqli $conn, string $action, array $ids, string $requestedBy=''): int {
    sq_ensure_schema($conn);
    $action = strtolower(trim($action));
    $ids = array_values(array_unique(array_filter(array_map('intval', $ids), fn($v)=>$v>0)));
    if (!$ids || !in_array($action, ['cut','restore','cancel','open_speed'], true)) throw new RuntimeException('طلب غير صالح.');
    if (count($ids) > 1000) throw new RuntimeException('عدد المشتركين كبير جداً للدفعة الواحدة.');
    $payload = sq_json(['ids'=>$ids, 'action'=>$action, 'created_ip'=>($_SERVER['REMOTE_ADDR'] ?? '')]);
    $requestedBy = trim($requestedBy) ?: sq_request_user();
    $st = $conn->prepare("INSERT INTO subscribers_action_jobs(action,status,total,payload_json,requested_by,message) VALUES(?,?,?,?,?,?)");
    if (!$st) throw new RuntimeException('تعذر تجهيز مهمة التنفيذ.');
    $status='queued'; $total=count($ids); $msg='بانتظار التنفيذ';
    $st->bind_param('ssisss', $action, $status, $total, $payload, $requestedBy, $msg);
    if (!$st->execute()) throw new RuntimeException('تعذر حفظ مهمة التنفيذ: '.$st->error);
    return (int)$conn->insert_id;
  }
}
if (!function_exists('sq_get_job')) {
  function sq_get_job(mysqli $conn, int $jobId): ?array {
    sq_ensure_schema($conn);
    $rs = @$conn->query('SELECT * FROM subscribers_action_jobs WHERE id='.(int)$jobId.' LIMIT 1');
    return ($rs && ($r=$rs->fetch_assoc())) ? $r : null;
  }
}
if (!function_exists('sq_latest_jobs')) {
  function sq_latest_jobs(mysqli $conn, int $limit=8): array {
    sq_ensure_schema($conn);
    $limit=max(1,min(30,$limit)); $out=[];
    $rs=@$conn->query('SELECT id,action,status,total,processed,succeeded,failed,message,requested_by,created_at,started_at,finished_at,updated_at FROM subscribers_action_jobs ORDER BY id DESC LIMIT '.$limit);
    if($rs) while($r=$rs->fetch_assoc()) $out[]=$r;
    return $out;
  }
}
if (!function_exists('sq_api_cmd')) {
  function sq_api_cmd($api, string $path, array $params = []) {
    if (!$api) return [];
    try {
      if (method_exists($api, 'command')) {
        $arr = [$path];
        foreach ($params as $k=>$v) {
          if ($k === '.id') $arr[] = '=.id='.$v;
          elseif (strlen((string)$k) && (($k[0] ?? '') === '?' || ($k[0] ?? '') === '~')) $arr[] = $k.'='.$v;
          else $arr[] = '='.$k.'='.$v;
        }
        return $api->command($arr);
      }
      if (method_exists($api, 'comm')) return $api->comm($path, $params);
    } catch (Throwable $e) { return [['.type'=>'!trap','message'=>$e->getMessage()]]; }
    return [];
  }
}
if (!function_exists('sq_rows_error')) {
  function sq_rows_error($rows, string &$msg=''): bool {
    if(!is_array($rows)) return false;
    foreach($rows as $row){ if(!is_array($row)) continue; $t=(string)($row['.type']??''); if($t==='!trap'||$t==='!fatal'){ $msg=(string)($row['message']??$row['category']??'RouterOS error'); return true; } }
    return false;
  }
}
if (!function_exists('sq_find_first_id_by_name')) {
  function sq_find_first_id_by_name($api, string $path, string $name): string {
    $name=trim($name); if($name==='') return '';
    $rows=sq_api_cmd($api,$path,['?name'=>$name]); $msg='';
    if(sq_rows_error($rows,$msg)) throw new RuntimeException($msg);
    if(!is_array($rows)||!$rows) $rows=sq_api_cmd($api,$path,[]);
    if(!is_array($rows)) return '';
    foreach($rows as $row){ if(!is_array($row)) continue; if(isset($row['.type'])&&$row['.type']!=='!re') continue; if(trim((string)($row['name']??''))===$name && trim((string)($row['.id']??''))!=='') return trim((string)$row['.id']); }
    return '';
  }
}
if (!function_exists('sq_disconnect_ppp')) {
  function sq_disconnect_ppp($api, string $pppUser): void {
    $aid=sq_find_first_id_by_name($api,'/ppp/active/print',$pppUser);
    if($aid!==''){ $rows=sq_api_cmd($api,'/ppp/active/remove',['.id'=>$aid]); $msg=''; if(sq_rows_error($rows,$msg)) throw new RuntimeException('فشل فصل الجلسة '.$pppUser.': '.$msg); }
  }
}
if (!function_exists('sq_set_secret_disabled')) {
  function sq_set_secret_disabled($api, string $pppUser, bool $disabled): void {
    $sid=sq_find_first_id_by_name($api,'/ppp/secret/print',$pppUser);
    if($sid==='') throw new RuntimeException('PPPoE secret غير موجود: '.$pppUser);
    $rows=sq_api_cmd($api,'/ppp/secret/set',['.id'=>$sid,'disabled'=>($disabled?'yes':'no')]); $msg='';
    if(sq_rows_error($rows,$msg)) throw new RuntimeException('فشل تعديل حالة '.$pppUser.': '.$msg);
  }
}
if (!function_exists('sq_remove_secret')) {
  function sq_remove_secret($api, string $pppUser): bool {
    $sid=sq_find_first_id_by_name($api,'/ppp/secret/print',$pppUser);
    if($sid==='') return false;
    $rows=sq_api_cmd($api,'/ppp/secret/remove',['.id'=>$sid]); $msg='';
    if(sq_rows_error($rows,$msg)) throw new RuntimeException('فشل حذف '.$pppUser.': '.$msg);
    return true;
  }
}
if (!function_exists('sq_set_open_speed')) {
  function sq_set_open_speed($api, string $pppUser): void {
    $sid=sq_find_first_id_by_name($api,'/ppp/secret/print',$pppUser);
    if($sid==='') throw new RuntimeException('لم يتم العثور على secret للحساب '.$pppUser);
    $rows=sq_api_cmd($api,'/ppp/secret/set',['.id'=>$sid,'profile'=>'OPEN-ALL-UNLIMITED','rate-limit'=>'']); $msg='';
    if(sq_rows_error($rows,$msg)) throw new RuntimeException('فشل فتح السرعة '.$pppUser.': '.$msg);
  }
}
if (!function_exists('sq_execute_one')) {
  function sq_execute_one(mysqli $conn, $api, string $action, array $sub): array {
    if (!function_exists('bulk_execute_one_with_api')) {
      throw new RuntimeException('مصدر تنفيذ المشتركين الموحد غير متاح.');
    }
    return bulk_execute_one_with_api($conn, $api, $action, $sub, ['source'=>'queue']);
  }
}
if (!function_exists('sq_claim_next_job')) {
  function sq_claim_next_job(mysqli $conn, string $worker): ?array {
    sq_ensure_schema($conn);
    // Release stale running locks after 10 minutes.
    @$conn->query("UPDATE subscribers_action_jobs SET status='queued', message='أعيدت للانتظار بعد توقف التنفيذ', locked_by=NULL, locked_at=NULL WHERE status='running' AND locked_at < (NOW() - INTERVAL 10 MINUTE)");
    $rs=@$conn->query("SELECT id FROM subscribers_action_jobs WHERE status='queued' ORDER BY id ASC LIMIT 1");
    if(!$rs || !($r=$rs->fetch_assoc())) return null;
    $id=(int)$r['id'];
    $w=$conn->real_escape_string($worker);
    @$conn->query("UPDATE subscribers_action_jobs SET status='running', locked_by='$w', locked_at=NOW(), started_at=IFNULL(started_at,NOW()), message='قيد التنفيذ' WHERE id=$id AND status='queued'");
    if($conn->affected_rows<1) return null;
    return sq_get_job($conn,$id);
  }
}
if (!function_exists('sq_run_worker')) {
  function sq_run_worker(mysqli $conn, int $maxJobs=1, int $maxSeconds=20): array {
    sq_ensure_schema($conn);
    $worker='worker-'.getmypid().'-'.substr(md5((string)microtime(true)),0,6);
    $started=microtime(true); $summary=['ok'=>true,'jobs'=>0,'changed'=>0,'failed'=>0,'messages'=>[]];
    for($j=0;$j<$maxJobs;$j++){
      if((microtime(true)-$started)>$maxSeconds) break;
      $job=sq_claim_next_job($conn,$worker); if(!$job) break;
      $summary['jobs']++;
      $jobId=(int)$job['id']; $action=(string)$job['action']; $payload=json_decode((string)($job['payload_json']??''),true) ?: [];
      $ids=array_values(array_unique(array_filter(array_map('intval',(array)($payload['ids']??[])),fn($v)=>$v>0)));
      $details=[]; $changed=0; $failed=0; $processed=0; $api=null;
      try{
        mt_require_router_online_or_throw($conn, 'تنفيذ Queue للمشتركين');
        $api=mt_api($conn); if(!$api) throw new RuntimeException((string)($GLOBALS['mt_last_error'] ?? 'فشل الاتصال بالمايكروتك.'));
        if($action==='open_speed' && function_exists('mt_ppp_profile_ensure') && !mt_ppp_profile_ensure($conn,'OPEN-ALL-UNLIMITED','')) throw new RuntimeException((string)($GLOBALS['mt_last_error'] ?? 'تعذر تجهيز بروفايل السرعة المفتوحة.'));
        $in=implode(',',array_map('intval',$ids)); $rowsById=[];
        $rs=@$conn->query('SELECT * FROM subscribers WHERE id IN ('.$in.')');
        if($rs) while($r=$rs->fetch_assoc()) $rowsById[(int)$r['id']]=$r;
        foreach($ids as $id){
          $processed++;
          try{
            $sub=$rowsById[$id]??null; if(!$sub) throw new RuntimeException('المشترك غير موجود في قاعدة البيانات.');
            $details[]=sq_execute_one($conn,$api,$action,$sub); $changed++;
          }catch(Throwable $e){ $failed++; $details[]=['id'=>$id,'ok'=>false,'message'=>$e->getMessage()]; sq_log('job='.$jobId.' item='.$id.' failed: '.$e->getMessage()); }
          if($processed % 5 === 0 || $processed === count($ids)){
            $msg='قيد التنفيذ: '.$processed.' / '.count($ids);
            $conn->query("UPDATE subscribers_action_jobs SET processed=$processed,succeeded=$changed,failed=$failed,result_json='".$conn->real_escape_string(sq_json($details))."',message='".$conn->real_escape_string($msg)."' WHERE id=$jobId");
          }
          if((microtime(true)-$started)>$maxSeconds) break 2;
        }
        $status=$failed>0?($changed>0?'partial':'failed'):'done';
        $msg='تم '.sq_action_label($action).' على '.$changed.' مشترك'.($failed>0?'، وفشل '.$failed:'');
        $conn->query("UPDATE subscribers_action_jobs SET status='".$conn->real_escape_string($status)."',processed=$processed,succeeded=$changed,failed=$failed,result_json='".$conn->real_escape_string(sq_json($details))."',message='".$conn->real_escape_string($msg)."',finished_at=NOW(),locked_by=NULL,locked_at=NULL WHERE id=$jobId");
        try{ ua_ppp_cache_rebuild($conn,0,true); }catch(Throwable $e){ sq_log('cache rebuild after job='.$jobId.' failed: '.$e->getMessage()); }
        $summary['changed']+=$changed; $summary['failed']+=$failed; $summary['messages'][]=$msg;
      }catch(Throwable $e){
        $msg='فشل التنفيذ: '.$e->getMessage(); sq_log('job='.$jobId.' fatal: '.$msg);
        $conn->query("UPDATE subscribers_action_jobs SET status='failed',message='".$conn->real_escape_string($msg)."',result_json='".$conn->real_escape_string(sq_json($details))."',finished_at=NOW(),locked_by=NULL,locked_at=NULL WHERE id=$jobId");
        $summary['failed']++; $summary['messages'][]=$msg;
      }finally{ try{ if($api && method_exists($api,'disconnect')) $api->disconnect(); }catch(Throwable $e){} }
    }
    return $summary;
  }
}
