gigafibre-fsm/services/targo-hub/lib/outbox.js

'use strict'
// ── Tampon d'envoi sortant (« undo send ») pour les courriels MULTI-DESTINATAIRES (notifs de file) ──────────
// But : avant qu'un fan-out parte, le retenir N secondes, l'AFFICHER (destinataires + sujet + compte à rebours)
// dans un panneau OPS, et permettre d'ANNULER. Filet anti-spam (voir [[feedback_inbox_queue_notify]]).
// Les réponses agent à 1 destinataire ne passent PAS ici (envoi immédiat, aucun risque). HOLD=0 → envoi direct.
const fs = require('fs')
const { log } = require('./helpers')
const sse = require('./sse')

const STORE = '/app/data/outbox.json'
const HOLD_MS = Math.max(0, (Number(process.env.OUTBOX_HOLD_SEC) || 20) * 1000)
const pending = new Map() // id -> { id, to, subject, sendAt, opts, meta, status, error? }
const timers = new Map()
let seq = 0

function persist () { try { fs.writeFileSync(STORE + '.tmp', JSON.stringify([...pending.values()])); fs.renameSync(STORE + '.tmp', STORE) } catch (e) { log('outbox persist: ' + e.message) } }
function recipientsOf (to) { return String(to || '').split(',').map(s => s.trim()).filter(Boolean) }
function view (e) { return { id: e.id, subject: e.subject, recipients: recipientsOf(e.to), count: recipientsOf(e.to).length, sendAt: e.sendAt, status: e.status || 'pending', error: e.error || null, kind: (e.meta && e.meta.kind) || 'email', label: (e.meta && e.meta.label) || '' } }
function list () { return [...pending.values()].sort((a, b) => a.sendAt - b.sendAt).map(view) }
function broadcast () { try { sse.broadcast('outbox', 'outbox', { pending: list(), holdSec: HOLD_MS / 1000 }) } catch (e) { /* */ } }

async function doSend (e) {
  const gmail = require('./gmail')
  const r = await gmail.sendMessage(e.opts)
  if (e.meta && e.meta.convToken) { try { require('./conversation').onOutboxSent(e.meta.convToken, r, e.meta) } catch (x) { /* */ } }
  return r
}
async function flush (id) {
  const e = pending.get(id); if (!e || e.status === 'sending') return
  e.status = 'sending'; clearTimeout(timers.get(id)); timers.delete(id)
  try {
    await doSend(e)
    pending.delete(id); persist(); broadcast()
    log('outbox SENT ' + id + ' → ' + e.to + ' [' + ((e.meta && e.meta.kind) || 'email') + ']')
  } catch (err) {
    e.status = 'failed'; e.error = err.message; persist(); broadcast() // reste visible (rouge) → Réessayer / Rejeter
    log('outbox FAILED ' + id + ': ' + err.message)
  }
}
function arm (e) { const t = setTimeout(() => flush(e.id).catch(() => {}), Math.max(0, e.sendAt - Date.now())); timers.set(e.id, t) }

// Met en file un envoi retenu. Renvoie une promesse qui résout TÔT (l'envoi réel se fait au flush).
function enqueue (opts, meta = {}) {
  if (!opts || !opts.to) return Promise.resolve({ ok: false, error: 'destinataire requis' })
  if (HOLD_MS === 0) return doSend({ opts, meta }).then(r => ({ ok: true, id: r && r.id, sent: true })).catch(e => ({ ok: false, error: e.message }))
  const id = 'ob_' + Date.now().toString(36) + '_' + (++seq)
  const e = { id, to: opts.to, subject: opts.subject || '', sendAt: Date.now() + HOLD_MS, opts, meta, status: 'pending' }
  pending.set(id, e); persist(); arm(e); broadcast()
  log('outbox HELD ' + id + ' → ' + recipientsOf(opts.to).length + ' dest. [' + (meta.kind || 'email') + '] « ' + e.subject.slice(0, 50) + ' »')
  return Promise.resolve({ ok: true, queued: true, id, sendAt: e.sendAt })
}
function cancel (id) { const e = pending.get(id); if (!e) return { ok: false, error: 'introuvable' }; clearTimeout(timers.get(id)); timers.delete(id); pending.delete(id); persist(); broadcast(); log('outbox CANCEL ' + id); return { ok: true, cancelled: id } }
function cancelAll () { const ids = [...pending.keys()]; for (const id of ids) { clearTimeout(timers.get(id)); timers.delete(id) } pending.clear(); persist(); broadcast(); log('outbox CANCEL-ALL (' + ids.length + ')'); return { ok: true, cancelled: ids.length } }
async function sendNow (id) { const e = pending.get(id); if (!e) return { ok: false, error: 'introuvable' }; if (e.status === 'failed') e.status = 'pending'; e.sendAt = Date.now(); await flush(id); const still = pending.get(id); return { ok: !still || still.status !== 'failed', id, error: still && still.error } }

function startup () {
  try { for (const e of JSON.parse(fs.readFileSync(STORE, 'utf8'))) pending.set(e.id, e) } catch (e) { return }
  let armed = 0
  for (const e of pending.values()) { if (e.status === 'failed') continue; arm(e); armed++ } // les en attente repartent (sendAt passé → flush quasi immédiat)
  if (pending.size) log('outbox: ' + pending.size + ' en file rechargés (' + armed + ' réarmés)')
  broadcast()
}

async function handle (req, res, method, p) {
  const json = (c, o) => { res.writeHead(c, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(o)) }
  const body = () => new Promise(r => { let d = ''; req.on('data', c => { d += c }); req.on('end', () => { try { r(JSON.parse(d || '{}')) } catch (e) { r({}) } }); req.on('error', () => r({})) })
  if (p === '/outbox' && method === 'GET') return json(200, { pending: list(), holdSec: HOLD_MS / 1000 })
  if (p === '/outbox/cancel' && method === 'POST') { const b = await body(); return json(200, cancel(b.id)) }
  if (p === '/outbox/cancel-all' && method === 'POST') return json(200, cancelAll())
  if (p === '/outbox/send-now' && method === 'POST') { const b = await body(); return json(200, await sendNow(b.id)) }
  return json(404, { error: 'not found' })
}

module.exports = { enqueue, cancel, cancelAll, sendNow, list, startup, handle }