feat: admin role guards, SSE connection cap, wallet-crypto tests

apps/api/src/modules/auth/tests/auth.middleware.test.ts

@@ -1,10 +1,11 @@
 import express from "express"
 import jwt from "jsonwebtoken"
 import request from "supertest"
-import { describe, expect, it } from "vitest"
+import { beforeEach, describe, expect, it } from "vitest"
 import { env } from "../../../shared/config/env.js"
 import { errorHandler } from "../../../shared/middleware/error-handler.js"
-import { requireAuth } from "../../../shared/middleware/auth.middleware.js"
+import { requireAuth, requireRole } from "../../../shared/middleware/auth.middleware.js"
+import { prisma } from "../../../shared/database/prisma.js"
 
 function buildTestApp() {
   const app = express()
@@ -51,3 +52,69 @@ describe("requireAuth middleware", () => {
     expect(response.body.error.code).toBe("UNAUTHORIZED")
   })
 })
+
+describe("requireRole middleware", () => {
+  beforeEach(async () => {
+    await prisma.user.deleteMany()
+  })
+
+  async function createUser(email: string, role: string): Promise<string> {
+    const { id } = await prisma.user.create({
+      data: {
+        email,
+        passwordHash: "not-used-in-this-test",
+        role,
+      },
+    })
+    return id
+  }
+
+  it("allows a caller whose role is in the allowlist", async () => {
+    const userId = await createUser("admin@test.com", "admin")
+    const app = express()
+    app.get("/admin", requireAuth, requireRole(["admin"]), (_req, res) => {
+      res.status(200).json({ ok: true })
+    })
+    app.use(errorHandler)
+
+    const token = jwt.sign({ sub: userId }, env.JWT_SECRET)
+    const response = await request(app)
+      .get("/admin")
+      .set("Authorization", `Bearer ${token}`)
+
+    expect(response.status).toBe(200)
+  })
+
+  it("rejects a caller whose role is not in the allowlist", async () => {
+    const userId = await createUser("fan@test.com", "user")
+    const app = express()
+    app.get("/admin", requireAuth, requireRole(["admin"]), (_req, res) => {
+      res.status(200).json({ ok: true })
+    })
+    app.use(errorHandler)
+
+    const token = jwt.sign({ sub: userId }, env.JWT_SECRET)
+    const response = await request(app)
+      .get("/admin")
+      .set("Authorization", `Bearer ${token}`)
+
+    expect(response.status).toBe(403)
+    expect(response.body.error.code).toBe("FORBIDDEN")
+  })
+
+  it("rejects a caller whose account no longer exists", async () => {
+    const app = express()
+    app.get("/admin", requireAuth, requireRole(["admin"]), (_req, res) => {
+      res.status(200).json({ ok: true })
+    })
+    app.use(errorHandler)
+
+    const token = jwt.sign({ sub: "does-not-exist" }, env.JWT_SECRET)
+    const response = await request(app)
+      .get("/admin")
+      .set("Authorization", `Bearer ${token}`)
+
+    expect(response.status).toBe(403)
+    expect(response.body.error.code).toBe("FORBIDDEN")
+  })
+})

apps/api/src/modules/reconciliation/middleware/reconciliation-rate-limit.ts

@@ -0,0 +1,30 @@
+import type { NextFunction, Request, Response } from "express"
+import { createRateLimiter } from "../../../shared/rate-limit/rate-limiter.js"
+import { env } from "../../../shared/config/env.js"
+
+// Manual reconciliation runs are an operator/ops action — cheap to trigger,
+// expensive to execute. This limiter bounds how often a single admin can fire
+// one. Uses the in-memory limiter: reconciliation is a single-instance,
+// rollup-style task and the cap here is about preventing accidental
+// over-triggering, not defending a high-traffic public endpoint.
+const reconciliationRunLimiter = createRateLimiter({
+  windowMs: env.RECONCILIATION_RUN_RATE_LIMIT_WINDOW_MS,
+  max: env.RECONCILIATION_RUN_RATE_LIMIT_MAX,
+})
+
+export function reconciliationRunRateLimit(
+  req: Request,
+  res: Response,
+  next: NextFunction
+): void {
+  if (!reconciliationRunLimiter.consume(`user:${req.userId ?? "unknown"}`)) {
+    res.status(429).json({
+      error: {
+        code: "RECONCILIATION_RUN_RATE_LIMITED",
+        message: "Too many reconciliation runs. Try again shortly.",
+      },
+    })
+    return
+  }
+  next()
+}

apps/api/src/modules/reconciliation/routes/reconciliation.routes.ts

@@ -1,9 +1,18 @@
 import { Router } from "express"
-import { requireAuth } from "../../../shared/middleware/auth.middleware.js"
+import { requireAuth, requireRole } from "../../../shared/middleware/auth.middleware.js"
 import { reconciliationController } from "../controllers/reconciliation.controller.js"
+import { reconciliationRunRateLimit } from "../middleware/reconciliation-rate-limit.js"
 
 export const reconciliationRouter: Router = Router()
 
 // Manual trigger, mainly for demos/ops; the scheduled loop in server.ts
 // calls reconciliationService.run() on the same interval automatically.
-reconciliationRouter.post("/run", requireAuth, reconciliationController.run)
+// Operator-only (admin) by design, and rate-limited per admin so a stray
+// script can't hammer actual reconciliation batches.
+reconciliationRouter.post(
+  "/run",
+  requireAuth,
+  requireRole(["admin"]),
+  reconciliationRunRateLimit,
+  reconciliationController.run
+)

apps/api/src/modules/reconciliation/tests/reconciliation.routes.test.ts

@@ -10,7 +10,13 @@ async function registerAndLogin(email: string) {
   const res = await request(app)
     .post("/api/auth/login")
     .send({ email, password: "Password1!" })
-  return { token: res.body.token as string }
+  return { token: res.body.token as string, userId: res.body.user.id as string }
+}
+
+async function registerAndLoginAdmin(email: string) {
+  const { token, userId } = await registerAndLogin(email)
+  await prisma.user.update({ where: { id: userId }, data: { role: "admin" } })
+  return { token, userId }
 }
 
 beforeEach(async () => {
@@ -24,9 +30,20 @@ describe("POST /api/reconciliation/run", () => {
     expect(res.status).toBe(401)
   })
 
-  it("runs reconciliation and returns a summary", async () => {
+  it("rejects non-admin (regular user) requests with 403", async () => {
     const { token } = await registerAndLogin("recon-route-fan@test.com")
 
+    const res = await request(app)
+      .post("/api/reconciliation/run")
+      .set("Authorization", `Bearer ${token}`)
+
+    expect(res.status).toBe(403)
+    expect(res.body.error.code).toBe("FORBIDDEN")
+  })
+
+  it("runs reconciliation and returns a summary when an admin triggers it", async () => {
+    const { token } = await registerAndLoginAdmin("recon-route-admin@test.com")
+
     const res = await request(app)
       .post("/api/reconciliation/run")
       .set("Authorization", `Bearer ${token}`)

apps/api/src/modules/streams/controllers/stream.controller.ts

@@ -1,4 +1,5 @@
 import type { NextFunction, Request, Response } from "express"
+import { env } from "../../../shared/config/env.js"
 import { AppError } from "../../../shared/errors/app-error.js"
 import { tipEventBus, type TipFeedEvent } from "../../../shared/realtime/tip-event-bus.js"
 import { tipPayoutRepository } from "../../tips/repositories/tip-payout.repository.js"
@@ -11,6 +12,37 @@ import { toStreamResponse } from "../types/stream.types.js"
 
 const HEARTBEAT_INTERVAL_MS = 20_000
 
+// Per-instance SSE connection accounting. Each open tip-stream connection
+// holds a socket, an event-bus subscription and a heartbeat timer, so we cap
+// them per instance and per user to keep an abusive client (or a flood of
+// legitimate ones) from exhausting server resources. Backpressure comes from
+// the limit itself: once a cap is hit, excess clients are rejected fast
+// instead of queuing unbounded state server-side.
+const sseConnectionsByUser = new Map<string, number>()
+let sseConnectionCount = 0
+
+function sseCanOpen(userId: string): boolean {
+  if (sseConnectionCount >= env.SSE_STREAM_MAX_CONNECTIONS) return false
+  const perUser = sseConnectionsByUser.get(userId) ?? 0
+  if (perUser >= env.SSE_STREAM_MAX_PER_USER) return false
+  return true
+}
+
+function sseOpen(userId: string): void {
+  sseConnectionCount += 1
+  sseConnectionsByUser.set(userId, (sseConnectionsByUser.get(userId) ?? 0) + 1)
+}
+
+function sseClose(userId: string): void {
+  sseConnectionCount = Math.max(0, sseConnectionCount - 1)
+  const perUser = (sseConnectionsByUser.get(userId) ?? 1) - 1
+  if (perUser <= 0) {
+    sseConnectionsByUser.delete(userId)
+  } else {
+    sseConnectionsByUser.set(userId, perUser)
+  }
+}
+
 type TipFeedPayload = Omit<TipFeedEvent, "seq" | "emittedAt">
 
 export function buildTipFeedPayloads(
@@ -84,40 +116,62 @@ export const streamController = {
         throw new AppError(404, "STREAM_NOT_FOUND", "Stream not found")
       }
 
-      res.writeHead(200, {
-        "Content-Type": "text/event-stream",
-        "Cache-Control": "no-cache",
-        Connection: "keep-alive",
-      })
-      // writeHead() only queues the headers; without an explicit flush they
-      // sit unsent until the first body write, which can be a long time
-      // away (empty backlog + only a periodic heartbeat). Force them out
-      // immediately so clients see the connection open right away.
-      res.flushHeaders()
-
-      // Late joiners get every tip's current state so they aren't blind to
-      // what already happened. Each tip appears once here (its latest
-      // status), so this can never race with / duplicate a live event for
-      // a tip that's still in flight.
-      const backlog = await tipRepository.findByStreamId(streamId!)
-      const payoutsByTipId = await loadBacklogPayouts(backlog.map((tip) => tip.id))
-      for (const payload of buildTipFeedPayloads(backlog, payoutsByTipId)) {
-        writeEvent(res, payload)
+      // Not counted until the stream is known to exist, so a flood of
+      // unknown stream ids can't be gamed into exhausting the cap.
+      if (!sseCanOpen(req.userId!)) {
+        res.status(503).json({
+          error: {
+            code: "SSE_LIMIT_REACHED",
+            message:
+              "Too many concurrent tip-stream connections. Reduce the number of open streams or try again shortly.",
+          },
+        })
+        return
+      }
+      sseOpen(req.userId!)
+      let acquired = true
+
+      try {
+        res.writeHead(200, {
+          "Content-Type": "text/event-stream",
+          "Cache-Control": "no-cache",
+          Connection: "keep-alive",
+        })
+        // writeHead() only queues the headers; without an explicit flush they
+        // sit unsent until the first body write, which can be a long time
+        // away (empty backlog + only a periodic heartbeat). Force them out
+        // immediately so clients see the connection open right away.
+        res.flushHeaders()
+
+        // Late joiners get every tip's current state so they aren't blind to
+        // what already happened. Each tip appears once here (its latest
+        // status), so this can never race with / duplicate a live event for
+        // a tip that's still in flight.
+        const backlog = await tipRepository.findByStreamId(streamId!)
+        const payoutsByTipId = await loadBacklogPayouts(backlog.map((tip) => tip.id))
+        for (const payload of buildTipFeedPayloads(backlog, payoutsByTipId)) {
+          writeEvent(res, payload)
+        }
+
+        const unsubscribe = tipEventBus.subscribe(streamId!, (event) => {
+          writeEvent(res, event, event.seq)
+        })
+
+        const heartbeat = setInterval(() => {
+          res.write(": heartbeat\n\n")
+        }, HEARTBEAT_INTERVAL_MS)
+
+        req.on("close", () => {
+          clearInterval(heartbeat)
+          unsubscribe()
+          sseClose(req.userId!)
+          res.end()
+        })
+        acquired = false
+      } catch (innerError) {
+        if (acquired) sseClose(req.userId!)
+        throw innerError
       }
-
-      const unsubscribe = tipEventBus.subscribe(streamId!, (event) => {
-        writeEvent(res, event, event.seq)
-      })
-
-      const heartbeat = setInterval(() => {
-        res.write(": heartbeat\n\n")
-      }, HEARTBEAT_INTERVAL_MS)
-
-      req.on("close", () => {
-        clearInterval(heartbeat)
-        unsubscribe()
-        res.end()
-      })
     } catch (error) {
       next(error)
     }

apps/api/src/modules/streams/tests/stream.tips-feed.test.ts

@@ -314,4 +314,35 @@ describe("GET /api/streams/:id/tips (SSE feed)", () => {
     expect(res.status).toBe(404)
     expect(res.body.error.code).toBe("STREAM_NOT_FOUND")
   })
+
+  it("rejects excess concurrent connections once the per-user SSE cap is hit", async () => {
+    const { token: hostToken, creator } = await createCreatorWithProfileAndWallet(
+      "sse-cap-creator@test.com",
+      "sse-cap-creator"
+    )
+
+    const streamRes = await request(app)
+      .post("/api/streams")
+      .set("Authorization", `Bearer ${hostToken}`)
+      .send({ title: "Capped" })
+    const streamId = streamRes.body.id as string
+    const path = `/api/streams/${streamId}/tips`
+
+    // Default per-user cap is 5 (SSE_STREAM_MAX_PER_USER); hold that many
+    // connections open so the next one is rejected.
+    const held: { req: http.ClientRequest }[] = []
+    for (let i = 0; i < 5; i++) {
+      const conn = await openSseConnection(path, hostToken)
+      held.push(conn)
+    }
+
+    const sixth = await request(app).get(path).set("Authorization", `Bearer ${hostToken}`)
+
+    expect(sixth.status).toBe(503)
+    expect(sixth.body.error.code).toBe("SSE_LIMIT_REACHED")
+
+    for (const conn of held) {
+      conn.req.destroy()
+    }
+  })
 })