fchaty/relay-server/src/index.js

const express = require('express');
const { WebSocketServer } = require('ws');
const { createServer } = require('http');
const { randomBytes, randomUUID } = require('crypto');
const multer = require('multer');
const fs = require('fs');
const path = require('path');

// ─── Config ───────────────────────────────────────────────────────────────────

const PORT = process.env.PORT || 3000;
const MAX_FILE_BYTES = parseInt(process.env.MAX_FILE_SIZE_MB ?? '25') * 1024 * 1024;
const UPLOADS_DIR = process.env.UPLOADS_DIR ?? '/data/files';
const QUEUE_DIR = process.env.QUEUE_DIR ?? '/data/queue';
const PAIRING_DIR = process.env.PAIRING_DIR ?? '/data/pairing';
const PAIRING_TTL_MS = 5 * 60 * 1000;
const QUEUE_TTL_MS = 7 * 24 * 60 * 60 * 1000;
const QUEUE_MAX = 500;
const FILE_TTL_DAYS = positiveInteger(process.env.FILE_TTL_DAYS, 30);
const FILE_CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
const ADMIN_TOKEN = process.env.ADMIN_TOKEN?.trim() ?? '';

fs.mkdirSync(UPLOADS_DIR, { recursive: true });
fs.mkdirSync(QUEUE_DIR, { recursive: true });
fs.mkdirSync(PAIRING_DIR, { recursive: true });

const app = express();
app.use(express.json());

// ─── File storage ─────────────────────────────────────────────────────────────

const storage = multer.diskStorage({
  destination: UPLOADS_DIR,
  filename: (_req, _file, cb) => cb(null, randomUUID()),
});
const upload = multer({ storage, limits: { fileSize: MAX_FILE_BYTES } });

// ─── In-memory state ──────────────────────────────────────────────────────────

const pairingSessions = new Map(); // code -> session
const connections = new Map();     // peerID -> { ws, name }
const fileRegistry = new Map();    // fileID -> { diskPath, originalName, size }

function positiveInteger(value, fallback) {
  const parsed = Number.parseInt(value ?? '', 10);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}

function cleanupFilesOlderThan(days = FILE_TTL_DAYS) {
  const cutoff = Date.now() - days * 24 * 60 * 60 * 1000;
  let deletedFiles = 0;
  let deletedBytes = 0;

  for (const entry of fs.readdirSync(UPLOADS_DIR, { withFileTypes: true })) {
    if (!entry.isFile()) continue;

    const diskPath = path.join(UPLOADS_DIR, entry.name);
    try {
      const stats = fs.statSync(diskPath);
      if (stats.mtimeMs >= cutoff) continue;

      fs.unlinkSync(diskPath);
      fileRegistry.delete(entry.name);
      deletedFiles += 1;
      deletedBytes += stats.size;
    } catch (error) {
      console.error(`[cleanup] failed to remove ${entry.name}: ${error.message}`);
    }
  }

  return { days, deletedFiles, deletedBytes };
}

function logCleanupResult(result, source) {
  if (result.deletedFiles > 0) {
    console.log(
      `[cleanup] ${source}: removed ${result.deletedFiles} file(s), ${result.deletedBytes} byte(s), older than ${result.days} day(s)`
    );
  }
}

logCleanupResult(cleanupFilesOlderThan(), 'startup');
setInterval(() => {
  logCleanupResult(cleanupFilesOlderThan(), 'scheduled');
}, FILE_CLEANUP_INTERVAL_MS);

// ─── Persistent queue (disk-backed) ──────────────────────────────────────────
//
// Layout on disk:
//   /data/queue/{peerID}/{messageID}.json
//
// A message is written to disk the moment it is queued.
// It is deleted from disk the moment it is delivered.
// On server startup all existing files are loaded back into memory.

function queueDir(peerID) {
  return path.join(QUEUE_DIR, peerID);
}

function queuePath(peerID, messageID) {
  return path.join(queueDir(peerID), `${messageID}.json`);
}

function persistMessage(peerID, messageID, envelope) {
  fs.mkdirSync(queueDir(peerID), { recursive: true });
  fs.writeFileSync(
    queuePath(peerID, messageID),
    JSON.stringify({ envelope, queuedAt: Date.now() })
  );
}

function deletePersistedMessage(peerID, messageID) {
  try { fs.unlinkSync(queuePath(peerID, messageID)); } catch (_) {}
}

function loadQueueFromDisk() {
  const queues = new Map();
  if (!fs.existsSync(QUEUE_DIR)) return queues;

  for (const peerID of fs.readdirSync(QUEUE_DIR)) {
    const dir = queueDir(peerID);
    if (!fs.statSync(dir).isDirectory()) continue;
    const entries = [];
    for (const file of fs.readdirSync(dir)) {
      if (!file.endsWith('.json')) continue;
      try {
        const raw = fs.readFileSync(path.join(dir, file), 'utf8');
        const { envelope, queuedAt } = JSON.parse(raw);
        // Drop messages older than TTL
        if (Date.now() - queuedAt > QUEUE_TTL_MS) {
          fs.unlinkSync(path.join(dir, file));
          continue;
        }
        const messageID = file.replace('.json', '');
        entries.push({ messageID, envelope, queuedAt });
      } catch (_) {}
    }
    if (entries.length > 0) {
      entries.sort((a, b) => a.queuedAt - b.queuedAt);
      queues.set(peerID, entries);
    }
  }
  return queues;
}

// In-memory queue mirrors disk — both are always in sync
const offlineQueues = loadQueueFromDisk();
console.log(`[queue] loaded ${[...offlineQueues.values()].reduce((s, q) => s + q.length, 0)} queued messages from disk`);

function enqueue(peerID, messageID, envelope) {
  if (!offlineQueues.has(peerID)) offlineQueues.set(peerID, []);
  const queue = offlineQueues.get(peerID);
  if (queue.length >= QUEUE_MAX) {
    const dropped = queue.shift();
    deletePersistedMessage(peerID, dropped.messageID);
  }
  queue.push({ messageID, envelope, queuedAt: Date.now() });
  persistMessage(peerID, messageID, envelope);
}

function flushQueue(peerID, ws) {
  const queue = offlineQueues.get(peerID);
  if (!queue || queue.length === 0) return;
  const now = Date.now();
  for (const { messageID, envelope, queuedAt } of queue) {
    if (now - queuedAt < QUEUE_TTL_MS) send(ws, envelope);
    deletePersistedMessage(peerID, messageID);
  }
  offlineQueues.delete(peerID);
}

// ─── Persistent pairing sessions (disk-backed) ────────────────────────────────
//
// Pairing sessions are written to disk so tokens survive a server restart.
// Without this, everyone would need to re-pair after every server update.

function pairingPath(code) {
  return path.join(PAIRING_DIR, `${code}.json`);
}

function persistSession(session) {
  fs.writeFileSync(pairingPath(session.code), JSON.stringify(session));
}

function deleteSession(code) {
  try { fs.unlinkSync(pairingPath(code)); } catch (_) {}
}

function loadSessionsFromDisk() {
  if (!fs.existsSync(PAIRING_DIR)) return;
  const now = Date.now();
  for (const file of fs.readdirSync(PAIRING_DIR)) {
    if (!file.endsWith('.json')) continue;
    try {
      const session = JSON.parse(fs.readFileSync(path.join(PAIRING_DIR, file), 'utf8'));
      if (session.expiresAt && session.expiresAt < now) {
        fs.unlinkSync(path.join(PAIRING_DIR, file));
        continue;
      }
      // Permanent sessions (joined pairs) have no expiry — keep them forever
      pairingSessions.set(session.code, session);
    } catch (_) {}
  }
}

loadSessionsFromDisk();
console.log(`[pairing] loaded ${pairingSessions.size} sessions from disk`);

// ─── Helpers ──────────────────────────────────────────────────────────────────

function generateCode() {
  const alphabet = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789';
  const bytes = randomBytes(6);
  let suffix = '';
  for (let i = 0; i < 6; i++) suffix += alphabet[bytes[i] % alphabet.length];
  return 'FCHT-' + suffix;
}

function generateToken() {
  return randomBytes(32).toString('hex');
}

function sessionByToken(token) {
  for (const s of pairingSessions.values()) {
    if (s.creatorToken === token || s.joinerToken === token) return s;
  }
  return null;
}

function peerIDFromToken(session, token) {
  return session.creatorToken === token ? session.creatorID : session.joinerID;
}

function peerPartnerID(session, peerID) {
  return session.creatorID === peerID ? session.joinerID : session.creatorID;
}

function send(ws, obj) {
  if (ws.readyState === 1) ws.send(JSON.stringify(obj));
}

function cleanExpiredSessions() {
  const now = Date.now();
  for (const [code, s] of pairingSessions) {
    // Only remove pending (not yet joined) sessions that expired
    if (!s.joinerID && s.expiresAt < now) {
      pairingSessions.delete(code);
      deleteSession(code);
    }
  }
}

setInterval(cleanExpiredSessions, 60_000);

// ─── Auth middleware (HTTP) ───────────────────────────────────────────────────

function requireToken(req, res, next) {
  const auth = req.headers.authorization ?? '';
  if (!auth.startsWith('Bearer ')) return res.status(401).json({ error: 'Unauthorized' });
  const token = auth.slice(7);
  const session = sessionByToken(token);
  if (!session) return res.status(401).json({ error: 'Invalid token' });
  req.peerID = peerIDFromToken(session, token);
  req.session = session;
  next();
}

function requireAdmin(req, res, next) {
  if (!ADMIN_TOKEN) {
    return res.status(503).json({ error: 'Admin cleanup is not configured' });
  }

  const auth = req.headers.authorization ?? '';
  if (auth !== `Bearer ${ADMIN_TOKEN}`) {
    return res.status(401).json({ error: 'Unauthorized' });
  }

  next();
}

// ─── Pairing routes ───────────────────────────────────────────────────────────

app.post('/pairing/create', (req, res) => {
  const name = req.body?.displayName?.trim();
  if (!name) return res.status(400).json({ error: 'displayName is required' });

  let code, tries = 0;
  do {
    code = generateCode();
    if (++tries > 200) return res.status(503).json({ error: 'Please try again' });
  } while (pairingSessions.has(code));

  const session = {
    code,
    creatorID: randomUUID(),
    creatorName: name,
    creatorToken: generateToken(),
    joinerID: null,
    joinerName: null,
    joinerToken: null,
    expiresAt: Date.now() + PAIRING_TTL_MS,
  };
  pairingSessions.set(code, session);
  persistSession(session);

  res.json({
    code,
    token: session.creatorToken,
    peerID: session.creatorID,
    expiresAt: new Date(session.expiresAt).toISOString(),
  });
});

app.post('/pairing/join', (req, res) => {
  const code = req.body?.code?.trim().toUpperCase();
  const name = req.body?.displayName?.trim();
  if (!code || !name) return res.status(400).json({ error: 'code and displayName are required' });

  const session = pairingSessions.get(code);
  if (!session || (!session.joinerID && Date.now() > session.expiresAt)) {
    pairingSessions.delete(code);
    deleteSession(code);
    return res.status(404).json({ error: 'Code not found or expired' });
  }
  if (session.joinerID) return res.status(409).json({ error: 'Code already used' });

  session.joinerID = randomUUID();
  session.joinerName = name;
  session.joinerToken = generateToken();
  delete session.expiresAt; // paired sessions never expire
  persistSession(session);

  res.json({
    token: session.joinerToken,
    peerID: session.joinerID,
    peer: { id: session.creatorID, displayName: session.creatorName },
  });
});

// ─── File routes ──────────────────────────────────────────────────────────────

app.post('/files', requireToken, upload.single('file'), (req, res) => {
  if (!req.file) return res.status(400).json({ error: 'No file provided' });
  const fileID = req.file.filename;
  fileRegistry.set(fileID, {
    diskPath: req.file.path,
    originalName: req.file.originalname,
    size: req.file.size,
  });
  res.json({ id: fileID, name: req.file.originalname, size: req.file.size });
});

app.get('/files/:id', requireToken, (req, res) => {
  const meta = fileRegistry.get(req.params.id);
  if (!meta) return res.status(404).json({ error: 'File not found' });
  res.setHeader('Content-Disposition', `attachment; filename="${encodeURIComponent(meta.originalName)}"`);
  res.sendFile(meta.diskPath);
});

// ─── Admin routes ───────────────────────────────────────────────────────────

app.post('/admin/cleanup', requireAdmin, (req, res) => {
  const requestedDays = req.body?.days;
  const days = requestedDays === undefined ? FILE_TTL_DAYS : Number(requestedDays);

  if (!Number.isInteger(days) || days < 1 || days > 3650) {
    return res.status(400).json({ error: 'days must be an integer between 1 and 3650' });
  }

  const result = cleanupFilesOlderThan(days);
  logCleanupResult(result, 'manual');
  res.json(result);
});

// ─── Health check ─────────────────────────────────────────────────────────────

app.get('/health', (_req, res) => res.json({
  ok: true,
  connections: connections.size,
  queued: [...offlineQueues.values()].reduce((sum, q) => sum + q.length, 0),
}));

// ─── WebSocket ────────────────────────────────────────────────────────────────

const server = createServer(app);
const wss = new WebSocketServer({ server, path: '/ws' });

wss.on('connection', (ws) => {
  let peerID = null;
  let peerName = null;
  let session = null;
  let authed = false;

  const authTimeout = setTimeout(() => {
    if (!authed) ws.close(1008, 'Auth timeout');
  }, 10_000);

  ws.on('message', (raw) => {
    let msg;
    try { msg = JSON.parse(raw.toString()); }
    catch { return ws.close(1003, 'Invalid JSON'); }

    // ─── Auth ─────────────────────────────────────────────────────────────────
    if (!authed) {
      if (msg.type !== 'auth' || !msg.token) {
        return ws.close(1008, 'Send { type: "auth", token: "..." } first');
      }
      session = sessionByToken(msg.token);
      if (!session) return ws.close(1008, 'Invalid token');

      clearTimeout(authTimeout);
      authed = true;
      peerID = peerIDFromToken(session, msg.token);
      peerName = session.creatorToken === msg.token ? session.creatorName : session.joinerName;
      connections.set(peerID, { ws, name: peerName });

      send(ws, { type: 'auth.ok', peerID });
      console.log(`[ws] connected: ${peerName} (${peerID})`);

      // Notify the partner that this peer is now online
      const partnerIDOnAuth = peerPartnerID(session, peerID);
      const partnerConnOnAuth = connections.get(partnerIDOnAuth);
      if (partnerConnOnAuth?.ws.readyState === 1) {
        send(partnerConnOnAuth.ws, { type: 'peer.joined', peerID, peerName });
      }

      // Deliver messages that arrived while this peer was offline
      flushQueue(peerID, ws);
      return;
    }

    // ─── Chat message ─────────────────────────────────────────────────────────
    if (msg.type === 'chat.message') {
      const partnerID = peerPartnerID(session, peerID);
      const messageID = msg.id ?? randomUUID();
      const outgoing = { ...msg, id: messageID, from: peerID, fromName: peerName };
      const target = connections.get(partnerID);

      if (target?.ws.readyState === 1) {
        send(target.ws, outgoing);
        send(ws, { type: 'delivered', id: messageID });
      } else {
        enqueue(partnerID, messageID, outgoing);
        send(ws, { type: 'queued', id: messageID });
      }
      return;
    }

    // ─── Read receipt ─────────────────────────────────────────────────────────
    // Client sends: { type: "read", messageIDs: ["id1", "id2", ...] }
    // Server forwards to the sender of those messages so they see "read" ticks
    if (msg.type === 'read') {
      const partnerID = peerPartnerID(session, peerID);
      const receipt = { type: 'read', messageIDs: msg.messageIDs, by: peerID };
      const target = connections.get(partnerID);

      if (target?.ws.readyState === 1) {
        send(target.ws, receipt);
      } else {
        // Queue the read receipt too — partner deserves to know
        enqueue(partnerID, `read-${randomUUID()}`, receipt);
      }
      return;
    }

    // ─── Ping ─────────────────────────────────────────────────────────────────
    if (msg.type === 'ping') {
      send(ws, { type: 'pong' });
    }
  });

  ws.on('close', () => {
    if (peerID) {
      connections.delete(peerID);
      console.log(`[ws] disconnected: ${peerName} (${peerID})`);
    }
    clearTimeout(authTimeout);
  });

  ws.on('error', (err) => {
    console.error('[ws] error:', err.message);
  });
});

// ─── Start ────────────────────────────────────────────────────────────────────

server.listen(PORT, () => {
  console.log(`Fchati Relay listening on port ${PORT}`);
});