- {items.map((item) => {
- const picked = resolutions[item.contactId];
+
+
+ {item.currentEmail}
+ {item.contactName && · {item.contactName}}
+ {item.candidatesTotal > item.candidates.length && (
+
+ {' '}
+ (
+ {t('superAdmin.accounts.import.reconcile.candidatesShown', {
+ shown: item.candidates.length,
+ total: item.candidatesTotal,
+ })}
+ )
+
+ )}
+
+
+ {item.candidates.map((candidate) => {
+ const pickedBy = pickedEmails.get(candidate.csvEmail.toLowerCase());
+ // An email reconciles ONE contact: blocked when already applied to
+ // another contact (server-computed) or picked on another item of
+ // this page (unsaved local decision).
+ const blocked = candidate.usedByContactId !== undefined || (pickedBy !== undefined && pickedBy !== item.contactId);
return (
-
-
- {item.currentEmail}
- {item.contactName && (
- · {item.contactName}
- )}
-
-
- {item.candidates.map((c) => (
-
- ))}
-
+ {picked === candidate.csvRowNumber &&
{t('superAdmin.accounts.import.reconcile.picked')}}
+
);
})}
+
+
+
+ );
+}
+
+const ITEMS_PAGE_SIZE = 25;
+
+// Flat, searchable "who matched what" table over every session item — auto
+// picks and ambiguous outcomes alike. Read-only: resolution stays in the
+// ambiguous queue above; this section exists for visibility and lookup.
+function ItemsSection({ jobId }: { jobId: string }) {
+ const { t } = useTranslation();
+ const [search, setSearch] = useState('');
+ const [q, setQ] = useState('');
+ const [kind, setKind] = useState<'all' | 'auto' | 'ambiguous'>('all');
+ const [status, setStatus] = useState<'all' | 'pending' | 'applied' | 'skipped' | 'failed'>('all');
+ const [offset, setOffset] = useState(0);
+
+ useEffect(() => {
+ const id = setTimeout(() => {
+ setQ(search.trim());
+ setOffset(0);
+ }, 400);
+ return () => clearTimeout(id);
+ }, [search]);
+
+ const pageQuery = useQuery({
+ queryKey: ['reconcile-items', jobId, offset, q, kind, status],
+ queryFn: () =>
+ reconcileGateway.itemsPage(jobId, {
+ offset,
+ limit: ITEMS_PAGE_SIZE,
+ q: q || undefined,
+ kind: kind === 'all' ? undefined : kind,
+ status: status === 'all' ? undefined : status,
+ }),
+ });
+
+ const total = pageQuery.data?.total ?? 0;
+ const items = pageQuery.data?.items ?? [];
+
+ return (
+
+
+
{t('superAdmin.accounts.import.reconcile.itemsTitle')}
+
{t('superAdmin.accounts.import.reconcile.itemsHint')}
+
+
+
+ setSearch(e.target.value)}
+ placeholder={t('superAdmin.accounts.import.reconcile.searchPlaceholder')}
+ className="h-8 min-w-48 flex-1"
+ />
+
+
-
+
+ {pageQuery.isLoading ? (
+
+ ) : items.length === 0 ? (
+
{t('superAdmin.accounts.import.reconcile.searchNoResults')}
+ ) : (
+
+
+
+
+ {t('superAdmin.accounts.import.reconcile.colContact')}
+ {t('superAdmin.accounts.import.reconcile.colCurrentEmail')}
+ {t('superAdmin.accounts.import.reconcile.colNewEmail')}
+ {t('superAdmin.accounts.import.reconcile.colType')}
+ {t('superAdmin.accounts.import.reconcile.colStatus')}
+
+
+
+ {items.map((item) => (
+
+
+ {item.contactName || #{item.contactId}}
+
+ {item.currentEmail}
+ {item.newEmail ?? '—'}
+
+ {item.kind === 'auto'
+ ? t('superAdmin.accounts.import.reconcile.kindAuto')
+ : t('superAdmin.accounts.import.reconcile.kindAmbiguous')}
+
+
+
+ {item.status === 'failed' && item.failureReason && (
+
+ {item.failureReason}
+
+ )}
+
+
+ ))}
+
+
+
+ )}
+
+
+
+
+ {t('superAdmin.accounts.import.reconcile.pageIndicator', {
+ from: Math.min(offset + 1, total).toLocaleString(),
+ to: Math.min(offset + ITEMS_PAGE_SIZE, total).toLocaleString(),
+ total: total.toLocaleString(),
+ })}
+
+
+
+
);
}
+
+function ItemStatusBadge({ status }: { status: ReconcileItemRow['status'] }) {
+ const { t } = useTranslation();
+ const variant = status === 'applied' ? 'default' : status === 'failed' ? 'destructive' : status === 'skipped' ? 'outline' : 'secondary';
+ const label =
+ status === 'applied'
+ ? t('superAdmin.accounts.import.reconcile.statusApplied')
+ : status === 'failed'
+ ? t('superAdmin.accounts.import.reconcile.statusFailed')
+ : status === 'skipped'
+ ? t('superAdmin.accounts.import.reconcile.statusSkipped')
+ : t('superAdmin.accounts.import.reconcile.statusPending');
+ return
{label};
+}
diff --git a/apps/frontend-react/src/features/super-admin/accounts/reconcile-gateway.ts b/apps/frontend-react/src/features/super-admin/accounts/reconcile-gateway.ts
index d7b6de65..b88bc61e 100644
--- a/apps/frontend-react/src/features/super-admin/accounts/reconcile-gateway.ts
+++ b/apps/frontend-react/src/features/super-admin/accounts/reconcile-gateway.ts
@@ -1,31 +1,77 @@
import { apiClient } from '@/lib/api-client';
-// EVO-1464 workaround — types must match
+// Types must match
// apps/msgops-api/src/modules/enterprise-import/email-reconcile.types.ts
export interface AmbiguousCandidate {
csvRowNumber: number;
csvName: string;
csvEmail: string;
+ // Name-similarity score (0..1) — candidates arrive sorted by it, best first.
+ score: number;
+ // created_at agreement with the contact: 2 = exact instant, 1 = same date,
+ // 0 = none. Absent on sessions created before the field existed.
+ timeMatch?: number;
+ // Present when this candidate's email was already applied to another
+ // contact in this session — picking it would collide with the per-account
+ // email uniqueness, so the UI disables it.
+ usedByContactId?: number;
}
export interface AmbiguousMatch {
contactId: number;
currentEmail: string;
contactName: string;
+ // Top candidates only ("showing candidates.length of candidatesTotal").
candidates: AmbiguousCandidate[];
+ candidatesTotal: number;
}
-export interface ReconcilePreview {
+export interface ReconcileSessionProgress {
+ jobId: string;
csvRows: number;
invalidCsvRows: number;
contactsMasked: number;
- uniqueMatches: number;
- ambiguousMatches: number;
- noMatches: number;
alreadyClean: number;
- ambiguousSample: AmbiguousMatch[];
+ noMatches: number;
noMatchSample: Array<{ contactId: number; currentEmail: string }>;
+ auto: { total: number; applied: number; failed: number; pending: number };
+ ambiguous: { total: number; applied: number; skipped: number; pending: number; failed: number };
+ createdAt: string;
+ updatedAt: string;
+}
+
+export interface AmbiguousPage {
+ totalPending: number;
+ offset: number;
+ items: AmbiguousMatch[];
+}
+
+// One row of the session items table — "who matched what".
+export interface ReconcileItemRow {
+ contactId: number;
+ contactName: string;
+ currentEmail: string;
+ kind: 'auto' | 'ambiguous';
+ status: 'pending' | 'applied' | 'skipped' | 'failed';
+ newEmail: string | null;
+ csvRowNumber: number | null;
+ failureReason: string | null;
+ candidatesTotal: number | null;
+}
+
+export interface ReconcileItemsPage {
+ total: number;
+ offset: number;
+ items: ReconcileItemRow[];
+}
+
+export interface ReconcileItemsQuery {
+ offset: number;
+ limit: number;
+ q?: string;
+ kind?: 'auto' | 'ambiguous';
+ status?: 'pending' | 'applied' | 'skipped' | 'failed';
}
export interface ApplyResolution {
@@ -34,21 +80,80 @@ export interface ApplyResolution {
csvRowNumber: number | null;
}
-export interface ApplyResult {
- updated: number;
- skippedAmbiguous: number;
- skippedNoMatch: number;
+export interface ResolveBatchResult {
+ applied: number;
+ skipped: number;
+ invalid: number;
failures: Array<{ contactId: number; reason: string }>;
+ progress: ReconcileSessionProgress;
+}
+
+export interface ApplyAutoChunkResult {
+ applied: number;
+ failed: number;
+ remaining: number;
+}
+
+export interface BulkResolveResult {
+ resolved: number;
+ unresolved: number;
+ nextAfterId: string | null;
+ remainingPending: number;
}
+// Batched reconciliation over a server-side session: the CSV uploads ONCE
+// (createSession) and every later step is an incremental, quantified batch.
export const reconcileGateway = {
- async preview(jobId: string, csv: string): Promise
{
- const { data } = await apiClient.post(`/imports/${jobId}/reconcile/preview`, { csv });
+ async getSession(jobId: string): Promise {
+ try {
+ const { data } = await apiClient.get(`/imports/${jobId}/reconcile/session`);
+ return data;
+ } catch (err: any) {
+ if (err?.response?.status === 404) return null;
+ throw err;
+ }
+ },
+
+ async createSession(jobId: string, csv: string, ignoreColumns: string[] = []): Promise {
+ // Parsing + matching + persisting 350k contacts takes a while — disable
+ // the client timeout and let the server/nginx budget govern.
+ const { data } = await apiClient.post(`/imports/${jobId}/reconcile/session`, { csv, ignoreColumns }, { timeout: 0 });
+ return data;
+ },
+
+ async ambiguousPage(jobId: string, offset: number, limit: number, q?: string): Promise {
+ const { data } = await apiClient.get(`/imports/${jobId}/reconcile/session/ambiguous`, {
+ params: { offset, limit, ...(q ? { q } : {}) },
+ });
+ return data;
+ },
+
+ async itemsPage(jobId: string, query: ReconcileItemsQuery): Promise {
+ const { data } = await apiClient.get(`/imports/${jobId}/reconcile/session/items`, {
+ params: query,
+ });
return data;
},
- async apply(jobId: string, csv: string, resolutions: ApplyResolution[]): Promise {
- const { data } = await apiClient.post(`/imports/${jobId}/reconcile/apply`, { csv, resolutions });
+ async resolve(jobId: string, resolutions: ApplyResolution[]): Promise {
+ const { data } = await apiClient.post(`/imports/${jobId}/reconcile/session/resolve`, { resolutions });
return data;
},
+
+ async applyAuto(jobId: string, limit: number): Promise {
+ const { data } = await apiClient.post(`/imports/${jobId}/reconcile/session/apply-auto`, { limit }, { timeout: 0 });
+ return data;
+ },
+
+ async bulkResolve(
+ jobId: string,
+ payload: { strategy: 'best-name'; threshold: number; limit: number; afterId?: string } | { strategy: 'skip-remaining' },
+ ): Promise {
+ const { data } = await apiClient.post(`/imports/${jobId}/reconcile/session/bulk-resolve`, payload, { timeout: 0 });
+ return data;
+ },
+
+ async deleteSession(jobId: string): Promise {
+ await apiClient.delete(`/imports/${jobId}/reconcile/session`);
+ },
};
diff --git a/apps/frontend-react/src/locales/en-US.json b/apps/frontend-react/src/locales/en-US.json
index 0eef8241..f80f4cf7 100644
--- a/apps/frontend-react/src/locales/en-US.json
+++ b/apps/frontend-react/src/locales/en-US.json
@@ -2496,6 +2496,8 @@
"applyError": "Failed to apply reconciliation",
"applyDoneToast": "{{updated}} contacts updated · {{ambiguous}} ambiguous · {{noMatch}} no match",
"fileReadError": "Failed to read the CSV file",
+ "fileTooLarge": "The file is {{size}} MB, which exceeds the {{max}} MB limit. Split the CSV into smaller parts and reconcile in batches.",
+ "payloadTooLarge": "The server rejected the upload: the file is too large ({{max}} MB limit). Split the CSV into smaller parts and reconcile in batches.",
"statCsvRows": "CSV rows",
"statContactsMasked": "Masked contacts",
"statUniqueMatches": "Unique matches",
@@ -2507,7 +2509,74 @@
"ambiguousShownLimited": "Showing {{shown}} of {{total}} ambiguous cases. Apply and re-preview to see more.",
"picked": "Picked",
"skipThis": "Skip this (leave masked)",
- "skipped": "Skipped"
+ "skipped": "Skipped",
+ "processCsv": "Process CSV",
+ "processingCsv": "Processing CSV…",
+ "uploadHint": "The CSV is processed once. From then on, progress is stored server-side — you can leave the page and pick up where you left off without re-uploading the file.",
+ "sessionCreateError": "Failed to process the CSV",
+ "sessionError": "Failed to load the reconciliation session",
+ "statAuto": "Automatic fixes",
+ "autoTitle": "Automatic fixes",
+ "autoProgress": "{{applied}} of {{total}} applied",
+ "autoFailed": "{{failed}} failures",
+ "autoDone": "All automatic fixes have been applied.",
+ "chunkSize": "Batch",
+ "applyAutoStart": "Apply automatic",
+ "applyAutoStop": "Pause",
+ "bulkTitle": "Bulk actions — ambiguous",
+ "bulkThreshold": "Minimum confidence",
+ "thresholdHigh": "High (80% by name)",
+ "thresholdMedium": "Medium (60% by name)",
+ "thresholdLow": "Low (50% by name)",
+ "bulkBestName": "Auto-resolve by name",
+ "bulkBestNameRunning": "Resolving…",
+ "bulkBestNameDone": "{{resolved}} auto-resolved · {{unresolved}} still pending",
+ "skipRemaining": "Skip all remaining",
+ "skipRemainingConfirm": "Mark all {{count}} pending ambiguous cases as skipped? The contacts keep their masked email.",
+ "skipRemainingDone": "{{count}} contacts skipped",
+ "bulkHint": "Auto-resolve applies the candidate with the highest name similarity when it clears the chosen confidence and there is no tie. The rest stay in the manual queue.",
+ "ambiguousAllDone": "All ambiguous cases handled: {{applied}} applied · {{skipped}} skipped.",
+ "ambiguousPendingHeader": "{{pending}} pending of {{total}} ambiguous cases. Resolve per page or use the bulk actions.",
+ "pageSize": "Per page",
+ "prevPage": "Previous",
+ "nextPage": "Next",
+ "pageIndicator": "{{from}}–{{to}} of {{total}}",
+ "candidatesShown": "showing {{shown}} of {{total}} candidates",
+ "score": "similarity {{pct}}%",
+ "saveDecisions": "Save decisions ({{count}})",
+ "decisionsSaved": "{{applied}} applied · {{skipped}} skipped",
+ "discardSession": "Discard session",
+ "discardConfirm": "Discard the reconciliation session? Fixes already applied to contacts remain — only the work queue is removed.",
+ "columnsDetected": "Columns detected in the file",
+ "columnRequired": "required",
+ "columnOptional": "optional",
+ "columnsMissingTitle": "Invalid file",
+ "columnsMissing": "The file is missing required columns: {{columns}}. Export the CSV from BMS with these columns and try again.",
+ "columnsIgnoredHint": "Required columns drive the match (email = key; created_at and the name break collision ties). The name may come as a name column or as the first_name + last_name pair. Unchecked optional columns are ignored during processing.",
+ "columnNameGroup": "name (or first_name + last_name)",
+ "missingColumnsError": "The CSV is missing required columns: {{columns}}",
+ "timeExact": "exact date/time",
+ "timeSameDay": "same date",
+ "decisionsFailed": "{{count}} not applied — {{reason}}",
+ "usedByContact": "already used by contact #{{id}}",
+ "pickedElsewhere": "picked for another contact",
+ "searchPlaceholder": "Search by name or email…",
+ "searchNoResults": "No items found for this search.",
+ "itemsTitle": "Session items",
+ "itemsHint": "Who matched what — auto and ambiguous matches with their apply status. Resolution stays in the queue above.",
+ "kindAll": "All kinds",
+ "kindAuto": "Automatic",
+ "kindAmbiguous": "Ambiguous",
+ "statusAll": "All statuses",
+ "statusPending": "Pending",
+ "statusApplied": "Applied",
+ "statusSkipped": "Skipped",
+ "statusFailed": "Failed",
+ "colContact": "Contact",
+ "colCurrentEmail": "Current email",
+ "colNewEmail": "New email",
+ "colType": "Kind",
+ "colStatus": "Status"
},
"progressSkipped": "skipped ({{reason}})",
"progressDone": "done",
diff --git a/apps/frontend-react/src/locales/es-ES.json b/apps/frontend-react/src/locales/es-ES.json
index 6c0bb2bb..9a1baf04 100644
--- a/apps/frontend-react/src/locales/es-ES.json
+++ b/apps/frontend-react/src/locales/es-ES.json
@@ -2524,6 +2524,8 @@
"applyError": "Error al aplicar la reconciliación",
"applyDoneToast": "{{updated}} contactos actualizados · {{ambiguous}} ambiguos · {{noMatch}} sin coincidencia",
"fileReadError": "Error al leer el archivo CSV",
+ "fileTooLarge": "El archivo pesa {{size}} MB y supera el límite de {{max}} MB. Divide el CSV en partes más pequeñas y reconcilia por etapas.",
+ "payloadTooLarge": "El servidor rechazó el envío: el archivo es demasiado grande (límite de {{max}} MB). Divide el CSV en partes más pequeñas y reconcilia por etapas.",
"statCsvRows": "Filas en el CSV",
"statContactsMasked": "Contactos enmascarados",
"statUniqueMatches": "Coincidencia única",
@@ -2535,7 +2537,74 @@
"ambiguousShownLimited": "Mostrando {{shown}} de {{total}} casos ambiguos. Aplique y rehaga la previsualización para ver los próximos.",
"picked": "Elegido",
"skipThis": "Saltar este (dejar enmascarado)",
- "skipped": "Saltado"
+ "skipped": "Saltado",
+ "processCsv": "Procesar CSV",
+ "processingCsv": "Procesando CSV…",
+ "uploadHint": "El CSV se procesa una sola vez. A partir de ahí, el progreso queda guardado en el servidor — puedes salir de la página y continuar donde lo dejaste sin volver a subir el archivo.",
+ "sessionCreateError": "Error al procesar el CSV",
+ "sessionError": "Error al cargar la sesión de reconciliación",
+ "statAuto": "Correcciones automáticas",
+ "autoTitle": "Correcciones automáticas",
+ "autoProgress": "{{applied}} de {{total}} aplicadas",
+ "autoFailed": "{{failed}} fallos",
+ "autoDone": "Todas las correcciones automáticas fueron aplicadas.",
+ "chunkSize": "Lote",
+ "applyAutoStart": "Aplicar automáticas",
+ "applyAutoStop": "Pausar",
+ "bulkTitle": "Acciones masivas — ambiguos",
+ "bulkThreshold": "Confianza mínima",
+ "thresholdHigh": "Alta (80% por nombre)",
+ "thresholdMedium": "Media (60% por nombre)",
+ "thresholdLow": "Baja (50% por nombre)",
+ "bulkBestName": "Auto-resolver por nombre",
+ "bulkBestNameRunning": "Resolviendo…",
+ "bulkBestNameDone": "{{resolved}} resueltos automáticamente · {{unresolved}} siguen pendientes",
+ "skipRemaining": "Saltar todos los restantes",
+ "skipRemainingConfirm": "¿Marcar los {{count}} casos ambiguos pendientes como saltados? Los contactos mantienen el email enmascarado.",
+ "skipRemainingDone": "{{count}} contactos saltados",
+ "bulkHint": "Auto-resolver aplica el candidato con mayor similitud de nombre cuando supera la confianza elegida y no hay empate. El resto sigue en la cola manual.",
+ "ambiguousAllDone": "Todos los casos ambiguos fueron tratados: {{applied}} aplicados · {{skipped}} saltados.",
+ "ambiguousPendingHeader": "{{pending}} pendientes de {{total}} casos ambiguos. Resuelve por página o usa las acciones masivas.",
+ "pageSize": "Por página",
+ "prevPage": "Anterior",
+ "nextPage": "Siguiente",
+ "pageIndicator": "{{from}}–{{to}} de {{total}}",
+ "candidatesShown": "mostrando {{shown}} de {{total}} candidatos",
+ "score": "similitud {{pct}}%",
+ "saveDecisions": "Guardar decisiones ({{count}})",
+ "decisionsSaved": "{{applied}} aplicados · {{skipped}} saltados",
+ "discardSession": "Descartar sesión",
+ "discardConfirm": "¿Descartar la sesión de reconciliación? Las correcciones ya aplicadas a los contactos permanecen — solo se elimina la cola de trabajo.",
+ "columnsDetected": "Columnas detectadas en el archivo",
+ "columnRequired": "obligatoria",
+ "columnOptional": "opcional",
+ "columnsMissingTitle": "Archivo inválido",
+ "columnsMissing": "Faltan columnas obligatorias en el archivo: {{columns}}. Exporta el CSV desde BMS con esas columnas e inténtalo de nuevo.",
+ "columnsIgnoredHint": "Las columnas obligatorias impulsan el match (email = clave; created_at y el nombre desempatan colisiones). El nombre puede venir como columna name o como el par first_name + last_name. Las columnas opcionales desmarcadas se ignoran durante el procesamiento.",
+ "columnNameGroup": "name (o first_name + last_name)",
+ "missingColumnsError": "Al CSV le faltan las columnas obligatorias: {{columns}}",
+ "timeExact": "fecha/hora idéntica",
+ "timeSameDay": "misma fecha",
+ "decisionsFailed": "{{count}} no aplicadas — {{reason}}",
+ "usedByContact": "ya usado por el contacto #{{id}}",
+ "pickedElsewhere": "elegido para otro contacto",
+ "searchPlaceholder": "Buscar por nombre o email…",
+ "searchNoResults": "No se encontraron elementos para esta búsqueda.",
+ "itemsTitle": "Elementos de la sesión",
+ "itemsHint": "Quién coincidió con qué — coincidencias automáticas y ambiguas con su estado de aplicación. La resolución sigue en la cola de arriba.",
+ "kindAll": "Todos los tipos",
+ "kindAuto": "Automático",
+ "kindAmbiguous": "Ambiguo",
+ "statusAll": "Todos los estados",
+ "statusPending": "Pendiente",
+ "statusApplied": "Aplicado",
+ "statusSkipped": "Saltado",
+ "statusFailed": "Falló",
+ "colContact": "Contacto",
+ "colCurrentEmail": "Email actual",
+ "colNewEmail": "Email nuevo",
+ "colType": "Tipo",
+ "colStatus": "Estado"
}
}
},
diff --git a/apps/frontend-react/src/locales/pt-BR.json b/apps/frontend-react/src/locales/pt-BR.json
index 7e8c2aa9..d196c093 100644
--- a/apps/frontend-react/src/locales/pt-BR.json
+++ b/apps/frontend-react/src/locales/pt-BR.json
@@ -2496,6 +2496,8 @@
"applyError": "Falha ao aplicar reconciliação",
"applyDoneToast": "{{updated}} contatos atualizados · {{ambiguous}} ambíguos · {{noMatch}} sem match",
"fileReadError": "Falha ao ler o arquivo CSV",
+ "fileTooLarge": "O arquivo tem {{size}} MB e excede o limite de {{max}} MB. Divida o CSV em partes menores e reconcilie em etapas.",
+ "payloadTooLarge": "O servidor recusou o envio: o arquivo é grande demais (limite de {{max}} MB). Divida o CSV em partes menores e reconcilie em etapas.",
"statCsvRows": "Linhas no CSV",
"statContactsMasked": "Contatos mascarados",
"statUniqueMatches": "Match único",
@@ -2507,7 +2509,74 @@
"ambiguousShownLimited": "Mostrando {{shown}} de {{total}} casos ambíguos. Aplique e refaça a pré-visualização pra ver os próximos.",
"picked": "Escolhido",
"skipThis": "Pular este (deixar mascarado)",
- "skipped": "Pulado"
+ "skipped": "Pulado",
+ "processCsv": "Processar CSV",
+ "processingCsv": "Processando CSV…",
+ "uploadHint": "O CSV é processado uma única vez. Depois disso, o progresso fica salvo no servidor — dá pra sair da página e continuar de onde parou, sem reenviar o arquivo.",
+ "sessionCreateError": "Falha ao processar o CSV",
+ "sessionError": "Falha ao carregar a sessão de reconciliação",
+ "statAuto": "Correções automáticas",
+ "autoTitle": "Correções automáticas",
+ "autoProgress": "{{applied}} de {{total}} aplicadas",
+ "autoFailed": "{{failed}} falhas",
+ "autoDone": "Todas as correções automáticas foram aplicadas.",
+ "chunkSize": "Lote",
+ "applyAutoStart": "Aplicar automáticas",
+ "applyAutoStop": "Pausar",
+ "bulkTitle": "Ações em massa — ambíguos",
+ "bulkThreshold": "Confiança mínima",
+ "thresholdHigh": "Alta (80% por nome)",
+ "thresholdMedium": "Média (60% por nome)",
+ "thresholdLow": "Baixa (50% por nome)",
+ "bulkBestName": "Auto-resolver por nome",
+ "bulkBestNameRunning": "Resolvendo…",
+ "bulkBestNameDone": "{{resolved}} resolvidos automaticamente · {{unresolved}} continuam pendentes",
+ "skipRemaining": "Pular todos os restantes",
+ "skipRemainingConfirm": "Marcar todos os {{count}} casos ambíguos pendentes como pulados? Os contatos continuam com o email mascarado.",
+ "skipRemainingDone": "{{count}} contatos pulados",
+ "bulkHint": "Auto-resolver aplica o candidato com maior similaridade de nome quando ele supera a confiança escolhida e não há empate. Os que não passam continuam na fila manual.",
+ "ambiguousAllDone": "Todos os casos ambíguos foram tratados: {{applied}} aplicados · {{skipped}} pulados.",
+ "ambiguousPendingHeader": "{{pending}} pendentes de {{total}} casos ambíguos. Resolva por página ou use as ações em massa.",
+ "pageSize": "Por página",
+ "prevPage": "Anterior",
+ "nextPage": "Próxima",
+ "pageIndicator": "{{from}}–{{to}} de {{total}}",
+ "candidatesShown": "mostrando {{shown}} de {{total}} candidatos",
+ "score": "similaridade {{pct}}%",
+ "saveDecisions": "Salvar decisões ({{count}})",
+ "decisionsSaved": "{{applied}} aplicados · {{skipped}} pulados",
+ "discardSession": "Descartar sessão",
+ "discardConfirm": "Descartar a sessão de reconciliação? As correções já aplicadas nos contatos permanecem — apenas a fila de trabalho é removida.",
+ "columnsDetected": "Colunas detectadas no arquivo",
+ "columnRequired": "obrigatória",
+ "columnOptional": "opcional",
+ "columnsMissingTitle": "Arquivo inválido",
+ "columnsMissing": "Faltam colunas obrigatórias no arquivo: {{columns}}. Exporte o CSV do BMS com essas colunas e tente de novo.",
+ "columnsIgnoredHint": "As colunas obrigatórias são usadas no match (email = chave; created_at e o nome desempatam colisões). O nome pode vir como coluna name ou como o par first_name + last_name. Colunas opcionais desmarcadas são ignoradas no processamento.",
+ "columnNameGroup": "name (ou first_name + last_name)",
+ "missingColumnsError": "O CSV não tem as colunas obrigatórias: {{columns}}",
+ "timeExact": "data/hora idêntica",
+ "timeSameDay": "mesma data",
+ "decisionsFailed": "{{count}} não aplicadas — {{reason}}",
+ "usedByContact": "já usado pelo contato #{{id}}",
+ "pickedElsewhere": "escolhido para outro contato",
+ "searchPlaceholder": "Buscar por nome ou email…",
+ "searchNoResults": "Nenhum item encontrado para essa busca.",
+ "itemsTitle": "Itens da sessão",
+ "itemsHint": "Quem casou com o quê — matches automáticos e ambíguos, com status de aplicação. A resolução continua na fila acima.",
+ "kindAll": "Todos os tipos",
+ "kindAuto": "Automático",
+ "kindAmbiguous": "Ambíguo",
+ "statusAll": "Todos os status",
+ "statusPending": "Pendente",
+ "statusApplied": "Aplicado",
+ "statusSkipped": "Pulado",
+ "statusFailed": "Falhou",
+ "colContact": "Contato",
+ "colCurrentEmail": "Email atual",
+ "colNewEmail": "Email novo",
+ "colType": "Tipo",
+ "colStatus": "Status"
},
"progressSkipped": "pulado ({{reason}})",
"progressDone": "concluído",
diff --git a/apps/msgops-api/src/entities/email-reconcile-item.entity.ts b/apps/msgops-api/src/entities/email-reconcile-item.entity.ts
new file mode 100644
index 00000000..9aeee299
--- /dev/null
+++ b/apps/msgops-api/src/entities/email-reconcile-item.entity.ts
@@ -0,0 +1,68 @@
+import { Column, Entity, Index, PrimaryGeneratedColumn } from 'typeorm';
+
+export type EmailReconcileItemKind = 'auto' | 'ambiguous';
+export type EmailReconcileItemStatus = 'pending' | 'applied' | 'skipped' | 'failed';
+
+export interface EmailReconcileStoredCandidate {
+ csvRowNumber: number;
+ csvName: string;
+ csvEmail: string;
+ // Jaccard token similarity between the contact name and csvName (0..1).
+ // Candidates are stored sorted by it, best first, so bulk best-name
+ // resolution reads candidates[0] without recomputing.
+ score: number;
+ // created_at agreement with the contact (2 exact, 1 same day, 0 none).
+ // Absent on sessions persisted before the field existed.
+ timeMatch?: number;
+}
+
+// One row per masked contact the reconcile session has an outcome for.
+// kind=auto → unique/confident match; new_email already decided.
+// kind=ambiguous → operator (or bulk strategy) must pick a candidate.
+// Status walks pending → applied|skipped|failed. Contacts with no CSV match
+// get no item — they are only counted on the session row.
+@Entity('email_reconcile_items')
+@Index('email_reconcile_items_job_kind_status_idx', ['jobId', 'kind', 'status'])
+@Index('email_reconcile_items_job_contact_uq', ['jobId', 'contactId'], { unique: true })
+export class EmailReconcileItemEntity {
+ @PrimaryGeneratedColumn('increment', { name: 'id', type: 'bigint' })
+ id: string;
+
+ @Column('uuid', { name: 'job_id' })
+ jobId: string;
+
+ @Column('integer', { name: 'contact_id' })
+ contactId: number;
+
+ @Column('varchar', { name: 'current_email', length: 255 })
+ currentEmail: string;
+
+ @Column('varchar', { name: 'contact_name', length: 255, nullable: true })
+ contactName: string | null;
+
+ @Column('varchar', { name: 'kind', length: 16 })
+ kind: EmailReconcileItemKind;
+
+ @Column('varchar', { name: 'status', length: 16, default: 'pending' })
+ status: EmailReconcileItemStatus;
+
+ // For kind=auto: filled at session create. For kind=ambiguous: filled when
+ // the item is resolved (operator pick or bulk strategy).
+ @Column('varchar', { name: 'new_email', length: 255, nullable: true })
+ newEmail: string | null;
+
+ @Column('integer', { name: 'csv_row_number', nullable: true })
+ csvRowNumber: number | null;
+
+ // kind=ambiguous only: top candidates sorted by score desc, capped — enough
+ // for the operator UI and for bulk best-name resolution.
+ @Column('jsonb', { name: 'candidates', nullable: true })
+ candidates: EmailReconcileStoredCandidate[] | null;
+
+ // Real candidate count before the cap, so the UI can say "showing 20 of N".
+ @Column('integer', { name: 'candidates_total', nullable: true })
+ candidatesTotal: number | null;
+
+ @Column('text', { name: 'failure_reason', nullable: true })
+ failureReason: string | null;
+}
diff --git a/apps/msgops-api/src/entities/email-reconcile-session.entity.ts b/apps/msgops-api/src/entities/email-reconcile-session.entity.ts
new file mode 100644
index 00000000..b2a04b67
--- /dev/null
+++ b/apps/msgops-api/src/entities/email-reconcile-session.entity.ts
@@ -0,0 +1,41 @@
+import { Column, CreateDateColumn, Entity, PrimaryColumn, UpdateDateColumn } from 'typeorm';
+
+// One reconcile working set per import job. The CSV is parsed and matched
+// ONCE (POST /imports/:jobId/reconcile/session); the outcome is persisted here
+// plus one row per masked contact in email_reconcile_items, so the operator
+// can resolve/apply in paginated batches without re-uploading the CSV.
+// Re-creating the session replaces the previous one (job_id is the PK).
+@Entity('email_reconcile_sessions')
+export class EmailReconcileSessionEntity {
+ @PrimaryColumn('uuid', { name: 'job_id' })
+ jobId: string;
+
+ @Column('integer', { name: 'account_id' })
+ accountId: number;
+
+ @Column('integer', { name: 'csv_rows' })
+ csvRows: number;
+
+ @Column('integer', { name: 'invalid_csv_rows' })
+ invalidCsvRows: number;
+
+ @Column('integer', { name: 'contacts_masked' })
+ contactsMasked: number;
+
+ @Column('integer', { name: 'already_clean' })
+ alreadyClean: number;
+
+ @Column('integer', { name: 'no_match_total' })
+ noMatchTotal: number;
+
+ // Small capped sample for display only — the full no-match set is derivable
+ // from contacts that stay masked after the session is exhausted.
+ @Column('jsonb', { name: 'no_match_sample', default: () => "'[]'" })
+ noMatchSample: Array<{ contactId: number; currentEmail: string }>;
+
+ @CreateDateColumn({ name: 'created_at', type: 'timestamp with time zone' })
+ createdAt: Date;
+
+ @UpdateDateColumn({ name: 'updated_at', type: 'timestamp with time zone' })
+ updatedAt: Date;
+}
diff --git a/apps/msgops-api/src/main.ts b/apps/msgops-api/src/main.ts
index dd03e420..2eb31f01 100644
--- a/apps/msgops-api/src/main.ts
+++ b/apps/msgops-api/src/main.ts
@@ -70,6 +70,21 @@ async function bootstrap() {
app.getHttpAdapter().getInstance().set('query parser', 'extended');
+ // The email-reconcile flow (/imports/:jobId/reconcile/*) receives whole CSV
+ // exports embedded in the JSON body — 350k contacts easily exceed 20mb — so
+ // these routes get their own higher limit instead of raising the global one,
+ // which also guards public webhook endpoints. Must stay registered BEFORE
+ // the global parser (body-parser skips requests already parsed). Keep in
+ // sync with client_max_body_size in apps/frontend-react/nginx.conf.
+ app.use(
+ '/imports',
+ bodyParser.json({
+ limit: '64mb',
+ verify: (req: any, _res, buf) => {
+ req.rawBody = buf;
+ },
+ }),
+ );
// Captures the raw body alongside the parsed JSON. Webhook controllers
// (e.g. POST /webhooks/meta and /webhooks/evolution-hub) need it to compute
// HMAC-SHA256 signatures from the exact bytes the sender hashed.
diff --git a/apps/msgops-api/src/migrations/1783000000000-create-email-reconcile-sessions.ts b/apps/msgops-api/src/migrations/1783000000000-create-email-reconcile-sessions.ts
new file mode 100644
index 00000000..7f6cf025
--- /dev/null
+++ b/apps/msgops-api/src/migrations/1783000000000-create-email-reconcile-sessions.ts
@@ -0,0 +1,75 @@
+import { MigrationInterface, QueryRunner, Table } from 'typeorm';
+
+/**
+ * Persisted reconcile working set (email reconciliation of masked imports).
+ *
+ * The previous flow was stateless: the whole CSV traveled on every preview and
+ * a single all-or-nothing apply. Real Enterprise exports (350k+ contacts)
+ * produce tens of thousands of ambiguous matches, which made that flow
+ * unusable — the CSV is now parsed/matched once into these tables and the
+ * operator resolves/applies in paginated, quantified batches.
+ *
+ * - `email_reconcile_sessions`: one working set per import job (PK job_id —
+ * re-running the match replaces the session). Carries the counters the UI
+ * shows plus a capped no-match sample.
+ * - `email_reconcile_items`: one row per matched masked contact. kind=auto
+ * rows are ready to apply; kind=ambiguous rows hold the top candidates
+ * (jsonb, sorted by name-similarity score) for operator/bulk resolution.
+ * Composite index (job_id, kind, status) drives pagination and batch
+ * selection; the (job_id, contact_id) unique index anchors resolutions.
+ */
+export class CreateEmailReconcileSessions1783000000000 implements MigrationInterface {
+ private readonly sessionsTable = new Table({
+ name: 'email_reconcile_sessions',
+ columns: [
+ { name: 'job_id', type: 'uuid', isPrimary: true },
+ { name: 'account_id', type: 'integer', isNullable: false },
+ { name: 'csv_rows', type: 'integer', isNullable: false },
+ { name: 'invalid_csv_rows', type: 'integer', isNullable: false },
+ { name: 'contacts_masked', type: 'integer', isNullable: false },
+ { name: 'already_clean', type: 'integer', isNullable: false },
+ { name: 'no_match_total', type: 'integer', isNullable: false },
+ { name: 'no_match_sample', type: 'jsonb', isNullable: false, default: "'[]'" },
+ { name: 'created_at', type: 'TIMESTAMP WITH TIME ZONE', isNullable: false, default: 'NOW()' },
+ { name: 'updated_at', type: 'TIMESTAMP WITH TIME ZONE', isNullable: false, default: 'NOW()' },
+ ],
+ foreignKeys: [{ columnNames: ['job_id'], referencedTableName: 'enterprise_import_jobs', referencedColumnNames: ['id'], onDelete: 'CASCADE' }],
+ });
+
+ private readonly itemsTable = new Table({
+ name: 'email_reconcile_items',
+ columns: [
+ { name: 'id', type: 'bigserial', isPrimary: true },
+ { name: 'job_id', type: 'uuid', isNullable: false },
+ { name: 'contact_id', type: 'integer', isNullable: false },
+ { name: 'current_email', type: 'varchar', length: '255', isNullable: false },
+ { name: 'contact_name', type: 'varchar', length: '255', isNullable: true },
+ { name: 'kind', type: 'varchar', length: '16', isNullable: false },
+ { name: 'status', type: 'varchar', length: '16', isNullable: false, default: "'pending'" },
+ { name: 'new_email', type: 'varchar', length: '255', isNullable: true },
+ { name: 'csv_row_number', type: 'integer', isNullable: true },
+ { name: 'candidates', type: 'jsonb', isNullable: true },
+ { name: 'candidates_total', type: 'integer', isNullable: true },
+ { name: 'failure_reason', type: 'text', isNullable: true },
+ ],
+ foreignKeys: [
+ { columnNames: ['job_id'], referencedTableName: 'email_reconcile_sessions', referencedColumnNames: ['job_id'], onDelete: 'CASCADE' },
+ // CASCADE: a deleted contact simply drops out of the working set.
+ { columnNames: ['contact_id'], referencedTableName: 'contacts', referencedColumnNames: ['id'], onDelete: 'CASCADE' },
+ ],
+ indices: [
+ { name: 'email_reconcile_items_job_kind_status_idx', columnNames: ['job_id', 'kind', 'status'] },
+ { name: 'email_reconcile_items_job_contact_uq', columnNames: ['job_id', 'contact_id'], isUnique: true },
+ ],
+ });
+
+ public async up(queryRunner: QueryRunner): Promise {
+ await queryRunner.createTable(this.sessionsTable, true);
+ await queryRunner.createTable(this.itemsTable, true);
+ }
+
+ public async down(queryRunner: QueryRunner): Promise {
+ await queryRunner.dropTable(this.itemsTable, true);
+ await queryRunner.dropTable(this.sessionsTable, true);
+ }
+}
diff --git a/apps/msgops-api/src/modules/contacts/contacts.service.ts b/apps/msgops-api/src/modules/contacts/contacts.service.ts
index 4c65b846..3c5925ae 100644
--- a/apps/msgops-api/src/modules/contacts/contacts.service.ts
+++ b/apps/msgops-api/src/modules/contacts/contacts.service.ts
@@ -24,7 +24,6 @@ import { SuppressedsPageDto } from './dto/suppressedsPage.dto';
import { ContactAutomationEntity } from 'src/entities/contact-automation.entity';
import { AuditService } from './../../utils/audits/audit.service';
import { ClickhouseProvider } from '../../providers/clickhouse.provider';
-import { maskEmail } from '../../utils/masking/email-masker';
import { EventPublisherService } from '../../providers/messaging/event-publisher.service';
import { EXCHANGES } from '@bms/messaging';
import { TagEntity } from 'src/entities/tag.entity';
@@ -93,7 +92,10 @@ export class ContactsService {
async findAll(): Promise> {
try {
- const contacts = await this.contactRepository.find({
+ // Returns the stored email verbatim. Reconciled contacts carry their
+ // real address (the send flow depends on it); never re-mask on read —
+ // contacts still pending reconcile hold the masked placeholder anyway.
+ return await this.contactRepository.find({
where: {
accountId: this.cls.get('accountId'),
},
@@ -101,10 +103,6 @@ export class ContactsService {
createdAtDate: 'DESC',
},
});
- return contacts.map((contact) => ({
- ...contact,
- email: contact.maskedEmail ?? maskEmail(contact.email),
- })) as ContactEntity[];
} catch (e) {
console.error(e);
throw new HttpException('Internal Server Error', HttpStatus.INTERNAL_SERVER_ERROR);
@@ -492,13 +490,10 @@ export class ContactsService {
return csvParser.parse(contacts);
}
- const maskedResults = results.map((result) => ({
- ...result,
- email: maskEmail(result.email),
- }));
-
+ // Stored email verbatim — same rationale as findAll/findOneById: the
+ // send flow and the operator need the real address after reconcile.
return new PaginationDto({
- results: maskedResults,
+ results,
total: results.length,
page: params.page,
itemsPerPage: params.itemsPerPage,
@@ -648,10 +643,6 @@ export class ContactsService {
returnObject[this.snakeToCamelCase(key)] = result[0][key];
});
- if (returnObject.email) {
- returnObject.email = maskEmail(returnObject.email);
- }
-
return returnObject;
} catch (e) {
console.error(e);
diff --git a/apps/msgops-api/src/modules/enterprise-import/__tests__/email-reconcile.service.spec.ts b/apps/msgops-api/src/modules/enterprise-import/__tests__/email-reconcile.service.spec.ts
index a1d6da0d..dccdf911 100644
--- a/apps/msgops-api/src/modules/enterprise-import/__tests__/email-reconcile.service.spec.ts
+++ b/apps/msgops-api/src/modules/enterprise-import/__tests__/email-reconcile.service.spec.ts
@@ -14,6 +14,9 @@ describe('EmailReconcileService', () => {
let service: EmailReconcileService;
let contactsRepo: {
update: jest.Mock;
+ create: jest.Mock;
+ save: jest.Mock;
+ findOne: jest.Mock;
createQueryBuilder: jest.Mock;
};
let jobsRepo: { findOne: jest.Mock };
@@ -28,6 +31,12 @@ describe('EmailReconcileService', () => {
beforeEach(async () => {
contactsRepo = {
update: jest.fn().mockResolvedValue({ affected: 1 }),
+ // create() passes the partial through so save() assertions can inspect
+ // exactly what would be persisted.
+ create: jest.fn((partial) => partial),
+ save: jest.fn().mockImplementation((entity) => Promise.resolve(entity)),
+ // No pre-existing holder of any email unless a test says otherwise.
+ findOne: jest.fn().mockResolvedValue(null),
createQueryBuilder: jest.fn(),
};
jobsRepo = { findOne: jest.fn() };
@@ -130,15 +139,202 @@ describe('EmailReconcileService', () => {
});
});
+ describe('required CSV columns', () => {
+ it('rejects a CSV missing created_at before processing any row', async () => {
+ setMaskedContacts([]);
+ const csv = ['name,email,status', 'Lucas Silva,lucassilva@gmail.com,Active'].join('\n');
+
+ const err = await service.preview('job-1', csv).catch((e) => e);
+ expect(err).toBeInstanceOf(BadRequestException);
+ expect(err.getResponse()).toMatchObject({ code: 'RECONCILE_MISSING_COLUMNS', missing: ['created_at'] });
+ });
+
+ it('lists every missing required column', async () => {
+ setMaskedContacts([]);
+ const csv = ['email,status', 'lucassilva@gmail.com,Active'].join('\n');
+
+ const err = await service.preview('job-1', csv).catch((e) => e);
+ expect(err).toBeInstanceOf(BadRequestException);
+ expect(err.getResponse()).toMatchObject({ missing: ['name (or first_name + last_name)', 'created_at'] });
+ });
+
+ it('accepts first_name/last_name in place of name, composing the full name', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira' }]);
+
+ // Collision decided by name only (contact has no createdAt) — proves the
+ // composed first+last name feeds the similarity scoring.
+ const csv = [
+ 'first_name,last_name,email,status,created_at',
+ 'Joao,Pereira,lucassilva@gmail.com,Active,2026-01-01',
+ 'Carlos,Melo,lucasrocha@gmail.com,Active,2026-01-02',
+ ].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ expect(out.ambiguousMatches).toBe(0);
+ });
+
+ it('rejects first_name without last_name when name is absent', async () => {
+ setMaskedContacts([]);
+ const csv = ['first_name,email,created_at', 'Lucas,lucassilva@gmail.com,2026-01-01'].join('\n');
+
+ const err = await service.preview('job-1', csv).catch((e) => e);
+ expect(err).toBeInstanceOf(BadRequestException);
+ expect(err.getResponse()).toMatchObject({ missing: ['name (or first_name + last_name)'] });
+ });
+
+ it('accepts headers regardless of case and semicolon delimiter', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Lucas', lastName: 'Silva' }]);
+ const csv = ['Name;Email;Status;Created_At', 'Lucas Silva;lucassilva@gmail.com;Active;2026-01-01'].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ });
+ });
+
+ describe('created_at disambiguation', () => {
+ it('auto-matches when exactly one collision candidate shares the exact timestamp', async () => {
+ // Names are useless here (both totally different people) — only the
+ // timestamp separates them.
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira', createdAt: new Date('2023-04-10T14:22:31Z') }]);
+
+ const csv = ['name,email,status,created_at', 'Ana Souza,lucassilva@gmail.com,Active,2023-04-10 14:22:31', 'Rita Melo,lucasrocha@gmail.com,Active,2024-08-01 09:00:00'].join(
+ '\n',
+ );
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ expect(out.ambiguousMatches).toBe(0);
+ });
+
+ it('tolerates a fixed timezone offset in offset-less timestamps', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira', createdAt: new Date('2023-04-10T14:22:31Z') }]);
+
+ // Export written in America/Sao_Paulo local time (UTC-3), no offset marker.
+ const csv = ['name,email,status,created_at', 'Ana Souza,lucassilva@gmail.com,Active,2023-04-10 11:22:31', 'Rita Melo,lucasrocha@gmail.com,Active,2024-08-01 09:00:00'].join(
+ '\n',
+ );
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ expect(out.ambiguousMatches).toBe(0);
+ });
+
+ it('uses date-only agreement when the CSV carries no time component', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira', createdAt: new Date('2023-04-10T14:22:31Z') }]);
+
+ const csv = ['name,email,status,created_at', 'Ana Souza,lucassilva@gmail.com,Active,2023-04-10', 'Rita Melo,lucasrocha@gmail.com,Active,2024-08-01'].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ });
+
+ it('keeps ambiguity but drops time-disagreeing candidates when several share the date', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira', createdAt: new Date('2023-04-10T14:22:31Z') }]);
+
+ const csv = [
+ 'name,email,status,created_at',
+ 'Ana Souza,lucassilva@gmail.com,Active,2023-04-10',
+ 'Rita Melo,lucasrocha@gmail.com,Active,2023-04-10',
+ 'Bia Costa,lucasbia@gmail.com,Active,2024-08-01',
+ ].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.ambiguousMatches).toBe(1);
+ // Only the two same-day candidates survive; each carries timeMatch=1.
+ expect(out.ambiguousSample[0]?.candidates).toHaveLength(2);
+ expect(out.ambiguousSample[0]?.candidates.every((c) => c.timeMatch === 1)).toBe(true);
+ });
+
+ it('compares at minute precision, ignoring seconds and milliseconds', async () => {
+ // DB keeps ms, export truncates to the minute — the instant tier must
+ // still discriminate between two same-day candidates.
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira', createdAt: new Date('2023-04-10T14:22:31.874Z') }]);
+
+ const csv = [
+ 'name,email,status,created_at',
+ // 11:22 local (UTC-3) == 14:22Z — same minute, no seconds written.
+ 'Ana Souza,lucassilva@gmail.com,Active,2023-04-10 11:22',
+ // Same day but a different minute — must lose to the row above.
+ 'Rita Melo,lucasrocha@gmail.com,Active,2023-04-10 09:15',
+ ].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ expect(out.ambiguousMatches).toBe(0);
+ });
+
+ it('matches names regardless of punctuation and token order', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Lucas', lastName: 'Silva' }]);
+
+ const csv = ['name,email,status,created_at', '"Silva, Lucas",lucassilva@gmail.com,Active,2026-01-01', 'Carlos Pereira,lucasrocha@gmail.com,Active,2026-01-02'].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.uniqueMatches).toBe(1);
+ expect(out.ambiguousMatches).toBe(0);
+ });
+
+ it('does not hand a lone CSV row to every contact sharing the mask (partial CSV)', async () => {
+ // Real incident: a 1-row test CSV auto-matched all 82 contacts whose
+ // mask collided with the row's — the same email assigned 82 times.
+ setMaskedContacts([
+ { id: 10, email: 'franc***@gmail.com', firstName: 'Francyne', lastName: 'Ferraz', createdAt: new Date('2026-07-09T07:48:16.039Z') },
+ { id: 11, email: 'franc***@gmail.com', firstName: 'Francisco', lastName: 'Picone', createdAt: new Date('2026-04-11T14:02:32.298Z') },
+ { id: 12, email: 'franc***@gmail.com', firstName: 'Franco', lastName: 'Caputo', createdAt: new Date('2026-03-16T16:44:14.642Z') },
+ ]);
+
+ // Export truncates seconds — 04:48 local (UTC-3) vs 07:48:16.039Z still
+ // agrees at minute precision (seconds/ms ignored on both sides).
+ const csv = ['first_name,last_name,email,status,created_at', 'Francyne,Ferraz,francferraz98@gmail.com,Active,2026-07-09 04:48'].join('\n');
+
+ const out = await service.preview('job-1', csv);
+
+ expect(out.uniqueMatches).toBe(1);
+ // The other two collision contacts must NOT auto-receive the same email;
+ // the operator sees the row as a (time-disagreeing) manual option.
+ expect(out.ambiguousMatches).toBe(2);
+ expect(out.ambiguousSample.map((a) => a.contactId).sort()).toEqual([11, 12]);
+ });
+
+ it('gives a CSV row claimed by two auto picks to the strongest agreement only', async () => {
+ // Bulk-created base: both contacts share the mask AND the creation
+ // minute, so both would auto-pick the lone row. Email is unique per
+ // account — the better name keeps it, the other goes to the operator.
+ setMaskedContacts([
+ { id: 10, email: 'lucas***@gmail.com', firstName: 'Lucas', lastName: 'Silva', createdAt: new Date('2026-01-05T12:30:10Z') },
+ { id: 11, email: 'lucas***@gmail.com', firstName: 'Pedro', lastName: 'Costa', createdAt: new Date('2026-01-05T12:30:40Z') },
+ ]);
+
+ const csv = ['name,email,status,created_at', 'Lucas Silva,lucassilva@gmail.com,Active,2026-01-05 12:30'].join('\n');
+
+ const out = await service.preview('job-1', csv);
+
+ expect(out.uniqueMatches).toBe(1);
+ expect(out.ambiguousMatches).toBe(1);
+ expect(out.ambiguousSample[0]?.contactId).toBe(11);
+ expect(out.ambiguousSample[0]?.candidates).toHaveLength(1);
+ });
+
+ it('falls back to name-only behavior when the contact has no createdAt', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Joao', lastName: 'Pereira' }]);
+
+ const csv = ['name,email,status,created_at', 'Lucas Silva,lucassilva@gmail.com,Active,2026-01-01', 'Lucas Souza,lucasrocha@gmail.com,Active,2026-01-02'].join('\n');
+
+ const out = await service.preview('job-1', csv);
+ expect(out.ambiguousMatches).toBe(1);
+ expect(out.ambiguousSample[0]?.candidates).toHaveLength(2);
+ });
+ });
+
describe('apply', () => {
- it('writes a row per unique match using repository.update so the BeforeUpdate listener fires', async () => {
+ it('writes a row per unique match using save() so the BeforeUpdate listener fires', async () => {
setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Lucas', lastName: 'Silva' }]);
const csv = ['name,email,status,created_at', 'Lucas Silva,lucassilva@gmail.com,Active,2026-01-01'].join('\n');
const out = await service.apply('job-1', csv, []);
- expect(contactsRepo.update).toHaveBeenCalledWith({ id: 10 }, { email: 'lucassilva@gmail.com' });
+ expect(contactsRepo.save).toHaveBeenCalledWith({ id: 10, email: 'lucassilva@gmail.com' });
expect(out.updated).toBe(1);
expect(out.skippedAmbiguous).toBe(0);
});
@@ -151,7 +347,7 @@ describe('EmailReconcileService', () => {
// Operator picks row 2 even though row 1 would otherwise be ambiguous.
const out = await service.apply('job-1', csv, [{ contactId: 10, csvRowNumber: 2 }]);
- expect(contactsRepo.update).toHaveBeenCalledWith({ id: 10 }, { email: 'lucasrocha@gmail.com' });
+ expect(contactsRepo.save).toHaveBeenCalledWith({ id: 10, email: 'lucasrocha@gmail.com' });
expect(out.updated).toBe(1);
});
@@ -162,17 +358,32 @@ describe('EmailReconcileService', () => {
const out = await service.apply('job-1', csv, [{ contactId: 10, csvRowNumber: null }]);
- expect(contactsRepo.update).not.toHaveBeenCalled();
+ expect(contactsRepo.save).not.toHaveBeenCalled();
expect(out.updated).toBe(0);
expect(out.skippedAmbiguous).toBe(1);
});
+ it('fails friendly when the email already belongs to another contact in the account', async () => {
+ setMaskedContacts([{ id: 10, email: 'lucas***@gmail.com', firstName: 'Lucas', lastName: 'Silva' }]);
+ // Someone else (a clean contact) already holds the address.
+ contactsRepo.findOne.mockResolvedValue({ id: 999 });
+
+ const csv = ['name,email,status,created_at', 'Lucas Silva,lucassilva@gmail.com,Active,2026-01-01'].join('\n');
+
+ const out = await service.apply('job-1', csv, []);
+
+ expect(contactsRepo.save).not.toHaveBeenCalled();
+ expect(out.updated).toBe(0);
+ expect(out.failures).toHaveLength(1);
+ expect(out.failures[0]?.reason).toContain('#999');
+ });
+
it('records failures without aborting the batch', async () => {
setMaskedContacts([
{ id: 10, email: 'lucas***@gmail.com', firstName: 'Lucas', lastName: 'Silva' },
{ id: 11, email: 'maria***@gmail.com', firstName: 'Maria', lastName: 'Souza' },
]);
- contactsRepo.update.mockImplementationOnce(() => Promise.reject(new Error('DB down')));
+ contactsRepo.save.mockImplementationOnce(() => Promise.reject(new Error('DB down')));
const csv = ['name,email,status,created_at', 'Lucas Silva,lucassilva@gmail.com,Active,2026-01-01', 'Maria Souza,mariasouza@gmail.com,Active,2026-01-02'].join('\n');
diff --git a/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile-session.dto.ts b/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile-session.dto.ts
new file mode 100644
index 00000000..67893635
--- /dev/null
+++ b/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile-session.dto.ts
@@ -0,0 +1,61 @@
+import * as Joi from 'joi';
+import { JoiSchema, JoiSchemaOptions } from 'nestjs-joi';
+
+// Same ceiling as the stateless preview/apply DTOs (reconcile.dto.ts) — the
+// CSV arrives once at session create and never travels again afterwards.
+const CSV_MAX_BYTES = 50 * 1024 * 1024;
+
+const resolutionSchema = Joi.object({
+ contactId: Joi.number().integer().positive().required(),
+ // null = explicit "skip / leave masked". Number = pick this CSV row.
+ csvRowNumber: Joi.number().integer().positive().allow(null).required(),
+});
+
+@JoiSchemaOptions({ stripUnknown: true })
+export class ReconcileSessionCreateDto {
+ @JoiSchema(Joi.string().min(10).max(CSV_MAX_BYTES).required())
+ csv: string;
+
+ // Optional columns the operator deselected in the upload form (e.g.
+ // `status`). Required columns are never ignorable — the service drops them
+ // from this list defensively.
+ @JoiSchema(Joi.array().items(Joi.string().max(100)).max(50).default([]))
+ ignoreColumns: string[];
+}
+
+@JoiSchemaOptions({ stripUnknown: true })
+export class ReconcileResolveBatchDto {
+ // Bounded to a UI page — decisions apply immediately, so each call must
+ // stay interactive.
+ @JoiSchema(Joi.array().items(resolutionSchema).min(1).max(500).required())
+ resolutions: Array<{ contactId: number; csvRowNumber: number | null }>;
+}
+
+@JoiSchemaOptions({ stripUnknown: true })
+export class ReconcileApplyAutoDto {
+ // Operator-tunable chunk: bigger = fewer round trips, smaller = more
+ // granular progress. Each row is an individual repo.update (BeforeUpdate
+ // listener), so the ceiling keeps a chunk comfortably under proxy timeouts.
+ @JoiSchema(Joi.number().integer().min(100).max(20000).default(5000))
+ limit: number;
+}
+
+@JoiSchemaOptions({ stripUnknown: true })
+export class ReconcileBulkResolveDto {
+ @JoiSchema(Joi.string().valid('best-name', 'skip-remaining').required())
+ strategy: 'best-name' | 'skip-remaining';
+
+ // best-name only: minimum score of the top candidate. 0.8 mirrors the
+ // automatic tie-break; lower values trade precision for coverage under
+ // explicit operator consent.
+ @JoiSchema(Joi.number().min(0.1).max(1).default(0.5))
+ threshold: number;
+
+ @JoiSchema(Joi.number().integer().min(100).max(20000).default(5000))
+ limit: number;
+
+ // Pagination cursor from the previous call (items left pending are not
+ // re-examined within one sweep).
+ @JoiSchema(Joi.string().pattern(/^\d+$/).optional())
+ afterId?: string;
+}
diff --git a/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile.dto.ts b/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile.dto.ts
index 66bdfd5d..9a2488fb 100644
--- a/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile.dto.ts
+++ b/apps/msgops-api/src/modules/enterprise-import/dtos/reconcile.dto.ts
@@ -1,7 +1,11 @@
import * as Joi from 'joi';
import { JoiSchema, JoiSchemaOptions } from 'nestjs-joi';
-const CSV_MAX_BYTES = 12 * 1024 * 1024; // 12MB — comfortably above 30k+ contact rows.
+// 50MB of CSV text — real Enterprise exports reach hundreds of thousands of
+// contacts (350k ≈ 20MB). Must stay under the 64mb body-parser limit on the
+// /imports routes (main.ts) and match MAX_CSV_FILE_MB in the frontend's
+// reconcile-emails-card.tsx, which pre-checks the file before uploading.
+const CSV_MAX_BYTES = 50 * 1024 * 1024;
const resolutionSchema = Joi.object({
contactId: Joi.number().integer().positive().required(),
diff --git a/apps/msgops-api/src/modules/enterprise-import/email-reconcile-session.service.ts b/apps/msgops-api/src/modules/enterprise-import/email-reconcile-session.service.ts
new file mode 100644
index 00000000..ef4474d9
--- /dev/null
+++ b/apps/msgops-api/src/modules/enterprise-import/email-reconcile-session.service.ts
@@ -0,0 +1,451 @@
+import { BadRequestException, Injectable, Logger, NotFoundException } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { In, Repository } from 'typeorm';
+import { ContactEntity } from '../../entities/contact.entity';
+import { EnterpriseImportJobEntity } from '../../entities/enterprise-import-job.entity';
+import { EmailReconcileItemEntity, EmailReconcileStoredCandidate } from '../../entities/email-reconcile-item.entity';
+import { EmailReconcileSessionEntity } from '../../entities/email-reconcile-session.entity';
+import { AMBIGUOUS_CANDIDATES_LIMIT, EmailReconcileService } from './email-reconcile.service';
+import type {
+ AmbiguousPageResult,
+ ApplyAutoChunkResult,
+ ApplyResolution,
+ BulkResolveResult,
+ ReconcileItemsPage,
+ ReconcileSessionProgress,
+ ResolveBatchResult,
+} from './email-reconcile.types';
+
+const NO_MATCH_SAMPLE_LIMIT = 50;
+const INSERT_CHUNK = 1000;
+
+/**
+ * Batched reconciliation over a persisted working set.
+ *
+ * The stateless preview/apply flow collapses on real Enterprise exports:
+ * 350k contacts yield tens of thousands of ambiguous matches, and both the
+ * response payload and the single all-or-nothing apply become unmanageable.
+ *
+ * Here the CSV is parsed and matched ONCE (createSession) and the outcome is
+ * persisted — one item per matched contact. From then on everything is
+ * incremental and quantified:
+ * - auto items are applied in operator-sized chunks (applyAutoChunk),
+ * - ambiguous items are reviewed in pages (getAmbiguousPage + resolveBatch),
+ * - bulk strategies (best-name / skip-remaining) clear the long tail.
+ * Progress is derivable at any point from item statuses, so the operator can
+ * leave and resume without re-uploading the CSV.
+ */
+@Injectable()
+export class EmailReconcileSessionService {
+ private readonly logger = new Logger(EmailReconcileSessionService.name);
+
+ constructor(
+ private readonly reconcileService: EmailReconcileService,
+ @InjectRepository(ContactEntity)
+ private readonly contactsRepo: Repository,
+ @InjectRepository(EnterpriseImportJobEntity)
+ private readonly jobsRepo: Repository,
+ @InjectRepository(EmailReconcileSessionEntity)
+ private readonly sessionsRepo: Repository,
+ @InjectRepository(EmailReconcileItemEntity)
+ private readonly itemsRepo: Repository,
+ ) {}
+
+ async createSession(jobId: string, csv: string, ignoreColumns: string[] = []): Promise {
+ const job = await this.requireJob(jobId);
+ const computation = await this.reconcileService.computeReconciliation(csv, job.accountId!, ignoreColumns);
+
+ // Replace any previous working set — items cascade with the session row.
+ await this.sessionsRepo.delete({ jobId });
+ await this.sessionsRepo.insert({
+ jobId,
+ accountId: job.accountId!,
+ csvRows: computation.csvRows,
+ invalidCsvRows: computation.invalidCsvRows,
+ contactsMasked: computation.contactsMasked,
+ alreadyClean: computation.alreadyClean,
+ noMatchTotal: computation.noMatches.length,
+ noMatchSample: computation.noMatches.slice(0, NO_MATCH_SAMPLE_LIMIT),
+ });
+
+ const items: Array> = [
+ ...computation.matches.map((m) => ({
+ jobId,
+ contactId: m.contactId,
+ currentEmail: m.currentEmail,
+ contactName: m.contactName || null,
+ kind: 'auto' as const,
+ status: 'pending' as const,
+ newEmail: m.newEmail,
+ csvRowNumber: m.csvRowNumber,
+ })),
+ ...computation.ambiguous.map((a) => ({
+ jobId,
+ contactId: a.contactId,
+ currentEmail: a.currentEmail,
+ contactName: a.contactName || null,
+ kind: 'ambiguous' as const,
+ status: 'pending' as const,
+ candidates: a.candidates.slice(0, AMBIGUOUS_CANDIDATES_LIMIT) as EmailReconcileStoredCandidate[],
+ candidatesTotal: a.candidatesTotal,
+ })),
+ ];
+
+ for (let i = 0; i < items.length; i += INSERT_CHUNK) {
+ await this.itemsRepo.insert(items.slice(i, i + INSERT_CHUNK));
+ }
+
+ this.logger.log(
+ `[email-reconcile] session created jobId=${jobId} accountId=${job.accountId} auto=${computation.matches.length} ambiguous=${computation.ambiguous.length} no_match=${computation.noMatches.length}`,
+ );
+ return this.getProgress(jobId);
+ }
+
+ async getProgress(jobId: string): Promise {
+ const session = await this.sessionsRepo.findOne({ where: { jobId } });
+ if (!session) throw new NotFoundException(`No reconcile session for job ${jobId}`);
+
+ const counts: Array<{ kind: string; status: string; count: string }> = await this.itemsRepo
+ .createQueryBuilder('i')
+ .select('i.kind', 'kind')
+ .addSelect('i.status', 'status')
+ .addSelect('COUNT(*)', 'count')
+ .where('i.job_id = :jobId', { jobId })
+ .groupBy('i.kind')
+ .addGroupBy('i.status')
+ .getRawMany();
+
+ const get = (kind: string, status: string): number => Number(counts.find((c) => c.kind === kind && c.status === status)?.count ?? 0);
+ const auto = {
+ applied: get('auto', 'applied'),
+ failed: get('auto', 'failed'),
+ pending: get('auto', 'pending'),
+ total: 0,
+ };
+ auto.total = auto.applied + auto.failed + auto.pending;
+ const ambiguous = {
+ applied: get('ambiguous', 'applied'),
+ skipped: get('ambiguous', 'skipped'),
+ pending: get('ambiguous', 'pending'),
+ failed: get('ambiguous', 'failed'),
+ total: 0,
+ };
+ ambiguous.total = ambiguous.applied + ambiguous.skipped + ambiguous.pending + ambiguous.failed;
+
+ return {
+ jobId,
+ csvRows: session.csvRows,
+ invalidCsvRows: session.invalidCsvRows,
+ contactsMasked: session.contactsMasked,
+ alreadyClean: session.alreadyClean,
+ noMatches: session.noMatchTotal,
+ noMatchSample: session.noMatchSample,
+ auto: { total: auto.total, applied: auto.applied, failed: auto.failed, pending: auto.pending },
+ ambiguous: {
+ total: ambiguous.total,
+ applied: ambiguous.applied,
+ skipped: ambiguous.skipped,
+ pending: ambiguous.pending,
+ failed: ambiguous.failed,
+ },
+ createdAt: session.createdAt.toISOString(),
+ updatedAt: session.updatedAt.toISOString(),
+ };
+ }
+
+ async getAmbiguousPage(jobId: string, offset: number, limit: number, q?: string): Promise {
+ await this.requireSession(jobId);
+ const qb = this.itemsRepo
+ .createQueryBuilder('i')
+ .where('i.job_id = :jobId', { jobId })
+ .andWhere("i.kind = 'ambiguous'")
+ .andWhere("i.status = 'pending'")
+ .orderBy('i.id', 'ASC')
+ .skip(offset)
+ .take(limit);
+ this.applySearch(qb, q);
+ const [rows, totalPending] = await qb.getManyAndCount();
+
+ const usedBy = await this.findUsedCandidates(jobId, rows);
+
+ return {
+ totalPending,
+ offset,
+ items: rows.map((r) => ({
+ contactId: r.contactId,
+ currentEmail: r.currentEmail,
+ contactName: r.contactName ?? '',
+ candidates: (r.candidates ?? []).map((c) => {
+ const usedByContactId = usedBy.get(c.csvEmail.toLowerCase());
+ return usedByContactId !== undefined && usedByContactId !== r.contactId ? { ...c, usedByContactId } : c;
+ }),
+ candidatesTotal: r.candidatesTotal ?? r.candidates?.length ?? 0,
+ })),
+ };
+ }
+
+ /**
+ * Emails from this page's candidates that were already applied in this
+ * session, mapped to the contact that received them. Keyed by email (not
+ * CSV row number) so duplicate CSV rows carrying the same address are
+ * flagged too. Lets the UI disable picks that would collide with the
+ * per-account email uniqueness instead of failing after the fact.
+ */
+ private async findUsedCandidates(jobId: string, rows: EmailReconcileItemEntity[]): Promise> {
+ const emails = new Set();
+ for (const r of rows) for (const c of r.candidates ?? []) emails.add(c.csvEmail.toLowerCase());
+ if (emails.size === 0) return new Map();
+
+ const applied: Array<{ new_email: string; contact_id: number }> = await this.itemsRepo
+ .createQueryBuilder('i')
+ .select('i.new_email', 'new_email')
+ .addSelect('i.contact_id', 'contact_id')
+ .where('i.job_id = :jobId', { jobId })
+ .andWhere("i.status = 'applied'")
+ .andWhere('LOWER(i.new_email) IN (:...emails)', { emails: [...emails] })
+ .getRawMany();
+
+ return new Map(applied.map((a) => [a.new_email.toLowerCase(), Number(a.contact_id)]));
+ }
+
+ /**
+ * Flat, searchable listing of session items — the operator's "who matched
+ * what" view. Covers every kind/status (auto picks, applied, failed,
+ * skipped, pending ambiguous); the resolution UI stays in the ambiguous
+ * queue, this is for visibility and lookup.
+ */
+ async getItemsPage(
+ jobId: string,
+ opts: { kind?: 'auto' | 'ambiguous'; status?: 'pending' | 'applied' | 'skipped' | 'failed'; q?: string; offset: number; limit: number },
+ ): Promise {
+ await this.requireSession(jobId);
+ const qb = this.itemsRepo.createQueryBuilder('i').where('i.job_id = :jobId', { jobId }).orderBy('i.id', 'ASC').skip(opts.offset).take(opts.limit);
+ if (opts.kind) qb.andWhere('i.kind = :kind', { kind: opts.kind });
+ if (opts.status) qb.andWhere('i.status = :status', { status: opts.status });
+ this.applySearch(qb, opts.q);
+ const [rows, total] = await qb.getManyAndCount();
+
+ return {
+ total,
+ offset: opts.offset,
+ items: rows.map((r) => ({
+ contactId: r.contactId,
+ contactName: r.contactName ?? '',
+ currentEmail: r.currentEmail,
+ kind: r.kind,
+ status: r.status,
+ newEmail: r.newEmail ?? null,
+ csvRowNumber: r.csvRowNumber ?? null,
+ failureReason: r.failureReason ?? null,
+ candidatesTotal: r.candidatesTotal ?? null,
+ })),
+ };
+ }
+
+ /**
+ * Case-insensitive contains-search over everything the operator can see:
+ * contact name, masked email, applied email, and (for ambiguous items) the
+ * candidates payload — so searching a raw CSV email finds the queue entry
+ * offering it.
+ */
+ private applySearch(qb: { andWhere: (clause: string, params: Record) => unknown }, q?: string): void {
+ const term = (q ?? '').trim();
+ if (!term) return;
+ const escaped = term.replace(/[\\%_]/g, (ch) => `\\${ch}`);
+ qb.andWhere('(i.contact_name ILIKE :q OR i.current_email ILIKE :q OR i.new_email ILIKE :q OR i.candidates::text ILIKE :q)', { q: `%${escaped}%` });
+ }
+
+ /**
+ * Applies a page of operator decisions immediately. Bounded by the DTO, so
+ * each call stays interactive; the contact write goes through repo.update
+ * per row on purpose — the @BeforeUpdate listener re-derives hashed_email.
+ */
+ async resolveBatch(jobId: string, resolutions: ApplyResolution[]): Promise {
+ const session = await this.requireSession(jobId);
+ const items = await this.itemsRepo.find({
+ where: { jobId, contactId: In(resolutions.map((r) => r.contactId)), kind: 'ambiguous', status: 'pending' },
+ });
+ const byContact = new Map(items.map((i) => [i.contactId, i]));
+
+ let applied = 0;
+ let skipped = 0;
+ let invalid = 0;
+ const failures: Array<{ contactId: number; reason: string }> = [];
+
+ for (const resolution of resolutions) {
+ const item = byContact.get(resolution.contactId);
+ if (!item) {
+ invalid++;
+ continue;
+ }
+
+ if (resolution.csvRowNumber === null) {
+ await this.itemsRepo.update({ id: item.id }, { status: 'skipped' });
+ skipped++;
+ continue;
+ }
+
+ const candidate = (item.candidates ?? []).find((c) => c.csvRowNumber === resolution.csvRowNumber);
+ if (!candidate) {
+ invalid++;
+ continue;
+ }
+
+ const outcome = await this.applyEmail(item, candidate.csvEmail, candidate.csvRowNumber, session.accountId);
+ if (outcome) failures.push(outcome);
+ else applied++;
+ }
+
+ return { applied, skipped, invalid, failures, progress: await this.getProgress(jobId) };
+ }
+
+ /** Applies the next `limit` pending auto items. The client loops until remaining=0. */
+ async applyAutoChunk(jobId: string, limit: number): Promise {
+ const session = await this.requireSession(jobId);
+ const chunk = await this.itemsRepo.find({
+ where: { jobId, kind: 'auto', status: 'pending' },
+ order: { id: 'ASC' },
+ take: limit,
+ });
+
+ let applied = 0;
+ let failed = 0;
+ for (const item of chunk) {
+ const outcome = await this.applyEmail(item, item.newEmail!, item.csvRowNumber, session.accountId);
+ if (outcome) failed++;
+ else applied++;
+ }
+
+ const remaining = await this.itemsRepo.count({ where: { jobId, kind: 'auto', status: 'pending' } });
+ this.logger.log(`[email-reconcile] auto chunk jobId=${jobId} applied=${applied} failed=${failed} remaining=${remaining}`);
+ return { applied, failed, remaining };
+ }
+
+ /**
+ * Clears pending ambiguous items in bulk.
+ * - best-name: applies candidates[0] (best score) when it clears the
+ * operator-chosen threshold AND strictly beats the runner-up. Items that
+ * don't qualify stay pending — the id cursor (afterId) keeps the client
+ * loop moving instead of re-examining them forever.
+ * - skip-remaining: marks every pending ambiguous item skipped (single SQL).
+ */
+ async bulkResolve(jobId: string, strategy: 'best-name' | 'skip-remaining', threshold: number, limit: number, afterId?: string): Promise {
+ const session = await this.requireSession(jobId);
+
+ if (strategy === 'skip-remaining') {
+ const result = await this.itemsRepo.update({ jobId, kind: 'ambiguous', status: 'pending' }, { status: 'skipped' });
+ return { resolved: result.affected ?? 0, unresolved: 0, nextAfterId: null, remainingPending: 0 };
+ }
+
+ const qb = this.itemsRepo
+ .createQueryBuilder('i')
+ .where('i.job_id = :jobId', { jobId })
+ .andWhere("i.kind = 'ambiguous'")
+ .andWhere("i.status = 'pending'")
+ .orderBy('i.id', 'ASC')
+ .take(limit);
+ if (afterId) qb.andWhere('i.id > :afterId', { afterId });
+ const chunk = await qb.getMany();
+
+ // Emails among the chunk's best candidates that some contact already owns
+ // (an applied auto pick, an earlier resolution, or a clean contact). The
+ // AUTOMATIC strategy must never consume those — the item stays pending in
+ // the similarity queue, where only an explicit operator decision settles
+ // it. Applying here would just burn the item as failed.
+ const taken = await this.findTakenEmails(
+ session.accountId,
+ chunk.map((i) => i.candidates?.[0]?.csvEmail).filter((e): e is string => Boolean(e)),
+ );
+
+ let resolved = 0;
+ let unresolved = 0;
+ for (const item of chunk) {
+ const best = item.candidates?.[0];
+ const runner = item.candidates?.[1];
+ const wins = best && best.score >= threshold && (!runner || runner.score < best.score);
+ if (!wins || taken.has(best.csvEmail.toLowerCase())) {
+ unresolved++;
+ continue;
+ }
+ const outcome = await this.applyEmail(item, best.csvEmail, best.csvRowNumber, session.accountId);
+ if (outcome) unresolved++;
+ else {
+ resolved++;
+ // Two chunk items may share the same best email — the win consumes it.
+ taken.add(best.csvEmail.toLowerCase());
+ }
+ }
+
+ const remainingPending = await this.itemsRepo.count({ where: { jobId, kind: 'ambiguous', status: 'pending' } });
+ const nextAfterId = chunk.length < limit ? null : chunk[chunk.length - 1].id;
+ this.logger.log(`[email-reconcile] bulk best-name jobId=${jobId} threshold=${threshold} resolved=${resolved} unresolved=${unresolved} remaining=${remainingPending}`);
+ return { resolved, unresolved, nextAfterId, remainingPending };
+ }
+
+ async deleteSession(jobId: string): Promise {
+ await this.requireSession(jobId);
+ await this.sessionsRepo.delete({ jobId });
+ }
+
+ // ─── helpers ────────────────────────────────────────────────────────────
+
+ /**
+ * Writes the contact + item status. Returns a failure record or null on
+ * success. save() over an entity instance (NOT repo.update, which skips
+ * entity listeners) so @BeforeUpdate re-derives hashed_email/email_provider
+ * from the new raw email — the SHA-256 contact lookup depends on it.
+ */
+ private async applyEmail(
+ item: EmailReconcileItemEntity,
+ newEmail: string,
+ csvRowNumber: number | null,
+ accountId: number,
+ ): Promise<{ contactId: number; reason: string } | null> {
+ try {
+ // Friendly pre-check before the unique index does it the hard way: the
+ // email may already belong to another contact in the account (a clean
+ // one, or one reconciled earlier in this session). The index still
+ // backs this up if a concurrent write slips through.
+ const holder = await this.contactsRepo.findOne({ where: { accountId, email: newEmail.toLowerCase() } });
+ if (holder && holder.id !== item.contactId) {
+ const reason = `email already in use by contact #${holder.id}`;
+ await this.itemsRepo.update({ id: item.id }, { status: 'failed', failureReason: reason });
+ return { contactId: item.contactId, reason };
+ }
+
+ await this.contactsRepo.save(this.contactsRepo.create({ id: item.contactId, email: newEmail }));
+ await this.itemsRepo.update({ id: item.id }, { status: 'applied', newEmail, csvRowNumber });
+ return null;
+ } catch (err: any) {
+ const reason = err?.message ?? String(err);
+ await this.itemsRepo.update({ id: item.id }, { status: 'failed', failureReason: reason });
+ return { contactId: item.contactId, reason };
+ }
+ }
+
+ /** Which of these emails already belong to some contact in the account (lowercased set). */
+ private async findTakenEmails(accountId: number, emails: string[]): Promise> {
+ const unique = [...new Set(emails.map((e) => e.toLowerCase()))];
+ if (unique.length === 0) return new Set();
+ const holders = await this.contactsRepo.find({
+ where: { accountId, email: In(unique) },
+ select: ['email'],
+ });
+ return new Set(holders.map((h) => h.email.toLowerCase()));
+ }
+
+ private async requireSession(jobId: string): Promise {
+ const session = await this.sessionsRepo.findOne({ where: { jobId } });
+ if (!session) throw new NotFoundException(`No reconcile session for job ${jobId}`);
+ return session;
+ }
+
+ private async requireJob(jobId: string): Promise {
+ const job = await this.jobsRepo.findOne({ where: { id: jobId } });
+ if (!job) throw new NotFoundException(`Import job ${jobId} not found`);
+ if (!job.accountId) {
+ throw new BadRequestException(`Import job ${jobId} has no accountId — cannot reconcile`);
+ }
+ return job;
+ }
+}
diff --git a/apps/msgops-api/src/modules/enterprise-import/email-reconcile.service.ts b/apps/msgops-api/src/modules/enterprise-import/email-reconcile.service.ts
index 608722f0..53d782fe 100644
--- a/apps/msgops-api/src/modules/enterprise-import/email-reconcile.service.ts
+++ b/apps/msgops-api/src/modules/enterprise-import/email-reconcile.service.ts
@@ -5,11 +5,33 @@ import { parseString } from 'fast-csv';
import { ContactEntity } from '../../entities/contact.entity';
import { EnterpriseImportJobEntity } from '../../entities/enterprise-import-job.entity';
import { maskEmail } from '../../utils/masking/email-masker';
-import type { AmbiguousMatch, ApplyResolution, ApplyResult, CsvRow, ReconcileMatch, ReconcilePreview } from './email-reconcile.types';
+import { parseCsvTimestamp, timeMatchLevel } from './reconcile-timestamp.util';
+import type { AmbiguousMatch, ApplyResolution, ApplyResult, CsvRow, ReconcileComputation, ReconcileMatch, ReconcilePreview, TimeMatchLevel } from './email-reconcile.types';
const AMBIGUOUS_SAMPLE_LIMIT = 100;
const NO_MATCH_SAMPLE_LIMIT = 50;
const NAME_TIE_BREAK_THRESHOLD = 0.8; // Jaccard token similarity above this counts as a confident match.
+// Columns the matching pipeline depends on: email is the mask key, created_at
+// and the name signal disambiguate collisions. A CSV missing any of them is
+// rejected before processing — a half-matched run over a wrong export is
+// worse than a hard error. The frontend blocks the same set pre-upload.
+// The name signal accepts either a single `name` column or the
+// first_name/last_name pair (exports vary between the two shapes).
+export const ALWAYS_REQUIRED_CSV_COLUMNS = ['email', 'created_at'] as const;
+export const NAME_SIGNAL_COLUMNS = ['name', 'first_name', 'last_name'] as const;
+const MISSING_NAME_SIGNAL_TOKEN = 'name (or first_name + last_name)';
+// Machine-readable marker on the 400 response so the UI can render a
+// column-specific message instead of the raw server string.
+export const MISSING_COLUMNS_ERROR_CODE = 'RECONCILE_MISSING_COLUMNS';
+// Masks keep only 5 chars of the local part, so short/common prefixes over a
+// large base collide by the thousands. Everything that leaves this service
+// (preview response, session storage) carries at most this many candidates
+// per ambiguous contact, sorted by name-similarity score.
+export const AMBIGUOUS_CANDIDATES_LIMIT = 20;
+
+// An auto decision plus how strongly it agrees with the contact — kept so
+// competing claims over the same CSV row can be arbitrated (dedupeAutoPicks).
+type AutoPick = { contact: ContactEntity; row: CsvRow; timeMatch: TimeMatchLevel; score: number };
/**
* Workaround service for EVO-1464.
@@ -25,10 +47,14 @@ const NAME_TIE_BREAK_THRESHOLD = 0.8; // Jaccard token similarity above this cou
* 1) Bucket CSV rows by their reconstructed mask (deterministic — same
* algorithm as `maskEmail`).
* 2) For each masked contact, look up its bucket. Zero rows → noMatch.
- * One row → uniqueMatch. Two or more → try name similarity; if a single
- * CSV row clears the threshold, treat it as confident. Otherwise the
+ * One row → uniqueMatch. Two or more → disambiguate by created_at
+ * agreement, then name similarity (see `matchContact`). Otherwise the
* operator decides via the resolutions payload on `apply`.
*
+ * The CSV must carry email, created_at and a name signal (`name` or the
+ * first_name/last_name pair); anything missing is a hard 400 before any row
+ * is processed.
+ *
* `apply` only writes contacts the operator has either uniquely matched or
* explicitly resolved. Anything missing from the payload is left alone.
*/
@@ -45,17 +71,47 @@ export class EmailReconcileService {
async preview(jobId: string, csv: string): Promise {
const job = await this.requireJob(jobId);
- const rows = await this.parseCsv(csv);
+ const computation = await this.computeReconciliation(csv, job.accountId);
+
+ return {
+ csvRows: computation.csvRows,
+ invalidCsvRows: computation.invalidCsvRows,
+ contactsMasked: computation.contactsMasked,
+ uniqueMatches: computation.matches.length,
+ ambiguousMatches: computation.ambiguous.length,
+ noMatches: computation.noMatches.length,
+ alreadyClean: computation.alreadyClean,
+ ambiguousSample: computation.ambiguous.slice(0, AMBIGUOUS_SAMPLE_LIMIT).map((a) => ({
+ ...a,
+ candidates: a.candidates.slice(0, AMBIGUOUS_CANDIDATES_LIMIT),
+ })),
+ noMatchSample: computation.noMatches.slice(0, NO_MATCH_SAMPLE_LIMIT),
+ };
+ }
+
+ /**
+ * Parses the CSV and matches it against the account's masked contacts.
+ * Pure read — persisting/applying is up to the caller (preview serves a
+ * capped view; the session flow stores it for batched resolution).
+ *
+ * The CSV must always be processed WHOLE: matching buckets rows by mask, so
+ * a partial read would misreport collisions from other parts of the file as
+ * unique matches.
+ */
+ async computeReconciliation(csv: string, accountId: number, ignoreColumns: string[] = []): Promise {
+ const rows = await this.parseCsv(csv, ignoreColumns);
const validRows = rows.filter((r) => r.email && r.email.includes('@'));
const invalidCsvRows = rows.length - validRows.length;
const bucketsByMask = this.bucketByMask(validRows);
- const maskedContacts = await this.loadMaskedContacts(job.accountId);
- const alreadyClean = await this.countCleanContacts(job.accountId);
+ const maskedContacts = await this.loadMaskedContacts(accountId);
+ const maskCollisions = this.countContactsByMask(maskedContacts);
+ const alreadyClean = await this.countCleanContacts(accountId);
const matches: ReconcileMatch[] = [];
const ambiguous: AmbiguousMatch[] = [];
const noMatches: Array<{ contactId: number; currentEmail: string }> = [];
+ const autoPicks: AutoPick[] = [];
for (const contact of maskedContacts) {
const candidates = bucketsByMask.get(contact.email) ?? [];
@@ -64,39 +120,52 @@ export class EmailReconcileService {
continue;
}
- if (candidates.length === 1) {
- matches.push({
- contactId: contact.id,
- currentEmail: contact.email,
- newEmail: candidates[0].email,
- csvRowNumber: candidates[0].rowNumber,
- });
- continue;
- }
-
- // Multiple CSV rows share this mask. Try name similarity to recover a
- // single confident pick before falling back to operator resolution.
- const contactName = `${contact.firstName ?? ''} ${contact.lastName ?? ''}`.trim();
- const confident = this.pickByName(contactName, candidates);
- if (confident) {
- matches.push({
- contactId: contact.id,
- currentEmail: contact.email,
- newEmail: confident.email,
- csvRowNumber: confident.rowNumber,
- });
+ const outcome = this.matchContact(contact, candidates, maskCollisions.get(contact.email) ?? 1);
+ if (outcome.kind === 'auto') {
+ autoPicks.push({ contact, row: outcome.row, timeMatch: outcome.timeMatch, score: outcome.score });
continue;
}
ambiguous.push({
contactId: contact.id,
currentEmail: contact.email,
- contactName,
- candidates: candidates.map((c) => ({
- csvRowNumber: c.rowNumber,
- csvName: c.name,
- csvEmail: c.email,
+ contactName: outcome.contactName,
+ candidates: outcome.scored.map((s) => ({
+ csvRowNumber: s.row.rowNumber,
+ csvName: s.row.name,
+ csvEmail: s.row.email,
+ score: Math.round(s.score * 1000) / 1000,
+ timeMatch: s.timeMatch,
})),
+ candidatesTotal: outcome.scored.length,
+ });
+ }
+
+ const { winners, demoted } = this.dedupeAutoPicks(autoPicks);
+ for (const w of winners) {
+ matches.push({
+ contactId: w.contact.id,
+ currentEmail: w.contact.email,
+ newEmail: w.row.email,
+ csvRowNumber: w.row.rowNumber,
+ contactName: `${w.contact.firstName ?? ''} ${w.contact.lastName ?? ''}`.trim(),
+ });
+ }
+ for (const d of demoted) {
+ ambiguous.push({
+ contactId: d.contact.id,
+ currentEmail: d.contact.email,
+ contactName: `${d.contact.firstName ?? ''} ${d.contact.lastName ?? ''}`.trim(),
+ candidates: [
+ {
+ csvRowNumber: d.row.rowNumber,
+ csvName: d.row.name,
+ csvEmail: d.row.email,
+ score: Math.round(d.score * 1000) / 1000,
+ timeMatch: d.timeMatch,
+ },
+ ],
+ candidatesTotal: 1,
});
}
@@ -104,12 +173,10 @@ export class EmailReconcileService {
csvRows: rows.length,
invalidCsvRows,
contactsMasked: maskedContacts.length,
- uniqueMatches: matches.length,
- ambiguousMatches: ambiguous.length,
- noMatches: noMatches.length,
alreadyClean,
- ambiguousSample: ambiguous.slice(0, AMBIGUOUS_SAMPLE_LIMIT),
- noMatchSample: noMatches.slice(0, NO_MATCH_SAMPLE_LIMIT),
+ matches,
+ ambiguous,
+ noMatches,
};
}
@@ -121,9 +188,11 @@ export class EmailReconcileService {
const bucketsByMask = this.bucketByMask(validRows);
const maskedContacts = await this.loadMaskedContacts(job.accountId);
+ const maskCollisions = this.countContactsByMask(maskedContacts);
const resolutionsById = new Map(resolutions.map((r) => [r.contactId, r.csvRowNumber]));
const updates: ReconcileMatch[] = [];
+ const autoPicks: AutoPick[] = [];
let skippedAmbiguous = 0;
let skippedNoMatch = 0;
@@ -159,40 +228,46 @@ export class EmailReconcileService {
skippedNoMatch++;
continue;
}
- if (candidates.length === 1) {
- updates.push({
- contactId: contact.id,
- currentEmail: contact.email,
- newEmail: candidates[0].email,
- csvRowNumber: candidates[0].rowNumber,
- });
- continue;
- }
- const contactName = `${contact.firstName ?? ''} ${contact.lastName ?? ''}`.trim();
- const confident = this.pickByName(contactName, candidates);
- if (confident) {
- updates.push({
- contactId: contact.id,
- currentEmail: contact.email,
- newEmail: confident.email,
- csvRowNumber: confident.rowNumber,
- });
+ const outcome = this.matchContact(contact, candidates, maskCollisions.get(contact.email) ?? 1);
+ if (outcome.kind === 'auto') {
+ autoPicks.push({ contact, row: outcome.row, timeMatch: outcome.timeMatch, score: outcome.score });
continue;
}
// Ambiguous and the operator didn't decide → leave alone.
skippedAmbiguous++;
}
+ // Same arbitration as the preview: one CSV row reconciles one contact.
+ const { winners, demoted } = this.dedupeAutoPicks(autoPicks);
+ for (const w of winners) {
+ updates.push({
+ contactId: w.contact.id,
+ currentEmail: w.contact.email,
+ newEmail: w.row.email,
+ csvRowNumber: w.row.rowNumber,
+ });
+ }
+ skippedAmbiguous += demoted.length;
+
const failures: ApplyResult['failures'] = [];
let updated = 0;
- // Updates run per-row through the repository so the @BeforeUpdate
- // listener (setUserDetails) fires — that re-derives hashed_email and
- // email_provider from the new raw email. A bulk UPDATE would leave both
- // stale and break the SHA-256 contact lookup downstream.
+ // Updates run per-row through save() over an entity instance so the
+ // @BeforeUpdate listener (setUserDetails) fires — that re-derives
+ // hashed_email and email_provider from the new raw email. repo.update()
+ // and bulk UPDATEs skip entity listeners and would leave both stale,
+ // breaking the SHA-256 contact lookup downstream.
for (const u of updates) {
try {
- await this.contactsRepo.update({ id: u.contactId }, { email: u.newEmail });
+ // Friendly pre-check before the unique index does it the hard way:
+ // the email may already belong to another contact in the account
+ // (a clean contact, or one reconciled earlier in this same batch).
+ const holder = await this.contactsRepo.findOne({ where: { accountId: job.accountId!, email: u.newEmail.toLowerCase() } });
+ if (holder && holder.id !== u.contactId) {
+ failures.push({ contactId: u.contactId, reason: `email already in use by contact #${holder.id}` });
+ continue;
+ }
+ await this.contactsRepo.save(this.contactsRepo.create({ id: u.contactId, email: u.newEmail }));
updated++;
} catch (err: any) {
failures.push({ contactId: u.contactId, reason: err?.message ?? String(err) });
@@ -217,19 +292,54 @@ export class EmailReconcileService {
return job;
}
- private parseCsv(csv: string): Promise {
+ private parseCsv(csv: string, ignoreColumns: string[] = []): Promise {
+ const notIgnorable = new Set([...ALWAYS_REQUIRED_CSV_COLUMNS, ...NAME_SIGNAL_COLUMNS]);
+ const ignored = new Set(ignoreColumns.map((c) => c.trim().toLowerCase()).filter((c) => !notIgnorable.has(c)));
return new Promise((resolve, reject) => {
const rows: CsvRow[] = [];
let lineNumber = 0;
- parseString(csv, { headers: true, trim: true })
+ // Exports vary between comma and semicolon; sniff the header line.
+ const headerLine = csv.slice(0, csv.indexOf('\n') === -1 ? csv.length : csv.indexOf('\n'));
+ const delimiter = headerLine.split(';').length > headerLine.split(',').length ? ';' : ',';
+ const stream = parseString(csv, {
+ delimiter,
+ trim: true,
+ // Normalize headers so `Email`/`EMAIL ` still map to `email`.
+ headers: (headers) => headers.map((h) => (h ?? '').trim().toLowerCase()),
+ })
.on('error', reject)
+ .on('headers', (headers: string[]) => {
+ const present = new Set(headers);
+ const missing: string[] = [];
+ if (!present.has('name') && !(present.has('first_name') && present.has('last_name'))) {
+ missing.push(MISSING_NAME_SIGNAL_TOKEN);
+ }
+ missing.push(...ALWAYS_REQUIRED_CSV_COLUMNS.filter((c) => !present.has(c)));
+ if (missing.length > 0) {
+ reject(
+ new BadRequestException({
+ statusCode: 400,
+ error: 'Bad Request',
+ code: MISSING_COLUMNS_ERROR_CODE,
+ missing,
+ message: `CSV is missing required column(s): ${missing.join(', ')}`,
+ }),
+ );
+ stream.destroy(); // don't parse 350k rows after rejecting
+ }
+ })
.on('data', (row: Record) => {
lineNumber++;
+ const createdAtRaw = row.created_at ?? '';
+ // Exports carry either a joined `name` or the first/last pair —
+ // compose per-row so a blank `name` still falls back to the pair.
+ const name = (row.name ?? '').trim() || [row.first_name, row.last_name].filter(Boolean).join(' ').trim();
rows.push({
- name: row.name ?? '',
+ name,
email: (row.email ?? '').toLowerCase(),
- status: row.status ?? '',
- created_at: row.created_at ?? '',
+ status: ignored.has('status') ? '' : (row.status ?? ''),
+ created_at: createdAtRaw,
+ createdAtTs: parseCsvTimestamp(createdAtRaw),
rowNumber: lineNumber,
});
})
@@ -260,24 +370,88 @@ export class EmailReconcileService {
return this.contactsRepo.createQueryBuilder('c').where('c.account_id = :accountId', { accountId }).andWhere("c.email NOT LIKE '%***%'").getCount();
}
+ /** How many masked contacts share each mask — the contact-side collision count. */
+ private countContactsByMask(contacts: ContactEntity[]): Map {
+ const map = new Map();
+ for (const c of contacts) map.set(c.email, (map.get(c.email) ?? 0) + 1);
+ return map;
+ }
+
/**
- * When two or more CSV rows map to the same mask, compare each candidate's
- * `name` to the contact's full name. Return a single confident pick or
- * null. Threshold: Jaccard ≥ 0.8 over lowercased name tokens.
+ * Decides one masked contact against its mask-collision bucket.
+ *
+ * Layered: created_at agreement first, name similarity second.
+ * 1) A single CSV row in the bucket → auto ONLY when the contact is also
+ * the only one with that mask. With N contacts sharing the mask, a
+ * partial CSV (or filtered export) would otherwise assign the same row
+ * to all N — the lone row must earn the pick via time/name like any
+ * other candidate.
+ * 2) Rank candidates by created_at agreement (exact instant > same day >
+ * none) and keep only the best tier. The import preserves the source
+ * created_at, so the true row agrees with the contact by construction —
+ * a lone candidate in a non-zero tier is a confident pick.
+ * 3) Within the surviving tier, name similarity decides as before
+ * (Jaccard ≥ 0.8 with no runner-up at the threshold).
+ * Anything left goes to the operator, carrying only the surviving tier —
+ * time-disagreeing candidates are noise, not options.
*/
- private pickByName(contactName: string, candidates: CsvRow[]): CsvRow | null {
+ private matchContact(
+ contact: ContactEntity,
+ candidates: CsvRow[],
+ maskCollisions: number,
+ ):
+ | { kind: 'auto'; row: CsvRow; timeMatch: TimeMatchLevel; score: number }
+ | { kind: 'ambiguous'; contactName: string; scored: Array<{ row: CsvRow; score: number; timeMatch: TimeMatchLevel }> } {
+ const contactName = `${contact.firstName ?? ''} ${contact.lastName ?? ''}`.trim();
const target = tokenize(contactName);
- if (target.size === 0) return null;
+ const scoreOf = (row: CsvRow) => (target.size === 0 ? 0 : jaccard(target, tokenize(row.name)));
+ // Auto picks always carry their agreement strength so the caller can
+ // arbitrate when two contacts claim the same CSV row (dedupeAutoPicks).
+ const auto = (row: CsvRow, timeMatch: TimeMatchLevel) => ({ kind: 'auto' as const, row, timeMatch, score: scoreOf(row) });
+
+ if (candidates.length === 1 && maskCollisions <= 1) {
+ const row = candidates[0];
+ return auto(row, timeMatchLevel(contact.createdAt, row.createdAtTs));
+ }
- const scored = candidates.map((c) => ({ row: c, score: jaccard(target, tokenize(c.name)) })).sort((a, b) => b.score - a.score);
+ const leveled = candidates.map((row) => ({ row, timeMatch: timeMatchLevel(contact.createdAt, row.createdAtTs) }));
+ const bestLevel = Math.max(...leveled.map((l) => l.timeMatch));
+ const pool = bestLevel > 0 ? leveled.filter((l) => l.timeMatch === bestLevel) : leveled;
+ if (bestLevel > 0 && pool.length === 1) return auto(pool[0].row, pool[0].timeMatch);
+
+ const scored = pool.map((l) => ({ row: l.row, score: scoreOf(l.row), timeMatch: l.timeMatch })).sort((a, b) => b.score - a.score);
const best = scored[0];
const runner = scored[1];
-
if (best.score >= NAME_TIE_BREAK_THRESHOLD && (!runner || runner.score < NAME_TIE_BREAK_THRESHOLD)) {
- return best.row;
+ return { kind: 'auto', row: best.row, timeMatch: best.timeMatch, score: best.score };
+ }
+ return { kind: 'ambiguous', contactName, scored };
+ }
+
+ /**
+ * Email is unique per account (contact_email_unique index), so one CSV row
+ * can reconcile at most ONE contact. When several contacts independently
+ * auto-pick the same row, only the strongest agreement (time tier, then
+ * name score) keeps the auto — the rest are demoted to the ambiguous queue
+ * for the operator, instead of failing later on the unique index.
+ */
+ private dedupeAutoPicks(picks: AutoPick[]): { winners: AutoPick[]; demoted: AutoPick[] } {
+ const byRow = new Map();
+ for (const p of picks) {
+ const group = byRow.get(p.row.rowNumber);
+ if (group) group.push(p);
+ else byRow.set(p.row.rowNumber, [p]);
+ }
+
+ const winners: AutoPick[] = [];
+ const demoted: AutoPick[] = [];
+ for (const group of byRow.values()) {
+ if (group.length > 1) group.sort((a, b) => b.timeMatch - a.timeMatch || b.score - a.score);
+ winners.push(group[0]);
+ demoted.push(...group.slice(1));
}
- return null;
+ return { winners, demoted };
}
}
@@ -287,6 +461,7 @@ function tokenize(name: string): Set {
.toLowerCase()
.normalize('NFD')
.replace(/[̀-ͯ]/g, '') // strip diacritics
+ .replace(/[^\p{L}\p{N}]+/gu, ' ') // punctuation → separator ("Silva, Lucas" tokenizes like "Silva Lucas")
.split(/\s+/)
.filter((t) => t.length > 1), // ignore single-letter middle initials
);
diff --git a/apps/msgops-api/src/modules/enterprise-import/email-reconcile.types.ts b/apps/msgops-api/src/modules/enterprise-import/email-reconcile.types.ts
index ed46fc01..fa1a90e5 100644
--- a/apps/msgops-api/src/modules/enterprise-import/email-reconcile.types.ts
+++ b/apps/msgops-api/src/modules/enterprise-import/email-reconcile.types.ts
@@ -9,31 +9,69 @@ export interface CsvRow {
email: string;
status: string;
created_at: string;
+ // Parsed form of created_at — null when the raw value is empty/unparseable.
+ createdAtTs: ParsedCsvTimestamp | null;
// Original 1-based row number in the CSV — surfaces in reports so the
// operator can find the row in the source file.
rowNumber: number;
}
+// created_at as written in the CSV. Exports may carry no timezone offset, so
+// the epoch is computed as-if-UTC and `hasOffset` tells consumers whether it
+// is trustworthy as an absolute instant.
+export interface ParsedCsvTimestamp {
+ epochMs: number;
+ hasTime: boolean;
+ hasOffset: boolean;
+ // Date part exactly as written (YYYY-MM-DD) — the export's local calendar date.
+ dateISO: string;
+}
+
+// How strongly a candidate's created_at agrees with the contact's:
+// 2 = same minute (seconds/milliseconds ignored on both sides — exports
+// truncate them — tolerating a fixed timezone offset)
+// 1 = same calendar date
+// 0 = no agreement / not comparable
+export type TimeMatchLevel = 0 | 1 | 2;
+
export interface ReconcileMatch {
contactId: number;
currentEmail: string; // masked email already in DB (e.g., lucas***@gmail.com)
newEmail: string; // raw email from CSV
csvRowNumber: number;
+ // Contact's full name — persisted on session items so the operator can
+ // search matches by name, not just by email.
+ contactName?: string;
}
export interface AmbiguousCandidate {
csvRowNumber: number;
csvName: string;
csvEmail: string;
+ // Jaccard token similarity between contact name and csvName (0..1).
+ // Candidates ship sorted by it, best first.
+ score: number;
+ // created_at agreement with the contact (see TimeMatchLevel). Typed as
+ // number because it round-trips through jsonb storage; optional because
+ // sessions persisted before the field existed have candidates without it.
+ timeMatch?: number;
+ // Set when this candidate's row/email was already applied to another
+ // contact in this session. Computed at read time (getAmbiguousPage), never
+ // stored — picking it again would violate the per-account email uniqueness.
+ usedByContactId?: number;
}
export interface AmbiguousMatch {
contactId: number;
currentEmail: string;
contactName: string;
- // CSV rows whose mask collides with this contact. Two or more is the
- // ambiguous case — UI prompts the operator to pick one or skip.
+ // CSV rows whose mask collides with this contact, sorted by score desc and
+ // CAPPED — short masks over big bases collide by the thousands, and an
+ // uncapped list once produced multi-hundred-MB responses. Two or more is
+ // the ambiguous case — UI prompts the operator to pick one or skip.
candidates: AmbiguousCandidate[];
+ // Real candidate count before the cap ("showing 20 of N").
+ candidatesTotal: number;
}
export interface ReconcilePreview {
@@ -64,3 +102,86 @@ export interface ApplyResult {
skippedNoMatch: number;
failures: Array<{ contactId: number; reason: string }>;
}
+
+// ─── Persisted session flow (batched reconciliation) ────────────────────────
+
+// Full in-memory outcome of parsing + matching one CSV against one account.
+// preview() serves a capped view of it; the session flow persists it.
+export interface ReconcileComputation {
+ csvRows: number;
+ invalidCsvRows: number;
+ contactsMasked: number;
+ alreadyClean: number;
+ matches: ReconcileMatch[];
+ // Candidates inside each entry are already scored/sorted but NOT capped —
+ // consumers cap for transport/storage.
+ ambiguous: AmbiguousMatch[];
+ noMatches: Array<{ contactId: number; currentEmail: string }>;
+}
+
+export interface ReconcileSessionProgress {
+ jobId: string;
+ csvRows: number;
+ invalidCsvRows: number;
+ contactsMasked: number;
+ alreadyClean: number;
+ noMatches: number;
+ noMatchSample: Array<{ contactId: number; currentEmail: string }>;
+ // Quantified progress the UI renders as bars/counters.
+ auto: { total: number; applied: number; failed: number; pending: number };
+ ambiguous: { total: number; applied: number; skipped: number; pending: number; failed: number };
+ createdAt: string;
+ updatedAt: string;
+}
+
+export interface AmbiguousPageResult {
+ totalPending: number;
+ offset: number;
+ items: AmbiguousMatch[];
+}
+
+// One row of the session items table — the operator-facing "who matched what"
+// listing (auto picks, applied/failed/skipped outcomes, ambiguous queue).
+export interface ReconcileItemRow {
+ contactId: number;
+ contactName: string;
+ currentEmail: string;
+ kind: 'auto' | 'ambiguous';
+ status: 'pending' | 'applied' | 'skipped' | 'failed';
+ newEmail: string | null;
+ csvRowNumber: number | null;
+ failureReason: string | null;
+ candidatesTotal: number | null;
+}
+
+export interface ReconcileItemsPage {
+ total: number;
+ offset: number;
+ items: ReconcileItemRow[];
+}
+
+export interface ResolveBatchResult {
+ applied: number;
+ skipped: number;
+ // Resolutions that referenced an unknown/already-decided item or a CSV row
+ // not among the stored candidates.
+ invalid: number;
+ failures: Array<{ contactId: number; reason: string }>;
+ progress: ReconcileSessionProgress;
+}
+
+export interface ApplyAutoChunkResult {
+ applied: number;
+ failed: number;
+ remaining: number;
+}
+
+export interface BulkResolveResult {
+ // Items decided+applied by the strategy in this call.
+ resolved: number;
+ // Items examined but left pending (no candidate cleared the threshold).
+ unresolved: number;
+ // Cursor for the next call — null when the pending set is exhausted.
+ nextAfterId: string | null;
+ remainingPending: number;
+}
diff --git a/apps/msgops-api/src/modules/enterprise-import/enterprise-import.controller.ts b/apps/msgops-api/src/modules/enterprise-import/enterprise-import.controller.ts
index ede13e07..34e5be2a 100644
--- a/apps/msgops-api/src/modules/enterprise-import/enterprise-import.controller.ts
+++ b/apps/msgops-api/src/modules/enterprise-import/enterprise-import.controller.ts
@@ -1,12 +1,23 @@
-import { Body, Controller, ForbiddenException, Get, Param, Post, Req, UseGuards } from '@nestjs/common';
+import { Body, Controller, Delete, ForbiddenException, Get, Param, Post, Query, Req, UseGuards } from '@nestjs/common';
import { RequireSuperAdmin } from '../authz/require-super-admin.decorator';
import { EnterpriseImportService } from './enterprise-import.service';
import { EmailReconcileService } from './email-reconcile.service';
+import { EmailReconcileSessionService } from './email-reconcile-session.service';
import { ImportAccountDto, ResumeImportDto } from './dtos/import-account.dto';
import { ImportStatusDto } from './dtos/import-status.dto';
import { ReconcileApplyDto, ReconcilePreviewDto } from './dtos/reconcile.dto';
+import { ReconcileApplyAutoDto, ReconcileBulkResolveDto, ReconcileResolveBatchDto, ReconcileSessionCreateDto } from './dtos/reconcile-session.dto';
import { EnterpriseImportEnabledGuard } from './enterprise-import.guard';
-import type { ApplyResult, ReconcilePreview } from './email-reconcile.types';
+import type {
+ AmbiguousPageResult,
+ ApplyAutoChunkResult,
+ ApplyResult,
+ BulkResolveResult,
+ ReconcileItemsPage,
+ ReconcilePreview,
+ ReconcileSessionProgress,
+ ResolveBatchResult,
+} from './email-reconcile.types';
@Controller()
@UseGuards(EnterpriseImportEnabledGuard)
@@ -14,6 +25,7 @@ export class EnterpriseImportController {
constructor(
private readonly service: EnterpriseImportService,
private readonly reconcileService: EmailReconcileService,
+ private readonly reconcileSessionService: EmailReconcileSessionService,
) {}
// Creates the account with skipDefaults:true and enqueues the import job.
@@ -67,4 +79,79 @@ export class EnterpriseImportController {
reconcileApply(@Param('jobId') jobId: string, @Body() body: ReconcileApplyDto): Promise {
return this.reconcileService.apply(jobId, body.csv, body.resolutions);
}
+
+ // ── Batched reconciliation over a persisted session ──────────────────────
+ // The CSV is parsed/matched once (POST session); from then on the operator
+ // applies auto matches in chunks, reviews ambiguous cases in pages and can
+ // bulk-resolve the tail — all without re-uploading the CSV. Progress is
+ // recomputed from item statuses, so the flow survives page reloads.
+
+ @Post('/imports/:jobId/reconcile/session')
+ @RequireSuperAdmin()
+ createReconcileSession(@Param('jobId') jobId: string, @Body() body: ReconcileSessionCreateDto): Promise {
+ return this.reconcileSessionService.createSession(jobId, body.csv, body.ignoreColumns);
+ }
+
+ @Get('/imports/:jobId/reconcile/session')
+ @RequireSuperAdmin()
+ getReconcileSession(@Param('jobId') jobId: string): Promise {
+ return this.reconcileSessionService.getProgress(jobId);
+ }
+
+ @Get('/imports/:jobId/reconcile/session/ambiguous')
+ @RequireSuperAdmin()
+ getReconcileAmbiguousPage(@Param('jobId') jobId: string, @Query('offset') offset?: string, @Query('limit') limit?: string, @Query('q') q?: string): Promise {
+ const parsedOffset = Math.max(0, Number(offset) || 0);
+ const parsedLimit = Math.min(200, Math.max(1, Number(limit) || 50));
+ return this.reconcileSessionService.getAmbiguousPage(jobId, parsedOffset, parsedLimit, (q ?? '').slice(0, 200));
+ }
+
+ // Flat "who matched what" listing over the session items — searchable by
+ // contact name / masked email / applied email / candidate payload, and
+ // filterable by kind and status. Read-only; resolution stays in /ambiguous.
+ @Get('/imports/:jobId/reconcile/session/items')
+ @RequireSuperAdmin()
+ getReconcileItemsPage(
+ @Param('jobId') jobId: string,
+ @Query('offset') offset?: string,
+ @Query('limit') limit?: string,
+ @Query('q') q?: string,
+ @Query('kind') kind?: string,
+ @Query('status') status?: string,
+ ): Promise {
+ const kinds = ['auto', 'ambiguous'] as const;
+ const statuses = ['pending', 'applied', 'skipped', 'failed'] as const;
+ return this.reconcileSessionService.getItemsPage(jobId, {
+ offset: Math.max(0, Number(offset) || 0),
+ limit: Math.min(200, Math.max(1, Number(limit) || 50)),
+ q: (q ?? '').slice(0, 200),
+ kind: kinds.find((k) => k === kind),
+ status: statuses.find((s) => s === status),
+ });
+ }
+
+ @Post('/imports/:jobId/reconcile/session/resolve')
+ @RequireSuperAdmin()
+ resolveReconcileBatch(@Param('jobId') jobId: string, @Body() body: ReconcileResolveBatchDto): Promise {
+ return this.reconcileSessionService.resolveBatch(jobId, body.resolutions);
+ }
+
+ @Post('/imports/:jobId/reconcile/session/apply-auto')
+ @RequireSuperAdmin()
+ applyReconcileAutoChunk(@Param('jobId') jobId: string, @Body() body: ReconcileApplyAutoDto): Promise {
+ return this.reconcileSessionService.applyAutoChunk(jobId, body.limit);
+ }
+
+ @Post('/imports/:jobId/reconcile/session/bulk-resolve')
+ @RequireSuperAdmin()
+ bulkResolveReconcile(@Param('jobId') jobId: string, @Body() body: ReconcileBulkResolveDto): Promise {
+ return this.reconcileSessionService.bulkResolve(jobId, body.strategy, body.threshold, body.limit, body.afterId);
+ }
+
+ @Delete('/imports/:jobId/reconcile/session')
+ @RequireSuperAdmin()
+ async deleteReconcileSession(@Param('jobId') jobId: string): Promise<{ deleted: true }> {
+ await this.reconcileSessionService.deleteSession(jobId);
+ return { deleted: true };
+ }
}
diff --git a/apps/msgops-api/src/modules/enterprise-import/enterprise-import.module.ts b/apps/msgops-api/src/modules/enterprise-import/enterprise-import.module.ts
index 17bcaaf6..fc99264b 100644
--- a/apps/msgops-api/src/modules/enterprise-import/enterprise-import.module.ts
+++ b/apps/msgops-api/src/modules/enterprise-import/enterprise-import.module.ts
@@ -6,20 +6,23 @@ import { EnterpriseImportJobEntity } from '../../entities/enterprise-import-job.
import { EnterpriseIdMappingEntity } from '../../entities/enterprise-id-mapping.entity';
import { AccountEntity } from '../../entities/account.entity';
import { ContactEntity } from '../../entities/contact.entity';
+import { EmailReconcileSessionEntity } from '../../entities/email-reconcile-session.entity';
+import { EmailReconcileItemEntity } from '../../entities/email-reconcile-item.entity';
import { QUEUE_ENTERPRISE_IMPORT } from '../../providers/queue/queue.constants';
import { EnterpriseImportController } from './enterprise-import.controller';
import { EnterpriseImportService } from './enterprise-import.service';
import { EmailReconcileService } from './email-reconcile.service';
+import { EmailReconcileSessionService } from './email-reconcile-session.service';
import { EnterpriseImportEnabledGuard } from './enterprise-import.guard';
@Module({
imports: [
BullModule.registerQueue({ name: QUEUE_ENTERPRISE_IMPORT }),
- TypeOrmModule.forFeature([EnterpriseImportJobEntity, EnterpriseIdMappingEntity, AccountEntity, ContactEntity]),
+ TypeOrmModule.forFeature([EnterpriseImportJobEntity, EnterpriseIdMappingEntity, AccountEntity, ContactEntity, EmailReconcileSessionEntity, EmailReconcileItemEntity]),
AccountsModule,
],
controllers: [EnterpriseImportController],
- providers: [EnterpriseImportService, EmailReconcileService, EnterpriseImportEnabledGuard],
+ providers: [EnterpriseImportService, EmailReconcileService, EmailReconcileSessionService, EnterpriseImportEnabledGuard],
exports: [EnterpriseImportService],
})
export class EnterpriseImportModule {}
diff --git a/apps/msgops-api/src/modules/enterprise-import/reconcile-timestamp.util.ts b/apps/msgops-api/src/modules/enterprise-import/reconcile-timestamp.util.ts
new file mode 100644
index 00000000..12dfec96
--- /dev/null
+++ b/apps/msgops-api/src/modules/enterprise-import/reconcile-timestamp.util.ts
@@ -0,0 +1,99 @@
+import type { ParsedCsvTimestamp, TimeMatchLevel } from './email-reconcile.types';
+
+// created_at is a match signal because the import worker preserves the
+// Enterprise timestamp verbatim (base.importer.ts keeps source createdAt on
+// insert and never overwrites it on re-import), and the BMS CSV export carries
+// that same source column. When both sides agree, the pair is effectively a
+// natural key on top of the email mask.
+
+// ISO / SQL-ish: YYYY-MM-DD[ T]HH:mm[:ss[.SSS]][Z|±HH[:]MM]
+const ISO_RE = /^(\d{4})-(\d{2})-(\d{2})(?:[ T](\d{2}):(\d{2})(?::(\d{2})(?:\.\d+)?)?\s*(Z|[+-]\d{2}:?\d{2})?)?$/i;
+// Brazilian export format: DD/MM/YYYY[ HH:mm[:ss]]
+const BR_RE = /^(\d{2})\/(\d{2})\/(\d{4})(?:[ T](\d{2}):(\d{2})(?::(\d{2}))?)?$/;
+
+export function parseCsvTimestamp(raw: string): ParsedCsvTimestamp | null {
+ const s = (raw ?? '').trim();
+ if (!s) return null;
+
+ let y: number, mo: number, d: number;
+ let hh = 0;
+ let mi = 0;
+ let ss = 0;
+ let offset: string | undefined;
+ let hasTime = false;
+
+ const iso = s.match(ISO_RE);
+ if (iso) {
+ y = Number(iso[1]);
+ mo = Number(iso[2]);
+ d = Number(iso[3]);
+ if (iso[4] !== undefined) {
+ hasTime = true;
+ hh = Number(iso[4]);
+ mi = Number(iso[5]);
+ ss = iso[6] !== undefined ? Number(iso[6]) : 0;
+ offset = iso[7];
+ }
+ } else {
+ const br = s.match(BR_RE);
+ if (!br) return null;
+ d = Number(br[1]);
+ mo = Number(br[2]);
+ y = Number(br[3]);
+ if (br[4] !== undefined) {
+ hasTime = true;
+ hh = Number(br[4]);
+ mi = Number(br[5]);
+ ss = br[6] !== undefined ? Number(br[6]) : 0;
+ }
+ }
+
+ if (mo < 1 || mo > 12 || d < 1 || d > 31 || hh > 23 || mi > 59 || ss > 59) return null;
+
+ let epochMs = Date.UTC(y, mo - 1, d, hh, mi, ss);
+ let hasOffset = false;
+ if (offset) {
+ hasOffset = true;
+ if (offset.toUpperCase() !== 'Z') {
+ const om = offset.match(/^([+-])(\d{2}):?(\d{2})$/);
+ if (om) {
+ const sign = om[1] === '-' ? -1 : 1;
+ epochMs -= sign * (Number(om[2]) * 60 + Number(om[3])) * 60_000;
+ }
+ }
+ }
+
+ const pad = (n: number) => String(n).padStart(2, '0');
+ return { epochMs, hasTime, hasOffset, dateISO: `${y}-${pad(mo)}-${pad(d)}` };
+}
+
+// Export timezone for BMS data — used only for the day-level comparison, where
+// the calendar date written in the CSV may differ from the UTC date stored in
+// the DB around midnight.
+const EXPORT_TZ_OFFSET_MS = -3 * 3600_000; // America/Sao_Paulo (no DST since 2019)
+
+export function timeMatchLevel(contactCreatedAt: Date | null | undefined, ts: ParsedCsvTimestamp | null): TimeMatchLevel {
+ if (!contactCreatedAt || !ts) return 0;
+ const contactMs = contactCreatedAt.getTime();
+ if (Number.isNaN(contactMs)) return 0;
+
+ if (ts.hasTime) {
+ // Compare at MINUTE precision: exports truncate seconds (and the DB keeps
+ // milliseconds), so anything finer than the minute never agrees between
+ // the two sides. Seconds in the CSV are a bonus we deliberately ignore —
+ // once the minute matches, name similarity breaks any remaining tie.
+ const MINUTE_MS = 60_000;
+ const deltaMin = Math.floor(ts.epochMs / MINUTE_MS) - Math.floor(contactMs / MINUTE_MS);
+ if (ts.hasOffset) {
+ if (deltaMin === 0) return 2;
+ } else {
+ // The export timezone is unknown but constant. Accept a same-minute
+ // match shifted by any fixed half-hour-aligned offset within ±14h.
+ if (deltaMin % 30 === 0 && Math.abs(deltaMin) <= 14 * 60) return 2;
+ }
+ }
+
+ const isoDate = (ms: number) => new Date(ms).toISOString().slice(0, 10);
+ if (ts.dateISO === isoDate(contactMs) || ts.dateISO === isoDate(contactMs + EXPORT_TZ_OFFSET_MS)) return 1;
+ return 0;
+}