import { scriptOutput } from '../lib/script-output.js';

/**
 * Affiliate analytics rollup — AE SQL API → referral_link_stats_daily.
 *
 * Called hourly (MINUTE === 0) from the affiliate tick endpoint.
 * Aggregates click data from Cloudflare Analytics Engine and combines with
 * PG signup/purchase aggregates, then UPSERTs into referral_link_stats_daily.
 *
 * AE SQL API URL:
 *   POST https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/analytics_engine/sql
 *
 * WHY SUM(_sample_interval) NOT COUNT(*):
 *   AE may sample high-volume datasets. COUNT(*) counts sampled rows, not real
 *   events. SUM(_sample_interval) weights each sample by its sampling rate,
 *   reconstructing the true event count even when sampling is active.
 *
 * Schedule: gated behind minute === 0 in tick.ts — NOT a separate cron.
 * The cron tick fires every 30 min; rollup only runs on the :00 tick.
 */

import { eq, sql } from 'drizzle-orm';
import type { DrizzleClient } from '@/server/db/client.js';
import { cronCursors } from '@/server/db/schema.js';
import { advanceCursor, getCursor } from '@/server/db/queries/cron-cursors.js';
import { normMinor } from '@/server/affiliate-module/backfill.js';
import { minorForApi } from '@/server/referrals/ledger.js';

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

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

const JOB_KEY = 'referral-analytics-rollup';
const MAX_BACKFILL_DAYS = 31;
const UPSERT_CHUNK_SIZE = 100;

/**
 * Build AE SQL query for click aggregates on a given calendar day.
 *
 * Uses SUM(_sample_interval) not COUNT(*) — AE may sample high-volume datasets.
 * COUNT(*) would undercount sampled rows by the inverse sampling rate.
 *
 * @param day - ISO date string "YYYY-MM-DD".
 */
export function buildAeQuery(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 = ''",
    'GROUP BY blob1',
  ].join(' ');
}

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

/** UTC midnight for the calendar day of `date`. */
export function utcMidnightFromDate(date: Date): Date {
  return new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate()));
}

/** UTC midnight for the calendar day `daysAgo` days before today (UTC). */
export function utcMidnightDaysAgo(daysAgo: number): Date {
  const now = new Date();
  return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() - daysAgo));
}

export function formatDayUtc(midnight: Date): string {
  return midnight.toISOString().slice(0, 10);
}

export function addUtcDays(midnight: Date, days: number): Date {
  const next = new Date(midnight);
  next.setUTCDate(next.getUTCDate() + days);
  return next;
}

/** Every calendar day strictly after `cursorMidnight` through `throughMidnight` inclusive. */
export function listDaysStrictlyAfterCursor(cursorMidnight: Date, throughMidnight: Date): string[] {
  const days: string[] = [];
  let current = addUtcDays(utcMidnightFromDate(cursorMidnight), 1);
  const through = utcMidnightFromDate(throughMidnight);

  while (current.getTime() <= through.getTime()) {
    days.push(formatDayUtc(current));
    current = addUtcDays(current, 1);
  }

  return days;
}

/**
 * Parse AE SQL API response rows into a link_id-keyed map.
 */
export function parseAeClickRows(
  rows: AeClickRow[],
): Record<string, { clicks: number; clicksSuspicious: number }> {
  const result: Record<string, { clicks: number; clicksSuspicious: number }> = {};
  for (const row of rows) {
    result[row.link_id] = {
      clicks: Number(row.clicks),
      clicksSuspicious: Number(row.clicks_suspicious),
    };
  }
  return result;
}

async function rollupSingleDay(
  db: DrizzleClient,
  env: RollupEnv,
  day: string,
): Promise<RollupResult> {
  // 1. Fetch click aggregates from AE SQL API.
  let 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: buildAeQuery(day) }),
      },
    );

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

    const aeData = (await aeResponse.json()) as { data?: AeClickRow[] };
    clicksByLink = parseAeClickRows(aeData.data ?? []);
  }

  // 2. Signup aggregates from PG.
  const signupRows = (await db.execute(sql`
    SELECT rl.id AS link_id, COUNT(*) AS signups
    FROM referrals r
    JOIN referral_links rl ON rl.id = r.link_id
    WHERE r.bound_at::date = ${day}
    GROUP BY rl.id
  `)) as { rows: Array<{ link_id: string; signups: string | number }> };
  const signupsByLink: Record<string, number> = {};
  for (const row of signupRows.rows as Array<{ link_id: string; signups: string | number }>) {
    signupsByLink[row.link_id] = Number(row.signups);
  }

  // 3. Order + commission aggregates from affiliate_entries (source_type='order_line' post-migration).
  const purchaseRows = (await db.execute(sql`
    SELECT
      rl.id AS link_id,
      COUNT(DISTINCT ae.source_id) AS purchases,
      COALESCE(SUM(le.delta) FILTER (WHERE ae.entry_type = 'affiliate_commission'), 0) AS commission_agorot,
      COALESCE(SUM(le.delta) FILTER (WHERE ae.entry_type = 'referral_reward'), 0) AS reward_agorot
    FROM referral_links rl
    JOIN referrals r ON r.link_id = rl.id
    JOIN affiliate_entries ae ON ae.referral_id = r.id
      AND ae.source_type IN ('order', 'order_line')
      AND ae.entry_type IN ('affiliate_commission', 'referral_reward')
      AND ae.created_at::date = ${day}
    LEFT JOIN ledger_entries le ON le.id = ae.entry_id
    GROUP BY rl.id
  `)) as {
    rows: Array<{
      link_id: string;
      purchases: string | number;
      commission_agorot: string | number;
      reward_agorot: string | number;
    }>;
  };
  const purchasesByLink: Record<
    string,
    { purchases: number; commissionAgorot: number; rewardAgorot: number }
  > = {};
  for (const row of purchaseRows.rows as Array<{
    link_id: string;
    purchases: string | number;
    commission_agorot: string | number;
    reward_agorot: string | number;
  }>) {
    purchasesByLink[row.link_id] = {
      purchases: Number(row.purchases),
      commissionAgorot: minorForApi(normMinor(row.commission_agorot)),
      rewardAgorot: minorForApi(normMinor(row.reward_agorot)),
    };
  }

  // 4. Collect all link IDs across all sources.
  const allLinkIds = new Set([
    ...Object.keys(clicksByLink),
    ...Object.keys(signupsByLink),
    ...Object.keys(purchasesByLink),
  ]);

  if (allLinkIds.size === 0) return { day, linksProcessed: 0, rowsUpserted: 0 };

  // 5. UPSERT into referral_link_stats_daily (chunked multi-row to avoid per-link round-trips).
  const statRows = [...allLinkIds].map((linkId) => ({
    linkId,
    clicks: clicksByLink[linkId]?.clicks ?? 0,
    clicksSuspicious: clicksByLink[linkId]?.clicksSuspicious ?? 0,
    signups: signupsByLink[linkId] ?? 0,
    purchases: purchasesByLink[linkId]?.purchases ?? 0,
    commissionAgorot: purchasesByLink[linkId]?.commissionAgorot ?? 0,
    rewardAgorot: purchasesByLink[linkId]?.rewardAgorot ?? 0,
  }));

  let rowsUpserted = 0;
  for (let i = 0; i < statRows.length; i += UPSERT_CHUNK_SIZE) {
    const chunk = statRows.slice(i, i + UPSERT_CHUNK_SIZE);
    const valueTuples = chunk.map(
      (row) =>
        sql`(${row.linkId}, ${day}, ${row.clicks}, ${row.clicksSuspicious}, ${row.signups}, ${row.purchases}, ${row.commissionAgorot}, ${row.rewardAgorot})`,
    );

    await db.execute(sql`
      INSERT INTO referral_link_stats_daily
        (link_id, day, clicks, clicks_suspicious, signups, purchases, commission_agorot, reward_agorot)
      VALUES ${sql.join(valueTuples, sql`, `)}
      ON CONFLICT (link_id, day) DO UPDATE SET
        clicks             = EXCLUDED.clicks,
        clicks_suspicious  = EXCLUDED.clicks_suspicious,
        signups            = EXCLUDED.signups,
        purchases          = EXCLUDED.purchases,
        commission_agorot  = EXCLUDED.commission_agorot,
        reward_agorot      = EXCLUDED.reward_agorot
    `);
    rowsUpserted += chunk.length;
  }

  return { day, linksProcessed: allLinkIds.size, rowsUpserted };
}

/**
 * Run the hourly analytics rollup with date-granular catch-up.
 *
 * Steps per calendar day (oldest → newest):
 *   1. Query AE SQL API for click aggregates.
 *   2. Query PG for signup aggregates.
 *   3. Query PG for purchase + commission aggregates.
 *   4. UPSERT all link stats into referral_link_stats_daily (chunked).
 *   5. Advance cursor to that day only after the day's UPSERT chunks complete.
 *
 * Only called when MINUTE === 0 (gated in tick.ts — not a separate cron).
 * If CF_ACCOUNT_ID or CF_AE_API_TOKEN are absent, AE step is skipped gracefully.
 */
export async function runAnalyticsRollup(db: DrizzleClient, env: RollupEnv): Promise<RollupResult> {
  const yesterdayMidnight = utcMidnightDaysAgo(1);
  const yesterdayDay = formatDayUtc(yesterdayMidnight);

  const [cursorRow] = await db
    .select()
    .from(cronCursors)
    .where(eq(cronCursors.jobKey, JOB_KEY))
    .limit(1);

  const cursor = await getCursor(db, JOB_KEY, yesterdayMidnight);

  let daysToProcess = listDaysStrictlyAfterCursor(cursor, yesterdayMidnight);
  if (!cursorRow) {
    // First run with no persisted cursor — process yesterday only (legacy behavior).
    daysToProcess = [yesterdayDay];
  }

  if (daysToProcess.length > MAX_BACKFILL_DAYS) {
    const deferredDays = daysToProcess.length - MAX_BACKFILL_DAYS;
    scriptOutput(
      JSON.stringify({
        event: 'referral_analytics_rollup_backfill_deferred',
        deferredDays,
        processingFrom: daysToProcess[0],
        processingThrough: daysToProcess[MAX_BACKFILL_DAYS - 1],
        deferredFrom: daysToProcess[MAX_BACKFILL_DAYS],
        deferredThrough: daysToProcess[daysToProcess.length - 1],
      }),
    );
    daysToProcess = daysToProcess.slice(0, MAX_BACKFILL_DAYS);
  }

  if (daysToProcess.length === 0) {
    return { day: yesterdayDay, linksProcessed: 0, rowsUpserted: 0 };
  }

  let totalLinksProcessed = 0;
  let totalRowsUpserted = 0;
  let lastDay = yesterdayDay;

  for (const day of daysToProcess) {
    const result = await rollupSingleDay(db, env, day);
    totalLinksProcessed += result.linksProcessed;
    totalRowsUpserted += result.rowsUpserted;
    lastDay = day;
    await advanceCursor(db, JOB_KEY, utcMidnightFromDate(new Date(`${day}T00:00:00.000Z`)));
  }

  return {
    day: lastDay,
    linksProcessed: totalLinksProcessed,
    rowsUpserted: totalRowsUpserted,
  };
}