TypeScript 61.9%
HTML 37.2%
SQL 0.7%
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 belowRiv = (await db.execute(sql`select l.id, l.title, l.ends_at, coalesce(au.auction_house, l.source_id) as house, l.all_in_bid_usd::float as bid, l.riv_usd_at_assessment::float as riv, l.bid_vs_riv::float as d, l.fee_basis51 from auction_lots l left join auctions au on au.id = l.auction_id52 where l.asset_id = ${assetId} and l.status in ('live','upcoming') and l.assessment_verdict = 'deal' and l.bid_vs_riv is not null and (l.ends_at is null or l.ends_at > now()) limit 5`)) as unknown as Array<{ id: string; title: string; ends_at: Date | null; house: string; bid: number; riv: number; d: number; fee_basis: string | null }>;53 const [newListings, lots, ending, record, baseline, pop] = await Promise.all([54 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))),55 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))),56 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)))),57 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),58 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 }>>,59 db.select().from(populationReports).where(and(eq(populationReports.assetId, assetId), gte(populationReports.createdAt, since))).orderBy(sql`${populationReports.reportDate} desc`).limit(2),60 ]);61 const s = row.stats;62 const priorAth = s?.athUsd ?? null;63 const rec = record[0];64 const newRecord = rec && (priorAth === null || rec.priceUsd >= priorAth) ? { priceUsd: rec.priceUsd, saleDate: rec.saleDate, sourceId: rec.sourceId } : null;65 let populationChange: AssetState['populationChange'] = null;66 if (pop.length) {67 const latest = pop[0]!;68 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);69 if (prev[0] && prev[0].total !== latest.total) populationChange = { grader: latest.grader, from: prev[0].total, to: latest.total, date: latest.reportDate };70 }71 return {72 title: row.asset.title,73 slug: row.asset.slug,74 rivUsd: s?.rivUsd ?? null,75 rivConfidence: s?.rivConfidence ?? null,76 rivSampleSize: s?.rivSampleSize ?? 0,77 athUsd: priorAth,78 latestSaleUsd: s?.latestSaleUsd ?? null,79 latestSaleAt: s?.latestSaleAt ?? null,80 sales30d: s?.sales30d ?? 0,81 baselineSales30d: baseline[0]?.n ?? null,82 newListings,83 newAuctionLots: lots,84 endingLots: ending.filter((l): l is typeof l & { endsAt: Date } => l.endsAt !== null),85 belowRivLots: belowRiv.map((l) => ({ id: l.id, title: l.title, endsAt: l.ends_at ? new Date(l.ends_at) : null, auctionHouse: l.house, allInBidUsd: Number(l.bid), rivUsd: Number(l.riv), bidVsRiv: Number(l.d), feeBasis: l.fee_basis })),86 newRecordSale: newRecord,87 populationChange,88 };89}9091async function categoryState(db: Database, slug: string, since: Date, now: Date): Promise<CategoryState | null> {92 const c = await db.select().from(categories).where(eq(categories.slug, slug)).limit(1);93 if (!c[0]) return null;94 const snap = await db.select().from(categorySnapshots).where(eq(categorySnapshots.categorySlug, slug)).orderBy(sql`${categorySnapshots.date} desc`).limit(1);95 const [rec, radar, lots, ending, base] = await Promise.all([96 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 }>>,97 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 }>>,98 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 }>>,99 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 }>>,100 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 }>>,101 ]);102 const r = rec[0];103 return {104 name: c[0].name,105 slug,106 change1d: snap[0]?.change1d ?? null,107 newRecordSale: r ? { assetTitle: r.title, assetSlug: r.slug, priceUsd: Number(r.price_usd), saleDate: new Date(r.sale_date) } : null,108 radarFindings: radar.map((x) => ({ assetTitle: x.title, assetSlug: x.slug, kind: x.kind, score: Number(x.score) })),109 newAuctionLots: lots[0]?.n ?? 0,110 endingLots: ending[0]?.n ?? 0,111 sales30d: snap[0]?.sales ?? 0,112 baselineSales30d: base[0]?.n ?? null,113 };114}115116async function indexState(db: Database, ticker: string): Promise<IndexState | null> {117 const i = await db.select().from(indices).where(eq(indices.ticker, ticker)).limit(1);118 if (!i[0]) return null;119 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);120 const change1d = vals.length === 2 && vals[1]!.value > 0 ? vals[0]!.value / vals[1]!.value - 1 : null;121 return { ticker, name: i[0].name, change1d, value: vals[0]?.value ?? null };122}123124/** Evaluate every active alert. `since` = last run time (defaults to 1 hour ago). */125export async function runAlerts(db: Database, opts: { since?: Date; now?: Date } = {}): Promise<{ evaluated: number; triggered: number }> {126 const now = opts.now ?? new Date();127 const since = opts.since ?? new Date(now.getTime() - 3600_000);128 const rows = await db.select().from(alerts).where(eq(alerts.active, true));129 const prefs = await loadUsers(db, [...new Set(rows.map((r) => r.userId))]);130 const cacheA = new Map<string, AssetState | null>();131 const cacheC = new Map<string, CategoryState | null>();132 const cacheI = new Map<string, IndexState | null>();133 let triggered = 0;134 for (const a of rows) {135 if (!prefs.has(a.userId)) continue;136 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 };137 let trigger: Trigger | null = null;138 try {139 if (a.targetType === 'asset') {140 if (!cacheA.has(a.targetId)) cacheA.set(a.targetId, await assetState(db, a.targetId, since, now));141 const s = cacheA.get(a.targetId);142 if (s) trigger = evaluateAssetAlert(row, s, now);143 } else if (a.targetType === 'category') {144 if (!cacheC.has(a.targetId)) cacheC.set(a.targetId, await categoryState(db, a.targetId, since, now));145 const s = cacheC.get(a.targetId);146 if (s) trigger = evaluateCategoryAlert(row, s, now);147 } else if (a.targetType === 'index') {148 if (!cacheI.has(a.targetId)) cacheI.set(a.targetId, await indexState(db, a.targetId));149 const s = cacheI.get(a.targetId);150 if (s) trigger = evaluateIndexAlert(row, s, now);151 }152 } catch (err) {153 log.error({ err: err instanceof Error ? err.message : String(err), alertId: a.id }, 'alert evaluation failed');154 continue;155 }156 if (trigger) {157 await deliver(db, { userId: a.userId, alertId: a.id, channel: a.channel, trigger, prefs: prefs.get(a.userId), now });158 triggered++;159 }160 }161 log.info({ evaluated: rows.length, triggered }, 'alerts evaluated');162 return { evaluated: rows.length, triggered };163}164165/** Price targets: notify once when reached (in-app + e-mail per prefs). */166export async function runTargets(db: Database, now = new Date()): Promise<number> {167 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));168 const prefs = await loadUsers(db, [...new Set(rows.map((r) => r.t.userId))]);169 let n = 0;170 for (const { t, asset, stats } of rows) {171 const riv = stats?.rivUsd ?? null;172 if (!targetHit(t.direction as 'above' | 'below', t.targetUsd, riv)) continue;173 const usd = (v: number) => new Intl.NumberFormat('en-US', { style: 'currency', currency: 'USD', maximumFractionDigits: 0 }).format(v);174 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)]] };175 await deliver(db, { userId: t.userId, alertId: null, channel: 'both', trigger, prefs: prefs.get(t.userId), kind: 'target_hit', now });176 await db.update(priceTargets).set({ hitAt: now, notifiedAt: now }).where(eq(priceTargets.id, t.id));177 n++;178 }179 return n;180}181