diff --git a/api/src/gateways/streams.gateway.spec.ts b/api/src/gateways/streams.gateway.spec.ts index ff22649..c8107bf 100644 --- a/api/src/gateways/streams.gateway.spec.ts +++ b/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 diff --git a/api/src/streams/streams.module.ts b/api/src/streams/streams.module.ts index d3d7174..e0a0d8e 100644 --- a/api/src/streams/streams.module.ts +++ b/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], diff --git a/api/src/streams/streams.service.spec.ts b/api/src/streams/streams.service.spec.ts index 402d707..d956900 100644 --- a/api/src/streams/streams.service.spec.ts +++ b/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 { @@ -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 = { diff --git a/api/src/streams/streams.service.ts b/api/src/streams/streams.service.ts index 0488198..6e31be4 100644 --- a/api/src/streams/streams.service.ts +++ b/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 { 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 { @@ -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 {