import { sql } from 'drizzle-orm';
import type { DrizzleClient } from '@/server/db/client.js';

export type ShareRollupResult = { day: string; linksProcessed: number; rowsUpserted: number };

export type ShareRollupEnv = {
  CF_ACCOUNT_ID?: string;
  CF_AE_API_TOKEN?: string;
};

/**
 * AE SQL query for share click aggregates on a given calendar day.
 * Discriminated from referral clicks by blob8 = 'share'.
 */
export function buildShareAeQuery(day: string): string {
  return [
    'SELECT blob1 AS link_id,',
    'SUM(_sample_interval) AS clicks,',
    'SUM(CASE WHEN double1 = 1 THEN _sample_interval ELSE 0 END) AS clicks_suspicious',
    'FROM `multideal-dev`',
    `WHERE toDate(timestamp) = '${day}'`,
    "  AND blob8 = 'share'",
    'GROUP BY blob1',
  ].join(' ');
}

type AeClickRow = {
  link_id: string;
  clicks: string | number;
  clicks_suspicious: string | number;
};

type AudienceAeRow = {
  device: string;
  country: string;
  clicks: string | number;
  clicks_suspicious: string | number;
};

/** AE SQL query for share click audience aggregates (device + country) on a given day. */
export function buildShareAudienceAeQuery(day: string): string {
  return [
    'SELECT blob3 AS device,',
    'blob4 AS country,',
    'SUM(_sample_interval) AS clicks,',
    'SUM(CASE WHEN double1 = 1 THEN _sample_interval ELSE 0 END) AS clicks_suspicious',
    'FROM `multideal-dev`',
    `WHERE toDate(timestamp) = '${day}'`,
    "  AND blob8 = 'share'",
    'GROUP BY blob3, blob4',
  ].join(' ');
}

/**
 * Run the hourly share analytics rollup.
 *
 * Called from affiliate-tick.ts at MINUTE === 0 (same gate, no new cron needed).
 * Aggregates yesterday's share clicks from AE and UPSERTs into share_link_stats_daily.
 * Denormalizes deal_id + channel from share_links at write time.
 */
export async function runShareAnalyticsRollup(
  db: DrizzleClient,
  env: ShareRollupEnv,
): Promise<ShareRollupResult> {
  const yesterday = new Date();
  yesterday.setUTCDate(yesterday.getUTCDate() - 1);
  const day = yesterday.toISOString().slice(0, 10);

  const clicksByLink: Record<string, { clicks: number; clicksSuspicious: number }> = {};

  if (env.CF_ACCOUNT_ID && env.CF_AE_API_TOKEN) {
    const aeResponse = await fetch(
      `https://api.cloudflare.com/client/v4/accounts/${env.CF_ACCOUNT_ID}/analytics_engine/sql`,
      {
        method: 'POST',
        headers: {
          Authorization: `Bearer ${env.CF_AE_API_TOKEN}`,
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({ query: buildShareAeQuery(day) }),
      },
    );

    if (!aeResponse.ok) {
      throw new Error(`AE SQL API error: ${aeResponse.status} ${await aeResponse.text()}`);
    }

    const aeData = (await aeResponse.json()) as { data?: AeClickRow[] };
    for (const row of aeData.data ?? []) {
      clicksByLink[row.link_id] = {
        clicks: Number(row.clicks),
        clicksSuspicious: Number(row.clicks_suspicious),
      };
    }

    // --- Audience aggregate (device + country breakdown) ---
    try {
      const audienceRes = await fetch(
        `https://api.cloudflare.com/client/v4/accounts/${env.CF_ACCOUNT_ID}/analytics_engine/sql`,
        {
          method: 'POST',
          headers: {
            Authorization: `Bearer ${env.CF_AE_API_TOKEN}`,
            'Content-Type': 'application/json',
          },
          body: JSON.stringify({ query: buildShareAudienceAeQuery(day) }),
        },
      );
      if (!audienceRes.ok) {
        throw new Error(`AE audience error: ${audienceRes.status} ${await audienceRes.text()}`);
      }
      const audienceData = (await audienceRes.json()) as { data?: AudienceAeRow[] };
      for (const row of audienceData.data ?? []) {
        await db.execute(sql`
          INSERT INTO share_audience_daily (day, device, country, clicks, clicks_suspicious)
          VALUES (${day}, ${row.device}, ${row.country}, ${Number(row.clicks)}, ${Number(row.clicks_suspicious)})
          ON CONFLICT (day, device, country) DO UPDATE
            SET clicks            = EXCLUDED.clicks,
                clicks_suspicious = EXCLUDED.clicks_suspicious
        `);
      }
    } catch (err) {
      console.warn('[share-rollup] audience aggregate failed:', err);
    }
  }

  const linkIds = Object.keys(clicksByLink);
  if (linkIds.length === 0) return { day, linksProcessed: 0, rowsUpserted: 0 };

  // Fetch deal_id + channel for each link_id from share_links for denormalization.
  const metaRows = (await db.execute<{ id: string; deal_id: string | null; channel: string }>(sql`
    SELECT id, deal_id, channel FROM share_links WHERE id = ANY(${linkIds}::uuid[])
  `)) as { rows: Array<{ id: string; deal_id: string | null; channel: string }> };
  const metaByLink: Record<string, { dealId: string | null; channel: string }> = {};
  for (const r of metaRows.rows) {
    metaByLink[r.id] = { dealId: r.deal_id, channel: r.channel };
  }

  let rowsUpserted = 0;
  for (const linkId of linkIds) {
    const { clicks, clicksSuspicious } = clicksByLink[linkId]!;
    const meta = metaByLink[linkId];
    const dealId = meta?.dealId ?? null;
    const channel = meta?.channel ?? 'other';

    await db.execute(sql`
      INSERT INTO share_link_stats_daily (link_id, day, clicks, clicks_suspicious, deal_id, channel)
      VALUES (${linkId}, ${day}, ${clicks}, ${clicksSuspicious}, ${dealId}, ${channel})
      ON CONFLICT (link_id, day) DO UPDATE SET
        clicks             = EXCLUDED.clicks,
        clicks_suspicious  = EXCLUDED.clicks_suspicious
    `);
    rowsUpserted++;
  }

  return { day, linksProcessed: linkIds.length, rowsUpserted };
}