gigafibre-fsm/services/targo-hub/lib/store-pg.js

'use strict'
/**
 * store-pg.js — magasin Postgres DURABLE et NORMALISÉ du système de communication omnichannel (comme les tickets).
 *
 * Phase 2 (normalisé) :
 *   comms.conversations : 1 ligne / conversation = ENTÊTE (colonnes indexées : token, thread_id, email,
 *     customer, queue, status, last_activity) + `meta` jsonb (= la conv SANS ses messages : readBy, draft,
 *     notes, triage, assistants, ticketState, linkedTickets…). Lossless.
 *   comms.messages : 1 ligne / message (colonnes indexées : conv_token, gmail_id, ts, sender…) + `data` jsonb
 *     (= le message complet, lossless). PK (conv_token, id).
 *
 * Postgres MAÎTRE : le hub charge la Map depuis Postgres au boot (cf. conversation.js). Le JSON reste écrit
 * en double comme sauvegarde / rollback. Migration auto depuis l'ancien schéma Phase 1 (colonne `data` sur
 * conversations) : on supprime l'ancienne table (les données sont réamorcées depuis la Map/JSON).
 *
 * Réutilise la connexion Postgres ERPNext (cf. address-db.js). Tout est BEST-EFFORT (try/catch) → Postgres
 * down ⇒ `available()` false ⇒ le hub continue sur le JSON. S'inspire des stores Chatwoot/FreeScout.
 */
const { Pool } = require('pg')
const { log } = require('./helpers')

let _pool = null
let _available = false

function pool () {
  if (!_pool) {
    _pool = new Pool({
      host: process.env.COMMS_DB_HOST || process.env.ADDR_DB_HOST || 'erpnext-db-1',
      port: +(process.env.COMMS_DB_PORT || process.env.ADDR_DB_PORT || 5432),
      user: process.env.COMMS_DB_USER || process.env.ADDR_DB_USER || 'postgres',
      password: process.env.COMMS_DB_PASS || process.env.ADDR_DB_PASS || '123',
      database: process.env.COMMS_DB_NAME || process.env.ADDR_DB_NAME || '_eb65bdc0c4b1b2d6',
      max: 5, idleTimeoutMillis: 30000, connectionTimeoutMillis: 6000,
    })
    _pool.on('error', (e) => log('comms-pg pool error:', e.message))
  }
  return _pool
}

const DDL_CONV = `CREATE TABLE IF NOT EXISTS comms.conversations (
  token text PRIMARY KEY, channel text, phone text, email text, customer text, customer_name text,
  thread_id text, subject text, queue text, assignee text, status text,
  last_activity timestamptz, created_at timestamptz, expires_at timestamptz, noise boolean DEFAULT false,
  meta jsonb NOT NULL, updated_at timestamptz DEFAULT now())`
const DDL_MSG = `CREATE TABLE IF NOT EXISTS comms.messages (
  conv_token text NOT NULL, id text NOT NULL, ord int,
  sender text, from_name text, from_email text, to_email text, via text, agent text, gmail_id text,
  ts timestamptz, email_date timestamptz, data jsonb NOT NULL, updated_at timestamptz DEFAULT now(),
  PRIMARY KEY (conv_token, id))`
const DDL_IDX = [
  `CREATE INDEX IF NOT EXISTS conv_thread_idx   ON comms.conversations (thread_id)`,
  `CREATE INDEX IF NOT EXISTS conv_email_idx    ON comms.conversations (email)`,
  `CREATE INDEX IF NOT EXISTS conv_customer_idx ON comms.conversations (customer)`,
  `CREATE INDEX IF NOT EXISTS conv_queue_idx    ON comms.conversations (queue)`,
  `CREATE INDEX IF NOT EXISTS conv_status_idx   ON comms.conversations (status)`,
  `CREATE INDEX IF NOT EXISTS conv_lastact_idx  ON comms.conversations (last_activity DESC)`,
  `CREATE INDEX IF NOT EXISTS msg_conv_idx   ON comms.messages (conv_token)`,
  `CREATE INDEX IF NOT EXISTS msg_gmail_idx  ON comms.messages (gmail_id)`,
  `CREATE INDEX IF NOT EXISTS msg_ts_idx     ON comms.messages (ts)`,
]

async function init () {
  try {
    await pool().query('CREATE SCHEMA IF NOT EXISTS comms')
    // Migration auto Phase 1 → Phase 2 : l'ancienne table avait une colonne `data` (conv entière). On la
    // supprime (les données reviennent via le réamorçage depuis la Map/JSON) puis on crée le schéma normalisé.
    const cols = await pool().query(`SELECT column_name FROM information_schema.columns WHERE table_schema='comms' AND table_name='conversations'`)
    const names = cols.rows.map(r => r.column_name)
    if (names.includes('data') && !names.includes('meta')) {
      await pool().query('DROP TABLE IF EXISTS comms.conversations')
      log('comms-pg: migration Phase 1→2 — ancienne table conversations supprimée (réamorçage depuis JSON)')
    }
    await pool().query(DDL_CONV)
    await pool().query(DDL_MSG)
    for (const q of DDL_IDX) await pool().query(q)
    // Recherche FLOUE (typo-tolérante) : pg_trgm + index trigram GIN (nom client, courriel, sujet, corps des messages).
    try {
      await pool().query('CREATE EXTENSION IF NOT EXISTS pg_trgm')
      await pool().query(`CREATE INDEX IF NOT EXISTS conv_name_trgm  ON comms.conversations USING gin (customer_name gin_trgm_ops)`)
      await pool().query(`CREATE INDEX IF NOT EXISTS conv_email_trgm ON comms.conversations USING gin (email gin_trgm_ops)`)
      await pool().query(`CREATE INDEX IF NOT EXISTS conv_subj_trgm  ON comms.conversations USING gin (subject gin_trgm_ops)`)
      await pool().query(`CREATE INDEX IF NOT EXISTS msg_text_trgm   ON comms.messages USING gin ((data->>'text') gin_trgm_ops)`)
      log('comms-pg: recherche floue prête (pg_trgm + index trigram)')
    } catch (e) { log('comms-pg: pg_trgm indisponible (recherche exacte seulement):', e.message) }
    _available = true
    log('comms-pg: schéma normalisé prêt (comms.conversations + comms.messages)')
  } catch (e) {
    _available = false
    log('comms-pg: init échouée → repli JSON :', e.message)
  }
  return _available
}
function available () { return _available }

function toDate (v) { try { return v ? new Date(v) : null } catch { return null } }

const UP_CONV = `INSERT INTO comms.conversations
  (token,channel,phone,email,customer,customer_name,thread_id,subject,queue,assignee,status,last_activity,created_at,expires_at,noise,meta,updated_at)
  VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,now())
  ON CONFLICT (token) DO UPDATE SET channel=EXCLUDED.channel,phone=EXCLUDED.phone,email=EXCLUDED.email,
    customer=EXCLUDED.customer,customer_name=EXCLUDED.customer_name,thread_id=EXCLUDED.thread_id,subject=EXCLUDED.subject,
    queue=EXCLUDED.queue,assignee=EXCLUDED.assignee,status=EXCLUDED.status,last_activity=EXCLUDED.last_activity,
    created_at=EXCLUDED.created_at,expires_at=EXCLUDED.expires_at,noise=EXCLUDED.noise,meta=EXCLUDED.meta,updated_at=now()`
const UP_MSG = `INSERT INTO comms.messages
  (conv_token,id,ord,sender,from_name,from_email,to_email,via,agent,gmail_id,ts,email_date,data,updated_at)
  VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,now())
  ON CONFLICT (conv_token,id) DO UPDATE SET ord=EXCLUDED.ord,sender=EXCLUDED.sender,from_name=EXCLUDED.from_name,
    from_email=EXCLUDED.from_email,to_email=EXCLUDED.to_email,via=EXCLUDED.via,agent=EXCLUDED.agent,
    gmail_id=EXCLUDED.gmail_id,ts=EXCLUDED.ts,email_date=EXCLUDED.email_date,data=EXCLUDED.data,updated_at=now()`

// Upsert d'UNE conversation (entête + messages) dans une transaction.
async function upsert (conv) {
  if (!_available || !conv || !conv.token) return
  const client = await pool().connect()
  try {
    await client.query('BEGIN')
    const meta = { ...conv }; delete meta.messages
    await client.query(UP_CONV, [conv.token, conv.channel || null, conv.phone || null, conv.email || null,
      conv.customer || null, conv.customerName || null, conv.threadId || null, conv.lastSubject || conv.subject || null,
      conv.queue || null, conv.assignee || null, conv.status || null, toDate(conv.lastActivity), toDate(conv.createdAt),
      toDate(conv.expiresAt), !!conv.noise, JSON.stringify(meta)])
    const msgs = Array.isArray(conv.messages) ? conv.messages : []
    const ids = msgs.map(m => String(m.id))
    // retire les messages qui ne sont plus dans la conv (suppression/scission)
    await client.query('DELETE FROM comms.messages WHERE conv_token=$1 AND NOT (id = ANY($2::text[]))', [conv.token, ids.length ? ids : ['']])
    let ord = 0
    for (const m of msgs) {
      await client.query(UP_MSG, [conv.token, String(m.id), ord++, m.from || null, m.fromName || null, m.fromEmail || null,
        m.toEmail || null, m.via || null, m.agent || null, m.gmail_id || null, toDate(m.ts), toDate(m.emailDate), JSON.stringify(m)])
    }
    await client.query('COMMIT')
  } catch (e) {
    try { await client.query('ROLLBACK') } catch { /* */ }
    log('comms-pg upsert:', e.message)
  } finally { client.release() }
}

async function upsertAll (convs) {
  if (!_available || !Array.isArray(convs)) return 0
  let n = 0
  for (const c of convs) { await upsert(c); n++ }
  return n
}

async function remove (token) {
  if (!_available || !token) return
  try {
    await pool().query('DELETE FROM comms.messages WHERE conv_token=$1', [token])
    await pool().query('DELETE FROM comms.conversations WHERE token=$1', [token])
  } catch (e) { log('comms-pg remove:', e.message) }
}

// Reconstruit toutes les conversations (entête meta + messages ordonnés) → tableau de convs identiques à l'original.
async function loadAll () {
  if (!_available) return null
  try {
    const convs = await pool().query('SELECT token, meta FROM comms.conversations')
    const msgs = await pool().query('SELECT conv_token, data FROM comms.messages ORDER BY conv_token, ord')
    const byConv = new Map()
    for (const r of msgs.rows) { const a = byConv.get(r.conv_token) || []; a.push(r.data); byConv.set(r.conv_token, a) }
    return convs.rows.map(r => ({ ...r.meta, messages: byConv.get(r.token) || [] }))
  } catch (e) { log('comms-pg loadAll:', e.message); return null }
}

async function count () {
  if (!_available) return -1
  try { const r = await pool().query('SELECT count(*)::int AS n FROM comms.conversations'); return r.rows[0].n } catch { return -1 }
}
async function msgCount () {
  if (!_available) return -1
  try { const r = await pool().query('SELECT count(*)::int AS n FROM comms.messages'); return r.rows[0].n } catch { return -1 }
}

// Recherche FLOUE omnichannel (SQL) : nom client / courriel / sujet (exact ILIKE + trigram `<%` typo-tolérant)
// + corps des messages (ILIKE indexé trigram). Classé : correspondance directe d'abord, puis similarité, puis récence.
async function search (q, limit = 30) {
  if (!_available) return null // PG indispo → l'appelant fera un repli mémoire
  const term = String(q || '').trim()
  if (!term) return []
  const like = '%' + term.replace(/[%_\\]/g, '') + '%'
  try {
    const r = await pool().query(`
      SELECT c.token, c.customer_name, c.email, c.subject, c.channel, c.queue, c.status, c.last_activity,
        GREATEST(word_similarity($1, coalesce(c.customer_name,'')), word_similarity($1, coalesce(c.email,'')), word_similarity($1, coalesce(c.subject,''))) AS score,
        (c.customer_name ILIKE $2 OR c.email ILIKE $2 OR c.subject ILIKE $2) AS direct,
        (SELECT m.data->>'text' FROM comms.messages m WHERE m.conv_token=c.token AND (m.data->>'text') ILIKE $2 ORDER BY m.ts DESC LIMIT 1) AS msg_hit
      FROM comms.conversations c
      WHERE c.customer_name ILIKE $2 OR c.email ILIKE $2 OR c.subject ILIKE $2
         OR $1 <% coalesce(c.customer_name,'') OR $1 <% coalesce(c.email,'') OR $1 <% coalesce(c.subject,'')
         OR EXISTS (SELECT 1 FROM comms.messages m WHERE m.conv_token=c.token AND (m.data->>'text') ILIKE $2)
      ORDER BY direct DESC, score DESC, c.last_activity DESC NULLS LAST
      LIMIT $3`, [term, like, limit])
    return r.rows.map(x => ({
      token: x.token, customerName: x.customer_name, email: x.email, subject: x.subject,
      channel: x.channel, queue: x.queue, status: x.status, lastActivity: x.last_activity,
      score: Number(x.score || 0), direct: !!x.direct,
      snippet: x.msg_hit ? String(x.msg_hit).replace(/\s+/g, ' ').slice(0, 140) : null,
    }))
  } catch (e) { log('comms-pg search:', e.message); return null }
}

module.exports = { init, available, upsert, upsertAll, remove, loadAll, count, msgCount, search }