diff --git a/prisma/schema/alert.prisma b/prisma/schema/alert.prisma index 05a5d6a2..198bbc00 100644 --- a/prisma/schema/alert.prisma +++ b/prisma/schema/alert.prisma @@ -10,6 +10,7 @@ model PriceAlert { callbackUrl String? // optional since #1012 isActive Boolean @default(true) triggeredAt DateTime? + triggeredPrice Decimal? @db.Decimal(20, 7) createdAt DateTime @default(now()) @@index([creatorId]) diff --git a/prisma/schema/migrations/20261017000000_add_price_alert_triggered_price/migration.sql b/prisma/schema/migrations/20261017000000_add_price_alert_triggered_price/migration.sql new file mode 100644 index 00000000..1de652d3 --- /dev/null +++ b/prisma/schema/migrations/20261017000000_add_price_alert_triggered_price/migration.sql @@ -0,0 +1,2 @@ +ALTER TABLE "PriceAlert" +ADD COLUMN "triggeredPrice" DECIMAL(20, 7); diff --git a/src/constants/notifications.constants.ts b/src/constants/notifications.constants.ts index 67b0cd95..1b84f65a 100644 --- a/src/constants/notifications.constants.ts +++ b/src/constants/notifications.constants.ts @@ -4,6 +4,7 @@ export const NOTIFICATION_TYPES = { TRADE_COMPLETED: 'trade_completed', LOCKUP_EXPIRING: 'lockup_expiring', PRICE_MOVED: 'price_moved', + PRICE_ALERT_TRIGGERED: 'price_alert_triggered', PAUSE_PROPOSAL_CREATED: 'pause_proposal_created', TRADING_PAUSED: 'trading_paused', KEY_DEPRECATED: 'key_deprecated', diff --git a/src/modules/alerts/alert.controllers.ts b/src/modules/alerts/alert.controllers.ts index 3e561e93..09e0612c 100644 --- a/src/modules/alerts/alert.controllers.ts +++ b/src/modules/alerts/alert.controllers.ts @@ -75,7 +75,8 @@ export async function httpListAlerts( ): Promise { try { const authWallet = (req as AuthenticatedRequest).user?.wallet; - const queryWallet = req.query.wallet_address || req.query.walletAddress; + const queryWallet = + req.query.wallet || req.query.wallet_address || req.query.walletAddress; const targetWallet = authWallet || queryWallet; if (!targetWallet || typeof targetWallet !== 'string') { diff --git a/src/modules/alerts/alert.schemas.ts b/src/modules/alerts/alert.schemas.ts index fa140419..256f3b8c 100644 --- a/src/modules/alerts/alert.schemas.ts +++ b/src/modules/alerts/alert.schemas.ts @@ -4,6 +4,7 @@ import { isValidStellarAddress } from '../wallet/wallet.utils'; export const CreateAlertSchema = z .object({ + key_id: z.string().min(1, 'key_id is required').optional(), keyId: z.string().min(1, 'keyId is required').optional(), creator_id: z.string().min(1, 'creator_id is required').optional(), creatorId: z.string().min(1, 'creatorId is required').optional(), @@ -34,7 +35,7 @@ export const CreateAlertSchema = z callbackUrl: z.string().url('callbackUrl must be a valid URL').optional(), }) .refine( - (data) => !!(data.keyId || data.creator_id || data.creatorId), + (data) => !!(data.key_id || data.keyId || data.creator_id || data.creatorId), { message: 'keyId is required', path: ['keyId'], @@ -48,7 +49,7 @@ export const CreateAlertSchema = z } ) .transform((data) => ({ - creator_id: (data.keyId || data.creator_id || data.creatorId)!, + creator_id: (data.key_id || data.keyId || data.creator_id || data.creatorId)!, wallet_address: data.wallet_address || data.walletAddress, target_price: (data.targetPrice ?? data.target_price)!, direction: data.direction, @@ -67,6 +68,12 @@ export type CreateAlertInput = { }; export const ListAlertsQuerySchema = z.object({ + wallet: z + .string() + .refine(isValidStellarAddress, { + message: 'Invalid Stellar wallet address', + }) + .optional(), wallet_address: z .string() .refine(isValidStellarAddress, { diff --git a/src/modules/alerts/alert.service.ts b/src/modules/alerts/alert.service.ts index ebe4ff0e..7a957c76 100644 --- a/src/modules/alerts/alert.service.ts +++ b/src/modules/alerts/alert.service.ts @@ -3,6 +3,7 @@ import { envConfig } from '../../config'; import { logger } from '../../utils/logger.utils'; import { CreateAlertInput } from './alert.schemas'; import { truncateWallet } from '../../utils/wallet-display.utils'; +import { SendMail } from '../../utils/mail.utils'; export type PriceMovement = { creatorId: string; @@ -42,7 +43,7 @@ export async function createAlert(input: CreateAlertInput) { walletAddress: input.wallet_address, targetPrice: input.target_price, direction: input.direction, - callbackUrl: input.callback_url ?? null, + callbackUrl: input.callback_url, }, }); @@ -71,21 +72,6 @@ export async function listAlerts(walletAddress: string) { }); } -/** - * Issue #1012: returns every triggered (not-necessarily-active) alert for a - * wallet so the notifications service can surface them in-app. Ordered by - * triggeredAt desc so newest fires first. - */ -export async function listTriggeredAlertsForWallet(walletAddress: string) { - return await prisma.priceAlert.findMany({ - where: { - walletAddress, - triggeredAt: { not: null }, - }, - orderBy: { triggeredAt: 'desc' }, - }); -} - /** * Deletes a price alert by id, scoped to the wallet address for authorization. * Returns the deleted record id or null if not found. @@ -179,6 +165,8 @@ function toNumber(value: number | string | { toString(): string }): number { return typeof value === 'number' ? value : Number(value.toString()); } + + function maskCallbackUrl(callbackUrl: string): string { try { const url = new URL(callbackUrl); @@ -203,17 +191,13 @@ async function deliverPriceAlertWebhook( walletAddress: string; targetPrice: unknown; direction: string; - callbackUrl: string | null; + callbackUrl: string; }, payload: Record ): Promise { - if (!alert.callbackUrl) { - return; - } - const maxAttempts = envConfig.WEBHOOK_RETRY_MAX_ATTEMPTS; const maskedUrl = maskCallbackUrl(alert.callbackUrl); - const threshold = toNumber(alert.targetPrice as string | number | { toString(): string}); + const threshold = toNumber(alert.targetPrice as string | number | { toString(): string }); const triggeredPrice = toNumber(payload.current_price as string | number | { toString(): string }); for (let attempt = 1; attempt <= maxAttempts; attempt++) { @@ -280,12 +264,6 @@ async function deliverPriceAlertWebhook( /** * Evaluates active alerts for a creator price movement and delivers only alerts * whose threshold was crossed in the registered direction. - * - * Issue #1012: alert disable is unconditional — even if webhook delivery - * throws, the alert is disabled and (via listTriggeredAlertsForWallet) becomes - * visible in the in-app notification feed. Webhook failure is logged, not - * propagated, so callers wiring this into the price pipeline can't roll back - * a trade. */ export async function evaluatePriceAlertsForMovement( movement: PriceMovement @@ -302,8 +280,6 @@ export async function evaluatePriceAlertsForMovement( }, }); - let alertsTriggered = 0; - for (const alert of alerts) { const targetPrice = toNumber(alert.targetPrice); const crossedAbove = @@ -319,53 +295,60 @@ export async function evaluatePriceAlertsForMovement( continue; } - // Disable + record trigger first — this is what the in-app - // notification reads. If this succeeds, the user sees the alert even - // if the webhook below fails. - await prisma.priceAlert.update({ - where: { id: alert.id }, + // Claim the alert atomically before delivery so overlapping indexer + // workers cannot notify the same wallet more than once. + const claimed = await prisma.priceAlert.updateMany({ + where: { id: alert.id, isActive: true, triggeredAt: null }, data: { isActive: false, triggeredAt: new Date(), + triggeredPrice: String(movement.currentPrice), }, }); - alertsTriggered += 1; + if (claimed.count === 0) continue; + + const payload = { + event_type: 'price_alert', + alert_id: alert.id, + creator_id: alert.creatorId, + wallet_address: alert.walletAddress, + target_price: targetPrice, + current_price: currentPrice, + direction: alert.direction, + }; try { - await deliverPriceAlertWebhook(alert, { - event_type: 'price_alert', - alert_id: alert.id, - creator_id: alert.creatorId, - wallet_address: alert.walletAddress, - target_price: targetPrice, - current_price: currentPrice, - direction: alert.direction, + await deliverPriceAlertWebhook(alert, payload); + } catch (error) { + // The trigger is already durable and visible in the in-app inbox; + // a webhook outage must not reactivate it or block other alerts. + logger.warn( + { alert_id: alert.id, error: getDeliveryErrorCode(error) }, + 'Price alert webhook delivery failed after trigger' + ); + } + + try { + const wallet = await prisma.stellarWallet.findUnique({ + where: { address: alert.walletAddress }, + select: { user: { select: { email: true, emailVerified: true } } }, }); - } catch (webhookError) { - logger.error( - { - alert_id: alert.id, - creator_id: alert.creatorId, - error_message: - webhookError instanceof Error - ? webhookError.message - : 'Unknown error', - }, - 'Price alert webhook failed after alert was disabled' + if (wallet?.user.emailVerified) { + const safeKey = escapeHtml(alert.creatorId); + await SendMail({ + to: wallet.user.email, + subject: `Price alert triggered for ${alert.creatorId}`, + text: `The ${alert.direction} price alert for key ${alert.creatorId} triggered at ${currentPrice}. Your target was ${targetPrice}.`, + html: `

Your ${escapeHtml(alert.direction)} price alert for key ${safeKey} triggered.

Current price: ${currentPrice}
Alert target: ${targetPrice}

`, + }); + } + } catch (error) { + logger.warn( + { alert_id: alert.id, error: (error as Error).message }, + 'Price alert email delivery failed' ); } } - - logger[alertsTriggered > 0 ? 'info' : 'debug']( - { - creator_id: movement.creatorId, - alerts_evaluated: alerts.length, - alerts_triggered: alertsTriggered, - new_price: currentPrice, - checked_at: new Date().toISOString(), - }, - 'Price alert check run' - ); } catch (err) { logger.error( { @@ -379,4 +362,17 @@ export async function evaluatePriceAlertsForMovement( ); throw err; } -} \ No newline at end of file +} + +function escapeHtml(value: string): string { + return value.replace(/[&<>"']/g, character => { + const entities: Record = { + '&': '&', + '<': '<', + '>': '>', + '"': '"', + "'": ''', + }; + return entities[character] ?? character; + }); +} diff --git a/src/modules/indexer/price-snapshot.service.ts b/src/modules/indexer/price-snapshot.service.ts index e1486b7d..5caee723 100644 --- a/src/modules/indexer/price-snapshot.service.ts +++ b/src/modules/indexer/price-snapshot.service.ts @@ -135,6 +135,21 @@ export async function upsertPriceSnapshot( }, }), ]); + try { + const { evaluatePriceAlertsForMovement } = await import('../alerts/alert.service'); + await evaluatePriceAlertsForMovement({ + creatorId, + previousPrice: existing.currentPrice.toString(), + currentPrice: price.toString(), + ledger_sequence: ledger, + }); + } catch (error) { + // Price history is the source of truth; alert delivery is best-effort. + logger.warn( + { creator_id: creatorId, error: (error as Error).message }, + 'price-snapshot: alert evaluation failed' + ); + } logger.debug( { creator_id: creatorId, diff --git a/src/modules/notifications/notification.service.ts b/src/modules/notifications/notification.service.ts index a9ce44d4..04d95955 100644 --- a/src/modules/notifications/notification.service.ts +++ b/src/modules/notifications/notification.service.ts @@ -1,8 +1,8 @@ - // src/modules/notifications/notification.service.ts import { prisma } from '../../utils/prisma.utils'; import { getRedis } from '../../utils/redis.utils'; import { + CIRCUIT_BREAKER_TRIP_NOTIFICATION_LIMIT, LOCKUP_WARNING_WINDOW_MS, NOTIFICATION_TYPES, REDIS_KEYS, @@ -16,8 +16,12 @@ import { NotificationItem } from './notification.types'; async function getLastReadAt(walletAddress: string): Promise { const redis = getRedis(); if (!redis) return null; - const raw = await redis.get(REDIS_KEYS.notificationsReadAt(walletAddress)); - if (!raw) return null; + const raw = await redis.get( + REDIS_KEYS.notificationsReadAt(walletAddress) + ); + if (!raw) { + return null; + } const parsed = new Date(raw); return Number.isNaN(parsed.getTime()) ? null : parsed; } @@ -26,108 +30,356 @@ function isRead(createdAt: Date, lastReadAt: Date | null): boolean { return lastReadAt !== null && createdAt.getTime() <= lastReadAt.getTime(); } -async function buildTradeCompleted(walletAddress: string, lastReadAt: Date | null): Promise { - const trades = await prisma.trade.findMany({ where: { buyer: walletAddress }, orderBy: { timestamp: 'desc' }, take: 50 }); - return trades.map((trade: { id: string; timestamp: Date; creatorId: string; quantity: unknown; price: unknown; txHash: string | null }) => ({ - id: "trade_completed:${trade.id}", - type: NOTIFICATION_TYPES.TRADE_COMPLETED, - createdAt: trade.timestamp.toISOString(), - read: isRead(trade.timestamp, lastReadAt), - payload: { tradeId: trade.id, keyId: trade.creatorId, quantity: trade.quantity, price: trade.price, txHash: trade.txHash }, - })); +async function buildTradeCompleted( + walletAddress: string, + lastReadAt: Date | null +): Promise { + const trades = await prisma.trade.findMany({ + where: { buyer: walletAddress }, + orderBy: { timestamp: 'desc' }, + take: 50, + }); + + return trades.map( + (trade: { + id: string; + timestamp: Date; + creatorId: string; + quantity: unknown; + price: unknown; + txHash: string | null; + }) => { + const createdAt = trade.timestamp; + return { + id: `trade_completed:${trade.id}`, + type: NOTIFICATION_TYPES.TRADE_COMPLETED, + createdAt: createdAt.toISOString(), + read: isRead(createdAt, lastReadAt), + payload: { + tradeId: trade.id, + keyId: trade.creatorId, + quantity: trade.quantity, + price: trade.price, + txHash: trade.txHash, + }, + }; + } + ); } -async function buildLockupExpiring(walletAddress: string, lastReadAt: Date | null, now: Date): Promise { +async function buildLockupExpiring( + walletAddress: string, + lastReadAt: Date | null, + now: Date +): Promise { const windowEnd = new Date(now.getTime() + LOCKUP_WARNING_WINDOW_MS); - const holdings = await prisma.keyOwnership.findMany({ where: { ownerAddress: walletAddress, balance: { gt: 0 }, lockupExpiresAt: { gt: now, lte: windowEnd } } }); - return holdings.map((holding: { id: string; creatorId: string; lockupExpiresAt: Date | null; balance: { toString(): string } }) => ({ - id: "lockup_expiring:${holding.id}", - type: NOTIFICATION_TYPES.LOCKUP_EXPIRING, - createdAt: holding.lockupExpiresAt!.toISOString(), - read: isRead(holding.lockupExpiresAt!, lastReadAt), - payload: { keyId: holding.creatorId, lockupExpiresAt: holding.lockupExpiresAt!.toISOString(), balance: holding.balance.toString() }, - })); + const holdings = await prisma.keyOwnership.findMany({ + where: { + ownerAddress: walletAddress, + balance: { gt: 0 }, + lockupExpiresAt: { + gt: now, + lte: windowEnd, + }, + }, + }); + + return holdings.map( + (holding: { + id: string; + creatorId: string; + lockupExpiresAt: Date | null; + balance: { toString(): string }; + }) => { + const createdAt = holding.lockupExpiresAt!; + return { + id: `lockup_expiring:${holding.id}`, + type: NOTIFICATION_TYPES.LOCKUP_EXPIRING, + createdAt: createdAt.toISOString(), + read: isRead(createdAt, lastReadAt), + payload: { + keyId: holding.creatorId, + lockupExpiresAt: holding.lockupExpiresAt!.toISOString(), + balance: holding.balance.toString(), + }, + }; + } + ); } -async function buildKeyDeprecated(walletAddress: string, lastReadAt: Date | null): Promise { - const holdings = await prisma.keyOwnership.findMany({ where: { ownerAddress: walletAddress, balance: { gt: 0 } }, select: { creatorId: true } }); - if (holdings.length === 0) return []; +async function buildKeyDeprecated( + walletAddress: string, + lastReadAt: Date | null +): Promise { + const holdings = await prisma.keyOwnership.findMany({ + where: { ownerAddress: walletAddress, balance: { gt: 0 } }, + select: { creatorId: true }, + }); + if (holdings.length === 0) { + return []; + } + const keyIds = holdings.map((h: { creatorId: string }) => h.creatorId); - const deprecatedKeys = await prisma.creatorProfile.findMany({ where: { id: { in: keyIds }, deprecatedAt: { not: null } }, select: { id: true, deprecatedAt: true, buybackPriceXlm: true, buybackExpiresAt: true } }); - return deprecatedKeys.map((key) => ({ - id: "key_deprecated:${key.id}", - type: NOTIFICATION_TYPES.KEY_DEPRECATED, - createdAt: key.deprecatedAt!.toISOString(), - read: isRead(key.deprecatedAt!, lastReadAt), - payload: { keyId: key.id, buybackPriceXlm: key.buybackPriceXlm != null ? String(key.buybackPriceXlm) : null, buybackExpiresAt: key.buybackExpiresAt ? key.buybackExpiresAt.toISOString() : null }, - })); + const deprecatedKeys = await prisma.creatorProfile.findMany({ + where: { id: { in: keyIds }, deprecatedAt: { not: null } }, + select: { + id: true, + deprecatedAt: true, + buybackPriceXlm: true, + buybackExpiresAt: true, + reason: true, + successorKeyId: true, + }, + }); + + return deprecatedKeys.map((key) => { + // deprecatedAt is guaranteed non-null by the where: { not: null } filter above + const createdAt = key.deprecatedAt!; + return { + id: `key_deprecated:${key.id}`, + type: NOTIFICATION_TYPES.KEY_DEPRECATED, + createdAt: createdAt.toISOString(), + read: isRead(createdAt, lastReadAt), + payload: { + keyId: key.id, + reason: key.reason ?? null, + successorKeyId: key.successorKeyId ?? null, + buybackPriceXlm: + key.buybackPriceXlm !== null && + key.buybackPriceXlm !== undefined + ? String(key.buybackPriceXlm) + : null, + buybackExpiresAt: key.buybackExpiresAt + ? key.buybackExpiresAt.toISOString() + : null, + }, + }; + } + ); } -export async function buildKeySunsetFlagged(walletAddress: string, lastReadAt: Date | null): Promise { - const holdings = await prisma.keyOwnership.findMany({ where: { ownerAddress: walletAddress, balance: { gt: 0 } }, select: { creatorId: true } }); - if (holdings.length === 0) return []; +export async function buildKeySunsetFlagged( + walletAddress: string, + lastReadAt: Date | null +): Promise { + const holdings = await prisma.keyOwnership.findMany({ + where: { ownerAddress: walletAddress, balance: { gt: 0 } }, + select: { creatorId: true }, + }); + if (holdings.length === 0) { + return []; + } + const keyIds = holdings.map((h: { creatorId: string }) => h.creatorId); - const sunsetKeys = await prisma.creatorProfile.findMany({ where: { id: { in: keyIds }, deprecatedAt: { not: null } }, select: { id: true, deprecatedAt: true, buybackPriceXlm: true, buybackExpiresAt: true } }); - return sunsetKeys.map((key) => ({ - id: "key_sunset_flagged:${key.id}", - type: NOTIFICATION_TYPES.KEY_SUNSET_FLAGGED, - createdAt: key.deprecatedAt!.toISOString(), - read: isRead(key.deprecatedAt!, lastReadAt), - payload: { keyId: key.id, sunsetDeadline: key.buybackExpiresAt ? key.buybackExpiresAt.toISOString() : null, buybackPriceXlm: key.buybackPriceXlm != null ? String(key.buybackPriceXlm) : null }, - })); + const sunsetKeys = await prisma.creatorProfile.findMany({ + where: { id: { in: keyIds }, deprecatedAt: { not: null } }, + select: { + id: true, + deprecatedAt: true, + buybackPriceXlm: true, + buybackExpiresAt: true, + }, + }); + + return sunsetKeys.map((key) => { + const createdAt = key.deprecatedAt!; + return { + id: `key_sunset_flagged:${key.id}`, + type: NOTIFICATION_TYPES.KEY_SUNSET_FLAGGED, + createdAt: createdAt.toISOString(), + read: isRead(createdAt, lastReadAt), + payload: { + keyId: key.id, + sunsetDeadline: key.buybackExpiresAt + ? key.buybackExpiresAt.toISOString() + : null, + buybackPriceXlm: + key.buybackPriceXlm !== null && key.buybackPriceXlm !== undefined + ? String(key.buybackPriceXlm) + : null, + }, + }; + }); +} + +/** + * Notifications for circuit breaker trips on keys the wallet created (#987). + * + * One item per indexed CircuitBreakerTrip row, newest first. The indexer only + * writes a row once per unique (txHash, eventIndex), so each trip event yields + * exactly one notification even across indexer replays. + */ +async function buildCircuitBreakerTripped( + walletAddress: string, + lastReadAt: Date | null +): Promise { + const trips = await prisma.circuitBreakerTrip.findMany({ + where: { creatorWallet: walletAddress }, + orderBy: [{ occurredAt: 'desc' }, { id: 'desc' }], + take: CIRCUIT_BREAKER_TRIP_NOTIFICATION_LIMIT, + }); + + return trips.map( + (trip: { + id: string; + keyId: string; + actualBps: number; + maxBps: number | null; + txHash: string; + occurredAt: Date; + }) => { + const createdAt = trip.occurredAt; + return { + id: `circuit_breaker_tripped:${trip.id}`, + type: NOTIFICATION_TYPES.CIRCUIT_BREAKER_TRIPPED, + createdAt: createdAt.toISOString(), + read: isRead(createdAt, lastReadAt), + payload: { + keyId: trip.keyId, + actualBps: trip.actualBps, + maxBps: trip.maxBps, + txHash: trip.txHash, + }, + }; + } + ); } -async function buildPriceMoved(walletAddress: string, lastReadAt: Date | null, now: Date): Promise { +async function buildPriceMoved( + walletAddress: string, + lastReadAt: Date | null, + now: Date +): Promise { const flaggedKeyIds = await getPriceMovedKeyIds(); - if (flaggedKeyIds.length === 0) return []; - const holdings = await prisma.keyOwnership.findMany({ where: { ownerAddress: walletAddress, balance: { gt: 0 }, creatorId: { in: flaggedKeyIds } }, select: { creatorId: true } }); - if (holdings.length === 0) return []; - const heldFlaggedIds = holdings.map((h: { creatorId: string }) => h.creatorId); - const snapshots = await prisma.creatorPriceSnapshot.findMany({ where: { creatorId: { in: heldFlaggedIds } }, select: { creatorId: true, currentPrice: true, price24hAgo: true, updatedAt: true } }); + if (flaggedKeyIds.length === 0) { + return []; + } + + const holdings = await prisma.keyOwnership.findMany({ + where: { + ownerAddress: walletAddress, + balance: { gt: 0 }, + creatorId: { in: flaggedKeyIds }, + }, + select: { creatorId: true }, + }); + + if (holdings.length === 0) { + return []; + } + + const heldFlaggedIds = holdings.map( + (h: { creatorId: string }) => h.creatorId + ); + const snapshots = await prisma.creatorPriceSnapshot.findMany({ + where: { creatorId: { in: heldFlaggedIds } }, + select: { + creatorId: true, + currentPrice: true, + price24hAgo: true, + updatedAt: true, + }, + }); + const notifications: NotificationItem[] = []; for (const snapshot of snapshots) { + const createdAt = snapshot.updatedAt ?? now; notifications.push({ - id: "price_moved:${snapshot.creatorId}", + id: `price_moved:${snapshot.creatorId}`, type: NOTIFICATION_TYPES.PRICE_MOVED, - createdAt: (snapshot.updatedAt ?? now).toISOString(), - read: isRead(snapshot.updatedAt ?? now, lastReadAt), - payload: { keyId: snapshot.creatorId, currentPrice: snapshot.currentPrice.toString(), price24hAgo: snapshot.price24hAgo.toString() }, + createdAt: createdAt.toISOString(), + read: isRead(createdAt, lastReadAt), + payload: { + keyId: snapshot.creatorId, + currentPrice: snapshot.currentPrice.toString(), + price24hAgo: snapshot.price24hAgo.toString(), + }, }); await markPriceMovedDelivered(snapshot.creatorId, walletAddress); } + return notifications; } -async function buildPriceAlertTriggered(walletAddress: string, lastReadAt: Date | null): Promise { - const triggeredAlerts = await prisma.priceAlert.findMany({ where: { walletAddress, triggeredAt: { not: null } }, orderBy: { triggeredAt: 'desc' }, take: 50 }); - if (triggeredAlerts.length === 0) return []; - const creatorIds = Array.from(new Set(triggeredAlerts.map((a: { creatorId: string }) => a.creatorId))); - const snapshots = await prisma.creatorPriceSnapshot.findMany({ where: { creatorId: { in: creatorIds } }, select: { creatorId: true, currentPrice: true } }); - const priceByCreator = new Map(snapshots.map((s: { creatorId: string; currentPrice: unknown }) => [s.creatorId, String(s.currentPrice)])); - return triggeredAlerts.map((alert: { id: string; creatorId: string; targetPrice: unknown; direction: string; triggeredAt: Date | null }) => ({ - id: "price_alert_triggered:${alert.id}", - type: NOTIFICATION_TYPES.PRICE_ALERT_TRIGGERED, - createdAt: alert.triggeredAt!.toISOString(), - read: isRead(alert.triggeredAt!, lastReadAt), - payload: { alertId: alert.id, keyId: alert.creatorId, targetPrice: String(alert.targetPrice), triggeredPrice: priceByCreator.get(alert.creatorId) ?? null, direction: alert.direction }, - })); -} +async function buildPriceAlertTriggered( + walletAddress: string, + lastReadAt: Date | null +): Promise { + const alerts = await prisma.priceAlert.findMany({ + where: { walletAddress, triggeredAt: { not: null } }, + orderBy: { triggeredAt: 'desc' }, + take: 50, + select: { + id: true, + creatorId: true, + direction: true, + targetPrice: true, + triggeredPrice: true, + triggeredAt: true, + }, + }); + return alerts.flatMap(alert => { + if (!alert.triggeredAt) return []; + return [{ + id: `price_alert_triggered:${alert.id}`, + type: NOTIFICATION_TYPES.PRICE_ALERT_TRIGGERED, + createdAt: alert.triggeredAt.toISOString(), + read: isRead(alert.triggeredAt, lastReadAt), + payload: { + alertId: alert.id, + keyId: alert.creatorId, + direction: alert.direction, + targetPrice: alert.targetPrice.toString(), + triggeredPrice: alert.triggeredPrice?.toString() ?? null, + }, + }]; + }); +} -export async function listNotifications(walletAddress: string, now: Date = new Date()): Promise { +export async function listNotifications( + walletAddress: string, + now: Date = new Date() +): Promise { const lastReadAt = await getLastReadAt(walletAddress); - const [trades, lockups, priceMoved, keyDeprecated, priceAlertTriggered] = await Promise.all([ + + const [ + trades, + lockups, + priceMoved, + priceAlertTriggered, + keyDeprecated, + circuitBreakerTrips, + ] = await Promise.all([ buildTradeCompleted(walletAddress, lastReadAt), buildLockupExpiring(walletAddress, lastReadAt, now), buildPriceMoved(walletAddress, lastReadAt, now), - buildKeyDeprecated(walletAddress, lastReadAt), buildPriceAlertTriggered(walletAddress, lastReadAt), + buildKeyDeprecated(walletAddress, lastReadAt), + buildCircuitBreakerTripped(walletAddress, lastReadAt), ]); - return [...trades, ...lockups, ...priceMoved, ...keyDeprecated, ...priceAlertTriggered].sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); + + return [ + ...trades, + ...lockups, + ...priceMoved, + ...priceAlertTriggered, + ...keyDeprecated, + ...circuitBreakerTrips, + ].sort( + (a, b) => + new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime() + ); } -export async function markAllNotificationsRead(walletAddress: string, now: Date = new Date()): Promise { +export async function markAllNotificationsRead( + walletAddress: string, + now: Date = new Date() +): Promise { const redis = getRedis(); if (!redis) return; - await redis.set(REDIS_KEYS.notificationsReadAt(walletAddress), now.toISOString()); + await redis.set( + REDIS_KEYS.notificationsReadAt(walletAddress), + now.toISOString() + ); }