feat(api): add POST /streams/events ingest endpoint and SDK publishEvent

api/src/contract-provider.spec.ts

@@ -38,6 +38,7 @@ import { StreamOwnershipGuard } from "./common/guards/stream-ownership.guard"
 import { StreamOwnershipService } from "./common/guards/stream-ownership.service"
 import createJwtConfig, { createRefreshJwtConfig } from "./config/jwt.config"
 import { StreamsRepository } from "./streams/repository/streams.repository"
+import { StreamApiKeyGuard } from "./streams/stream-api-key.guard"
 import { StreamsController } from "./streams/streams.controller"
 import { StreamsService } from "./streams/streams.service"
 import { TagsRepository } from "./tags/repository/tags.repository"
@@ -46,6 +47,7 @@ import { TagsService } from "./tags/tags.service"
 import { WebhooksService } from "./webhooks/webhooks.service"
 
 process.env.JWT_SECRET ??= "test-secret"
+process.env.STREAM_API_KEY ??= "test-stream-key"
 
 /** In-memory double for the Postgres-backed UsersRepository. */
 class InMemoryUsersRepository {
@@ -124,6 +126,7 @@ describe("Contract provider verification (api)", () => {
         AuthGuard,
         JwtExtractorService,
         StreamOwnershipGuard,
+        StreamApiKeyGuard,
         { provide: StreamOwnershipService, useValue: streamOwnershipService },
         { provide: TokenDenylistService, useValue: tokenDenylistService },
         AuthService,
@@ -212,8 +215,21 @@ describe("Contract provider verification (api)", () => {
     if (contract.request.authenticated) {
       req = req.set("Authorization", `Bearer ${accessToken}`)
     }
+    if (contract.request.apiKey) {
+      req = req.set("X-Stream-Api-Key", process.env.STREAM_API_KEY ?? "")
+    }
     if (contract.request.body !== undefined) {
-      req = req.send(contract.request.body as object)
+      // Substitute stream-id placeholders inside the request body (e.g. the
+      // `streamId` field of the ingest contract) the same way path params are.
+      const body = JSON.parse(JSON.stringify(contract.request.body)) as Record<
+        string,
+        unknown
+      >
+      for (const [k, v] of Object.entries(body)) {
+        if (v === PLACEHOLDER.EXISTING_STREAM_ID) body[k] = existingStreamId
+        if (v === PLACEHOLDER.MISSING_STREAM_ID) body[k] = "999999"
+      }
+      req = req.send(body)
     }
     return req
   }

api/src/database.integration.spec.ts

@@ -6,6 +6,7 @@ import {
   destroyTestApp,
   TestAppContext,
 } from "./database/test-utils"
+import { StreamsDbRepository } from "./streams/repository/streams-db.repository"
 
 describe("Database Integration Tests", () => {
   let ctx: TestAppContext
@@ -338,6 +339,49 @@ describe("Database Integration Tests", () => {
     })
   })
 
+  describe("Stream data ingestion (issue #514)", () => {
+    let streamsDb: StreamsDbRepository
+    let streamId: number
+
+    beforeEach(async () => {
+      streamsDb = new StreamsDbRepository(pool)
+      const user = await pool.query(
+        `INSERT INTO users (username, email, password_hash)
+         VALUES ('ingester', 'ingest@test.com', 'hash')
+         RETURNING id`,
+      )
+      const stream = await pool.query(
+        `INSERT INTO streams (user_id, name) VALUES ($1, 'Ingest Stream')
+         RETURNING id`,
+        [user.rows[0].id],
+      )
+      streamId = stream.rows[0].id
+    })
+
+    it("insertPendingEvent writes a row to stream_data that getPendingEvents returns", async () => {
+      const pending = await streamsDb.insertPendingEvent(
+        streamId,
+        { viewerId: "u1", kind: "data" },
+        new Date("2026-08-01T00:00:00Z"),
+      )
+
+      expect(pending.streamId).toBe(String(streamId))
+      expect(pending.data).toEqual({ viewerId: "u1", kind: "data" })
+      expect(pending.timestamp).toBe("2026-08-01T00:00:00.000Z")
+
+      // The row is visible to the worker's poll source.
+      const { data } = await streamsDb.getPendingEvents(100, 0)
+      expect(data).toHaveLength(1)
+      expect(data[0]).toEqual(pending)
+    })
+
+    it("insertPendingEvent throws NotFoundException for a nonexistent stream", async () => {
+      await expect(
+        streamsDb.insertPendingEvent(999999, { foo: "bar" }, new Date()),
+      ).rejects.toThrow(/not found/)
+    })
+  })
+
   describe("Indexes", () => {
     it("has an index on streams.user_id", async () => {
       const result = await pool.query(

api/src/streams/dto/ingest-stream-event.dto.ts

@@ -0,0 +1,29 @@
+import { ApiProperty } from "@nestjs/swagger"
+import { IsOptional, IsString, Matches } from "class-validator"
+
+/**
+ * Payload accepted by `POST /streams/events` (issue #514).
+ *
+ * The server stamps the arrival timestamp itself — trusting client
+ * clocks for latency metrics is a correctness risk, so the wire shape
+ * deliberately has no timestamp field.
+ */
+export class IngestStreamEventDto {
+  @ApiProperty({
+    description: "Numeric stream id, as a string (matches the StreamEvent wire shape)",
+    example: "42",
+  })
+  @IsString()
+  @Matches(/^\d+$/, { message: "streamId must be a numeric string" })
+  streamId!: string
+
+  @ApiProperty({
+    description: "Free-form event payload, persisted as JSONB in stream_data",
+    example: { viewerId: "user_42" },
+  })
+  // Kept as a decorated (thus whitelist-preserved) optional field; the
+  // service enforces presence + plain-object shape so the DTO stays
+  // dependency-light (the installed class-validator d.ts lacks object decorators).
+  @IsOptional()
+  data?: Record<string, unknown>
+}

api/src/streams/repository/streams-db.repository.ts

@@ -3,12 +3,13 @@ import {
   Injectable,
   InternalServerErrorException,
   Logger,
+  NotFoundException,
   ServiceUnavailableException,
 } from "@nestjs/common"
 import { Pool } from "pg"
+
 import { PG_POOL } from "../../database/database.module"
 import { StreamAnalyticsDto } from "../dto/stream-analytics.dto"
-import type { StreamVisibility } from "../dto/visibility"
 import { Stream } from "../stream.entity"
 import {
   StreamsRepository,
@@ -18,6 +19,8 @@ import {
   type StreamUpdateChanges,
 } from "./streams.repository"
 
+import type { StreamVisibility } from "../dto/visibility"
+
 /**
  * PostgreSQL-backed streams repository.
  *
@@ -227,6 +230,41 @@ export class StreamsDbRepository {
     }
   }
 
+  /**
+   * Append an ingested event to `stream_data` — the worker's poll source
+   * (issue #514). The server stamps the timestamp; a nonexistent stream
+   * id surfaces as a foreign-key violation, mapped to 404.
+   */
+  async insertPendingEvent(
+    streamId: number,
+    data: Record<string, unknown>,
+    timestamp: Date,
+  ): Promise<PendingStreamEvent> {
+    try {
+      const { rows } = await this.pool.query<{
+        stream_id: number
+        data: Record<string, unknown>
+        timestamp: Date
+      }>(
+        `INSERT INTO stream_data (stream_id, data, timestamp)
+         VALUES ($1, $2, $3)
+         RETURNING stream_id, data, timestamp`,
+        [streamId, data, timestamp],
+      )
+      return {
+        streamId: String(rows[0].stream_id),
+        data: rows[0].data,
+        timestamp: rows[0].timestamp.toISOString(),
+      }
+    } catch (err) {
+      // 23503 = foreign_key_violation — the stream does not exist.
+      if ((err as { code?: string }).code === "23503") {
+        throw new NotFoundException(`stream ${streamId} not found`)
+      }
+      this.handleDbError(err, "insertPendingEvent")
+    }
+  }
+
   /**
    * Returns a paginated slice of unprocessed stream-data rows ordered by
    * insertion time (oldest first) so the worker processes events in FIFO order.

api/src/streams/repository/streams.repository.ts

@@ -1,9 +1,11 @@
-import { Injectable } from "@nestjs/common"
-import type { StreamEventRecord } from "@xstreamroll/types"
+import { Injectable, NotFoundException } from "@nestjs/common"
+
 import { StreamAnalyticsDto } from "../dto/stream-analytics.dto"
-import type { StreamVisibility } from "../dto/visibility"
 import { Stream } from "../stream.entity"
 
+import type { StreamVisibility } from "../dto/visibility"
+import type { StreamEventRecord } from "@xstreamroll/types"
+
 export interface StreamCreateParams {
   userId: number
   name: string
@@ -71,6 +73,8 @@ export class StreamsRepository {
   /** Per-stream append-only event log, mirroring the `stream_events` table. */
   private readonly eventsByStream = new Map<number, StreamEventRecord[]>()
   private nextEventId = 1
+  /** Pending (unprocessed) events, mirroring the `stream_data` table (issue #514). */
+  private readonly pendingEvents: PendingStreamEvent[] = []
 
   async findById(id: number): Promise<Stream | undefined> {
     return this.streamsById.get(id)
@@ -162,15 +166,41 @@ export class StreamsRepository {
   }
 
   /**
-   * Returns a paginated slice of pending (unprocessed) stream events.
-   * In-memory stub: always returns an empty array because the in-memory
-   * repository has no stream_data store. Used only in unit tests.
+   * Append an ingested event to the pending queue (issue #514). Mirrors
+   * the `INSERT INTO stream_data …` query a Postgres-backed repository
+   * runs; the server stamps the timestamp.
+   */
+  async insertPendingEvent(
+    streamId: number,
+    data: Record<string, unknown>,
+    timestamp: Date,
+  ): Promise<PendingStreamEvent> {
+    if (!this.streamsById.has(streamId)) {
+      throw new NotFoundException(`stream ${streamId} not found`)
+    }
+    const event: PendingStreamEvent = {
+      streamId: String(streamId),
+      data,
+      timestamp: timestamp.toISOString(),
+    }
+    this.pendingEvents.push(event)
+    return event
+  }
+
+  /**
+   * Returns a paginated slice of pending (unprocessed) stream events in
+   * FIFO order. Backed by the in-memory pending queue so tests and local
+   * development can exercise the full ingest → poll flow.
    */
   async getPendingEvents(
-    _limit: number,
-    _offset: number,
+    limit: number,
+    offset: number,
   ): Promise<{ data: PendingStreamEvent[]; nextCursor: number | null }> {
-    return { data: [], nextCursor: null }
+    const data = this.pendingEvents.slice(offset, offset + limit)
+    return {
+      data,
+      nextCursor: data.length < limit ? null : offset + data.length,
+    }
   }
 
   async getAnalytics(streamId: number): Promise<StreamAnalyticsDto> {

api/src/streams/stream-api-key.guard.spec.ts

@@ -0,0 +1,64 @@
+import { ExecutionContext, UnauthorizedException } from "@nestjs/common"
+
+import { StreamApiKeyGuard } from "./stream-api-key.guard"
+
+const VALID_KEY = "sk-test-123"
+
+function contextWithHeader(
+  value: string | string[] | undefined,
+): ExecutionContext {
+  const req = { headers: { "x-stream-api-key": value } }
+  return {
+    switchToHttp: () => ({
+      getRequest: () => req,
+    }),
+  } as unknown as ExecutionContext
+}
+
+describe("StreamApiKeyGuard (issue #514)", () => {
+  let guard: StreamApiKeyGuard
+
+  beforeEach(() => {
+    process.env.STREAM_API_KEY = VALID_KEY
+    guard = new StreamApiKeyGuard()
+  })
+
+  afterEach(() => {
+    delete process.env.STREAM_API_KEY
+  })
+
+  it("allows a request with the correct API key", () => {
+    expect(guard.canActivate(contextWithHeader(VALID_KEY))).toBe(true)
+  })
+
+  it("rejects a request with a missing header", () => {
+    expect(() => guard.canActivate(contextWithHeader(undefined))).toThrow(
+      UnauthorizedException,
+    )
+    expect(() => guard.canActivate(contextWithHeader(undefined))).toThrow(
+      "missing stream API key",
+    )
+  })
+
+  it("rejects a request with an empty key", () => {
+    expect(() => guard.canActivate(contextWithHeader(""))).toThrow(
+      UnauthorizedException,
+    )
+  })
+
+  it("rejects a request with the wrong key (fixed-length compare)", () => {
+    expect(() =>
+      guard.canActivate(contextWithHeader("sk-test-999")),
+    ).toThrow(UnauthorizedException)
+    expect(() =>
+      guard.canActivate(contextWithHeader("a-short-key")),
+    ).toThrow(UnauthorizedException)
+  })
+
+  it("rejects every request when STREAM_API_KEY is not configured", () => {
+    delete process.env.STREAM_API_KEY
+    expect(() => guard.canActivate(contextWithHeader(VALID_KEY))).toThrow(
+      "stream API key is not configured",
+    )
+  })
+})

api/src/streams/stream-api-key.guard.ts

@@ -0,0 +1,46 @@
+import { createHash, timingSafeEqual } from "crypto"
+
+import {
+  CanActivate,
+  ExecutionContext,
+  Injectable,
+  UnauthorizedException,
+} from "@nestjs/common"
+
+import type { Request } from "express"
+
+/**
+ * Authenticates event-ingestion requests (`POST /streams/events`) via
+ * the `X-Stream-Api-Key` header (issue #514).
+ *
+ * The expected key is read from the same `STREAM_API_KEY` env var that
+ * `config/env.ts` validates. Comparison is constant-time (SHA-256 both
+ * sides first so `timingSafeEqual` gets equal-length buffers). When the
+ * env var is unset — which `validateEnv()` prevents in practice — every
+ * request is rejected, which is the safe default for a public endpoint.
+ */
+@Injectable()
+export class StreamApiKeyGuard implements CanActivate {
+  canActivate(context: ExecutionContext): boolean {
+    const request = context.switchToHttp().getRequest<Request>()
+    const provided = request.headers["x-stream-api-key"]
+    if (typeof provided !== "string" || provided.length === 0) {
+      throw new UnauthorizedException("missing stream API key")
+    }
+
+    const expected = process.env.STREAM_API_KEY ?? ""
+    if (expected.length === 0) {
+      throw new UnauthorizedException("stream API key is not configured")
+    }
+
+    const providedHash = createHash("sha256").update(provided).digest()
+    const expectedHash = createHash("sha256").update(expected).digest()
+    if (
+      providedHash.length !== expectedHash.length ||
+      !timingSafeEqual(providedHash, expectedHash)
+    ) {
+      throw new UnauthorizedException("invalid stream API key")
+    }
+    return true
+  }
+}