diff --git a/frontend/src/pages/WacrmMessages.tsx b/frontend/src/pages/WacrmMessages.tsx index ad65079..06d684f 100644 --- a/frontend/src/pages/WacrmMessages.tsx +++ b/frontend/src/pages/WacrmMessages.tsx @@ -6,6 +6,7 @@ import Layout from '../components/Layout'; import { Card, Button, Input, EmptyState, Skeleton, toast, Badge, Modal } from '../components/ui'; import { odooApi, type WacrmConversation, type WacrmMessage, type Patient, type WacrmTag } from '../services/odoo'; import { useAuth } from '../lib/auth'; +import { useAutoRefresh, AutoRefreshBadge } from '../lib/autoRefresh'; import WacrmMetrics from '../components/wacrm/WacrmMetrics'; import ConversationNotas from '../components/wacrm/ConversationNotas'; import ConversationEtiquetas from '../components/wacrm/ConversationEtiquetas'; @@ -18,9 +19,35 @@ interface Member { role: string; } +// El backend devuelve marcas ISO con microsegundos: se recortan a 3 dígitos +// para que `new Date()` las entienda en todos los navegadores. +const parseMarca = (iso?: string | null): Date | null => { + if (!iso) return null; + const limpio = iso.replace(/(\.\d{3})\d+/, '$1'); + const fecha = new Date(limpio); + return Number.isNaN(fecha.getTime()) ? null : fecha; +}; + +const hhmmss = (fecha: Date | null) => + fecha ? fecha.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit', second: '2-digit' }) : null; + +// Huella de los campos que la bandeja pinta: sirve para no re-renderizar +// cuando el refresco trae exactamente lo mismo. +const firmaConversacion = (c: WacrmConversation) => JSON.stringify([ + c.last_message_at, c.last_message_text, c.unread_count, c.status, + c.assigned_agent_id, c.assigned_agent, (c.tags || []).map((t) => t.id), +]); + +// Cada cuánto se sincroniza el caché del servidor. Cada ciclo es incremental +// (milisegundos), así que 30 s es holgado y deja la bandeja viva. +const SYNC_INTERVAL_MS = 30000; + const WacrmMessages: FC = () => { const [conversations, setConversations] = useState([]); const [messages, setMessages] = useState([]); + // Espejo de `messages` para que `loadMessages` no dependa del estado (ver + // comentario en loadMessages). + const messagesRef = useRef([]); const [msgPage, setMsgPage] = useState(1); const [msgHasMore, setMsgHasMore] = useState(false); const [loadingOlder, setLoadingOlder] = useState(false); @@ -31,7 +58,8 @@ const WacrmMessages: FC = () => { const cercaDelFondoRef = useRef(true); const [members, setMembers] = useState([]); const [loading, setLoading] = useState(true); - const [syncing, setSyncing] = useState(false); + // Hora (del servidor) de la última sincronización del caché con WACRM. + const [lastSyncAt, setLastSyncAt] = useState(null); const [search, setSearch] = useState(''); const [selected, setSelected] = useState(null); const [assignOpen, setAssignOpen] = useState(false); @@ -77,6 +105,10 @@ const WacrmMessages: FC = () => { searchRef.current = search; }, [search]); + useEffect(() => { + messagesRef.current = messages; + }, [messages]); + useEffect(() => { if (currentAgentId && typeof window !== 'undefined') { localStorage.setItem('skeen_current_agent_id', currentAgentId); @@ -88,11 +120,13 @@ const WacrmMessages: FC = () => { const res = await odooApi.getWacrmConversations(searchRef.current); if (res.status === 'success') { setConversations(res.conversations); + if (res.last_sync_at) setLastSyncAt(res.last_sync_at); const currentSelected = selectedRef.current; if (currentSelected) { const updated = res.conversations.find((c) => c.external_id === currentSelected.external_id); - if (updated && updated.external_id !== currentSelected.external_id) { - // Solo actualizar si realmente cambió la referencia (datos nuevos) + // La conversación abierta también se refresca (último mensaje, no + // leídos, asignación...), pero solo si de verdad cambió algo. + if (updated && firmaConversacion(updated) !== firmaConversacion(currentSelected)) { setSelected(updated); } } else if (!userSelectedRef.current && res.conversations.length > 0) { @@ -138,9 +172,10 @@ const WacrmMessages: FC = () => { setMsgHasMore(res.has_more); if (page === 1) { // Merge: conserva el historial antiguo ya cargado (scroll manual) + const previos = messagesRef.current; const known = new Set(res.messages.map((m) => m.external_id)); const minTs = res.messages.length ? res.messages[0].created_at : ''; - const older = messages.filter((m) => !known.has(m.external_id) && (m.created_at ?? '') < (minTs ?? '')); + const older = previos.filter((m) => !known.has(m.external_id) && (m.created_at ?? '') < (minTs ?? '')); setMessages([...older, ...res.messages]); // Solo baja al final si cambió de conversación o el usuario ya estaba // abajo; si estaba leyendo mensajes anteriores, se respeta su posición. @@ -148,7 +183,7 @@ const WacrmMessages: FC = () => { scrollThreadToBottom(); } } else { - const prev = messages; + const prev = messagesRef.current; const known = new Set(prev.map((m) => m.external_id)); const older = res.messages.filter((m) => !known.has(m.external_id)); setMessages([...older, ...prev]); @@ -163,7 +198,10 @@ const WacrmMessages: FC = () => { toast.error('Error al cargar mensajes'); console.error(err); } - }, [messages]); + // Sin dependencia de `messages` a propósito: si la tuviera, cada mensaje + // nuevo la recrearía y el efecto que la llama volvería a pedir el hilo en + // bucle (una petición por render). + }, []); const loadMembers = useCallback(async () => { try { @@ -176,25 +214,35 @@ const WacrmMessages: FC = () => { } }, []); - const sync = async () => { + // Un ciclo = sincronizar el caché del servidor y volver a leer la bandeja y + // el hilo abierto. El hook de auto-refresco se encarga de que nunca haya dos + // ciclos a la vez y de que no corra con la pestaña oculta. + const refreshAll = useCallback(async () => { try { - setSyncing(true); const res = await odooApi.syncWacrm(); if (res.status === 'success') { - toast.success(`Sincronizado: ${JSON.stringify(res.result)}`); - await loadConversations(); + if (res.result?.last_sync_at) setLastSyncAt(String(res.result.last_sync_at)); } else if (res.status === 'busy') { - // La auto-sincronización cada 15 s ya está corriendo - toast.info('Ya hay una sincronización en curso'); - await loadConversations(); + // Otro ciclo (esta u otra pestaña) está corriendo: no es un error. + console.info('Sincronización WACRM en curso; se reintenta al siguiente ciclo'); } } catch (err) { - toast.error('Error al sincronizar con WACRM'); - console.error(err); - } finally { - setSyncing(false); + // Un fallo de sincronización no debe romper la bandeja: se sigue viendo + // el caché y se reintenta en el siguiente ciclo. + console.error('Error al sincronizar con WACRM', err); } - }; + await loadConversations(); + const current = selectedRef.current; + if (current) await loadMessages(current.external_id); + }, [loadConversations, loadMessages]); + + const auto = useAutoRefresh({ + onRefresh: refreshAll, + intervalMs: SYNC_INTERVAL_MS, + storageKey: 'skeen_wacrm_autorefresh_paused', + }); + + const sync = useCallback(() => auto.refreshNow(), [auto]); const loadTags = useCallback(async () => { try { @@ -289,17 +337,9 @@ const WacrmMessages: FC = () => { } }, [selected, loadMessages]); - // Auto-sync cada 15 segundos - useEffect(() => { - const interval = setInterval(() => { - loadConversations(); - const currentSelected = selectedRef.current; - if (currentSelected) { - loadMessages(currentSelected.external_id); - } - }, 15000); - return () => clearInterval(interval); - }, [loadConversations, loadMessages]); + // El ciclo de sincronización + refresco lo lleva `useAutoRefresh` (ver + // `auto` arriba): espera a que termine el anterior, no corre con la pestaña + // oculta y se puede pausar desde el indicador. const handleSelect = useCallback((c: WacrmConversation) => { userSelectedRef.current = true; @@ -402,7 +442,7 @@ const WacrmMessages: FC = () => { className="pl-10" /> -
+
- diff --git a/frontend/src/services/odoo.ts b/frontend/src/services/odoo.ts index cb15bc7..a2da3d9 100644 --- a/frontend/src/services/odoo.ts +++ b/frontend/src/services/odoo.ts @@ -5403,7 +5403,7 @@ export const odooApi = { // ============================================================ // WACRM Proxy // ============================================================ - async syncWacrm(): Promise<{ status: string; result: Record }> { + async syncWacrm(): Promise<{ status: string; result: Record }> { const { data } = await axios.post(`${ODOO_BASE}/skeen/frontend/v1/wacrm/sync`); return data; }, @@ -5420,7 +5420,7 @@ export const odooApi = { return data; }, - async getWacrmConversations(search?: string): Promise<{ status: string; conversations: WacrmConversation[] }> { + async getWacrmConversations(search?: string): Promise<{ status: string; conversations: WacrmConversation[]; last_sync_at?: string | null }> { const { data } = await axios.get(`${ODOO_BASE}/skeen/frontend/v1/wacrm/conversations`, { params: search ? { search } : {} }); return data; }, diff --git a/odoo-addons/skeen_whatsapp/controllers/wacrm_proxy.py b/odoo-addons/skeen_whatsapp/controllers/wacrm_proxy.py index cb05b85..745e64d 100644 --- a/odoo-addons/skeen_whatsapp/controllers/wacrm_proxy.py +++ b/odoo-addons/skeen_whatsapp/controllers/wacrm_proxy.py @@ -9,6 +9,7 @@ from odoo.http import request from .frontend import json_response, _parse_json_body from .auth import require_role, require_perm import os +import hmac import logging import requests import re @@ -18,6 +19,32 @@ from datetime import datetime, timedelta _logger = logging.getLogger(__name__) +# ============================================================ +# Sincronización incremental +# +# El caché local se llena desde Supabase (WACRM). Recorrer las 9.750 filas +# de `contacts` y todos los mensajes en cada ciclo costaba ~75 s, así que la +# pantalla nunca alcanzaba a sincronizar. Ahora: +# - los contactos se traen SOLO si están referenciados por una conversación +# o un deal, o si cambiaron desde la última marca; +# - conversaciones/mensajes/deals se traen por fecha de actualización; +# - las escrituras van en lote (una búsqueda + un create múltiple); +# - la poda de lo que ya no existe en el origen SOLO corre en un ciclo +# completo (cada FULL_SYNC_INTERVAL segundos), nunca en un parcial. +# Las marcas viven en ir.config_parameter y solo avanzan, nunca retroceden. +# ============================================================ +PARAM_CONV_MARK = 'skeen.wacrm.last_sync' # marca de conversations +PARAM_MSG_MARK = 'skeen.wacrm.last_msg_sync' # marca de messages +PARAM_CONTACT_MARK = 'skeen.wacrm.last_contact_sync' +PARAM_DEAL_MARK = 'skeen.wacrm.last_deal_sync' +PARAM_LAST_SYNC_AT = 'skeen.wacrm.last_sync_at' # hora local del último ciclo ok +PARAM_LAST_FULL_AT = 'skeen.wacrm.last_full_sync_at' +PARAM_NOTIFY_KEY = 'skeen.wacrm.notify_key' # llave del aviso puntual +FULL_SYNC_INTERVAL = 300 # segundos entre ciclos completos (poda) +INCREMENTAL_CONTACT_PAGE = 1000 # tope de contactos cambiados por ciclo +SUPABASE_PAGE_SIZE = 1000 +EPOCH_SUPABASE = '1970-01-01T00:00:00.000000+00:00' + def _get_supabase_config(): """Lee configuración de Supabase desde variables de entorno o ir.config_parameter.""" @@ -84,6 +111,26 @@ def _hermes_bridge_request(path, json=None): return response.json() +def _supabase_server_time(): + """Hora del propio Supabase (cabecera Date), en UTC naive. + + Se usa para arrancar las marcas con el reloj del origen y no con el de + Odoo, que puede estar desfasado. + """ + url, key = _get_supabase_config() + try: + from email.utils import parsedate_to_datetime + from datetime import timezone + response = requests.get( + f'{url}/rest/v1/', + headers={'apikey': key, 'Authorization': f'Bearer {key}'}, + timeout=10) + return parsedate_to_datetime(response.headers['Date']).astimezone(timezone.utc).replace(tzinfo=None) + except Exception as e: + _logger.warning('No se pudo leer la hora de Supabase: %s', e) + return None + + def _supabase_request_paginated(table, params=None, page_size=1000): """Itera sobre todas las páginas de resultados de PostgREST. @@ -120,6 +167,141 @@ def _parse_datetime(value): return None +def _to_supabase_dt(dt): + """Datetime naive UTC → texto ISO 8601 que PostgREST acepta en filtros.""" + return dt.strftime('%Y-%m-%dT%H:%M:%S.%f') + '+00:00' + + +def _max_dt(rows, field): + """Máximo (naive UTC) del campo en las filas, o None si no hay ninguno.""" + mayor = None + for row in rows: + dt = _parse_datetime(row.get(field)) + if dt and (mayor is None or dt > mayor): + mayor = dt + return mayor + + +def _chunks(items, size): + for i in range(0, len(items), size): + yield items[i:i + size] + + +def _changed_vals(record, vals): + """Subconjunto de `vals` que difiere de lo que ya tiene el registro.""" + cambios = {} + for field, valor in vals.items(): + actual = record[field] + if hasattr(actual, 'ids'): + actual = actual.id or False + valor = valor or False + # Un campo vacío en Odoo se lee como False; el origen manda None. Se + # igualan para no reescribir la fila en cada ciclo. + if valor is None: + valor = False + if actual is None: + actual = False + if isinstance(actual, datetime) and isinstance(valor, datetime): + if abs((actual - valor).total_seconds()) < 0.001: + continue + elif actual == valor: + continue + cambios[field] = valor + return cambios + + +def _bulk_upsert(model, rows, key='external_id'): + """Crea/actualiza en lote. Devuelve {clave: registro}. + + Una sola búsqueda por lote, un `create` múltiple y escrituras agrupadas + por valores idénticos: nada de una consulta por fila. + """ + if not rows: + return {} + by_key = {rec[key]: rec for rec in model.search([(key, 'in', [r[key] for r in rows])])} + nuevos = [r for r in rows if r[key] not in by_key] + nuevos_ids = set() + if nuevos: + for rec in model.create(nuevos): + by_key[rec[key]] = rec + nuevos_ids.add(rec.id) + grupos = {} + for row in rows: + rec = by_key.get(row[key]) + if rec is None or rec.id in nuevos_ids: + continue + cambios = _changed_vals(rec, {k: v for k, v in row.items() if k != key}) + if not cambios: + continue + firma = tuple(sorted((k, str(v)) for k, v in cambios.items())) + previos = grupos.get(firma) + grupos[firma] = (cambios, (previos[1] | rec) if previos else rec) + for cambios, recs in grupos.values(): + recs.write(cambios) + return by_key + + +def _contacto_a_vals(item): + return { + 'external_id': item['id'], + 'account_id': item.get('account_id'), + 'phone': item.get('phone'), + 'name': item.get('name') or item.get('phone'), + 'email': item.get('email'), + 'company': item.get('company'), + 'avatar_url': item.get('avatar_url'), + 'created_at': _parse_datetime(item.get('created_at')), + 'updated_at': _parse_datetime(item.get('updated_at')), + } + + +def _conversacion_a_vals(item, contacts_map): + vals = { + 'external_id': item['id'], + 'status': item.get('status') or 'open', + 'assigned_agent_id': item.get('assigned_agent_id') or None, + 'assigned_agent': str(item.get('assigned_agent_id')) if item.get('assigned_agent_id') else None, + 'last_message_text': item.get('last_message_text'), + 'last_message_at': _parse_datetime(item.get('last_message_at')), + 'unread_count': item.get('unread_count', 0), + 'created_at': _parse_datetime(item.get('created_at')), + 'updated_at': _parse_datetime(item.get('updated_at')), + } + contacto = contacts_map.get(item.get('contact_id')) + if contacto: + vals['contact_id'] = contacto.id + elif not item.get('contact_id'): + vals['contact_id'] = False + # Si el contacto no vino en el lote no se toca el vínculo ya guardado. + return vals + + +def _traer_contactos(env, contact_ids): + """Upserta los contactos indicados trayéndolos del origen. Devuelve el mapa.""" + if not contact_ids: + return {} + items = [] + for trozo in _chunks(sorted(contact_ids), 150): + items.extend(_supabase_request('GET', 'contacts', { + 'select': '*', 'id': 'in.(%s)' % ','.join(trozo)})) + return _bulk_upsert(env['skeen.wacrm.contact'].sudo(), + [_contacto_a_vals(i) for i in items]) + + +def _traer_conversaciones(env, conv_ids, contacts_map): + """Trae del origen las conversaciones indicadas (y sus contactos).""" + if not conv_ids: + return {} + items = [] + for trozo in _chunks(sorted(conv_ids), 100): + items.extend(_supabase_request('GET', 'conversations', { + 'select': '*', 'id': 'in.(%s)' % ','.join(trozo)})) + faltan = {i['contact_id'] for i in items if i.get('contact_id')} - set(contacts_map) + contacts_map.update(_traer_contactos(env, faltan)) + return _bulk_upsert(env['skeen.wacrm.conversation'].sudo(), + [_conversacion_a_vals(i, contacts_map) for i in items]) + + class SkeenWacrmProxyController(http.Controller): """Endpoints proxy WACRM para el frontend.""" @@ -128,32 +310,41 @@ class SkeenWacrmProxyController(http.Controller): # ============================================================ @http.route('/skeen/frontend/v1/wacrm/sync', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False) def sync_wacrm(self, **kw): - """Sincroniza contactos, conversaciones, mensajes, pipelines y deals desde Supabase hacia Odoo. + """Sincroniza el caché local con Supabase (WACRM). - El frontend tiene auto-sync cada 15 s y un botón manual: si ambos - coinciden, dos transacciones escriben las mismas filas del caché y - Postgres aborta una con "could not serialize access" (500). Se - reintenta con la transacción limpia, que es lo que hace Odoo en sus - propios reintentos. + Por defecto corre en modo incremental: solo trae lo que cambió desde + la última marca. Cada FULL_SYNC_INTERVAL segundos —o con `full=1`, o + la primera vez— corre un ciclo completo, que además poda del caché lo + que ya no existe en el origen. + + Un solo sync a la vez: dos transacciones escribiendo las mismas filas + se abortan mutuamente ("could not serialize access"), así que el + segundo responde 'busy' y ante un conflicto real se reintenta. """ - # Un solo sync a la vez: el frontend auto-sincroniza cada 15 s y tiene - # botón manual; dos syncs simultáneos chocan en las mismas filas del - # caché y Postgres aborta uno. El lock de transacción se libera solo. + # El lock de transacción se libera solo al terminar la petición. try: request.env.cr.execute( "SELECT pg_try_advisory_xact_lock(hashtext('skeen_wacrm_sync'))") libre = request.env.cr.fetchone()[0] except Exception: libre = True + ICP = request.env['ir.config_parameter'].sudo() if not libre: return json_response({ 'status': 'busy', 'message': 'Ya hay una sincronización en curso; se actualizará en unos segundos', + 'last_sync_at': ICP.get_param(PARAM_LAST_SYNC_AT), }) + pedir_full = str(kw.get('full', '')).lower() in ('1', 'true', 'yes', 'si') + if not pedir_full: + try: + pedir_full = bool((_parse_json_body() or {}).get('full')) + except Exception: + pedir_full = False ultimo = None for intento in range(3): try: - result = self._do_sync() + result = self._do_sync(full=pedir_full) return json_response({'status': 'success', 'result': result}) except Exception as e: ultimo = e @@ -165,168 +356,164 @@ class SkeenWacrmProxyController(http.Controller): _logger.exception('Error sincronizando WACRM') return json_response({'status': 'error', 'message': str(ultimo)}, 500) - def _do_sync(self): + def _do_sync(self, full=False): + inicio = time.monotonic() + ICP = request.env['ir.config_parameter'].sudo() + Contact = request.env['skeen.wacrm.contact'].sudo() + Conv = request.env['skeen.wacrm.conversation'].sudo() + Msg = request.env['skeen.wacrm.message'].sudo() + Deal = request.env['skeen.wacrm.deal'].sudo() + + marca_conv = ICP.get_param(PARAM_CONV_MARK) + marca_msg = ICP.get_param(PARAM_MSG_MARK) + marca_contacto = ICP.get_param(PARAM_CONTACT_MARK) + marca_deal = ICP.get_param(PARAM_DEAL_MARK) + ultimo_full = _parse_datetime(ICP.get_param(PARAM_LAST_FULL_AT)) + + # Sin marcas previas (primer arranque) o con el ciclo completo vencido, + # se hace una pasada completa: es la única que poda lo que ya no existe. + if not marca_conv or not marca_msg: + full = True + elif not ultimo_full or (fields.Datetime.now() - ultimo_full).total_seconds() >= FULL_SYNC_INTERVAL: + full = True + modo = 'full' if full else 'incremental' account_id = None - # 1. Contacts (paginado por si PostgREST limita a 1000 filas) - contacts_data = _supabase_request_paginated('contacts', {'select': '*'}) - contacts_map = {} + # 1. Conversaciones (completas la primera vez y en cada ciclo completo; + # después solo las que cambiaron desde la marca). + params_conv = {'select': '*', 'order': 'id.asc' if full else 'updated_at.asc,id.asc'} + if not full: + params_conv['updated_at'] = 'gte.%s' % marca_conv + conversations_data = _supabase_request_paginated('conversations', params_conv) + + # 2. Deals (igual criterio) + params_deal = {'select': '*', 'order': 'id.asc' if full else 'updated_at.asc,id.asc'} + if not full: + params_deal['updated_at'] = 'gte.%s' % (marca_deal or EPOCH_SUPABASE) + deals_data = _supabase_request_paginated('deals', params_deal) + + # 3. Contactos: solo los que hacen falta. Los referenciados por lo que + # se acaba de traer, más los que cambiaron desde la última marca. + # La marca arranca en la hora DEL ORIGEN en el primer ciclo: el + # histórico de pacientes importados no se recorre, solo se siguen + # los cambios posteriores (acotado a una página por ciclo, para que + # un reimport masivo no arrastre las 9.750 filas de golpe). + if not marca_contacto: + servidor = _supabase_server_time() + if servidor: + marca_contacto = _to_supabase_dt(servidor) + ICP.set_param(PARAM_CONTACT_MARK, marca_contacto) + referenciados = [i['contact_id'] for i in conversations_data + deals_data if i.get('contact_id')] + referenciados = list(dict.fromkeys(referenciados)) + contacts_data = [] + for trozo in _chunks(referenciados, 150): + contacts_data.extend(_supabase_request('GET', 'contacts', { + 'select': '*', 'id': 'in.(%s)' % ','.join(trozo)})) + contactos_barridos = [] + if not full and marca_contacto: + contactos_barridos = _supabase_request('GET', 'contacts', { + 'select': '*', 'updated_at': 'gte.%s' % marca_contacto, + 'order': 'updated_at.asc,id.asc', 'limit': INCREMENTAL_CONTACT_PAGE}) + contacts_data.extend(contactos_barridos) + contacts_data = list({i['id']: i for i in contacts_data}.values()) for item in contacts_data: if not account_id and item.get('account_id'): account_id = item['account_id'] - vals = { - 'external_id': item['id'], - 'account_id': item.get('account_id'), - 'phone': item.get('phone'), - 'name': item.get('name') or item.get('phone'), - 'email': item.get('email'), - 'company': item.get('company'), - 'avatar_url': item.get('avatar_url'), - 'created_at': _parse_datetime(item.get('created_at')), - 'updated_at': _parse_datetime(item.get('updated_at')), - } - existing = request.env['skeen.wacrm.contact'].sudo().search([('external_id', '=', item['id'])], limit=1) - if existing: - existing.write(vals) - contacts_map[item['id']] = existing - else: - contacts_map[item['id']] = request.env['skeen.wacrm.contact'].sudo().create(vals) + if not account_id: + fila = _supabase_request('GET', 'contacts', {'select': 'account_id', 'limit': 1}) + account_id = fila[0].get('account_id') if fila else None + contacts_map = _bulk_upsert(Contact, [_contacto_a_vals(i) for i in contacts_data]) - # 2. Conversations - # Además del upsert, se REFLEJAN ELIMINACIONES: lo que ya no exista - # en Supabase se borra del caché local (antes el caché acumulaba - # conversaciones borradas y la bandeja de SKEEN mostraba chats muertos). - conversations_data = _supabase_request_paginated('conversations', {'select': '*'}) - conversations_map = {} - for item in conversations_data: - vals = { - 'external_id': item['id'], - 'contact_id': contacts_map.get(item.get('contact_id'), request.env['skeen.wacrm.contact']).id if contacts_map.get(item.get('contact_id')) else False, - 'status': item.get('status', 'open'), - 'assigned_agent_id': item.get('assigned_agent_id') or None, - 'assigned_agent': str(item.get('assigned_agent_id')) if item.get('assigned_agent_id') else None, - 'last_message_text': item.get('last_message_text'), - 'last_message_at': _parse_datetime(item.get('last_message_at')), - 'unread_count': item.get('unread_count', 0), - 'created_at': _parse_datetime(item.get('created_at')), - 'updated_at': _parse_datetime(item.get('updated_at')), - } - existing = request.env['skeen.wacrm.conversation'].sudo().search([('external_id', '=', item['id'])], limit=1) - if existing: - existing.write(vals) - conversations_map[item['id']] = existing - else: - conversations_map[item['id']] = request.env['skeen.wacrm.conversation'].sudo().create(vals) + # 4. Conversaciones: el vínculo con el contacto se resuelve con el mapa + # anterior. Si un contacto no vino en el lote, no se pisa el vínculo. + conversations_map = _bulk_upsert( + Conv, [_conversacion_a_vals(i, contacts_map) for i in conversations_data]) - remote_conv_ids = set(conversations_map.keys()) - stale_convs = request.env['skeen.wacrm.conversation'].sudo().search([ - ('external_id', 'not in', list(remote_conv_ids))]) if remote_conv_ids else \ - request.env['skeen.wacrm.conversation'].sudo().search([]) - if stale_convs: - stale_convs.unlink() - conversations_map = {k: v for k, v in conversations_map.items()} + # 5. Mensajes: completos la primera vez y en cada ciclo completo; luego + # solo los nuevos, en orden ascendente (con descendente + offset los + # mensajes que van entrando desplazan las páginas y se duplican). + params_msg = {'select': '*', 'order': 'created_at.asc,id.asc'} + if not full: + params_msg['created_at'] = 'gte.%s' % marca_msg + messages_data = _supabase_request_paginated('messages', params_msg, page_size=SUPABASE_PAGE_SIZE) - # 3. Messages (todos los del origen, sin recorte) - messages_data = _supabase_request_paginated( - 'messages', - {'select': '*', 'order': 'created_at.desc'}, - page_size=1000 - ) - remote_msg_ids = {item['id'] for item in messages_data} - stale_msgs = request.env['skeen.wacrm.message'].sudo().search([ - ('external_id', 'not in', list(remote_msg_ids))]) if remote_msg_ids else \ - request.env['skeen.wacrm.message'].sudo().search([]) - stale_msgs.unlink() + # Una conversación que no vino en el lote puede tener mensajes nuevos + # (lo normal en incremental): se resuelve contra el caché y, si tampoco + # está, se trae del origen junto con su contacto. + pendientes = {m['conversation_id'] for m in messages_data if m.get('conversation_id')} - set(conversations_map) + if pendientes: + locales = Conv.search([('external_id', 'in', list(pendientes))]) + for conv in locales: + conversations_map[conv.external_id] = conv + pendientes -= set(locales.mapped('external_id')) + if pendientes: + conversations_map.update(_traer_conversaciones(request.env, pendientes, contacts_map)) + + msg_vals = [] for item in messages_data: conversation = conversations_map.get(item.get('conversation_id')) if not conversation: continue - vals = { + msg_vals.append({ 'external_id': item['id'], 'conversation_id': conversation.id, - 'sender_type': {'contact': 'customer', 'user': 'agent'}.get(item.get('sender_type', 'customer'), item.get('sender_type', 'customer')), - 'content_type': item.get('content_type', 'text'), + 'sender_type': {'contact': 'customer', 'user': 'agent'}.get( + item.get('sender_type', 'customer'), item.get('sender_type', 'customer')), + 'content_type': item.get('content_type') or 'text', 'content_text': item.get('content_text') or item.get('content'), 'media_url': item.get('media_url'), 'template_name': item.get('template_name'), 'status': item.get('status', 'sent'), 'created_at': _parse_datetime(item.get('created_at')), - } - existing = request.env['skeen.wacrm.message'].sudo().search([('external_id', '=', item['id'])], limit=1) - if existing: - existing.write(vals) - else: - request.env['skeen.wacrm.message'].sudo().create(vals) + }) + _bulk_upsert(Msg, msg_vals) - # 4. Pipelines & Stages - pipelines_data = _supabase_request_paginated('pipelines', {'select': '*'}) - pipelines_map = {} - stages_map = {} - for item in pipelines_data: - vals = { - 'external_id': item['id'], - 'name': item.get('name'), - 'created_at': _parse_datetime(item.get('created_at')), - } - existing = request.env['skeen.wacrm.pipeline'].sudo().search([('external_id', '=', item['id'])], limit=1) - if existing: - existing.write(vals) - pipelines_map[item['id']] = existing - else: - pipelines_map[item['id']] = request.env['skeen.wacrm.pipeline'].sudo().create(vals) + # 6. Pipelines & Stages (catálogo diminuto: siempre completo) + pipelines_data = _supabase_request_paginated('pipelines', {'select': '*', 'order': 'id.asc'}) + pipelines_map = _bulk_upsert(request.env['skeen.wacrm.pipeline'].sudo(), [{ + 'external_id': item['id'], + 'name': item.get('name'), + 'created_at': _parse_datetime(item.get('created_at')), + } for item in pipelines_data]) - stages_data = _supabase_request_paginated('pipeline_stages', {'select': '*'}) - for item in stages_data: - vals = { - 'external_id': item['id'], - 'pipeline_id': pipelines_map.get(item.get('pipeline_id'), request.env['skeen.wacrm.pipeline']).id if pipelines_map.get(item.get('pipeline_id')) else False, - 'name': item.get('name'), - 'position': item.get('position', 0), - 'color': item.get('color', '#3b82f6'), - } - existing = request.env['skeen.wacrm.stage'].sudo().search([('external_id', '=', item['id'])], limit=1) - if existing: - existing.write(vals) - stages_map[item['id']] = existing - else: - stages_map[item['id']] = request.env['skeen.wacrm.stage'].sudo().create(vals) + stages_data = _supabase_request_paginated('pipeline_stages', {'select': '*', 'order': 'id.asc'}) + stages_map = _bulk_upsert(request.env['skeen.wacrm.stage'].sudo(), [{ + 'external_id': item['id'], + 'pipeline_id': pipelines_map[item['pipeline_id']].id if pipelines_map.get(item.get('pipeline_id')) else False, + 'name': item.get('name'), + 'position': item.get('position', 0), + 'color': item.get('color') or '#3b82f6', + } for item in stages_data]) - # 5. Deals - # Mapa profiles.id → user_id (Supabase guarda assigned_to como profiles.id; - # localmente usamos user_id, consistente con conversations.assigned_agent_id) + # 7. Deals — mapa profiles.id → user_id (Supabase guarda assigned_to como + # profiles.id; localmente usamos user_id, como assigned_agent_id). profiles_usermap = {} try: for p in _supabase_request('GET', 'profiles', params={'select': 'id,user_id'}): profiles_usermap[p['id']] = p['user_id'] except Exception: pass - deals_data = _supabase_request_paginated('deals', {'select': '*'}) - for item in deals_data: - vals = { - 'external_id': item['id'], - 'title': item.get('title') or 'Sin título', - 'contact_id': contacts_map.get(item.get('contact_id'), request.env['skeen.wacrm.contact']).id if contacts_map.get(item.get('contact_id')) else False, - 'conversation_id': item.get('conversation_id'), - 'pipeline_id': pipelines_map.get(item.get('pipeline_id'), request.env['skeen.wacrm.pipeline']).id if pipelines_map.get(item.get('pipeline_id')) else False, - 'stage_id': stages_map.get(item.get('stage_id'), request.env['skeen.wacrm.stage']).id if stages_map.get(item.get('stage_id')) else False, - 'value': float(item.get('value', 0) or 0), - 'currency': item.get('currency', 'USD'), - 'status': item.get('status', 'open'), - 'expected_close_date': item.get('expected_close_date'), - 'notes': item.get('notes'), - 'assigned_to': profiles_usermap.get(item.get('assigned_to')) if item.get('assigned_to') else None, - 'created_at': _parse_datetime(item.get('created_at')), - 'updated_at': _parse_datetime(item.get('updated_at')), - } - existing = request.env['skeen.wacrm.deal'].sudo().search([('external_id', '=', item['id'])], limit=1) - if existing: - existing.write(vals) - else: - request.env['skeen.wacrm.deal'].sudo().create(vals) + _bulk_upsert(Deal, [{ + 'external_id': item['id'], + 'title': item.get('title') or 'Sin título', + 'contact_id': contacts_map[item['contact_id']].id if contacts_map.get(item.get('contact_id')) else False, + 'conversation_id': item.get('conversation_id'), + 'pipeline_id': pipelines_map[item['pipeline_id']].id if pipelines_map.get(item.get('pipeline_id')) else False, + 'stage_id': stages_map[item['stage_id']].id if stages_map.get(item.get('stage_id')) else False, + 'value': float(item.get('value', 0) or 0), + 'currency': item.get('currency', 'USD'), + 'status': item.get('status', 'open'), + 'expected_close_date': fields.Date.to_date(item.get('expected_close_date')), + 'notes': item.get('notes'), + 'assigned_to': profiles_usermap.get(item.get('assigned_to')) if item.get('assigned_to') else None, + 'created_at': _parse_datetime(item.get('created_at')), + 'updated_at': _parse_datetime(item.get('updated_at')), + } for item in deals_data]) - # 6. Auto-crear deals: contactos con conversación sin deal abierto. - # Se crean EN SUPABASE (la app WACRM nativa los ve) y se espejan local. - # Idempotente: un deal abierto por contacto. + # 8. Auto-crear deals: contactos con conversación sin deal abierto. + # Se crean EN SUPABASE (la app WACRM nativa los ve) y se espejan local. + # Idempotente: un deal por contacto. En incremental `deals_data` es + # parcial, así que la lista de contactos con deal se pide al origen. deals_created = 0 try: if conversations_map and pipelines_map and account_id: @@ -338,9 +525,15 @@ class SkeenWacrmProxyController(http.Controller): 'select': 'user_id', 'order': 'created_at.asc', 'limit': 1}) owner_id = owner[0]['user_id'] if owner else None if stage and owner_id: - Deal = request.env['skeen.wacrm.deal'].sudo() - # Un deal por contacto: no crear si ya tiene cualquier deal (open o ganado) + # Un deal por contacto: no crear si ya tiene cualquiera (open o ganado) contacts_with_deals = {str(d.get('contact_id')) for d in deals_data if d.get('contact_id')} + if not full: + con_conv = [c.external_id for c in conversations_map.values() if c.contact_id] + for trozo in _chunks(con_conv, 100): + remotos = _supabase_request('GET', 'deals', { + 'select': 'contact_id', + 'contact_id': 'in.(%s)' % ','.join(trozo)}) + contacts_with_deals |= {str(d['contact_id']) for d in remotos if d.get('contact_id')} for conv in conversations_map.values(): contact = conv.contact_id if not contact: @@ -382,7 +575,46 @@ class SkeenWacrmProxyController(http.Controller): except Exception as e: _logger.warning('Auto-deals omitidos: %s', e) + # 9. Poda: SOLO en el ciclo completo. Comparar contra un lote parcial + # (incremental o una página suelta) borraría filas válidas; por eso + # el borrado queda reservado a la pasada que trajo TODO el origen. + # Los mensajes de una conversación borrada caen por cascade. + borrados = {'conversations': 0, 'messages': 0} + if full: + remotos_conv = set(conversations_map) + sobran_conv = Conv.search([]).filtered(lambda c: c.external_id not in remotos_conv) + if sobran_conv: + borrados['conversations'] = len(sobran_conv) + sobran_conv.unlink() + remotos_msg = {m['id'] for m in messages_data} + sobran_msg = Msg.search([]).filtered(lambda m: m.external_id not in remotos_msg) + if sobran_msg: + borrados['messages'] = len(sobran_msg) + sobran_msg.unlink() + + # 10. Marcas: avanzan al máximo visto en cada tabla, nunca retroceden. + ahora = fields.Datetime.now() + ICP.set_param(PARAM_LAST_SYNC_AT, _to_supabase_dt(ahora)) + if full: + ICP.set_param(PARAM_LAST_FULL_AT, _to_supabase_dt(ahora)) + + def _avanzar(param, nuevo): + if not nuevo: + return + previo = _parse_datetime(ICP.get_param(param)) + if previo is None or nuevo > previo: + ICP.set_param(param, _to_supabase_dt(nuevo)) + + _avanzar(PARAM_CONV_MARK, _max_dt(conversations_data, 'updated_at')) + _avanzar(PARAM_MSG_MARK, _max_dt(messages_data, 'created_at')) + # Solo la página del barrido mueve la marca de contactos: los + # referenciados se traen siempre aparte, y usarlos aquí saltaría + # contactos intermedios del barrido. + _avanzar(PARAM_CONTACT_MARK, _max_dt(contactos_barridos, 'updated_at')) + _avanzar(PARAM_DEAL_MARK, _max_dt(deals_data, 'updated_at')) + return { + 'mode': modo, 'contacts': len(contacts_data), 'conversations': len(conversations_data), 'messages': len(messages_data), @@ -390,9 +622,75 @@ class SkeenWacrmProxyController(http.Controller): 'stages': len(stages_data), 'deals': len(deals_data), 'deals_created': deals_created, + 'deleted': borrados, 'account_id': account_id, + 'duration_ms': int((time.monotonic() - inicio) * 1000), + 'last_sync_at': ICP.get_param(PARAM_LAST_SYNC_AT), } + def _sync_conversacion(self, external_id): + """Alta puntual de UNA conversación, su contacto y sus mensajes.""" + Conv = request.env['skeen.wacrm.conversation'].sudo() + convs = _supabase_request('GET', 'conversations', { + 'select': '*', 'id': 'eq.%s' % external_id}) + if convs: + contacts_map = _traer_contactos( + request.env, {c['contact_id'] for c in convs if c.get('contact_id')}) + _bulk_upsert(Conv, [_conversacion_a_vals(c, contacts_map) for c in convs]) + conv = Conv.search([('external_id', '=', external_id)], limit=1) + if not conv: + return {'conversation': 0, 'messages': 0} + msgs = _supabase_request_paginated('messages', { + 'select': '*', 'conversation_id': 'eq.%s' % external_id, + 'order': 'created_at.asc,id.asc'}) + _bulk_upsert(request.env['skeen.wacrm.message'].sudo(), [{ + 'external_id': m['id'], + 'conversation_id': conv.id, + 'sender_type': {'contact': 'customer', 'user': 'agent'}.get( + m.get('sender_type', 'customer'), m.get('sender_type', 'customer')), + 'content_type': m.get('content_type') or 'text', + 'content_text': m.get('content_text') or m.get('content'), + 'media_url': m.get('media_url'), + 'template_name': m.get('template_name'), + 'status': m.get('status', 'sent'), + 'created_at': _parse_datetime(m.get('created_at')), + } for m in msgs]) + return {'conversation': len(convs), 'messages': len(msgs)} + + @http.route('/skeen/frontend/v1/wacrm/notify', type='http', auth='none', methods=['POST', 'OPTIONS'], csrf=False) + def wacrm_notify(self, **kw): + """Aviso puntual del puente de Hermes: "llegó un mensaje a esta conversación". + + Da de alta solo esa conversación, su contacto y sus mensajes, para que + la bandeja se actualice en segundos sin esperar el ciclo. Es aditivo y + a prueba de fallos: si el aviso falla o se pierde, la sincronización + periódica cubre el mensaje igual. Se protege con llave compartida. + """ + if request.httprequest.method == 'OPTIONS': + return json_response({}, 200) + try: + esperada = os.environ.get('SKEEN_WACRM_NOTIFY_KEY') or \ + request.env['ir.config_parameter'].sudo().get_param(PARAM_NOTIFY_KEY) + if not esperada: + return json_response({'status': 'error', 'message': 'Aviso puntual deshabilitado'}, 503) + if not hmac.compare_digest(request.httprequest.headers.get('X-Skeen-Key') or '', esperada): + return json_response({'status': 'error', 'message': 'Llave inválida'}, 401) + external_id = (_parse_json_body().get('conversation_id') or '').strip() + if not external_id: + return json_response({'status': 'error', 'message': 'Falta conversation_id'}, 400) + try: + request.env.cr.execute( + "SELECT pg_try_advisory_xact_lock(hashtext('skeen_wacrm_sync'))") + libre = request.env.cr.fetchone()[0] + except Exception: + libre = True + if not libre: + return json_response({'status': 'busy'}) + return json_response({'status': 'success', 'result': self._sync_conversacion(external_id)}) + except Exception as e: + _logger.exception('Error en el aviso puntual de WACRM') + return json_response({'status': 'error', 'message': str(e)}, 500) + # ============================================================ # Mensajes # ============================================================ @@ -522,6 +820,7 @@ class SkeenWacrmProxyController(http.Controller): return json_response({ 'status': 'success', 'conversations': [self._conversation_to_dict(c) for c in conversations], + 'last_sync_at': request.env['ir.config_parameter'].sudo().get_param(PARAM_LAST_SYNC_AT), }) except Exception as e: _logger.exception('Error listando conversaciones WACRM')