feat: add realtime

apps/api/src/app.module.ts

@@ -12,6 +12,7 @@ import { AuditModule } from './audit/audit.module';
 import { JobsModule } from './jobs/jobs.module';
 import { PlannedTabsModule } from './planned-tabs/planned-tabs.module';
 import { RemindersModule } from './reminders/reminders.module';
+import { RealtimeModule } from './realtime/realtime.module';
 import { HealthController } from './health.controller';
 
 @Module({
@@ -19,6 +20,7 @@ import { HealthController } from './health.controller';
     ThrottlerModule.forRoot([{ ttl: 60000, limit: 100 }]),
     ScheduleModule.forRoot(),
     PrismaModule,
+    RealtimeModule,
     AuthModule,
     VenuesModule,
     DevicesModule,

apps/api/src/auth/auth.service.ts

@@ -5,6 +5,7 @@ import * as crypto from 'crypto';
 import { v4 as uuidv4 } from 'uuid';
 import { PrismaService } from '../prisma/prisma.service';
 import { AuditService } from '../audit/audit.service';
+import { RealtimeService } from '../realtime/realtime.service';
 import { TOKEN_EXPIRY_DAYS, TOKEN_BYTE_LENGTH, TOKEN_PREFIX_LENGTH } from '@kooltap/shared';
 
 @Injectable()
@@ -13,6 +14,7 @@ export class AuthService {
     private prisma: PrismaService,
     private jwtService: JwtService,
     private auditService: AuditService,
+    private realtime: RealtimeService,
   ) {}
 
   async login(email: string, password: string) {
@@ -172,6 +174,8 @@ export class AuthService {
 
     await this.auditService.log(venueId, 'device', deviceId, 'device_registered', { deviceName: name });
 
+    this.realtime.notify(venueId, 'device');
+
     return { deviceId, token: rawToken, tokenExpiresAt: expiresAt };
   }
 
@@ -199,6 +203,8 @@ export class AuthService {
       select: { id: true, email: true, name: true, role: true, venueId: true, createdAt: true },
     });
 
+    this.realtime.notify(venueId, 'users');
+
     return user;
   }
 

apps/api/src/commands/commands.service.ts

@@ -7,6 +7,7 @@ import {
 import { v4 as uuidv4 } from 'uuid';
 import { PrismaService } from '../prisma/prisma.service';
 import { AuditService } from '../audit/audit.service';
+import { RealtimeService } from '../realtime/realtime.service';
 import { CreateCommandDto } from './dto/create-command.dto';
 import { COMMAND_EXPIRY_HOURS, DEVICE_OFFLINE_THRESHOLD_MS } from '@kooltap/shared';
 
@@ -15,6 +16,7 @@ export class CommandsService {
   constructor(
     private prisma: PrismaService,
     private auditService: AuditService,
+    private realtime: RealtimeService,
   ) {}
 
   async create(venueId: string, dto: CreateCommandDto, user: any) {
@@ -130,6 +132,8 @@ export class CommandsService {
       };
     });
 
+    this.realtime.notify(venueId, 'commands');
+
     return result;
   }
 

apps/api/src/devices/devices.controller.ts

@@ -9,7 +9,7 @@ import {
   HttpStatus,
   Req,
 } from '@nestjs/common';
-import { Throttle } from '@nestjs/throttler';
+import { SkipThrottle, Throttle } from '@nestjs/throttler';
 import { ApiTags, ApiOperation, ApiBearerAuth, ApiResponse } from '@nestjs/swagger';
 import { DevicesService } from './devices.service';
 import { AuthService } from '../auth/auth.service';
@@ -37,7 +37,10 @@ export class DevicesController {
     return this.authService.registerDevice(dto.venueId, dto.name, dto.registrationCode);
   }
 
+  // Authenticated by the device token; the till must never be rate-limited out by other
+  // traffic from the venue's shared IP
   @Post(':deviceId/sync')
+  @SkipThrottle()
   @UseGuards(DeviceAuthGuard)
   @ApiOperation({ summary: 'Sync device state with server' })
   @ApiResponse({ status: 200, description: 'Sync successful, returns pending commands' })
@@ -46,6 +49,7 @@ export class DevicesController {
   }
 
   @Post(':deviceId/commands/:commandId/ack')
+  @SkipThrottle()
   @UseGuards(DeviceAuthGuard)
   @ApiOperation({ summary: 'Acknowledge a command result' })
   @ApiResponse({ status: 200, description: 'Command acknowledged' })

apps/api/src/devices/devices.service.spec.ts

@@ -0,0 +1,103 @@
+import { DevicesService } from './devices.service';
+
+type Snapshot = { tabId: string; customerName: string; version: number; totalInCents: number; itemCount: number };
+
+function createService({ previous = [] as Snapshot[], delivered = 0, expired = 0 } = {}) {
+  const tx = {
+    tabOwnership: { deleteMany: jest.fn(), upsert: jest.fn() },
+    openTabSnapshot: { deleteMany: jest.fn(), upsert: jest.fn() },
+    plannedTab: { updateMany: jest.fn().mockResolvedValue({ count: delivered }) },
+    device: { update: jest.fn() },
+  };
+  const prisma = {
+    openTabSnapshot: { findMany: jest.fn().mockResolvedValue(previous) },
+    tabOwnership: { findMany: jest.fn().mockResolvedValue([]) },
+    $transaction: jest.fn((fn: (t: typeof tx) => unknown) => fn(tx)),
+    remoteCommand: { updateMany: jest.fn().mockResolvedValue({ count: expired }), findMany: jest.fn().mockResolvedValue([]) },
+    plannedTab: { findMany: jest.fn().mockResolvedValue([{ customerName: 'Els' }]) },
+  };
+  const audit = { log: jest.fn() };
+  const realtime = { notify: jest.fn() };
+  const service = new DevicesService(prisma as any, audit as any, realtime as any);
+  return { service, prisma, tx, realtime };
+}
+
+const device = { id: 'd1', venueId: 'v1', revokedAt: null, lastSeenAt: new Date() };
+const tab = (id: string, customerName: string) => ({ id, customerName, version: 1, totalInCents: 0, itemCount: 0 });
+
+describe('DevicesService.sync planned customers', () => {
+  it('marks planned customers delivered once the device reports a tab with their name', async () => {
+    const { service, tx } = createService();
+
+    await service.sync(device, { deviceId: 'd1', openTabs: [tab('t1', 'Jan'), tab('t2', 'Piet')] });
+
+    expect(tx.plannedTab.updateMany).toHaveBeenCalledWith({
+      where: { venueId: 'v1', deliveredAt: null, customerName: { in: ['Jan', 'Piet'] } },
+      data: { deliveredAt: expect.any(Date) },
+    });
+  });
+
+  it('only sends customers that have not been delivered yet', async () => {
+    const { service, prisma } = createService();
+
+    const result = await service.sync(device, { deviceId: 'd1', openTabs: [] });
+
+    expect(prisma.plannedTab.findMany).toHaveBeenCalledWith(
+      expect.objectContaining({ where: { venueId: 'v1', deliveredAt: null } }),
+    );
+    expect(result.plannedTabNames).toEqual(['Els']);
+  });
+
+  it('does not mark anything delivered when the device reports no tabs', async () => {
+    const { service, tx } = createService();
+
+    await service.sync(device, { deviceId: 'd1', openTabs: [] });
+
+    expect(tx.plannedTab.updateMany).not.toHaveBeenCalled();
+  });
+});
+
+describe('DevicesService.sync live update signals', () => {
+  const jan = { id: 't1', customerName: 'Jan', version: 2, totalInCents: 500, itemCount: 1 };
+  const janSnapshot = { tabId: 't1', customerName: 'Jan', version: 2, totalInCents: 500, itemCount: 1 };
+
+  it('sends nothing when the sync changed nothing', async () => {
+    const { service, realtime } = createService({ previous: [janSnapshot] });
+
+    await service.sync(device, { deviceId: 'd1', openTabs: [jan] });
+
+    expect(realtime.notify).not.toHaveBeenCalled();
+  });
+
+  it('signals "tabs" when a tab changed', async () => {
+    const { service, realtime } = createService({ previous: [janSnapshot] });
+
+    await service.sync(device, { deviceId: 'd1', openTabs: [{ ...jan, version: 3, totalInCents: 900, itemCount: 2 }] });
+
+    expect(realtime.notify).toHaveBeenCalledWith('v1', 'tabs');
+  });
+
+  it('signals "tabs" when a tab disappeared', async () => {
+    const { service, realtime } = createService({ previous: [janSnapshot] });
+
+    await service.sync(device, { deviceId: 'd1', openTabs: [] });
+
+    expect(realtime.notify).toHaveBeenCalledWith('v1', 'tabs');
+  });
+
+  it('signals delivered customers and expired commands', async () => {
+    const { service, realtime } = createService({ previous: [janSnapshot], delivered: 1, expired: 2 });
+
+    await service.sync(device, { deviceId: 'd1', openTabs: [jan] });
+
+    expect(realtime.notify).toHaveBeenCalledWith('v1', 'planned-tabs', 'commands');
+  });
+
+  it('signals "device" when a terminal comes back online', async () => {
+    const { service, realtime } = createService({ previous: [janSnapshot] });
+
+    await service.sync({ ...device, lastSeenAt: new Date(Date.now() - 10 * 60 * 1000) }, { deviceId: 'd1', openTabs: [jan] });
+
+    expect(realtime.notify).toHaveBeenCalledWith('v1', 'device');
+  });
+});

apps/api/src/devices/devices.service.ts

@@ -7,6 +7,8 @@ import {
 import { v4 as uuidv4 } from 'uuid';
 import { PrismaService } from '../prisma/prisma.service';
 import { AuditService } from '../audit/audit.service';
+import { RealtimeService, RealtimeTopic } from '../realtime/realtime.service';
+import { DEVICE_ONLINE_THRESHOLD_MS } from '@kooltap/shared';
 import { SyncDto } from './dto/sync.dto';
 import { AckCommandDto } from './dto/ack-command.dto';
 
@@ -15,6 +17,7 @@ export class DevicesService {
   constructor(
     private prisma: PrismaService,
     private auditService: AuditService,
+    private realtime: RealtimeService,
   ) {}
 
   async sync(device: any, dto: SyncDto) {
@@ -29,6 +32,21 @@ export class DevicesService {
     const openTabs = dto.openTabs || [];
     const openTabIds = openTabs.map((t) => t.id);
 
+    // Devices sync often, so dashboards are only told about what actually changed
+    const previous = await this.prisma.openTabSnapshot.findMany({
+      where: { deviceId: device.id },
+      select: { tabId: true, customerName: true, version: true, totalInCents: true, itemCount: true },
+    });
+    const previousById = new Map(previous.map((s) => [s.tabId, s]));
+    const tabsChanged =
+      previous.length !== openTabs.length ||
+      openTabs.some((t) => {
+        const p = previousById.get(t.id);
+        return !p || p.customerName !== t.customerName || p.version !== t.version
+          || p.totalInCents !== t.totalInCents || p.itemCount !== t.itemCount;
+      });
+    let plannedDelivered = 0;
+
     // Ownership changes are detected up front so audit logging stays outside the transaction
     const takenOver = await this.prisma.tabOwnership.findMany({
       where: { venueId: device.venueId, tabId: { in: openTabIds }, deviceId: { not: device.id } },
@@ -90,6 +108,19 @@ export class DevicesService {
         });
       }
 
+      // A planned customer counts as delivered once the device reports a tab with that exact name
+      // (the terminal matches names exactly when creating tabs)
+      if (openTabs.length > 0) {
+        ({ count: plannedDelivered } = await tx.plannedTab.updateMany({
+          where: {
+            venueId: device.venueId,
+            deliveredAt: null,
+            customerName: { in: openTabs.map((t) => t.customerName) },
+          },
+          data: { deliveredAt: now },
+        }));
+      }
+
       await tx.device.update({
         where: { id: device.id },
         data: { lastSeenAt: now },
@@ -97,7 +128,7 @@ export class DevicesService {
     }, { timeout: 15000 });
 
     // Expire old commands
-    await this.prisma.remoteCommand.updateMany({
+    const expired = await this.prisma.remoteCommand.updateMany({
       where: {
         deviceId: device.id,
         status: 'pending',
@@ -131,13 +162,21 @@ export class DevicesService {
       orderBy: { createdAt: 'asc' },
     });
 
-    // Fetch planned tab names for this venue
+    // Only customers the terminal has not opened a tab for yet
     const plannedTabs = await this.prisma.plannedTab.findMany({
-      where: { venueId: device.venueId },
+      where: { venueId: device.venueId, deliveredAt: null },
       orderBy: { position: 'asc' },
       select: { customerName: true },
     });
 
+    const topics: RealtimeTopic[] = [];
+    if (tabsChanged) topics.push('tabs');
+    if (plannedDelivered > 0) topics.push('planned-tabs');
+    if (expired.count > 0) topics.push('commands');
+    // Only when the device comes back online; dashboards poll the device status themselves
+    if (now.getTime() - new Date(device.lastSeenAt).getTime() > DEVICE_ONLINE_THRESHOLD_MS) topics.push('device');
+    if (topics.length > 0) this.realtime.notify(device.venueId, ...topics);
+
     return {
       cursor,
       commands: commands.map((c) => ({
@@ -198,6 +237,8 @@ export class DevicesService {
       { commandId, status: dto.status, result: dto.result },
     );
 
+    this.realtime.notify(device.venueId, 'commands', 'tabs');
+
     return { ok: true };
   }
 
@@ -229,6 +270,8 @@ export class DevicesService {
 
     await this.auditService.log(venueId, 'system', deviceId, 'device_revoked', { deviceName: device.name });
 
+    this.realtime.notify(venueId, 'device', 'commands');
+
     return { ok: true };
   }
 }

apps/api/src/jobs/command-expiry.job.ts

@@ -1,13 +1,17 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { Cron, CronExpression } from '@nestjs/schedule';
 import { PrismaService } from '../prisma/prisma.service';
+import { RealtimeService } from '../realtime/realtime.service';
 import { v4 as uuidv4 } from 'uuid';
 
 @Injectable()
 export class CommandExpiryJob {
   private readonly logger = new Logger(CommandExpiryJob.name);
 
-  constructor(private prisma: PrismaService) {}
+  constructor(
+    private prisma: PrismaService,
+    private realtime: RealtimeService,
+  ) {}
 
   @Cron(CronExpression.EVERY_HOUR)
   async handleExpiredCommands() {
@@ -48,6 +52,10 @@ export class CommandExpiryJob {
       });
     }
 
+    for (const venueId of new Set(expiredCommands.map((c) => c.venueId))) {
+      this.realtime.notify(venueId, 'commands');
+    }
+
     this.logger.log(`Expired ${expiredCommands.length} commands`);
   }
 }

apps/api/src/planned-tabs/planned-tabs.service.spec.ts

@@ -24,7 +24,7 @@ function createService(initial: Omit<Row, 'venueId'>[]) {
     },
   };
   const prisma = { $transaction: (fn: (t: typeof tx) => unknown) => fn(tx) };
-  const service = new PlannedTabsService(prisma as any);
+  const service = new PlannedTabsService(prisma as any, { notify: jest.fn() } as any);
   return { service, rows: () => rows };
 }
 

apps/api/src/planned-tabs/planned-tabs.service.ts

@@ -1,16 +1,20 @@
 import { Injectable, NotFoundException, ConflictException, BadRequestException } from '@nestjs/common';
 import { PrismaService } from '../prisma/prisma.service';
+import { RealtimeService } from '../realtime/realtime.service';
 import { v4 as uuidv4 } from 'uuid';
 
 @Injectable()
 export class PlannedTabsService {
-  constructor(private prisma: PrismaService) {}
+  constructor(
+    private prisma: PrismaService,
+    private realtime: RealtimeService,
+  ) {}
 
   async findAll(venueId: string) {
     const tabs = await this.prisma.plannedTab.findMany({
       where: { venueId },
       orderBy: { position: 'asc' },
-      select: { id: true, customerName: true, email: true, position: true, createdAt: true },
+      select: { id: true, customerName: true, email: true, position: true, createdAt: true, deliveredAt: true },
     });
     return { plannedTabs: tabs };
   }
@@ -44,6 +48,7 @@ export class PlannedTabsService {
       select: { id: true, customerName: true, email: true, position: true, createdAt: true },
     });
 
+    this.realtime.notify(venueId, 'planned-tabs');
     return tab;
   }
 
@@ -67,6 +72,7 @@ export class PlannedTabsService {
       select: { id: true, customerName: true, email: true, position: true, createdAt: true },
     });
 
+    this.realtime.notify(venueId, 'planned-tabs', 'tabs');
     return result;
   }
 
@@ -86,7 +92,7 @@ export class PlannedTabsService {
       incoming.set(key, { customerName, email: entry.email || previous?.email });
     }
 
-    return this.prisma.$transaction(async (tx) => {
+    const result = await this.prisma.$transaction(async (tx) => {
       const existing = await tx.plannedTab.findMany({ where: { venueId }, orderBy: { position: 'asc' } });
       const existingByKey = new Map(existing.map((t) => [t.customerName.toLowerCase(), t]));
       let nextPosition = (existing[existing.length - 1]?.position ?? -1) + 1;
@@ -120,6 +126,9 @@ export class PlannedTabsService {
 
       return { created, updated, unchanged, removed };
     }, { timeout: 30000 });
+
+    this.realtime.notify(venueId, 'planned-tabs', 'tabs');
+    return result;
   }
 
   async remove(venueId: string, id: string) {
@@ -127,6 +136,7 @@ export class PlannedTabsService {
     if (!tab) throw new NotFoundException({ code: 'not_found', message: 'Planned tab not found' });
 
     await this.prisma.plannedTab.delete({ where: { id } });
+    this.realtime.notify(venueId, 'planned-tabs', 'tabs');
     return { ok: true };
   }
 }

apps/api/src/realtime/realtime.gateway.ts

@@ -0,0 +1,35 @@
+import { JwtService } from '@nestjs/jwt';
+import { OnGatewayConnection, OnGatewayInit, WebSocketGateway } from '@nestjs/websockets';
+import { Namespace, Socket } from 'socket.io';
+import { RealtimeService, venueRoom } from './realtime.service';
+
+@WebSocketGateway({
+  namespace: '/realtime',
+  cors: { origin: process.env.CORS_ORIGIN || 'http://localhost:3000', credentials: true },
+})
+export class RealtimeGateway implements OnGatewayInit, OnGatewayConnection {
+  constructor(
+    private jwtService: JwtService,
+    private realtime: RealtimeService,
+  ) {}
+
+  afterInit(namespace: Namespace) {
+    this.realtime.attach(namespace);
+
+    // Authenticate during the handshake so rejected clients never count as connected;
+    // the dashboard refreshes its access token on `unauthorized` and reconnects
+    namespace.use(async (socket, next) => {
+      try {
+        const payload = await this.jwtService.verifyAsync(String(socket.handshake.auth?.token ?? ''));
+        socket.data.venueId = payload.venueId;
+        next();
+      } catch {
+        next(new Error('unauthorized'));
+      }
+    });
+  }
+
+  async handleConnection(client: Socket) {
+    await client.join(venueRoom(client.data.venueId));
+  }
+}

apps/api/src/realtime/realtime.module.ts

@@ -0,0 +1,13 @@
+import { Global, Module } from '@nestjs/common';
+import { JwtModule } from '@nestjs/jwt';
+import { getJwtSecret } from '../auth/jwt-secret';
+import { RealtimeGateway } from './realtime.gateway';
+import { RealtimeService } from './realtime.service';
+
+@Global()
+@Module({
+  imports: [JwtModule.register({ secret: getJwtSecret() })],
+  providers: [RealtimeGateway, RealtimeService],
+  exports: [RealtimeService],
+})
+export class RealtimeModule {}

apps/api/src/realtime/realtime.service.ts

@@ -0,0 +1,24 @@
+import { Injectable } from '@nestjs/common';
+import { Namespace } from 'socket.io';
+
+/** What changed; the dashboard refetches the matching data over the normal API. */
+export type RealtimeTopic = 'tabs' | 'commands' | 'planned-tabs' | 'device' | 'reminders' | 'users';
+
+export const venueRoom = (venueId: string) => `venue:${venueId}`;
+
+/**
+ * Tells connected dashboards of a venue that data changed. Events only carry topic names,
+ * never data, so a socket can't see anything its owner couldn't fetch through the API.
+ */
+@Injectable()
+export class RealtimeService {
+  private namespace: Namespace | null = null;
+
+  attach(namespace: Namespace) {
+    this.namespace = namespace;
+  }
+
+  notify(venueId: string, ...topics: RealtimeTopic[]) {
+    this.namespace?.to(venueRoom(venueId)).emit('invalidate', { topics });
+  }
+}

apps/api/src/reminders/reminders.service.ts

@@ -1,5 +1,6 @@
 import { Injectable, BadRequestException, NotFoundException } from '@nestjs/common';
 import { PrismaService } from '../prisma/prisma.service';
+import { RealtimeService } from '../realtime/realtime.service';
 
 // Tabs stay open per customer (closing only settles them), so a customer can be reminded again
 // once this many days have passed since their last reminder.
@@ -19,7 +20,10 @@ export class RemindersService {
   private resendApiKey: string;
   private fromEmail: string;
 
-  constructor(private prisma: PrismaService) {
+  constructor(
+    private prisma: PrismaService,
+    private realtime: RealtimeService,
+  ) {
     this.resendApiKey = process.env.RESEND_API_KEY || '';
     this.fromEmail = process.env.RESEND_FROM_EMAIL || 'KoolTap <onboarding@resend.dev>';
   }
@@ -137,6 +141,8 @@ export class RemindersService {
       }
     }
 
+    if (sent > 0) this.realtime.notify(venueId, 'reminders');
+
     return { sent, total: candidates.length, errors: errors.length > 0 ? errors : undefined };
   }
 

apps/api/src/tabs/tabs.service.ts

@@ -22,6 +22,13 @@ export class TabsService {
     });
     const snapshotByKey = new Map(snapshots.map((s) => [`${s.deviceId}:${s.tabId}`, s]));
 
+    // Customer details (email for reminders) live on the planned customer with the same name
+    const planned = await this.prisma.plannedTab.findMany({
+      where: { venueId },
+      select: { id: true, customerName: true, email: true },
+    });
+    const plannedByName = new Map(planned.map((p) => [p.customerName, p]));
+
     const tabs = [];
     for (const ownership of ownerships) {
       const snapshot = snapshotByKey.get(`${ownership.deviceId}:${ownership.tabId}`);
@@ -38,6 +45,8 @@ export class TabsService {
           deviceLastSeenAt: ownership.device.lastSeenAt,
           deviceRevokedAt: ownership.device.revokedAt,
           lastReportedAt: snapshot.reportedAt,
+          plannedTabId: plannedByName.get(snapshot.customerName)?.id ?? null,
+          email: plannedByName.get(snapshot.customerName)?.email ?? null,
         });
       }
     }