feat(sse-module): server sent event

CI/E2E/docker-compose-local.yaml

@@ -2,10 +2,22 @@ version: "3.2"
 services:
   mongodb:
     image: mongo:latest
+    command: ["--replSet", "rs0", "--bind_ip_all"]
+    environment:
+      MONGO_RS_HOST: "localhost:27017"
     volumes:
       - "mongodb_data:/data/db"
     ports:
       - "27017:27017"
+    healthcheck:
+      test: >
+        mongosh --quiet --eval "
+          try { if (rs.status().ok !== 1) quit(1) }
+          catch { rs.initiate({ _id: 'rs0', members: [{ _id: 0, host: '$$MONGO_RS_HOST' }] }) }
+        "
+      interval: 10s
+      start_period: 40s
+      retries: 5
 
   scichat-loopback:
     image: dacat/scichat-loopback:e2e

CI/E2E/docker-compose.yaml

@@ -8,17 +8,23 @@ services:
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
   mongodb:
-    image: "mongo:latest"
-    ports:
-      - "27017:27017"
+    image: mongo:latest
+    command: ["--replSet", "rs0", "--bind_ip_all"]
+    environment:
+      MONGO_RS_HOST: "localhost:27017"
     volumes:
       - "mongodb_data:/data/db"
+    ports:
+      - "27017:27017"
     healthcheck:
-      test: echo 'db.runCommand("ping").ok' | mongosh mongodb:27017/test --quiet
+      test: >
+        mongosh --quiet --eval "
+           try { if (rs.status().ok !== 1) quit(1) }
+          catch { rs.initiate({ _id: 'rs0', members: [{ _id: 0, host: '$$MONGO_RS_HOST' }] }) }
+        "
       interval: 10s
-      timeout: 10s
-      retries: 5
       start_period: 40s
+      retries: 5
   scichat-loopback:
     image: dacat/scichat-loopback:e2e
     command:

src/app.module.ts

@@ -34,7 +34,6 @@ import { JobConfigModule } from "./config/job-config/jobconfig.module";
 import { CoreJobActionCreators } from "./config/job-config/actions/corejobactioncreators.module";
 import { HttpModule, HttpService } from "@nestjs/axios";
 import { MSGraphMailTransport } from "./common/graph-mail";
-import { TransportType } from "@nestjs-modules/mailer/dist/interfaces/mailer-options.interface";
 import { MetricsModule } from "./metrics/metrics.module";
 import {
   GenericHistory,
@@ -46,6 +45,8 @@ import { RuntimeConfigModule } from "./config/runtime-config/runtime-config.modu
 import { MetadataKeysModule } from "./metadata-keys/metadatakeys.module";
 import { OidcClientModule } from "./common/openid-client/openid-client.module";
 import { ThrottlerModule } from "@nestjs/throttler";
+import { SseModule } from "./serverSentEvent/sse.module";
+import type { MailerOptions } from "@nestjs-modules/mailer";
 
 @Module({
   imports: [
@@ -55,6 +56,7 @@ import { ThrottlerModule } from "@nestjs/throttler";
       cache: true,
     }),
     AuthModule,
+    SseModule,
     OidcClientModule,
     CaslModule,
     AttachmentsModule,
@@ -88,7 +90,7 @@ import { ThrottlerModule } from "@nestjs/throttler";
         configService: ConfigService,
         httpService: HttpService,
       ) => {
-        let transport: TransportType;
+        let transport: MailerOptions["transport"];
         const transportType = configService
           .get<string>("email.type")
           ?.toLowerCase();

src/auth/auth.service.ts

@@ -19,6 +19,7 @@ import { ReturnedUserDto } from "src/users/dto/returned-user.dto";
 import { CreateUserSettingsDto } from "src/users/dto/create-user-settings.dto";
 import { OidcClientService } from "../common/openid-client/openid-client.service";
 import { OidcAuthService } from "src/common/openid-client/openid-auth.service";
+import { JWTUser } from "./interfaces/jwt-user.interface";
 
 @Injectable()
 export class AuthService {
@@ -94,6 +95,13 @@ export class AuthService {
     };
   }
 
+  createSseTicket(user: JWTUser): string {
+    return this.jwtService.sign(
+      { ...user, purpose: "sse" },
+      { expiresIn: "60s" },
+    );
+  }
+
   async logout(req: Request) {
     const logoutURL = this.configService.get<string>("logoutURL") || "";
     const expressSessionSecret = this.configService.get<string>(

src/auth/strategies/jwt.strategy.ts

@@ -1,11 +1,17 @@
 import { ExtractJwt, Strategy } from "passport-jwt";
 import { PassportStrategy } from "@nestjs/passport";
-import { Injectable } from "@nestjs/common";
+import { Injectable, UnauthorizedException } from "@nestjs/common";
+import { Request } from "express";
 import { RolesService } from "src/users/roles.service";
 import { UsersService } from "src/users/users.service";
 import { JWTUser } from "../interfaces/jwt-user.interface";
 import { User } from "src/users/schemas/user.schema";
 import { ConfigService } from "@nestjs/config";
+import {
+  SSE_STREAM_PATH,
+  fromSseTicket,
+  requireJwtSecret,
+} from "src/auth/utils/jwt.util";
 
 @Injectable()
 export class JwtStrategy extends PassportStrategy(Strategy) {
@@ -15,13 +21,26 @@ export class JwtStrategy extends PassportStrategy(Strategy) {
     private usersService: UsersService,
   ) {
     super({
-      jwtFromRequest: ExtractJwt.fromAuthHeaderAsBearerToken(),
+      jwtFromRequest: ExtractJwt.fromExtractors([
+        ExtractJwt.fromAuthHeaderAsBearerToken(),
+        fromSseTicket,
+      ]),
       ignoreExpiration: false,
-      secretOrKey: configService.get<string>("jwt.secret") || "defaultSecret",
+      secretOrKey: requireJwtSecret(configService),
+      passReqToCallback: true,
     });
   }
 
-  async validate(payload: Omit<User, "password">) {
+  async validate(
+    request: Request,
+    payload: Omit<User, "password"> & { purpose?: string },
+  ) {
+    const isSsePath = request.path === SSE_STREAM_PATH;
+    const isSseTicket = payload.purpose === "sse";
+    if (isSsePath !== isSseTicket) {
+      throw new UnauthorizedException();
+    }
+
     const roles = await this.rolesService.find({ userId: payload._id });
 
     const userIdentity = await this.usersService.findByIdUserIdentity(

src/auth/utils/jwt.util.ts

@@ -0,0 +1,21 @@
+import { Request } from "express";
+import { ConfigService } from "@nestjs/config";
+
+export const SSE_STREAM_PATH = "/api/v3/events/stream";
+
+export const fromSseTicket = (req: Request): string | null => {
+  if (req.path !== SSE_STREAM_PATH) return null;
+  const ticket = req.query?.ticket;
+  return typeof ticket === "string" ? ticket : null;
+};
+
+export const requireJwtSecret = (configService: ConfigService): string => {
+  const secret = configService.get<string>("jwt.secret");
+  if (secret) return secret;
+
+  if (process.env.NODE_ENV === "production") {
+    throw new Error("jwt.secret must be configured in production");
+  }
+  console.warn("jwt.secret is not set, falling back to an insecure default");
+  return "defaultSecret";
+};

src/config/configuration.ts

@@ -5,6 +5,7 @@ import { DEFAULT_PROPOSAL_TYPE } from "src/proposals/schemas/proposal.schema";
 import localconfiguration from "./localconfiguration";
 
 const configuration = () => {
+  const jwtSecret = process.env.JWT_SECRET;
   const accessGroupsStaticValues =
     process.env.ACCESS_GROUPS_STATIC_VALUES || "";
   const adminGroups = process.env.ADMIN_GROUPS || "";
@@ -326,7 +327,7 @@ const configuration = () => {
     httpMaxRedirects: process.env.HTTP_MAX_REDIRECTS ?? 5,
     httpTimeOut: process.env.HTTP_TIMEOUT ?? 5000,
     jwt: {
-      secret: process.env.JWT_SECRET,
+      secret: jwtSecret,
       expiresIn: parseInt(process.env.JWT_EXPIRES_IN ?? "3600", 10),
       neverExpires: process.env.JWT_NEVER_EXPIRES ?? "100y",
     },

src/config/frontend.config.json

@@ -5,6 +5,7 @@
   },
   "statusBannerMessage": "",
   "statusBannerCode": "INFO",
+  "realTimeUpdatesEnabled": false,
   "autoApplyFilters": false,
   "accessTokenPrefix": "Bearer ",
   "addDatasetEnabled": false,

src/serverSentEvent/interfaces/sse-event.interface.ts

@@ -0,0 +1,30 @@
+import { Subject } from "rxjs";
+import { JWTUser } from "src/auth/interfaces/jwt-user.interface";
+import { MessageEvent } from "@nestjs/common";
+
+export interface SseClient {
+  user: JWTUser;
+  subject: Subject<MessageEvent>;
+}
+
+/**
+ * Event action names emitted over SSE. Only "created" is produced today
+ * (insert operations); "updated"/"deleted" are reserved for future use.
+ */
+export type SseAction = "created" | "updated" | "deleted";
+
+export interface HasAccessGroups {
+  ownerGroup?: string;
+  accessGroups?: string[];
+}
+
+export interface SseEvent<T extends HasAccessGroups = HasAccessGroups> {
+  entity: string;
+  action: SseAction;
+  message: T;
+}
+
+export interface SseConnectionsReport {
+  connections: number;
+  users: Record<string, number>;
+}

src/serverSentEvent/interfaces/sse-registry.interface.ts

@@ -0,0 +1,31 @@
+import { ClassConstructor } from "class-transformer";
+
+/**
+ * MongoDB collection names that the SSE listener may watch.
+ */
+export type WatchableCollection =
+  | "Attachment"
+  | "RuntimeConfig"
+  | "Dataset"
+  | "Proposal"
+  | "Sample"
+  | "PublishedData"
+  | "MetadataKeys"
+  | "Datablock"
+  | "Instrument"
+  | "Job"
+  | "OrigDatablock"
+  | "History";
+
+/**
+ * How a watched collection maps to an SSE entity name and the DTO used to
+ * serialize raw MongoDB documents before they are pushed to clients.
+ */
+export interface SseRegistryEntry {
+  entity: string;
+  dto: ClassConstructor<object>;
+}
+
+export type SseRegistry = Partial<
+  Record<WatchableCollection, SseRegistryEntry>
+>;

src/serverSentEvent/sse.controller.ts

@@ -0,0 +1,83 @@
+import {
+  Controller,
+  Sse,
+  MessageEvent,
+  UseGuards,
+  Req,
+  Get,
+  Post,
+} from "@nestjs/common";
+import { Observable } from "rxjs";
+
+import { Request } from "express";
+import {
+  ApiBearerAuth,
+  ApiOperation,
+  ApiQuery,
+  ApiTags,
+} from "@nestjs/swagger";
+import { PoliciesGuard } from "src/casl/guards/policies.guard";
+import { Action } from "src/casl/action.enum";
+import { AppAbility } from "src/casl/casl-ability.factory";
+import { CheckPolicies } from "src/casl/decorators/check-policies.decorator";
+import { DatasetClass } from "src/datasets/schemas/dataset.schema";
+import { JWTUser } from "src/auth/interfaces/jwt-user.interface";
+import { RuntimeConfig } from "src/config/runtime-config/schemas/runtime-config.schema";
+import { SseService } from "./sse.service";
+import { AuthService } from "src/auth/auth.service";
+
+@ApiTags("events")
+@Controller("events")
+@ApiBearerAuth()
+export class SseController {
+  constructor(
+    private readonly sseService: SseService,
+    private readonly authService: AuthService,
+  ) {}
+
+  @Sse("stream")
+  @UseGuards(PoliciesGuard)
+  @CheckPolicies("datasets", (ability: AppAbility) =>
+    ability.can(Action.DatasetRead, DatasetClass),
+  )
+  @ApiOperation({
+    summary: "Subscribe to server-sent events.",
+    description:
+      "Opens a text/event-stream connection pushing events for documents the authenticated user is allowed to read.",
+  })
+  @ApiQuery({
+    name: "token",
+    required: false,
+    description:
+      "JWT access token. Alternative to the Authorization header for EventSource clients that cannot set request headers.",
+  })
+  stream(@Req() request: Request): Observable<MessageEvent> {
+    const user = request.user as JWTUser;
+    return this.sseService.getEvents(user);
+  }
+
+  @Get("connections")
+  @UseGuards(PoliciesGuard)
+  @CheckPolicies(
+    "runtimeconfig",
+    (ability: AppAbility) =>
+      ability.can(Action.RuntimeConfigUpdate, RuntimeConfig), //TODO: define a correct policy for monitoring connections
+  )
+  @ApiOperation({
+    summary: "List active SSE connections on this instance.",
+  })
+  connections() {
+    return this.sseService.getAllConnections();
+  }
+
+  @UseGuards(PoliciesGuard)
+  @CheckPolicies("datasets", (ability: AppAbility) =>
+    ability.can(Action.DatasetRead, DatasetClass),
+  )
+  @Post("ticket")
+  async createTicket(@Req() request: Request): Promise<{ ticket: string }> {
+    return {
+      ticket: this.authService.createSseTicket(request.user as JWTUser),
+    };
+  }
+}