import 'server-only';
import { getDb, sql } from '@rareindex/database';

type Row = Record<string, unknown>;
const run = async <T = Row>(q: ReturnType<typeof sql>): Promise<T[]> => (await getDb().execute(q)) as unknown as T[];
const one = async <T = Row>(q: ReturnType<typeof sql>): Promise<T | null> => (await run<T>(q))[0] ?? null;

export async function dashboard() {
  const counts = await one(sql`select
    (select count(*)::int from assets) as assets,
    (select count(*)::int from asset_variants) as variants,
    (select count(*)::int from sales) as sales,
    (select count(*)::int from sales where status = 'flagged') as sales_flagged,
    (select count(*)::int from sales where status = 'excluded') as sales_excluded,
    (select count(*)::int from listings where availability = 'available') as listings_live,
    (select count(*)::int from price_observations) as observations,
    (select count(*)::int from raw_records) as raw_records,
    (select count(*)::int from raw_records where processed_at is null) as raw_unprocessed,
    (select count(*)::int from raw_records where process_error is not null) as raw_errors,
    (select count(*)::int from normalized_records where status = 'pending') as normalized_pending,
    (select count(*)::int from normalized_records where status = 'unmatched') as normalized_unmatched,
    (select count(*)::int from connectors where status = 'active') as connectors_active,
    (select count(*)::int from connectors where status = 'paused') as connectors_paused,
    (select count(*)::int from connector_health where status in ('failing','degraded')) as connectors_unhealthy,
    (select count(*)::int from taxonomy_proposals where status = 'pending') as proposals_pending,
    (select max(sale_date) from sales) as latest_sale,
    (select max(fetched_at) from raw_records) as latest_fetch,
    (select max(computed_at) from valuations) as latest_valuation,
    (select max(date) from index_values) as latest_index,
    (select coalesce(sum(usd_est),0) from costs where occurred_at >= now() - interval '24 hours') as cost_24h_usd,
    (select coalesce(sum(credits),0) from costs where occurred_at >= now() - interval '24 hours' and kind = 'firecrawl') as firecrawl_credits_24h,
    (select coalesce(sum(credits),0) from costs where occurred_at >= now() - interval '24 hours' and kind = 'scrapfly') as scrapfly_credits_24h`);
  const recentRuns = await run(sql`select r.id, r.connector_id, r.status, r.trigger, r.started_at, r.finished_at, r.pages_attempted, r.pages_success, r.records_raw, r.records_normalized, r.records_duplicate, r.error from connector_runs r order by r.started_at desc limit 12`);
  const recentEvents = await run(sql`select type, count(*)::int as n from events where created_at >= now() - interval '24 hours' group by type order by n desc`);
  return { counts, recentRuns, recentEvents };
}

export async function connectorsOverview() {
  return run(sql`with last_run as (
      select distinct on (connector_id) * from connector_runs order by connector_id, started_at desc),
    day as (
      select connector_id, sum(pages_attempted)::int as pages, sum(records_raw)::int as records, sum(records_duplicate)::int as duplicates, sum(cost_credits)::float as credits,
        greatest(1, extract(epoch from (max(coalesce(finished_at, now())) - min(started_at))) / 60.0) as minutes
      from connector_runs where started_at >= now() - interval '24 hours' group by connector_id),
    conf as (
      select connector_id, avg((payload->>'confidence')::float) as parser_confidence, count(*)::int as normalized_24h from normalized_records where created_at >= now() - interval '24 hours' group by connector_id)
    select c.id, c.display_name, c.source_id, c.status, c.priority, c.engine_priority, c.categories, c.refresh_frequency_minutes, c.schema_version, c.connector_version, c.last_run_at, c.last_success_at, c.next_run_at,
      h.status as health_status, h.computed_at as health_at, h.health,
      lr.id as last_run_id, lr.status as last_run_status, lr.error as last_error, lr.started_at as last_started_at, lr.finished_at as last_finished_at, lr.pages_attempted as last_pages, lr.records_raw as last_records, lr.anomalies as last_anomalies,
      d.pages as pages_24h, d.records as records_24h, d.duplicates as duplicates_24h, d.credits as credits_24h, (coalesce(d.pages,0) / coalesce(d.minutes,1)) as pages_per_min,
      cf.parser_confidence, cf.normalized_24h
    from connectors c
    left join connector_health h on h.connector_id = c.id
    left join last_run lr on lr.connector_id = c.id
    left join day d on d.connector_id = c.id
    left join conf cf on cf.connector_id = c.id
    order by c.priority = 'high' desc, c.id`);
}

export async function connectorDetail(id: string) {
  const connector = await one(sql`select * from connectors where id = ${id}`);
  if (!connector) return null;
  const source = await one(sql`select * from sources where id = ${connector.source_id as string}`);
  const health = await one(sql`select * from connector_health where connector_id = ${id}`);
  const runs = await run(sql`select * from connector_runs where connector_id = ${id} order by started_at desc limit 25`);
  const rawSample = await run(sql`select id, kind, url, external_id, engine, fetched_at, http_status, processed_at, process_error, parser_version from raw_records where connector_id = ${id} order by fetched_at desc limit 10`);
  const normalized = await run(sql`select status, count(*)::int as n from normalized_records where connector_id = ${id} group by status`);
  const normalizedSample = await run(sql`select id, kind, status, match_method, match_confidence, asset_id, reject_reason, created_at, payload->>'rawTitle' as raw_title, payload->>'price' as price, payload->>'currency' as currency from normalized_records where connector_id = ${id} order by created_at desc limit 10`);
  const costs = await run(sql`select date_trunc('day', occurred_at)::date as day, kind, sum(credits)::float as credits, sum(usd_est)::float as usd from costs where connector_id = ${id} and occurred_at >= now() - interval '30 days' group by 1, 2 order by 1 desc`);
  const outputs = await one(sql`select (select count(*)::int from sales where connector_id = ${id}) as sales, (select count(*)::int from listings where connector_id = ${id}) as listings, (select count(*)::int from price_observations where connector_id = ${id}) as observations`);
  const anomalies = runs.flatMap((r) => (Array.isArray(r.anomalies) ? (r.anomalies as string[]) : [])).slice(0, 30);
  return { connector, source, health, runs, rawSample, normalized, normalizedSample, costs, outputs, anomalies };
}

export async function costsOverview(days = 30) {
  const byDay = await run(sql`select date_trunc('day', occurred_at)::date as day, kind, sum(credits)::float as credits, sum(usd_est)::float as usd, count(*)::int as events from costs where occurred_at >= now() - (${days}::int || ' days')::interval group by 1, 2 order by 1`);
  const byKind = await run(sql`select kind, coalesce(provider,'') as provider, sum(credits)::float as credits, sum(usd_est)::float as usd, count(*)::int as events from costs where occurred_at >= now() - (${days}::int || ' days')::interval group by 1, 2 order by usd desc`);
  const byConnector = await run(sql`select coalesce(connector_id,'—') as connector_id, sum(credits)::float as credits, sum(usd_est)::float as usd, count(*)::int as events from costs where occurred_at >= now() - (${days}::int || ' days')::interval group by 1 order by usd desc, credits desc limit 30`);
  const byCategory = await run(sql`select coalesce(category_slug,'—') as category_slug, sum(usd_est)::float as usd, sum(credits)::float as credits from costs where occurred_at >= now() - (${days}::int || ' days')::interval group by 1 order by usd desc limit 30`);
  const byEndpoint = await run(sql`select coalesce(endpoint,'—') as endpoint, coalesce(metadata->>'model','') as model, sum(usd_est)::float as usd, count(*)::int as events, sum((metadata->'usage'->>'inputTokens')::float) as input_tokens, sum((metadata->'usage'->>'outputTokens')::float) as output_tokens from costs where kind = 'ai' and occurred_at >= now() - (${days}::int || ' days')::interval group by 1, 2 order by usd desc limit 30`);
  const totals = await one(sql`select sum(usd_est)::float as usd, sum(credits) filter (where kind='firecrawl')::float as firecrawl_credits, sum(credits) filter (where kind='scrapfly')::float as scrapfly_credits, sum(usd_est) filter (where kind='ai')::float as ai_usd, count(*)::int as events from costs where occurred_at >= now() - (${days}::int || ' days')::interval`);
  const perRecord = await one(sql`select (select count(*)::int from raw_records where fetched_at >= now() - (${days}::int || ' days')::interval) as raw_records`);
  return { byDay, byKind, byConnector, byCategory, byEndpoint, totals, perRecord };
}

export async function taxonomyQueue() {
  const pending = await run(sql`select * from taxonomy_proposals where status = 'pending' order by volume_estimate desc nulls last, created_at`);
  const decided = await run(sql`select * from taxonomy_proposals where status <> 'pending' order by decided_at desc nulls last limit 30`);
  const families = await run(sql`select slug, name from categories where level = 0 and active order by sort_order`);
  return { pending, decided, families };
}

export async function dataQuality() {
  const flagged = await run(sql`select s.id, s.asset_id, a.slug as asset_slug, a.title, s.status, s.flags, s.price, s.currency, s.price_usd, s.sale_date, s.source_id, s.source_url, s.raw_title, s.confidence, s.data_quality,
      (select reason from audit_log l where l.entity_type = 'sale' and l.entity_id = s.id order by l.created_at desc limit 1) as last_reason
    from sales s join assets a on a.id = s.asset_id where s.status in ('flagged','excluded') order by s.created_at desc limit 100`);
  const unmatched = await run(sql`select n.id, n.connector_id, n.kind, n.reject_reason, n.match_confidence, n.created_at, n.payload->>'rawTitle' as raw_title, n.payload->>'price' as price, n.payload->>'currency' as currency, n.payload->>'sourceUrl' as source_url, n.payload->'attributes'->>'categorySlug' as category_slug
    from normalized_records n where n.status in ('unmatched','rejected') order by n.created_at desc limit 100`);
  const duplicates = await run(sql`select cross_listing_group_id, count(*)::int as n, min(raw_title) as sample_title, array_agg(source_id) as sources from listings where cross_listing_group_id is not null group by cross_listing_group_id having count(*) > 1 order by n desc limit 30`);
  const quality = await one(sql`select avg(data_quality)::float as sales_avg, percentile_cont(0.1) within group (order by data_quality) as sales_p10, (select avg(data_quality)::float from assets) as assets_avg, (select count(*)::int from sales where confidence < 0.6) as low_confidence_sales from sales`);
  return { flagged, unmatched, duplicates, quality };
}

export async function matchCandidatesForRecord(recordId: string) {
  const rec = await one(sql`select id, payload from normalized_records where id = ${recordId}`);
  if (!rec) return { record: null, candidates: [] };
  const payload = rec.payload as { rawTitle?: string; attributes?: { categorySlug?: string } };
  const title = payload.rawTitle ?? '';
  const candidates = await run(sql`select id, slug, title, category_slug, similarity(title, ${title}) as score from assets where title % ${title} ${payload.attributes?.categorySlug ? sql`and category_slug = ${payload.attributes.categorySlug}` : sql``} order by score desc limit 10`);
  return { record: rec, candidates };
}

export async function auditLog(opts: { entityType?: string; limit?: number } = {}) {
  return run(sql`select * from audit_log where true ${opts.entityType ? sql`and entity_type = ${opts.entityType}` : sql``} order by created_at desc limit ${opts.limit ?? 200}`);
}

export async function eventsLog(opts: { type?: string; limit?: number } = {}) {
  const rows = await run(sql`select * from events where true ${opts.type ? sql`and type = ${opts.type}` : sql``} order by created_at desc limit ${opts.limit ?? 200}`);
  const types = await run(sql`select type, count(*)::int as n from events where created_at >= now() - interval '7 days' group by type order by n desc`);
  return { rows, types };
}