fix(api): broadcast stream status transitions over the streams socket

api/src/gateways/streams.gateway.spec.ts

@@ -387,6 +387,38 @@ describe("StreamsGateway", () => {
       ])
     })
 
+    // Issue #519: the broadcast must land on the exact room a client
+    // joined via `stream:subscribe` — this ties the subscribe handshake
+    // to the emit helpers end-to-end.
+    it("broadcasts a status emit to the room a subscribed socket joined", () => {
+      const { server, events } = makeServer()
+      gateway.server = server as unknown as any
+
+      const socket = makeSocket({ data: { userId: 55 } })
+      gateway.handleSubscribe(socket as unknown as any, { streamId: "abc" })
+
+      gateway.emitStarted({
+        streamId: "abc",
+        userId: 55,
+        startedAt: "2026-06-16T00:00:00Z",
+      })
+
+      expect(events).toEqual([
+        {
+          room: "stream:abc",
+          event: STREAM_EVENTS.STARTED,
+          payload: {
+            streamId: "abc",
+            userId: 55,
+            startedAt: "2026-06-16T00:00:00Z",
+          },
+        },
+      ])
+      // The emit targets the stream room, not the user room the socket
+      // joined during the connection handshake.
+      expect(socket.join).toHaveBeenCalledWith("stream:abc")
+    })
+
     it("broadcasts notifications only to the target user's room", () => {
       const { server, events } = makeServer()
       gateway.server = server as unknown as any

api/src/streams/streams.module.ts

@@ -1,16 +1,18 @@
 import { CacheModule } from "@nestjs/cache-manager"
 import { Module } from "@nestjs/common"
-import { streamsCacheConfig } from "../config/cache.config"
+
+import { StreamsController } from "./streams.controller"
 import { AuthModule } from "../auth/auth.module"
+import { StreamsDbRepository } from "./repository/streams-db.repository"
+import { StreamsRepository } from "./repository/streams.repository"
+import { StreamsService } from "./streams.service"
 import { AuthGuard } from "../common/guards/auth.guard"
 import { StreamOwnershipGuard } from "../common/guards/stream-ownership.guard"
 import { StreamOwnershipService } from "../common/guards/stream-ownership.service"
+import { streamsCacheConfig } from "../config/cache.config"
+import { GatewaysModule } from "../gateways/gateways.module"
 import { TagsModule } from "../tags/tags.module"
 import { WebhooksModule } from "../webhooks/webhooks.module"
-import { StreamsDbRepository } from "./repository/streams-db.repository"
-import { StreamsRepository } from "./repository/streams.repository"
-import { StreamsController } from "./streams.controller"
-import { StreamsService } from "./streams.service"
 
 /**
  * Injection token used to swap the streams repository implementation.
@@ -32,6 +34,9 @@ const isTest = process.env.NODE_ENV === "test"
     // TagsModule provides TagsService so StreamsService.list() can
     // batch-load tags for the streams on the current page (#330).
     TagsModule,
+    // GatewaysModule provides StreamsGateway so StreamsService can
+    // broadcast status transitions to subscribed sockets (#519).
+    GatewaysModule,
     CacheModule.register(streamsCacheConfig()),
   ],
   controllers: [StreamsController],

api/src/streams/streams.service.spec.ts

@@ -1,10 +1,13 @@
 import { ConflictException, NotFoundException } from "@nestjs/common"
+import * as fc from "fast-check"
+
+import { Stream } from "./stream.entity"
+import { StreamsGateway } from "../gateways/streams.gateway"
 import { Tag } from "../tags/tag.entity"
 import { TagsService } from "../tags/tags.service"
-import { Stream } from "./stream.entity"
+import { WebhooksService } from "../webhooks/webhooks.service"
 import { StreamsRepository } from "./repository/streams.repository"
 import { StreamsService } from "./streams.service"
-import { WebhooksService } from "../webhooks/webhooks.service"
 
 describe("StreamsService", () => {
   let service: StreamsService
@@ -19,6 +22,11 @@ describe("StreamsService", () => {
   }
   let mockWebhooksService: { dispatchStreamEvent: jest.Mock }
   let mockTagsService: { listForStreamIds: jest.Mock }
+  let mockGateway: {
+    emitStarted: jest.Mock
+    emitStopped: jest.Mock
+    emitError: jest.Mock
+  }
 
   /** Helper to build a fully-typed Stream with the new visibility field. */
   function streamFixture(overrides: Partial<Stream> = {}): Stream {
@@ -51,10 +59,16 @@ describe("StreamsService", () => {
     mockTagsService = {
       listForStreamIds: jest.fn().mockResolvedValue(new Map()),
     }
+    mockGateway = {
+      emitStarted: jest.fn(),
+      emitStopped: jest.fn(),
+      emitError: jest.fn(),
+    }
     service = new StreamsService(
       mockRepo as unknown as StreamsRepository,
       mockWebhooksService as unknown as WebhooksService,
       mockTagsService as unknown as TagsService,
+      mockGateway as unknown as StreamsGateway,
     )
   })
 
@@ -204,6 +218,117 @@ describe("StreamsService", () => {
     )
   })
 
+  // ── Issue #519 — socket broadcasts on status transitions ────────────────
+
+  it("update inactive -> active emits stream:started to the gateway", async () => {
+    const existing = streamFixture({ status: "inactive", userId: 7 })
+    const updated = { ...existing, status: "active" }
+    mockRepo.findById.mockResolvedValue(existing)
+    mockRepo.update.mockResolvedValue(updated)
+
+    await service.update(1, { status: "active" })
+
+    expect(mockGateway.emitStarted).toHaveBeenCalledWith({
+      streamId: 1,
+      userId: 7,
+      startedAt: expect.any(String),
+    })
+    expect(mockGateway.emitStopped).not.toHaveBeenCalled()
+    expect(mockGateway.emitError).not.toHaveBeenCalled()
+  })
+
+  it("update active -> inactive emits stream:stopped to the gateway", async () => {
+    const existing = streamFixture({ status: "active", userId: 7 })
+    const updated = { ...existing, status: "inactive" }
+    mockRepo.findById.mockResolvedValue(existing)
+    mockRepo.update.mockResolvedValue(updated)
+
+    await service.update(1, { status: "inactive" })
+
+    expect(mockGateway.emitStopped).toHaveBeenCalledWith({
+      streamId: 1,
+      userId: 7,
+      stoppedAt: expect.any(String),
+    })
+    expect(mockGateway.emitStarted).not.toHaveBeenCalled()
+    expect(mockGateway.emitError).not.toHaveBeenCalled()
+  })
+
+  it("update * -> error emits stream:error to the gateway with code and message", async () => {
+    const existing = streamFixture({ status: "inactive", userId: 7 })
+    const updated = { ...existing, status: "error" }
+    mockRepo.findById.mockResolvedValue(existing)
+    mockRepo.update.mockResolvedValue(updated)
+
+    await service.update(1, { status: "error" })
+
+    expect(mockGateway.emitError).toHaveBeenCalledWith({
+      streamId: 1,
+      userId: 7,
+      occurredAt: expect.any(String),
+      code: "STREAM_ERROR",
+      message: "stream 1 entered error state",
+    })
+    expect(mockGateway.emitStarted).not.toHaveBeenCalled()
+    expect(mockGateway.emitStopped).not.toHaveBeenCalled()
+  })
+
+  it("update error -> inactive emits stream:stopped to the gateway", async () => {
+    const existing = streamFixture({ status: "error", userId: 7 })
+    const updated = { ...existing, status: "inactive" }
+    mockRepo.findById.mockResolvedValue(existing)
+    mockRepo.update.mockResolvedValue(updated)
+
+    await service.update(1, { status: "inactive" })
+
+    expect(mockGateway.emitStopped).toHaveBeenCalledWith({
+      streamId: 1,
+      userId: 7,
+      stoppedAt: expect.any(String),
+    })
+  })
+
+  it("update without a status change emits nothing to the gateway", async () => {
+    const existing = streamFixture({ status: "active", userId: 7, name: "old" })
+    const updated = { ...existing, name: "renamed" }
+    mockRepo.findById.mockResolvedValue(existing)
+    mockRepo.update.mockResolvedValue(updated)
+
+    await service.update(1, { name: "renamed" })
+
+    expect(mockGateway.emitStarted).not.toHaveBeenCalled()
+    expect(mockGateway.emitStopped).not.toHaveBeenCalled()
+    expect(mockGateway.emitError).not.toHaveBeenCalled()
+  })
+
+  it("invalid transitions emit nothing to the gateway", async () => {
+    const existing = streamFixture({ status: "active" })
+    mockRepo.findById.mockResolvedValue(existing)
+
+    await expect(service.update(2, { status: "active" })).rejects.toThrow(
+      ConflictException,
+    )
+    expect(mockGateway.emitStarted).not.toHaveBeenCalled()
+    expect(mockGateway.emitStopped).not.toHaveBeenCalled()
+    expect(mockGateway.emitError).not.toHaveBeenCalled()
+  })
+
+  it("a rejected webhook dispatch does not suppress the socket emit", async () => {
+    const existing = streamFixture({ status: "inactive", userId: 7 })
+    const updated = { ...existing, status: "active" }
+    mockRepo.findById.mockResolvedValue(existing)
+    mockRepo.update.mockResolvedValue(updated)
+    mockWebhooksService.dispatchStreamEvent.mockRejectedValue(
+      new Error("subscriber unreachable"),
+    )
+
+    await service.update(1, { status: "active" })
+
+    // The webhook failure is swallowed (fire-and-forget) and the socket
+    // emit still fires — the two side effects are independent.
+    expect(mockGateway.emitStarted).toHaveBeenCalledTimes(1)
+  })
+
   it("update status active -> inactive dispatches a stream:stopped webhook event", async () => {
     const existing = streamFixture({ status: "active", userId: 7 })
     const updated = { ...existing, status: "inactive" }
@@ -344,12 +469,14 @@ describe("StreamsService", () => {
 // ============================================
 // PROPERTY-BASED TESTS FOR STREAM STATUS TRANSITIONS + VISIBILITY TRANSITIONS
 // ============================================
-import * as fc from "fast-check";
+// These legacy property-based blocks were authored with loose `any`
+// mocks; the disable is scoped to this section (it runs to EOF) and
+// mirrors the file-level disable convention in streams.gateway.spec.ts.
+/* eslint-disable @typescript-eslint/no-explicit-any */
 
 describe("StreamsService - Property-Based Tests", () => {
   let service: StreamsService;
   let mockRepo: any;
-  let mockWebhooksService: { dispatchStreamEvent: jest.Mock };
 
   beforeEach(() => {
     mockRepo = {

api/src/streams/streams.service.ts

@@ -2,22 +2,26 @@ import {
   ConflictException,
   Injectable,
   NotFoundException,
+  Optional,
 } from "@nestjs/common"
-import type { StreamEventRecord } from "@xstreamroll/types"
+
+import { Stream } from "./stream.entity"
 import { PaginatedResult } from "../common/dto/pagination.dto"
-import type {
-  StreamListFilter,
-  StreamUpdateChanges,
-  StreamCreateParams,
-} from "./repository/streams.repository"
-import { PendingStreamEvent } from "./repository/streams.repository"
 import { STREAM_EVENTS } from "../gateways/stream-events"
+import { StreamsGateway } from "../gateways/streams.gateway"
 import { TagsService } from "../tags/tags.service"
 import { WebhooksService } from "../webhooks/webhooks.service"
-import { StreamsRepository } from "./repository/streams.repository"
 import { StreamAnalyticsDto } from "./dto/stream-analytics.dto"
-import { Stream } from "./stream.entity"
+import { PendingStreamEvent } from "./repository/streams.repository"
+import { StreamsRepository } from "./repository/streams.repository"
+
 import type { StreamVisibility } from "./dto/visibility"
+import type {
+  StreamListFilter,
+  StreamUpdateChanges,
+  StreamCreateParams,
+} from "./repository/streams.repository"
+import type { StreamEventRecord } from "@xstreamroll/types"
 
 export interface PagedStreams extends PaginatedResult<Stream> {
   hasMore: boolean
@@ -36,6 +40,10 @@ export class StreamsService {
     private readonly repo: StreamsRepository,
     private readonly webhooksService: WebhooksService,
     private readonly tagsService: TagsService,
+    // Optional so unit tests and any consumer that predates the socket
+    // wiring can construct the service without a gateway. Mirrors the
+    // `NotificationsService` injection pattern.
+    @Optional() private readonly gateway?: StreamsGateway,
   ) {}
 
   async create(params: StreamCreateParams): Promise<Stream> {
@@ -115,35 +123,63 @@ export class StreamsService {
     })
 
     if (changes.status !== undefined && changes.status !== stream.status) {
-      this.dispatchStatusWebhook(updated, changes.status)
+      this.dispatchStatusSideEffects(updated, changes.status)
     }
 
     return updated
   }
 
   /**
-   * Fires the webhook event matching a stream's new status. Runs in the
-   * background — a slow or unreachable subscriber must never delay the
-   * status transition response.
+   * Fires the side effects that accompany a stream status transition:
+   * the matching webhook event (fire-and-forget) and the matching
+   * socket broadcast scoped to the stream's room (issue #519). Both
+   * paths derive their payloads from the same `now`/`base` values so
+   * they cannot drift, and each is independent — a failed webhook
+   * dispatch never suppresses the socket emit.
    */
-  private dispatchStatusWebhook(stream: Stream, newStatus: string): void {
+  private dispatchStatusSideEffects(stream: Stream, newStatus: string): void {
     const event = STATUS_TO_WEBHOOK_EVENT[newStatus]
     if (!event) return
 
     const now = new Date().toISOString()
-    const payload =
+    const base = { streamId: stream.id, userId: stream.userId } as const
+
+    const webhookPayload =
       newStatus === "error"
-        ? { streamId: stream.id, userId: stream.userId, occurredAt: now }
+        ? { ...base, occurredAt: now }
         : newStatus === "active"
-          ? { streamId: stream.id, userId: stream.userId, startedAt: now }
-          : { streamId: stream.id, userId: stream.userId, stoppedAt: now }
+          ? { ...base, startedAt: now }
+          : { ...base, stoppedAt: now }
 
+    // Webhook fan-out is fire-and-forget — a slow or unreachable
+    // subscriber must never delay the status transition response.
     this.webhooksService
-      .dispatchStreamEvent(stream.id, event, payload)
+      .dispatchStreamEvent(stream.id, event, webhookPayload)
       .catch(() => {
         // dispatchStreamEvent already logs; swallow here so a webhook
         // fan-out failure never surfaces as an update() error.
       })
+
+    // Socket broadcast is independent of webhook delivery: a failed
+    // webhook dispatch must not suppress the live status update.
+    switch (newStatus) {
+      case "active":
+        this.gateway?.emitStarted({ ...base, startedAt: now })
+        break
+      case "inactive":
+        this.gateway?.emitStopped({ ...base, stoppedAt: now })
+        break
+      case "error":
+        // The webhook payload has no code/message; the socket wire
+        // contract requires them, so the emit supplies defaults.
+        this.gateway?.emitError({
+          ...base,
+          occurredAt: now,
+          code: "STREAM_ERROR",
+          message: `stream ${stream.id} entered error state`,
+        })
+        break
+    }
   }
 
   async delete(id: number): Promise<void> {