/**
 * Titan Messenger — Real-Time Messaging Server (v3.1)
 * ==================================================
 * Production-ready WebSocket + REST server for Titan Connect encrypted chat.
 * v3.1: Merged Titan Cloud Relay (upload/download) from standalone Space.
 *       Relay routes available at /relay/* — see cloudRelay.js
 *
 * Stack  : Node.js 20 LTS · Express 4 · ws 8 · better-sqlite3 9 · firebase-admin 12
 *          node-fetch 2 (CommonJS) · zlib (built-in) · multer · node-cron
 * Port   : 7860
 * Storage: /data/titan_messenger.db  (WAL-mode SQLite, HF Spaces persistent volume)
 * Archive: Private Hugging Face Dataset (HF_DATASET_REPO) — gzipped JSONL
 *
 * Security guarantees
 *  • Two-layer auth on every protected request (x-titan-key + Firebase ID Token)
 *  • uid ALWAYS derived from verified Firebase token — never from client payload
 *  • chat_id membership validated on every read / write
 *  • Parameterised SQL throughout — NO string interpolation
 *  • Encrypted payloads are NEVER logged; only innocuous metadata reaches stdout
 *
 * Changelog v4 (reliability / sync pass — "no message is ever lost")
 *  • ADDED: monotonic `rev` / `crev` columns. EVERY state change (insert, edit,
 *    delete, reaction, delivered, seen) bumps a global revision counter. Clients
 *    keep a cursor and call `sync_request {since_rev}` (WS) or GET /sync — the
 *    server replays everything the user missed (Telegram "getDifference" style).
 *  • ADDED: `recipient_uid` column (+ indexes) so sync is an index lookup.
 *  • CHANGED: message_ack is sent IMMEDIATELY after the DB commit. Push (FCM)
 *    and the Firestore ping run in the background and can no longer delay / fail
 *    the ack (they were awaited before, so a slow FCM call stalled the sender).
 *  • CHANGED: commit + live delivery is one synchronous block → frames reach a
 *    client in strictly increasing rev order (lets the client advance its cursor
 *    from live frames safely).
 *  • CHANGED: `delivered` is only set on a real client confirmation
 *    (`message_received`) or when the message is seen. The old 8 s timer marked
 *    messages delivered even if the device never got them.
 *  • FIXED (security): react_message / mark_seen (WS + REST) had no membership
 *    check — any logged-in user could react to / mark-seen any message id.
 *  • FIXED: message_updated frames never carried the new payload, so edits did
 *    not reach the other side. They do now.
 *  • FIXED: per-socket frame ordering (frames of one socket are processed
 *    sequentially), group delivery (recipientUidFrom returned garbage for
 *    groups), stale /history cache after new messages, unbounded WS payload,
 *    no send rate-limit, retry re-pushing FCM on every resend.
 *  • ADDED: error frames carry `code` (invalid | forbidden | rate_limited |
 *    internal) so the client knows what is retryable.
 *  • ADDED: archive_floor_rev (tells a long-offline client to refetch history),
 *    graceful shutdown closes sockets with 1001 and checkpoints the WAL.
 *
 * Changelog v3
 *  • ADDED: message_delivered receipt — sets delivered_at in DB and sends
 *    { type: "message_delivered", data: { id, delivered_at } } to the sender
 *    immediately after a successful live WebSocket delivery to the recipient.
 *  • FIXED: retryPendingArchives — pending_archive now stores both the local tmp
 *    file path and the dataset path so the gzip buffer can actually be re-uploaded.
 *  • FIXED: ALL SQLite statements are pre-compiled in `stmts` — no inline
 *    db.prepare() calls in hot paths (HTTP handlers, WS message loop).
 *  • FIXED: archive path-date parsing uses correct segment indices after split("/").
 *  • FIXED: message_id inputs from REST body are explicitly coerced to integers.
 *  • ADDED: guard against editing/deleting already-deleted messages (409 response).
 *  • ADDED: temp_id type-guard (must be string, else null).
 *  • FIXED: mark_seen runs inside a SQLite transaction for atomicity.
 *  • FIXED: chat_id split uses indexOf("_") — safe against UIDs that contain "_".
 */

"use strict";

// ─────────────────────────────────────────────────────────────────────────────
// 0. Imports
// ─────────────────────────────────────────────────────────────────────────────
const http          = require("http");
const fs            = require("fs");
const path          = require("path");
const zlib          = require("zlib");
const os            = require("os");
const { promisify } = require("util");

const express                        = require("express");
const { WebSocketServer, WebSocket } = require("ws");
const Database                       = require("better-sqlite3");
const admin                          = require("firebase-admin");
const fetch                          = require("node-fetch");

const gzip   = promisify(zlib.gzip);
const gunzip = promisify(zlib.gunzip);

// ─────────────────────────────────────────────────────────────────────────────
// 1. Environment variables & validation
// ─────────────────────────────────────────────────────────────────────────────
const PORT                 = parseInt(process.env.PORT || "7860", 10);
const TITAN_API_KEY        = process.env.TITAN_API_KEY;
const FB_SA_JSON           = process.env.FIREBASE_SERVICE_ACCOUNT;
const HF_TOKEN             = process.env.HF_TOKEN               || "";
const HF_DATASET_REPO      = process.env.HF_DATASET_REPO        || "";
const ARCHIVE_DAYS         = parseInt(process.env.ARCHIVE_DAYS           || "4",   10);
const ARCHIVE_INTERVAL_HRS = parseInt(process.env.ARCHIVE_INTERVAL_HOURS || "6",   10);
const ARCHIVE_DAILY_HOUR   = parseInt(process.env.ARCHIVE_DAILY_HOUR     || "3",   10);
const EMERGENCY_SIZE_MB    = parseInt(process.env.EMERGENCY_SIZE_MB      || "800", 10);

if (!TITAN_API_KEY) {
  console.error("[FATAL] TITAN_API_KEY env var is not set.");
  process.exit(1);
}
if (!FB_SA_JSON) {
  console.error("[FATAL] FIREBASE_SERVICE_ACCOUNT env var is not set.");
  process.exit(1);
}
if (!HF_TOKEN || !HF_DATASET_REPO) {
  console.warn("[WARN] HF_TOKEN or HF_DATASET_REPO not set — cold-storage archiving will be disabled.");
}

// ─────────────────────────────────────────────────────────────────────────────
// 2. Firebase Admin initialisation
// ─────────────────────────────────────────────────────────────────────────────
let serviceAccount;
try {
  serviceAccount = JSON.parse(FB_SA_JSON);
} catch (err) {
  console.error("[FATAL] FIREBASE_SERVICE_ACCOUNT is not valid JSON:", err.message);
  process.exit(1);
}

admin.initializeApp({ credential: admin.credential.cert(serviceAccount) });
console.log("[INFO] Firebase Admin initialised.");

// ─────────────────────────────────────────────────────────────────────────────
// 3. SQLite database initialisation
// ─────────────────────────────────────────────────────────────────────────────
const DATA_DIR = process.env.DATA_DIR || "/data";
const DB_PATH  = path.join(DATA_DIR, "titan_messenger.db");

if (!fs.existsSync(DATA_DIR)) {
  fs.mkdirSync(DATA_DIR, { recursive: true });
}

let db;
try {
  db = new Database(DB_PATH);
} catch (err) {
  console.error("[FATAL] Cannot open SQLite database at", DB_PATH, "—", err.message);
  process.exit(1);
}

// Performance + safety pragmas
db.pragma("journal_mode = WAL");
db.pragma("busy_timeout = 5000");
db.pragma("synchronous = NORMAL"); // safe with WAL, far fewer fsyncs per message
db.pragma("foreign_keys = ON");

// ── Full schema ───────────────────────────────────────────────────────────────
db.exec(`
  CREATE TABLE IF NOT EXISTS messages (
    id                INTEGER PRIMARY KEY AUTOINCREMENT,
    chat_id           TEXT    NOT NULL,
    sender_uid        TEXT    NOT NULL,
    encrypted_payload TEXT    NOT NULL,
    type              TEXT    NOT NULL DEFAULT 'text',
    edited            INTEGER NOT NULL DEFAULT 0,
    edited_at         REAL,
    delivered_at      REAL,
    seen_at           REAL,
    created_at        REAL    NOT NULL DEFAULT (unixepoch())
  );

  CREATE INDEX IF NOT EXISTS idx_messages_chat_created
    ON messages(chat_id, created_at);

  CREATE TABLE IF NOT EXISTS fcm_tokens (
    user_id    TEXT PRIMARY KEY,
    token      TEXT NOT NULL,
    updated_at REAL NOT NULL DEFAULT (unixepoch())
  );

  CREATE TABLE IF NOT EXISTS meta (
    key   TEXT PRIMARY KEY,
    value TEXT
  );

  -- Stores both the local tmp file path (for re-uploading on retry)
  -- and the dataset destination path.
  CREATE TABLE IF NOT EXISTS pending_archive (
    id                INTEGER PRIMARY KEY AUTOINCREMENT,
    tmp_file_path     TEXT    NOT NULL,
    dataset_file_path TEXT    NOT NULL UNIQUE,
    created_at        REAL    NOT NULL DEFAULT (unixepoch())
  );
`);
// Idempotent migrations
const _migs = [
  `ALTER TABLE messages ADD COLUMN reactions TEXT`,
  `ALTER TABLE messages ADD COLUMN reply_to TEXT`,
  `ALTER TABLE messages ADD COLUMN status TEXT NOT NULL DEFAULT 'sent'`,
  `CREATE TABLE IF NOT EXISTS group_members (
     group_id TEXT NOT NULL, user_id TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'member',
     joined_at REAL NOT NULL DEFAULT (unixepoch()), PRIMARY KEY (group_id, user_id))`,
  `CREATE INDEX IF NOT EXISTS idx_gm_group ON group_members(group_id)`,
  // v3.2 — reliability pass ────────────────────────────────────────────────
  // client_msg_id: echoes the client's temp_id. Lets processMessage() treat a
  // retried send_message as a replay instead of inserting a duplicate row.
  `ALTER TABLE messages ADD COLUMN client_msg_id TEXT`,
  // id is already the natural monotonic order (AUTOINCREMENT, no ties, unlike
  // created_at which only has 1-second resolution). This index makes
  // "WHERE chat_id = ? AND id < ? ORDER BY id DESC" (used for pagination) fast.
  `CREATE INDEX IF NOT EXISTS idx_messages_chat_id ON messages(chat_id, id)`,
  // v4 — sync pass ───────────────────────────────────────────────────────────
  // rev  : bumped on EVERY change (what the SENDER cares about: receipts too)
  // crev : bumped only on content-level changes (insert/edit/delete/reaction)
  //        — what the RECIPIENT cares about (they don't need a re-send because
  //        somebody flipped a delivered/seen flag).
  `ALTER TABLE messages ADD COLUMN recipient_uid TEXT`,
  `ALTER TABLE messages ADD COLUMN rev INTEGER NOT NULL DEFAULT 0`,
  `ALTER TABLE messages ADD COLUMN crev INTEGER NOT NULL DEFAULT 0`,
  `CREATE INDEX IF NOT EXISTS idx_messages_sender_rev ON messages(sender_uid, rev)`,
  `CREATE INDEX IF NOT EXISTS idx_messages_rcpt_crev ON messages(recipient_uid, crev)`,
];
for (const sql of _migs) { try { db.exec(sql); } catch {} }
// Partial unique index — must run after the column exists, and can't use
// "IF NOT EXISTS" safely inside the generic try/catch loop above on every
// startup without risking a silent skip if the column migration hasn't
// landed yet on some older deployments, so it's kept as its own guarded step.
try {
  db.exec(
    `CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_dedup
       ON messages(chat_id, sender_uid, client_msg_id)
       WHERE client_msg_id IS NOT NULL`
  );
} catch (err) {
  console.error("[ERROR] Could not create idx_messages_dedup:", err.message);
}

console.log("[INFO] SQLite ready — WAL mode enabled.");

// ─────────────────────────────────────────────────────────────────────────────
// 4. Pre-compiled SQL statements
//    ALL statements are compiled here at startup. No db.prepare() calls in
//    hot paths (request handlers, WS message loop).
// ─────────────────────────────────────────────────────────────────────────────
const MSG_COLS =
  "id, chat_id, sender_uid, encrypted_payload, type, edited, edited_at, delivered_at, seen_at, " +
  "created_at, reactions, reply_to, client_msg_id, recipient_uid, rev, crev";

const stmts = {

  // ── FCM tokens ──────────────────────────────────────────────────────────────
  upsertFcmToken : db.prepare(
    "INSERT OR REPLACE INTO fcm_tokens (user_id, token, updated_at) VALUES (?, ?, unixepoch())"
  ),
  lookupFcmToken : db.prepare(
    "SELECT token FROM fcm_tokens WHERE user_id = ?"
  ),
  deleteFcmToken : db.prepare(
    "DELETE FROM fcm_tokens WHERE user_id = ?"
  ),

  // ── Messages — insert & full read-back ─────────────────────────────────────
  insertMessage : db.prepare(
    `INSERT INTO messages (chat_id, sender_uid, encrypted_payload, type, reply_to, client_msg_id, recipient_uid, rev, crev)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
  ),
  selectMessageFull : db.prepare(`SELECT ${MSG_COLS} FROM messages WHERE id = ?`),
  // Lightweight ownership / participant check — does NOT return encrypted_payload
  selectMessageMeta : db.prepare(
    "SELECT id, chat_id, sender_uid, type, recipient_uid, seen_at, delivered_at FROM messages WHERE id = ?"
  ),
  // Idempotency lookup — a retried send_message (same temp_id) finds the row
  // that was already created instead of inserting a duplicate.
  selectByClientMsgId : db.prepare(
    `SELECT ${MSG_COLS} FROM messages WHERE chat_id = ? AND sender_uid = ? AND client_msg_id = ?`
  ),

  // ── History ─────────────────────────────────────────────────────────────────
  selectHistory : db.prepare(
    `SELECT ${MSG_COLS} FROM messages
     WHERE chat_id = ? AND created_at < ?
     ORDER BY created_at DESC, id DESC
     LIMIT ?`
  ),
  selectHistoryBeforeId : db.prepare(
    `SELECT ${MSG_COLS} FROM messages
     WHERE chat_id = ? AND id < ?
     ORDER BY id DESC
     LIMIT ?`
  ),

  // ── Receipts (every flip bumps rev so the SENDER's sync sees it) ───────────
  setDelivered : db.prepare(
    "UPDATE messages SET delivered_at = unixepoch(), rev = ? WHERE id = ? AND delivered_at IS NULL"
  ),
  markOneSeen : db.prepare(
    `UPDATE messages SET seen_at = unixepoch(), rev = ?,
            delivered_at = COALESCE(delivered_at, unixepoch())
     WHERE id = ? AND seen_at IS NULL`
  ),
  selectUnseenInChat : db.prepare(
    "SELECT id FROM messages WHERE chat_id = ? AND sender_uid != ? AND seen_at IS NULL AND type != 'deleted'"
  ),

  // ── Edit / soft delete / reaction (content-level → bump rev AND crev) ──────
  editMessage : db.prepare(
    `UPDATE messages SET encrypted_payload = ?, edited = 1, edited_at = unixepoch(), rev = ?, crev = ?
     WHERE id = ? AND sender_uid = ? AND type != 'deleted'`
  ),
  selectEditedAt : db.prepare(
    "SELECT id, chat_id, edited, edited_at, rev FROM messages WHERE id = ?"
  ),
  softDeleteMessage : db.prepare(
    `UPDATE messages SET type = 'deleted', encrypted_payload = '[deleted]', rev = ?, crev = ?
     WHERE id = ? AND sender_uid = ? AND type != 'deleted'`
  ),
  updateReaction : db.prepare(
    "UPDATE messages SET reactions = ?, rev = ?, crev = ? WHERE id = ? AND type != 'deleted'"
  ),

  // ── Sync (the heart of v4) ─────────────────────────────────────────────────
  // sender sees a row when `rev` moves (incl. receipts); recipient when `crev`
  // moves; group members (recipient_uid IS NULL) via group_members.
  selectSync : db.prepare(
    `SELECT ${MSG_COLS},
            CASE WHEN sender_uid = @uid THEN rev ELSE crev END AS sync_rev
     FROM messages
     WHERE (sender_uid = @uid AND rev > @since)
        OR (recipient_uid = @uid AND crev > @since)
        OR (recipient_uid IS NULL AND sender_uid != @uid AND crev > @since
            AND chat_id IN (SELECT group_id FROM group_members WHERE user_id = @uid))
     ORDER BY sync_rev ASC
     LIMIT @lim`
  ),
  // First-ever sync (cursor 0): don't replay history, just the unread backlog.
  selectBaselineUnseen : db.prepare(
    `SELECT ${MSG_COLS}, crev AS sync_rev FROM messages
     WHERE recipient_uid = ? AND seen_at IS NULL AND type != 'deleted'
     ORDER BY crev ASC LIMIT ?`
  ),

  // ── Archive — read ─────────────────────────────────────────────────────────
  selectOldMessages : db.prepare(
    `SELECT ${MSG_COLS} FROM messages WHERE created_at < ? ORDER BY created_at ASC`
  ),
  selectOldMessagesLimit : db.prepare(
    `SELECT ${MSG_COLS} FROM messages WHERE created_at < ? ORDER BY created_at ASC LIMIT ?`
  ),
  // For emergency archive: find the timestamp of the Nth oldest message (0-indexed offset)
  selectNthOldestTs : db.prepare(
    "SELECT created_at FROM messages ORDER BY created_at ASC LIMIT 1 OFFSET ?"
  ),

  // ── Archive — delete ───────────────────────────────────────────────────────
  deleteMessageById : db.prepare(
    "DELETE FROM messages WHERE id = ?"
  ),

  // ── Meta ────────────────────────────────────────────────────────────────────
  getMeta : db.prepare("SELECT value FROM meta WHERE key = ?"),
  setMeta : db.prepare("INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)"),

  // ── Pending archive ─────────────────────────────────────────────────────────
  insertPending : db.prepare(
    "INSERT OR IGNORE INTO pending_archive (tmp_file_path, dataset_file_path) VALUES (?, ?)"
  ),
  allPending : db.prepare(
    "SELECT id, tmp_file_path, dataset_file_path FROM pending_archive ORDER BY created_at ASC"
  ),
  deletePending : db.prepare(
    "DELETE FROM pending_archive WHERE id = ?"
  ),

  // ── Group members ───────────────────────────────────────────────────────────
  isGroupMember : db.prepare(
    "SELECT 1 FROM group_members WHERE group_id = ? AND user_id = ? LIMIT 1"
  ),
  insertGroupMember : db.prepare(
    "INSERT OR IGNORE INTO group_members (group_id, user_id, role) VALUES (?, ?, ?)"
  ),
  getGroupMembers : db.prepare(
    "SELECT user_id FROM group_members WHERE group_id = ?"
  ),
};

// ─────────────────────────────────────────────────────────────────────────────
// 4b. Revision counter + one-time backfill
//     `rev` is a single global, strictly increasing integer. It is persisted in
//     the meta table so it can never go backwards across restarts (even after
//     old rows were archived away).
// ─────────────────────────────────────────────────────────────────────────────
let _rev = 0;
(function initRevCounter() {
  const a = db.prepare("SELECT COALESCE(MAX(rev), 0) AS m FROM messages").get().m;
  const metaRow = stmts.getMeta.get("rev_counter");
  const b = parseInt((metaRow && metaRow.value) || "0", 10) || 0;
  _rev = Math.max(a, b);
})();

function nextRev() {
  _rev += 1;
  stmts.setMeta.run("rev_counter", String(_rev));
  return _rev;
}

// Rows written by v3 have rev = 0: give them a rev (in id order) and a recipient.
(function backfillRevs() {
  const rows = db.prepare("SELECT id, chat_id, sender_uid FROM messages WHERE rev = 0 ORDER BY id ASC").all();
  if (!rows.length) return;
  const upd = db.prepare("UPDATE messages SET recipient_uid = ?, rev = ?, crev = ? WHERE id = ?");
  db.transaction(() => {
    for (const r of rows) {
      const sep = r.chat_id.indexOf("_");
      let rcpt = null;
      if (sep > 0 && sep < r.chat_id.length - 1) {
        const a = r.chat_id.slice(0, sep), b = r.chat_id.slice(sep + 1);
        rcpt = a === r.sender_uid ? b : a;
      }
      const rv = nextRev();
      upd.run(rcpt, rv, rv, r.id);
    }
  })();
  console.log(`[INFO] Backfilled rev/recipient for ${rows.length} legacy message(s).`);
})();

// ─────────────────────────────────────────────────────────────────────────────
// 5. Utility helpers
// ─────────────────────────────────────────────────────────────────────────────

/** Returns the first 8 chars of a string for safe debug logging. Never logs full IDs. */
function redact(s) {
  return typeof s === "string" ? s.slice(0, 8) + "…" : "<nil>";
}

/**
 * Validate that `uid` is one of the two participants encoded in `chatId`.
 * chatId format: "<uidA>_<uidB>" where uidA < uidB lexicographically.
 * Uses indexOf to find the first underscore, supporting UIDs that contain "_".
 */
async function isChatMemberAsync(chatId, uid) {
  if (!chatId || !uid) return false;
  try { if (stmts.isGroupMember.get(chatId, uid)) return true; } catch {}
  const sep = chatId.indexOf("_");
  if (sep > 0 && sep < chatId.length - 1) {
    // FIX: split on the FIRST underscore only (same as recipientUidFrom).
    // The previous `chatId.split("_")` broke for any UID that itself
    // contains an underscore: split() cuts on EVERY "_", so a 3+-segment
    // result meant `[a, b]` never equalled either real UID and BOTH
    // legitimate participants were wrongly denied membership (silent
    // "Forbidden" on send_message / history for those accounts).
    const a = chatId.slice(0, sep);
    const b = chatId.slice(sep + 1);
    return a === uid || b === uid;
  }
  // Fallback to Firestore for Legacy Groups
  try {
    const doc = await admin.firestore().collection('titan_groups').doc(chatId).get();
    if (doc.exists) {
      const data = doc.data();
      if (data.members && data.members.includes(uid)) {
        // Cache EVERY member (not just the caller) so group fan-out knows
        // all recipients without another Firestore round-trip.
        for (const m of data.members) { try { stmts.insertGroupMember.run(chatId, String(m), 'member'); } catch {} }
        return true;
      }
    }
  } catch (e) { console.error("[ERROR] Firestore Group Sync:", e.message); }
  return false;
}

/** Return the other participant in a two-person chat. */
function recipientUidFrom(chatId, uid) {
  const sep = chatId.indexOf("_");
  const a   = chatId.slice(0, sep);
  const b   = chatId.slice(sep + 1);
  return a === uid ? b : a;
}

/** Current Unix timestamp (whole seconds). */
function nowSec() {
  return Math.floor(Date.now() / 1000);
}

/** Safely coerce `v` to a positive integer, or return null. */
function toInt(v) {
  const n = parseInt(v, 10);
  return Number.isFinite(n) && n > 0 ? n : null;
}

// ─────────────────────────────────────────────────────────────────────────────
// 6. Authentication helpers
// ─────────────────────────────────────────────────────────────────────────────

function checkTitanKey(headerValue) {
  return typeof headerValue === "string" && headerValue === TITAN_API_KEY;
}

async function verifyFirebaseToken(bearerHeader) {
  if (!bearerHeader || !bearerHeader.startsWith("Bearer ")) {
    throw new Error("Missing or malformed Authorization header.");
  }
  const idToken = bearerHeader.slice(7).trim();
  return admin.auth().verifyIdToken(idToken);
}

/**
 * Express middleware: enforces both auth layers.
 * Attaches req.uid (from verified Firebase token ONLY) on success.
 */
async function authMiddleware(req, res, next) {
  if (!checkTitanKey(req.headers["x-titan-key"])) {
    return res.status(401).json({ error: "Unauthorized" });
  }
  try {
    const decoded = await verifyFirebaseToken(req.headers["authorization"]);
    req.uid = decoded.uid; // uid NEVER from request body
    next();
  } catch {
    return res.status(401).json({ error: "Unauthorized" });
  }
}

// ─────────────────────────────────────────────────────────────────────────────
// 7. FCM push notifications
// ─────────────────────────────────────────────────────────────────────────────
const FCM_BODY_MAP = {
  text          : "رسالة جديدة 📩",
  image         : "📷 صورة",
  video         : "🎥 فيديو",
  voice         : "🎤 رسالة صوتية",
  file          : "📁 ملف",
  audio         : "🎵 صوت",
  sticker       : "🎨 ستيكر",
  "cloud-drop" : "☁️ ملف مشترك",
};

/**
 * Sends an FCM push notification to `recipientUid`.
 * Silently skips if no token is registered.
 * Removes stale tokens automatically on registration-not-found errors.
 */
async function sendPushNotification(recipientUid, senderName, type, extraData = {}) {
  const row = stmts.lookupFcmToken.get(recipientUid);
  if (!row) return;

  const body = FCM_BODY_MAP[type] || "رسالة جديدة 📩";
  // For text messages, show the message type as body and sender as title
  // (content stays encrypted — never include encrypted_payload in FCM)
  try {
    await admin.messaging().send({
      token        : row.token,
      notification : { title: senderName || "Titan", body },
      // data payload: available to service worker for badge/list update
      // ALL values must be strings for FCM data payloads
      data: {
        type       : type  || "text",
        senderUid  : String(extraData.senderUid   || ""),
        chatId     : String(extraData.chatId       || ""),
        messageId  : String(extraData.messageId    || ""),
        senderName : String(senderName             || ""),
        senderIcon : String(extraData.senderIcon   || ""),  // ✅ avatar URL for SW grouped notif
      },
      // Android: high priority so notification wakes device
      android: { priority: "high" },
      // APNS (iOS): alert + badge increment
      apns: { payload: { aps: { badge: 1, sound: "default" } } },
      // Web push: set icon + badge
      webpush: {
        notification: { icon: "/icons/icon-192x192.png", badge: "/icons/badge-72x72.png" },
        headers     : { Urgency: "high" }
      },
    });
    console.log("[INFO] FCM push sent to", redact(recipientUid));
  } catch (err) {
    if (err.code === "messaging/registration-token-not-found" ||
        err.code === "messaging/registration-token-not-registered") {
      stmts.deleteFcmToken.run(recipientUid);
      console.log("[INFO] Stale FCM token removed for", redact(recipientUid));
      // Tell the recipient's device (if online) to re-register a fresh token
      wsDeliver(recipientUid, { type: "refresh_fcm_token" });
    } else {
      console.error("[ERROR] FCM send failed:", err.code || err.message);
    }
  }
}

// ─────────────────────────────────────────────────────────────────────────────
// 8. In-memory WebSocket connection map
//    uid → Set<WebSocket>  (a user may have multiple device connections)
// ─────────────────────────────────────────────────────────────────────────────
/** @type {Map<string, Set<WebSocket>>} */
const wsConnections = new Map();

/** Cache of uid → displayName to avoid repeated admin.auth().getUser() calls. */
const wsNames = new Map();
/** Cache of uid → photoURL (avatar) for FCM notification icons. */
const wsIcons = new Map();

// Slow-consumer guard: a socket whose send buffer exceeds this is dead weight.
// Terminating it is safe — the client reconnects and sync_request replays anything it missed.
const WS_MAX_BUFFERED = 4 * 1024 * 1024;

function wsRegister(uid, ws) {
  if (!wsConnections.has(uid)) wsConnections.set(uid, new Set());
  wsConnections.get(uid).add(ws);
}

function wsUnregister(uid, ws) {
  const set = wsConnections.get(uid);
  if (!set) return;
  set.delete(ws);
  if (set.size === 0) wsConnections.delete(uid);
}

/**
 * Deliver a JSON frame to all open sockets of `uid`.
 * @returns {boolean} true if at least one socket accepted the frame.
 */
function wsDeliver(uid, frame) {
  const set = wsConnections.get(uid);
  if (!set || set.size === 0) return false;
  const payload = JSON.stringify(frame);
  let delivered = false;
  for (const ws of set) {
    if (ws.readyState !== WebSocket.OPEN) continue;
    if (ws.bufferedAmount > WS_MAX_BUFFERED) {
      console.warn("[WARN] Slow WS consumer — terminating", redact(uid));
      try { ws.terminate(); } catch {}
      continue;
    }
    try { ws.send(payload); delivered = true; } catch {}
  }
  return delivered;
}

/** Deliver one frame to several users (deduplicated). */
function wsDeliverMany(uids, frame) {
  let any = false;
  for (const u of new Set(uids)) { if (u && wsDeliver(u, frame)) any = true; }
  return any;
}

/** Safely serialise and send an object to a single WebSocket. */
function wsSend(ws, obj) {
  if (ws && ws.readyState === WebSocket.OPEN) {
    ws.send(JSON.stringify(obj));
  }
}

/** Broadcast a frame to every currently connected client (used for presence). */
function wsBroadcast(frame) {
  const payload = JSON.stringify(frame);
  for (const set of wsConnections.values()) {
    for (const ws of set) {
      if (ws.readyState === WebSocket.OPEN) ws.send(payload);
    }
  }
}

// ─────────────────────────────────────────────────────────────────────────────
// 9. Core message processing & state mutations (shared by HTTP and WS paths)
//
//    INVARIANT: every mutation below is  [bump rev + write row + live-deliver]
//    in ONE synchronous block (no `await` in between). Therefore frames reach a
//    given client in strictly increasing `rev` order, which is what lets the
//    client treat "rev of the last frame I processed" as a safe sync cursor.
// ─────────────────────────────────────────────────────────────────────────────

const MAX_PAYLOAD_CHARS = 64_000;
const SYNC_MAX_ROWS     = 300;
const SYNC_MAX_BYTES    = 1_500_000;

// ── /history response cache (burst-reconnect protection) ─────────────────────
// Key: `${uid}:${chat_id}` — TTL 5 s, non-paginated requests only. v4 invalidates
// it on every mutation so a fresh message can never be hidden for 5 s.
const _historyCache = new Map();
const _HIST_TTL = 5_000;
setInterval(() => {
  const now = Date.now();
  for (const [k, v] of _historyCache) if (now - v.ts > _HIST_TTL) _historyCache.delete(k);
}, 60_000).unref();
function invalidateHistoryCache(chatId, uids) {
  for (const u of uids) _historyCache.delete(`${u}:${chatId}`);
}

// ── Chat resolution ──────────────────────────────────────────────────────────
/** Who else (besides the sender) should receive a message in `chatId`. */
function resolveChat(chatId, senderUid) {
  let members = [];
  try { members = stmts.getGroupMembers.all(chatId).map(r => r.user_id); } catch {}
  if (members.length) return { recipients: members.filter(u => u !== senderUid), direct: false };
  const sep = chatId.indexOf("_");
  if (sep > 0 && sep < chatId.length - 1) {
    const a = chatId.slice(0, sep), b = chatId.slice(sep + 1);
    return { recipients: [a === senderUid ? b : a], direct: true };
  }
  return { recipients: [], direct: false };
}

/** Everyone who should hear about a change to message `meta` (sender included). */
function audienceOf(meta) {
  return [meta.sender_uid, ...resolveChat(meta.chat_id, meta.sender_uid).recipients];
}

/** Sync (no Firestore) participant test for an existing message row. */
function isParticipant(meta, uid) {
  if (meta.sender_uid === uid) return true;
  if (meta.recipient_uid && meta.recipient_uid === uid) return true;
  try { return !!stmts.isGroupMember.get(meta.chat_id, uid); } catch { return false; }
}

/** Strip server-internal columns before a row leaves the server. */
function publicRow(r) {
  if (!r) return r;
  const { recipient_uid, crev, ...rest } = r; // eslint-disable-line no-unused-vars
  return rest;
}

// ── Input validation ─────────────────────────────────────────────────────────
function validateSend(chatId, payload, type, tempId) {
  if (!chatId || typeof chatId !== "string" || chatId.length > 200) return "chat_id is required (string, <= 200 chars).";
  if (!payload || typeof payload !== "string")                       return "encrypted_payload (non-empty string) is required.";
  if (payload.length > MAX_PAYLOAD_CHARS)                            return "encrypted_payload too large.";
  if (type !== undefined && type !== null && (typeof type !== "string" || type.length > 32)) return "type must be a short string.";
  if (tempId && (typeof tempId !== "string" || tempId.length > 100)) return "temp_id must be a string <= 100 chars.";
  return null;
}

// ── Per-user send rate limit (flood / runaway-retry protection) ──────────────
const _sendBuckets = new Map();
function allowSend(uid) {
  const now = Date.now();
  let b = _sendBuckets.get(uid);
  if (!b || now - b.t > 5000) { b = { t: now, n: 0 }; _sendBuckets.set(uid, b); }
  b.n += 1;
  return b.n <= 40; // 40 messages / 5 s per user
}
setInterval(() => {
  const now = Date.now();
  for (const [k, v] of _sendBuckets) if (now - v.t > 60_000) _sendBuckets.delete(k);
}, 60_000).unref();

const fail = (status, code, message) => ({ ok: false, status, code, message });
const OK   = { ok: true, status: 200 };

// ── Background notifications (never block the ack) ───────────────────────────
async function notifyRecipient(recipient, senderName, type, extra, senderUid) {
  // FCM — sent ALWAYS (even when a socket is open): the socket can belong to an
  // app that is backgrounded. The service worker de-duplicates if the chat is visible.
  try { await sendPushNotification(recipient, senderName, type || "text", extra); }
  catch (e) { console.error("[ERROR] push:", e.message); }

  // Firestore "wake" ping (recipient's UI refetches via /sync or /history).
  try {
    await admin.firestore().collection("titan_sync").doc(recipient).set({
      lastEvent : "new_msg",
      from      : senderUid,
      ts        : Date.now(),
      _trigger  : Math.random(),
    }, { merge: true });
  } catch (syncErr) { console.error("[ERROR] Firebase Sync Ping Failed:", syncErr.message); }
}

// ── Insert (atomic: rev bump + row) ──────────────────────────────────────────
const commitInsert = db.transaction((chatId, senderUid, payload, type, replyToJSON, clientMsgId, recipientUid) => {
  const rev = nextRev();
  const r = stmts.insertMessage.run(chatId, senderUid, payload, type, replyToJSON, clientMsgId, recipientUid, rev, rev);
  return stmts.selectMessageFull.get(r.lastInsertRowid);
});

/**
 * Store a message (idempotent on temp_id), ACK the sender immediately, deliver
 * live, and schedule push notifications in the background.
 * Synchronous on purpose — see the INVARIANT above. Returns { stored, ack }.
 */
function processMessage({ chatId, senderUid, senderName, senderIcon = "", encryptedPayload, type, tempId, replyTo, senderWs }) {
  const clientMsgId = (typeof tempId === "string" && tempId) ? tempId : null;
  const { recipients, direct } = resolveChat(chatId, senderUid);

  let stored   = clientMsgId ? stmts.selectByClientMsgId.get(chatId, senderUid, clientMsgId) : undefined;
  const isReplay = !!stored;

  let created = false;
  if (!stored) {
    const replyToJSON = replyTo ? JSON.stringify(replyTo) : null;
    try {
      stored  = commitInsert(chatId, senderUid, encryptedPayload, type || "text", replyToJSON,
                             clientMsgId, direct ? recipients[0] : null);
      created = true;
    } catch (err) {
      // Lost a race with a concurrent identical send → reuse the winner's row.
      if (clientMsgId && /UNIQUE/i.test(err.message || "")) {
        stored = stmts.selectByClientMsgId.get(chatId, senderUid, clientMsgId);
      }
      if (!stored) throw err;
    }
    invalidateHistoryCache(chatId, [senderUid, ...recipients]);
    console.log(`[INFO] Message stored (chat=${redact(chatId)}, len=${encryptedPayload.length}, rev=${stored.rev})`);
  }

  // ── 1. ACK the sender FIRST — the message is durable, nothing below may delay it
  const ack = {
    type       : "message_ack",
    temp_id    : tempId || null,
    message_id : stored.id,
    server_ts  : stored.created_at,
    rev        : stored.rev,
  };
  if (senderWs) wsSend(senderWs, ack);   // HTTP path merges `ack` into its JSON response

  // A replay of a message the recipient already confirmed: nothing more to do.
  if (isReplay && stored.delivered_at) return { stored, ack };

  // ── 2. Live delivery (sync, so frame order == rev order)
  const frame = { type: "new_message", data: publicRow(stored) };
  for (const r of recipients) wsDeliver(r, frame);
  // Delivered receipt is NOT set here — only when the device confirms with
  // `message_received` (see doDelivered) or when the message is seen.

  // ── 3. Background push + Firestore ping (first send only → retries can't spam)
  if (created) {
    for (const r of recipients) {
      notifyRecipient(r, senderName, type || "text", {
        senderUid, chatId, messageId: stored.id, senderIcon: senderIcon || "",
      }, senderUid).catch(() => {});
    }
  }
  return { stored, ack };
}

// ── Receipts ─────────────────────────────────────────────────────────────────
/** Recipient confirms the device actually processed these message ids. */
function doDelivered(uid, rawIds) {
  const out = [];
  db.transaction((list) => {
    for (const raw of list) {
      const id = toInt(raw); if (!id) continue;
      const m = stmts.selectMessageMeta.get(id);
      if (!m || m.sender_uid === uid || m.delivered_at || m.type === "deleted") continue;
      if (!isParticipant(m, uid)) continue;
      const rev = nextRev();
      if (stmts.setDelivered.run(rev, id).changes) out.push({ id, sender: m.sender_uid, chat_id: m.chat_id, rev });
    }
  })(rawIds);
  for (const o of out) {
    wsDeliver(o.sender, { type: "message_delivered", message_id: o.id, chat_id: o.chat_id, delivered_at: nowSec(), rev: o.rev });
    invalidateHistoryCache(o.chat_id, [o.sender, uid]);
  }
  return OK;
}

/** Recipient read these messages. Authorisation: must be a participant and not the sender. */
function doSeen(uid, rawIds) {
  const groups = new Map(); // `${sender}|${chat}` → { sender, chat_id, ids[], rev }
  db.transaction((list) => {
    for (const raw of list) {
      const id = toInt(raw); if (!id) continue;
      const m = stmts.selectMessageMeta.get(id);
      if (!m || m.type === "deleted" || m.seen_at || m.sender_uid === uid) continue;
      if (!isParticipant(m, uid)) continue;
      const rev = nextRev();
      if (!stmts.markOneSeen.run(rev, id).changes) continue;
      const k = `${m.sender_uid}|${m.chat_id}`;
      if (!groups.has(k)) groups.set(k, { sender: m.sender_uid, chat_id: m.chat_id, ids: [], rev: 0 });
      const g = groups.get(k); g.ids.push(id); g.rev = rev; // rev of the LAST row in the batch
    }
  })(rawIds);
  for (const g of groups.values()) {
    wsDeliver(g.sender, { type: "message_seen", message_ids: g.ids, seen_by: uid, chat_id: g.chat_id, rev: g.rev });
    invalidateHistoryCache(g.chat_id, [g.sender, uid]);
  }
  return OK;
}

function doSeenChat(uid, chatId) {
  return doSeen(uid, stmts.selectUnseenInChat.all(chatId, uid).map(r => r.id));
}

// ── Edit / delete / react ────────────────────────────────────────────────────
function doEdit(uid, rawId, payload) {
  const msgId = toInt(rawId);
  if (!msgId || !payload || typeof payload !== "string" || payload.length > MAX_PAYLOAD_CHARS)
    return fail(400, "invalid", "message_id (positive integer) and a payload string (<= 64k) are required.");
  const m = stmts.selectMessageMeta.get(msgId);
  if (!m)                       return fail(404, "not_found", "Message not found.");
  if (m.sender_uid !== uid)     return fail(403, "forbidden", "Forbidden.");
  if (m.type === "deleted")     return fail(409, "conflict",  "Cannot edit a deleted message.");

  const rev = db.transaction(() => { const r = nextRev(); stmts.editMessage.run(payload, r, r, msgId, uid); return r; })();
  const u = stmts.selectEditedAt.get(msgId);
  const aud = audienceOf(m);
  wsDeliverMany(aud, {
    type: "message_updated", message_id: u.id, chat_id: u.chat_id, rev,
    // v4: the NEW payload travels with the frame (it was missing before, so the
    // other side never saw the edited text).
    data: { id: u.id, chat_id: u.chat_id, edited: true, edited_at: u.edited_at, encrypted_payload: payload },
  });
  invalidateHistoryCache(m.chat_id, aud);
  return OK;
}

function doDelete(uid, rawId) {
  const msgId = toInt(rawId);
  if (!msgId) return fail(400, "invalid", "message_id (positive integer) is required.");
  const m = stmts.selectMessageMeta.get(msgId);
  if (!m)                   return fail(404, "not_found", "Message not found.");
  if (m.sender_uid !== uid) return fail(403, "forbidden", "Forbidden.");
  if (m.type === "deleted") return fail(409, "conflict",  "Message already deleted.");

  const rev = db.transaction(() => { const r = nextRev(); stmts.softDeleteMessage.run(r, r, msgId, uid); return r; })();
  const aud = audienceOf(m);
  wsDeliverMany(aud, { type: "message_deleted", message_id: msgId, chat_id: m.chat_id, rev });
  invalidateHistoryCache(m.chat_id, aud);
  return OK;
}

function doReact(uid, rawId, reaction) {
  const msgId = toInt(rawId);
  if (!msgId) return fail(400, "invalid", "message_id (positive integer) is required.");
  if (reaction !== undefined && reaction !== null && (typeof reaction !== "string" || reaction.length > 32))
    return fail(400, "invalid", "reaction must be a short string.");
  const m = stmts.selectMessageMeta.get(msgId);
  if (!m)                      return fail(404, "not_found", "Message not found.");
  if (!isParticipant(m, uid))  return fail(403, "forbidden", "Forbidden.");   // v4: was missing
  if (m.type === "deleted")    return fail(409, "conflict",  "Message was deleted.");

  const rev = db.transaction(() => {
    const r = nextRev();
    stmts.updateReaction.run(reaction ? JSON.stringify({ emoji: reaction }) : null, r, r, msgId);
    return r;
  })();
  const aud = audienceOf(m);
  wsDeliverMany(aud, { type: "react_message", message_id: msgId, chat_id: m.chat_id, reaction: reaction || null, from: uid, rev });
  invalidateHistoryCache(m.chat_id, aud);

  // Offline peers: wake them through Firestore so their UI refetches.
  for (const r of aud) {
    if (r === uid) continue;
    admin.firestore().collection("titan_sync").doc(r).set({
      lastEvent: "react_msg", from: uid, msgId, reaction: reaction || null, ts: Date.now(), _trigger: Math.random(),
    }, { merge: true }).catch(() => {});
  }
  return OK;
}

// ── Sync ─────────────────────────────────────────────────────────────────────
/**
 * Everything user `uid` missed since revision `sinceRev`.
 *  baseline : cursor was 0 (fresh install) → only the unread backlog, no history replay.
 *  reset    : the cursor is older than what was archived to cold storage → the
 *             client must refetch history for the chats it cares about.
 *  upto_rev : the cursor value the client should store after applying `rows`.
 */
function buildSync(uid, sinceRev, limit) {
  const lim     = Math.min(Math.max(toInt(limit) || 100, 1), SYNC_MAX_ROWS);
  const since   = Math.max(parseInt(sinceRev, 10) || 0, 0);
  const counter = _rev;
  const floorRow = stmts.getMeta.get("archive_floor_rev");
  const floor    = parseInt((floorRow && floorRow.value) || "0", 10) || 0;

  if (since === 0) {
    const rows = stmts.selectBaselineUnseen.all(uid, 500).map(r => ({ ...publicRow(r) }));
    return { rows, has_more: false, upto_rev: counter, baseline: true, reset: false };
  }

  let rows = stmts.selectSync.all({ uid, since, lim: lim + 1 });
  let has_more = rows.length > lim;
  if (has_more) rows.length = lim;

  // Byte cap — keep at least one row so progress is always made.
  let bytes = 0, keep = 0;
  for (const r of rows) {
    bytes += (r.encrypted_payload ? r.encrypted_payload.length : 0) + 400;
    if (keep > 0 && bytes > SYNC_MAX_BYTES) break;
    keep++;
  }
  if (keep < rows.length) { rows.length = keep; has_more = true; }

  const upto = has_more ? rows[rows.length - 1].sync_rev : counter;
  return { rows: rows.map(publicRow), has_more, upto_rev: upto, baseline: false, reset: since < floor };
}


// ─────────────────────────────────────────────────────────────────────────────
// 10. Archive — Hugging Face Dataset cold storage
// ─────────────────────────────────────────────────────────────────────────────
const ARCHIVE_BASE = "archive";
const INDEX_PATH   = "archive/.index.json";

/** Full URL for reading a file from the private dataset (raw content). */
function hfRawUrl(filePath) {
  return `https://huggingface.co/datasets/${HF_DATASET_REPO}/resolve/main/${filePath}`;
}

/**
 * Upload a single Buffer to the private HF Dataset using the commit endpoint.
 *
 * The old /upload/ endpoint was retired (HTTP 410). All uploads now go through
 * the NDJSON commit API — the same path used by hfBatchCommit and backupDatabase.
 *
 * @param {string}      datasetFilePath  destination path inside the dataset repo
 * @param {Buffer}      contentBuffer
 * @param {string|null} tmpFilePath      local tmp path for retry tracking (null = no retry)
 * @returns {Promise<boolean>}
 */
async function uploadToDataset(datasetFilePath, contentBuffer, tmpFilePath) {
  if (!HF_TOKEN || !HF_DATASET_REPO) {
    console.warn("[WARN] Archive upload skipped — HF credentials not configured.");
    return false;
  }

  const commitUrl = `https://huggingface.co/api/datasets/${HF_DATASET_REPO}/commit/main`;

  const ndjson = [
    JSON.stringify({ key: "header", value: { summary: `Titan upload: ${datasetFilePath}` } }),
    JSON.stringify({
      key  : "file",
      value: {
        path    : datasetFilePath,
        encoding: "base64",
        content : contentBuffer.toString("base64"),
      },
    }),
  ].join("\n");

  try {
    const res = await fetch(commitUrl, {
      method : "POST",
      headers: {
        Authorization  : `Bearer ${HF_TOKEN}`,
        "Content-Type" : "application/x-ndjson",
      },
      body: ndjson,
    });

    if (!res.ok) {
      const text = await res.text().catch(() => "");
      console.error(`[ERROR] HF upload failed (${res.status}): ${text.slice(0, 120)}`);
      if (tmpFilePath) stmts.insertPending.run(tmpFilePath, datasetFilePath);
      return false;
    }

    console.log(`[INFO] Archive uploaded: ${datasetFilePath}`);
    return true;
  } catch (err) {
    console.error("[ERROR] HF upload exception:", err.message);
    if (tmpFilePath) stmts.insertPending.run(tmpFilePath, datasetFilePath);
    return false;
  }
}

/**
 * Download a file from the HF Dataset and decompress if it ends with .gz.
 * @param {string} filePath
 * @returns {Promise<string|null>}  raw string content, or null on error/404
 */
async function downloadFromDataset(filePath) {
  if (!HF_TOKEN || !HF_DATASET_REPO) return null;
  try {
    const res = await fetch(hfRawUrl(filePath), {
      headers: { Authorization: `Bearer ${HF_TOKEN}` },
    });
    if (res.status === 404) return null;
    if (!res.ok) {
      console.error(`[ERROR] HF download failed (${res.status}) for ${filePath}`);
      return null;
    }
    const buf = await res.buffer();
    if (filePath.endsWith(".gz")) {
      const decompressed = await gunzip(buf);
      return decompressed.toString("utf8");
    }
    return buf.toString("utf8");
  } catch (err) {
    console.error("[ERROR] HF download exception:", err.message);
    return null;
  }
}

/**
 * Archive messages older than `olderThanTimestamp`.
 * Groups rows by chat_id → gzip JSONL → upload → update index → delete from SQLite.
 * Failed uploads are queued in pending_archive (both tmp + dataset paths stored).
 *
 * @param {number}      olderThanTimestamp  Unix seconds
 * @param {number|null} limit               cap for emergency runs; null = no cap
 */
async function archiveMessages(olderThanTimestamp, limit = null) {
  const rows = limit
    ? stmts.selectOldMessagesLimit.all(olderThanTimestamp, limit)
    : stmts.selectOldMessages.all(olderThanTimestamp);

  if (rows.length === 0) {
    console.log("[INFO] Archive: no messages to move.");
    return;
  }
  console.log(`[INFO] Archive: processing ${rows.length} message(s).`);

  // Group by chat_id, preserving the ASC order from the query
  const byChatId = new Map();
  for (const row of rows) {
    if (!byChatId.has(row.chat_id)) byChatId.set(row.chat_id, []);
    byChatId.get(row.chat_id).push(row);
  }

  // Fetch the current index object (create a blank one if it doesn't exist yet)
  let indexObj = { last_updated: new Date().toISOString(), files: {} };
  const indexRaw = await downloadFromDataset(INDEX_PATH);
  if (indexRaw) {
    try { indexObj = JSON.parse(indexRaw); }
    catch { /* leave default */ }
  }
  if (!indexObj.files || typeof indexObj.files !== "object") indexObj.files = {};

  const successIds = []; // IDs to delete from SQLite after all uploads succeed
  let maxArchivedRev = 0;  // v4: highest rev that leaves SQLite → archive_floor_rev

  for (const [chatId, messages] of byChatId) {
    // Use the date of the OLDEST message in this batch as the path date
    const msgDate     = new Date(messages[0].created_at * 1000);
    const year        = msgDate.getUTCFullYear();
    const month       = String(msgDate.getUTCMonth() + 1).padStart(2, "0");
    const day         = String(msgDate.getUTCDate()).padStart(2, "0");
    const datasetPath = `${ARCHIVE_BASE}/${year}/${month}/${day}/${chatId}.jsonl.gz`;

    // Build JSONL — one JSON object per line, no array wrapper
    const jsonl = messages.map(m => JSON.stringify(m)).join("\n");

    // Compress
    let compressed;
    try {
      compressed = await gzip(Buffer.from(jsonl, "utf8"));
    } catch (err) {
      console.error("[ERROR] Archive gzip failed for chat", redact(chatId), "—", err.message);
      continue;
    }

    // Write to a local tmp file.
    // This path is stored in pending_archive so retryPendingArchives can re-read the buffer.
    const tmpPath = path.join(os.tmpdir(), `titan_arc_${chatId}_${Date.now()}.jsonl.gz`);
    try {
      fs.writeFileSync(tmpPath, compressed);
    } catch (err) {
      console.error("[ERROR] Tmp file write failed:", err.message);
      continue;
    }

    // Attempt upload — on failure, insertPending is called with both paths
    const ok = await uploadToDataset(datasetPath, compressed, tmpPath);
    if (!ok) {
      // Tmp file is intentionally left on disk for retry
      continue;
    }

    // Update in-memory index
    if (!indexObj.files[chatId]) indexObj.files[chatId] = [];
    if (!indexObj.files[chatId].includes(datasetPath)) {
      indexObj.files[chatId].push(datasetPath);
    }

    successIds.push(...messages.map(m => m.id));
    for (const m of messages) if ((m.rev || 0) > maxArchivedRev) maxArchivedRev = m.rev;

    // Upload succeeded — tmp file no longer needed
    try { fs.unlinkSync(tmpPath); } catch { /* best-effort */ }
  }

  // Upload the updated index (reflects only successfully archived files)
  if (successIds.length > 0) {
    indexObj.last_updated = new Date().toISOString();
    const indexBuf = Buffer.from(JSON.stringify(indexObj, null, 2), "utf8");
    // Index is small and non-critical; no retry tracking for it
    await uploadToDataset(INDEX_PATH, indexBuf, null);
  }

  // Delete successfully archived messages from SQLite inside a single transaction
  if (successIds.length > 0) {
    const txDelete = db.transaction((ids) => {
      for (const id of ids) stmts.deleteMessageById.run(id);
    });
    txDelete(successIds);
    if (maxArchivedRev > 0) {
      const fr  = stmts.getMeta.get("archive_floor_rev");
      const old = parseInt((fr && fr.value) || "0", 10) || 0;
      stmts.setMeta.run("archive_floor_rev", String(Math.max(old, maxArchivedRev)));
    }
    console.log(`[INFO] Archive: ${successIds.length} message(s) moved to cold storage.`);
  }
}

/**
 * Retry uploads that previously failed.
 *
 * In v3, pending_archive stores the local tmp file path alongside the dataset path,
 * so the gzip buffer can be re-read and re-uploaded without going back to SQLite
 * (the messages may have already been deleted on a partial success run).
 */
async function retryPendingArchives() {
  const pending = stmts.allPending.all();
  if (pending.length === 0) return;

  console.log(`[INFO] Retrying ${pending.length} pending archive upload(s).`);

  for (const row of pending) {
    const { id, tmp_file_path, dataset_file_path } = row;

    if (!fs.existsSync(tmp_file_path)) {
      // OS cleaned up the tmp file (or container was replaced) — data is unrecoverable
      console.warn(`[WARN] Tmp archive file missing: ${path.basename(tmp_file_path)} — removing stale entry.`);
      stmts.deletePending.run(id);
      continue;
    }

    let compressed;
    try {
      compressed = fs.readFileSync(tmp_file_path);
    } catch (err) {
      console.error("[ERROR] Cannot read tmp archive file:", err.message);
      stmts.deletePending.run(id);
      continue;
    }

    // Retry upload — do NOT pass tmpFilePath so we don't create a duplicate pending entry
    const ok = await uploadToDataset(dataset_file_path, compressed, null);
    if (ok) {
      stmts.deletePending.run(id);
      try { fs.unlinkSync(tmp_file_path); } catch { /* best-effort */ }
      console.log(`[INFO] Retry upload succeeded: ${dataset_file_path}`);
    } else {
      console.warn(`[WARN] Retry upload failed: ${dataset_file_path} — will try again next cycle.`);
    }
  }
}

/**
 * Emergency archive: archive the oldest 100,000 messages when the DB file
 * exceeds EMERGENCY_SIZE_MB in size.
 */
async function checkEmergencyArchive() {
  try {
    const stat   = fs.statSync(DB_PATH);
    const sizeMB = stat.size / (1024 * 1024);
    if (sizeMB > EMERGENCY_SIZE_MB) {
      console.log(`[INFO] Emergency archive triggered (DB: ${sizeMB.toFixed(1)} MB > limit: ${EMERGENCY_SIZE_MB} MB).`);
      // Find the created_at of the 100,000th oldest message (offset 99999 is 0-indexed)
      const row       = stmts.selectNthOldestTs.get(99999);
      const threshold = row ? row.created_at : nowSec();
      await archiveMessages(threshold, 100000);
    }
  } catch (err) {
    console.error("[ERROR] Emergency archive check failed:", err.message);
  }
}

/**
 * Daily archive: move all messages older than ARCHIVE_DAYS days to cold storage,
 * then attempt to complete any previously failed uploads.
 */
async function dailyArchive() {
  console.log(`[INFO] Daily archive started (threshold: ${ARCHIVE_DAYS} day(s) ago).`);
  const threshold = nowSec() - ARCHIVE_DAYS * 86400;
  await archiveMessages(threshold);
  await retryPendingArchives();
  console.log("[INFO] Daily archive complete.");
}

// ─────────────────────────────────────────────────────────────────────────────
// 11. Scheduler
// ─────────────────────────────────────────────────────────────────────────────

// In-memory guard (resets on restart — the meta table is the durable source of truth)
let lastDailyArchiveDate = null;

// Every 30 minutes: check whether the daily archive is due
setInterval(async () => {
  const now      = new Date();
  const todayStr = now.toISOString().slice(0, 10); // "YYYY-MM-DD"
  const hour     = now.getUTCHours();

  if (hour !== ARCHIVE_DAILY_HOUR) return;
  if (lastDailyArchiveDate === todayStr) return; // already ran in-process today

  // Double-check the durable meta table (guards against duplicate runs after a restart)
  const metaRow = stmts.getMeta.get("last_daily_archive");
  if (metaRow && metaRow.value === todayStr) {
    lastDailyArchiveDate = todayStr; // sync in-memory guard
    return;
  }

  // Mark as done BEFORE starting (prevents a second run if the job is slow)
  lastDailyArchiveDate = todayStr;
  stmts.setMeta.run("last_daily_archive", todayStr);

  await dailyArchive().catch(err => console.error("[ERROR] Daily archive job:", err.message));

}, 30 * 60 * 1000);

// Every ARCHIVE_INTERVAL_HRS hours: check DB size and trigger emergency archive if needed
setInterval(async () => {
  await checkEmergencyArchive().catch(err => console.error("[ERROR] Emergency archive check:", err.message));
}, ARCHIVE_INTERVAL_HRS * 60 * 60 * 1000);

// On startup: attempt to complete any previously failed uploads from the last run
retryPendingArchives().catch(err => console.error("[ERROR] Startup retry archives:", err.message));

// ─────────────────────────────────────────────────────────────────────────────
// 11b. Database Backup — gzip snapshot → Hugging Face Dataset
// ─────────────────────────────────────────────────────────────────────────────
const BACKUP_KEEP   = 3;          // keep only the 3 most recent backups
const BACKUP_FOLDER = "backups";

async function backupDatabase() {
  if (!HF_TOKEN || !HF_DATASET_REPO) {
    console.warn("[WARN] backupDatabase: HF_TOKEN or HF_DATASET_REPO not set — skipping backup.");
    return;
  }

  const timestamp = new Date().toISOString().replace(/[:.]/g, "-"); // safe for filenames
  const remotePath = `${BACKUP_FOLDER}/titan_messenger_${timestamp}.db.gz`;

  try {
    // Read the live DB file
    const dbBuf = fs.readFileSync(DB_PATH);

    // Compress with gzip
    const compressed = await gzip(dbBuf);

    // Upload to Hugging Face Dataset
    await uploadToDataset(remotePath, compressed, null);
    console.log(`[INFO] backupDatabase: snapshot uploaded → ${remotePath}`);

    // Prune old backups — keep only the BACKUP_KEEP most recent
    try {
      const entries = await listDatasetFolder(BACKUP_FOLDER);
      // entries are objects with a .path property; sort ascending by path (ISO timestamp sorts lexically)
      const backupFiles = entries
        .filter(e => e.path && e.path.startsWith(BACKUP_FOLDER + "/") && e.path.endsWith(".db.gz"))
        .map(e => e.path)
        .sort();

      if (backupFiles.length > BACKUP_KEEP) {
        const toDelete = backupFiles.slice(0, backupFiles.length - BACKUP_KEEP);
        await deleteFromDataset(toDelete);
        console.log(`[INFO] backupDatabase: pruned ${toDelete.length} old backup(s), kept ${BACKUP_KEEP}`);
      }
    } catch (pruneErr) {
      console.error("[ERROR] backupDatabase: prune step failed:", pruneErr.message);
    }

  } catch (err) {
    console.error("[ERROR] backupDatabase:", err.message);
  }
}

/** List files inside a dataset folder via the HF API tree endpoint. */
async function listDatasetFolder(folder) {
  const url = `https://huggingface.co/api/datasets/${HF_DATASET_REPO}/tree/main/${encodeURIComponent(folder)}?recursive=false`;
  const res = await fetch(url, { headers: { Authorization: `Bearer ${HF_TOKEN}` } });
  if (!res.ok) return [];
  const data = await res.json().catch(() => []);
  return Array.isArray(data) ? data : [];
}

/** Delete a list of file paths from the dataset via a single NDJSON commit. */
async function deleteFromDataset(filePaths) {
  if (!filePaths || !filePaths.length) return;
  const lines = [
    JSON.stringify({ key: "header", value: { summary: `Titan backup prune: delete ${filePaths.length} file(s)` } }),
    ...filePaths.map(p => JSON.stringify({ key: "deletedFile", value: { path: p } })),
  ].join("\n");

  const res = await fetch(`https://huggingface.co/api/datasets/${HF_DATASET_REPO}/commit/main`, {
    method:  "POST",
    headers: { Authorization: `Bearer ${HF_TOKEN}`, "Content-Type": "application/x-ndjson" },
    body:    lines,
  });
  if (!res.ok) console.error(`[ERROR] deleteFromDataset: HF commit failed (${res.status})`);
}

// Run every 6 hours
setInterval(() => {
  backupDatabase().catch(err => console.error("[ERROR] Scheduled backup:", err.message));
}, 6 * 60 * 60 * 1000);

// Also run 30 seconds after startup
setTimeout(() => {
  backupDatabase().catch(err => console.error("[ERROR] Startup backup:", err.message));
}, 30_000);

// ─────────────────────────────────────────────────────────────────────────────
// 12. Express application & REST routes
// ─────────────────────────────────────────────────────────────────────────────
const app = express();

// 🟢 إضافة CORS Middleware للسماح للـ PWA بالتواصل مع السيرفر
app.use((req, res, next) => {
  res.header("Access-Control-Allow-Origin", "*");
  res.header("Access-Control-Allow-Methods", "GET, POST, OPTIONS");
  res.header("Access-Control-Allow-Headers", "Content-Type, Authorization, x-titan-key");
  
  // الرد فوراً على طلبات الـ Preflight اللي بيبعتها المتصفح
  if (req.method === "OPTIONS") {
    return res.status(200).end();
  }
  next();
});

app.use(express.json({ limit: "64kb" })); // Payloads are ciphertext — 64 KB is generous

// ── Cloud Relay (merged from titan-cloud-relay Space) ────────────────────────
// Handles media/file uploads + downloads via HF Dataset.
// Routes: POST /relay/upload-chunk, GET /relay/download-chunk, GET /relay/health
const cloudRelay = require("./cloudRelay");
app.use("/relay", cloudRelay);

// ── GET /health ── public, no auth ───────────────────────────────────────────
app.get("/health", (_req, res) => {
  res.json({ status: "ok", uptime: process.uptime() });
});

// ── GET /history ─────────────────────────────────────────────────────────────
// Pagination: prefer `before_id` (exact, no ties — see selectHistoryBeforeId).
// `before` (a Unix-seconds timestamp) is kept for backward compatibility with
// any client that hasn't been rebuilt yet; created_at only has 1-second
// resolution so it can tie between messages, which is why before_id exists.
app.get("/history", authMiddleware, async (req, res) => {
  const { chat_id, limit, before, before_id } = req.query;

  if (!chat_id) return res.status(400).json({ error: "chat_id is required." });

  const beforeId   = toInt(before_id); // null if absent/invalid
  const isPaginated = !!before || !!beforeId;

  // Dedup guard: same user + same chat + no pagination cursor = serve cache
  const _ckey   = `${req.uid}:${chat_id}`;
  const _cached = _historyCache.get(_ckey);
  if (_cached && !isPaginated && (Date.now() - _cached.ts < _HIST_TTL)) {
    return res.json(_cached.result); // silent — no log spam
  }

  console.log(`[INFO] Fetching history for chat: ${redact(chat_id)}`);

  if (!(await isChatMemberAsync(chat_id, req.uid))) return res.status(403).json({ error: "Forbidden." });

  const safeLimit  = Math.min(Math.max(parseInt(limit, 10) || 50, 1), 200);
  const safeBefore = parseFloat(before) || nowSec() + 9_999_999;

  // ── 1. Hot data from SQLite ─────────────────────────────────────────────
  let messages = beforeId
    ? stmts.selectHistoryBeforeId.all(chat_id, beforeId, safeLimit)
    : stmts.selectHistory.all(chat_id, safeBefore, safeLimit);

  // ── 2. Cold data from archive (only if the page is incomplete) ──────────
  if (messages.length < safeLimit && HF_TOKEN && HF_DATASET_REPO) {
    try {
      // We want archive rows strictly older than the oldest hot row.
      // In id mode, compare by id (exact); in legacy timestamp mode, by created_at.
      const oldestHotId = messages.length > 0 ? messages[messages.length - 1].id         : beforeId;
      const oldestHotTs  = messages.length > 0 ? messages[messages.length - 1].created_at : safeBefore;

      const indexRaw = await downloadFromDataset(INDEX_PATH);
      if (indexRaw) {
        const indexObj  = JSON.parse(indexRaw);
        const filePaths = (indexObj.files && indexObj.files[chat_id]) || [];

        // Sort descending by path (lexicographic order = chronological for YYYY/MM/DD paths)
        // so we read the most-recent archive files first.
        const sorted = [...filePaths].sort().reverse();

        const archiveMsgs = [];
        for (const fp of sorted) {
          if (archiveMsgs.length + messages.length >= safeLimit) break;

          // Path format: "archive/YYYY/MM/DD/chatId.jsonl.gz"
          // After split("/"): [0]="archive" [1]=YYYY [2]=MM [3]=DD [4]="chatId.jsonl.gz"
          const segs = fp.split("/");
          if (segs.length < 5) continue;

          // Convert the file's date to a Unix timestamp (start of that UTC day)
          // Only usable as a skip-optimisation in legacy timestamp mode — id
          // has no fixed relationship to a calendar date, so skip the shortcut
          // (still correct, just reads a few more archive files) in id mode.
          if (!beforeId) {
            const fileStartTs = Date.UTC(+segs[1], +segs[2] - 1, +segs[3]) / 1000;
            if (fileStartTs >= safeBefore) continue;
          }

          const jsonl = await downloadFromDataset(fp);
          if (!jsonl) continue;

          const archRows = jsonl
            .split("\n")
            .filter(Boolean)
            .map(line => { try { return JSON.parse(line); } catch { return null; } })
            .filter(r => r !== null && (beforeId ? r.id < oldestHotId : r.created_at < oldestHotTs))
            .sort((a, b) => (beforeId ? b.id - a.id : b.created_at - a.created_at)); // newest first, matching hot rows

          archiveMsgs.push(...archRows);
        }

        const remaining = safeLimit - messages.length;
        messages = [...messages, ...archiveMsgs.slice(0, remaining)];
      }
    } catch (err) {
      // Archive failure is non-fatal; return whatever SQLite provided
      console.error("[ERROR] Archive history fetch:", err.message);
    }
  }

  // v4: never leak server-internal columns (archive rows carry them too)
  messages = messages.map(publicRow);

  // Cache the result for the dedup guard (only for non-paginated requests)
  if (!isPaginated) {
    _historyCache.set(_ckey, { ts: Date.now(), result: { messages } });
  }

  // oldest_id: exact cursor for the next page (pass back as before_id).
  // null once fewer than a full page came back — that's the "no more" signal.
  const oldestId = messages.length > 0 ? Math.min(...messages.map(m => m.id)) : null;
  res.json({ messages, oldest_id: oldestId, has_more: messages.length >= safeLimit });
});

// ── POST /send_message ────────────────────────────────────────────────────────
app.post("/send_message", authMiddleware, async (req, res) => {
  const { chat_id, encrypted_payload, type, temp_id, reply_to } = req.body || {};

  const bad = validateSend(chat_id, encrypted_payload, type, temp_id);
  if (bad) return res.status(400).json({ error: bad, code: "invalid" });
  if (!allowSend(req.uid)) return res.status(429).json({ error: "Too many messages.", code: "rate_limited", retry_after_ms: 3000 });
  if (!(await isChatMemberAsync(chat_id, req.uid))) {
    return res.status(403).json({ error: "Forbidden.", code: "forbidden" });
  }

  try {
    // Look up displayName from WS connection cache first (avoids Firebase Admin call per message)
    let senderName = wsNames.get(req.uid) || "Someone";
    if (senderName === "Someone") {
      try {
        const userRecord = await admin.auth().getUser(req.uid);
        senderName = userRecord.displayName || "Someone";
        const _photoURL = userRecord.photoURL || "";
        if (senderName !== "Someone") wsNames.set(req.uid, senderName);
        if (_photoURL) wsIcons.set(req.uid, _photoURL);
      } catch { /* non-fatal */ }
    }

    const { stored, ack } = processMessage({
      chatId           : chat_id,
      senderUid        : req.uid,
      senderName,
      senderIcon       : wsIcons.get(req.uid) || "",
      encryptedPayload : encrypted_payload,
      type             : type || "text",
      tempId           : (typeof temp_id === "string" && temp_id) ? temp_id : null,
      replyTo          : reply_to || null,
      senderWs         : null, // HTTP path — ACK merged into response below
    });

    res.json({ success: true, message_id: stored.id, ...ack });
  } catch (err) {
    console.error("[ERROR] /send_message:", err.message);
    res.status(500).json({ error: "Internal server error.", code: "internal" });
  }
});

const sendResult = (res, r) => r.ok
  ? res.json({ success: true })
  : res.status(r.status).json({ error: r.message, code: r.code });

// ── POST /edit_message ────────────────────────────────────────────────────────
app.post("/edit_message", authMiddleware, (req, res) => {
  const { message_id, new_encrypted_payload } = req.body || {};
  sendResult(res, doEdit(req.uid, message_id, new_encrypted_payload));
});

// ── POST /delete_message ──────────────────────────────────────────────────────
app.post("/delete_message", authMiddleware, (req, res) => {
  sendResult(res, doDelete(req.uid, (req.body || {}).message_id));
});

// ── POST /mark_seen ───────────────────────────────────────────────────────────
// v4: only participants can mark (and never their own messages) — previously
// ANY authenticated user could mark ANY message id as seen.
app.post("/mark_seen", authMiddleware, (req, res) => {
  const { message_ids } = req.body || {};
  if (!Array.isArray(message_ids) || message_ids.length === 0) {
    return res.status(400).json({ error: "message_ids must be a non-empty array.", code: "invalid" });
  }
  sendResult(res, doSeen(req.uid, message_ids.slice(0, 500)));
});

// ── POST /ack_delivered ── native/background wake path: "I have these" ───────
app.post("/ack_delivered", authMiddleware, (req, res) => {
  const { message_ids } = req.body || {};
  if (!Array.isArray(message_ids) || message_ids.length === 0) {
    return res.status(400).json({ error: "message_ids must be a non-empty array.", code: "invalid" });
  }
  sendResult(res, doDelivered(req.uid, message_ids.slice(0, 500)));
});

// ── GET /sync?since_rev=N&limit=M ─────────────────────────────────────────────
// Same payload as the WS `sync_batch` frame. Lets a push-woken native shell /
// service worker pull what it missed without opening a socket.
app.get("/sync", authMiddleware, (req, res) => {
  try {
    res.json(buildSync(req.uid, req.query.since_rev, req.query.limit));
  } catch (err) {
    console.error("[ERROR] /sync:", err.message);
    res.status(500).json({ error: "Internal server error.", code: "internal" });
  }
});

// ── POST /admin/backup — manually trigger a database backup ──────────────────
// Requires both x-titan-key and a valid Firebase ID Token (authMiddleware).
// For emergency use before a planned Factory Rebuild.
app.post("/admin/backup", authMiddleware, async (req, res) => {
  console.log(`[INFO] /admin/backup triggered by uid=${redact(req.uid)}`);
  try {
    await backupDatabase();
    res.json({ success: true, message: "Backup completed successfully." });
  } catch (err) {
    console.error("[ERROR] /admin/backup:", err.message);
    res.status(500).json({ error: err.message });
  }
});
// ─────────────────────────────────────────────────────────────────────────────
const httpServer = http.createServer(app);

// ─────────────────────────────────────────────────────────────────────────────
// 14. WebSocket server  (/ws)
// ─────────────────────────────────────────────────────────────────────────────
// maxPayload: ws defaults to 100 MB per frame — messages are small ciphertext (media goes via /relay).
const wss = new WebSocketServer({ server: httpServer, path: "/ws", maxPayload: 256 * 1024 });

wss.on("connection", (ws) => {
  console.log("[INFO] WS connected (pending auth)");

  let uid           = null;
  let displayName   = "Someone";
  let authenticated = false;
  let pingTimer     = null;
  let pongTimer     = null;

  // FIX: Auth deadline — if no auth frame arrives within 15 s, close the socket.
  // Prevents ghost connections that appear as "connected (pending auth)" indefinitely
  // and eliminates the repeated connect→disconnect(unauthenticated) loop caused by
  // clients that open a WS connection but fail to send a valid token in time.
  const authDeadline = setTimeout(() => {
    if (!authenticated) {
      console.log("[INFO] WS auth deadline exceeded — closing unauthenticated socket");
      ws.close();
    }
  }, 15_000);

  // ── Application-level heartbeat: ping every 30 s, 10 s pong deadline ──────
  function startHeartbeat() {
    pingTimer = setInterval(() => {
      if (ws.readyState !== WebSocket.OPEN) return;
      wsSend(ws, { type: "ping" });
      // FIX: increased pong deadline from 10 s → 25 s.
      // Mobile browsers throttle JS timers when screen is locked,
      // causing legitimate clients to miss the 10 s window → false termination.
      pongTimer = setTimeout(() => {
        console.log("[INFO] WS pong timeout — terminating", redact(uid));
        ws.terminate();
      }, 25_000);
    }, 30_000);
  }

  // Handle native WebSocket pong frames (sent by some clients/browsers)
  ws.on("pong", () => {
    if (pongTimer) { clearTimeout(pongTimer); pongTimer = null; }
  });

  // ── Incoming frame handler ─────────────────────────────────────────────────
  // Frames of ONE socket are processed strictly in order (an `await` in one
  // handler used to let the next frame overtake it).
  let _chain = Promise.resolve();
  const handleFrame = async (raw) => {
    let frame;
    try { frame = JSON.parse(raw); }
    catch {
      wsSend(ws, { type: "error", message: "Invalid JSON." });
      return;
    }

    // ── Auth frame (must arrive first) ───────────────────────────────────────
    if (!authenticated) {
      if (frame.type !== "auth") {
        wsSend(ws, { type: "auth_error", message: "Send an auth frame first." });
        ws.close();
        return;
      }

      // Layer A — Titan API Key
      if (!checkTitanKey(frame.x_titan_key)) {
        wsSend(ws, { type: "auth_error", message: "Invalid Titan key." });
        ws.close();
        return;
      }

      // Layer B — Firebase ID Token
      try {
        const decoded = await admin.auth().verifyIdToken(frame.firebase_token || "");
        uid = decoded.uid; // ALWAYS from the verified token — never from the frame body

        try {
          const userRecord  = await admin.auth().getUser(uid);
          displayName       = userRecord.displayName || "Someone";
          const photoURL    = userRecord.photoURL    || "";
          if (displayName !== "Someone") wsNames.set(uid, displayName);
          // ✅ Cache avatar URL so FCM can include it in notifications
          if (photoURL) wsIcons.set(uid, photoURL);
        } catch { /* non-fatal */ }

        authenticated = true;
        clearTimeout(authDeadline);   // FIX: cancel auth deadline — client authenticated in time
        wsRegister(uid, ws);
        wsSend(ws, { type: "auth_ok", server_rev: _rev });
        startHeartbeat();
        console.log("[INFO] WS authenticated:", redact(uid));

        // Optional: register FCM token sent inline with the auth frame.
        // This ensures the token is always current after every reconnect,
        // even when the REST /register_fcm_token call hasn't fired yet.
        if (frame.fcm_token && typeof frame.fcm_token === "string" && frame.fcm_token.trim()) {
          stmts.upsertFcmToken.run(uid, frame.fcm_token.trim());
          console.log("[INFO] FCM token registered (WS auth) for", redact(uid));
        }

        // Presence: tell all connected clients this user is online
        wsBroadcast({ type: "user_online", user_id: uid });
      } catch {
        wsSend(ws, { type: "auth_error", message: "Firebase token invalid or expired." });
        ws.close();
      }
      return;
    }

    // ── Application-level pong ────────────────────────────────────────────────
    if (frame.type === "pong") {
      if (pongTimer) { clearTimeout(pongTimer); pongTimer = null; }
      return;
    }

    // ── message_received — the device CONFIRMS it processed these messages.
    // This (and `seen`) is the only thing that sets delivered_at. Accepts one id
    // or a batch (used after applying a sync batch). ─────────────────────────
    if (frame.type === "message_received") {
      const ids = Array.isArray(frame.message_ids)
        ? frame.message_ids
        : [frame.message_id ?? (frame.data && frame.data.message_id)];
      doDelivered(uid, ids.slice(0, 500));
      return;
    }

    // ── sync_request — "give me everything since rev N" ─────────────────────
    if (frame.type === "sync_request") {
      const d = frame.data || {};
      try {
        wsSend(ws, { type: "sync_batch", ...buildSync(uid, d.since_rev, d.limit) });
      } catch (e) {
        console.error("[ERROR] sync_request:", e.message);
        wsSend(ws, { type: "error", code: "internal", message: "sync failed" });
      }
      return;
    }

    // ── update_fcm_token — propagate a rotated push token mid-session. ───────
    if (frame.type === "update_fcm_token") {
      const tok = (frame.data && frame.data.fcm_token) || frame.fcm_token;
      if (typeof tok === "string" && tok.trim()) {
        try { stmts.upsertFcmToken.run(uid, tok.trim()); } catch (e) { console.error("[ERROR] update_fcm_token:", e.message); }
      }
      return;
    }

    // ── send_message ──────────────────────────────────────────────────────────
    if (frame.type === "send_message") {
      const data    = frame.data || {};
      const { chat_id, encrypted_payload, type, reply_to } = data;
      const temp_id = typeof data.temp_id === "string" ? data.temp_id : null;

      const bad = validateSend(chat_id, encrypted_payload, type, temp_id);
      if (bad) { wsSend(ws, { type: "error", code: "invalid", temp_id, message: bad }); return; }

      if (!allowSend(uid)) {
        // Transient — the client keeps the message in its outbox and retries.
        wsSend(ws, { type: "error", code: "rate_limited", temp_id, retry_after_ms: 3000, message: "Too many messages — slow down." });
        return;
      }
      if (!(await isChatMemberAsync(chat_id, uid))) {
        wsSend(ws, { type: "error", code: "forbidden", temp_id, message: "Forbidden: not a member of this chat." });
        return;
      }

      try {
        processMessage({
          chatId           : chat_id,
          senderUid        : uid,
          senderName       : displayName,
          senderIcon       : wsIcons.get(uid) || "",
          encryptedPayload : encrypted_payload,
          type             : type || "text",
          tempId           : temp_id,
          replyTo          : reply_to || null,
          senderWs         : ws,
        });
      } catch (err) {
        console.error("[ERROR] WS send_message:", err.message);
        // `internal` = transient (disk busy, ...) → the client retries with the same temp_id.
        wsSend(ws, { type: "error", code: "internal", temp_id, message: "Failed to store message." });
      }
      return;
    }

    // ── mark_seen ─────────────────────────────────────────────────────────────
    if (frame.type === "mark_seen") {
      const data = frame.data || {};
      if (data.chat_id && !data.message_ids) {
        if (typeof data.chat_id !== "string" || !(await isChatMemberAsync(data.chat_id, uid))) {
          wsSend(ws, { type: "error", code: "forbidden", message: "Forbidden." });
          return;
        }
        try { doSeenChat(uid, data.chat_id); } catch (e) { console.error("[ERROR] mark_seen by chat_id:", e.message); }
        return;
      }
      if (!Array.isArray(data.message_ids) || data.message_ids.length === 0) {
        wsSend(ws, { type: "error", code: "invalid", message: "data.message_ids must be a non-empty array." });
        return;
      }
      try { doSeen(uid, data.message_ids.slice(0, 500)); } catch (e) { console.error("[ERROR] mark_seen:", e.message); }
      return;
    }

    // ── edit_message / delete_message / react_message ────────────────────────
    if (frame.type === "edit_message" || frame.type === "delete_message" || frame.type === "react_message") {
      const d = frame.data || {};
      let r;
      try {
        if (frame.type === "edit_message")        r = doEdit(uid, d.message_id, d.encrypted_payload);
        else if (frame.type === "delete_message") r = doDelete(uid, d.message_id);
        else                                      r = doReact(uid, d.message_id, d.reaction);
      } catch (e) {
        console.error(`[ERROR] ${frame.type}:`, e.message);
        r = fail(500, "internal", "Internal error.");
      }
      if (!r.ok) wsSend(ws, { type: "error", code: r.code, message: r.message });
      return;
    }

    // ── get_presence — query online status of a specific user ────────────────
    if (frame.type === "get_presence") {
      const targetUid = (frame.data || {}).user_id;
      if (!targetUid) {
        wsSend(ws, { type: "error", message: "data.user_id is required for get_presence." });
        return;
      }
      const isOnline = wsConnections.has(targetUid) && wsConnections.get(targetUid).size > 0;
      // FIX: Send flat frame — client reads frame.user_id at top-level
      wsSend(ws, {
        type      : isOnline ? "user_online" : "user_offline",
        user_id   : targetUid,
        requested : true,
      });
      return;
    }

    wsSend(ws, { type: "error", code: "invalid", message: `Unknown frame type: ${frame.type}` });
  };

  ws.on("message", (raw) => {
    _chain = _chain.then(() => handleFrame(raw))
                   .catch(err => console.error("[ERROR] WS frame handler:", err.message));
  });

  // ── Disconnect ─────────────────────────────────────────────────────────────
  ws.on("close", () => {
    clearTimeout(authDeadline);   // FIX: cleanup on any close
    if (pingTimer) { clearInterval(pingTimer); pingTimer = null; }
    if (pongTimer) { clearTimeout(pongTimer);  pongTimer = null; }

    if (uid) {
      wsUnregister(uid, ws);
      // Broadcast offline presence only when this was the user's last socket
      if (!wsConnections.has(uid)) {
        // FIX: Client reads frame.user_id directly (top-level), not frame.data.user_id
        wsBroadcast({ type: "user_offline", user_id: uid, last_seen: nowSec() });
      }
      console.log("[INFO] WS disconnected:", redact(uid));
    } else {
      console.log("[INFO] WS disconnected (unauthenticated)");
    }
  });

  ws.on("error", (err) => {
    console.error("[ERROR] WS socket error:", err.message);
  });
});

// ─────────────────────────────────────────────────────────────────────────────
// 15. Start listening
// ─────────────────────────────────────────────────────────────────────────────
httpServer.listen(PORT, "0.0.0.0", () => {
  console.log(`[INFO] Titan Messenger v4 listening on port ${PORT}`);
  console.log(
    `[INFO] Archive — DAYS=${ARCHIVE_DAYS}, ` +
    `INTERVAL=${ARCHIVE_INTERVAL_HRS}h, DAILY_HOUR=${ARCHIVE_DAILY_HOUR}UTC, ` +
    `EMERGENCY=${EMERGENCY_SIZE_MB}MB`
  );
});

// ─────────────────────────────────────────────────────────────────────────────
// 16. Graceful shutdown
// ─────────────────────────────────────────────────────────────────────────────
function shutdown(signal) {
  console.log(`[INFO] ${signal} received — shutting down gracefully.`);
  // 1001 "going away": clients reconnect and `sync_request` replays whatever they missed.
  for (const set of wsConnections.values()) {
    for (const ws of set) { try { ws.close(1001, "server restarting"); } catch {} }
  }
  httpServer.close(() => {
    try { db.pragma("wal_checkpoint(TRUNCATE)"); } catch {}
    try { db.close(); } catch {}
    console.log("[INFO] HTTP server closed, DB checkpointed and released.");
    process.exit(0);
  });
  // Force exit after 10 s if lingering connections prevent a clean close
  setTimeout(() => {
    console.error("[ERROR] Forced exit after shutdown timeout.");
    process.exit(1);
  }, 10_000).unref();
}

process.on("SIGTERM", () => shutdown("SIGTERM"));
process.on("SIGINT",  () => shutdown("SIGINT"));
process.on("unhandledRejection", (reason) => {
  console.error("[ERROR] Unhandled rejection:", reason && reason.message ? reason.message : reason);
});