Implement and test real-time Socket.io notifications.

backend/RealTimeNotifications.tsx

@@ -1,5 +1,5 @@
 import React, { useEffect, useRef } from 'react';
-import { useSocket } from '../hooks/useSocket';
+import { useSocket } from './useSocket';
 import { toast } from 'react-hot-toast'; // Assuming react-hot-toast is used
 
 interface Props {

backend/SocketServer.ts

@@ -14,6 +14,10 @@
 import { Server as HttpServer } from 'http';
 import { Server, Socket } from 'socket.io';
 import jwt from 'jsonwebtoken';
+import { PrismaClient } from '@prisma/client';
+import { AuthenticationService } from './services/AuthenticationService';
+
+const prisma = new PrismaClient();
 
 // ─── Types ────────────────────────────────────────────────────────────────────
 
@@ -47,6 +51,15 @@ export const ROOMS = {
   notifications: 'room_notifications',
 } as const;
 
+/** JWT payload shapes supported by socket authentication. */
+interface SocketJwtPayload {
+  userId?: string;
+  id?: string;
+  email?: string;
+  role?: string;
+  sessionId?: string;
+}
+
 /** Canonical event names emitted by the server. */
 export const SERVER_EVENTS = {
   NOTIFICATION: 'notification',
@@ -126,6 +139,16 @@ export class SocketServer {
     return SocketServer.instance;
   }
 
+  /**
+   * Reset the singleton instance. Intended for test isolation only.
+   */
+  public static resetForTesting(): void {
+    if (SocketServer.instance) {
+      SocketServer.instance.shutdown();
+      delete (SocketServer as { instance?: SocketServer }).instance;
+    }
+  }
+
   /**
    * Return the raw Socket.IO `Server` instance.
    * Throws if `getInstance(httpServer)` has not been called yet.
@@ -142,8 +165,7 @@ export class SocketServer {
   // ─── Authentication middleware ─────────────────────────────────────────────
 
   private setupAuthMiddleware(): void {
-    this.io.use((socket: AuthSocket, next) => {
-      // Accept token from handshake auth object or Authorization header
+    this.io.use(async (socket: AuthSocket, next) => {
       const token =
         socket.handshake.auth?.token ||
         socket.handshake.headers?.authorization?.replace(/^Bearer\s+/i, '');
@@ -159,12 +181,11 @@ export class SocketServer {
       }
 
       try {
-        const decoded = jwt.verify(token, secret) as AuthenticatedUser & Record<string, unknown>;
-        socket.user = {
-          id: decoded.id,
-          email: decoded.email,
-          role: decoded.role,
-        };
+        const user = await this.authenticateToken(token, secret);
+        if (!user) {
+          return next(new Error('Authentication error: invalid token'));
+        }
+        socket.user = user;
         next();
       } catch (err) {
         const message =
@@ -176,6 +197,57 @@ export class SocketServer {
     });
   }
 
+  /**
+   * Resolve the authenticated user from a JWT.
+   * Supports production session tokens, legacy id/email/role payloads, and test tokens.
+   */
+  private async authenticateToken(
+    token: string,
+    secret: string
+  ): Promise<AuthenticatedUser | null> {
+    const decoded = jwt.verify(token, secret) as SocketJwtPayload & jwt.JwtPayload;
+    const userId = decoded.userId ?? decoded.id;
+
+    if (!userId) {
+      return null;
+    }
+
+    // Production path: validate active session when sessionId is present
+    if (decoded.sessionId) {
+      const authService = new AuthenticationService();
+      const result = await authService.verifyToken(token);
+      if (result.user) {
+        return {
+          id: result.user.id,
+          email: result.user.email,
+          role: result.user.role,
+        };
+      }
+      return null;
+    }
+
+    // Test / legacy tokens that include email and role directly in the JWT
+    if (decoded.email && decoded.role) {
+      return {
+        id: userId,
+        email: decoded.email,
+        role: decoded.role,
+      };
+    }
+
+    // Fallback: load user profile from the database
+    const user = await prisma.user.findUnique({ where: { id: userId } });
+    if (!user) {
+      return null;
+    }
+
+    return {
+      id: user.id,
+      email: user.email,
+      role: user.role,
+    };
+  }
+
   // ─── Connection handlers ───────────────────────────────────────────────────
 
   private setupConnectionHandlers(): void {
@@ -223,6 +295,9 @@ export class SocketServer {
       // Auto-join the user's private room
       this.joinRoom(socket, info, ROOMS.user(socket.user.id));
 
+      // Auto-join the notifications room for system-wide alerts
+      this.joinRoom(socket, info, ROOMS.notifications);
+
       // Auto-join the admin room for privileged users
       if (socket.user.role === 'ADMIN' || socket.user.role === 'SUPER_ADMIN') {
         this.joinRoom(socket, info, ROOMS.admins);

backend/prisma/migrations/20240824_add_notifications/migration.sql

@@ -0,0 +1,51 @@
+-- CreateTable
+CREATE TABLE "notifications" (
+    "id" TEXT NOT NULL,
+    "userId" TEXT NOT NULL,
+    "type" TEXT NOT NULL,
+    "title" TEXT NOT NULL,
+    "message" TEXT NOT NULL,
+    "data" JSONB NOT NULL DEFAULT '{}',
+    "priority" TEXT NOT NULL DEFAULT 'MEDIUM',
+    "category" TEXT,
+    "actionUrl" TEXT,
+    "actionText" TEXT,
+    "isRead" BOOLEAN NOT NULL DEFAULT false,
+    "readAt" TIMESTAMP(3),
+    "expiresAt" TIMESTAMP(3),
+    "sound" TEXT,
+    "icon" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "notifications_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "notification_preferences" (
+    "userId" TEXT NOT NULL,
+    "email" BOOLEAN NOT NULL DEFAULT true,
+    "sms" BOOLEAN NOT NULL DEFAULT false,
+    "push" BOOLEAN NOT NULL DEFAULT true,
+    "inApp" BOOLEAN NOT NULL DEFAULT true,
+    "soundEnabled" BOOLEAN NOT NULL DEFAULT true,
+    "desktopNotifications" BOOLEAN NOT NULL DEFAULT true,
+    "quietHours" JSONB,
+    "categories" JSONB,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "notification_preferences_pkey" PRIMARY KEY ("userId")
+);
+
+-- CreateIndex
+CREATE INDEX "notifications_userId_idx" ON "notifications"("userId");
+
+-- CreateIndex
+CREATE INDEX "notifications_userId_isRead_idx" ON "notifications"("userId", "isRead");
+
+-- AddForeignKey
+ALTER TABLE "notifications" ADD CONSTRAINT "notifications_userId_fkey" FOREIGN KEY ("userId") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "notification_preferences" ADD CONSTRAINT "notification_preferences_userId_fkey" FOREIGN KEY ("userId") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE;

backend/prisma/migrations/migration_lock.toml

@@ -0,0 +1,3 @@
+# Please do not edit this file manually
+# It should be added in your version-control system (i.e. Git)
+provider = "postgresql"

backend/schema.prisma

@@ -26,6 +26,8 @@ model User {
   bills       Bill[]
   payments    Payment[]
   sessions    UserSession[]
+  notifications Notification[]
+  notificationPreference NotificationPreference?
 
   @@map("users")
 }
@@ -138,4 +140,49 @@ enum PaymentMethod {
   STELLAR
   CREDIT_CARD
   BANK_TRANSFER
+}
+
+// In-app notifications persisted for offline delivery
+model Notification {
+  id          String    @id @default(cuid())
+  userId      String
+  type        String
+  title       String
+  message     String
+  data        Json      @default("{}")
+  priority    String    @default("MEDIUM")
+  category    String?
+  actionUrl   String?
+  actionText  String?
+  isRead      Boolean   @default(false)
+  readAt      DateTime?
+  expiresAt   DateTime?
+  sound       String?
+  icon        String?
+  createdAt   DateTime  @default(now())
+  updatedAt   DateTime  @updatedAt
+
+  user User @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  @@index([userId])
+  @@index([userId, isRead])
+  @@map("notifications")
+}
+
+model NotificationPreference {
+  userId               String   @id
+  email                Boolean  @default(true)
+  sms                  Boolean  @default(false)
+  push                 Boolean  @default(true)
+  inApp                Boolean  @default(true)
+  soundEnabled         Boolean  @default(true)
+  desktopNotifications Boolean  @default(true)
+  quietHours           Json?
+  categories           Json?
+  createdAt            DateTime @default(now())
+  updatedAt            DateTime @updatedAt
+
+  user User @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  @@map("notification_preferences")
 }

backend/services/NotificationService.ts

@@ -100,12 +100,10 @@ export class RealTimeNotificationService {
       const userId: string | undefined = (socket as any).user?.id;
       if (!userId) return;
 
-      // Deliver any queued notifications immediately on connect
-      const queued = this.notificationQueue.get(userId) ?? [];
-      if (queued.length > 0) {
-        socket.emit('notifications', queued);
-        this.notificationQueue.delete(userId);
-      }
+      // Deliver persisted and queued notifications when the user reconnects
+      this.deliverPendingNotifications(userId, socket).catch((err) => {
+        console.error('[NotificationService] deliverPendingNotifications error:', err);
+      });
 
       // Push current unread count
       this.getUnreadCount(userId).then((count) => {
@@ -206,6 +204,44 @@ export class RealTimeNotificationService {
     return Promise.all(notifications.map((n) => this.sendNotification(n)));
   }
 
+  /**
+   * Deliver unread notifications from the database plus any in-memory queue
+   * entries that were created while the user was offline in this process.
+   */
+  async deliverPendingNotifications(userId: string, socket: { emit: (event: string, data: unknown) => void }): Promise<void> {
+    const unreadFromDb = await this.getUserNotifications(userId, {
+      unreadOnly: true,
+      limit: 100,
+    });
+
+    const queued = this.notificationQueue.get(userId) ?? [];
+    const seenIds = new Set(unreadFromDb.map((notification) => notification.id));
+    const merged = [
+      ...unreadFromDb,
+      ...queued.filter((notification) => !notification.id || !seenIds.has(notification.id)),
+    ];
+
+    if (merged.length > 0) {
+      socket.emit('notifications', merged);
+    }
+
+    this.notificationQueue.delete(userId);
+  }
+
+  /**
+   * Expose the in-memory offline queue for testing.
+   */
+  getQueuedNotifications(userId: string): NotificationData[] {
+    return [...(this.notificationQueue.get(userId) ?? [])];
+  }
+
+  /**
+   * Reset the in-memory offline queue. Intended for test isolation only.
+   */
+  resetQueueForTesting(): void {
+    this.notificationQueue.clear();
+  }
+
   // ─── Database helpers ─────────────────────────────────────────────────────
 
   async getUserNotifications(