4ventures-client-portal/src/lib/ads-sync.ts

/** Pull Meta / Google Ads insights into ad_daily + roll up kpi_weekly. */

import { serviceHeaders, supabaseUrl } from '@/lib/agency';
import { resolveMetaAccessTokens } 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<string> {
  return resolveGoogleAdsAccessToken();
}

function gadsHeaders(accessToken: string, loginCustomerId?: string) {
  const h: Record<string, string> = {
    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<AdDailyInsert[]> {
  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<AdDailyInsert[]> {
  const tokens = await resolveMetaAccessTokens();
  if (!tokens.length) 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 errors: string[] = [];
  // Prefer agency_oauth (broader BM access); Dokploy env token may only cover BPS.
  for (const token of tokens) {
    try {
      const out: AdDailyInsert[] = [];
      for (const w of windows) {
        out.push(
          ...(await fetchMetaInsightsWindow(
            act,
            token,
            tenantId,
            w.since,
            w.until
          ))
        );
      }
      return out;
    } catch (e) {
      const msg = e instanceof Error ? e.message : String(e);
      errors.push(msg);
      // Try next token on permission / auth failures
      if (!/#200|permission|OAuthException|ads_read|ads_management|invalid/i.test(msg)) {
        throw e;
      }
    }
  }
  throw new Error(errors[errors.length - 1] || 'Meta sync error');
}

async function fetchGoogleWindow(
  cid: string,
  tenantId: string,
  sinceStr: string,
  untilStr: string,
  accessToken: string
): Promise<AdDailyInsert[]> {
  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<NonNullable<typeof j>['results']>[number];
          j = {
            results: parsed.flatMap(
              (chunk: { results?: Row[] }) => chunk.results || []
            ),
          };
        } else {
          j = parsed;
        }
      } catch {
        // NDJSON
        const results: NonNullable<typeof j>['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<AdDailyInsert[]> {
  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<string, Agg>();
  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 };
}