diff --git a/docs/OPERATOR.md b/docs/OPERATOR.md index a4b9131..95160f3 100644 --- a/docs/OPERATOR.md +++ b/docs/OPERATOR.md @@ -67,7 +67,9 @@ Endpoints individuales (mismo secret): - `/api/cron/organic-sync` — GA4 + GSC → `organic_daily` - `/api/cron/ads-sync` — Meta + Google Ads → `ad_daily` (API v22) - `/api/cron/superhuman-sync` — Seguimiento / Google Docs -- `/api/cron/pipeline-publish` — piezas `aprobado` + `scheduled_at` vencida → Meta IG/FB +- `/api/cron/clientify-sync` — reconciliación CRM Clientify (upsert 60d; diario 05:15 UTC) + (script: `python3 scripts/dokploy_ensure_clientify_sync_cron.py`) +- `/api/cron/pipeline-publish` — piezas `aprobado` + `scheduled_at` vencida → Meta IG/FB (cada ~10 min; script: `python3 scripts/dokploy_ensure_pipeline_publish_cron.py`) ```bash diff --git a/scripts/dokploy_ensure_clientify_sync_cron.py b/scripts/dokploy_ensure_clientify_sync_cron.py new file mode 100644 index 0000000..3930527 --- /dev/null +++ b/scripts/dokploy_ensure_clientify_sync_cron.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""Ensure Dokploy schedule for daily Clientify CRM reconciliation.""" +from __future__ import annotations + +import json +import re +import sys +from pathlib import Path + +from playwright.sync_api import sync_playwright + +ROOT = Path("/Users/4ventures/4ventures-client-portal") +BPS = Path("/Users/4ventures/Projects/bps-facturacion") +BASE = "https://dokploy.4ventures.es" +APP_ID = "NZS797t9Rnq9Pj4mHiii3" +SCHEDULE_NAME = "clientify-sync" +# 05:15 UTC daily — before typical morning review in Spain (07:15) +CRON_EXPR = "15 5 * * *" +CMD = ( + "bash -lc " + "'curl -sS -H \"Authorization: Bearer $CRON_SECRET\" " + "\"https://app.4ventures.es/api/cron/clientify-sync?lookback=60\"'" +) + + +def cred(key: str) -> str: + import os + + v = os.environ.get(key, "").strip() + if v: + return v + for p in (BPS / ".env", ROOT / ".env.local"): + if not p.is_file(): + continue + for line in p.read_text(encoding="utf-8").splitlines(): + if line.startswith(f"{key}="): + return line.split("=", 1)[1].strip().strip('"').strip("'") + return "" + + +def main() -> int: + email = cred("DOKPLOY_EMAIL") or "marketing@4ventures.es" + password = cred("DOKPLOY_PASSWORD") + if not password: + print("missing DOKPLOY_PASSWORD", flush=True) + return 2 + + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + page = browser.new_page(viewport={"width": 1500, "height": 1200}) + page.goto(BASE + "/", wait_until="domcontentloaded", timeout=90000) + page.wait_for_timeout(600) + if "/dashboard" not in page.url: + page.locator('input[type="email"], input[name="email"]').first.fill(email) + page.locator('input[type="password"]').first.fill(password) + page.locator('button[type="submit"]').first.click() + page.wait_for_url(re.compile(r"/dashboard"), timeout=90000) + print("login ok", flush=True) + + result = page.evaluate( + """async ({ appId, name, cron, command }) => { + const post = async (path, body) => { + const r = await fetch(path, { + method: 'POST', + credentials: 'include', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }); + const text = await r.text(); + let json = null; + try { json = JSON.parse(text); } catch {} + return { ok: r.ok, status: r.status, text: text.slice(0, 1200), json }; + }; + const get = async (path) => { + const r = await fetch(path, { credentials: 'include' }); + const text = await r.text(); + let json = null; + try { json = JSON.parse(text); } catch {} + return { ok: r.ok, status: r.status, text: text.slice(0, 2000), json }; + }; + + const listed = await get( + '/api/trpc/schedule.list?batch=1&input=' + + encodeURIComponent(JSON.stringify({ + '0': { json: { id: appId, scheduleType: 'application' } } + })) + ); + const rows = + listed.json?.[0]?.result?.data?.json || + listed.json?.[0]?.result?.data || + []; + const existing = Array.isArray(rows) + ? rows.filter((r) => (r?.name || '') === name) + : []; + + for (const row of existing) { + const sid = row.scheduleId || row.id; + if (!sid) continue; + await post('/api/trpc/schedule.delete?batch=1', { + '0': { json: { scheduleId: sid } }, + }); + } + + const created = await post('/api/trpc/schedule.create?batch=1', { + '0': { + json: { + name, + cronExpression: cron, + shellType: 'bash', + command, + applicationId: appId, + scheduleType: 'application', + enabled: true, + }, + }, + }); + return { deleted: existing.length, created, listedCount: Array.isArray(rows) ? rows.length : 0 }; + }""", + { + "appId": APP_ID, + "name": SCHEDULE_NAME, + "cron": CRON_EXPR, + "command": CMD, + }, + ) + print(json.dumps(result, indent=2)[:4000], flush=True) + ok = bool(result.get("created", {}).get("ok")) + browser.close() + return 0 if ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/app/api/cron/clientify-sync/route.ts b/src/app/api/cron/clientify-sync/route.ts new file mode 100644 index 0000000..5da7291 --- /dev/null +++ b/src/app/api/cron/clientify-sync/route.ts @@ -0,0 +1,56 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { assertCronAuth } from '@/lib/cron-auth'; +import { syncAllClientifyTenants } from '@/lib/clientify-sync'; + +export const dynamic = 'force-dynamic'; +export const maxDuration = 300; + +/** + * Daily Clientify reconciliation (source of truth for CRM deals/leads). + * Webhooks are best-effort; WooCommerce→Clientify often skips outbound events. + * + * Dokploy (daily ~05:15 UTC): + * curl -sS -H "Authorization: Bearer $CRON_SECRET" \ + * "https://app.4ventures.es/api/cron/clientify-sync?lookback=60" + */ +export async function GET(req: NextRequest) { + const denied = assertCronAuth(req); + if (denied) return denied; + + const lookbackDays = Math.min( + 120, + Math.max(14, Number(req.nextUrl.searchParams.get('lookback')) || 60) + ); + const tenantId = req.nextUrl.searchParams.get('tenant_id') || undefined; + + try { + const { results } = await syncAllClientifyTenants({ + tenantId, + lookbackDays, + }); + const withErrors = results.filter((r) => r.error); + return NextResponse.json( + { + ok: withErrors.length === 0, + at: new Date().toISOString(), + lookbackDays, + synced: results.filter((r) => !r.error).length, + withErrors: withErrors.length, + results, + }, + { status: withErrors.length ? 207 : 200 } + ); + } catch (e) { + return NextResponse.json( + { + ok: false, + error: e instanceof Error ? e.message : 'clientify-sync failed', + }, + { status: 500 } + ); + } +} + +export async function POST(req: NextRequest) { + return GET(req); +} diff --git a/src/app/api/cron/daily-sync/route.ts b/src/app/api/cron/daily-sync/route.ts index 0739515..8398bf0 100644 --- a/src/app/api/cron/daily-sync/route.ts +++ b/src/app/api/cron/daily-sync/route.ts @@ -92,10 +92,10 @@ export async function GET(req: NextRequest) { }; } - // 4) Clientify CRM → leads/deals + // 4) Clientify CRM → leads/deals (prefer dedicated /api/cron/clientify-sync) try { const [clientify, gohighlevel] = await Promise.all([ - syncAllClientifyTenants(), + syncAllClientifyTenants({ lookbackDays: Math.max(days * 2, 30) }), syncAllGhlTenants(), ]); const results = [...clientify.results, ...gohighlevel.results]; diff --git a/src/app/api/webhooks/clientify/route.ts b/src/app/api/webhooks/clientify/route.ts index 067ab4a..dcf273c 100644 --- a/src/app/api/webhooks/clientify/route.ts +++ b/src/app/api/webhooks/clientify/route.ts @@ -329,7 +329,7 @@ async function upsertDeal(tenantId: string, d: Json) { else if (/google/.test(tagBlob)) source = 'google'; else if ( /woocommerce|ecommerce/.test(tagBlob) || - /^order\s+\d+/i.test(dealName) + /^(order|pedido)\s+\d+/i.test(dealName) ) { source = 'ecommerce'; } diff --git a/src/lib/clientify-sync.ts b/src/lib/clientify-sync.ts index f231a73..083e217 100644 --- a/src/lib/clientify-sync.ts +++ b/src/lib/clientify-sync.ts @@ -1,4 +1,10 @@ -/** Pull Clientify contacts/deals → warehouse (leads/deals). */ +/** Pull Clientify contacts/deals → warehouse (leads/deals). + * + * Reliability model: + * - Webhooks = near-realtime (best effort; WooCommerce→Clientify often skips them) + * - This sync = source of truth: incremental upsert by stable UUID, no wipe + * - Daily cron pulls everything modified in the lookback window (default 60d) + */ import { createHash } from 'crypto'; import { serviceHeaders, supabaseUrl } from '@/lib/agency'; @@ -23,7 +29,10 @@ function clientifyKeyForSlug(slug: string): string | null { ); } -async function cfyGet(token: string, path: string): Promise> { +async function cfyGet( + token: string, + path: string +): Promise> { const r = await fetch(`${CFY}${path}`, { headers: { Authorization: `Token ${token}`, @@ -37,21 +46,67 @@ async function cfyGet(token: string, path: string): Promise; } -async function cfyPaginate( +function dayKey(s: unknown): string | null { + if (!s || typeof s !== 'string') return null; + const d = s.slice(0, 10); + return /^\d{4}-\d{2}-\d{2}$/.test(d) ? d : null; +} + +function rowActivityDay(row: Record): string | null { + return ( + dayKey(row.modified) || + dayKey(row.actual_closed_date) || + dayKey(row.created) + ); +} + +/** + * Paginate Clientify list endpoints newest-first until the lookback window + * is exhausted (or maxPages). Never deletes warehouse rows. + */ +async function cfyPaginateRecent( token: string, path: string, - maxPages = 40 + opts: { + sinceDay: string; + maxPages?: number; + ordering?: string; + } ): Promise[]> { const out: Record[] = []; + const seen = new Set(); const base = path.split('?')[0] || path; - const params = new URLSearchParams(path.includes('?') ? path.split('?')[1] : ''); + const params = new URLSearchParams( + path.includes('?') ? path.split('?')[1] : '' + ); params.set('page_size', '100'); + params.set('ordering', opts.ordering || '-modified'); + const maxPages = opts.maxPages ?? 80; + for (let page = 1; page <= maxPages; page++) { params.set('page', String(page)); const j = await cfyGet(token, `${base}?${params}`); const batch = (j.results as Record[]) || []; - out.push(...batch); - if (!j.next || !batch.length) break; + if (!batch.length) break; + + let newestOnPage: string | null = null; + for (const row of batch) { + const id = String(row.id ?? ''); + const activity = rowActivityDay(row); + if (activity && (!newestOnPage || activity > newestOnPage)) { + newestOnPage = activity; + } + // Keep rows in window, or without dates (safer than dropping) + if (activity && activity < opts.sinceDay) continue; + if (id && seen.has(id)) continue; + if (id) seen.add(id); + out.push(row); + } + + // Newest item on this page is already older than lookback → further + // pages (with -modified) cannot add anything in-window. + if (newestOnPage && newestOnPage < opts.sinceDay) break; + if (!j.next) break; } return out; } @@ -105,11 +160,99 @@ function contactIdFromUrl(url: unknown): string | null { return parts[parts.length - 1] || null; } +function lookbackSinceDay(days: number): string { + const d = new Date(); + d.setUTCDate(d.getUTCDate() - Math.max(1, days)); + return d.toISOString().slice(0, 10); +} + +async function upsertChunks( + url: string, + headers: Record, + rows: Record[], + label: string +): Promise { + for (let i = 0; i < rows.length; i += 100) { + const chunk = rows.slice(i, i + 100); + const r = await fetch(url, { + method: 'POST', + headers: { + ...headers, + 'Content-Type': 'application/json', + // Stable uuid5 primary keys → safe upsert, no wipe + Prefer: 'resolution=merge-duplicates,return=minimal', + }, + body: JSON.stringify(chunk), + }); + if (!r.ok) { + throw new Error(`${label}: ${(await r.text()).slice(0, 220)}`); + } + } +} + +function dealSource(d: Record): string | null { + let rawSrc = String(d.deal_source || d.contact_source || '') + .trim() + .slice(0, 80); + const tags = Array.isArray(d.tags) + ? d.tags.map(String).join(' ').toLowerCase() + : ''; + const name = String(d.name || '').slice(0, 200); + if (!rawSrc) { + if (/facebook|instagram|meta|\bfb\b/.test(tags)) rawSrc = 'meta'; + else if (/google/.test(tags)) rawSrc = 'google'; + else if ( + /woocommerce/.test(tags) || + /^(order|pedido)\s+\d+/i.test(name) + ) + rawSrc = 'ecommerce'; + } + return rawSrc ? rawSrc.toLowerCase() : null; +} + +function mapDealRow( + tid: string, + d: Record, + contactToLead: Map +): Record { + const did = String(d.id); + const stage = mapStage(d); + const cid = contactIdFromUrl(d.contact); + // Only link lead_id when the lead row exists (FK). Missing contacts + // stay null; a later sync/webhook that upserts the contact can fill it. + const leadId = cid ? contactToLead.get(cid) || null : null; + const amount = Number(d.amount || d.amount_user || 0) || 0; + // Won without actual_closed_date still counts (Woo often sets created only) + const closed = + parseDate(d.actual_closed_date) || + (stage === 'won' ? parseDate(d.created) : null); + const pipeUrl = String(d.pipeline || ''); + const pipeId = pipeUrl.includes('/pipelines/') + ? pipeUrl.replace(/\/$/, '').split('/').pop() || null + : null; + const name = String(d.name || '').slice(0, 200); + return { + id: uuid5(`clientify-deal:${tid}:${did}`), + tenant_id: tid, + lead_id: leadId, + crm_deal_id: `clientify:${did}`, + stage, + amount, + won_at: stage === 'won' ? closed : null, + lost_at: stage === 'lost' ? closed : null, + pipeline_id: pipeId, + pipeline_name: d.pipeline_desc ? String(d.pipeline_desc) : null, + source: dealSource(d), + name: name || null, + }; +} + export type ClientifySyncResult = { tenantId: string; tenantSlug: string; leads: number; deals: number; + lookbackDays?: number; error?: string; }; @@ -117,15 +260,19 @@ export async function syncClientifyTenant(opts: { tenantId: string; slug: string; apiKey?: string | null; + /** Days of modified contacts/deals to reconcile (default 60). */ + lookbackDays?: number; }): Promise { const headers = serviceHeaders(); const base = supabaseUrl(); + const lookbackDays = opts.lookbackDays ?? 60; if (!headers || !base) { return { tenantId: opts.tenantId, tenantSlug: opts.slug, leads: 0, deals: 0, + lookbackDays, error: 'no service role', }; } @@ -137,29 +284,49 @@ export async function syncClientifyTenant(opts: { tenantSlug: opts.slug, leads: 0, deals: 0, + lookbackDays, error: `Falta CLIENTIFY_API_KEY (o CLIENTIFY_API_KEY_${opts.slug.toUpperCase().replace(/-/g, '_')})`, }; } const tid = opts.tenantId; - try { - const contacts = await cfyPaginate(token, '/contacts/'); - await fetch( - `${base}/rest/v1/leads?tenant_id=eq.${encodeURIComponent(tid)}&crm=eq.clientify`, - { method: 'DELETE', headers: { ...headers, Prefer: 'return=minimal' } } - ); + const sinceDay = lookbackSinceDay(lookbackDays); + try { + // 0) Existing warehouse leads so deal FK links survive when a contact + // was last modified outside the lookback window. const contactToLead = new Map(); + for (let offset = 0; offset < 20000; offset += 1000) { + const lr = await fetch( + `${base}/rest/v1/leads?tenant_id=eq.${encodeURIComponent(tid)}&crm=eq.clientify&select=id,crm_contact_id&limit=1000&offset=${offset}`, + { headers, cache: 'no-store' } + ); + if (!lr.ok) break; + const rows = (await lr.json()) as { + id: string; + crm_contact_id?: string | null; + }[]; + if (!rows.length) break; + for (const row of rows) { + if (row.crm_contact_id) contactToLead.set(row.crm_contact_id, row.id); + } + if (rows.length < 1000) break; + } + + // 1) Contacts modified in window → upsert leads (no DELETE) + const contacts = await cfyPaginateRecent(token, '/contacts/', { + sinceDay, + maxPages: 100, + }); const leadRows = contacts.map((c) => { const cid = String(c.id); const lid = uuid5(`clientify-lead:${tid}:${cid}`); contactToLead.set(cid, lid); - const source = String( - c.contact_source || c.channel || c.medium || 'clientify' - ) - .trim() - .toLowerCase() - .slice(0, 80) || 'clientify'; + const source = + String(c.contact_source || c.channel || c.medium || 'clientify') + .trim() + .toLowerCase() + .slice(0, 80) || 'clientify'; const status = String(c.status || 'new') .trim() @@ -183,82 +350,28 @@ export async function syncClientifyTenant(opts: { created_at: parseDate(c.created) || new Date().toISOString(), }; }); + await upsertChunks(`${base}/rest/v1/leads`, headers, leadRows, 'upsert leads'); - for (let i = 0; i < leadRows.length; i += 200) { - const chunk = leadRows.slice(i, i + 200); - const r = await fetch(`${base}/rest/v1/leads`, { - method: 'POST', - headers: { - ...headers, - 'Content-Type': 'application/json', - Prefer: 'return=minimal', - }, - body: JSON.stringify(chunk), - }); - if (!r.ok) { - throw new Error(`insert leads: ${(await r.text()).slice(0, 180)}`); - } + // 2) Deals: two passes — all recent + won-only (status=3) for sales safety net + const dealById = new Map>(); + for (const d of await cfyPaginateRecent(token, '/deals/', { + sinceDay, + maxPages: 120, + })) { + dealById.set(String(d.id), d); + } + for (const d of await cfyPaginateRecent(token, '/deals/?status=3', { + sinceDay, + maxPages: 120, + ordering: '-modified', + })) { + dealById.set(String(d.id), d); } - const deals = await cfyPaginate(token, '/deals/'); - await fetch( - `${base}/rest/v1/deals?tenant_id=eq.${encodeURIComponent(tid)}&crm_deal_id=like.clientify:*`, - { method: 'DELETE', headers: { ...headers, Prefer: 'return=minimal' } } + const dealRows = [...dealById.values()].map((d) => + mapDealRow(tid, d, contactToLead) ); - - const dealRows = deals.map((d) => { - const did = String(d.id); - const stage = mapStage(d); - const cid = contactIdFromUrl(d.contact); - const leadId = cid ? contactToLead.get(cid) || null : null; - const amount = Number(d.amount || d.amount_user || 0) || 0; - const closed = parseDate(d.actual_closed_date); - const pipeUrl = String(d.pipeline || ''); - const pipeId = pipeUrl.includes('/pipelines/') - ? pipeUrl.replace(/\/$/, '').split('/').pop() || null - : null; - let rawSrc = String(d.deal_source || d.contact_source || '') - .trim() - .slice(0, 80); - const tags = Array.isArray(d.tags) ? d.tags.map(String).join(' ').toLowerCase() : ''; - const name = String(d.name || '').slice(0, 200); - if (!rawSrc) { - if (/facebook|instagram|meta|\bfb\b/.test(tags)) rawSrc = 'meta'; - else if (/google/.test(tags)) rawSrc = 'google'; - else if (/woocommerce/.test(tags) || /^order /i.test(name)) - rawSrc = 'ecommerce'; - } - return { - id: uuid5(`clientify-deal:${tid}:${did}`), - tenant_id: tid, - lead_id: leadId, - crm_deal_id: `clientify:${did}`, - stage, - amount, - won_at: stage === 'won' ? closed : null, - lost_at: stage === 'lost' ? closed : null, - pipeline_id: pipeId, - pipeline_name: d.pipeline_desc ? String(d.pipeline_desc) : null, - source: rawSrc ? rawSrc.toLowerCase() : null, - name: name || null, - }; - }); - - for (let i = 0; i < dealRows.length; i += 200) { - const chunk = dealRows.slice(i, i + 200); - const r = await fetch(`${base}/rest/v1/deals`, { - method: 'POST', - headers: { - ...headers, - 'Content-Type': 'application/json', - Prefer: 'return=minimal', - }, - body: JSON.stringify(chunk), - }); - if (!r.ok) { - throw new Error(`insert deals: ${(await r.text()).slice(0, 180)}`); - } - } + await upsertChunks(`${base}/rest/v1/deals`, headers, dealRows, 'upsert deals'); const now = new Date().toISOString(); await fetch( @@ -274,6 +387,13 @@ export async function syncClientifyTenant(opts: { status: 'connected', last_sync_at: now, error_message: null, + meta: { + source: 'cron', + synced_at: now, + lookback_days: lookbackDays, + leads_upserted: leadRows.length, + deals_upserted: dealRows.length, + }, }), } ); @@ -283,6 +403,7 @@ export async function syncClientifyTenant(opts: { tenantSlug: opts.slug, leads: leadRows.length, deals: dealRows.length, + lookbackDays, }; } catch (e) { return { @@ -290,6 +411,7 @@ export async function syncClientifyTenant(opts: { tenantSlug: opts.slug, leads: 0, deals: 0, + lookbackDays, error: e instanceof Error ? e.message : 'clientify sync failed', }; } @@ -298,6 +420,7 @@ export async function syncClientifyTenant(opts: { /** Sync all tenants with clientify integration (+ API key available). */ export async function syncAllClientifyTenants(opts?: { tenantId?: string; + lookbackDays?: number; }): Promise<{ results: ClientifySyncResult[] }> { const headers = serviceHeaders(); const base = supabaseUrl(); @@ -308,7 +431,11 @@ export async function syncAllClientifyTenants(opts?: { { headers, cache: 'no-store' } ); if (!tRes.ok) throw new Error(await tRes.text()); - let tenants = (await tRes.json()) as { id: string; slug: string; name: string }[]; + let tenants = (await tRes.json()) as { + id: string; + slug: string; + name: string; + }[]; if (opts?.tenantId) { tenants = tenants.filter((t) => t.id === opts.tenantId); } @@ -329,7 +456,6 @@ export async function syncAllClientifyTenants(opts?: { const results: ClientifySyncResult[] = []; for (const t of tenants) { const integ = byTenant.get(t.id); - // Sync if marked connected/pending OR we have a key for this slug const key = (integ?.meta && typeof integ.meta.api_key === 'string' ? integ.meta.api_key @@ -352,6 +478,7 @@ export async function syncAllClientifyTenants(opts?: { tenantId: t.id, slug: t.slug, apiKey: key, + lookbackDays: opts?.lookbackDays, }) ); } diff --git a/src/lib/fetch-won-deals.ts b/src/lib/fetch-won-deals.ts index 0117007..8c31f4e 100644 --- a/src/lib/fetch-won-deals.ts +++ b/src/lib/fetch-won-deals.ts @@ -16,7 +16,7 @@ async function enrichDealSources( ...d, source: d.source || - (/^order\s+\d+/i.test(d.name || '') ? 'ecommerce' : d.source) || + (/^(order|pedido)\s+\d+/i.test(d.name || '') ? 'ecommerce' : d.source) || null, })); } @@ -62,7 +62,8 @@ async function enrichDealSources( return deals.map((d) => { let source = d.source || (d.lead_id ? leadSource.get(d.lead_id) : null) || null; - if (!source && /^order\s+\d+/i.test(d.name || '')) source = 'ecommerce'; + if (!source && /^(order|pedido)\s+\d+/i.test(d.name || '')) + source = 'ecommerce'; return { ...d, source }; }); } diff --git a/src/lib/metrics.ts b/src/lib/metrics.ts index a1024dd..4101050 100644 --- a/src/lib/metrics.ts +++ b/src/lib/metrics.ts @@ -742,7 +742,7 @@ export function inferDealSource(d: { if (/google/.test(tags)) return 'google'; if ( /woocommerce|ecommerce/.test(tags) || - /^order\s+\d+/i.test(d.name || '') + /^(order|pedido)\s+\d+/i.test(d.name || '') ) { return 'ecommerce'; }