test(contracts): cover webhook, event-replay, analytics, and notification endpoints

api/src/contract-provider.spec.ts

@@ -18,6 +18,7 @@ import { Test, TestingModule } from "@nestjs/testing"
 import {
   authContracts,
   loginBody,
+  notificationsContracts,
   PLACEHOLDER,
   registerBody,
   resolvePath,
@@ -38,6 +39,9 @@ import { JwtExtractorService } from "./common/guards/jwt-extractor.service"
 import { StreamOwnershipGuard } from "./common/guards/stream-ownership.guard"
 import { StreamOwnershipService } from "./common/guards/stream-ownership.service"
 import createJwtConfig, { createRefreshJwtConfig } from "./config/jwt.config"
+import { NotificationsController } from "./notifications/notifications.controller"
+import { NotificationsService } from "./notifications/notifications.service"
+import { NotificationsRepository } from "./notifications/repository/notifications.repository"
 import { StreamsRepository } from "./streams/repository/streams.repository"
 import { StreamApiKeyGuard } from "./streams/stream-api-key.guard"
 import { StreamsController } from "./streams/streams.controller"
@@ -132,6 +136,7 @@ describe("Contract provider verification (api)", () => {
         StreamTagsController,
         AuthController,
         WebhooksController,
+        NotificationsController,
       ],
       providers: [
         StreamsService,
@@ -147,6 +152,8 @@ describe("Contract provider verification (api)", () => {
           provide: WebhookDeliveriesRepository,
           useValue: deliveriesRepository,
         },
+        NotificationsService,
+        NotificationsRepository,
         AuthGuard,
         JwtExtractorService,
         StreamOwnershipGuard,
@@ -253,6 +260,20 @@ describe("Contract provider verification (api)", () => {
     delivery.nextAttemptAt = null
     delivery.lastError = "connection refused"
     existingDeliveryId = String(delivery.id)
+
+    // Record one processed event so `list-stream-events` validates the
+    // non-empty shape — most importantly the stringified id/streamId.
+    await streamsRepository.recordEvent(stream.id, {
+      eventType: "stream:started",
+      payload: { streamId: stream.id },
+      occurredAt: "2026-08-01T00:00:00.000Z",
+    })
+
+    // Seed one unread notification so `list-notifications` validates the
+    // non-empty shape (numeric ids, ISO timestamps).
+    await moduleFixture
+      .get(NotificationsRepository)
+      .create(userId, "stream:started", { streamId: stream.id })
   })
 
   afterAll(async () => {
@@ -362,6 +383,22 @@ describe("Contract provider verification (api)", () => {
     })
   })
 
+  describe.each(notificationsContracts)("$name", (contract) => {
+    it(contract.description, async () => {
+      const res = await execute(contract)
+
+      expect(res.status).toBe(contract.response.status)
+      const result = contract.response.schema.safeParse(res.body)
+      if (!result.success) {
+        throw new Error(
+          `${contract.name}: response did not satisfy the contract schema\n` +
+            `${JSON.stringify(result.error.format(), null, 2)}\n` +
+            `body: ${JSON.stringify(res.body, null, 2)}`,
+        )
+      }
+    })
+  })
+
   it("login contract uses credentials the register contract actually created", () => {
     // Sanity check that the two contract fixtures stay in sync with each
     // other — if this ever fails, `auth.contract.ts` was edited so the

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

@@ -5,7 +5,6 @@ import {
 } 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"

tests/contracts/src/auth.contract.ts

@@ -1,7 +1,8 @@
-import type { CreateUserDto } from "@xstreamroll/types"
-import type { Contract } from "./contract"
 import { authResponseSchema } from "./schemas"
 
+import type { Contract } from "./contract"
+import type { CreateUserDto } from "@xstreamroll/types"
+
 export const registerBody: CreateUserDto = {
   username: "contractuser",
   email: "[email protected]",

tests/contracts/src/index.ts

@@ -18,8 +18,10 @@ export * from "./schemas"
 export * from "./streams.contract"
 export * from "./auth.contract"
 export * from "./webhooks.contract"
+export * from "./notifications.contract"
 
 import { authContracts } from "./auth.contract"
+import { notificationsContracts } from "./notifications.contract"
 import { streamsContracts } from "./streams.contract"
 import { webhooksContracts } from "./webhooks.contract"
 
@@ -30,4 +32,5 @@ export const allContracts: Contract[] = [
   ...streamsContracts,
   ...authContracts,
   ...webhooksContracts,
+  ...notificationsContracts,
 ]

tests/contracts/src/notifications.contract.ts

@@ -0,0 +1,27 @@
+import { type Contract } from "./contract"
+import { notificationsPageSchema } from "./schemas"
+
+/**
+ * Contract coverage for `api/src/notifications/notifications.controller.ts`
+ * (issue #534). The provider suite seeds one unread notification for the
+ * fixture user in beforeAll, so this contract validates the non-empty
+ * shape — including the numeric ids and ISO-string timestamps.
+ */
+export const notificationsContracts: Contract[] = [
+  {
+    name: "list-notifications",
+    description: "GET /notifications returns the caller's unread notifications in the paginated envelope",
+    consumer: "xstreamroll-sdk",
+    provider: "api",
+    request: {
+      method: "GET",
+      path: "/notifications",
+      query: { page: 1, limit: 20 },
+      authenticated: true,
+    },
+    response: {
+      status: 200,
+      schema: notificationsPageSchema,
+    },
+  },
+]

tests/contracts/src/schemas.ts

@@ -4,6 +4,7 @@ import type {
   ApiErrorResponse,
   PaginatedResponse,
   Stream,
+  StreamEventRecord,
   Tag,
   User,
 } from "@xstreamroll/types"
@@ -85,6 +86,95 @@ export const authResponseSchema = z.object({
   refreshToken: z.string(),
 })
 
+/** The closed set of stream lifecycle/data event types. */
+export const streamEventTypeSchema = z.enum([
+  "stream:started",
+  "stream:stopped",
+  "stream:error",
+  "viewer:joined",
+  "viewer:left",
+  "data",
+])
+
+/**
+ * A single persisted stream event, as returned by `GET /streams/:id/events`
+ * (issue #396). The `id`/`streamId` are strings on the wire — the same
+ * stringification that already bit `Stream` and `User` — and the schema
+ * pins that choice so a regression fails the provider suite.
+ */
+export const streamEventRecordSchema = typed<StreamEventRecord>()(
+  z.object({
+    id: z.string(),
+    streamId: z.string(),
+    eventType: streamEventTypeSchema,
+    payload: z.record(z.string(), z.unknown()),
+    occurredAt: z.string(),
+  }),
+)
+
+export const paginatedStreamEventsSchema = z.object({
+  data: z.array(streamEventRecordSchema),
+  page: z.number(),
+  limit: z.number(),
+  total: z.number(),
+  hasMore: z.boolean(),
+})
+
+/**
+ * Aggregate analytics returned by `GET /streams/:id/analytics`.
+ * Mirrors `StreamAnalyticsDto` in the API — no shared
+ * `@xstreamroll/types` interface exists yet, so the schema is written
+ * by hand and pinned by the consumer test's type assertion.
+ */
+export const streamAnalyticsSchema = z.object({
+  streamId: z.number(),
+  totalEventsProcessed: z.object({
+    last24h: z.number(),
+    last7d: z.number(),
+    last30d: z.number(),
+  }),
+  errorRate: z.object({
+    window: z.literal("30d"),
+    totalEvents: z.number(),
+    errorEvents: z.number(),
+    percentage: z.number(),
+  }),
+  processingLatency: z.object({
+    window: z.literal("30d"),
+    averageMs: z.number().nullable(),
+    p99Ms: z.number().nullable(),
+  }),
+  eventsPerMinute: z.array(
+    z.object({
+      minute: z.string(),
+      count: z.number(),
+    }),
+  ),
+  generatedAt: z.string(),
+})
+
+/**
+ * A single unread notification, as returned by `GET /notifications`.
+ * Numeric ids and ISO-string timestamps on the wire.
+ */
+export const notificationSchema = z.object({
+  id: z.number(),
+  userId: z.number(),
+  type: z.string(),
+  payload: z.record(z.string(), z.unknown()),
+  readAt: z.string().nullable(),
+  createdAt: z.string(),
+  expiresAt: z.string(),
+})
+
+export const notificationsPageSchema = z.object({
+  data: z.array(notificationSchema),
+  page: z.number(),
+  limit: z.number(),
+  total: z.number(),
+  unreadCount: z.number(),
+})
+
 /**
  * Shape returned by `POST /streams/events` (issue #514) and, per row,
  * by `GET /streams/pending` — the `stream_data` wire shape the worker
@@ -98,14 +188,18 @@ export const pendingStreamEventSchema = z.object({
 
 /**
  * A webhook subscription as returned by `POST /webhooks` — the only
- * response that includes the signing `secret`.
+ * response that includes the signing `secret`. Field types mirror the
+ * SDK's `WebhookSubscription` exactly (ids accept string or number on
+ * the wire; `events` is pinned to the closed {@link streamEventTypeSchema}
+ * union the SDK's `StreamEventType` declares) so a server-side type
+ * change fails CI (issue #534).
  */
 export const webhookSubscriptionSchema = z.object({
   id: z.union([z.string(), z.number()]),
   userId: z.union([z.string(), z.number()]),
   streamId: z.union([z.string(), z.number()]),
   url: z.string(),
-  events: z.array(z.string()),
+  events: z.array(streamEventTypeSchema),
   secret: z.string(),
   active: z.boolean(),
   createdAt: z.string(),
@@ -127,11 +221,16 @@ export const paginatedWebhookSubscriptionsSchema = z.object({
   limit: z.number(),
 })
 
-/** A single webhook delivery, as returned by the deliveries endpoints. */
+/**
+ * A single webhook delivery, as returned by the deliveries endpoints.
+ * Field types mirror the SDK's `WebhookDelivery` exactly (issue #534),
+ * including the `id`/`webhookSubscriptionId` string-vs-number choice and
+ * the closed `event` union.
+ */
 export const webhookDeliverySchema = z.object({
   id: z.union([z.string(), z.number()]),
   webhookSubscriptionId: z.union([z.string(), z.number()]),
-  event: z.string(),
+  event: streamEventTypeSchema,
   payload: z.record(z.string(), z.unknown()),
   status: z.enum(["pending", "success", "failed"]),
   attemptCount: z.number(),
@@ -143,6 +242,13 @@ export const webhookDeliverySchema = z.object({
   createdAt: z.string(),
 })
 
+export const paginatedWebhookDeliveriesSchema = z.object({
+  data: z.array(webhookDeliverySchema),
+  total: z.number(),
+  page: z.number(),
+  limit: z.number(),
+})
+
 export const apiErrorSchema = typed<ApiErrorResponse>()(
   z.object({
     statusCode: z.number(),

tests/contracts/src/streams.contract.ts

@@ -1,8 +1,10 @@
 import { PLACEHOLDER, type Contract } from "./contract"
 import {
   apiErrorSchema,
+  paginatedStreamEventsSchema,
   paginatedStreamsSchema,
   pendingStreamEventSchema,
+  streamAnalyticsSchema,
   streamSchema,
 } from "./schemas"
 
@@ -109,42 +111,39 @@ export const streamsContracts: Contract[] = [
     },
   },
   {
-    // Runs after the seed stream exists (created in the provider suite's
-    // beforeAll). `q=seed` matches the seed stream's name
-    // case-insensitively; the response is the standard paginated
-    // envelope over the filtered set (issue #532).
-    name: "list-streams-search",
-    description: "GET /streams?q=… returns only streams matching the search, in the paginated envelope",
+    // The provider suite records one event on the seeded stream in
+    // beforeAll, so this contract validates the non-empty shape — most
+    // importantly the stringified `id`/`streamId` fields (issue #534).
+    name: "list-stream-events",
+    description: "GET /streams/:id/events replays the stream's event log in the paginated envelope",
     consumer: "xstreamroll-sdk",
     provider: "api",
     request: {
       method: "GET",
-      path: "/streams",
-      query: { page: 1, limit: 20, q: "seed" },
+      path: "/streams/:id/events",
+      pathParams: { id: PLACEHOLDER.EXISTING_STREAM_ID },
+      query: { page: 1, limit: 50 },
       authenticated: true,
     },
     response: {
       status: 200,
-      schema: paginatedStreamsSchema,
+      schema: paginatedStreamEventsSchema,
     },
   },
   {
-    // `tag=live-streaming` matches the seed stream, which the provider
-    // suite tags in beforeAll. Unknown tags return an empty page, so the
-    // shape contract holds either way (issue #532).
-    name: "list-streams-by-tag",
-    description: "GET /streams?tag=… returns only streams carrying the tag, in the paginated envelope",
+    name: "get-stream-analytics",
+    description: "GET /streams/:id/analytics returns the aggregate analytics shape",
     consumer: "xstreamroll-sdk",
     provider: "api",
     request: {
       method: "GET",
-      path: "/streams",
-      query: { page: 1, limit: 20, tag: "live-streaming" },
+      path: "/streams/:id/analytics",
+      pathParams: { id: PLACEHOLDER.EXISTING_STREAM_ID },
       authenticated: true,
     },
     response: {
       status: 200,
-      schema: paginatedStreamsSchema,
+      schema: streamAnalyticsSchema,
     },
   },
   {