102 lines
3.3 KiB
TypeScript
102 lines
3.3 KiB
TypeScript
import {
|
|
WebSocketGateway,
|
|
WebSocketServer,
|
|
OnGatewayConnection,
|
|
OnGatewayDisconnect,
|
|
SubscribeMessage,
|
|
MessageBody,
|
|
ConnectedSocket,
|
|
} from "@nestjs/websockets";
|
|
import { Server, Socket } from "socket.io";
|
|
import { Logger } from "@nestjs/common";
|
|
import { JwtService } from "@nestjs/jwt";
|
|
|
|
/**
|
|
* NotificationsGateway — C-6
|
|
* Real-time push of package status changes to connected portal clients.
|
|
*
|
|
* Connection flow:
|
|
* 1. Client connects with `auth: { token: "<jwt>" }` in socket options.
|
|
* 2. Gateway verifies JWT → places socket in room `user:<userId>`.
|
|
* 3. On package status change, NotificationsService calls `emitStatusChange()`.
|
|
* 4. All sockets in that user room receive `package:status` event.
|
|
*
|
|
* Client (Next.js portal):
|
|
* const socket = io("http://localhost:3001", { auth: { token: localStorage.getItem("mw_access") } });
|
|
* socket.on("package:status", (data) => { ... });
|
|
*/
|
|
@WebSocketGateway({
|
|
cors: {
|
|
origin: (process.env.CORS_ORIGINS ?? "http://localhost:3000").split(",").map(o => o.trim()),
|
|
credentials: true,
|
|
},
|
|
namespace: "/ws",
|
|
transports: ["websocket", "polling"],
|
|
})
|
|
export class NotificationsGateway implements OnGatewayConnection, OnGatewayDisconnect {
|
|
@WebSocketServer() server!: Server;
|
|
private readonly logger = new Logger(NotificationsGateway.name);
|
|
|
|
constructor(private jwtService: JwtService) {}
|
|
|
|
async handleConnection(client: Socket) {
|
|
try {
|
|
const token =
|
|
(client.handshake.auth as any)?.token ??
|
|
client.handshake.headers?.authorization?.replace("Bearer ", "");
|
|
|
|
if (!token) {
|
|
this.logger.warn(`[WS] Client ${client.id} rejected — no token`);
|
|
client.disconnect(true);
|
|
return;
|
|
}
|
|
|
|
const secret = process.env.JWT_SECRET ?? "changeme";
|
|
const payload = this.jwtService.verify(token, { secret });
|
|
const userId: string = payload.sub;
|
|
|
|
// Join personal room so we can target by userId
|
|
await client.join(`user:${userId}`);
|
|
client.data.userId = userId;
|
|
client.data.tenantId = payload.tenantId;
|
|
|
|
this.logger.log(`[WS] Connected: ${client.id} → user:${userId}`);
|
|
} catch {
|
|
this.logger.warn(`[WS] Client ${client.id} rejected — invalid token`);
|
|
client.disconnect(true);
|
|
}
|
|
}
|
|
|
|
handleDisconnect(client: Socket) {
|
|
this.logger.log(`[WS] Disconnected: ${client.id}`);
|
|
}
|
|
|
|
/** Emitted by NotificationsService on every package status change */
|
|
emitStatusChange(userId: string, pkg: {
|
|
id: string;
|
|
trackingId: string;
|
|
status: string;
|
|
description?: string;
|
|
}) {
|
|
this.server.to(`user:${userId}`).emit("package:status", {
|
|
packageId: pkg.id,
|
|
trackingId: pkg.trackingId,
|
|
status: pkg.status,
|
|
description: pkg.description ?? null,
|
|
at: new Date().toISOString(),
|
|
});
|
|
this.logger.log(`[WS] Emitted package:status to user:${userId} — ${pkg.trackingId} → ${pkg.status}`);
|
|
}
|
|
|
|
/** Broadcast to all sockets in a tenant room */
|
|
emitToTenant(tenantId: string, event: string, data: any) {
|
|
this.server.to(`tenant:${tenantId}`).emit(event, data);
|
|
}
|
|
|
|
/** Ping/pong — optional keep-alive */
|
|
@SubscribeMessage("ping")
|
|
handlePing(@ConnectedSocket() client: Socket, @MessageBody() _data: any) {
|
|
client.emit("pong", { at: new Date().toISOString() });
|
|
}
|
|
}
|