feat(notifications): web push + telegram/discord stream alerts

backend/package.json

@@ -49,6 +49,7 @@
     "prom-client": "^15.1.3",
     "swagger-jsdoc": "^6.3.0",
     "swagger-ui-express": "^5.0.1",
+    "web-push": "^3.6.7",
     "winston": "^3.11.0",
     "zod": "^4.4.3"
   },
@@ -60,6 +61,7 @@
     "@types/supertest": "^6.0.3",
     "@types/swagger-jsdoc": "^6.0.4",
     "@types/swagger-ui-express": "^4.1.6",
+    "@types/web-push": "^3.6.4",
     "@vitest/coverage-v8": "^3.2.7",
     "eventsource": "^2.0.2",
     "nodemon": "^3.1.11",

backend/prisma/schema.prisma

@@ -245,3 +245,51 @@ model IndexerDeadLetterEvent {
   @@index([attempts])
   @@index([eventType])
 }
+
+// ---------------------------------------------------------------------------
+// Notifications (Web Push + Telegram / Discord)
+// ---------------------------------------------------------------------------
+
+model PushSubscription {
+  id        String   @id @default(cuid())
+  address   String
+  endpoint  String   @unique
+  p256dh    String
+  auth      String
+  userAgent String?
+  createdAt DateTime @default(now())
+
+  @@index([address])
+}
+
+model NotificationChannel {
+  id             String   @id @default(cuid())
+  address        String   @unique
+  telegramChatId String?
+  discordWebhook String?
+  events         String[] @default(["STREAM_CREATED", "CLIFF_REACHED", "RUNWAY_WARNING", "TOKENS_CLAIMED"])
+  createdAt      DateTime @default(now())
+  updatedAt      DateTime @updatedAt
+}
+
+model LinkCode {
+  code      String    @id
+  address   String
+  expiresAt DateTime
+  usedAt    DateTime?
+  createdAt DateTime  @default(now())
+
+  @@index([address])
+}
+
+model NotificationLog {
+  id        String   @id @default(cuid())
+  address   String
+  event     String
+  channel   String
+  status    String
+  error     String?
+  createdAt DateTime @default(now())
+
+  @@index([address, createdAt])
+}

backend/src/routes/v1/notifications.routes.ts

@@ -0,0 +1,52 @@
+import { Router } from 'express';
+import { notificationService } from '../../services/notification.service.js';
+
+export const notificationsRouter = Router();
+
+notificationsRouter.post('/subscribe', async (req, res) => {
+  try {
+    const result = await notificationService.subscribeWebPush(req.body);
+    res.status(201).json(result);
+  } catch (err) {
+    res.status(400).json({ error: err instanceof Error ? err.message : 'Bad request' });
+  }
+});
+
+notificationsRouter.post('/unsubscribe', async (req, res) => {
+  try {
+    await notificationService.unsubscribeWebPush(req.body.address, req.body.endpoint);
+    res.json({ ok: true });
+  } catch (err) {
+    res.status(400).json({ error: err instanceof Error ? err.message : 'Bad request' });
+  }
+});
+
+notificationsRouter.post('/link-code', async (req, res) => {
+  try {
+    const result = await notificationService.generateLinkCode(req.body.address);
+    res.json(result);
+  } catch (err) {
+    res.status(400).json({ error: err instanceof Error ? err.message : 'Bad request' });
+  }
+});
+
+notificationsRouter.post('/telegram/link', async (req, res) => {
+  try {
+    const result = await notificationService.linkTelegram(req.body.code, req.body.chatId);
+    res.json(result);
+  } catch (err) {
+    res.status(400).json({ error: err instanceof Error ? err.message : 'Bad request' });
+  }
+});
+
+notificationsRouter.post('/discord', async (req, res) => {
+  try {
+    const result = await notificationService.setDiscordWebhook(
+      req.body.address,
+      req.body.webhookUrl,
+    );
+    res.json(result);
+  } catch (err) {
+    res.status(400).json({ error: err instanceof Error ? err.message : 'Bad request' });
+  }
+});

backend/src/services/notification.service.ts

@@ -0,0 +1,161 @@
+import { prisma } from '../lib/prisma.js';
+// eslint-disable-next-line @typescript-eslint/no-var-requires
+import webpush from 'web-push';
+
+export type StreamEventType =
+  | 'STREAM_CREATED'
+  | 'CLIFF_REACHED'
+  | 'RUNWAY_WARNING'
+  | 'TOKENS_CLAIMED';
+
+export interface StreamEvent {
+  type: StreamEventType;
+  address: string;
+  title: string;
+  body: string;
+  url?: string;
+  data?: Record<string, unknown>;
+}
+
+const TELEGRAM_API = 'https://api.telegram.org';
+
+export class NotificationService {
+  private vapidReady = false;
+
+  constructor() {
+    const publicKey = process.env.VAPID_PUBLIC_KEY;
+    const privateKey = process.env.VAPID_PRIVATE_KEY;
+    const subject = process.env.VAPID_SUBJECT ?? 'mailto:alerts@flowfi.app';
+    if (publicKey && privateKey) {
+      webpush.setVapidDetails(subject, publicKey, privateKey);
+      this.vapidReady = true;
+    }
+  }
+
+  async subscribeWebPush(input: {
+    address: string;
+    endpoint: string;
+    keys: { p256dh: string; auth: string };
+    userAgent?: string;
+  }) {
+    return prisma.pushSubscription.upsert({
+      where: { endpoint: input.endpoint },
+      update: {
+        address: input.address,
+        p256dh: input.keys.p256dh,
+        auth: input.keys.auth,
+        userAgent: input.userAgent ?? null,
+      },
+      create: {
+        address: input.address,
+        endpoint: input.endpoint,
+        p256dh: input.keys.p256dh,
+        auth: input.keys.auth,
+        userAgent: input.userAgent ?? null,
+      },
+    });
+  }
+
+  async unsubscribeWebPush(address: string, endpoint: string) {
+    return prisma.pushSubscription.deleteMany({ where: { address, endpoint } });
+  }
+
+  async generateLinkCode(address: string) {
+    const code = Math.random().toString(36).slice(2, 8).toUpperCase();
+    const expiresAt = new Date(Date.now() + 15 * 60 * 1000);
+    return prisma.linkCode.create({ data: { code, address, expiresAt } });
+  }
+
+  async linkTelegram(code: string, chatId: string) {
+    const record = await prisma.linkCode.findUnique({ where: { code } });
+    if (!record || record.usedAt || record.expiresAt < new Date()) {
+      throw new Error('Link code is invalid or expired');
+    }
+    await prisma.linkCode.update({ where: { code }, data: { usedAt: new Date() } });
+    return prisma.notificationChannel.upsert({
+      where: { address: record.address },
+      update: { telegramChatId: chatId },
+      create: { address: record.address, telegramChatId: chatId },
+    });
+  }
+
+  async setDiscordWebhook(address: string, webhookUrl: string) {
+    if (!/^https:\/\/(discord|discordapp)\.com\/api\/webhooks\//.test(webhookUrl)) {
+      throw new Error('Invalid Discord webhook URL');
+    }
+    return prisma.notificationChannel.upsert({
+      where: { address },
+      update: { discordWebhook: webhookUrl },
+      create: { address, discordWebhook: webhookUrl },
+    });
+  }
+
+  async dispatch(event: StreamEvent): Promise<void> {
+    const [subs, channel] = await Promise.all([
+      prisma.pushSubscription.findMany({ where: { address: event.address } }),
+      prisma.notificationChannel.findUnique({ where: { address: event.address } }),
+    ]);
+
+    const tasks: Array<Promise<unknown>> = [];
+    for (const sub of subs) tasks.push(this.sendWebPush(sub, event));
+    if (channel?.telegramChatId) tasks.push(this.sendTelegram(channel.telegramChatId, event));
+    if (channel?.discordWebhook) tasks.push(this.sendDiscord(channel.discordWebhook, event));
+
+    await Promise.allSettled(tasks);
+  }
+
+  private async sendWebPush(
+    sub: { endpoint: string; p256dh: string; auth: string },
+    event: StreamEvent,
+  ) {
+    if (!this.vapidReady) return;
+    try {
+      await webpush.sendNotification(
+        { endpoint: sub.endpoint, keys: { p256dh: sub.p256dh, auth: sub.auth } },
+        JSON.stringify({
+          title: event.title,
+          body: event.body,
+          url: event.url ?? '/',
+          event: event.type,
+          tag: `flowfi-${event.type.toLowerCase()}`,
+        }),
+      );
+    } catch (error) {
+      const statusCode = (error as { statusCode?: number })?.statusCode;
+      if (statusCode === 404 || statusCode === 410) {
+        await prisma.pushSubscription.deleteMany({ where: { endpoint: sub.endpoint } });
+      }
+    }
+  }
+
+  private async sendTelegram(chatId: string, event: StreamEvent) {
+    const token = process.env.TELEGRAM_BOT_TOKEN;
+    if (!token) return;
+    await fetch(`${TELEGRAM_API}/bot${token}/sendMessage`, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({
+        chat_id: chatId,
+        text: `*${event.title}*\n${event.body}`,
+        parse_mode: 'Markdown',
+      }),
+    }).catch(() => undefined);
+  }
+
+  private async sendDiscord(webhookUrl: string, event: StreamEvent) {
+    await fetch(webhookUrl, {
+      method: 'POST',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify({
+        embeds: [{
+          title: event.title,
+          description: event.body,
+          color: event.type === 'RUNWAY_WARNING' ? 0xf59e0b : 0x6366f1,
+          timestamp: new Date().toISOString(),
+        }],
+      }),
+    }).catch(() => undefined);
+  }
+}
+
+export const notificationService = new NotificationService();

backend/src/workers/notification-bot.worker.ts

@@ -0,0 +1,87 @@
+import { prisma } from '../lib/prisma.js';
+import { notificationService } from '../services/notification.service.js';
+
+const POLL_INTERVAL_MS = 60_000;
+const RUNWAY_WARNING_HOURS = 24;
+
+export class NotificationBotWorker {
+  private timer: ReturnType<typeof setInterval> | null = null;
+  private readonly warned = new Set<string>();
+
+  start() {
+    if (this.timer) return;
+    this.timer = setInterval(() => {
+      void this.pollRunwayWarnings().catch(() => undefined);
+    }, POLL_INTERVAL_MS);
+  }
+
+  stop() {
+    if (this.timer) clearInterval(this.timer);
+    this.timer = null;
+  }
+
+  async notify(type: string, address: string, data: Record<string, any>): Promise<void> {
+    const map: Record<string, { title: string; body: string; url: string }> = {
+      STREAM_CREATED: {
+        title: '🌊 New stream received',
+        body: `You are receiving a new stream of ${data.amount ?? ''} ${data.symbol ?? ''} from ${shorten(data.sender)}`,
+        url: `/streams/${data.streamId ?? ''}`,
+      },
+      CLIFF_REACHED: {
+        title: '🎯 Vesting cliff reached',
+        body: `${data.amount ?? ''} ${data.symbol ?? ''} tokens are now claimable.`,
+        url: `/streams/${data.streamId ?? ''}`,
+      },
+      RUNWAY_WARNING: {
+        title: '⚠️ Low runway warning',
+        body: `Stream #${data.streamId ?? ''} has less than ${RUNWAY_WARNING_HOURS} hours of runway remaining. Refuel now.`,
+        url: `/streams/${data.streamId ?? ''}`,
+      },
+      TOKENS_CLAIMED: {
+        title: '✅ Withdrawal confirmed',
+        body: `Successfully withdrawn ${data.amount ?? ''} ${data.symbol ?? ''} to your wallet.`,
+        url: `/streams/${data.streamId ?? ''}`,
+      },
+    };
+    const build = map[type];
+    if (!build) return;
+    await notificationService.dispatch({ type: type as any, address, ...build, data });
+  }
+
+  private async pollRunwayWarnings(): Promise<void> {
+    const now = Date.now();
+    const horizon = now + RUNWAY_WARNING_HOURS * 60 * 60 * 1000;
+
+    const streams = await prisma.stream.findMany({
+      where: {
+        isActive: true,
+        isPaused: false,
+        endTime: {
+          gt: BigInt(Math.floor(now / 1000)),
+          lt: BigInt(Math.floor(horizon / 1000)),
+        },
+      },
+      take: 200,
+    });
+
+    for (const stream of streams) {
+      const key = `${stream.streamId}:${stream.sender}:RUNWAY_WARNING`;
+      if (this.warned.has(key)) continue;
+      this.warned.add(key);
+      await notificationService.dispatch({
+        type: 'RUNWAY_WARNING',
+        address: stream.sender,
+        title: '⚠️ Stream runway below 24 hours',
+        body: `Stream #${stream.streamId} is running low. Refuel to avoid a payroll halt.`,
+        url: `/streams/${stream.streamId}`,
+      });
+    }
+  }
+}
+
+function shorten(address?: string): string {
+  if (!address) return 'unknown';
+  return address.length > 12 ? `${address.slice(0, 6)}…${address.slice(-4)}` : address;
+}
+
+export const notificationBotWorker = new NotificationBotWorker();