Hugging Face's logo

Spaces:
M-hv1
/
ghost-link
Paused

App Files Community
ghost-link / server.js
M-hv1's picture
M-hv1
Upload server.js
5af5bfd verified
Raw History Blame Contribute Delete
92.5 kB
/**
* 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);
});