import { Model } from "objection"; import { config } from "../config/index.js"; import { Atendimento } from "../models/Atendimento.model.js"; import { AtendimentoCliente } from "../models/AtendimentoCliente.model.js"; import { AtendimentoMensagem } from "../models/AtendimentoMensagem.model.js"; import { atendimentoPayloadSchema, protocolosAbertosSchema, protocoloMensagensSchema } from "../middleware/schemas/Atendimento.Schema.js"; import { assertPublicHttpsUrl } from "./ingestService.js"; import { BadRequestError } from "../shared/errors/index.js"; const MAX_REDIRECTS = 5; const MENSAGENS_BATCH = 500; const SYNC_CONCURRENCY = 4; function ifbotAuthHeaders() { const { token, authHeader } = config.ifbot; if (!token) return {}; const value = authHeader === "Authorization" && !token.includes(" ") ? `Bearer ${token}` : token; return { [authHeader]: value }; } async function fetchJson(urlStr, extraHeaders = {}) { let currentUrl = urlStr; let res; for (let hop = 0; ; hop += 1) { assertPublicHttpsUrl(currentUrl); res = await fetch(currentUrl, { headers: { "User-Agent": "Mozilla/5.0 star-oraculo/1.0", Accept: "application/json", ...extraHeaders }, redirect: "manual", signal: AbortSignal.timeout(15_000) }); if (res.status < 300 || res.status >= 400) break; const location = res.headers.get("location"); if (!location || hop >= MAX_REDIRECTS) { const err = new Error(`url_fetch_error:${res.status}`); err.statusCode = 502; throw err; } currentUrl = new URL(location, currentUrl).toString(); } if (!res.ok) { const err = new Error(`url_fetch_error:${res.status}`); err.statusCode = 502; throw err; } try { return await res.json(); } catch { const err = new Error("url_invalid_json"); err.statusCode = 502; throw err; } } export async function fetchAtendimentoJson(urlStr) { const json = await fetchJson(urlStr); const parsed = atendimentoPayloadSchema.safeParse(json); if (!parsed.success) { throw new BadRequestError("invalid_atendimento_payload"); } return parsed.data; } // prefixo do código do protocolo identifica o setor (ex.: SUP0000014690/2026 → SUP) export function setorFromCodigo(codigo) { const m = /^([A-Z]{3})\d/.exec(String(codigo ?? "")); return m ? m[1] : null; } function toDate(value) { if (!value) return null; const d = new Date(value); return Number.isNaN(d.getTime()) ? null : d; } export async function salvarAtendimento({ payload, setor, sourceUrl, abertura, telefone }) { const now = new Date(); const trx = await Model.startTransaction(); try { const c = payload.Cliente; const clienteMerge = ["Nome", "Foto", "Wid", "Ultima2", "Ultimacliente2", "Deadmensage2", "UpdatedAt"]; const clienteTelefone = telefone ?? c.Telefone ?? null; if (clienteTelefone) clienteMerge.push("Telefone"); await AtendimentoCliente.query(trx) .insert({ Id: c.Id, Nome: c.Nome, Foto: c.Foto ?? null, Wid: c.Wid ?? null, Telefone: clienteTelefone, Ultima2: c.Ultima2 ?? null, Ultimacliente2: c.Ultimacliente2 ?? null, Deadmensage2: c.Deadmensage2 ?? null, UpdatedAt: now }) .onConflict("Id") .merge(clienteMerge); const p = payload.Protocolo; const atendimentoMerge = ["Codigo", "Status", "ClienteId", "IngestedAt", "UpdatedAt"]; if (setor) atendimentoMerge.push("Setor"); if (sourceUrl) atendimentoMerge.push("SourceUrl"); if (abertura) atendimentoMerge.push("Abertura"); await Atendimento.query(trx) .insert({ Id: p.Id, Codigo: p.Codigo, Status: p.Status, ClienteId: c.Id, Setor: setor ?? null, SourceUrl: sourceUrl ?? null, Abertura: toDate(abertura), IngestedAt: now, UpdatedAt: now }) .onConflict("Id") .merge(atendimentoMerge); const mensagens = (payload.Mensagens ?? []).map((m) => ({ Id: m.Id, AtendimentoId: p.Id, Body: m.Body ?? null, Resposta: m.Resposta ?? null, Timestamp: toDate(m.Timestamp), Tipodemidia: m.Tipodemidia ?? null, Midia: m.Midia ?? null, Autor: m.Autor ?? null, Citacao: m.Citacao ?? null })); // batch insert via knex: o Objection não suporta insert em lote no MySQL for (let i = 0; i < mensagens.length; i += MENSAGENS_BATCH) { await trx(AtendimentoMensagem.tableName) .insert(mensagens.slice(i, i + MENSAGENS_BATCH)) .onConflict("Id") .merge(["AtendimentoId", "Body", "Resposta", "Timestamp", "Tipodemidia", "Midia", "Autor", "Citacao"]); } await trx.commit(); return { protocoloId: p.Id, codigo: p.Codigo, clienteId: c.Id, mensagens: mensagens.length }; } catch (err) { await trx.rollback(); throw err; } } async function importarProtocolo(item, baseUrl) { const mensagensUrl = `${baseUrl}/protocolos/${item.Id}/mensagens`; const json = await fetchJson(mensagensUrl, ifbotAuthHeaders()); // resposta completa (Protocolo + Cliente + Mensagens) ou só a lista de mensagens let payload; const full = atendimentoPayloadSchema.safeParse(json); if (full.success) { payload = full.data; } else { const soMensagens = protocoloMensagensSchema.safeParse(json); if (!soMensagens.success) throw new BadRequestError("invalid_mensagens_payload"); payload = { Protocolo: { Id: item.Id, Codigo: item.Codigo, Status: 1 }, Cliente: item.Cliente, Mensagens: soMensagens.data.Mensagens }; } return salvarAtendimento({ payload, setor: setorFromCodigo(item.Codigo), sourceUrl: mensagensUrl, abertura: item.Abertura ?? null, telefone: item.Cliente?.Telefone ?? null }); } export async function sincronizarProtocolosAbertos({ pagina = 1, limite = 50, inicio, fim } = {}) { const baseUrl = config.ifbot.baseUrl.replace(/\/+$/, ""); const params = new URLSearchParams({ pagina: String(pagina), limite: String(limite) }); if (inicio) params.set("inicio", inicio); if (fim) params.set("fim", fim); const listaJson = await fetchJson(`${baseUrl}/protocolos/abertos?${params.toString()}`, ifbotAuthHeaders()); const lista = protocolosAbertosSchema.safeParse(listaJson); if (!lista.success) throw new BadRequestError("invalid_protocolos_payload"); const protocolos = lista.data.Protocolos; const erros = []; let importados = 0; let mensagens = 0; for (let i = 0; i < protocolos.length; i += SYNC_CONCURRENCY) { const chunk = protocolos.slice(i, i + SYNC_CONCURRENCY); const resultados = await Promise.allSettled(chunk.map((item) => importarProtocolo(item, baseUrl))); resultados.forEach((r, idx) => { if (r.status === "fulfilled") { importados += 1; mensagens += r.value.mensagens; } else { erros.push({ protocoloId: chunk[idx].Id, codigo: chunk[idx].Codigo, erro: r.reason?.message ?? "erro_desconhecido" }); } }); } return { total: lista.data.Total ?? protocolos.length, pagina, limite, processados: protocolos.length, importados, mensagens, erros }; } let fullSyncEmAndamento = false; export async function sincronizarTodosProtocolosAbertos({ limite = 200, maxPaginas = 50 } = {}) { if (fullSyncEmAndamento) { const err = new Error("sync_em_andamento"); err.statusCode = 409; throw err; } fullSyncEmAndamento = true; try { const agregado = { total: 0, paginas: 0, processados: 0, importados: 0, mensagens: 0, erros: [] }; for (let pagina = 1; pagina <= maxPaginas; pagina += 1) { const r = await sincronizarProtocolosAbertos({ pagina, limite }); agregado.total = r.total; agregado.paginas = pagina; agregado.processados += r.processados; agregado.importados += r.importados; agregado.mensagens += r.mensagens; agregado.erros.push(...r.erros); if (r.processados < limite || agregado.processados >= agregado.total) break; } return agregado; } finally { fullSyncEmAndamento = false; } } export async function listarAtendimentos({ page = 1, pageSize = 20, setor, status } = {}) { const query = Atendimento.query() .withGraphFetched("cliente") .orderBy("IngestedAt", "desc") .page(page - 1, pageSize); if (setor) query.where("Setor", setor); if (status !== undefined) query.where("Status", status); const { results, total } = await query; return { results, total, page, pageSize }; } export async function obterAtendimento(id) { return Atendimento.query() .findById(id) .withGraphFetched("[cliente, mensagens]") .modifyGraph("mensagens", (q) => q.orderBy("Timestamp", "asc").orderBy("Id", "asc")); }