feat(realtime): typed event contract, rate limiting, heartbeat + notifications listing

backend/docs/REALTIME.md

@@ -0,0 +1,94 @@
+# Realtime Module
+
+Socket.IO gateway that pushes tip and notification events to connected
+clients. See `src/realtime/`.
+
+## Components
+
+| File | Purpose |
+|------|---------|
+| `types.ts` | Shared typed contract: event names + payloads, `SocketData` |
+| `auth.ts` | Handshake middleware — verifies the same JWT the REST API issues |
+| `rateLimit.ts` | Per-IP connection throttling and per-socket event throttling |
+| `gateway.ts` | Server init, room management, `emitTipCreated` / `emitNotificationCreated` |
+
+## Connecting
+
+Clients authenticate on the handshake, not via a separate event:
+
+```ts
+import { io } from 'socket.io-client';
+
+const socket = io(API_URL, {
+  auth: { token: accessToken }, // the same access token used for REST calls
+  transports: ['websocket'],
+});
+
+socket.on('connected', ({ userId }) => { /* handshake accepted */ });
+socket.on('error', ({ code, message }) => { /* FORBIDDEN, RATE_LIMITED, ... */ });
+```
+
+A connection with a missing or invalid token is rejected before `connection`
+fires — the client only sees `connect_error`.
+
+## Event contract
+
+The full typed contract lives in `types.ts` (`ServerToClientEvents`,
+`ClientToServerEvents`). Summary:
+
+- **Client → server:** `subscribe:creator`, `subscribe:notifications`,
+  `unsubscribe:creator`, `unsubscribe:notifications`
+- **Server → client:** `connected`, `error`, `tip.created`,
+  `notification.created`
+
+Subscribing to another user's `notifications` room is rejected with an
+`error` event (`code: 'FORBIDDEN'`) — a socket may only subscribe to its own
+`user:<userId>` room.
+
+## Heartbeat
+
+The server pings each client on a fixed interval and disconnects it if no
+pong is received within the timeout (configured in `gateway.ts`):
+
+- `pingInterval`: 25s — how often the server probes the connection
+- `pingTimeout`: 20s — how long it waits for a response before dropping it
+
+These are handled by Socket.IO's engine (no application code needed) and
+detect dead connections (e.g. a laptop going to sleep or a dropped Wi-Fi
+network) faster than waiting on a TCP timeout.
+
+## Reconnection
+
+`socket.io-client` reconnects automatically by default (exponential backoff,
+capped, with jitter). Important behavior to build clients against:
+
+- **Rooms are not restored automatically.** On `reconnect`, re-issue
+  `subscribe:creator` / `subscribe:notifications` for anything the client
+  still cares about — the server has no memory of a socket's prior rooms
+  once it disconnects.
+- **The auth token is re-sent on every reconnect attempt** (it's read from
+  the `auth` option passed to `io(...)`, not cached per-connection). If the
+  token expires while offline, refresh it before the client comes back
+  online so reconnection doesn't loop into `connect_error`.
+- Recommended client wiring:
+
+```ts
+socket.on('reconnect', () => {
+  socket.emit('subscribe:notifications', currentUserId);
+  socket.emit('subscribe:creator', watchedCreatorAddress);
+});
+```
+
+## Rate limiting
+
+Two independent limiters, both in-memory per server process (see
+`rateLimit.ts`):
+
+- **Connections:** 20 new connections per IP per 60s window, enforced as a
+  handshake middleware before auth runs.
+- **Events:** 30 client→server events per socket per 10s window, enforced at
+  the top of every event handler.
+
+Exceeding the connection limit rejects the handshake (`connect_error`).
+Exceeding the event limit emits `error` (`code: 'RATE_LIMITED'`) and drops
+that event; the socket stays connected.

backend/prisma/schema.prisma

@@ -27,32 +27,21 @@ model User {
   scopes         String[] @default([])
   createdAt      DateTime @default(now())
   updatedAt      DateTime @updatedAt
-  apiKeys        ApiKey[]
-  refreshTokens  RefreshToken[]
-  goals          Goal[]
-  sentTips       Tip[]                @relation("SentTips")
-  receivedTips   Tip[]                @relation("ReceivedTips")
-  leaderboardSnapshots LeaderboardSnapshot[]
-  streak         Streak?
-  notifications  Notification[]
-
-  tipperSubscriptions  Subscription[] @relation("SubscriptionTipper")
-  creatorSubscriptions Subscription[] @relation("SubscriptionCreator")
-  /// Soft-delete marker: non-null means the record is logically deleted.
-  deletedAt      DateTime?
   apiKeys              ApiKey[]
   refreshTokens        RefreshToken[]
   goals                Goal[]
   sentTips             Tip[]                @relation("SentTips")
   receivedTips         Tip[]                @relation("ReceivedTips")
+  leaderboardSnapshots LeaderboardSnapshot[]
+  streak               Streak?
+  notifications        Notification[]
   tipperSubscriptions  Subscription[]       @relation("SubscriptionTipper")
   creatorSubscriptions Subscription[]       @relation("SubscriptionCreator")
   creditScore          CreditScore?
   creditScoreHistory   CreditScoreHistory[]
   withdrawals          Withdrawal[]
-  notifications        Notification[]
-  leaderboardSnapshots LeaderboardSnapshot[]
-  streak               Streak?
+  /// Soft-delete marker: non-null means the record is logically deleted.
+  deletedAt            DateTime?
 
   @@index([createdAt])
 }
@@ -200,13 +189,6 @@ model XAccount {
   updatedAt  DateTime @updatedAt
 }
 
-/// Status of a tip transaction.
-enum TipStatus {
-  CONFIRMED
-  PENDING
-  REFUNDED
-}
-
 /// Leaderboard period tracked by snapshots.
 enum Period {
   WEEKLY

backend/src/app.ts

@@ -7,14 +7,6 @@ import { env } from './config/env.js';
 import {
   errorHandler,
   notFoundHandler,
-} from "./common/middleware/errorHandler.js";
-import { logger } from "./common/utils/logger.js";
-import { openApiDocument } from "./docs/openapi.js";
-import { authRouter } from "./modules/auth/auth.routes.js";
-import { profilesRouter } from "./modules/profiles/profiles.routes.js";
-import { creditRouter } from "./modules/credit/credit.routes.js";
-import { leaderboardRouter } from "./modules/leaderboard/leaderboard.routes.js";
-import { xRouter } from "./modules/x/x.routes.js";
 } from './common/middleware/errorHandler.js';
 import { logger } from './common/utils/logger.js';
 import { openApiDocument } from './docs/openapi.js';

backend/src/common/errors/AppError.ts

@@ -40,12 +40,6 @@ export class ConflictError extends AppError {
     super(409, message, "CONFLICT");
   }
 }
-export class ServiceUnavailableError extends AppError {
-  constructor(message = "Service unavailable") {
-    super(503, message, "SERVICE_UNAVAILABLE");
-  }
-}
-
 export class BadGatewayError extends AppError {
   constructor(message = 'Bad gateway', details?: unknown) {
     super(502, message, 'BAD_GATEWAY', details);

backend/src/realtime/auth.ts

@@ -1,26 +1,27 @@
-import type { Socket } from 'socket.io';
-import jwt from 'jsonwebtoken';
-import { config } from '../config/index.js';
+import type { Socket, ExtendedError } from 'socket.io';
+import { verifyAccessToken } from '../modules/auth/auth.service.js';
 import { logger } from '../common/utils/logger.js';
-import type { AuthUser } from '../modules/auth/auth.types.js';
+import type {
+  ClientToServerEvents,
+  ServerToClientEvents,
+  InterServerEvents,
+  SocketData,
+} from './types.js';
 
-interface JwtPayload {
-  sub: string;
-  stellarAddress: string;
-}
-
-export interface AuthenticatedSocket extends Socket {
-  authUser?: AuthUser;
-}
-
-declare module 'socket.io' {
-  interface Socket {
-    authUser?: AuthUser;
-  }
-}
+type AuthSocket = Socket<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>;
 
-export function socketAuth(socket: AuthenticatedSocket, next: (err?: Error) => void): void {
-  const token = socket.handshake.auth?.token as string | undefined;
+/**
+ * Socket.IO handshake middleware: authenticates a connecting socket using the
+ * same JWT access token issued by the REST auth module.
+ *
+ * The client must send the token as `socket.handshake.auth.token`, e.g.:
+ *   io(url, { auth: { token: accessToken } })
+ *
+ * On success, the decoded payload is attached to `socket.data.auth`.
+ * On failure, the connection is rejected before `connection` fires.
+ */
+export function socketAuth(socket: AuthSocket, next: (err?: ExtendedError) => void): void {
+  const token = socket.handshake.auth?.['token'] as string | undefined;
 
   if (!token) {
     logger.warn({ socketId: socket.id }, 'Socket connection rejected: no token');
@@ -29,16 +30,14 @@ export function socketAuth(socket: AuthenticatedSocket, next: (err?: Error) => v
   }
 
   try {
-    const payload = jwt.verify(token, config.auth.jwtSecret) as JwtPayload;
-    socket.authUser = {
-      id: payload.sub,
-      stellarAddress: payload.stellarAddress,
-      username: null,
-    };
-    logger.debug({ socketId: socket.id, userId: payload.sub }, 'Socket authenticated');
+    socket.data.auth = verifyAccessToken(token);
+    logger.debug(
+      { socketId: socket.id, userId: socket.data.auth.userId },
+      'Socket authenticated',
+    );
     next();
-  } catch (err) {
-    logger.warn({ socketId: socket.id, err }, 'Socket connection rejected: invalid token');
+  } catch {
+    logger.warn({ socketId: socket.id }, 'Socket connection rejected: invalid token');
     next(new Error('Invalid or expired token'));
   }
 }

backend/src/realtime/gateway.ts

@@ -2,46 +2,66 @@ import type { Server as HttpServer } from 'node:http';
 import { Server as SocketIOServer } from 'socket.io';
 import { env } from '../config/env.js';
 import { logger } from '../common/utils/logger.js';
+import { registerClosable } from '../common/utils/lifecycle.js';
 import { socketAuth } from './auth.js';
+import { connectionRateLimit, eventLimiter, guardEventRate } from './rateLimit.js';
 import type {
   ServerToClientEvents,
   ClientToServerEvents,
+  InterServerEvents,
+  SocketData,
   NotificationPayload,
 } from './types.js';
 import type { TipResponseDto } from '../modules/tips/tips.dto.js';
 
-let io: SocketIOServer<ClientToServerEvents, ServerToClientEvents> | null = null;
+export type RealtimeServer = SocketIOServer<
+  ClientToServerEvents,
+  ServerToClientEvents,
+  InterServerEvents,
+  SocketData
+>;
+
+let io: RealtimeServer | null = null;
+
+/**
+ * Heartbeat tuning (see docs/REALTIME.md): how often the server pings each
+ * client, and how long it waits for a pong before considering it disconnected.
+ */
+const HEARTBEAT_PING_INTERVAL_MS = 25_000;
+const HEARTBEAT_PING_TIMEOUT_MS = 20_000;
 
-export function initRealtime(httpServer: HttpServer): SocketIOServer<ClientToServerEvents, ServerToClientEvents> {
-  io = new SocketIOServer<ClientToServerEvents, ServerToClientEvents>(httpServer, {
-    cors: {
-      origin: env.CORS_ORIGIN.split(','),
-      methods: ['GET', 'POST'],
+export function initRealtime(httpServer: HttpServer): RealtimeServer {
+  io = new SocketIOServer<ClientToServerEvents, ServerToClientEvents, InterServerEvents, SocketData>(
+    httpServer,
+    {
+      cors: {
+        origin: env.CORS_ORIGIN.split(','),
+        methods: ['GET', 'POST'],
+      },
+      pingInterval: HEARTBEAT_PING_INTERVAL_MS,
+      pingTimeout: HEARTBEAT_PING_TIMEOUT_MS,
     },
-  });
+  );
 
+  io.use(connectionRateLimit);
   io.use(socketAuth);
 
   io.on('connection', (socket) => {
-    logger.info({ socketId: socket.id, userId: socket.authUser?.id }, 'Client connected');
+    const { userId } = socket.data.auth;
+    logger.info({ socketId: socket.id, userId }, 'Client connected');
+    socket.emit('connected', { userId });
 
     socket.on('subscribe:creator', (creatorAddress: string) => {
-      if (!socket.authUser) {
-        (socket as any).emit('error', { message: 'Not authenticated' });
-        return;
-      }
+      if (!guardEventRate(socket)) return;
       const room = `creator:${creatorAddress}`;
       void socket.join(room);
       logger.debug({ socketId: socket.id, room }, 'Subscribed to creator room');
     });
 
     socket.on('subscribe:notifications', (userId: string) => {
-      if (!socket.authUser) {
-        (socket as any).emit('error', { message: 'Not authenticated' });
-        return;
-      }
-      if (socket.authUser.id !== userId) {
-        (socket as any).emit('error', { message: 'Forbidden' });
+      if (!guardEventRate(socket)) return;
+      if (socket.data.auth.userId !== userId) {
+        socket.emit('error', { code: 'FORBIDDEN', message: 'Cannot subscribe to another user' });
         return;
       }
       const room = `user:${userId}`;
@@ -50,22 +70,35 @@ export function initRealtime(httpServer: HttpServer): SocketIOServer<ClientToSer
     });
 
     socket.on('unsubscribe:creator', (creatorAddress: string) => {
+      if (!guardEventRate(socket)) return;
       const room = `creator:${creatorAddress}`;
       void socket.leave(room);
       logger.debug({ socketId: socket.id, room }, 'Unsubscribed from creator room');
     });
 
     socket.on('unsubscribe:notifications', (userId: string) => {
+      if (!guardEventRate(socket)) return;
       const room = `user:${userId}`;
       void socket.leave(room);
       logger.debug({ socketId: socket.id, room }, 'Unsubscribed from notifications room');
     });
 
-    socket.on('disconnect', () => {
-      logger.info({ socketId: socket.id }, 'Client disconnected');
+    socket.on('disconnect', (reason) => {
+      logger.info({ socketId: socket.id, reason }, 'Client disconnected');
     });
   });
 
+  const sweepInterval = setInterval(() => eventLimiter.sweep(), 60_000);
+  sweepInterval.unref();
+
+  registerClosable({
+    name: 'Socket.IO',
+    close: async () => {
+      clearInterval(sweepInterval);
+      await new Promise<void>((resolve) => io?.close(() => resolve()));
+    },
+  });
+
   logger.info('Realtime gateway initialized');
   return io;
 }
@@ -79,9 +112,12 @@ export function emitTipCreated(tip: TipResponseDto): void {
 export function emitNotificationCreated(notification: NotificationPayload): void {
   if (!io) return;
   io.to(`user:${notification.userId}`).emit('notification.created', notification);
-  logger.debug({ notificationId: notification.id, room: `user:${notification.userId}` }, 'Emitted notification.created');
+  logger.debug(
+    { notificationId: notification.id, room: `user:${notification.userId}` },
+    'Emitted notification.created',
+  );
 }
 
-export function getIO(): SocketIOServer<ClientToServerEvents, ServerToClientEvents> | null {
+export function getIO(): RealtimeServer | null {
   return io;
 }

backend/src/realtime/index.ts

@@ -1,2 +1,8 @@
-export { initRealtime, emitTipCreated, emitNotificationCreated, getIO } from "./gateway.js";
-export type { ServerToClientEvents, ClientToServerEvents, NotificationPayload } from "./types.js";
+export { initRealtime, emitTipCreated, emitNotificationCreated, getIO } from './gateway.js';
+export type {
+  ServerToClientEvents,
+  ClientToServerEvents,
+  InterServerEvents,
+  SocketData,
+  NotificationPayload,
+} from './types.js';