gigafibre-fsm/services/targo-hub/lib/sync-orchestrator.js

'use strict'
// ─────────────────────────────────────────────────────────────────────────────
// ORCHESTRATEUR DE SYNCHRONISATION F → ERPNext — colonne vertébrale unifiée.
//
// Pattern : moteurs MODULAIRES (un par domaine) + orchestration + vue UNIFIÉES.
//   (1) MANIFEST du DAG de dépendances → garantit l'ordre (un module ne tourne jamais
//       avant ses prérequis : c'est la leçon « comptes avant factures » généralisée).
//   (2) status() → agrège l'état LÉGER de chaque module (compteurs PG + MAX(id) F +
//       coherenceStatus billing), mis en cache 60s. Base du tableau de bord unifié.
//
// L'EXÉCUTION ordonnée vit dans /opt/targo-sync/run.sh (l'hôte a docker pour le step
// Python des factures). Le DAG ci-dessous DOCUMENTE et VALIDE cet ordre.
// Règle de statut client : disabled/is_suspended = DÉRIVÉ des services, jamais sync
// depuis F (cf. feedback_customer_status_policy) → classés « dérivé » côté dashboard.
// ─────────────────────────────────────────────────────────────────────────────
const { json, log, parseBody } = require('./helpers')

// DAG : chaque module déclare ses prérequis. Ordre topologique garanti par l'orchestrateur.
// Dépendances VÉRIFIÉES contre les colonnes *_id de F (aucune FK déclarée) :
//   delivery.account_id→account · service.{delivery_id,device_id,product_id} · invoice.account_id
//   invoice_item.{invoice_id,service_id} · payment.account_id · payment_item.{payment_id,invoice_id}
//   ticket.{account_id,delivery_id} · account_memo.account_id
const MODULES = [
  { key: 'clients', label: 'Clients', deps: [], engine: 'legacy-sync', creator: 'createCustomers' },
  { key: 'adresses', label: 'Adresses', deps: ['clients'], engine: 'legacy-sync', note: 'Service Location (delivery.account_id)' },
  { key: 'services', label: 'Services', deps: ['clients', 'adresses'], engine: 'legacy-sync', note: 'Service Subscription (service.delivery_id)' },
  { key: 'tickets', label: 'Tickets', deps: ['clients', 'adresses'], engine: 'legacy-dispatch-sync', note: 'ticket.{account_id,delivery_id} — cadence propre' },
  { key: 'factures', label: 'Factures', deps: ['clients', 'adresses'], engine: 'sync_invoices_incremental.py', note: 'invoice.account_id ; items→service_location' },
  { key: 'paiements', label: 'Paiements', deps: ['factures'], engine: 'legacy-payments', note: 'payment_item.invoice_id' },
]

// Ordre topologique (Kahn) — sert à la fois de doc et de garde-fou (détecte un cycle).
function topoOrder () {
  const done = new Set(); const order = []
  let guard = 0
  while (order.length < MODULES.length && guard++ < 50) {
    for (const m of MODULES) { if (!done.has(m.key) && m.deps.every(d => done.has(d))) { order.push(m.key); done.add(m.key) } }
  }
  return order
}

let _cache = null; let _cacheAt = 0
async function status ({ fresh = false } = {}) {
  if (!fresh && _cache && (Date.now() - _cacheAt) < 60000) return { ..._cache, from_cache: true }
  const lp = require('./legacy-payments')
  const ls = require('./legacy-sync')

  let dc = null; try { dc = await ls.domainCounts() } catch (e) { dc = null } // set-diff EXACT (à créer + orphelins)
  let coh = {}; try { coh = await lp.coherenceStatus() } catch (e) { coh = { error: e.message } }

  const mods = {}
  const dom = (k, extra) => (dc && dc[k]) ? { erp: dc[k].erp, f: dc[k].f, to_create: dc[k].missing, orphans: dc[k].orphans, ...(extra || {}) } : { error: 'F indisponible' }
  mods.clients = dom('clients', { status_note: 'disabled/is_suspended = dérivé (non synchronisé)' })
  mods.adresses = dom('adresses')
  mods.services = dom('services', { note: 'orphans = SS dont le service F est supprimé (≠ annulé)' })
  mods.factures = coh.invoices ? { lag: coh.invoices.lag, synced: coh.invoices.live_synced, ar_open: coh.invoices.ar_open } : { error: coh.error }
  mods.paiements = coh.payments ? { lag: coh.payments.lag, total: coh.payments.total } : { error: coh.error }
  mods.tickets = { engine: 'legacy-dispatch-sync', note: 'moteur séparé (osTicket → Dispatch Job)' }

  const out = {
    checked_at: new Date().toISOString(),
    order: topoOrder(),
    dag: MODULES.map(m => ({ key: m.key, label: m.label, deps: m.deps, engine: m.engine, note: m.note })),
    modules: mods,
    billing_health: { healthy: coh.healthy, last_cycle: coh.last_cycle, last_cycle_min_ago: coh.last_cycle_min_ago },
    note: 'Moteurs modulaires, orchestration + vue unifiées. Exécution ordonnée : /opt/targo-sync/run.sh (horaire).',
  }
  _cache = out; _cacheAt = Date.now()
  return out
}

// ── SYNC MANUEL (« Synchroniser maintenant ») — étapes HUB-SIDE, progression pollable ──
// Couvre les couches qui changent en continu (clients/paiements/soldes/tickets), in-process.
// Les FACTURES + SERVICES sont créés par le cron horaire (scripts Python sur l'hôte) → hors de ce bouton ;
// l'UI affiche leur lag + le prochain passage auto.
// Cœur (rapide, ~30s) : couches qui changent en continu.
const CORE_STEPS = [
  { key: 'clients', label: 'Comptes manquants', fn: async () => { const r = await require('./legacy-payments').ensureBillingCustomers({ confirm: 'F-WINS' }); return `${r.created || 0} créé(s)` } },
  { key: 'paiements', label: 'Paiements', fn: async () => { const r = await require('./legacy-payments').syncPayments({ confirm: 'F-WINS', window: 5000, limit: 20000 }); return `${r.applied || 0} paiement(s), ${r.references || 0} alloc.` } },
  { key: 'soldes', label: 'Soldes des factures', fn: async () => { const r = await require('./legacy-payments').refreshOpenInvoices({ confirm: 'F-WINS' }); return `${r.updated || 0} soldé(s)` } },
]
// Tickets : SÉPARÉ (lourd ~70s : pull legacy + géocodage ; déjà sur scheduler 15 min + sync d'assignation live en Planification).
const TICKET_STEP = { key: 'tickets', label: 'Tickets (dispatch)', fn: async () => { const lds = require('./legacy-dispatch-sync'); const r = await lds.sync({ dryRun: false }); const n = (r && (r.created ?? r.ingested ?? r.jobs_created)) || 0; let a = null; try { a = await lds.ingestAssigned({ dryRun: false, allDates: true }) } catch (e) { /* pool créé quand même */ }; const na = (a && a.created) || 0; return `${n + na} ticket(s)${na ? ` (dont ${na} assigné(s))` : ''}` } }
// Créer TOUS les comptes F manquants (pas seulement ceux référencés par une facture récente comme le step `clients`).
// Lourd (scan complet des comptes F + création throttlée) → passe par le run ASYNC (polling), jamais une requête HTTP bloquante.
const CREATE_ALL_STEP = { key: 'clients-all', label: 'Tous les comptes manquants (F→OPS)', fn: async () => { const r = await require('./legacy-sync').createCustomers({ confirm: 'F-WINS', limit: 100000 }); return `${r.created || 0} créé(s)${r.errors ? ` · ${r.errors} err` : ''}` } }
function stepsFor (scope) { return scope === 'tickets' ? [TICKET_STEP] : scope === 'customers' ? [CREATE_ALL_STEP] : scope === 'all' ? [...CORE_STEPS, TICKET_STEP] : CORE_STEPS }

let _run = { running: false, scope: null, steps: [], started_at: null, finished_at: null }
async function runManual ({ confirm, scope = 'core' }) {
  if (confirm !== 'F-WINS') return { error: "confirm:'F-WINS' requis" }
  if (_run.running) return { already_running: true }
  const defs = stepsFor(scope)
  _run = { running: true, scope, started_at: new Date().toISOString(), finished_at: null, steps: defs.map(s => ({ key: s.key, label: s.label, status: 'pending', detail: null })) }
  ;(async () => {
    for (let i = 0; i < defs.length; i++) {
      _run.steps[i].status = 'running'
      try { _run.steps[i].detail = await defs[i].fn(); _run.steps[i].status = 'done' }
      catch (e) { _run.steps[i].status = 'error'; _run.steps[i].detail = String(e.message).slice(0, 160); log('sync run step ' + defs[i].key + ' err: ' + e.message) }
    }
    _run.running = false; _run.finished_at = new Date().toISOString(); _cacheAt = 0 // force /sync/status frais
  })()
  return { started: true, scope, steps: _run.steps.length }
}
function runStatus () {
  const done = _run.steps.filter(s => s.status === 'done' || s.status === 'error').length
  const ms = _run.started_at ? ((_run.finished_at ? Date.parse(_run.finished_at) : Date.now()) - Date.parse(_run.started_at)) : 0
  return { ..._run, total: _run.steps.length, done, elapsed_s: Math.round(ms / 1000), note: 'Factures & services : créés au cycle horaire (cron :15).' }
}

async function handle (req, res, method, p, url) {
  if (p === '/sync/run' && method === 'POST') {
    try { const b = await parseBody(req); return json(res, 200, await runManual({ confirm: b.confirm, scope: b.scope })) } catch (e) { return json(res, 500, { error: e.message }) }
  }
  if (p === '/sync/run-status' && method === 'GET') return json(res, 200, runStatus())
  if (p === '/sync/status' && method === 'GET') {
    try { const fresh = url && url.searchParams.get('fresh') === '1'; return json(res, 200, await status({ fresh })) }
    catch (e) { log('sync-orchestrator status error:', e.message); return json(res, 500, { error: e.message }) }
  }
  if (p === '/sync/dag' && method === 'GET') return json(res, 200, { order: topoOrder(), dag: MODULES })
  return json(res, 404, { error: 'route sync inconnue' })
}

module.exports = { handle, status, topoOrder, MODULES }