How to Build a Notification System with Replit

1// Type this into Replit Agent:
2// Build a multi-channel notification system with Express and PostgreSQL using Drizzle ORM.
3// Tables:
4// - notifications: id serial pk, user_id text not null, channel text not null
5//   (enum: in_app/email/sms), title text not null, body text not null,
6//   type text not null (enum: info/warning/error/success), reference_type text,
7//   reference_id text, status text default 'pending'
8//   (enum: pending/sent/delivered/failed/read), read_at timestamp,
9//   sent_at timestamp, created_at timestamp default now()
10// - notification_preferences: id serial pk, user_id text not null, channel text not null,
11//   event_type text not null, enabled boolean default true,
12//   unique(user_id, channel, event_type)
13// - notification_templates: id serial pk, event_type text unique not null,
14//   title_template text not null, body_template text not null, channels text[] not null
15// - notification_queue: id serial pk, notification_id integer FK notifications,
16//   channel text not null, payload jsonb not null, attempts integer default 0,
17//   max_attempts integer default 3, next_retry_at timestamp default now(),
18//   error_message text, created_at timestamp default now()
19// Routes: GET /api/notifications (user's in-app with pagination), PATCH /api/notifications/:id/read,
20// PATCH /api/notifications/read-all, GET /api/notifications/unread-count,
21// GET /api/notifications/preferences, PUT /api/notifications/preferences,
22// POST /api/notifications/dispatch (internal fan-out engine),
23// GET /api/notifications/stream (SSE for real-time in-app delivery).
24// React frontend: bell icon with unread badge, dropdown panel, preferences settings page.
25// Use Replit Auth. Bind server to 0.0.0.0.

1const express = require('express');
2const { db } = require('../db');
3const { notifications, notificationPreferences, notificationTemplates, notificationQueue } = require('../../shared/schema');
4const { eq, and } = require('drizzle-orm');
5
6const router = express.Router();
7
8// Simple template variable interpolation: {{variable}} -> value
9function interpolate(template, variables) {
10  return template.replace(/{{(\w+)}}/g, (_, key) => variables[key] || '');
11}
12
13// POST /api/notifications/dispatch — fan-out engine
14// Body: { eventType, userId, variables: { orderId: '123', amount: '$99' } }
15router.post('/dispatch', async (req, res) => {
16  const { eventType, userId, variables = {} } = req.body;
17  if (!eventType || !userId) {
18    return res.status(400).json({ error: 'eventType and userId required' });
19  }
20
21  // Load template for this event
22  const [template] = await db.select().from(notificationTemplates)
23    .where(eq(notificationTemplates.eventType, eventType));
24  if (!template) {
25    return res.status(404).json({ error: `No template for event type: ${eventType}` });
26  }
27
28  const title = interpolate(template.titleTemplate, variables);
29  const body = interpolate(template.bodyTemplate, variables);
30
31  const createdNotifications = [];
32
33  // Fan out to each channel this template supports
34  for (const channel of template.channels) {
35    // Check user preference for this channel + event type
36    const [pref] = await db.select().from(notificationPreferences)
37      .where(and(
38        eq(notificationPreferences.userId, userId),
39        eq(notificationPreferences.channel, channel),
40        eq(notificationPreferences.eventType, eventType)
41      ));
42
43    // Default to enabled if no preference record exists
44    if (pref && !pref.enabled) continue;
45
46    const [notification] = await db.insert(notifications).values({
47      userId,
48      channel,
49      title,
50      body,
51      type: variables.notificationType || 'info',
52      referenceType: variables.referenceType || null,
53      referenceId: variables.referenceId || null,
54    }).returning();
55
56    // Queue email and SMS for async delivery (in_app is real-time via SSE)
57    if (channel !== 'in_app') {
58      await db.insert(notificationQueue).values({
59        notificationId: notification.id,
60        channel,
61        payload: { userId, title, body, ...variables },
62      });
63    }
64
65    createdNotifications.push(notification);
66  }
67
68  res.status(201).json({ dispatched: createdNotifications.length, notifications: createdNotifications });
69});
70
71module.exports = router;

1const axios = require('axios');
2const sgMail = require('@sendgrid/mail');
3const twilio = require('twilio');
4const { db } = require('../db');
5const { notificationQueue, notifications } = require('../../shared/schema');
6const { eq, sql } = require('drizzle-orm');
7
8sgMail.setApiKey(process.env.SENDGRID_API_KEY);
9const twilioClient = twilio(process.env.TWILIO_ACCOUNT_SID, process.env.TWILIO_AUTH_TOKEN);
10
11async function processQueue() {
12  // Atomic claim: mark as 'processing' to prevent duplicates
13  const claimed = await db.execute(sql`
14    UPDATE notification_queue
15    SET error_message = 'processing'
16    WHERE id IN (
17      SELECT id FROM notification_queue
18      WHERE error_message IS DISTINCT FROM 'processing'
19        AND attempts < max_attempts
20        AND next_retry_at <= NOW()
21      LIMIT 10
22      FOR UPDATE SKIP LOCKED
23    )
24    RETURNING *
25  `);
26
27  for (const job of claimed.rows) {
28    const { id, channel, payload, attempts } = job;
29    let success = false;
30    let errorMsg = null;
31
32    try {
33      if (channel === 'email' && payload.userEmail) {
34        await sgMail.send({
35          to: payload.userEmail,
36          from: process.env.FROM_EMAIL || 'noreply@yourdomain.com',
37          subject: payload.title,
38          text: payload.body,
39        });
40        success = true;
41      } else if (channel === 'sms' && payload.userPhone) {
42        await twilioClient.messages.create({
43          body: `${payload.title}: ${payload.body}`,
44          from: process.env.TWILIO_PHONE_NUMBER,
45          to: payload.userPhone,
46        });
47        success = true;
48      }
49    } catch (err) {
50      errorMsg = err.message;
51    }
52
53    if (success) {
54      // Mark delivered
55      await db.execute(sql`DELETE FROM notification_queue WHERE id = ${id}`);
56      await db.execute(sql`UPDATE notifications SET status = 'sent', sent_at = NOW() WHERE id = ${job.notification_id}`);
57    } else {
58      // Exponential backoff: 1min, 2min, 4min
59      const nextRetrySeconds = Math.pow(2, attempts) * 60;
60      await db.execute(sql`
61        UPDATE notification_queue
62        SET attempts = attempts + 1,
63            next_retry_at = NOW() + INTERVAL '${nextRetrySeconds} seconds',
64            error_message = ${errorMsg || 'Unknown error'}
65        WHERE id = ${id}
66      `);
67
68      if (attempts + 1 >= job.max_attempts) {
69        await db.execute(sql`UPDATE notifications SET status = 'failed' WHERE id = ${job.notification_id}`);
70      }
71    }
72  }
73}
74
75// Run processor every 30 seconds on Reserved VM
76setInterval(processQueue, 30000);
77processQueue(); // Run once on startup
78
79module.exports = { processQueue };

1// In-memory SSE subscriber map: userId -> res
2const inAppSubscribers = new Map();
3
4// GET /api/notifications/stream — SSE for in-app delivery
5router.get('/stream', (req, res) => {
6  const userId = req.user?.id;
7  if (!userId) return res.status(401).end();
8
9  res.writeHead(200, {
10    'Content-Type': 'text/event-stream',
11    'Cache-Control': 'no-cache',
12    'Connection': 'keep-alive',
13    'X-Accel-Buffering': 'no',
14  });
15  res.write(`data: {"type":"connected"}\n\n`);
16
17  const pingInterval = setInterval(() => {
18    try { res.write(': ping\n\n'); } catch (e) { clearInterval(pingInterval); }
19  }, 30000);
20
21  inAppSubscribers.set(userId, res);
22
23  req.on('close', () => {
24    clearInterval(pingInterval);
25    inAppSubscribers.delete(userId);
26  });
27});
28
29// Helper: push in-app notification to connected user
30exports.pushInApp = (userId, notification) => {
31  const res = inAppSubscribers.get(userId);
32  if (res) {
33    try {
34      res.write(`data: ${JSON.stringify(notification)}\n\n`);
35    } catch (e) {
36      inAppSubscribers.delete(userId);
37    }
38  }
39};
40
41// Call pushInApp from the dispatch route after creating in_app notification:
42// const { pushInApp } = require('./stream');
43// if (channel === 'in_app') pushInApp(userId, notification);

1// GET /api/notifications/unread-count
2router.get('/unread-count', async (req, res) => {
3  const userId = req.user?.id;
4  if (!userId) return res.status(401).json({ error: 'Login required' });
5
6  const result = await db.execute(sql`
7    SELECT COUNT(*) as count FROM notifications
8    WHERE user_id = ${userId}
9      AND channel = 'in_app'
10      AND status != 'read'
11  `);
12  res.json({ count: parseInt(result.rows[0]?.count || 0) });
13});
14
15// PATCH /api/notifications/:id/read
16router.patch('/:id/read', async (req, res) => {
17  const userId = req.user?.id;
18  await db.update(notifications)
19    .set({ status: 'read', readAt: new Date() })
20    .where(and(
21      eq(notifications.id, parseInt(req.params.id)),
22      eq(notifications.userId, userId)
23    ));
24  res.json({ ok: true });
25});
26
27// PATCH /api/notifications/read-all
28router.patch('/read-all', async (req, res) => {
29  const userId = req.user?.id;
30  await db.execute(sql`
31    UPDATE notifications
32    SET status = 'read', read_at = NOW()
33    WHERE user_id = ${userId} AND channel = 'in_app' AND status != 'read'
34  `);
35  res.json({ ok: true });
36});
37
38// server/index.js — require the queue processor to start on Reserved VM
39const { processQueue } = require('./queue/processor');
40// processQueue() is called inside processor.js via setInterval + immediate call
41console.log('Notification queue processor started');