TypeScript 61.9%
HTML 37.2%
SQL 0.7%
1import 'server-only';2import { getDb, sql } from '@rareindex/database';34type Row = Record<string, unknown>;5const run = async <T = Row>(q: ReturnType<typeof sql>): Promise<T[]> => (await getDb().execute(q)) as unknown as T[];6const one = async <T = Row>(q: ReturnType<typeof sql>): Promise<T | null> => (await run<T>(q))[0] ?? null;78export async function dashboard() {9 const counts = await one(sql`select10 (select count(*)::int from assets) as assets,11 (select count(*)::int from asset_variants) as variants,12 (select count(*)::int from sales) as sales,13 (select count(*)::int from sales where status = 'flagged') as sales_flagged,14 (select count(*)::int from sales where status = 'excluded') as sales_excluded,15 (select count(*)::int from listings where availability = 'available') as listings_live,16 (select count(*)::int from price_observations) as observations,17 (select count(*)::int from raw_records) as raw_records,18 (select count(*)::int from raw_records where processed_at is null) as raw_unprocessed,19 (select count(*)::int from raw_records where process_error is not null) as raw_errors,20 (select count(*)::int from normalized_records where status = 'pending') as normalized_pending,21 (select count(*)::int from normalized_records where status = 'unmatched') as normalized_unmatched,22 (select count(*)::int from connectors where status = 'active') as connectors_active,23 (select count(*)::int from connectors where status = 'paused') as connectors_paused,24 (select count(*)::int from connector_health where status in ('failing','degraded')) as connectors_unhealthy,25 (select count(*)::int from taxonomy_proposals where status = 'pending') as proposals_pending,26 (select max(sale_date) from sales) as latest_sale,27 (select max(fetched_at) from raw_records) as latest_fetch,28 (select max(computed_at) from valuations) as latest_valuation,29 (select max(date) from index_values) as latest_index,30 (select coalesce(sum(usd_est),0) from costs where occurred_at >= now() - interval '24 hours') as cost_24h_usd,31 (select coalesce(sum(credits),0) from costs where occurred_at >= now() - interval '24 hours' and kind = 'firecrawl') as firecrawl_credits_24h,32 (select coalesce(sum(credits),0) from costs where occurred_at >= now() - interval '24 hours' and kind = 'scrapfly') as scrapfly_credits_24h`);33 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`);34 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`);35 return { counts, recentRuns, recentEvents };36}3738export async function connectorsOverview() {39 return run(sql`with last_run as (40 select distinct on (connector_id) * from connector_runs order by connector_id, started_at desc),41 day as (42 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,43 greatest(1, extract(epoch from (max(coalesce(finished_at, now())) - min(started_at))) / 60.0) as minutes44 from connector_runs where started_at >= now() - interval '24 hours' group by connector_id),45 conf as (46 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),47 spend as (48 select connector_id, sum(credits) filter (where kind = 'firecrawl')::float as fc_credits_24h, sum(credits) filter (where kind = 'scrapfly')::float as sf_credits_24h from costs where occurred_at >= now() - interval '24 hours' and connector_id is not null group by connector_id),49 bf as (50 select distinct on (connector_id) connector_id, id as backfill_id, status as backfill_status, percent as backfill_percent, pages_processed as backfill_pages, items_processed as backfill_items, updated_at as backfill_updated_at, reached_date as backfill_reached_date from connector_backfills order by connector_id, started_at desc)51 select c.id, c.display_name, c.source_id, c.status, c.priority, c.engine_priority, c.categories, c.regions, c.refresh_frequency_minutes, c.schema_version, c.connector_version, c.last_run_at, c.last_success_at, c.next_run_at, c.meta,52 h.status as health_status, h.computed_at as health_at, h.health,53 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,54 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,55 cf.parser_confidence, cf.normalized_24h,56 sp.fc_credits_24h, sp.sf_credits_24h,57 bf.backfill_id, bf.backfill_status, bf.backfill_percent, bf.backfill_pages, bf.backfill_items, bf.backfill_updated_at, bf.backfill_reached_date58 from connectors c59 left join connector_health h on h.connector_id = c.id60 left join last_run lr on lr.connector_id = c.id61 left join day d on d.connector_id = c.id62 left join conf cf on cf.connector_id = c.id63 left join spend sp on sp.connector_id = c.id64 left join bf on bf.connector_id = c.id65 order by c.priority = 'high' desc, c.id`);66}6768export async function connectorDetail(id: string) {69 const connector = await one(sql`select * from connectors where id = ${id}`);70 if (!connector) return null;71 const source = await one(sql`select * from sources where id = ${connector.source_id as string}`);72 const health = await one(sql`select * from connector_health where connector_id = ${id}`);73 const runs = await run(sql`select * from connector_runs where connector_id = ${id} order by started_at desc limit 25`);74 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`);75 const normalized = await run(sql`select status, count(*)::int as n from normalized_records where connector_id = ${id} group by status`);76 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`);77 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`);78 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, (select count(*)::int from certificates where ${id} = any(source_ids) or last_source_url like '%' || ${connector.source_id as string} || '%') as certificates`);79 const backfills = await run(sql`select * from connector_backfills where connector_id = ${id} order by started_at desc limit 10`);80 const fieldStats = await run(sql`select field, day::text as day, total, nulls from connector_field_stats where connector_id = ${id} and day >= (current_date - interval '7 days')::date order by day desc, field`);81 const anomalies = runs.flatMap((r) => (Array.isArray(r.anomalies) ? (r.anomalies as string[]) : [])).slice(0, 30);82 return { connector, source, health, runs, rawSample, normalized, normalizedSample, costs, outputs, anomalies, backfills, fieldStats };83}8485export async function rawRecordDetail(id: string) {86 return one(sql`select id, connector_id, source_id, run_id, engine, url, external_id, kind, fetched_at, content_hash, http_status, payload, snapshot_ref, parser_version, connector_version, processed_at, process_error from raw_records where id = ${id}`);87}8889export async function normalizedRecordDetail(id: string) {90 return one(sql`select n.*, r.url as raw_url, r.engine as raw_engine from normalized_records n left join raw_records r on r.id = n.raw_record_id where n.id = ${id}`);91}9293/** Global coverage dashboard (SPEC §30). */94export async function coverage() {95 const totals = await one(sql`select96 (select count(*)::int from connectors) as connectors,97 (select count(*)::int from connectors where status = 'active') as connectors_active,98 (select count(*)::int from listings where availability = 'available') as live_listings,99 (select count(*)::int from sales) as sales,100 (select count(*)::int from sales where sale_date < now() - interval '1 year') as sales_older_1y,101 (select min(sale_date) from sales) as oldest_sale,102 (select count(*)::int from auction_lots) as auction_lots,103 (select count(distinct auction_house) from auctions) as auction_houses,104 (select count(distinct grader) from certificates) as graders_with_certs,105 (select count(*)::int from certificates) as certificates,106 (select count(*)::int from population_reports) as population_reports,107 (select count(*)::int from assets) as assets,108 (select count(*)::int from asset_variants) as variants,109 (select count(*)::int from price_observations) as observations,110 (select count(*)::int from raw_records where fetched_at >= now() - interval '24 hours') as raw_24h,111 (select count(*)::int from normalized_records where created_at >= now() - interval '24 hours') as normalized_24h,112 (select count(*)::int from sales where created_at >= now() - interval '24 hours') as sales_24h,113 (select count(*)::int from listings where first_seen_at >= now() - interval '24 hours') as listings_24h,114 (select count(*)::int from connector_backfills where status = 'running') as backfills_running,115 (select count(*)::int from connector_backfills where status = 'completed') as backfills_completed`);116 const byStatus = await run(sql`select coalesce(h.status, 'unknown') as status, count(*)::int as n from connectors c left join connector_health h on h.connector_id = c.id group by 1 order by n desc`);117 const byCategory = await run(sql`select cat as category, count(distinct c.id)::int as connectors, count(distinct c.id) filter (where h.status = 'healthy')::int as healthy from connectors c cross join unnest(c.categories) as cat left join connector_health h on h.connector_id = c.id group by 1 order by connectors desc, 1`);118 const byCountry = await run(sql`select coalesce(c.meta->>'country', c.regions[1], 'global') as country, count(*)::int as connectors from connectors c group by 1 order by connectors desc`);119 const bySourceType = await run(sql`select s.source_type, count(*)::int as connectors from connectors c join sources s on s.id = c.source_id group by 1 order by 2 desc`);120 const byEngine = await run(sql`select engine_priority[1] as engine, count(*)::int as connectors from connectors group by 1 order by 2 desc`);121 const byCapability = await run(sql`select cap as capability, count(*)::int as connectors from connectors c cross join jsonb_array_elements_text(coalesce(c.meta->'capabilities', '[]'::jsonb)) as cap group by 1 order by 2 desc`);122 const salesByFamily = await run(sql`select a.family_slug, count(*)::int as sales, count(*) filter (where s.sale_date >= now() - interval '30 days')::int as sales_30d, min(s.sale_date)::date as oldest from sales s join assets a on a.id = s.asset_id group by 1 order by sales desc`);123 const listingsByFamily = await run(sql`select a.family_slug, count(*)::int as live from listings l join assets a on a.id = l.asset_id where l.availability = 'available' group by 1 order by live desc`);124 const daily = await run(sql`select d::date as day,125 (select count(*)::int from raw_records r where r.fetched_at >= d and r.fetched_at < d + interval '1 day') as raw,126 (select count(*)::int from sales s where s.created_at >= d and s.created_at < d + interval '1 day') as sales,127 (select count(*)::int from listings l where l.first_seen_at >= d and l.first_seen_at < d + interval '1 day') as listings128 from generate_series(current_date - interval '13 days', current_date, interval '1 day') d order by 1`);129 const salesByYear = await run(sql`select extract(year from sale_date)::int as year, count(*)::int as sales from sales group by 1 order by 1`);130 return { totals, byStatus, byCategory, byCountry, bySourceType, byEngine, byCapability, salesByFamily, listingsByFamily, daily, salesByYear };131}132133export async function costsOverview(days = 30) {134 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`);135 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`);136 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`);137 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`);138 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`);139 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`);140 const perRecord = await one(sql`select (select count(*)::int from raw_records where fetched_at >= now() - (${days}::int || ' days')::interval) as raw_records`);141 return { byDay, byKind, byConnector, byCategory, byEndpoint, totals, perRecord };142}143144export async function taxonomyQueue() {145 const pending = await run(sql`select * from taxonomy_proposals where status = 'pending' order by volume_estimate desc nulls last, created_at`);146 const decided = await run(sql`select * from taxonomy_proposals where status <> 'pending' order by decided_at desc nulls last limit 30`);147 const families = await run(sql`select slug, name from categories where level = 0 and active order by sort_order`);148 return { pending, decided, families };149}150151export async function dataQuality() {152 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,153 (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_reason154 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`);155 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_slug156 from normalized_records n where n.status in ('unmatched','rejected') order by n.created_at desc limit 100`);157 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`);158 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`);159 return { flagged, unmatched, duplicates, quality };160}161162export async function matchCandidatesForRecord(recordId: string) {163 const rec = await one(sql`select id, payload from normalized_records where id = ${recordId}`);164 if (!rec) return { record: null, candidates: [] };165 const payload = rec.payload as { rawTitle?: string; attributes?: { categorySlug?: string } };166 const title = payload.rawTitle ?? '';167 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`);168 return { record: rec, candidates };169}170171export async function auditLog(opts: { entityType?: string; limit?: number } = {}) {172 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}`);173}174175export async function eventsLog(opts: { type?: string; limit?: number } = {}) {176 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}`);177 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`);178 return { rows, types };179}180