import { and, eq, gte, inArray, sql } from 'drizzle-orm';
import { assetStats, assets, categories, categorySnapshots, correlations, indexConstituents, indexValues, indices, listings, populationReports, priceSnapshots, radarFindings, sales, variantStats } from '@rareindex/database';
import { chainLinkedIndex, correlationMatrix, horizonReturns, type ConstituentSeries } from '@rareindex/indices';
import { logger, median, newId, round, toDateOnly } from '@rareindex/shared';
import { db } from '../lib/db.ts';
import { emit } from '../lib/events.ts';

const log = logger.child({ component: 'indices' });
const DAY = 86_400_000;

interface Constituent {
  assetId: string;
  variantId: string;
  familySlug: string;
  categorySlug: string;
  sales12m: number;
  confidence: number;
}

/** Eligible constituents: valuation confidence ≥ 0.4 and ≥ 3 valid sales in the last 12 months; one variant per asset (most sales). */
async function eligibleConstituents(now: Date): Promise<Constituent[]> {
  const since = new Date(now.getTime() - 365 * DAY);
  const rows = await db()
    .select({ assetId: variantStats.assetId, variantId: variantStats.variantId, confidence: variantStats.rivConfidence, familySlug: assets.familySlug, categorySlug: assets.categorySlug, sales12m: sql<number>`(select count(*)::int from sales s where s.variant_id = ${variantStats.variantId} and s.status = 'valid' and s.sale_date >= ${since.toISOString()}::timestamptz)` })
    .from(variantStats)
    .innerJoin(assets, eq(assets.id, variantStats.assetId))
    .where(and(sql`${variantStats.rivConfidence} >= 0.4`, sql`${variantStats.rivUsd} is not null`));
  const best = new Map<string, Constituent>();
  for (const r of rows) {
    if (r.sales12m < 3) continue;
    const c: Constituent = { assetId: r.assetId, variantId: r.variantId, familySlug: r.familySlug, categorySlug: r.categorySlug, sales12m: r.sales12m, confidence: Number(r.confidence) };
    const prev = best.get(r.assetId);
    if (!prev || c.sales12m > prev.sales12m) best.set(r.assetId, c);
  }
  return [...best.values()];
}

async function seriesFor(cons: Constituent[], baseDate: string): Promise<ConstituentSeries[]> {
  if (!cons.length) return [];
  const out: ConstituentSeries[] = [];
  for (let i = 0; i < cons.length; i += 200) {
    const chunk = cons.slice(i, i + 200);
    const rows = await db()
      .select({ variantId: priceSnapshots.variantId, date: priceSnapshots.date, value: priceSnapshots.rivUsd })
      .from(priceSnapshots)
      .where(and(inArray(priceSnapshots.variantId, chunk.map((c) => c.variantId)), gte(priceSnapshots.date, baseDate), sql`${priceSnapshots.rivUsd} is not null`))
      .orderBy(priceSnapshots.variantId, priceSnapshots.date);
    const byV = new Map<string, Array<{ date: string; value: number }>>();
    for (const r of rows) byV.set(r.variantId, [...(byV.get(r.variantId) ?? []), { date: r.date, value: Number(r.value) }]);
    for (const c of chunk) out.push({ id: c.variantId, points: byV.get(c.variantId) ?? [], weight: c.sales12m });
  }
  return out;
}

interface DayStats {
  transactions: number;
  volume: number | null;
  medianSale: number | null;
  avgSale: number | null;
}
async function dailySaleStats(variantIds: string[], dates: string[]): Promise<Map<string, DayStats>> {
  const m = new Map<string, DayStats>();
  if (!variantIds.length || !dates.length) return m;
  const rows = await db().execute(sql`
    select to_char(sale_date at time zone 'UTC', 'YYYY-MM-DD') as d, count(*)::int as n, sum(coalesce(all_in_usd, price_usd))::float as vol,
           percentile_cont(0.5) within group (order by coalesce(all_in_usd, price_usd))::float as med, avg(coalesce(all_in_usd, price_usd))::float as avg
    from sales where status = 'valid' and variant_id in ${variantIds} and sale_date >= ${dates[0]}::date
    group by 1`);
  for (const r of rows as unknown as Array<{ d: string; n: number; vol: number; med: number; avg: number }>) m.set(r.d, { transactions: r.n, volume: r.vol, medianSale: r.med, avgSale: r.avg });
  return m;
}

/** Market cap estimate (§124): Σ population × RIV where population is known; confidence by coverage. */
async function marketCap(assetIds: string[]): Promise<{ value: number | null; confidence: string }> {
  if (!assetIds.length) return { value: null, confidence: 'insufficient' };
  const rows = await db().execute(sql`
    select coalesce(sum(p.total * s.riv_usd), 0)::float as cap, count(p.asset_id)::int as covered
    from asset_stats s
    left join lateral (select total, asset_id from population_reports pr where pr.asset_id = s.asset_id order by report_date desc limit 1) p on true
    where s.asset_id in ${assetIds} and s.riv_usd is not null`);
  const r = (rows as unknown as Array<{ cap: number; covered: number }>)[0];
  if (!r || r.covered === 0) return { value: null, confidence: 'insufficient' };
  const coverage = r.covered / assetIds.length;
  return { value: r.cap, confidence: coverage >= 0.8 ? 'medium' : coverage >= 0.3 ? 'low' : 'insufficient' };
}

export async function runIndices(opts: { now?: Date } = {}): Promise<{ published: number; skipped: string[] }> {
  const now = opts.now ?? new Date();
  const defs = await db().select().from(indices).where(eq(indices.active, true));
  const cons = await eligibleConstituents(now);
  log.info({ constituents: cons.length }, 'eligible constituents');
  const published: string[] = [];
  const skipped: string[] = [];
  const subSeries: Record<string, Array<{ date: string; value: number; transactions: number; constituents: number }>> = {};
  const today = toDateOnly(now);

  for (const def of defs.filter((d) => !d.isFlagship)) {
    const members = cons.filter((c) => def.familySlugs.includes(c.familySlug));
    const series = await seriesFor(members, def.baseDate);
    const points = chainLinkedIndex(series, { baseDate: def.baseDate, baseValue: def.baseValue, minConstituents: def.minConstituents, weighting: def.weighting === 'liquidity' ? 'liquidity' : 'equal' });
    await syncConstituents(def.id, members, today);
    if (!points.length) {
      skipped.push(`${def.ticker} (${members.length}/${def.minConstituents} constituents)`);
      continue;
    }
    const stats = await dailySaleStats(members.map((m) => m.variantId), points.map((p) => p.date));
    const cap = await marketCap(members.map((m) => m.assetId));
    const [tracked] = await db().select({ n: sql<number>`count(*)::int` }).from(assets).where(inArray(assets.familySlug, def.familySlugs));
    const [liq] = await db().select({ l: sql<number | null>`avg(${assetStats.liquidityScore})::float` }).from(assetStats).where(inArray(assetStats.assetId, members.map((m) => m.assetId)));
    const h = horizonReturns(points);
    const breadth = new Set(members.map((m) => m.categorySlug)).size;
    await db().delete(indexValues).where(eq(indexValues.indexId, def.id)); // series is fully recomputed each run
    const rows = points.map((p) => {
      const s = stats.get(p.date);
      return { indexId: def.id, date: p.date, value: p.value, constituentsCount: p.constituents, transactions: s?.transactions ?? 0, volumeUsd: s?.volume ?? null, medianSaleUsd: s?.medianSale ?? null, avgSaleUsd: s?.avgSale ?? null, marketCapEstUsd: p.date === points.at(-1)!.date ? cap.value : null, marketCapConfidence: p.date === points.at(-1)!.date ? cap.confidence : null, liquidityScore: liq?.l ?? null, momentum: p.date === points.at(-1)!.date ? h['30d'] : null, breadth, trackedAssets: tracked?.n ?? 0, coverage: p.coverage };
    });
    for (let i = 0; i < rows.length; i += 500) {
      await db()
        .insert(indexValues)
        .values(rows.slice(i, i + 500))
        .onConflictDoUpdate({ target: [indexValues.indexId, indexValues.date], set: { value: sql`excluded.value`, constituentsCount: sql`excluded.constituents_count`, transactions: sql`excluded.transactions`, volumeUsd: sql`excluded.volume_usd`, medianSaleUsd: sql`excluded.median_sale_usd`, avgSaleUsd: sql`excluded.avg_sale_usd`, marketCapEstUsd: sql`excluded.market_cap_est_usd`, marketCapConfidence: sql`excluded.market_cap_confidence`, liquidityScore: sql`excluded.liquidity_score`, momentum: sql`excluded.momentum`, breadth: sql`excluded.breadth`, trackedAssets: sql`excluded.tracked_assets`, coverage: sql`excluded.coverage` } });
    }
    subSeries[def.ticker] = rows.map((r) => ({ date: r.date, value: r.value, transactions: r.transactions, constituents: r.constituentsCount }));
    published.push(def.ticker);
    await emit('index_updated', { type: 'index', id: def.id }, { ticker: def.ticker, value: points.at(-1)!.value, date: points.at(-1)!.date, constituents: points.at(-1)!.constituents });
  }

  // RARE: transaction-weighted average of subindex daily log returns
  const flagship = defs.find((d) => d.isFlagship);
  if (flagship) {
    const tickers = Object.keys(subSeries);
    // The flagship publishes only when its own minimum (25 distinct priced constituents across the
    // published subindices, as stated on /rareindex) is met — not as soon as one subindex exists.
    const flagshipConstituents = tickers.reduce((a, t) => a + (subSeries[t]!.at(-1)?.constituents ?? 0), 0);
    if (tickers.length >= 1 && flagshipConstituents >= flagship.minConstituents) {
      const dates = [...new Set(tickers.flatMap((t) => subSeries[t]!.map((p) => p.date)))].sort();
      const last = new Map<string, number>();
      let level = flagship.baseValue;
      const rows: Array<typeof indexValues.$inferInsert> = [];
      let started = false;
      for (const date of dates) {
        let num = 0;
        let den = 0;
        let cnt = 0;
        let tx = 0;
        for (const t of tickers) {
          const p = subSeries[t]!.find((x) => x.date === date);
          if (!p) continue;
          cnt++;
          tx += p.transactions;
          const prev = last.get(t);
          if (prev) {
            const w = 1 + p.transactions;
            num += Math.log(p.value / prev) * w;
            den += w;
          }
          last.set(t, p.value);
        }
        if (!started) {
          started = true;
          rows.push({ indexId: flagship.id, date, value: level, constituentsCount: cnt, transactions: tx, breadth: cnt, coverage: cnt / Math.max(1, defs.length - 1) });
          continue;
        }
        if (den > 0) level = level * Math.exp(num / den);
        rows.push({ indexId: flagship.id, date, value: round(level, 4), constituentsCount: cnt, transactions: tx, breadth: cnt, coverage: cnt / Math.max(1, defs.length - 1) });
      }
      await db().delete(indexValues).where(eq(indexValues.indexId, flagship.id));
      for (let i = 0; i < rows.length; i += 500) {
        await db().insert(indexValues).values(rows.slice(i, i + 500)).onConflictDoUpdate({ target: [indexValues.indexId, indexValues.date], set: { value: sql`excluded.value`, constituentsCount: sql`excluded.constituents_count`, transactions: sql`excluded.transactions`, breadth: sql`excluded.breadth`, coverage: sql`excluded.coverage` } });
      }
      published.push(flagship.ticker);
    } else skipped.push(tickers.length ? `RARE (${flagshipConstituents}/${flagship.minConstituents} constituents across ${tickers.length} subindices)` : 'RARE (no subindex published)');
  }

  // correlations between subindices
  const corrSeries: Record<string, Array<{ date: string; value: number }>> = {};
  for (const [t, s] of Object.entries(subSeries)) corrSeries[t] = s.map((p) => ({ date: p.date, value: p.value }));
  for (const win of [30, 90, 365]) {
    const since = toDateOnly(new Date(now.getTime() - win * DAY));
    for (const c of correlationMatrix(corrSeries, since)) {
      await db().insert(correlations).values({ a: c.a, b: c.b, windowDays: win, coefficient: c.r, observations: c.n, computedAt: now }).onConflictDoUpdate({ target: [correlations.a, correlations.b, correlations.windowDays], set: { coefficient: c.r, observations: c.n, computedAt: now } });
    }
  }
  log.info({ published, skipped }, 'indices run done');
  return { published: published.length, skipped };
}

async function syncConstituents(indexId: string, members: Constituent[], today: string): Promise<void> {
  const current = await db().select({ assetId: indexConstituents.assetId, variantId: indexConstituents.variantId, addedAt: indexConstituents.addedAt }).from(indexConstituents).where(and(eq(indexConstituents.indexId, indexId), sql`${indexConstituents.removedAt} is null`));
  const currentKeys = new Set(current.map((c) => `${c.assetId}|${c.variantId}`));
  const memberKeys = new Set(members.map((m) => `${m.assetId}|${m.variantId}`));
  for (const m of members) {
    if (currentKeys.has(`${m.assetId}|${m.variantId}`)) continue;
    await db().insert(indexConstituents).values({ indexId, assetId: m.assetId, variantId: m.variantId, weight: m.sales12m, addedAt: today, reason: `confidence ${m.confidence.toFixed(2)}, ${m.sales12m} sales/12m` }).onConflictDoNothing();
  }
  for (const c of current) {
    if (memberKeys.has(`${c.assetId}|${c.variantId}`)) continue;
    await db().update(indexConstituents).set({ removedAt: today }).where(and(eq(indexConstituents.indexId, indexId), eq(indexConstituents.assetId, c.assetId), eq(indexConstituents.variantId, c.variantId), eq(indexConstituents.addedAt, c.addedAt)));
  }
}

/** Category snapshots (§125, §151) for every category with tracked assets. */
export async function runCategorySnapshots(opts: { now?: Date } = {}): Promise<number> {
  const now = opts.now ?? new Date();
  const today = toDateOnly(now);
  const since30 = new Date(now.getTime() - 30 * DAY);
  const cats = await db().select({ slug: categories.slug, level: categories.level, family: categories.familySlug }).from(categories);
  const cons = await eligibleConstituents(now);
  let n = 0;
  for (const cat of cats) {
    const scope = cat.level === 0 ? eq(assets.familySlug, cat.slug) : eq(assets.categorySlug, cat.slug);
    const [agg] = await db()
      .select({ tracked: sql<number>`count(*)::int`, valued: sql<number>`count(s.riv_usd)::int` })
      .from(assets)
      .leftJoin(sql`asset_stats s`, sql`s.asset_id = ${assets.id}`)
      .where(scope);
    if (!agg || agg.tracked === 0) continue;
    const [sal] = await db()
      .select({ n: sql<number>`count(*)::int`, vol: sql<number | null>`sum(${sales.priceUsd})::float`, med: sql<number | null>`percentile_cont(0.5) within group (order by ${sales.priceUsd})::float` })
      .from(sales)
      .innerJoin(assets, eq(assets.id, sales.assetId))
      .where(and(scope, eq(sales.status, 'valid'), gte(sales.saleDate, since30)));
    const [lst] = await db().select({ n: sql<number>`count(*)::int` }).from(listings).innerJoin(assets, eq(assets.id, listings.assetId)).where(and(scope, eq(listings.availability, 'available')));
    const [liq] = await db().select({ l: sql<number | null>`avg(s.liquidity_score)::float` }).from(assets).innerJoin(sql`asset_stats s`, sql`s.asset_id = ${assets.id}`).where(scope);
    const members = cons.filter((c) => (cat.level === 0 ? c.familySlug === cat.slug : c.categorySlug === cat.slug));
    let indexValue: number | null = null;
    let ch: Record<string, number | null> = {};
    if (members.length >= 5) {
      const series = await seriesFor(members, '2024-01-01');
      const pts = chainLinkedIndex(series, { baseDate: '2024-01-01', baseValue: 1000, minConstituents: 5 });
      if (pts.length) {
        indexValue = pts.at(-1)!.value;
        ch = horizonReturns(pts);
      }
    }
    const idsForCap = (await db().select({ id: assets.id }).from(assets).where(scope).limit(5000)).map((r) => r.id);
    const cap = await marketCap(idsForCap);
    await db()
      .insert(categorySnapshots)
      .values({ categorySlug: cat.slug, date: today, indexValue, trackedAssets: agg.tracked, assetsWithValuation: agg.valued, sales: sal?.n ?? 0, volumeUsd: sal?.vol ?? null, medianSaleUsd: sal?.med ?? null, activeListings: lst?.n ?? 0, marketCapEstUsd: cap.value, liquidityScore: liq?.l ?? null, change1d: ch['1d'] ?? null, change7d: ch['7d'] ?? null, change30d: ch['30d'] ?? null, change1y: ch['1y'] ?? null })
      .onConflictDoUpdate({ target: [categorySnapshots.categorySlug, categorySnapshots.date], set: { indexValue, trackedAssets: agg.tracked, assetsWithValuation: agg.valued, sales: sal?.n ?? 0, volumeUsd: sal?.vol ?? null, medianSaleUsd: sal?.med ?? null, activeListings: lst?.n ?? 0, marketCapEstUsd: cap.value, liquidityScore: liq?.l ?? null, change1d: ch['1d'] ?? null, change7d: ch['7d'] ?? null, change30d: ch['30d'] ?? null, change1y: ch['1y'] ?? null } });
    n++;
  }
  log.info({ categories: n }, 'category snapshots done');
  return n;
}

/** Rare Radar + record sales (§154–§155). Derived strictly from stored transactions/listings. */
export async function runRadar(opts: { now?: Date } = {}): Promise<number> {
  const now = opts.now ?? new Date();
  let n = 0;
  // record sales per category (all-time and 1y)
  const records = await db().execute(sql`
    select distinct on (a.category_slug) a.category_slug, s.id as sale_id, s.asset_id, s.price_usd::float as price, s.sale_date
    from sales s join assets a on a.id = s.asset_id where s.status = 'valid'
    order by a.category_slug, s.price_usd desc`);
  for (const r of records as unknown as Array<{ category_slug: string; sale_id: string; asset_id: string; price: number; sale_date: Date }>) {
    await db().insert(radarFindings).values({ id: newId('event'), assetId: r.asset_id, kind: 'record_sale', score: Math.log10(1 + r.price), evidence: { categorySlug: r.category_slug, priceUsd: r.price, saleDate: r.sale_date, scope: 'all_time' }, entityType: 'sale', entityId: r.sale_id, detectedAt: now }).onConflictDoUpdate({ target: [radarFindings.kind, radarFindings.entityType, radarFindings.entityId], set: { detectedAt: now, evidence: { categorySlug: r.category_slug, priceUsd: r.price, saleDate: r.sale_date, scope: 'all_time' } } });
    n++;
  }
  // price discrepancy: active listing ≥ 25% below RIV with confidence ≥ 0.6
  const cheap = await db().execute(sql`
    select l.id, l.asset_id, l.discount_to_riv::float as d, l.price_usd::float as ask, s.riv_usd::float as riv, s.riv_confidence::float as conf
    from listings l join asset_stats s on s.asset_id = l.asset_id
    where l.availability = 'available' and l.discount_to_riv <= -0.25 and l.discount_to_riv >= -0.5 and not ('riv_review' = any(l.flags)) and s.riv_confidence >= 0.6 and s.riv_sample_size >= 5 and l.price_usd > 0 limit 2000`);
  for (const r of cheap as unknown as Array<{ id: string; asset_id: string; d: number; ask: number; riv: number; conf: number }>) {
    await db().insert(radarFindings).values({ id: newId('event'), assetId: r.asset_id, kind: 'price_discrepancy', score: -r.d * r.conf, evidence: { askUsd: r.ask, rivUsd: r.riv, discount: r.d, confidence: r.conf }, entityType: 'listing', entityId: r.id, detectedAt: now, expiresAt: new Date(now.getTime() + 7 * DAY) }).onConflictDoUpdate({ target: [radarFindings.kind, radarFindings.entityType, radarFindings.entityId], set: { detectedAt: now, score: -r.d * r.conf, evidence: { askUsd: r.ask, rivUsd: r.riv, discount: r.d, confidence: r.conf } } });
    n++;
  }
  // auction lots whose buyer-pays bid is ≥ 10 % below the variant RIV (§33–§35), ending within 7 days.
  // Neutral wording (§41–§42): a gap between a bid and a valuation, never a statement about the seller.
  const lots = await db().execute(sql`
    select l.id, l.asset_id, l.bid_vs_riv::float as d, l.all_in_bid_usd::float as bid, l.riv_usd_at_assessment::float as riv, l.fee_basis, l.ends_at, l.url, au.auction_house,
           coalesce(vs.riv_confidence, st.riv_confidence)::float as conf
    from auction_lots l join auctions au on au.id = l.auction_id
    left join variant_stats vs on vs.variant_id = l.variant_id left join asset_stats st on st.asset_id = l.asset_id
    where l.status in ('live','upcoming') and l.assessment_verdict = 'deal' and l.bid_vs_riv is not null and l.all_in_bid_usd > 0
      and l.ends_at between now() and now() + interval '7 days' and coalesce(vs.riv_confidence, st.riv_confidence) >= 0.6
    limit 2000`);
  for (const r of lots as unknown as Array<{ id: string; asset_id: string; d: number; bid: number; riv: number; fee_basis: string | null; ends_at: Date; url: string; auction_house: string; conf: number }>) {
    const evidence = { allInBidUsd: r.bid, rivUsd: r.riv, bidVsRiv: r.d, feeBasis: r.fee_basis, endsAt: r.ends_at, house: r.auction_house, url: r.url };
    await db().insert(radarFindings).values({ id: newId('event'), assetId: r.asset_id, kind: 'auction_below_riv', score: -r.d * r.conf, evidence, entityType: 'auction_lot', entityId: r.id, detectedAt: now, expiresAt: new Date(r.ends_at) }).onConflictDoUpdate({ target: [radarFindings.kind, radarFindings.entityType, radarFindings.entityId], set: { detectedAt: now, score: -r.d * r.conf, evidence, expiresAt: new Date(r.ends_at) } });
    n++;
  }
  // first listing in years: asset with a new listing (first_seen last 7d) and no listing/sale in the prior 2 years
  const firsts = await db().execute(sql`
    select l.id, l.asset_id from listings l
    where l.first_seen_at >= now() - interval '7 days'
      and not exists (select 1 from listings l2 where l2.asset_id = l.asset_id and l2.id <> l.id and l2.first_seen_at >= now() - interval '2 years')
      and exists (select 1 from sales s where s.asset_id = l.asset_id and s.sale_date < now() - interval '2 years')
      and not exists (select 1 from sales s where s.asset_id = l.asset_id and s.sale_date >= now() - interval '2 years')
    limit 500`);
  for (const r of firsts as unknown as Array<{ id: string; asset_id: string }>) {
    await db().insert(radarFindings).values({ id: newId('event'), assetId: r.asset_id, kind: 'first_listing_in_years', score: 1, evidence: {}, entityType: 'listing', entityId: r.id, detectedAt: now, expiresAt: new Date(now.getTime() + 14 * DAY) }).onConflictDoNothing();
    n++;
  }
  // ultra-low population: latest population report total ≤ 10 with a valuation
  const lowPop = await db()
    .select({ assetId: populationReports.assetId, total: populationReports.total })
    .from(populationReports)
    .where(sql`${populationReports.total} <= 10 and ${populationReports.reportDate} = (select max(report_date) from population_reports p2 where p2.asset_id = ${populationReports.assetId})`)
    .limit(1000);
  for (const r of lowPop) {
    await db().insert(radarFindings).values({ id: newId('event'), assetId: r.assetId, kind: 'ultra_low_population', score: 1 / (1 + r.total), evidence: { population: r.total }, entityType: 'asset', entityId: r.assetId, detectedAt: now }).onConflictDoUpdate({ target: [radarFindings.kind, radarFindings.entityType, radarFindings.entityId], set: { detectedAt: now, evidence: { population: r.total } } });
    n++;
  }
  log.info({ findings: n }, 'radar done');
  return n;
}

export { median };