spb/rareindex

1/**2 * Backfill of the buyer-pays price on existing sales (§35): recomputes all_in_usd / fee_basis /3 * buyer_premium_rate from the connector flag, the auction house and the fee schedule. Idempotent4 * and resumable: rows that already carry a fee_basis are skipped unless `force`. Touched assets are5 * marked for revaluation (asset_stats.updated_at pushed back, like `ri regrade`).6 */7import { sql } from 'drizzle-orm';8import { logger } from '@rareindex/shared';9import { db } from './lib/db.ts';10import { saleAllIn } from './entity-resolution/writers.ts';1112const log = logger.child({ component: 'fees-backfill' });1314export interface FeesBackfillResult {15  scanned: number;16  updated: number;17  byBasis: Record<string, number>;18  assets: number;19}2021interface Row {22  id: string;23  asset_id: string;24  price_usd: number;25  currency: string;26  buyer_premium_included: boolean | null;27  auction_house: string | null;28  sale_type: string;29  source_id: string;30  connector_id: string;31  sale_date: Date | string;32  fee_basis: string | null;33}3435export async function backfillFees(opts: { limit?: number; connectorId?: string; force?: boolean; batch?: number } = {}): Promise<FeesBackfillResult> {36  const limit = opts.limit ?? 5_000_000;37  const batch = opts.batch ?? 5000;38  const res: FeesBackfillResult = { scanned: 0, updated: 0, byBasis: {}, assets: 0 };39  const touched = new Set<string>();40  let lastId = '';41  while (res.scanned < limit) {42    const rows = (await db().execute(sql`43      select s.id, s.asset_id, s.price_usd::float as price_usd, s.currency, s.buyer_premium_included, s.auction_house, s.sale_type, s.source_id, s.connector_id, s.sale_date, s.fee_basis44      from sales s45      where s.id > ${lastId}46        ${opts.connectorId ? sql`and s.connector_id = ${opts.connectorId}` : sql``}47        ${opts.force ? sql`` : sql`and s.fee_basis is null`}48      order by s.id limit ${Math.min(batch, limit - res.scanned)}`)) as unknown as Row[];49    if (!rows.length) break;50    res.scanned += rows.length;51    lastId = rows[rows.length - 1]!.id;52    const updates: Array<{ id: string; allIn: number; basis: string; rate: number }> = [];53    for (const r of rows) {54      const f = await saleAllIn({ priceUsd: Number(r.price_usd), currency: r.currency, buyerPremiumIncluded: r.buyer_premium_included, auctionHouse: r.auction_house, saleType: r.sale_type, sourceId: r.source_id, connectorId: r.connector_id, saleDate: new Date(r.sale_date) });55      res.byBasis[f.feeBasis] = (res.byBasis[f.feeBasis] ?? 0) + 1;56      updates.push({ id: r.id, allIn: f.allInUsd, basis: f.feeBasis, rate: f.buyerPremiumRate });57      if (f.feeBasis.startsWith('added_')) touched.add(r.asset_id);58    }59    // one statement per batch; the payload travels as JSON (drizzle serialises JS arrays as records, not SQL arrays)60    await db().execute(sql`61      update sales s set all_in_usd = u.all_in, fee_basis = u.basis, buyer_premium_rate = u.rate62      from json_to_recordset(${JSON.stringify(updates.map((u) => ({ id: u.id, all_in: u.allIn, basis: u.basis, rate: u.rate })))}::json) as u(id text, all_in numeric, basis text, rate real)63      where s.id = u.id`);64    res.updated += updates.length;65    log.info({ scanned: res.scanned, updated: res.updated, lastId }, 'fees backfill progress');66  }67  res.assets = touched.size;68  const ids = [...touched];69  for (let i = 0; i < ids.length; i += 1000) await db().execute(sql`update asset_stats set updated_at = '1970-01-01' where asset_id in ${ids.slice(i, i + 1000)}`);70  return res;71}72