/** Pull Meta / Google Ads insights into ad_daily + roll up kpi_weekly. */ import { serviceHeaders, supabaseUrl } from '@/lib/agency'; import { resolveMetaAccessToken } from '@/lib/ads-discovery'; import { leadFromMetaActions, linkClicksFromMetaActions, nMeta } from '@/lib/meta-leads'; import { resolveGoogleAdsAccessToken } from '@/lib/google-ads-auth'; const META_GRAPH = 'https://graph.facebook.com/v21.0'; const GADS_API = 'https://googleads.googleapis.com/v22'; export type SyncTenantResult = { tenantId: string; tenantName: string; metaRows: number; googleRows: number; weeklyRows: number; errors: string[]; }; type IntegrationRow = { tenant_id: string; provider: string; external_account_id: string | null; status?: string; meta?: { extra_ad_accounts?: { account_id?: string; external_account_id?: string }[]; alternate_accounts?: { external_account_id?: string }[]; } | null; }; function extraAccountIds(row: IntegrationRow | undefined): string[] { if (!row) return []; const out: string[] = []; for (const x of row.meta?.extra_ad_accounts || []) { const id = x.account_id || x.external_account_id; if (id) out.push(id); } for (const x of row.meta?.alternate_accounts || []) { if (x.external_account_id) out.push(x.external_account_id); } return out; } type TenantRow = { id: string; name: string }; type AdDailyInsert = { tenant_id: string; date: string; platform: string; campaign_id: string | null; campaign_name: string | null; ad_id: string | null; ad_name: string | null; impressions: number; clicks: number; spend: number; platform_leads: number; /** Meta: clics en el enlace. Google: igual que clicks. */ link_clicks?: number; }; const n = nMeta; function isoWeekMonday(d: Date): string { const x = new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate())); const day = x.getUTCDay() || 7; if (day !== 1) x.setUTCDate(x.getUTCDate() - (day - 1)); return x.toISOString().slice(0, 10); } async function googleAccessToken(): Promise { return resolveGoogleAdsAccessToken(); } function gadsHeaders(accessToken: string, loginCustomerId?: string) { const h: Record = { Authorization: `Bearer ${accessToken}`, 'developer-token': process.env.GOOGLE_ADS_DEVELOPER_TOKEN!.trim(), 'Content-Type': 'application/json', }; const login = loginCustomerId || process.env.GOOGLE_ADS_LOGIN_CUSTOMER_ID?.trim()?.replace(/-/g, '') || '5963150101'; // MCC 4Ventures if (login) h['login-customer-id'] = login; return h; } function isoDayUTC(d: Date): string { return d.toISOString().slice(0, 10); } /** Inclusive day windows of at most `chunkDays` from `since` → today. */ function dayWindows(days: number, chunkDays: number): { since: string; until: string }[] { const until = new Date(); const start = new Date(); start.setUTCDate(start.getUTCDate() - days); start.setUTCHours(0, 0, 0, 0); until.setUTCHours(0, 0, 0, 0); const out: { since: string; until: string }[] = []; let cursor = new Date(start); while (cursor <= until) { const end = new Date(cursor); end.setUTCDate(end.getUTCDate() + chunkDays - 1); if (end > until) end.setTime(until.getTime()); out.push({ since: isoDayUTC(cursor), until: isoDayUTC(end) }); cursor = new Date(end); cursor.setUTCDate(cursor.getUTCDate() + 1); } return out; } async function fetchMetaInsightsWindow( act: string, token: string, tenantId: string, since: string, until: string ): Promise { const timeRange = JSON.stringify({ since, until }); const out: AdDailyInsert[] = []; let url: string | null = `${META_GRAPH}/${act}/insights` + // Campaign level: ad-level daily lead sums inflate vs Ads Manager period totals. `?fields=campaign_id,campaign_name,impressions,clicks,spend,actions,date_start` + `&level=campaign&time_increment=1&time_range=${encodeURIComponent(timeRange)}` + `&limit=500&access_token=${encodeURIComponent(token)}`; while (url) { const r = await fetch(url, { cache: 'no-store' }); if (!r.ok) { throw new Error(`Meta insights ${act}: ${(await r.text()).slice(0, 220)}`); } const j = (await r.json()) as { data?: { campaign_id?: string; campaign_name?: string; impressions?: string; clicks?: string; spend?: string; actions?: { action_type?: string; value?: string }[]; date_start?: string; }[]; paging?: { next?: string }; }; for (const row of j.data || []) { if (!row.date_start) continue; const campaignId = row.campaign_id || null; out.push({ tenant_id: tenantId, date: row.date_start, platform: 'meta', campaign_id: campaignId, campaign_name: row.campaign_name || null, // campaign-level rows: reuse campaign id as ad_id (NOT NULL column) ad_id: campaignId || '', ad_name: '', impressions: n(row.impressions), // Ads Manager «Clics (todos)» — link clicks se exponen aparte en UI Meta. clicks: n(row.clicks), spend: Math.round(n(row.spend) * 100) / 100, platform_leads: leadFromMetaActions(row.actions), link_clicks: linkClicksFromMetaActions(row.actions, n(row.clicks)), }); } url = j.paging?.next || null; } return out; } async function fetchMetaAdDaily( accountId: string, tenantId: string, days: number ): Promise { const token = await resolveMetaAccessToken(); if (!token) throw new Error('Sin token Meta'); const act = accountId.startsWith('act_') ? accountId : `act_${accountId}`; // Meta rejects huge ad-level ranges — pull in ~30d windows const windows = dayWindows(days, 30); const out: AdDailyInsert[] = []; for (const w of windows) { out.push( ...(await fetchMetaInsightsWindow(act, token, tenantId, w.since, w.until)) ); } return out; } async function fetchGoogleWindow( cid: string, tenantId: string, sinceStr: string, untilStr: string, accessToken: string ): Promise { const query = ` SELECT campaign.id, campaign.name, segments.date, metrics.impressions, metrics.clicks, metrics.cost_micros, metrics.conversions FROM campaign WHERE segments.date BETWEEN '${sinceStr}' AND '${untilStr}' `; // Try with MCC login, then as self const attempts = [ gadsHeaders(accessToken), gadsHeaders(accessToken, cid), ]; let j: { results?: { campaign?: { id?: string; name?: string }; segments?: { date?: string }; metrics?: { impressions?: string; clicks?: string; costMicros?: string; conversions?: number; }; }[]; } | null = null; let lastErr = ''; for (const headers of attempts) { const r = await fetch(`${GADS_API}/customers/${cid}/googleAds:searchStream`, { method: 'POST', headers, body: JSON.stringify({ query }), }); if (r.ok) { const raw = await r.text(); // searchStream returns NDJSON array chunks or JSON array try { const parsed = JSON.parse(raw); if (Array.isArray(parsed)) { type Row = NonNullable['results']>[number]; j = { results: parsed.flatMap( (chunk: { results?: Row[] }) => chunk.results || [] ), }; } else { j = parsed; } } catch { // NDJSON const results: NonNullable['results'] = []; for (const line of raw.split('\n')) { if (!line.trim()) continue; try { const chunk = JSON.parse(line) as { results?: typeof results }; results.push(...(chunk.results || [])); } catch { /* skip */ } } j = { results }; } break; } lastErr = await r.text(); } if (!j) { // fallback search (non-stream) for (const headers of attempts) { const r = await fetch(`${GADS_API}/customers/${cid}/googleAds:search`, { method: 'POST', headers, body: JSON.stringify({ query }), }); if (r.ok) { j = (await r.json()) as typeof j; break; } lastErr = await r.text(); } } if (!j) { throw new Error(`Google Ads ${cid}: ${lastErr.slice(0, 220)}`); } const rawRows: { date: string; campaignId: string | null; campaignName: string | null; impressions: number; clicks: number; spend: number; conversions: number; }[] = []; for (const row of j.results || []) { const date = row.segments?.date; if (!date) continue; const spend = Math.round((n(row.metrics?.costMicros) / 1_000_000) * 100) / 100; const campaignId = row.campaign?.id ? String(row.campaign.id) : null; rawRows.push({ date, campaignId, campaignName: row.campaign?.name || null, impressions: n(row.metrics?.impressions), clicks: n(row.metrics?.clicks), spend, conversions: n(row.metrics?.conversions), }); } // Keep fractional Google conversions (numeric column). UI totals use // Math.round(sum) once so any date range matches Ads Manager. return rawRows.map((r) => ({ tenant_id: tenantId, date: r.date, platform: 'google', campaign_id: r.campaignId, campaign_name: r.campaignName, ad_id: r.campaignId || '', ad_name: '', impressions: r.impressions, clicks: r.clicks, spend: r.spend, platform_leads: Math.round(r.conversions * 1e6) / 1e6, link_clicks: r.clicks, })); } async function fetchGoogleCampaignDaily( customerId: string, tenantId: string, days: number, accessToken: string ): Promise { const cid = customerId.replace(/-/g, ''); const windows = dayWindows(days, 90); const out: AdDailyInsert[] = []; for (const w of windows) { out.push( ...(await fetchGoogleWindow(cid, tenantId, w.since, w.until, accessToken)) ); } return out; } async function replaceDaily( tenantId: string, platforms: string[], rows: AdDailyInsert[], days: number ) { const headers = serviceHeaders(); if (!headers) throw new Error('no service role'); const base = supabaseUrl(); const since = new Date(); since.setUTCDate(since.getUTCDate() - days); const sinceStr = since.toISOString().slice(0, 10); for (const platform of platforms) { await fetch( `${base}/rest/v1/ad_daily?tenant_id=eq.${encodeURIComponent(tenantId)}&platform=eq.${platform}&date=gte.${sinceStr}`, { method: 'DELETE', headers: { ...headers, Prefer: 'return=minimal' } } ); } // chunk inserts — prefer link_clicks column; fallback to Meta `__lc:N` marker for (let i = 0; i < rows.length; i += 400) { const full = rows.slice(i, i + 400).map((r) => ({ ...r, link_clicks: Math.round(r.link_clicks ?? r.clicks), ad_name: r.ad_name || '', })); let r = await fetch(`${base}/rest/v1/ad_daily`, { method: 'POST', headers: { ...headers, 'Content-Type': 'application/json', Prefer: 'return=minimal', }, body: JSON.stringify(full), }); if (!r.ok) { const errText = await r.text(); if (!/link_clicks/i.test(errText)) { throw new Error(`insert ad_daily: ${errText.slice(0, 220)}`); } const legacy = full.map((row) => { const link = Math.round(row.link_clicks ?? row.clicks); const { link_clicks: _lc, ...rest } = row; if (row.platform === 'meta') { return { ...rest, ad_name: `__lc:${link}` }; } return rest; }); r = await fetch(`${base}/rest/v1/ad_daily`, { method: 'POST', headers: { ...headers, 'Content-Type': 'application/json', Prefer: 'return=minimal', }, body: JSON.stringify(legacy), }); if (!r.ok) { throw new Error(`insert ad_daily: ${(await r.text()).slice(0, 220)}`); } } } } async function upsertWeeklyFromDaily(tenantId: string, days: number) { const headers = serviceHeaders(); if (!headers) throw new Error('no service role'); const base = supabaseUrl(); const since = new Date(); since.setUTCDate(since.getUTCDate() - days); const sinceStr = since.toISOString().slice(0, 10); const daily: { date?: string; platform?: string; spend?: number; platform_leads?: number; }[] = []; const pageSize = 1000; for (let offset = 0; offset < 200000; offset += pageSize) { const r = await fetch( `${base}/rest/v1/ad_daily?tenant_id=eq.${encodeURIComponent(tenantId)}&date=gte.${sinceStr}&select=date,platform,spend,platform_leads&order=date.asc&limit=${pageSize}&offset=${offset}`, { headers, cache: 'no-store' } ); if (!r.ok) throw new Error(await r.text()); const chunk = (await r.json()) as typeof daily; daily.push(...chunk); if (chunk.length < pageSize) break; } type Agg = { spend: number; leads: number }; const map = new Map(); for (const row of daily) { if (!row.date) continue; const week = isoWeekMonday(new Date(`${row.date}T00:00:00Z`)); const channel = (row.platform || 'other').toLowerCase(); const key = `${week}|${channel}`; const cur = map.get(key) || { spend: 0, leads: 0 }; cur.spend += n(row.spend); cur.leads += n(row.platform_leads); map.set(key, cur); } // wipe weekly ads channels in window then insert (keep non-ads channels if any) const weeks = [...new Set([...map.keys()].map((k) => k.split('|')[0]))]; for (const week of weeks) { await fetch( `${base}/rest/v1/kpi_weekly?tenant_id=eq.${encodeURIComponent(tenantId)}&week=eq.${week}&channel=in.(meta,google)`, { method: 'DELETE', headers: { ...headers, Prefer: 'return=minimal' } } ); } const payload = [...map.entries()].map(([key, agg]) => { const [week, channel] = key.split('|'); return { tenant_id: tenantId, week: `${week}T00:00:00+00:00`, channel, total_leads: Math.round(agg.leads), total_spend: Math.round(agg.spend * 100) / 100, cohort_deals_won: 0, cohort_revenue_won: 0, cac_cohort: 0, roas_cohort: 0, }; }); if (payload.length) { // kpi_weekly may be a materialized view — best-effort only for (let i = 0; i < payload.length; i += 400) { const chunk = payload.slice(i, i + 400); const ins = await fetch(`${base}/rest/v1/kpi_weekly`, { method: 'POST', headers: { ...headers, 'Content-Type': 'application/json', Prefer: 'resolution=merge-duplicates,return=minimal', }, body: JSON.stringify(chunk), }); if (!ins.ok) { const msg = await ins.text(); if (msg.includes('materialized view')) { // UI falls back to ad_daily; skip hard failure return 0; } throw new Error(`insert kpi_weekly: ${msg.slice(0, 220)}`); } } } return payload.length; } /** Max lookback for intentional backfills (≈ Jan 2024 → now). Cron stays short. */ export const ADS_BACKFILL_MAX_DAYS = 1000; export async function syncAdsForTenants(opts?: { tenantId?: string; days?: number; }): Promise<{ results: SyncTenantResult[]; days: number }> { const headers = serviceHeaders(); if (!headers) throw new Error('no service role'); const base = supabaseUrl(); const days = Math.min( ADS_BACKFILL_MAX_DAYS, Math.max(1, opts?.days ?? 30) ); const tRes = await fetch( `${base}/rest/v1/tenants?select=id,name&agency_followed=neq.false&order=name.asc`, { headers, cache: 'no-store' } ); if (!tRes.ok) throw new Error(await tRes.text()); let tenants = (await tRes.json()) as TenantRow[]; if (opts?.tenantId) { // Explicit id may be archived satellite — still allow const all = await fetch( `${base}/rest/v1/tenants?select=id,name&order=name.asc`, { headers, cache: 'no-store' } ); const allTenants = all.ok ? ((await all.json()) as TenantRow[]) : tenants; tenants = allTenants.filter((t) => t.id === opts.tenantId); } const iRes = await fetch( `${base}/rest/v1/integrations?provider=in.(meta_ads,google_ads)&select=tenant_id,provider,external_account_id,status,meta&external_account_id=not.is.null`, { headers, cache: 'no-store' } ); if (!iRes.ok) throw new Error(await iRes.text()); const integrations = (await iRes.json()) as IntegrationRow[]; let gToken: string | null = null; try { gToken = await googleAccessToken(); } catch { gToken = null; } const results: SyncTenantResult[] = []; for (const tenant of tenants) { const result: SyncTenantResult = { tenantId: tenant.id, tenantName: tenant.name, metaRows: 0, googleRows: 0, weeklyRows: 0, errors: [], }; const mine = integrations.filter((i) => i.tenant_id === tenant.id); const rows: AdDailyInsert[] = []; const platforms: string[] = []; const meta = mine.find((i) => i.provider === 'meta_ads' && i.external_account_id); const metaIds = [ ...(meta?.external_account_id ? [meta.external_account_id] : []), ...extraAccountIds(meta), ]; for (const accountId of [...new Set(metaIds)]) { try { const mrows = await fetchMetaAdDaily(accountId, tenant.id, days); rows.push(...mrows); result.metaRows += mrows.length; } catch (e) { result.errors.push( e instanceof Error ? e.message : 'meta sync error' ); } } if (result.metaRows > 0) platforms.push('meta'); const gads = mine.find( (i) => i.provider === 'google_ads' && i.external_account_id ); const googleIds = [ ...(gads?.external_account_id ? [gads.external_account_id] : []), ...extraAccountIds(gads), ]; if (googleIds.length) { if (!gToken) { result.errors.push('Google Ads sin token OAuth'); } else { for (const customerId of [...new Set(googleIds)]) { try { const grows = await fetchGoogleCampaignDaily( customerId, tenant.id, days, gToken ); rows.push(...grows); result.googleRows += grows.length; } catch (e) { result.errors.push( e instanceof Error ? e.message : 'google sync error' ); } } } } if (result.googleRows > 0) platforms.push('google'); if (rows.length && platforms.length) { try { await replaceDaily(tenant.id, platforms, rows, days); result.weeklyRows = await upsertWeeklyFromDaily(tenant.id, days); const now = new Date().toISOString(); for (const p of platforms) { const provider = p === 'meta' ? 'meta_ads' : 'google_ads'; await fetch( `${base}/rest/v1/integrations?tenant_id=eq.${encodeURIComponent(tenant.id)}&provider=eq.${provider}`, { method: 'PATCH', headers: { ...headers, 'Content-Type': 'application/json', Prefer: 'return=minimal', }, body: JSON.stringify({ last_sync_at: now, error_message: null, }), } ); } } catch (e) { result.errors.push( e instanceof Error ? e.message : 'warehouse write error' ); } } results.push(result); } return { results, days }; }