feat: add per-user rate limiting and concurrent job limits

Redis sliding window (sorted set) enforces per-user request rate limits
via the rateLimitPerUser DB setting. Concurrent job limits checked at
the HTTP layer before enqueue via maxConcurrentJobsPerUser setting.
Both default to 0 (unlimited).

apps/api/src/index.ts

@@ -317,6 +317,7 @@ await app.register(cookie, {
 
 // IP allowlist (enterprise -- must run before auth to reject early)
 import { registerIpAllowlist } from "./plugins/ip-allowlist.js";
+import { registerPerUserRateLimit } from "./plugins/per-user-rate-limit.js";
 
 await registerIpAllowlist(app);
 
@@ -326,6 +327,9 @@ await configRoutes(app);
 // Auth middleware (must be registered before routes it protects)
 await authMiddleware(app);
 
+// Per-user rate limiting (after auth so request.user is populated)
+await registerPerUserRateLimit(app);
+
 // Auth routes
 await authRoutes(app);
 

apps/api/src/plugins/per-user-rate-limit.ts

@@ -0,0 +1,57 @@
+/**
+ * Per-user rate limiting using Redis sliding window (sorted sets).
+ *
+ * Runs AFTER auth middleware so `request.user` is populated.
+ * Only applies to authenticated users on /api/ routes.
+ * The limit is controlled by the `rateLimitPerUser` DB setting (0 = unlimited).
+ */
+import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
+import { sharedRedis } from "../jobs/connection.js";
+import { getSettingNumber } from "../lib/settings-helpers.js";
+import { getAuthUser } from "./auth.js";
+
+const WINDOW_MS = 60_000; // 1-minute sliding window
+
+export async function registerPerUserRateLimit(app: FastifyInstance): Promise<void> {
+  app.addHook("preHandler", async (request: FastifyRequest, reply: FastifyReply) => {
+    const user = getAuthUser(request);
+    if (!user) return; // Anonymous/public requests skip per-user limits
+
+    // Only rate-limit API routes
+    if (!request.url.startsWith("/api/")) return;
+
+    const rateLimitPerUser = await getSettingNumber("rateLimitPerUser", 0);
+    if (rateLimitPerUser <= 0) return; // 0 = unlimited
+
+    const redis = sharedRedis();
+    const key = `ratelimit:user:${user.id}`;
+    const now = Date.now();
+
+    // Sliding window using Redis sorted set:
+    // 1. Remove entries older than the window
+    // 2. Add current request with timestamp as score
+    // 3. Count entries in the window
+    // 4. Set TTL slightly longer than window for cleanup
+    const multi = redis.multi();
+    multi.zremrangebyscore(key, 0, now - WINDOW_MS);
+    multi.zadd(key, now, `${now}:${Math.random()}`);
+    multi.zcard(key);
+    multi.expire(key, 61);
+    const results = await multi.exec();
+
+    // multi.exec() returns [[err, result], ...] for each command
+    const requestCount = (results?.[2]?.[1] as number) ?? 0;
+
+    // Set standard rate limit headers
+    reply.header("X-RateLimit-Limit", rateLimitPerUser);
+    reply.header("X-RateLimit-Remaining", Math.max(0, rateLimitPerUser - requestCount));
+    reply.header("X-RateLimit-Reset", Math.ceil((now + WINDOW_MS) / 1000));
+
+    if (requestCount > rateLimitPerUser) {
+      return reply.status(429).send({
+        error: "Rate limit exceeded",
+        retryAfter: Math.ceil(WINDOW_MS / 1000),
+      });
+    }
+  });
+}

apps/api/src/routes/tool-factory.ts

@@ -3,15 +3,18 @@ import { mkdir, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { extname, join } from "node:path";
 import { ANALYTICS_EVENTS, getBundleForTool, TOOL_BUNDLE_MAP, TOOLS } from "@snapotter/shared";
+import { and, inArray, sql } from "drizzle-orm";
 import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
 import type { z } from "zod";
 import { env } from "../config.js";
+import { db, schema } from "../db/index.js";
 import { enqueueToolJob, waitForJob } from "../jobs/enqueue.js";
 import { trackEvent } from "../lib/analytics.js";
 import { formatZodErrors, stripInternalPaths } from "../lib/errors.js";
 import { isToolInstalled } from "../lib/feature-status.js";
 import { getObjectBuffer, putObject } from "../lib/object-storage.js";
 import { resolveToolPool, shouldSkipSyncWindow } from "../lib/pool.js";
+import { getSettingNumber } from "../lib/settings-helpers.js";
 import { type ReceivedUpload, receiveUpload } from "../lib/upload-stream.js";
 import { InputValidationError } from "../modality/contract.js";
 import { inputHandlerFor } from "../modality/input-handler.js";
@@ -441,6 +444,29 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
           });
         }
 
+        // Check per-user concurrent job limit before enqueuing
+        const userId = getAuthUser(request)?.id ?? null;
+        const maxConcurrent = await getSettingNumber("maxConcurrentJobsPerUser", 0);
+        if (maxConcurrent > 0 && userId) {
+          const activeJobs = await db
+            .select({ count: sql<number>`count(*)::int` })
+            .from(schema.jobs)
+            .where(
+              and(
+                sql`${schema.jobs.userId} = ${userId}`,
+                inArray(schema.jobs.status, ["queued", "processing"]),
+              ),
+            );
+
+          if (activeJobs[0].count >= maxConcurrent) {
+            return reply.status(429).send({
+              error: "Too many concurrent jobs. Please wait for existing jobs to complete.",
+              activeJobs: activeJobs[0].count,
+              limit: maxConcurrent,
+            });
+          }
+        }
+
         const startTime = Date.now();
         const pool = resolveToolPool(config.toolId);
 
@@ -451,7 +477,7 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
         await enqueueToolJob({
           jobId,
           toolId: config.toolId,
-          userId: getAuthUser(request)?.id ?? null,
+          userId,
           pool,
           inputRefs,
           filename,