<?php
// admin/messages_lib.php (AUTO-MIGRATE)

if (session_status() === PHP_SESSION_NONE) session_start();

function ml_col_exists(mysqli $conn, string $table, string $col): bool {
  $t = $conn->real_escape_string($table);
  $c = $conn->real_escape_string($col);
  $rs = $conn->query("SHOW COLUMNS FROM `$t` LIKE '$c'");
  return $rs && $rs->num_rows > 0;
}

function ml_table_exists(mysqli $conn, string $table): bool {
  $tableEsc = $conn->real_escape_string($table);
  $rs = @$conn->query("SHOW TABLES LIKE '{$tableEsc}'");
  return $rs && $rs->num_rows > 0;
}

function ensure_message_tables(mysqli $conn): void {
  $conn->query("
    CREATE TABLE IF NOT EXISTS message_templates (
      id INT AUTO_INCREMENT PRIMARY KEY,
      template_key VARCHAR(50) NULL,
      title VARCHAR(150) NULL,
      body TEXT NULL,
      is_enabled TINYINT(1) NOT NULL DEFAULT 1,
      channel VARCHAR(20) NOT NULL DEFAULT 'whatsapp',
      created_at DATETIME NULL DEFAULT CURRENT_TIMESTAMP,
      updated_at DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
  ");
  try { $conn->query("ALTER TABLE message_templates CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); } catch (Throwable $e) {}

  // ⚠️ قد تكون الجدول موجودة من نسخة قديمة بدون template_key.
  // نضيف الأعمدة بشكل آمن مع تجاهل أخطاء التكرار.
  if (!ml_col_exists($conn,'message_templates','template_key')) {
    try { $conn->query("ALTER TABLE message_templates ADD COLUMN template_key VARCHAR(50) NULL"); } catch (Throwable $e) {}
  }
  if (!ml_col_exists($conn,'message_templates','title')) $conn->query("ALTER TABLE message_templates ADD COLUMN title VARCHAR(150) NULL");
  if (!ml_col_exists($conn,'message_templates','body')) $conn->query("ALTER TABLE message_templates ADD COLUMN body TEXT NULL");
  if (!ml_col_exists($conn,'message_templates','is_enabled')) $conn->query("ALTER TABLE message_templates ADD COLUMN is_enabled TINYINT(1) NOT NULL DEFAULT 1");
  if (!ml_col_exists($conn,'message_templates','channel')) $conn->query("ALTER TABLE message_templates ADD COLUMN channel VARCHAR(20) NOT NULL DEFAULT 'whatsapp'");
  try { $conn->query("CREATE UNIQUE INDEX uq_message_templates_key ON message_templates(template_key)"); } catch (Throwable $e) {}

  $conn->query("
    CREATE TABLE IF NOT EXISTS message_logs (
      id INT AUTO_INCREMENT PRIMARY KEY,
      template_key VARCHAR(50) NULL,
      subscriber_id INT NULL,
      phone VARCHAR(50) NULL,
      channel VARCHAR(20) NULL,
      status VARCHAR(30) NULL,
      response TEXT NULL,
      created_at DATETIME NULL DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
  ");
  try { $conn->query("ALTER TABLE message_logs CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); } catch (Throwable $e) {}
  if (!ml_col_exists($conn,'message_logs','template_key')) $conn->query("ALTER TABLE message_logs ADD COLUMN template_key VARCHAR(50) NULL");
}

function default_templates(): array {
  return [
    'MANUAL_CUT'            => ['تم فصل الخدمة', "عزيزي {name}، تم فصل خدمة الإنترنت على حساب {ppp_username}.\nالسبب: {reason}\nالتاريخ: {datetime}"],
    'MANUAL_RESTORE'        => ['تمت إعادة الخدمة', "عزيزي {name}، تمت إعادة تفعيل خدمة الإنترنت على حساب {ppp_username}.\nالتاريخ: {datetime}"],
    'LINE_CUT_UNPAID'       => ['قطع الخدمة لعدم الدفع', 'تنبيه: تم إيقاف الخدمة بسبب عدم تسديد الفاتورة رقم {invoice_id}. يرجى التسديد لإعادة التفعيل.'],
    'DUE_SOON_WARNING'      => ['تنبيه قبل القطع', 'تنبيه: فاتورتك رقم {invoice_id} غير مدفوعة، سيتم قطع الخدمة خلال {days} يوم/أيام إن لم يتم التسديد.'],
    // ✅ LIMIT بسبب عدم التسديد
    'LINE_LIMIT_UNPAID'     => ['تخفيض السرعة لعدم التسديد', "تنبيه: بسبب عدم تسديد الفاتورة رقم {invoice_id} (شهر {cycle_month}) تم تخفيض سرعة الإنترنت مؤقتًا.\nالمبلغ المطلوب: {amount}.\nسيتم قطع الخدمة بعد {days} يوم/أيام إذا لم يتم التسديد."],
    // ✅ LIMIT بسبب انتهاء الباقة (لا يوجد قطع، فقط تخفيض)
    'LINE_LIMIT_PACKAGE_EXPIRED' => ['تخفيض السرعة لانتهاء الباقة', "تنبيه: انتهت مدة الباقة الخاصة بك بتاريخ {package_end_date}.\nتم تخفيض سرعة الإنترنت مؤقتًا لحين التجديد.\nللتجديد يرجى التواصل معنا."],
    'SPEED_UPGRADED'        => ['رفع/استعادة السرعة', 'تمت استعادة السرعة إلى {speed} للمشترك {name}.'],
    'NEW_SUBSCRIBER_ACTIVE' => ['تفعيل مشترك جديد', 'تم تفعيل اشتراكك. اسم المستخدم: {username} كلمة المرور: {password} السرعة: {speed} الباقة: {package}.'],
    'SUBSCRIBER_ID_INFO'    => ['بيانات المشترك', "عزيزي {name}،\nرقم المعرّف الخاص باشتراكك هو: {subscriber_id}."],
    'PACKAGE_RENEWED'       => ['تجديد الباقة', 'تم تجديد الباقة الإضافية بنجاح: {addon}.'],
    'INVOICE_PAID'          => ['تسديد الفاتورة', 'تم تسديد الفاتورة رقم {invoice_id} بنجاح. شكراً لك.'],
    // ✅ انتهاء حجم الباقة الشهرية
    'QUOTA_EXPIRED_LIMIT'   => ['انتهى حجم الباقة - تم تخفيض السرعة', "تنبيه: تم استهلاك حجم الباقة لهذا الشهر ({cycle_month}).\nتم تخفيض السرعة مؤقتاً إلى {limit_speed} حتى بداية الشهر الجديد.\nيمكنك شراء باقة إضافية لاستعادة السرعة وإضافة حجم حتى نهاية الشهر."],
    // ✅ تطبيق باقة إضافية (حجم)
    'QUOTA_ADDON_APPLIED'   => ['تم إضافة حجم للباقة', "تمت إضافة {addon_gb}GB إلى اشتراكك لهذا الشهر ({cycle_month}).\nتمت استعادة السرعة إلى {speed}.\nصلاحية الإضافة حتى نهاية الشهر."],

    // ✅ تقرير أسبوعي للإدارة
    'ADMIN_WEEKLY_REPORT'   => ['تقرير أسبوعي للإدارة',
      "📊 تقرير أسبوعي ({from} → {to})\n"
    . "— إجمالي الفواتير: {invoices_total}\n"
    . "— إجمالي المدفوع: {paid_total}\n"
    . "— عدد غير المدفوع: {unpaid_count}\n"
    . "— عدد المشتركين: {subs_total}\n"],
    'ADMIN_DEVICE_DOWN'     => ['تنبيه الإدارة: انقطاع جهاز',
      "⚠️ تنبيه شبكة من {company_name}\nانقطع الاتصال مع {device_type}: {device_name}\nالعنوان: {device_host}\nوقت الرصد: {event_time}"],
    'ADMIN_DEVICE_UP'       => ['تنبيه الإدارة: عودة جهاز',
      "✅ عودة اتصال من {company_name}\nعاد {device_type}: {device_name} إلى العمل\nالعنوان: {device_host}\nوقت العودة: {event_time}\nمدة الانقطاع: {offline_duration}"],
  ];
}

function get_message_templates(mysqli $conn): array {
  ensure_message_tables($conn);

  // إن كان column مازال غير موجود لأي سبب (صلاحيات/نسخة قديمة)، نمنع انهيار الصفحة
  $hasKey = ml_col_exists($conn,'message_templates','template_key');

  $defs = default_templates();
  foreach ($defs as $k => $row) {
    $k2 = $conn->real_escape_string($k);
    $title = $conn->real_escape_string($row[0]);
    $body  = $conn->real_escape_string($row[1]);

    if($hasKey){
      $conn->query("
        INSERT INTO message_templates (template_key, title, body, is_enabled, channel)
        SELECT '$k2', '$title', '$body', 1, 'whatsapp'
        WHERE NOT EXISTS (SELECT 1 FROM message_templates WHERE template_key='$k2')
      ");
    } else {
      // fallback: نمنع التكرار حسب العنوان فقط
      $conn->query("
        INSERT INTO message_templates (title, body, is_enabled, channel)
        SELECT '$title', '$body', 1, 'whatsapp'
        WHERE NOT EXISTS (SELECT 1 FROM message_templates WHERE title='$title')
      ");
    }
  }

  $out = [];
  $rs = $conn->query("SELECT * FROM message_templates ORDER BY id ASC");
  while ($r = $rs->fetch_assoc()) {
    $key = $r['template_key'] ?? ('T'.$r['id']);
    $out[$key] = $r;
  }
  return $out;
}

function save_message_template(
  mysqli $conn,
  string $templateKey,
  string $title,
  string $body,
  string $channel,
  int $isEnabled
): bool {
  ensure_message_tables($conn);

  $templateKey = trim($templateKey);
  if ($templateKey === '') {
    throw new InvalidArgumentException('Message template key is required.');
  }

  $channel = strtolower(trim($channel));
  if (!in_array($channel, ['whatsapp', 'sms', 'both'], true)) {
    $channel = 'whatsapp';
  }
  $isEnabled = $isEnabled === 1 ? 1 : 0;

  // Legacy rows without template_key are exposed by get_message_templates() as T{id}.
  if (preg_match('/^T([1-9][0-9]*)$/', $templateKey, $match)) {
    $templateId = (int)$match[1];
    $stmt = $conn->prepare(
      'UPDATE message_templates
       SET title = ?, body = ?, channel = ?, is_enabled = ?
       WHERE id = ?'
    );
    if (!$stmt) {
      throw new RuntimeException('Unable to prepare legacy message template update.');
    }
    $stmt->bind_param('sssii', $title, $body, $channel, $isEnabled, $templateId);
    $ok = $stmt->execute();
    $matched = $stmt->affected_rows > 0;
    $stmt->close();

    if (!$ok) return false;
    if ($matched) return true;

    $check = $conn->prepare('SELECT id FROM message_templates WHERE id = ? LIMIT 1');
    if (!$check) {
      throw new RuntimeException('Unable to verify legacy message template.');
    }
    $check->bind_param('i', $templateId);
    $check->execute();
    $check->store_result();
    $exists = $check->num_rows === 1;
    $check->close();
    return $exists;
  }

  if (!ml_col_exists($conn, 'message_templates', 'template_key')) {
    throw new RuntimeException('The message_templates.template_key column is unavailable.');
  }

  $stmt = $conn->prepare(
    'UPDATE message_templates
     SET title = ?, body = ?, channel = ?, is_enabled = ?
     WHERE template_key = ?'
  );
  if (!$stmt) {
    throw new RuntimeException('Unable to prepare message template update.');
  }
  $stmt->bind_param('sssis', $title, $body, $channel, $isEnabled, $templateKey);
  $ok = $stmt->execute();
  $matched = $stmt->affected_rows > 0;
  $stmt->close();

  if (!$ok) return false;
  if ($matched) return true;

  // affected_rows is zero when the submitted values are unchanged, so verify
  // existence before deciding that this is a new template.
  $check = $conn->prepare('SELECT id FROM message_templates WHERE template_key = ? LIMIT 1');
  if (!$check) {
    throw new RuntimeException('Unable to verify message template.');
  }
  $check->bind_param('s', $templateKey);
  $check->execute();
  $check->store_result();
  $exists = $check->num_rows === 1;
  $check->close();
  if ($exists) return true;

  $insert = $conn->prepare(
    'INSERT INTO message_templates
      (template_key, title, body, channel, is_enabled)
     VALUES (?, ?, ?, ?, ?)'
  );
  if (!$insert) {
    throw new RuntimeException('Unable to prepare message template insert.');
  }
  $insert->bind_param('ssssi', $templateKey, $title, $body, $channel, $isEnabled);
  $ok = $insert->execute();
  $insert->close();
  return $ok;
}

/* =========================================================
   Outbox Queue + Campaigns (Phase 2)
========================================================= */

function ensure_message_queue_tables(mysqli $conn): void {
  static $ensuredConnections = [];
  $connectionId = spl_object_id($conn);
  if (isset($ensuredConnections[$connectionId])) return;

  ensure_message_tables($conn);

  $conn->query("CREATE TABLE IF NOT EXISTS message_outbox (
    id INT AUTO_INCREMENT PRIMARY KEY,
    campaign_id BIGINT NULL,
    template_key VARCHAR(50) NULL,
    subscriber_id INT NULL,
    to_phone VARCHAR(50) NULL,
    channel VARCHAR(20) NOT NULL DEFAULT 'whatsapp',
    body TEXT NULL,
    vars_json LONGTEXT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'QUEUED',
    try_count INT NOT NULL DEFAULT 0,
    last_error TEXT NULL,
    provider_message_id VARCHAR(120) NULL,
    created_by INT NULL,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    sent_at DATETIME NULL,
    delivered_at DATETIME NULL
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;");
  $outboxColumns = [
    'campaign_id' => 'BIGINT NULL',
    'vars_json' => 'LONGTEXT NULL',
    'created_by' => 'INT NULL',
    'try_count' => 'INT NOT NULL DEFAULT 0',
    'last_error' => 'TEXT NULL',
    'sent_at' => 'DATETIME NULL',
    'delivered_at' => 'DATETIME NULL',
    'run_id' => 'BIGINT NULL',
    'status_updated_at' => 'DATETIME NULL',
  ];
  foreach ($outboxColumns as $column => $definition) {
    if (!ml_col_exists($conn, 'message_outbox', $column)) {
      try { $conn->query("ALTER TABLE message_outbox ADD COLUMN `{$column}` {$definition}"); } catch (Throwable $e) {}
    }
  }

  $conn->query("CREATE TABLE IF NOT EXISTS message_campaigns (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    title VARCHAR(180) NOT NULL,
    template_key VARCHAR(50) NULL,
    channel VARCHAR(20) NOT NULL DEFAULT 'whatsapp',
    filters_json LONGTEXT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'DRAFT',
    total_targets INT NOT NULL DEFAULT 0,
    queued_count INT NOT NULL DEFAULT 0,
    sent_count INT NOT NULL DEFAULT 0,
    failed_count INT NOT NULL DEFAULT 0,
    created_by INT NULL,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;");
  $campaignColumns = [
    'template_key' => 'VARCHAR(50) NULL',
    'channel' => "VARCHAR(20) NOT NULL DEFAULT 'whatsapp'",
    'filters_json' => 'LONGTEXT NULL',
    'status' => "VARCHAR(30) NOT NULL DEFAULT 'DRAFT'",
    'total_targets' => 'INT NOT NULL DEFAULT 0',
    'queued_count' => 'INT NOT NULL DEFAULT 0',
    'sent_count' => 'INT NOT NULL DEFAULT 0',
    'failed_count' => 'INT NOT NULL DEFAULT 0',
    'created_by' => 'INT NULL',
    'updated_at' => 'DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP',
  ];
  foreach ($campaignColumns as $column => $definition) {
    if (!ml_col_exists($conn, 'message_campaigns', $column)) {
      try { $conn->query("ALTER TABLE message_campaigns ADD COLUMN `{$column}` {$definition}"); } catch (Throwable $e) {}
    }
  }
  // Some early builds used fail_count. Preserve their data while standardising
  // every reader and writer on failed_count.
  if (ml_col_exists($conn, 'message_campaigns', 'fail_count')
      && ml_col_exists($conn, 'message_campaigns', 'failed_count')) {
    try {
      $conn->query("UPDATE message_campaigns
        SET failed_count=GREATEST(COALESCE(failed_count,0),COALESCE(fail_count,0))");
    } catch (Throwable $e) {}
  }

  $conn->query("CREATE TABLE IF NOT EXISTS message_queue_runs (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    status VARCHAR(32) NOT NULL DEFAULT 'TESTING',
    test_phone VARCHAR(50) NULL,
    test_body LONGTEXT NULL,
    test_channel VARCHAR(20) NULL,
    confirmation_required TINYINT(1) NOT NULL DEFAULT 1,
    test_sent TINYINT(1) NOT NULL DEFAULT 0,
    test_error TEXT NULL,
    total_count INT NOT NULL DEFAULT 0,
    processed_count INT NOT NULL DEFAULT 0,
    sent_count INT NOT NULL DEFAULT 0,
    failed_count INT NOT NULL DEFAULT 0,
    started_by INT NULL,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    started_at DATETIME NULL,
    approved_at DATETIME NULL,
    completed_at DATETIME NULL,
    updated_at DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_message_queue_runs_status (status, id)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;");
  if (!ml_col_exists($conn, 'message_queue_runs', 'test_channel')) {
    try { $conn->query("ALTER TABLE message_queue_runs ADD COLUMN test_channel VARCHAR(20) NULL AFTER test_body"); } catch (Throwable $e) {}
  }

  try { $conn->query("CREATE INDEX idx_outbox_status ON message_outbox(status, id)"); } catch (Throwable $e) {}
  try { $conn->query("CREATE INDEX idx_outbox_sub ON message_outbox(subscriber_id)"); } catch (Throwable $e) {}
  try { $conn->query("CREATE INDEX idx_outbox_run ON message_outbox(run_id, status, id)"); } catch (Throwable $e) {}
  try { $conn->query("CREATE INDEX idx_outbox_campaign ON message_outbox(campaign_id, subscriber_id, channel)"); } catch (Throwable $e) {}
  $ensuredConnections[$connectionId] = true;
}

function ml_safe_json_decode(?string $s): array {
  if (!$s) return [];
  $j = json_decode($s, true);
  return (json_last_error() === JSON_ERROR_NONE && is_array($j)) ? $j : [];
}

function messages_render(string $body, array $vars): string {
  $vars = array_merge([
    'date' => date('Y-m-d'),
    'time' => date('H:i'),
    'datetime' => date('Y-m-d H:i'),
  ], $vars);

  return preg_replace_callback('/\\{([a-zA-Z0-9_\\.\\-]+)\\}/u', function($m) use ($vars){
    $k = $m[1];
    if (array_key_exists($k, $vars)) return (string)$vars[$k];
    if (strpos($k,'.') !== false) {
      $parts = explode('.', $k);
      $v = $vars;
      foreach($parts as $p){
        if (is_array($v) && array_key_exists($p, $v)) $v = $v[$p]; else { $v = null; break; }
      }
      if ($v !== null) return (string)$v;
    }
    return $m[0];
  }, (string)$body);
}

function messages_load_template(mysqli $conn, string $templateKey): ?array {
  ensure_message_tables($conn);
  $tKeyEsc = $conn->real_escape_string($templateKey);
  $rs = $conn->query("SELECT template_key,title,body,is_enabled,channel FROM message_templates WHERE template_key='$tKeyEsc' LIMIT 1");
  if(!$rs || !$rs->num_rows){
    if(function_exists('get_message_templates')) { @get_message_templates($conn); }
    $rs = $conn->query("SELECT template_key,title,body,is_enabled,channel FROM message_templates WHERE template_key='$tKeyEsc' LIMIT 1");
  }
  if(!$rs || !$rs->num_rows) return null;
  $t = $rs->fetch_assoc();
  if ((int)($t['is_enabled'] ?? 1) !== 1) return null;
  return $t;
}

function messages_build_vars_from_sub(mysqli $conn, int $subscriberId, array $vars): array {
  $subscriberId = (int)$subscriberId;
  if ($subscriberId <= 0) return $vars;

  $selects = [];
  foreach (['id','full_name','username','ppp_username','phone','ip_address','base_speed','current_speed'] as $col) {
    $selects[] = ml_col_exists($conn, 'subscribers', $col) ? "`$col`" : "NULL AS `$col`";
  }
  if (ml_col_exists($conn, 'subscribers', 'speed')) {
    $selects[] = "`speed`";
  } elseif (ml_col_exists($conn, 'subscribers', 'speed_name')) {
    $selects[] = "`speed_name` AS `speed`";
  } else {
    $selects[] = "NULL AS `speed`";
  }
  if (ml_col_exists($conn, 'subscribers', 'package')) {
    $selects[] = "`package`";
  } elseif (ml_col_exists($conn, 'subscribers', 'package_name')) {
    $selects[] = "`package_name` AS `package`";
  } else {
    $selects[] = "NULL AS `package`";
  }

  $rs = $conn->query("SELECT " . implode(',', $selects) . " FROM subscribers WHERE id=$subscriberId LIMIT 1");
  if($rs && $rs->num_rows){
    $s = $rs->fetch_assoc();
    $vars = array_merge([
      'subscriber_id' => $s['id'],
      'name' => $s['full_name'] ?? '',
      'full_name' => $s['full_name'] ?? '',
      'username' => $s['username'] ?? '',
      'ppp_username' => $s['ppp_username'] ?? ($s['username'] ?? ''),
      'phone' => $s['phone'] ?? '',
      'ip' => $s['ip_address'] ?? '',
      'speed' => $s['speed'] ?? ($s['current_speed'] ?? ''),
      'package' => $s['package'] ?? '',
      'base_speed' => $s['base_speed'] ?? '',
      'current_speed' => $s['current_speed'] ?? '',
      'sub' => $s,
    ], $vars);
  }
  return $vars;
}

function messages_phone_digits(string $phone): string {
  $digits = (string)preg_replace('/\D+/', '', trim($phone));
  if (str_starts_with($digits, '00')) $digits = substr($digits, 2);
  return $digits;
}

function messages_phone_is_valid(string $phone): bool {
  $digits = messages_phone_digits($phone);
  return strlen($digits) >= 8 && strlen($digits) <= 15;
}

function messages_normalize_recipient(mysqli $conn, string $phone, string $channel): string {
  $digits = messages_phone_digits($phone);
  // Validate the already-normalized value directly. Calling
  // messages_phone_is_valid() here normalized a second time and could strip
  // another leading "00", rejecting otherwise acceptable stored numbers.
  if (strlen($digits) < 8 || strlen($digits) > 15) return '';
  if (strtolower(trim($channel)) !== 'whatsapp') return $digits;

  require_once __DIR__ . '/settings_store.php';
  $whatsapp = setting_get($conn, 'whatsapp_cloud', []);
  $country = is_array($whatsapp)
    ? messages_phone_digits((string)($whatsapp['default_country_code'] ?? ''))
    : '';
  if ($country === '') {
    $sms = setting_get($conn, 'sms_provider', []);
    if (is_array($sms)) $country = messages_phone_digits((string)($sms['default_country_code'] ?? '963'));
  }
  if ($country === '') $country = '963';
  if (str_starts_with($digits, '0')) $digits = $country . substr($digits, 1);
  return strlen($digits) >= 8 && strlen($digits) <= 15 ? $digits : '';
}

function messages_campaign_normalize_filters($filters): array {
  if (is_string($filters)) $filters = ['audience'=>$filters];
  if (!is_array($filters)) $filters = [];
  $audience = strtolower(trim((string)($filters['audience'] ?? 'all')));
  if (!in_array($audience, ['all','active','stopped','has_phone'], true)) $audience = 'all';
  $allowed = ['unpaid_invoices','no_unpaid_invoices','complete_profile','incomplete_profile'];
  $exclude = $filters['exclude'] ?? [];
  if (!is_array($exclude)) $exclude = [$exclude];
  $exclude = array_values(array_unique(array_filter(array_map(
    static fn($value): string => strtolower(trim((string)$value)),
    $exclude
  ), static fn(string $value): bool => in_array($value, $allowed, true))));
  $subscriberIds = $filters['subscriber_ids'] ?? [];
  if (!is_array($subscriberIds)) $subscriberIds = preg_split('/[^0-9]+/', (string)$subscriberIds, -1, PREG_SPLIT_NO_EMPTY);
  $subscriberIds = array_values(array_slice(array_unique(array_filter(array_map(
    static fn($value): int => (int)$value,
    $subscriberIds
  ), static fn(int $value): bool => $value > 0)), 0, 1000));
  return ['audience'=>$audience, 'exclude'=>$exclude, 'subscriber_ids'=>$subscriberIds];
}

function messages_campaign_profile_complete_sql(mysqli $conn, string $alias = 's'): string {
  $prefix = $alias !== '' ? $alias . '.' : '';
  $requirements = [];
  foreach (['full_name','phone','address_area'] as $column) {
    if (ml_col_exists($conn, 'subscribers', $column)) {
      $requirements[] = "TRIM(COALESCE({$prefix}`{$column}`,''))<>''";
    }
  }
  $userFields = [];
  foreach (['ppp_username','username'] as $column) {
    if (ml_col_exists($conn, 'subscribers', $column)) $userFields[] = "{$prefix}`{$column}`";
  }
  if ($userFields) $requirements[] = "TRIM(COALESCE(" . implode(',', $userFields) . ",''))<>''";
  $documentFields = [];
  foreach (['document_number','national_id'] as $column) {
    if (ml_col_exists($conn, 'subscribers', $column)) $documentFields[] = "{$prefix}`{$column}`";
  }
  if ($documentFields) $requirements[] = "TRIM(COALESCE(" . implode(',', $documentFields) . ",''))<>''";
  return $requirements ? '(' . implode(' AND ', $requirements) . ')' : '1=1';
}

function messages_campaign_unpaid_invoice_sql(mysqli $conn, string $subscriberExpression): string {
  if (!ml_table_exists($conn, 'invoices') || !ml_col_exists($conn, 'invoices', 'subscriber_id')) return '0=1';
  $notPaid = [];
  if (ml_col_exists($conn, 'invoices', 'payment_status')) {
    $notPaid[] = "LOWER(TRIM(COALESCE(i.payment_status,''))) NOT IN ('paid','settled')";
  }
  if (ml_col_exists($conn, 'invoices', 'status')) {
    $notPaid[] = "LOWER(TRIM(COALESCE(i.status,''))) NOT IN ('paid','مدفوع')";
  }
  $amount = ml_col_exists($conn, 'invoices', 'remaining_amount')
    ? 'COALESCE(i.remaining_amount,i.amount,0)>0'
    : (ml_col_exists($conn, 'invoices', 'amount') ? 'COALESCE(i.amount,0)>0' : '1=1');
  $state = $notPaid ? '(' . implode(' AND ', $notPaid) . ')' : '1=1';
  return "EXISTS (SELECT 1 FROM invoices i
    WHERE i.subscriber_id={$subscriberExpression} AND {$state} AND {$amount})";
}

function messages_campaign_audience_where(mysqli $conn, $filters, string $alias = 's'): string {
  $normalized = messages_campaign_normalize_filters($filters);
  $audience = $normalized['audience'];
  $exclude = $normalized['exclude'];
  $prefix = $alias !== '' ? $alias . '.' : '';
  $parts = ['1=1'];
  if (!empty($normalized['subscriber_ids'])) {
    $parts[] = "{$prefix}id IN (" . implode(',', array_map('intval', $normalized['subscriber_ids'])) . ")";
  }
  if (ml_col_exists($conn, 'subscribers', 'is_deleted')) {
    $parts[] = "COALESCE({$prefix}is_deleted,0)=0";
  }
  if (ml_col_exists($conn, 'subscribers', 'status')) {
    $active = "LOWER(TRIM(COALESCE({$prefix}status,''))) IN ('active','نشط','1')";
    if ($audience === 'active') $parts[] = $active;
    if ($audience === 'stopped') $parts[] = "NOT ({$active})";
  }
  $parts[] = "{$prefix}phone IS NOT NULL";
  $parts[] = "TRIM({$prefix}phone)<>''";
  $profileComplete = messages_campaign_profile_complete_sql($conn, $alias);
  $hasUnpaid = messages_campaign_unpaid_invoice_sql($conn, "{$prefix}id");
  if (in_array('unpaid_invoices', $exclude, true)) $parts[] = "NOT ({$hasUnpaid})";
  if (in_array('no_unpaid_invoices', $exclude, true)) $parts[] = $hasUnpaid;
  if (in_array('complete_profile', $exclude, true)) $parts[] = "NOT ({$profileComplete})";
  if (in_array('incomplete_profile', $exclude, true)) $parts[] = $profileComplete;
  return 'WHERE ' . implode(' AND ', $parts);
}

function messages_campaign_eligible_targets(mysqli $conn, $filters): array {
  if (!ml_col_exists($conn, 'subscribers', 'phone')) return [];
  $where = messages_campaign_audience_where($conn, $filters, 's');
  $columns = ['id','phone','full_name','username','ppp_username','ip_address',
    'speed','speed_name','package','package_name','base_speed','current_speed'];
  $select = [];
  foreach ($columns as $column) {
    $select[] = ml_col_exists($conn, 'subscribers', $column)
      ? "s.`{$column}`"
      : "NULL AS `{$column}`";
  }
  $rs = @$conn->query("SELECT " . implode(',', $select) . " FROM subscribers s {$where} ORDER BY s.id ASC");
  if (!$rs) return [];
  $rows = [];
  while ($row = $rs->fetch_assoc()) {
    if (!messages_phone_is_valid((string)($row['phone'] ?? ''))) continue;
    $row['id'] = (int)$row['id'];
    $rows[] = $row;
  }
  return $rows;
}

function messages_campaign_vars_from_row(array $subscriber, array $vars = []): array {
  return array_merge([
    'subscriber_id'=>(int)($subscriber['id'] ?? 0),
    'name'=>(string)($subscriber['full_name'] ?? ''),
    'full_name'=>(string)($subscriber['full_name'] ?? ''),
    'username'=>(string)($subscriber['username'] ?? ''),
    'ppp_username'=>(string)($subscriber['ppp_username'] ?? $subscriber['username'] ?? ''),
    'phone'=>(string)($subscriber['phone'] ?? ''),
    'ip'=>(string)($subscriber['ip_address'] ?? ''),
    'speed'=>(string)($subscriber['speed'] ?? $subscriber['speed_name'] ?? $subscriber['current_speed'] ?? ''),
    'package'=>(string)($subscriber['package'] ?? $subscriber['package_name'] ?? ''),
    'base_speed'=>(string)($subscriber['base_speed'] ?? ''),
    'current_speed'=>(string)($subscriber['current_speed'] ?? ''),
    'sub'=>$subscriber,
  ], $vars);
}

function messages_campaign_enqueue_targets(
  mysqli $conn,
  int $campaignId,
  string $templateKey,
  string $channel,
  $filters,
  int $createdBy = 0
): array {
  ensure_message_queue_tables($conn);
  $template = messages_load_template($conn, $templateKey);
  if (!$template) return ['ok'=>false,'error'=>'قالب الحملة غير موجود أو غير مفعّل.'];
  $channel = strtolower(trim($channel));
  if (!in_array($channel, ['sms','whatsapp','both'], true)) {
    return ['ok'=>false,'error'=>'قناة الحملة غير صالحة.'];
  }
  $channels = $channel === 'both' ? ['sms','whatsapp'] : [$channel];
  $targets = messages_campaign_eligible_targets($conn, $filters);
  $existing = [];
  $channelSql = implode(',', array_map(
    static fn(string $value): string => "'" . $conn->real_escape_string($value) . "'",
    $channels
  ));
  $existingRs = @$conn->query("SELECT subscriber_id,channel FROM message_outbox
    WHERE campaign_id={$campaignId} AND subscriber_id IS NOT NULL AND channel IN ({$channelSql})");
  if ($existingRs) {
    while ($row = $existingRs->fetch_assoc()) {
      $existing[(int)$row['subscriber_id'] . '|' . (string)$row['channel']] = true;
    }
  }

  $rows = [];
  $skipped = 0;
  $invalid = 0;
  foreach ($targets as $subscriber) {
    $subscriberId = (int)$subscriber['id'];
    $vars = messages_campaign_vars_from_row($subscriber, ['campaign_id'=>$campaignId]);
    $body = messages_render((string)($template['body'] ?? ''), $vars);
    $varsJson = json_encode($vars, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
    if (!is_string($varsJson)) $varsJson = '{}';
    if ($body === '') {
      $invalid++;
      continue;
    }
    foreach ($channels as $targetChannel) {
      $key = $subscriberId . '|' . $targetChannel;
      if (isset($existing[$key])) {
        $skipped++;
        continue;
      }
      $phone = messages_normalize_recipient($conn, (string)$subscriber['phone'], $targetChannel);
      if ($phone === '') {
        $invalid++;
        continue;
      }
      $rows[] = [
        'subscriber_id'=>$subscriberId,
        'phone'=>$phone,
        'channel'=>$targetChannel,
        'body'=>$body,
        'vars_json'=>$varsJson,
      ];
      $existing[$key] = true;
    }
  }

  $queued = 0;
  try {
    $conn->begin_transaction();
    foreach (array_chunk($rows, 100) as $chunk) {
      $values = [];
      foreach ($chunk as $row) {
        $tpl = $conn->real_escape_string($templateKey);
        $phone = $conn->real_escape_string((string)$row['phone']);
        $targetChannel = $conn->real_escape_string((string)$row['channel']);
        $body = $conn->real_escape_string((string)$row['body']);
        $varsJson = $conn->real_escape_string((string)$row['vars_json']);
        $subscriberId = (int)$row['subscriber_id'];
        $creatorSql = $createdBy > 0 ? (string)$createdBy : 'NULL';
        $values[] = "({$campaignId},'{$tpl}',{$subscriberId},'{$phone}','{$targetChannel}','{$body}','{$varsJson}','QUEUED',{$creatorSql})";
      }
      if (!$values) continue;
      $ok = $conn->query("INSERT INTO message_outbox
        (campaign_id,template_key,subscriber_id,to_phone,channel,body,vars_json,status,created_by)
        VALUES " . implode(',', $values));
      if (!$ok) throw new RuntimeException($conn->error);
      $queued += max(0, (int)$conn->affected_rows);
    }
    $conn->commit();
  } catch (Throwable $e) {
    try { $conn->rollback(); } catch (Throwable $ignored) {}
    error_log('[messages] bulk campaign queue failed: ' . $e->getMessage());
    return ['ok'=>false,'error'=>'تعذر إدراج جمهور الحملة كاملًا في الطابور. لم يُحفظ إدراج جزئي.'];
  }

  $coveredTargets = [];
  foreach ($existing as $key => $_) {
    [$subscriberId] = explode('|', $key, 2);
    $coveredTargets[(int)$subscriberId] = ($coveredTargets[(int)$subscriberId] ?? 0) + 1;
  }
  $remaining = 0;
  foreach ($targets as $subscriber) {
    if (($coveredTargets[(int)$subscriber['id']] ?? 0) < count($channels)) $remaining++;
  }
  return [
    'ok'=>true,
    'queued'=>$queued,
    'skipped'=>$skipped,
    'invalid'=>$invalid,
    'total_targets'=>count($targets),
    'remaining_targets'=>$remaining,
  ];
}

function messages_enqueue_template(mysqli $conn, string $templateKey, string $toPhone, array $vars = [], ?int $subscriberId = null, ?string $forceChannel = null, ?int $createdBy = null): array {
  ensure_message_queue_tables($conn);
  $t = messages_load_template($conn, $templateKey);
  if(!$t) return ['ok'=>false,'error'=>'Template غير موجود/معطّل: '.$templateKey];

  $subscriberId = $subscriberId ? (int)$subscriberId : null;
  if ($subscriberId) $vars = messages_build_vars_from_sub($conn, $subscriberId, $vars);

  $body = messages_render((string)($t['body'] ?? ''), $vars);
  $channel = strtolower(trim((string)($forceChannel ?: ($t['channel'] ?? 'whatsapp'))));
  if (!in_array($channel, ['sms','whatsapp','both'], true)) {
    return ['ok'=>false,'error'=>'قناة الرسالة غير صالحة.'];
  }
  $toPhone = trim($toPhone);
  if (!messages_phone_is_valid($toPhone) || $body === '') {
    return ['ok'=>false,'error'=>'رقم الهاتف أو نص الرسالة مفقود.'];
  }

  $channels = $channel === 'both' ? ['whatsapp','sms'] : [$channel];
  $campaignId = max(0, (int)($vars['campaign_id'] ?? 0));
  $varsJson = json_encode($vars, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
  if ($varsJson === false) $varsJson = '{}';
  $campaignValue = $campaignId > 0 ? $campaignId : null;
  $subscriberValue = $subscriberId ?: null;
  $creatorValue = $createdBy && $createdBy > 0 ? $createdBy : null;
  $ids = [];
  $skipped = 0;

  $duplicate = null;
  if ($campaignId > 0 && $subscriberId) {
    $duplicate = $conn->prepare("SELECT id FROM message_outbox
      WHERE campaign_id=? AND subscriber_id=? AND channel=? LIMIT 1");
  }
  $insert = $conn->prepare("INSERT INTO message_outbox
    (campaign_id,template_key,subscriber_id,to_phone,channel,body,vars_json,status,created_by)
    VALUES (?,?,?,?,?,?,?,'QUEUED',?)");
  if (!$insert) {
    error_log('[NetworkCustomers] message outbox prepare failed: ' . $conn->error);
    return ['ok'=>false,'error'=>'تعذر تجهيز إضافة الرسالة إلى الطابور.'];
  }

  foreach ($channels as $targetChannel) {
    $channelPhone = messages_normalize_recipient($conn, $toPhone, $targetChannel);
    if ($channelPhone === '') {
      if ($duplicate) $duplicate->close();
      $insert->close();
      return ['ok'=>false,'error'=>'رقم الهاتف غير صالح لقناة الإرسال المحددة.'];
    }
    if ($duplicate) {
      $duplicate->bind_param('iis', $campaignId, $subscriberId, $targetChannel);
      if ($duplicate->execute()) {
        $existing = $duplicate->get_result();
        if ($existing && $existing->num_rows) {
          $skipped++;
          continue;
        }
      }
    }
    $insert->bind_param(
      'isissssi',
      $campaignValue,
      $templateKey,
      $subscriberValue,
      $channelPhone,
      $targetChannel,
      $body,
      $varsJson,
      $creatorValue
    );
    if (!$insert->execute()) {
      error_log('[NetworkCustomers] message outbox insert failed: ' . $insert->error);
      if ($duplicate) $duplicate->close();
      $insert->close();
      return ['ok'=>false,'error'=>'فشل إدراج الرسالة في الطابور.'];
    }
    $ids[] = (int)$insert->insert_id;
  }
  if ($duplicate) $duplicate->close();
  $insert->close();

  return [
    'ok'=>true,
    'id'=>$ids[0] ?? 0,
    'ids'=>$ids,
    'queued'=>count($ids),
    'skipped'=>$skipped,
    'status'=>$ids ? 'QUEUED' : 'EXISTS',
  ];
}

function messages_send_outbox_row(mysqli $conn, array $row): array {
  $channel = (string)($row['channel'] ?? 'whatsapp');
  $to = (string)($row['to_phone'] ?? '');
  $body = (string)($row['body'] ?? '');

  if ($to === '' || $body === '') {
    return ['ok'=>false,'error'=>'بيانات الرسالة ناقصة'];
  }

  if ($channel === 'whatsapp') {
    require_once __DIR__ . '/whatsapp_cloud.php';
    $sender = new WhatsAppCloudSender($conn);
    return $sender->sendText($to, $body);
  }

  require_once __DIR__ . '/sms_provider.php';
  return sms_provider_send($conn, $to, $body, [
    'template_key' => $row['template_key'] ?? null,
    'subscriber_id' => $row['subscriber_id'] ?? null,
  ]);
}

function messages_sync_campaign_counters(mysqli $conn, ?int $campaignId = null): void {
  ensure_message_queue_tables($conn);
  $scope = $campaignId && $campaignId > 0 ? ' AND c.id=' . (int)$campaignId : '';
  try {
    $conn->query("UPDATE message_campaigns c
      LEFT JOIN (
        SELECT campaign_id,
          COUNT(DISTINCT subscriber_id) AS total_rows,
          COUNT(DISTINCT CASE WHEN status IN ('QUEUED','RETRY','SENDING') THEN subscriber_id END) AS pending_rows,
          COUNT(DISTINCT CASE WHEN status='SENT' THEN subscriber_id END) AS sent_rows,
          COUNT(DISTINCT CASE WHEN status='FAILED' THEN subscriber_id END) AS failed_rows
        FROM message_outbox
        WHERE campaign_id IS NOT NULL
        GROUP BY campaign_id
      ) q ON q.campaign_id=c.id
      SET c.queued_count=COALESCE(q.total_rows,0),
          c.sent_count=COALESCE(q.sent_rows,0),
          c.failed_count=COALESCE(q.failed_rows,0),
          c.status=CASE
            WHEN c.status='PAUSED' THEN 'PAUSED'
            WHEN COALESCE(q.pending_rows,0)>0 THEN 'QUEUED'
            WHEN COALESCE(q.total_rows,0)=0 THEN c.status
            WHEN COALESCE(q.failed_rows,0)>0 THEN 'COMPLETED_WITH_ERRORS'
            ELSE 'COMPLETED'
          END
      WHERE 1=1 {$scope}");
  } catch (Throwable $e) {
    error_log('[messages] campaign counter sync failed: ' . $e->getMessage());
  }
}

function messages_campaign_recount_runs(mysqli $conn, array $runIds): void {
  $runIds = array_values(array_unique(array_filter(array_map('intval', $runIds), static fn(int $id): bool => $id > 0)));
  foreach ($runIds as $runId) {
    $runRs = @$conn->query("SELECT status FROM message_queue_runs WHERE id={$runId} LIMIT 1");
    if (!$runRs || !$runRs->num_rows) continue;
    $run = $runRs->fetch_assoc();

    $statsRs = @$conn->query("SELECT
      COUNT(*) AS total_count,
      SUM(status IN ('QUEUED','RETRY','SENDING')) AS pending_count,
      SUM(status='SENT') AS sent_count,
      SUM(status='FAILED') AS failed_count
      FROM message_outbox WHERE run_id={$runId}");
    $stats = $statsRs && $statsRs->num_rows ? $statsRs->fetch_assoc() : [];
    $total = max(0, (int)($stats['total_count'] ?? 0));
    $pending = max(0, (int)($stats['pending_count'] ?? 0));
    $sent = max(0, (int)($stats['sent_count'] ?? 0));
    $failed = max(0, (int)($stats['failed_count'] ?? 0));
    $processed = $sent + $failed;
    $status = (string)($run['status'] ?? '');
    $completionSql = '';

    if (messages_queue_run_is_active($status) && $total === 0) {
      $status = 'CANCELLED';
      $completionSql = ',completed_at=NOW()';
    } elseif (messages_queue_run_is_active($status) && $pending === 0) {
      $status = 'COMPLETED';
      $completionSql = ',completed_at=NOW()';
    }
    $statusEsc = $conn->real_escape_string($status);
    @$conn->query("UPDATE message_queue_runs SET
      status='{$statusEsc}',total_count={$total},processed_count={$processed},
      sent_count={$sent},failed_count={$failed}{$completionSql}
      WHERE id={$runId} LIMIT 1");
  }
}

function messages_campaign_pause(mysqli $conn, int $campaignId): array {
  ensure_message_queue_tables($conn);
  $campaignId = (int)$campaignId;
  if ($campaignId <= 0) return ['ok'=>false,'error'=>'الحملة غير صالحة.'];
  if (!messages_queue_named_lock($conn, 1)) {
    return ['ok'=>false,'error'=>'يجري تنفيذ الطابور الآن. حاول إيقاف الحملة بعد لحظات.'];
  }

  $runIds = [];
  try {
    $conn->begin_transaction();
    $campaignRs = $conn->query("SELECT status FROM message_campaigns WHERE id={$campaignId} LIMIT 1 FOR UPDATE");
    if (!$campaignRs || !$campaignRs->num_rows) {
      $conn->rollback();
      return ['ok'=>false,'error'=>'الحملة غير موجودة.'];
    }
    $campaign = $campaignRs->fetch_assoc();
    if ((string)($campaign['status'] ?? '') === 'PAUSED') {
      $conn->commit();
      return ['ok'=>true,'message'=>'الحملة موقوفة بالفعل.'];
    }

    $sendingRs = $conn->query("SELECT COUNT(*) AS c FROM message_outbox
      WHERE campaign_id={$campaignId} AND status='SENDING'");
    $sending = (int)(($sendingRs ? $sendingRs->fetch_assoc() : [])['c'] ?? 0);
    if ($sending > 0) {
      $conn->rollback();
      return ['ok'=>false,'error'=>'توجد رسالة من الحملة قيد الإرسال الآن. حاول مجدداً بعد اكتمالها.'];
    }

    $runsRs = $conn->query("SELECT DISTINCT run_id FROM message_outbox
      WHERE campaign_id={$campaignId} AND run_id IS NOT NULL
        AND status IN ('QUEUED','RETRY')");
    if ($runsRs) while ($run = $runsRs->fetch_assoc()) $runIds[] = (int)$run['run_id'];

    $conn->query("UPDATE message_outbox SET status='PAUSED',run_id=NULL,
      last_error=NULL,status_updated_at=NOW()
      WHERE campaign_id={$campaignId} AND status IN ('QUEUED','RETRY')");
    $paused = max(0, (int)$conn->affected_rows);
    $conn->query("UPDATE message_campaigns SET status='PAUSED' WHERE id={$campaignId} LIMIT 1");
    $conn->commit();

    messages_campaign_recount_runs($conn, $runIds);
    return [
      'ok'=>true,
      'paused'=>$paused,
      'message'=>$paused > 0
        ? "تم إيقاف الحملة وحجز {$paused} رسالة غير مرسلة."
        : 'تم إيقاف الحملة. لا توجد رسائل معلّقة فيها.',
    ];
  } catch (Throwable $e) {
    try { $conn->rollback(); } catch (Throwable $ignored) {}
    error_log('[messages] campaign pause failed: ' . $e->getMessage());
    return ['ok'=>false,'error'=>'تعذر إيقاف الحملة.'];
  } finally {
    messages_queue_named_unlock($conn);
  }
}

function messages_campaign_resume(mysqli $conn, int $campaignId): array {
  ensure_message_queue_tables($conn);
  $campaignId = (int)$campaignId;
  if ($campaignId <= 0) return ['ok'=>false,'error'=>'الحملة غير صالحة.'];
  if (!messages_queue_named_lock($conn, 1)) {
    return ['ok'=>false,'error'=>'يجري تنفيذ الطابور الآن. حاول استئناف الحملة بعد لحظات.'];
  }

  try {
    $conn->begin_transaction();
    $campaignRs = $conn->query("SELECT status FROM message_campaigns WHERE id={$campaignId} LIMIT 1 FOR UPDATE");
    if (!$campaignRs || !$campaignRs->num_rows) {
      $conn->rollback();
      return ['ok'=>false,'error'=>'الحملة غير موجودة.'];
    }
    $campaign = $campaignRs->fetch_assoc();
    if ((string)($campaign['status'] ?? '') !== 'PAUSED') {
      $conn->rollback();
      return ['ok'=>false,'error'=>'الحملة ليست في حالة إيقاف.'];
    }

    $conn->query("UPDATE message_outbox SET status='QUEUED',run_id=NULL,
      last_error=NULL,status_updated_at=NOW()
      WHERE campaign_id={$campaignId} AND status='PAUSED'");
    $resumed = max(0, (int)$conn->affected_rows);
    $nextStatus = $resumed > 0 ? 'QUEUED' : 'DRAFT';
    $conn->query("UPDATE message_campaigns SET status='{$nextStatus}' WHERE id={$campaignId} LIMIT 1");
    $conn->commit();
    messages_sync_campaign_counters($conn, $campaignId);

    return [
      'ok'=>true,
      'resumed'=>$resumed,
      'message'=>$resumed > 0
        ? "تم استئناف الحملة وإعادة {$resumed} رسالة إلى الطابور."
        : 'تم استئناف الحملة، وهي جاهزة لإضافة جمهورها إلى الطابور.',
    ];
  } catch (Throwable $e) {
    try { $conn->rollback(); } catch (Throwable $ignored) {}
    error_log('[messages] campaign resume failed: ' . $e->getMessage());
    return ['ok'=>false,'error'=>'تعذر استئناف الحملة.'];
  } finally {
    messages_queue_named_unlock($conn);
  }
}

function messages_campaign_delete(mysqli $conn, int $campaignId): array {
  ensure_message_queue_tables($conn);
  $campaignId = (int)$campaignId;
  if ($campaignId <= 0) return ['ok'=>false,'error'=>'الحملة غير صالحة.'];
  if (!messages_queue_named_lock($conn, 1)) {
    return ['ok'=>false,'error'=>'يجري تنفيذ الطابور الآن. حاول حذف الحملة بعد لحظات.'];
  }

  $runIds = [];
  try {
    $conn->begin_transaction();
    $campaignRs = $conn->query("SELECT id FROM message_campaigns WHERE id={$campaignId} LIMIT 1 FOR UPDATE");
    if (!$campaignRs || !$campaignRs->num_rows) {
      $conn->rollback();
      return ['ok'=>false,'error'=>'الحملة غير موجودة.'];
    }

    $sendingRs = $conn->query("SELECT COUNT(*) AS c FROM message_outbox
      WHERE campaign_id={$campaignId} AND status='SENDING'");
    $sending = (int)(($sendingRs ? $sendingRs->fetch_assoc() : [])['c'] ?? 0);
    if ($sending > 0) {
      $conn->rollback();
      return ['ok'=>false,'error'=>'لا يمكن حذف الحملة أثناء إرسال إحدى رسائلها. أوقفها بعد اكتمال الرسالة الحالية.'];
    }

    $runsRs = $conn->query("SELECT DISTINCT run_id FROM message_outbox
      WHERE campaign_id={$campaignId} AND run_id IS NOT NULL");
    if ($runsRs) while ($run = $runsRs->fetch_assoc()) $runIds[] = (int)$run['run_id'];

    // Preserve the immutable delivery history while removing unsent work.
    $conn->query("UPDATE message_outbox SET campaign_id=NULL
      WHERE campaign_id={$campaignId} AND status IN ('SENT','FAILED')");
    $preserved = max(0, (int)$conn->affected_rows);
    $conn->query("DELETE FROM message_outbox WHERE campaign_id={$campaignId}");
    $removed = max(0, (int)$conn->affected_rows);
    $conn->query("DELETE FROM message_campaigns WHERE id={$campaignId} LIMIT 1");
    if ($conn->affected_rows !== 1) throw new RuntimeException('Campaign row was not deleted.');
    $conn->commit();

    messages_campaign_recount_runs($conn, $runIds);
    return [
      'ok'=>true,
      'removed'=>$removed,
      'preserved'=>$preserved,
      'message'=>'تم حذف الحملة ورسائلها غير المرسلة، مع الاحتفاظ بسجل الرسائل التي سبق إرسالها أو فشل إرسالها.',
    ];
  } catch (Throwable $e) {
    try { $conn->rollback(); } catch (Throwable $ignored) {}
    error_log('[messages] campaign delete failed: ' . $e->getMessage());
    return ['ok'=>false,'error'=>'تعذر حذف الحملة.'];
  } finally {
    messages_queue_named_unlock($conn);
  }
}

function messages_process_queue(mysqli $conn, int $limit = 50): array {
  ensure_message_queue_tables($conn);
  $limit = max(1, min(200, (int)$limit));
  $rs = $conn->query("SELECT * FROM message_outbox WHERE status IN ('QUEUED','RETRY') ORDER BY id ASC LIMIT $limit");
  if(!$rs) return ['ok'=>false,'error'=>'فشل قراءة الطابور'];
  $sent=0; $fail=0;
  while($r = $rs->fetch_assoc()){
    $id = (int)$r['id'];
    $conn->query("UPDATE message_outbox SET status='SENDING', try_count=try_count+1, status_updated_at=NOW() WHERE id=$id LIMIT 1");
    $res = messages_send_outbox_row($conn, $r);
    $tpl = $conn->real_escape_string((string)($r['template_key'] ?? ''));
    $ph  = $conn->real_escape_string((string)($r['to_phone'] ?? ''));
    $ch  = $conn->real_escape_string((string)($r['channel'] ?? ''));
    $sid = isset($r['subscriber_id']) && $r['subscriber_id'] !== null ? (int)$r['subscriber_id'] : 'NULL';

    if(!empty($res['ok'])){
      $sent++;
      $pid = '';
      if (isset($res['data']['messages'][0]['id'])) $pid = (string)$res['data']['messages'][0]['id'];
      if (isset($res['message_id'])) $pid = (string)$res['message_id'];
      $pidEsc = $conn->real_escape_string($pid);
      $resp = json_encode($res['data'] ?? $res, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES);
      $respEsc = $conn->real_escape_string((string)$resp);
      $conn->query("UPDATE message_outbox SET status='SENT', provider_message_id='$pidEsc', last_error=NULL, sent_at=NOW(), status_updated_at=NOW() WHERE id=$id LIMIT 1");
      $conn->query("INSERT INTO message_logs (template_key,subscriber_id,phone,channel,status,response) VALUES ('$tpl',$sid,'$ph','$ch','SENT','$respEsc')");
    } else {
      $fail++;
      $err = (string)($res['error'] ?? 'فشل الإرسال');
      $errEsc = $conn->real_escape_string($err);
      $tries = (int)($r['try_count'] ?? 0) + 1;
      $newStatus = ($tries < 3) ? 'RETRY' : 'FAILED';
      $conn->query("UPDATE message_outbox SET status='$newStatus', last_error='$errEsc', status_updated_at=NOW() WHERE id=$id LIMIT 1");
      $conn->query("INSERT INTO message_logs (template_key,subscriber_id,phone,channel,status,response) VALUES ('$tpl',$sid,'$ph','$ch','$newStatus','$errEsc')");
    }
  }
  messages_sync_campaign_counters($conn);
  return ['ok'=>true,'sent'=>$sent,'failed'=>$fail];
}

/**
 * Persistent queue runs
 *
 * A run owns a stable set of outbox rows through message_outbox.run_id.
 * This makes approval, progress reporting and retries deterministic even when
 * more messages are added while the current run is active.
 */

function messages_queue_run_is_active(string $status): bool {
  return in_array($status, ['TESTING','WAITING_CONFIRMATION','RUNNING','TEST_FAILED'], true);
}

function messages_queue_named_lock(mysqli $conn, int $timeout = 0): bool {
  $timeout = max(0, min(10, $timeout));
  $rs = @$conn->query("SELECT GET_LOCK('network_customers_message_queue_run', {$timeout}) AS locked");
  $row = $rs ? $rs->fetch_assoc() : null;
  return (int)($row['locked'] ?? 0) === 1;
}

function messages_queue_named_unlock(mysqli $conn): void {
  try { @$conn->query("SELECT RELEASE_LOCK('network_customers_message_queue_run')"); } catch (Throwable $e) {}
}

function messages_queue_run_public(?array $run): ?array {
  if (!$run) return null;
  $total = max(0, (int)($run['total_count'] ?? 0));
  $processed = max(0, (int)($run['processed_count'] ?? 0));
  $sent = max(0, (int)($run['sent_count'] ?? 0));
  $failed = max(0, (int)($run['failed_count'] ?? 0));
  return [
    'id' => (int)($run['id'] ?? 0),
    'status' => (string)($run['status'] ?? ''),
    'test_phone' => (string)($run['test_phone'] ?? ''),
    'test_channel' => (string)($run['test_channel'] ?? ''),
    'confirmation_required' => (int)($run['confirmation_required'] ?? 0) === 1,
    'test_sent' => (int)($run['test_sent'] ?? 0) === 1,
    'test_error' => (string)($run['test_error'] ?? ''),
    'total' => $total,
    'processed' => $processed,
    'queued' => max(0, $total - $processed),
    'sent' => $sent,
    'failed' => $failed,
    'percent' => $total > 0 ? min(100, (int)floor(($processed / $total) * 100)) : 0,
    'created_at' => (string)($run['created_at'] ?? ''),
    'started_at' => (string)($run['started_at'] ?? ''),
    'approved_at' => (string)($run['approved_at'] ?? ''),
    'completed_at' => (string)($run['completed_at'] ?? ''),
  ];
}

function messages_queue_get_run(mysqli $conn, int $runId): ?array {
  ensure_message_queue_tables($conn);
  $runId = (int)$runId;
  if ($runId <= 0) return null;
  $rs = @$conn->query("SELECT * FROM message_queue_runs WHERE id={$runId} LIMIT 1");
  return $rs && $rs->num_rows ? $rs->fetch_assoc() : null;
}

function messages_queue_active_run(mysqli $conn, bool $includeRecent = false): ?array {
  ensure_message_queue_tables($conn);
  $active = "'TESTING','WAITING_CONFIRMATION','RUNNING','TEST_FAILED'";
  $rs = @$conn->query("SELECT * FROM message_queue_runs WHERE status IN ({$active}) ORDER BY id DESC LIMIT 1");
  if ($rs && $rs->num_rows) return $rs->fetch_assoc();
  if (!$includeRecent) return null;
  $rs = @$conn->query("SELECT * FROM message_queue_runs ORDER BY id DESC LIMIT 1");
  return $rs && $rs->num_rows ? $rs->fetch_assoc() : null;
}

function messages_queue_send_test(mysqli $conn, string $phone, string $body, string $channel = 'sms', ?callable $sender = null): array {
  if ($sender !== null) return (array)$sender($conn, $phone, $body);
  if ($channel === 'whatsapp') {
    require_once __DIR__ . '/whatsapp_cloud.php';
    $whatsapp = new WhatsAppCloudSender($conn);
    return $whatsapp->sendText($phone, $body);
  }
  require_once __DIR__ . '/sms_provider.php';
  return sms_provider_send($conn, $phone, $body, ['source'=>'queue_test']);
}

function messages_queue_create_run(
  mysqli $conn,
  int $startedBy = 0,
  ?callable $testSender = null,
  ?int $campaignId = null
): array {
  ensure_message_queue_tables($conn);
  require_once __DIR__ . '/sms_provider.php';
  require_once __DIR__ . '/settings_store.php';

  if (!messages_queue_named_lock($conn, 1)) {
    return ['ok'=>false, 'error'=>'يجري تحديث طابور الرسائل الآن. حاول بعد لحظات.'];
  }

  try {
    $active = messages_queue_active_run($conn, false);
    if ($active) return ['ok'=>true, 'existing'=>true, 'run'=>messages_queue_run_public($active)];

    $cfg = sms_provider_normalize_config(setting_get($conn, 'sms_provider', []));
    $testPhone = trim((string)($cfg['test_phone'] ?? ''));
    if (preg_replace('/\D+/', '', $testPhone) === '') {
      return ['ok'=>false, 'error'=>'احفظ رقم الاختبار أولاً من صفحة إعدادات SMS.'];
    }

    $campaignScope = $campaignId && $campaignId > 0 ? ' AND campaign_id=' . (int)$campaignId : '';
    $countRs = @$conn->query("SELECT COUNT(*) AS c FROM message_outbox
      WHERE run_id IS NULL AND status IN ('QUEUED','RETRY'){$campaignScope}");
    $total = (int)(($countRs ? $countRs->fetch_assoc() : [])['c'] ?? 0);
    if ($total <= 0) return ['ok'=>false, 'empty'=>true, 'error'=>'لا توجد رسائل جاهزة في الطابور.'];

    $sampleRs = @$conn->query("SELECT body,channel FROM message_outbox
      WHERE run_id IS NULL AND status IN ('QUEUED','RETRY'){$campaignScope}
      ORDER BY id ASC LIMIT 1");
    $sample = $sampleRs && $sampleRs->num_rows ? $sampleRs->fetch_assoc() : null;
    $testBody = trim((string)($sample['body'] ?? ''));
    $testChannel = strtolower(trim((string)($sample['channel'] ?? 'sms')));
    if (!in_array($testChannel, ['sms','whatsapp'], true)) $testChannel = 'sms';
    if ($testBody === '') return ['ok'=>false, 'error'=>'تعذر تجهيز نص رسالة الاختبار من الطابور.'];

    $phoneEsc = $conn->real_escape_string($testPhone);
    $bodyEsc = $conn->real_escape_string($testBody);
    $channelEsc = $conn->real_escape_string($testChannel);
    $confirm = (string)($cfg['queue_confirmation_required'] ?? '1') === '1' ? 1 : 0;
    $actorSql = $startedBy > 0 ? (string)$startedBy : 'NULL';
    $ok = @$conn->query("INSERT INTO message_queue_runs
      (status,test_phone,test_body,test_channel,confirmation_required,total_count,started_by)
      VALUES ('TESTING','{$phoneEsc}','{$bodyEsc}','{$channelEsc}',{$confirm},{$total},{$actorSql})");
    if (!$ok) return ['ok'=>false, 'error'=>'تعذر إنشاء دورة إرسال جديدة.'];
    $runId = (int)$conn->insert_id;

    @$conn->query("UPDATE message_outbox SET run_id={$runId}
      WHERE run_id IS NULL AND status IN ('QUEUED','RETRY'){$campaignScope}
      ORDER BY id ASC LIMIT {$total}");
    $assigned = max(0, (int)$conn->affected_rows);
    if ($assigned <= 0) {
      @$conn->query("DELETE FROM message_queue_runs WHERE id={$runId}");
      return ['ok'=>false, 'empty'=>true, 'error'=>'لم تعد هناك رسائل جاهزة في الطابور.'];
    }
    if ($assigned !== $total) {
      $total = $assigned;
      @$conn->query("UPDATE message_queue_runs SET total_count={$total} WHERE id={$runId} LIMIT 1");
    }
  } finally {
    messages_queue_named_unlock($conn);
  }

  $testResult = messages_queue_send_test($conn, $testPhone, $testBody, $testChannel, $testSender);
  if (empty($testResult['ok'])) {
    $error = (string)($testResult['error'] ?? 'فشل إرسال رسالة الاختبار.');
    $errorEsc = $conn->real_escape_string($error);
    @$conn->query("UPDATE message_queue_runs SET status='TEST_FAILED',test_error='{$errorEsc}' WHERE id={$runId} LIMIT 1");
    $run = messages_queue_get_run($conn, $runId);
    return ['ok'=>false, 'test_failed'=>true, 'error'=>$error, 'run'=>messages_queue_run_public($run)];
  }

  $nextStatus = $confirm === 1 ? 'WAITING_CONFIRMATION' : 'RUNNING';
  @$conn->query("UPDATE message_queue_runs SET status='{$nextStatus}',test_sent=1,test_error=NULL,
    started_at=IF('{$nextStatus}'='RUNNING',NOW(),started_at) WHERE id={$runId} LIMIT 1");
  $run = messages_queue_get_run($conn, $runId);
  return [
    'ok'=>true,
    'test_sent'=>true,
    'requires_confirmation'=>$confirm === 1,
    'message'=>$confirm === 1
      ? 'تم إرسال رسالة الاختبار. راجع مضمونها ثم وافق على متابعة الطابور.'
      : 'تم إرسال رسالة الاختبار وبدأ تنفيذ الطابور.',
    'run'=>messages_queue_run_public($run),
  ];
}

function messages_queue_control_run(mysqli $conn, int $runId, string $action): array {
  ensure_message_queue_tables($conn);
  $runId = (int)$runId;
  $action = strtolower(trim($action));
  if ($runId <= 0) return ['ok'=>false,'error'=>'دورة الإرسال غير صالحة.'];
  if (!messages_queue_named_lock($conn, 1)) return ['ok'=>false,'error'=>'الطابور مشغول الآن.'];

  try {
    $run = messages_queue_get_run($conn, $runId);
    if (!$run) return ['ok'=>false,'error'=>'دورة الإرسال غير موجودة.'];
    $status = (string)$run['status'];

    if ($action === 'approve') {
      if ($status !== 'WAITING_CONFIRMATION') return ['ok'=>false,'error'=>'دورة الإرسال لا تنتظر موافقة.'];
      @$conn->query("UPDATE message_queue_runs SET status='RUNNING',approved_at=NOW(),started_at=COALESCE(started_at,NOW()) WHERE id={$runId} LIMIT 1");
    } elseif ($action === 'cancel') {
      if (!messages_queue_run_is_active($status)) return ['ok'=>false,'error'=>'دورة الإرسال منتهية بالفعل.'];
      @$conn->query("UPDATE message_outbox SET run_id=NULL WHERE run_id={$runId} AND status IN ('QUEUED','RETRY')");
      @$conn->query("UPDATE message_queue_runs SET status='CANCELLED',completed_at=NOW() WHERE id={$runId} LIMIT 1");
    } else {
      return ['ok'=>false,'error'=>'الإجراء غير معروف.'];
    }
    return ['ok'=>true,'run'=>messages_queue_run_public(messages_queue_get_run($conn, $runId))];
  } finally {
    messages_queue_named_unlock($conn);
  }
}

function messages_queue_process_step(mysqli $conn, int $runId, int $limit = 2, ?callable $sender = null): array {
  ensure_message_queue_tables($conn);
  $runId = (int)$runId;
  $limit = max(1, min(10, $limit));
  if ($runId <= 0) return ['ok'=>false,'error'=>'دورة الإرسال غير صالحة.'];
  if (!messages_queue_named_lock($conn, 0)) {
    $run = messages_queue_get_run($conn, $runId);
    return ['ok'=>true,'busy'=>true,'run'=>messages_queue_run_public($run)];
  }

  try {
    $run = messages_queue_get_run($conn, $runId);
    if (!$run) return ['ok'=>false,'error'=>'دورة الإرسال غير موجودة.'];
    if ((string)$run['status'] !== 'RUNNING') {
      return ['ok'=>true,'run'=>messages_queue_run_public($run)];
    }

    // Recover a row if PHP/the provider was interrupted after marking it as
    // SENDING. The named lock itself is released automatically on disconnect.
    @$conn->query("UPDATE message_outbox SET status='RETRY',
      last_error=COALESCE(NULLIF(last_error,''),'انقطع عامل الإرسال قبل استلام النتيجة'),
      status_updated_at=NOW()
      WHERE run_id={$runId} AND status='SENDING'
        AND (status_updated_at IS NULL OR status_updated_at < DATE_SUB(NOW(), INTERVAL 5 MINUTE))");

    $sentDelta = 0;
    $failedDelta = 0;
    for ($i = 0; $i < $limit; $i++) {
      $rs = @$conn->query("SELECT * FROM message_outbox
        WHERE run_id={$runId} AND status IN ('QUEUED','RETRY') ORDER BY id ASC LIMIT 1");
      if (!$rs || !$rs->num_rows) break;
      $row = $rs->fetch_assoc();
      $id = (int)$row['id'];
      @$conn->query("UPDATE message_outbox SET status='SENDING',try_count=try_count+1,status_updated_at=NOW() WHERE id={$id} AND run_id={$runId} LIMIT 1");

      $result = $sender !== null ? (array)$sender($conn, $row) : messages_send_outbox_row($conn, $row);
      $tpl = $conn->real_escape_string((string)($row['template_key'] ?? ''));
      $phone = $conn->real_escape_string((string)($row['to_phone'] ?? ''));
      $channel = $conn->real_escape_string((string)($row['channel'] ?? ''));
      $subscriberSql = isset($row['subscriber_id']) && $row['subscriber_id'] !== null ? (int)$row['subscriber_id'] : 'NULL';

      if (!empty($result['ok'])) {
        $providerId = (string)($result['message_id'] ?? ($result['data']['messages'][0]['id'] ?? ''));
        $providerEsc = $conn->real_escape_string($providerId);
        $responseEsc = $conn->real_escape_string((string)json_encode($result['data'] ?? $result, JSON_UNESCAPED_UNICODE|JSON_UNESCAPED_SLASHES));
        @$conn->query("UPDATE message_outbox SET status='SENT',provider_message_id='{$providerEsc}',last_error=NULL,sent_at=NOW(),status_updated_at=NOW() WHERE id={$id} LIMIT 1");
        @$conn->query("INSERT INTO message_logs (template_key,subscriber_id,phone,channel,status,response)
          VALUES ('{$tpl}',{$subscriberSql},'{$phone}','{$channel}','SENT','{$responseEsc}')");
        $sentDelta++;
        continue;
      }

      $error = (string)($result['error'] ?? 'فشل الإرسال');
      $errorEsc = $conn->real_escape_string($error);
      $tries = (int)($row['try_count'] ?? 0) + 1;
      $newStatus = $tries < 3 ? 'RETRY' : 'FAILED';
      @$conn->query("UPDATE message_outbox SET status='{$newStatus}',last_error='{$errorEsc}',status_updated_at=NOW() WHERE id={$id} LIMIT 1");
      @$conn->query("INSERT INTO message_logs (template_key,subscriber_id,phone,channel,status,response)
        VALUES ('{$tpl}',{$subscriberSql},'{$phone}','{$channel}','{$newStatus}','{$errorEsc}')");
      if ($newStatus === 'FAILED') $failedDelta++;
    }

    if ($sentDelta > 0 || $failedDelta > 0) {
      $processedDelta = $sentDelta + $failedDelta;
      @$conn->query("UPDATE message_queue_runs SET
        processed_count=processed_count+{$processedDelta},
        sent_count=sent_count+{$sentDelta},
        failed_count=failed_count+{$failedDelta}
        WHERE id={$runId} LIMIT 1");
    }

    $pendingRs = @$conn->query("SELECT COUNT(*) AS c FROM message_outbox WHERE run_id={$runId} AND status IN ('QUEUED','RETRY','SENDING')");
    $pending = (int)(($pendingRs ? $pendingRs->fetch_assoc() : [])['c'] ?? 0);
    if ($pending === 0) {
      @$conn->query("UPDATE message_queue_runs SET status='COMPLETED',processed_count=total_count,completed_at=NOW() WHERE id={$runId} LIMIT 1");
    }
    messages_sync_campaign_counters($conn);

    return ['ok'=>true,'run'=>messages_queue_run_public(messages_queue_get_run($conn, $runId))];
  } finally {
    messages_queue_named_unlock($conn);
  }
}

function messages_queue_statistics(mysqli $conn): array {
  ensure_message_queue_tables($conn);
  $stats = ['missing_phone'=>0,'sent'=>0,'failed'=>0];
  $missingWhere = ml_col_exists($conn, 'subscribers', 'phone')
    ? "(phone IS NULL OR TRIM(phone)='')"
    : '1=0';
  if (ml_col_exists($conn, 'subscribers', 'is_deleted')) $missingWhere .= " AND COALESCE(is_deleted,0)=0";
  $rs = @$conn->query("SELECT COUNT(*) AS c FROM subscribers WHERE {$missingWhere}");
  if ($rs) $stats['missing_phone'] = (int)($rs->fetch_assoc()['c'] ?? 0);
  $rs = @$conn->query("SELECT status,COUNT(*) AS c FROM message_outbox WHERE status IN ('SENT','FAILED') GROUP BY status");
  if ($rs) while ($row = $rs->fetch_assoc()) {
    if ($row['status'] === 'SENT') $stats['sent'] = (int)$row['c'];
    if ($row['status'] === 'FAILED') $stats['failed'] = (int)$row['c'];
  }
  return $stats;
}

function messages_queue_detail_rows(mysqli $conn, string $type, int $limit = 200): array {
  ensure_message_queue_tables($conn);
  $limit = max(1, min(500, $limit));
  if ($type === 'missing_phone') {
    $nameParts = [];
    if (ml_col_exists($conn,'subscribers','full_name')) $nameParts[] = "NULLIF(full_name,'')";
    if (ml_col_exists($conn,'subscribers','first_name')) {
      $father = ml_col_exists($conn,'subscribers','father_name') ? 'father_name' : "''";
      $last = ml_col_exists($conn,'subscribers','last_name') ? 'last_name' : "''";
      $nameParts[] = "NULLIF(TRIM(CONCAT_WS(' ',first_name,{$father},{$last})),'')";
    }
    if (ml_col_exists($conn,'subscribers','ppp_username')) $nameParts[] = "NULLIF(ppp_username,'')";
    if (ml_col_exists($conn,'subscribers','username')) $nameParts[] = "NULLIF(username,'')";
    $nameExpr = $nameParts ? 'COALESCE('.implode(',', $nameParts).",'مشترك')" : "'مشترك'";
    $userExpr = ml_col_exists($conn,'subscribers','ppp_username') ? 'ppp_username'
      : (ml_col_exists($conn,'subscribers','username') ? 'username' : "''");
    $notDeleted = ml_col_exists($conn,'subscribers','is_deleted') ? ' AND COALESCE(is_deleted,0)=0' : '';
    $rs = @$conn->query("SELECT id,{$nameExpr} AS subscriber_name,{$userExpr} AS subscriber_username
      FROM subscribers WHERE (phone IS NULL OR TRIM(phone)=''){$notDeleted} ORDER BY id DESC LIMIT {$limit}");
  } else {
    $status = $type === 'failed' ? 'FAILED' : 'SENT';
    $rs = @$conn->query("SELECT o.id,o.subscriber_id,o.to_phone,o.channel,o.template_key,o.status,
      o.try_count,o.last_error,o.created_at,o.sent_at,
      COALESCE(NULLIF(s.full_name,''),NULLIF(s.ppp_username,''),NULLIF(s.username,''),'—') AS subscriber_name
      FROM message_outbox o LEFT JOIN subscribers s ON s.id=o.subscriber_id
      WHERE o.status='{$status}' ORDER BY o.id DESC LIMIT {$limit}");
  }
  return $rs ? $rs->fetch_all(MYSQLI_ASSOC) : [];
}
