mirror of
https://github.com/snapotter-hq/SnapOtter.git
synced 2026-08-03 07:46:42 +02:00
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).
This commit is contained in:
@@ -317,6 +317,7 @@ await app.register(cookie, {
|
|||||||
|
|
||||||
// IP allowlist (enterprise -- must run before auth to reject early)
|
// IP allowlist (enterprise -- must run before auth to reject early)
|
||||||
import { registerIpAllowlist } from "./plugins/ip-allowlist.js";
|
import { registerIpAllowlist } from "./plugins/ip-allowlist.js";
|
||||||
|
import { registerPerUserRateLimit } from "./plugins/per-user-rate-limit.js";
|
||||||
|
|
||||||
await registerIpAllowlist(app);
|
await registerIpAllowlist(app);
|
||||||
|
|
||||||
@@ -326,6 +327,9 @@ await configRoutes(app);
|
|||||||
// Auth middleware (must be registered before routes it protects)
|
// Auth middleware (must be registered before routes it protects)
|
||||||
await authMiddleware(app);
|
await authMiddleware(app);
|
||||||
|
|
||||||
|
// Per-user rate limiting (after auth so request.user is populated)
|
||||||
|
await registerPerUserRateLimit(app);
|
||||||
|
|
||||||
// Auth routes
|
// Auth routes
|
||||||
await authRoutes(app);
|
await authRoutes(app);
|
||||||
|
|
||||||
|
|||||||
@@ -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),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
@@ -3,15 +3,18 @@ import { mkdir, rm } from "node:fs/promises";
|
|||||||
import { tmpdir } from "node:os";
|
import { tmpdir } from "node:os";
|
||||||
import { extname, join } from "node:path";
|
import { extname, join } from "node:path";
|
||||||
import { ANALYTICS_EVENTS, getBundleForTool, TOOL_BUNDLE_MAP, TOOLS } from "@snapotter/shared";
|
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 { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
|
||||||
import type { z } from "zod";
|
import type { z } from "zod";
|
||||||
import { env } from "../config.js";
|
import { env } from "../config.js";
|
||||||
|
import { db, schema } from "../db/index.js";
|
||||||
import { enqueueToolJob, waitForJob } from "../jobs/enqueue.js";
|
import { enqueueToolJob, waitForJob } from "../jobs/enqueue.js";
|
||||||
import { trackEvent } from "../lib/analytics.js";
|
import { trackEvent } from "../lib/analytics.js";
|
||||||
import { formatZodErrors, stripInternalPaths } from "../lib/errors.js";
|
import { formatZodErrors, stripInternalPaths } from "../lib/errors.js";
|
||||||
import { isToolInstalled } from "../lib/feature-status.js";
|
import { isToolInstalled } from "../lib/feature-status.js";
|
||||||
import { getObjectBuffer, putObject } from "../lib/object-storage.js";
|
import { getObjectBuffer, putObject } from "../lib/object-storage.js";
|
||||||
import { resolveToolPool, shouldSkipSyncWindow } from "../lib/pool.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 { type ReceivedUpload, receiveUpload } from "../lib/upload-stream.js";
|
||||||
import { InputValidationError } from "../modality/contract.js";
|
import { InputValidationError } from "../modality/contract.js";
|
||||||
import { inputHandlerFor } from "../modality/input-handler.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 startTime = Date.now();
|
||||||
const pool = resolveToolPool(config.toolId);
|
const pool = resolveToolPool(config.toolId);
|
||||||
|
|
||||||
@@ -451,7 +477,7 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
|
|||||||
await enqueueToolJob({
|
await enqueueToolJob({
|
||||||
jobId,
|
jobId,
|
||||||
toolId: config.toolId,
|
toolId: config.toolId,
|
||||||
userId: getAuthUser(request)?.id ?? null,
|
userId,
|
||||||
pool,
|
pool,
|
||||||
inputRefs,
|
inputRefs,
|
||||||
filename,
|
filename,
|
||||||
|
|||||||
Reference in New Issue
Block a user