Marketing/4ventures-client-portal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make Clientify CRM sync daily and reliable via incremental upsert.

Browse Source 
Webhooks alone miss WooCommerce deals; replace wipe+cap sync with lookback upsert and a dedicated Dokploy cron.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
4Ventures
2026-10-08 11:02:51 +02:00
parent b4e843762e
commit 4baf256750
8 changed files with 419 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

313
src/lib/clientify-sync.ts
View File

@@ -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<Record<string, unknown>> {
async function cfyGet(
token: string,
path: string
): Promise<Record<string, unknown>> {
const r = await fetch(`${CFY}${path}`, {
headers: {
Authorization: `Token ${token}`,
@@ -37,21 +46,67 @@ async function cfyGet(token: string, path: string): Promise<Record<string, unkno
return (await r.json()) as Record<string, unknown>;
}
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, unknown>): 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<Record<string, unknown>[]> {
const out: Record<string, unknown>[] = [];
const seen = new Set<string>();
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<string, unknown>[]) || [];
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<string, string>,
rows: Record<string, unknown>[],
label: string
): Promise<void> {
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, unknown>): 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<string, unknown>,
contactToLead: Map<string, string>
): Record<string, unknown> {
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<ClientifySyncResult> {
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<string, string>();
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<string, Record<string, unknown>>();
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,
})
);
}

5
src/lib/fetch-won-deals.ts
View File

@@ -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 };
});
}

2
src/lib/metrics.ts
View File

@@ -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';
}