From cfbec49081a2089bdd43b2a4fb88e1a26ebbd723 Mon Sep 17 00:00:00 2001 From: Joseph-tech10 <243628124+Joseph-tech10@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:40:43 +0100 Subject: [PATCH] feat(alerts): notify users when price thresholds trigger --- prisma/schema/alert.prisma | 1 + .../migration.sql | 2 + src/constants/notifications.constants.ts | 1 + src/modules/alerts/alert.controllers.ts | 3 +- src/modules/alerts/alert.schemas.ts | 11 ++- src/modules/alerts/alert.service.ts | 68 ++++++++++++++++--- src/modules/indexer/price-snapshot.service.ts | 15 ++++ .../notifications/notification.service.ts | 39 +++++++++++ 8 files changed, 128 insertions(+), 12 deletions(-) create mode 100644 prisma/schema/migrations/20261017000000_add_price_alert_triggered_price/migration.sql diff --git a/prisma/schema/alert.prisma b/prisma/schema/alert.prisma index e6ba7c2e..eae6a653 100644 --- a/prisma/schema/alert.prisma +++ b/prisma/schema/alert.prisma @@ -9,6 +9,7 @@ model PriceAlert { callbackUrl String 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 6ca8f962..039ef0ce 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 3c3d3bdf..381cb286 100644 --- a/src/modules/alerts/alert.schemas.ts +++ b/src/modules/alerts/alert.schemas.ts @@ -3,6 +3,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(), @@ -33,7 +34,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'], @@ -47,7 +48,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, @@ -66,6 +67,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 f9750714..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; @@ -294,7 +295,19 @@ export async function evaluatePriceAlertsForMovement( continue; } - await deliverPriceAlertWebhook(alert, { + // 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), + }, + }); + if (claimed.count === 0) continue; + + const payload = { event_type: 'price_alert', alert_id: alert.id, creator_id: alert.creatorId, @@ -302,15 +315,39 @@ export async function evaluatePriceAlertsForMovement( target_price: targetPrice, current_price: currentPrice, direction: alert.direction, - }); + }; - await prisma.priceAlert.update({ - where: { id: alert.id }, - data: { - isActive: false, - triggeredAt: new Date(), - }, - }); + try { + 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 } } }, + }); + 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' + ); + } } } catch (err) { logger.error( @@ -326,3 +363,16 @@ export async function evaluatePriceAlertsForMovement( throw err; } } + +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 83a1cdad..b8b2d2c3 100644 --- a/src/modules/indexer/price-snapshot.service.ts +++ b/src/modules/indexer/price-snapshot.service.ts @@ -134,6 +134,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 2f51900b..04d95955 100644 --- a/src/modules/notifications/notification.service.ts +++ b/src/modules/notifications/notification.service.ts @@ -301,6 +301,42 @@ async function buildPriceMoved( return notifications; } +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() @@ -311,12 +347,14 @@ export async function listNotifications( trades, lockups, priceMoved, + priceAlertTriggered, keyDeprecated, circuitBreakerTrips, ] = await Promise.all([ buildTradeCompleted(walletAddress, lastReadAt), buildLockupExpiring(walletAddress, lastReadAt, now), buildPriceMoved(walletAddress, lastReadAt, now), + buildPriceAlertTriggered(walletAddress, lastReadAt), buildKeyDeprecated(walletAddress, lastReadAt), buildCircuitBreakerTripped(walletAddress, lastReadAt), ]); @@ -325,6 +363,7 @@ export async function listNotifications( ...trades, ...lockups, ...priceMoved, + ...priceAlertTriggered, ...keyDeprecated, ...circuitBreakerTrips, ].sort(