1import { and, eq, gt, gte, inArray, isNull, lt, sql } from 'drizzle-orm';2import type { Database } from '@rareindex/database';3import { alerts, alertEvents, assets, assetStats, categories, categorySnapshots, indices, indexValues, listings, auctionLots, sales, radarFindings, populationReports, notifications, priceTargets, users } from '@rareindex/database';4import { newId, logger } from '@rareindex/shared';5import { sendMail, alertEmail } from '@rareindex/notify';6import { evaluateAssetAlert, evaluateCategoryAlert, evaluateIndexAlert, inQuietHours, targetHit, type AlertRow, type AssetState, type CategoryState, type IndexState, type Trigger } from './evaluate.js';78const log = logger.child({ job: 'account.alerts' });910interface UserPrefs {11 email: string;12 emailAlerts: boolean;13 quietStart: number | null;14 quietEnd: number | null;15}1617async function loadUsers(db: Database, ids: string[]): Promise<Map<string, UserPrefs>> {18 if (!ids.length) return new Map();19 const rows = await db.select({ id: users.id, email: users.email, prefs: users.preferences, deletedAt: users.deletedAt }).from(users).where(inArray(users.id, ids));20 return new Map(rows.filter((r) => !r.deletedAt).map((r) => {21 const p = (r.prefs ?? {}) as Record<string, unknown>;22 return [r.id, { email: r.email, emailAlerts: p.emailAlerts !== false, quietStart: (p.quietStart as number | null) ?? null, quietEnd: (p.quietEnd as number | null) ?? null }];23 }));24}2526/** Persist a trigger: notification row, alert_events audit row, alert bookkeeping, optional e-mail. */27export async function deliver(db: Database, opts: { userId: string; alertId: string | null; channel: string; trigger: Trigger; prefs: UserPrefs | undefined; kind?: string; now?: Date }): Promise<void> {28 const now = opts.now ?? new Date();29 const wantsEmail = (opts.channel === 'email' || opts.channel === 'both') && opts.prefs?.emailAlerts !== false && opts.prefs?.email;30 const hold = opts.prefs ? inQuietHours(opts.prefs, now) : false;31 let emailedAt: Date | null = null;32 if (wantsEmail && !hold) {33 const res = await sendMail({ to: opts.prefs!.email, ...alertEmail({ title: opts.trigger.title, body: opts.trigger.body, href: `${(process.env.NEXT_PUBLIC_SITE_URL ?? 'https://www.rareindex.io').replace(/\/$/, '')}${opts.trigger.href}`, facts: opts.trigger.facts }), tags: [{ name: 'kind', value: 'alert' }] });34 if (res.ok) emailedAt = now;35 else log.warn({ err: res.error, userId: opts.userId }, 'alert e-mail failed');36 }37 if (opts.channel !== 'email' || !emailedAt) {38 await db.insert(notifications).values({ id: newId('event'), userId: opts.userId, kind: opts.kind ?? 'alert', title: opts.trigger.title, body: opts.trigger.body, href: opts.trigger.href, payload: { facts: opts.trigger.facts, alertId: opts.alertId, heldForQuietHours: hold && Boolean(wantsEmail) }, emailedAt });39 }40 if (opts.alertId) {41 await db.insert(alertEvents).values({ id: newId('event'), alertId: opts.alertId, userId: opts.userId, message: opts.trigger.title, payload: { facts: opts.trigger.facts, href: opts.trigger.href } });42 await db.update(alerts).set({ lastTriggeredAt: now, triggerCount: sql`${alerts.triggerCount} + 1` }).where(eq(alerts.id, opts.alertId));43 }44}4546async function assetState(db: Database, assetId: string, since: Date, now: Date): Promise<AssetState | null> {47 const a = await db.select({ asset: assets, stats: assetStats }).from(assets).leftJoin(assetStats, eq(assetStats.assetId, assets.id)).where(eq(assets.id, assetId)).limit(1);48 const row = a[0];49 if (!row) return null;50 const [newListings, lots, ending, record, baseline, pop] = await Promise.all([51 db.select({ id: listings.id, priceUsd: listings.priceUsd, sourceId: listings.sourceId, firstSeenAt: listings.firstSeenAt }).from(listings).where(and(eq(listings.assetId, assetId), eq(listings.availability, 'available'), gte(listings.firstSeenAt, since))),52 db.select({ id: auctionLots.id, title: auctionLots.title, endsAt: auctionLots.endsAt, auctionHouse: sql<string>`coalesce(${auctionLots.sourceId}, '')` }).from(auctionLots).where(and(eq(auctionLots.assetId, assetId), gte(auctionLots.createdAt, since))),53 db.select({ id: auctionLots.id, title: auctionLots.title, endsAt: auctionLots.endsAt, auctionHouse: sql<string>`coalesce(${auctionLots.sourceId}, '')` }).from(auctionLots).where(and(eq(auctionLots.assetId, assetId), gt(auctionLots.endsAt, now), lt(auctionLots.endsAt, new Date(now.getTime() + 24 * 3600_000)))),54 db.select({ priceUsd: sales.priceUsd, saleDate: sales.saleDate, sourceId: sales.sourceId }).from(sales).where(and(eq(sales.assetId, assetId), eq(sales.status, 'valid'), gte(sales.createdAt, since))).orderBy(sql`${sales.priceUsd} desc`).limit(1),55 db.execute(sql`select count(*)::float / 3 as n from sales where asset_id = ${assetId} and status = 'valid' and sale_date >= ${new Date(now.getTime() - 120 * 86_400_000).toISOString()}::timestamptz and sale_date < ${new Date(now.getTime() - 30 * 86_400_000).toISOString()}::timestamptz`) as unknown as Promise<Array<{ n: number }>>,56 db.select().from(populationReports).where(and(eq(populationReports.assetId, assetId), gte(populationReports.createdAt, since))).orderBy(sql`${populationReports.reportDate} desc`).limit(2),57 ]);58 const s = row.stats;59 const priorAth = s?.athUsd ?? null;60 const rec = record[0];61 const newRecord = rec && (priorAth === null || rec.priceUsd >= priorAth) ? { priceUsd: rec.priceUsd, saleDate: rec.saleDate, sourceId: rec.sourceId } : null;62 let populationChange: AssetState['populationChange'] = null;63 if (pop.length) {64 const latest = pop[0]!;65 const prev = await db.select({ total: populationReports.total }).from(populationReports).where(and(eq(populationReports.assetId, assetId), eq(populationReports.grader, latest.grader), lt(populationReports.reportDate, latest.reportDate))).orderBy(sql`${populationReports.reportDate} desc`).limit(1);66 if (prev[0] && prev[0].total !== latest.total) populationChange = { grader: latest.grader, from: prev[0].total, to: latest.total, date: latest.reportDate };67 }68 return {69 title: row.asset.title,70 slug: row.asset.slug,71 rivUsd: s?.rivUsd ?? null,72 rivConfidence: s?.rivConfidence ?? null,73 rivSampleSize: s?.rivSampleSize ?? 0,74 athUsd: priorAth,75 latestSaleUsd: s?.latestSaleUsd ?? null,76 latestSaleAt: s?.latestSaleAt ?? null,77 sales30d: s?.sales30d ?? 0,78 baselineSales30d: baseline[0]?.n ?? null,79 newListings,80 newAuctionLots: lots,81 endingLots: ending.filter((l): l is typeof l & { endsAt: Date } => l.endsAt !== null),82 newRecordSale: newRecord,83 populationChange,84 };85}8687async function categoryState(db: Database, slug: string, since: Date, now: Date): Promise<CategoryState | null> {88 const c = await db.select().from(categories).where(eq(categories.slug, slug)).limit(1);89 if (!c[0]) return null;90 const snap = await db.select().from(categorySnapshots).where(eq(categorySnapshots.categorySlug, slug)).orderBy(sql`${categorySnapshots.date} desc`).limit(1);91 const [rec, radar, lots, ending, base] = await Promise.all([92 db.execute(sql`select a.title, a.slug, s.price_usd, s.sale_date from sales s join assets a on a.id = s.asset_id join asset_stats st on st.asset_id = a.id where a.category_slug = ${slug} and s.status = 'valid' and s.created_at >= ${since.toISOString()}::timestamptz and (st.ath_usd is null or s.price_usd >= st.ath_usd) order by s.price_usd desc limit 1`) as unknown as Promise<Array<{ title: string; slug: string; price_usd: number; sale_date: Date }>>,93 db.execute(sql`select a.title, a.slug, r.kind, r.score from radar_findings r join assets a on a.id = r.asset_id where a.category_slug = ${slug} and r.detected_at >= ${since.toISOString()}::timestamptz order by r.score desc limit 5`) as unknown as Promise<Array<{ title: string; slug: string; kind: string; score: number }>>,94 db.execute(sql`select count(*)::int as n from auction_lots l join assets a on a.id = l.asset_id where a.category_slug = ${slug} and l.created_at >= ${since.toISOString()}::timestamptz`) as unknown as Promise<Array<{ n: number }>>,95 db.execute(sql`select count(*)::int as n from auction_lots l join assets a on a.id = l.asset_id where a.category_slug = ${slug} and l.ends_at > ${now.toISOString()}::timestamptz and l.ends_at < ${new Date(now.getTime() + 24 * 3600_000).toISOString()}::timestamptz`) as unknown as Promise<Array<{ n: number }>>,96 db.execute(sql`select count(*)::float / 3 as n from sales s join assets a on a.id = s.asset_id where a.category_slug = ${slug} and s.status = 'valid' and s.sale_date >= ${new Date(now.getTime() - 120 * 86_400_000).toISOString()}::timestamptz and s.sale_date < ${new Date(now.getTime() - 30 * 86_400_000).toISOString()}::timestamptz`) as unknown as Promise<Array<{ n: number }>>,97 ]);98 const r = rec[0];99 return {100 name: c[0].name,101 slug,102 change1d: snap[0]?.change1d ?? null,103 newRecordSale: r ? { assetTitle: r.title, assetSlug: r.slug, priceUsd: Number(r.price_usd), saleDate: new Date(r.sale_date) } : null,104 radarFindings: radar.map((x) => ({ assetTitle: x.title, assetSlug: x.slug, kind: x.kind, score: Number(x.score) })),105 newAuctionLots: lots[0]?.n ?? 0,106 endingLots: ending[0]?.n ?? 0,107 sales30d: snap[0]?.sales ?? 0,108 baselineSales30d: base[0]?.n ?? null,109 };110}111112async function indexState(db: Database, ticker: string): Promise<IndexState | null> {113 const i = await db.select().from(indices).where(eq(indices.ticker, ticker)).limit(1);114 if (!i[0]) return null;115 const vals = await db.select({ date: indexValues.date, value: indexValues.value }).from(indexValues).where(eq(indexValues.indexId, i[0].id)).orderBy(sql`${indexValues.date} desc`).limit(2);116 const change1d = vals.length === 2 && vals[1]!.value > 0 ? vals[0]!.value / vals[1]!.value - 1 : null;117 return { ticker, name: i[0].name, change1d, value: vals[0]?.value ?? null };118}119120/** Evaluate every active alert. `since` = last run time (defaults to 1 hour ago). */121export async function runAlerts(db: Database, opts: { since?: Date; now?: Date } = {}): Promise<{ evaluated: number; triggered: number }> {122 const now = opts.now ?? new Date();123 const since = opts.since ?? new Date(now.getTime() - 3600_000);124 const rows = await db.select().from(alerts).where(eq(alerts.active, true));125 const prefs = await loadUsers(db, [...new Set(rows.map((r) => r.userId))]);126 const cacheA = new Map<string, AssetState | null>();127 const cacheC = new Map<string, CategoryState | null>();128 const cacheI = new Map<string, IndexState | null>();129 let triggered = 0;130 for (const a of rows) {131 if (!prefs.has(a.userId)) continue;132 const row: AlertRow = { id: a.id, userId: a.userId, alertType: a.alertType, targetType: a.targetType, targetId: a.targetId, threshold: a.threshold, active: a.active, lastTriggeredAt: a.lastTriggeredAt, cooldownMinutes: a.cooldownMinutes, name: a.name, channel: a.channel };133 let trigger: Trigger | null = null;134 try {135 if (a.targetType === 'asset') {136 if (!cacheA.has(a.targetId)) cacheA.set(a.targetId, await assetState(db, a.targetId, since, now));137 const s = cacheA.get(a.targetId);138 if (s) trigger = evaluateAssetAlert(row, s, now);139 } else if (a.targetType === 'category') {140 if (!cacheC.has(a.targetId)) cacheC.set(a.targetId, await categoryState(db, a.targetId, since, now));141 const s = cacheC.get(a.targetId);142 if (s) trigger = evaluateCategoryAlert(row, s, now);143 } else if (a.targetType === 'index') {144 if (!cacheI.has(a.targetId)) cacheI.set(a.targetId, await indexState(db, a.targetId));145 const s = cacheI.get(a.targetId);146 if (s) trigger = evaluateIndexAlert(row, s, now);147 }148 } catch (err) {149 log.error({ err: err instanceof Error ? err.message : String(err), alertId: a.id }, 'alert evaluation failed');150 continue;151 }152 if (trigger) {153 await deliver(db, { userId: a.userId, alertId: a.id, channel: a.channel, trigger, prefs: prefs.get(a.userId), now });154 triggered++;155 }156 }157 log.info({ evaluated: rows.length, triggered }, 'alerts evaluated');158 return { evaluated: rows.length, triggered };159}160161/** Price targets: notify once when reached (in-app + e-mail per prefs). */162export async function runTargets(db: Database, now = new Date()): Promise<number> {163 const rows = await db.select({ t: priceTargets, asset: assets, stats: assetStats }).from(priceTargets).innerJoin(assets, eq(assets.id, priceTargets.assetId)).leftJoin(assetStats, eq(assetStats.assetId, assets.id)).where(isNull(priceTargets.notifiedAt));164 const prefs = await loadUsers(db, [...new Set(rows.map((r) => r.t.userId))]);165 let n = 0;166 for (const { t, asset, stats } of rows) {167 const riv = stats?.rivUsd ?? null;168 if (!targetHit(t.direction as 'above' | 'below', t.targetUsd, riv)) continue;169 const usd = (v: number) => new Intl.NumberFormat('en-US', { style: 'currency', currency: 'USD', maximumFractionDigits: 0 }).format(v);170 const trigger: Trigger = { kind: 'alert', title: `Target reached: ${asset.title}`, body: `RIV is ${usd(riv!)} — your ${t.direction === 'above' ? 'sell' : 'buy'} target was ${usd(t.targetUsd)}.`, href: `/asset/${asset.slug}`, facts: [['RIV', usd(riv!)], ['Target', usd(t.targetUsd)]] };171 await deliver(db, { userId: t.userId, alertId: null, channel: 'both', trigger, prefs: prefs.get(t.userId), kind: 'target_hit', now });172 await db.update(priceTargets).set({ hitAt: now, notifiedAt: now }).where(eq(priceTargets.id, t.id));173 n++;174 }175 return n;176}177