diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 360b5029..25ce0577 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -566,6 +566,7 @@ app.get("/api/v1/readyz", async (_request, reply) => { // Cancel a job (authenticated) app.post( "/api/v1/jobs/:jobId/cancel", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async ( request: import("fastify").FastifyRequest<{ Params: { jobId: string } }>, reply: import("fastify").FastifyReply, diff --git a/apps/api/src/plugins/auth.ts b/apps/api/src/plugins/auth.ts index d192acdb..a3c59a03 100644 --- a/apps/api/src/plugins/auth.ts +++ b/apps/api/src/plugins/auth.ts @@ -547,72 +547,78 @@ export async function authRoutes(app: FastifyInstance): Promise { }); // POST /api/auth/change-password - app.post("/api/auth/change-password", async (request: FastifyRequest, reply: FastifyReply) => { - const authUser = requireAuth(request, reply); - if (!authUser) return; + app.post( + "/api/auth/change-password", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const authUser = requireAuth(request, reply); + if (!authUser) return; - const parsed = changePasswordSchema.safeParse(request.body); - if (!parsed.success) { - return reply.status(400).send({ - error: "Current password and new password are required", - code: "VALIDATION_ERROR", - }); - } - const body = parsed.data; + const parsed = changePasswordSchema.safeParse(request.body); + if (!parsed.success) { + return reply.status(400).send({ + error: "Current password and new password are required", + code: "VALIDATION_ERROR", + }); + } + const body = parsed.data; - const pwError = await validatePasswordStrength(body.newPassword); - if (pwError) { - return reply.status(400).send({ - error: pwError, - code: "VALIDATION_ERROR", - }); - } + const pwError = await validatePasswordStrength(body.newPassword); + if (pwError) { + return reply.status(400).send({ + error: pwError, + code: "VALIDATION_ERROR", + }); + } - const [user] = await db.select().from(schema.users).where(eq(schema.users.id, authUser.id)); + const [user] = await db.select().from(schema.users).where(eq(schema.users.id, authUser.id)); - if (!user) { - return reply.status(404).send({ error: "User not found", code: "NOT_FOUND" }); - } + if (!user) { + return reply.status(404).send({ error: "User not found", code: "NOT_FOUND" }); + } - if (!user.passwordHash) { - return reply.status(400).send({ - error: "Password changes are managed by your identity provider.", - code: "OIDC_NO_PASSWORD", - }); - } + if (!user.passwordHash) { + return reply.status(400).send({ + error: "Password changes are managed by your identity provider.", + code: "OIDC_NO_PASSWORD", + }); + } - const valid = await verifyPassword(body.currentPassword, user.passwordHash); - if (!valid) { - return reply - .status(401) - .send({ error: "Current password is incorrect", code: "INVALID_PASSWORD" }); - } + const valid = await verifyPassword(body.currentPassword, user.passwordHash); + if (!valid) { + return reply + .status(401) + .send({ error: "Current password is incorrect", code: "INVALID_PASSWORD" }); + } - const newHash = await hashPassword(body.newPassword); + const newHash = await hashPassword(body.newPassword); - await db - .update(schema.users) - .set({ passwordHash: newHash, mustChangePassword: false, updatedAt: new Date() }) - .where(eq(schema.users.id, authUser.id)); - - // Invalidate all other sessions for this user - const currentToken = extractToken(request); - if (currentToken) { await db - .delete(schema.sessions) - .where(and(eq(schema.sessions.userId, authUser.id), ne(schema.sessions.id, currentToken))); - } + .update(schema.users) + .set({ passwordHash: newHash, mustChangePassword: false, updatedAt: new Date() }) + .where(eq(schema.users.id, authUser.id)); - // Revoke all API keys - if credentials were compromised, keys must be rotated too - await db.delete(schema.apiKeys).where(eq(schema.apiKeys.userId, authUser.id)); + // Invalidate all other sessions for this user + const currentToken = extractToken(request); + if (currentToken) { + await db + .delete(schema.sessions) + .where( + and(eq(schema.sessions.userId, authUser.id), ne(schema.sessions.id, currentToken)), + ); + } - await auditFromRequest(request)("PASSWORD_CHANGED", { - userId: authUser.id, - username: authUser.username, - }); + // Revoke all API keys - if credentials were compromised, keys must be rotated too + await db.delete(schema.apiKeys).where(eq(schema.apiKeys.userId, authUser.id)); - return reply.send({ ok: true }); - }); + await auditFromRequest(request)("PASSWORD_CHANGED", { + userId: authUser.id, + username: authUser.username, + }); + + return reply.send({ ok: true }); + }, + ); // GET /api/auth/users (admin only) app.get("/api/auth/users", async (request: FastifyRequest, reply: FastifyReply) => { diff --git a/apps/api/src/plugins/mfa.ts b/apps/api/src/plugins/mfa.ts index 94fa27d8..0a5880b4 100644 --- a/apps/api/src/plugins/mfa.ts +++ b/apps/api/src/plugins/mfa.ts @@ -110,290 +110,315 @@ export function isMfaRequiredForUser(policy: MfaPolicy, userRole: string): boole export async function registerMfa(app: FastifyInstance): Promise { // POST /api/auth/mfa/enroll -- start MFA enrollment - app.post("/api/auth/mfa/enroll", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.post( + "/api/auth/mfa/enroll", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - // Check enterprise feature gate - let mfaLicensed = false; - try { - const { isFeatureEnabled } = await import("@snapotter/enterprise"); - mfaLicensed = isFeatureEnabled("mfa"); - } catch { - // Enterprise package not available - } + // Check enterprise feature gate + let mfaLicensed = false; + try { + const { isFeatureEnabled } = await import("@snapotter/enterprise"); + mfaLicensed = isFeatureEnabled("mfa"); + } catch { + // Enterprise package not available + } - if (!mfaLicensed) { - return reply.status(403).send({ - error: "MFA requires an enterprise license", - code: "FEATURE_NOT_LICENSED", - }); - } + if (!mfaLicensed) { + return reply.status(403).send({ + error: "MFA requires an enterprise license", + code: "FEATURE_NOT_LICENSED", + }); + } - // Check if already enrolled - const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, user.id)); - if (!dbUser) { - return reply.status(404).send({ error: "User not found", code: "NOT_FOUND" }); - } - if (dbUser.totpEnabled) { - return reply.status(409).send({ - error: "MFA is already enabled. Disable it first to re-enroll.", - code: "MFA_ALREADY_ENABLED", - }); - } + // Check if already enrolled + const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, user.id)); + if (!dbUser) { + return reply.status(404).send({ error: "User not found", code: "NOT_FOUND" }); + } + if (dbUser.totpEnabled) { + return reply.status(409).send({ + error: "MFA is already enabled. Disable it first to re-enroll.", + code: "MFA_ALREADY_ENABLED", + }); + } - // Check if there's already a pending (unverified) enrollment - if (dbUser.totpSecret && !dbUser.totpEnabled) { - return reply.status(409).send({ - error: - "MFA enrollment already pending. Complete verification first or contact an admin to reset.", - code: "MFA_ENROLLMENT_PENDING", - }); - } + // Check if there's already a pending (unverified) enrollment + if (dbUser.totpSecret && !dbUser.totpEnabled) { + return reply.status(409).send({ + error: + "MFA enrollment already pending. Complete verification first or contact an admin to reset.", + code: "MFA_ENROLLMENT_PENDING", + }); + } - // Generate TOTP secret - const totp = createTotp(user.username); - const uri = totp.toString(); + // Generate TOTP secret + const totp = createTotp(user.username); + const uri = totp.toString(); - // Generate recovery codes - const recoveryCodes = generateRecoveryCodes(); - const recoveryHash = hashRecoveryCodes(recoveryCodes); + // Generate recovery codes + const recoveryCodes = generateRecoveryCodes(); + const recoveryHash = hashRecoveryCodes(recoveryCodes); - // Encrypt TOTP secret for storage - const encryptedSecret = await encryptSecret(totp.secret.base32); + // Encrypt TOTP secret for storage + const encryptedSecret = await encryptSecret(totp.secret.base32); - // Store pending enrollment (not yet active) - await db - .update(schema.users) - .set({ - totpSecret: encryptedSecret, - totpEnabled: false, - recoveryCodesHash: recoveryHash, - updatedAt: new Date(), - }) - .where(eq(schema.users.id, user.id)); + // Store pending enrollment (not yet active) + await db + .update(schema.users) + .set({ + totpSecret: encryptedSecret, + totpEnabled: false, + recoveryCodesHash: recoveryHash, + updatedAt: new Date(), + }) + .where(eq(schema.users.id, user.id)); - return reply.send({ uri, recoveryCodes }); - }); + return reply.send({ uri, recoveryCodes }); + }, + ); // POST /api/auth/mfa/verify -- confirm enrollment with a TOTP code - app.post("/api/auth/mfa/verify", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.post( + "/api/auth/mfa/verify", + { + config: { rateLimit: { max: 15, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const parsed = verifyCodeSchema.safeParse(request.body); - if (!parsed.success) { - return reply.status(400).send({ - error: "A valid TOTP code is required", - code: "VALIDATION_ERROR", - }); - } - const { code } = parsed.data; + const parsed = verifyCodeSchema.safeParse(request.body); + if (!parsed.success) { + return reply.status(400).send({ + error: "A valid TOTP code is required", + code: "VALIDATION_ERROR", + }); + } + const { code } = parsed.data; - const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, user.id)); - if (!dbUser?.totpSecret) { - return reply.status(400).send({ - error: "No pending MFA enrollment found. Call /api/auth/mfa/enroll first.", - code: "NO_PENDING_ENROLLMENT", - }); - } - if (dbUser.totpEnabled) { - return reply.status(409).send({ - error: "MFA is already verified and active", - code: "MFA_ALREADY_ENABLED", - }); - } + const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, user.id)); + if (!dbUser?.totpSecret) { + return reply.status(400).send({ + error: "No pending MFA enrollment found. Call /api/auth/mfa/enroll first.", + code: "NO_PENDING_ENROLLMENT", + }); + } + if (dbUser.totpEnabled) { + return reply.status(409).send({ + error: "MFA is already verified and active", + code: "MFA_ALREADY_ENABLED", + }); + } - // Decrypt the stored secret - const secretBase32 = await decryptSecret(dbUser.totpSecret); - if (!secretBase32) { - return reply.status(500).send({ - error: "Failed to decrypt TOTP secret", - code: "DECRYPTION_FAILED", - }); - } + // Decrypt the stored secret + const secretBase32 = await decryptSecret(dbUser.totpSecret); + if (!secretBase32) { + return reply.status(500).send({ + error: "Failed to decrypt TOTP secret", + code: "DECRYPTION_FAILED", + }); + } - // Validate the code - if (!verifyTotpCode(secretBase32, code)) { - return reply.status(401).send({ - error: "Invalid TOTP code", - code: "INVALID_CODE", - }); - } + // Validate the code + if (!verifyTotpCode(secretBase32, code)) { + return reply.status(401).send({ + error: "Invalid TOTP code", + code: "INVALID_CODE", + }); + } - // Activate MFA - await db - .update(schema.users) - .set({ totpEnabled: true, updatedAt: new Date() }) - .where(eq(schema.users.id, user.id)); + // Activate MFA + await db + .update(schema.users) + .set({ totpEnabled: true, updatedAt: new Date() }) + .where(eq(schema.users.id, user.id)); - const audit = auditFromRequest(request); - await audit("MFA_ENROLLED", { userId: user.id, username: user.username }); + const audit = auditFromRequest(request); + await audit("MFA_ENROLLED", { userId: user.id, username: user.username }); - return reply.send({ ok: true }); - }); + return reply.send({ ok: true }); + }, + ); // POST /api/auth/mfa/complete -- complete login with TOTP code - app.post("/api/auth/mfa/complete", async (request: FastifyRequest, reply: FastifyReply) => { - const parsed = completeSchema.safeParse(request.body); - if (!parsed.success) { - return reply.status(400).send({ - error: "MFA token and code are required", - code: "VALIDATION_ERROR", - }); - } - const { mfaToken, code } = parsed.data; - - // Look up the pending MFA challenge in Redis - const redis = sharedRedis(); - const userId = await redis.get(`mfa:${mfaToken}`); - if (!userId) { - return reply.status(401).send({ - error: "MFA challenge expired or invalid", - code: "MFA_EXPIRED", - }); - } - - // Load user - const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, userId)); - if (!dbUser?.totpSecret) { - return reply.status(401).send({ - error: "User not found or MFA not configured", - code: "MFA_NOT_CONFIGURED", - }); - } - - // Decrypt the stored secret - const secretBase32 = await decryptSecret(dbUser.totpSecret); - if (!secretBase32) { - return reply.status(500).send({ - error: "Failed to decrypt TOTP secret", - code: "DECRYPTION_FAILED", - }); - } - - const audit = auditFromRequest(request); - let verified = false; - let recoveryUsed = false; - - // Try TOTP code first - if (verifyTotpCode(secretBase32, code)) { - verified = true; - } - - // Try recovery code if TOTP failed - if (!verified && dbUser.recoveryCodesHash) { - const result = verifyRecoveryCode(code, dbUser.recoveryCodesHash); - if (result.valid) { - verified = true; - recoveryUsed = true; - // Consume the recovery code - await db - .update(schema.users) - .set({ recoveryCodesHash: result.remaining || null, updatedAt: new Date() }) - .where(eq(schema.users.id, userId)); + app.post( + "/api/auth/mfa/complete", + { + config: { rateLimit: { max: 15, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + const parsed = completeSchema.safeParse(request.body); + if (!parsed.success) { + return reply.status(400).send({ + error: "MFA token and code are required", + code: "VALIDATION_ERROR", + }); } - } + const { mfaToken, code } = parsed.data; - if (!verified) { - await audit("MFA_VERIFY_FAILED", { userId, username: dbUser.username }); - return reply.status(401).send({ - error: "Invalid TOTP or recovery code", - code: "INVALID_CODE", + // Look up the pending MFA challenge in Redis + const redis = sharedRedis(); + const userId = await redis.get(`mfa:${mfaToken}`); + if (!userId) { + return reply.status(401).send({ + error: "MFA challenge expired or invalid", + code: "MFA_EXPIRED", + }); + } + + // Load user + const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, userId)); + if (!dbUser?.totpSecret) { + return reply.status(401).send({ + error: "User not found or MFA not configured", + code: "MFA_NOT_CONFIGURED", + }); + } + + // Decrypt the stored secret + const secretBase32 = await decryptSecret(dbUser.totpSecret); + if (!secretBase32) { + return reply.status(500).send({ + error: "Failed to decrypt TOTP secret", + code: "DECRYPTION_FAILED", + }); + } + + const audit = auditFromRequest(request); + let verified = false; + let recoveryUsed = false; + + // Try TOTP code first + if (verifyTotpCode(secretBase32, code)) { + verified = true; + } + + // Try recovery code if TOTP failed + if (!verified && dbUser.recoveryCodesHash) { + const result = verifyRecoveryCode(code, dbUser.recoveryCodesHash); + if (result.valid) { + verified = true; + recoveryUsed = true; + // Consume the recovery code + await db + .update(schema.users) + .set({ recoveryCodesHash: result.remaining || null, updatedAt: new Date() }) + .where(eq(schema.users.id, userId)); + } + } + + if (!verified) { + await audit("MFA_VERIFY_FAILED", { userId, username: dbUser.username }); + return reply.status(401).send({ + error: "Invalid TOTP or recovery code", + code: "INVALID_CODE", + }); + } + + // Delete the challenge token + await redis.del(`mfa:${mfaToken}`); + + // Create session (same as normal login completion) + const token = createSessionToken(); + const expiresAt = new Date(Date.now() + SESSION_DURATION_MS); + + await db.insert(schema.sessions).values({ + id: token, + userId: dbUser.id, + expiresAt, }); - } - // Delete the challenge token - await redis.del(`mfa:${mfaToken}`); - - // Create session (same as normal login completion) - const token = createSessionToken(); - const expiresAt = new Date(Date.now() + SESSION_DURATION_MS); - - await db.insert(schema.sessions).values({ - id: token, - userId: dbUser.id, - expiresAt, - }); - - await audit(recoveryUsed ? "MFA_RECOVERY_USED" : "MFA_VERIFIED", { - userId: dbUser.id, - username: dbUser.username, - }); - - const [teamRow] = await db.select().from(schema.teams).where(eq(schema.teams.id, dbUser.team)); - - return reply.send({ - token, - user: { - id: dbUser.id, + await audit(recoveryUsed ? "MFA_RECOVERY_USED" : "MFA_VERIFIED", { + userId: dbUser.id, username: dbUser.username, - role: dbUser.role, - mustChangePassword: env.SKIP_MUST_CHANGE_PASSWORD ? false : dbUser.mustChangePassword, - permissions: await getPermissions(dbUser.role), - teamName: teamRow?.name ?? dbUser.team, - analyticsEnabled: dbUser.analyticsEnabled ?? null, - analyticsConsentShownAt: dbUser.analyticsConsentShownAt?.getTime() ?? null, - analyticsConsentRemindAt: dbUser.analyticsConsentRemindAt?.getTime() ?? null, - }, - expiresAt: expiresAt.toISOString(), - }); - }); + }); + + const [teamRow] = await db + .select() + .from(schema.teams) + .where(eq(schema.teams.id, dbUser.team)); + + return reply.send({ + token, + user: { + id: dbUser.id, + username: dbUser.username, + role: dbUser.role, + mustChangePassword: env.SKIP_MUST_CHANGE_PASSWORD ? false : dbUser.mustChangePassword, + permissions: await getPermissions(dbUser.role), + teamName: teamRow?.name ?? dbUser.team, + analyticsEnabled: dbUser.analyticsEnabled ?? null, + analyticsConsentShownAt: dbUser.analyticsConsentShownAt?.getTime() ?? null, + analyticsConsentRemindAt: dbUser.analyticsConsentRemindAt?.getTime() ?? null, + }, + expiresAt: expiresAt.toISOString(), + }); + }, + ); // POST /api/auth/mfa/disable -- disable MFA (self-service) - app.post("/api/auth/mfa/disable", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.post( + "/api/auth/mfa/disable", + { + config: { rateLimit: { max: 15, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const parsed = disableSchema.safeParse(request.body); - if (!parsed.success) { - return reply.status(400).send({ - error: "Current TOTP code is required to disable MFA", - code: "VALIDATION_ERROR", - }); - } - const { code } = parsed.data; + const parsed = disableSchema.safeParse(request.body); + if (!parsed.success) { + return reply.status(400).send({ + error: "Current TOTP code is required to disable MFA", + code: "VALIDATION_ERROR", + }); + } + const { code } = parsed.data; - const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, user.id)); - if (!dbUser?.totpEnabled || !dbUser.totpSecret) { - return reply.status(400).send({ - error: "MFA is not enabled", - code: "MFA_NOT_ENABLED", - }); - } + const [dbUser] = await db.select().from(schema.users).where(eq(schema.users.id, user.id)); + if (!dbUser?.totpEnabled || !dbUser.totpSecret) { + return reply.status(400).send({ + error: "MFA is not enabled", + code: "MFA_NOT_ENABLED", + }); + } - // Decrypt and verify the code - const secretBase32 = await decryptSecret(dbUser.totpSecret); - if (!secretBase32) { - return reply.status(500).send({ - error: "Failed to decrypt TOTP secret", - code: "DECRYPTION_FAILED", - }); - } + // Decrypt and verify the code + const secretBase32 = await decryptSecret(dbUser.totpSecret); + if (!secretBase32) { + return reply.status(500).send({ + error: "Failed to decrypt TOTP secret", + code: "DECRYPTION_FAILED", + }); + } - if (!verifyTotpCode(secretBase32, code)) { - return reply.status(401).send({ - error: "Invalid TOTP code", - code: "INVALID_CODE", - }); - } + if (!verifyTotpCode(secretBase32, code)) { + return reply.status(401).send({ + error: "Invalid TOTP code", + code: "INVALID_CODE", + }); + } - // Clear MFA data - await db - .update(schema.users) - .set({ - totpSecret: null, - totpEnabled: false, - recoveryCodesHash: null, - updatedAt: new Date(), - }) - .where(eq(schema.users.id, user.id)); + // Clear MFA data + await db + .update(schema.users) + .set({ + totpSecret: null, + totpEnabled: false, + recoveryCodesHash: null, + updatedAt: new Date(), + }) + .where(eq(schema.users.id, user.id)); - const audit = auditFromRequest(request); - await audit("MFA_DISABLED", { userId: user.id, username: user.username }); + const audit = auditFromRequest(request); + await audit("MFA_DISABLED", { userId: user.id, username: user.username }); - return reply.send({ ok: true }); - }); + return reply.send({ ok: true }); + }, + ); // POST /api/auth/users/:id/mfa/reset -- admin reset app.post( diff --git a/apps/api/src/plugins/oidc.ts b/apps/api/src/plugins/oidc.ts index fdc1dc78..2bd8cfa6 100644 --- a/apps/api/src/plugins/oidc.ts +++ b/apps/api/src/plugins/oidc.ts @@ -103,197 +103,205 @@ export async function oidcRoutes(app: FastifyInstance): Promise { if (!env.OIDC_ENABLED) return; // GET /api/auth/oidc/login - app.get("/api/auth/oidc/login", async (request: FastifyRequest, reply: FastifyReply) => { - let config: oidc.Configuration; - try { - config = await getOrDiscoverConfig(); - } catch (err) { - request.log.error({ err }, "OIDC discovery failed"); - return redirectToLogin(reply, "oidc_provider_unreachable"); - } + app.get( + "/api/auth/oidc/login", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + let config: oidc.Configuration; + try { + config = await getOrDiscoverConfig(); + } catch (err) { + request.log.error({ err }, "OIDC discovery failed"); + return redirectToLogin(reply, "oidc_provider_unreachable"); + } - const state = oidc.randomState(); - const nonce = oidc.randomNonce(); - const codeVerifier = oidc.randomPKCECodeVerifier(); - const codeChallenge = await oidc.calculatePKCECodeChallenge(codeVerifier); + const state = oidc.randomState(); + const nonce = oidc.randomNonce(); + const codeVerifier = oidc.randomPKCECodeVerifier(); + const codeChallenge = await oidc.calculatePKCECodeChallenge(codeVerifier); - const redirectUri = `${env.EXTERNAL_URL}/api/auth/oidc/callback`; + const redirectUri = `${env.EXTERNAL_URL}/api/auth/oidc/callback`; - // Store OIDC state in a signed cookie - const statePayload: OidcStateCookie = { state, nonce, codeVerifier }; - const cookieValue = reply.signCookie(JSON.stringify(statePayload)); + // Store OIDC state in a signed cookie + const statePayload: OidcStateCookie = { state, nonce, codeVerifier }; + const cookieValue = reply.signCookie(JSON.stringify(statePayload)); - reply.setCookie("oidc-state", cookieValue, { - httpOnly: true, - sameSite: "lax", - secure: isSecure(), - path: "/api/auth/oidc", - maxAge: 600, // 10 minutes - signed: false, // already signed manually - }); + reply.setCookie("oidc-state", cookieValue, { + httpOnly: true, + sameSite: "lax", + secure: isSecure(), + path: "/api/auth/oidc", + maxAge: 600, // 10 minutes + signed: false, // already signed manually + }); - const authorizationUrl = oidc.buildAuthorizationUrl(config, { - redirect_uri: redirectUri, - scope: env.OIDC_SCOPES, - state, - nonce, - code_challenge: codeChallenge, - code_challenge_method: "S256", - }); + const authorizationUrl = oidc.buildAuthorizationUrl(config, { + redirect_uri: redirectUri, + scope: env.OIDC_SCOPES, + state, + nonce, + code_challenge: codeChallenge, + code_challenge_method: "S256", + }); - return reply.redirect(authorizationUrl.href); - }); + return reply.redirect(authorizationUrl.href); + }, + ); // GET /api/auth/oidc/callback - app.get("/api/auth/oidc/callback", async (request: FastifyRequest, reply: FastifyReply) => { - // 1. Validate state from signed cookie - const rawCookie = request.cookies?.["oidc-state"]; - if (!rawCookie) { - request.log.warn("OIDC callback: missing state cookie"); - return redirectToLogin(reply, "oidc_session_expired"); - } - - // Clear the cookie immediately - reply.clearCookie("oidc-state", { - path: "/api/auth/oidc", - httpOnly: true, - sameSite: "lax", - secure: isSecure(), - }); - - const unsigned = request.unsignCookie(rawCookie); - if (!unsigned.valid || !unsigned.value) { - request.log.warn("OIDC callback: invalid cookie signature"); - return redirectToLogin(reply, "oidc_session_expired"); - } - - let storedState: OidcStateCookie; - try { - storedState = JSON.parse(unsigned.value) as OidcStateCookie; - } catch { - request.log.warn("OIDC callback: malformed state cookie"); - return redirectToLogin(reply, "oidc_session_expired"); - } - - // Validate state parameter matches - const query = request.query as Record; - if (query.state !== storedState.state) { - request.log.warn("OIDC callback: state mismatch"); - return redirectToLogin(reply, "oidc_session_expired"); - } - - const audit = auditFromRequest(request); - - // Check for error response from the IdP - if (query.error) { - request.log.warn( - { error: query.error, description: query.error_description }, - "OIDC IdP returned error", - ); - authAttempts.inc({ method: "oidc", result: "failure" }); - await audit("OIDC_LOGIN_FAILED", { - reason: sanitizeAuditInput(String(query.error)), - }); - return redirectToLogin(reply, "oidc_auth_failed"); - } - - // 2. Exchange authorization code for tokens - let config: oidc.Configuration; - try { - config = await getOrDiscoverConfig(); - } catch (err) { - request.log.error({ err }, "OIDC discovery failed during callback"); - return redirectToLogin(reply, "oidc_provider_unreachable"); - } - - let tokenResponse: Awaited>; - try { - const callbackUrl = new URL(`${env.EXTERNAL_URL}/api/auth/oidc/callback`); - // Copy the query parameters from the actual request - for (const [key, value] of Object.entries(query)) { - callbackUrl.searchParams.set(key, value); + app.get( + "/api/auth/oidc/callback", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + // 1. Validate state from signed cookie + const rawCookie = request.cookies?.["oidc-state"]; + if (!rawCookie) { + request.log.warn("OIDC callback: missing state cookie"); + return redirectToLogin(reply, "oidc_session_expired"); } - tokenResponse = await oidc.authorizationCodeGrant(config, callbackUrl, { - pkceCodeVerifier: storedState.codeVerifier, - expectedNonce: storedState.nonce, - expectedState: storedState.state, + // Clear the cookie immediately + reply.clearCookie("oidc-state", { + path: "/api/auth/oidc", + httpOnly: true, + sameSite: "lax", + secure: isSecure(), }); - } catch (err) { - request.log.error({ err }, "OIDC token exchange failed"); - authAttempts.inc({ method: "oidc", result: "failure" }); - await audit("OIDC_LOGIN_FAILED", { reason: "token_exchange_failed" }); - return redirectToLogin(reply, "oidc_auth_failed"); - } - // 3. Extract claims from ID token - const claims = tokenResponse.claims(); - if (!claims) { - request.log.error("OIDC callback: no ID token claims"); - authAttempts.inc({ method: "oidc", result: "failure" }); - await audit("OIDC_LOGIN_FAILED", { reason: "no_id_token" }); - return redirectToLogin(reply, "oidc_auth_failed"); - } - - const sub = claims.sub; - const email = typeof claims.email === "string" ? claims.email : undefined; - const emailVerified = claims.email_verified === true; - const rawUsername = deriveUsername(claims as Record); - const derivedUsername = sanitizeUsername(rawUsername); - const idToken = tokenResponse.id_token ?? null; - - // 4. User resolution (delegated to shared resolver) - const result = await resolveExternalUser({ - provider: "oidc", - externalId: sub, - email, - emailVerified, - username: derivedUsername, - autoCreate: env.OIDC_AUTO_CREATE_USERS, - autoLink: env.OIDC_AUTO_LINK_USERS, - defaultRole: env.OIDC_DEFAULT_ROLE, - logger: request.log, - ip: request.ip, - requestId: request.id, - }); - - if (result.action === "denied" || !result.user) { - authAttempts.inc({ method: "oidc", result: "failure" }); - if (result.deniedReason === "user_limit_reached") { - return redirectToLogin(reply, "oidc_user_limit_reached"); + const unsigned = request.unsignCookie(rawCookie); + if (!unsigned.valid || !unsigned.value) { + request.log.warn("OIDC callback: invalid cookie signature"); + return redirectToLogin(reply, "oidc_session_expired"); } - return redirectToLogin(reply, "oidc_user_not_authorized"); - } - const resolvedUser = result.user; + let storedState: OidcStateCookie; + try { + storedState = JSON.parse(unsigned.value) as OidcStateCookie; + } catch { + request.log.warn("OIDC callback: malformed state cookie"); + return redirectToLogin(reply, "oidc_session_expired"); + } - // 5. Create session - const token = createSessionToken(); - const expiresAt = new Date(Date.now() + SESSION_DURATION_MS); + // Validate state parameter matches + const query = request.query as Record; + if (query.state !== storedState.state) { + request.log.warn("OIDC callback: state mismatch"); + return redirectToLogin(reply, "oidc_session_expired"); + } - await db.insert(schema.sessions).values({ - id: token, - userId: resolvedUser.id, - expiresAt, - idToken, - }); + const audit = auditFromRequest(request); - authAttempts.inc({ method: "oidc", result: "success" }); - await audit("OIDC_LOGIN_SUCCESS", { - userId: resolvedUser.id, - username: resolvedUser.username, - }); + // Check for error response from the IdP + if (query.error) { + request.log.warn( + { error: query.error, description: query.error_description }, + "OIDC IdP returned error", + ); + authAttempts.inc({ method: "oidc", result: "failure" }); + await audit("OIDC_LOGIN_FAILED", { + reason: sanitizeAuditInput(String(query.error)), + }); + return redirectToLogin(reply, "oidc_auth_failed"); + } - // 6. Set session cookie - reply.setCookie("snapotter-session", token, { - httpOnly: true, - sameSite: "strict", - secure: isSecure(), - path: "/", - maxAge: env.SESSION_DURATION_HOURS * 3600, - }); + // 2. Exchange authorization code for tokens + let config: oidc.Configuration; + try { + config = await getOrDiscoverConfig(); + } catch (err) { + request.log.error({ err }, "OIDC discovery failed during callback"); + return redirectToLogin(reply, "oidc_provider_unreachable"); + } - // 7. Redirect to app - return reply.redirect("/"); - }); + let tokenResponse: Awaited>; + try { + const callbackUrl = new URL(`${env.EXTERNAL_URL}/api/auth/oidc/callback`); + // Copy the query parameters from the actual request + for (const [key, value] of Object.entries(query)) { + callbackUrl.searchParams.set(key, value); + } + + tokenResponse = await oidc.authorizationCodeGrant(config, callbackUrl, { + pkceCodeVerifier: storedState.codeVerifier, + expectedNonce: storedState.nonce, + expectedState: storedState.state, + }); + } catch (err) { + request.log.error({ err }, "OIDC token exchange failed"); + authAttempts.inc({ method: "oidc", result: "failure" }); + await audit("OIDC_LOGIN_FAILED", { reason: "token_exchange_failed" }); + return redirectToLogin(reply, "oidc_auth_failed"); + } + + // 3. Extract claims from ID token + const claims = tokenResponse.claims(); + if (!claims) { + request.log.error("OIDC callback: no ID token claims"); + authAttempts.inc({ method: "oidc", result: "failure" }); + await audit("OIDC_LOGIN_FAILED", { reason: "no_id_token" }); + return redirectToLogin(reply, "oidc_auth_failed"); + } + + const sub = claims.sub; + const email = typeof claims.email === "string" ? claims.email : undefined; + const emailVerified = claims.email_verified === true; + const rawUsername = deriveUsername(claims as Record); + const derivedUsername = sanitizeUsername(rawUsername); + const idToken = tokenResponse.id_token ?? null; + + // 4. User resolution (delegated to shared resolver) + const result = await resolveExternalUser({ + provider: "oidc", + externalId: sub, + email, + emailVerified, + username: derivedUsername, + autoCreate: env.OIDC_AUTO_CREATE_USERS, + autoLink: env.OIDC_AUTO_LINK_USERS, + defaultRole: env.OIDC_DEFAULT_ROLE, + logger: request.log, + ip: request.ip, + requestId: request.id, + }); + + if (result.action === "denied" || !result.user) { + authAttempts.inc({ method: "oidc", result: "failure" }); + if (result.deniedReason === "user_limit_reached") { + return redirectToLogin(reply, "oidc_user_limit_reached"); + } + return redirectToLogin(reply, "oidc_user_not_authorized"); + } + + const resolvedUser = result.user; + + // 5. Create session + const token = createSessionToken(); + const expiresAt = new Date(Date.now() + SESSION_DURATION_MS); + + await db.insert(schema.sessions).values({ + id: token, + userId: resolvedUser.id, + expiresAt, + idToken, + }); + + authAttempts.inc({ method: "oidc", result: "success" }); + await audit("OIDC_LOGIN_SUCCESS", { + userId: resolvedUser.id, + username: resolvedUser.username, + }); + + // 6. Set session cookie + reply.setCookie("snapotter-session", token, { + httpOnly: true, + sameSite: "strict", + secure: isSecure(), + path: "/", + maxAge: env.SESSION_DURATION_HOURS * 3600, + }); + + // 7. Redirect to app + return reply.redirect("/"); + }, + ); } diff --git a/apps/api/src/plugins/saml.ts b/apps/api/src/plugins/saml.ts index 1f403e4c..41b05446 100644 --- a/apps/api/src/plugins/saml.ts +++ b/apps/api/src/plugins/saml.ts @@ -79,105 +79,117 @@ export async function registerSaml(app: FastifyInstance): Promise { }); // GET /api/auth/saml/login -- SP-initiated login redirect - app.get("/api/auth/saml/login", async (_request: FastifyRequest, reply: FastifyReply) => { - try { - const saml = getSamlInstance(); - const loginUrl = await saml.getAuthorizeUrlAsync("", undefined, {}); - return reply.redirect(loginUrl); - } catch (err) { - _request.log.error({ err }, "SAML login redirect failed"); - return redirectToLogin(reply, "saml_auth_failed"); - } - }); + app.get( + "/api/auth/saml/login", + { + config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, + }, + async (_request: FastifyRequest, reply: FastifyReply) => { + try { + const saml = getSamlInstance(); + const loginUrl = await saml.getAuthorizeUrlAsync("", undefined, {}); + return reply.redirect(loginUrl); + } catch (err) { + _request.log.error({ err }, "SAML login redirect failed"); + return redirectToLogin(reply, "saml_auth_failed"); + } + }, + ); // POST /api/auth/saml/callback -- Assertion Consumer Service (ACS) - app.post("/api/auth/saml/callback", async (request: FastifyRequest, reply: FastifyReply) => { - const saml = getSamlInstance(); - const audit = auditFromRequest(request); + app.post( + "/api/auth/saml/callback", + { + config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + const saml = getSamlInstance(); + const audit = auditFromRequest(request); - let profile: Awaited>["profile"]; - try { - const result = await saml.validatePostResponseAsync(request.body as Record); - profile = result.profile; - } catch (err) { - request.log.error({ err }, "SAML assertion validation failed"); - authAttempts.inc({ method: "saml", result: "failure" }); - await audit("SAML_LOGIN_FAILED", { - error: err instanceof Error ? err.message : "Unknown error", + let profile: Awaited>["profile"]; + try { + const result = await saml.validatePostResponseAsync(request.body as Record); + profile = result.profile; + } catch (err) { + request.log.error({ err }, "SAML assertion validation failed"); + authAttempts.inc({ method: "saml", result: "failure" }); + await audit("SAML_LOGIN_FAILED", { + error: err instanceof Error ? err.message : "Unknown error", + }); + return redirectToLogin(reply, "saml_auth_failed"); + } + + if (!profile?.nameID) { + request.log.warn("SAML callback: no profile or nameID in assertion"); + authAttempts.inc({ method: "saml", result: "failure" }); + await audit("SAML_LOGIN_FAILED", { reason: "missing_profile" }); + return redirectToLogin(reply, "saml_auth_failed"); + } + + // Extract claims from SAML assertion + const externalId = profile.nameID; + const email = profile[env.SAML_EMAIL_ATTRIBUTE] as string | undefined; + const usernameAttr = env.SAML_USERNAME_ATTRIBUTE + ? (profile[env.SAML_USERNAME_ATTRIBUTE] as string | undefined) + : undefined; + + // Derive a username from available claims + const rawUsername = usernameAttr || email?.split("@")[0] || profile.nameID; + let username = sanitizeUsername(rawUsername); + username = await findUniqueUsername(username); + + // Resolve user via shared external-auth resolver + const result = await resolveExternalUser({ + provider: "saml", + externalId, + email, + emailVerified: true, // SAML assertions from a trusted IdP are considered verified + username, + autoCreate: env.SAML_AUTO_CREATE_USERS, + autoLink: env.SAML_AUTO_LINK_USERS, + defaultRole: env.SAML_DEFAULT_ROLE, + logger: request.log, + ip: request.ip, + requestId: request.id, }); - return redirectToLogin(reply, "saml_auth_failed"); - } - if (!profile?.nameID) { - request.log.warn("SAML callback: no profile or nameID in assertion"); - authAttempts.inc({ method: "saml", result: "failure" }); - await audit("SAML_LOGIN_FAILED", { reason: "missing_profile" }); - return redirectToLogin(reply, "saml_auth_failed"); - } + if (result.action === "denied" || !result.user) { + authAttempts.inc({ method: "saml", result: "failure" }); + const errorParam = + result.deniedReason === "user_limit_reached" + ? "saml_user_limit_reached" + : "saml_user_not_authorized"; + return redirectToLogin(reply, errorParam); + } - // Extract claims from SAML assertion - const externalId = profile.nameID; - const email = profile[env.SAML_EMAIL_ATTRIBUTE] as string | undefined; - const usernameAttr = env.SAML_USERNAME_ATTRIBUTE - ? (profile[env.SAML_USERNAME_ATTRIBUTE] as string | undefined) - : undefined; + const resolvedUser = result.user; - // Derive a username from available claims - const rawUsername = usernameAttr || email?.split("@")[0] || profile.nameID; - let username = sanitizeUsername(rawUsername); - username = await findUniqueUsername(username); + // Create session (same pattern as OIDC) + const token = createSessionToken(); + const expiresAt = new Date(Date.now() + SESSION_DURATION_MS); - // Resolve user via shared external-auth resolver - const result = await resolveExternalUser({ - provider: "saml", - externalId, - email, - emailVerified: true, // SAML assertions from a trusted IdP are considered verified - username, - autoCreate: env.SAML_AUTO_CREATE_USERS, - autoLink: env.SAML_AUTO_LINK_USERS, - defaultRole: env.SAML_DEFAULT_ROLE, - logger: request.log, - ip: request.ip, - requestId: request.id, - }); + await db.insert(schema.sessions).values({ + id: token, + userId: resolvedUser.id, + expiresAt, + }); - if (result.action === "denied" || !result.user) { - authAttempts.inc({ method: "saml", result: "failure" }); - const errorParam = - result.deniedReason === "user_limit_reached" - ? "saml_user_limit_reached" - : "saml_user_not_authorized"; - return redirectToLogin(reply, errorParam); - } + authAttempts.inc({ method: "saml", result: "success" }); + await audit("SAML_LOGIN_SUCCESS", { + userId: resolvedUser.id, + username: resolvedUser.username, + }); - const resolvedUser = result.user; + // Set session cookie and redirect to app + reply.setCookie("snapotter-session", token, { + httpOnly: true, + sameSite: "strict", + secure: isSecure(), + path: "/", + maxAge: env.SESSION_DURATION_HOURS * 3600, + }); - // Create session (same pattern as OIDC) - const token = createSessionToken(); - const expiresAt = new Date(Date.now() + SESSION_DURATION_MS); - - await db.insert(schema.sessions).values({ - id: token, - userId: resolvedUser.id, - expiresAt, - }); - - authAttempts.inc({ method: "saml", result: "success" }); - await audit("SAML_LOGIN_SUCCESS", { - userId: resolvedUser.id, - username: resolvedUser.username, - }); - - // Set session cookie and redirect to app - reply.setCookie("snapotter-session", token, { - httpOnly: true, - sameSite: "strict", - secure: isSecure(), - path: "/", - maxAge: env.SESSION_DURATION_HOURS * 3600, - }); - - return reply.redirect("/"); - }); + return reply.redirect("/"); + }, + ); } diff --git a/apps/api/src/routes/analytics.ts b/apps/api/src/routes/analytics.ts index f78c2815..abc275a8 100644 --- a/apps/api/src/routes/analytics.ts +++ b/apps/api/src/routes/analytics.ts @@ -38,45 +38,49 @@ export async function analyticsRoutes(app: FastifyInstance): Promise { }; }); - app.put("/api/v1/user/analytics", async (request, reply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.put( + "/api/v1/user/analytics", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, + async (request, reply) => { + const user = requireAuth(request, reply); + if (!user) return; - const parsed = analyticsConsentSchema.safeParse(request.body ?? {}); - if (!parsed.success) { - return reply.status(400).send({ - error: parsed.error.issues.map((i) => i.message).join("; "), - code: "VALIDATION_ERROR", - }); - } - const body = parsed.data; + const parsed = analyticsConsentSchema.safeParse(request.body ?? {}); + if (!parsed.success) { + return reply.status(400).send({ + error: parsed.error.issues.map((i) => i.message).join("; "), + code: "VALIDATION_ERROR", + }); + } + const body = parsed.data; - const now = new Date(); + const now = new Date(); - if (body.remindLater) { - const remindAt = new Date(Date.now() + 7 * 24 * 60 * 60 * 1000); + if (body.remindLater) { + const remindAt = new Date(Date.now() + 7 * 24 * 60 * 60 * 1000); + await db + .update(schema.users) + .set({ + analyticsEnabled: null, + analyticsConsentShownAt: now, + analyticsConsentRemindAt: remindAt, + updatedAt: now, + }) + .where(eq(schema.users.id, user.id)); + return reply.send({ ok: true, analyticsEnabled: null }); + } + + const enabled = body.enabled === true; await db .update(schema.users) .set({ - analyticsEnabled: null, + analyticsEnabled: enabled, analyticsConsentShownAt: now, - analyticsConsentRemindAt: remindAt, + analyticsConsentRemindAt: null, updatedAt: now, }) .where(eq(schema.users.id, user.id)); - return reply.send({ ok: true, analyticsEnabled: null }); - } - - const enabled = body.enabled === true; - await db - .update(schema.users) - .set({ - analyticsEnabled: enabled, - analyticsConsentShownAt: now, - analyticsConsentRemindAt: null, - updatedAt: now, - }) - .where(eq(schema.users.id, user.id)); - return reply.send({ ok: true, analyticsEnabled: enabled }); - }); + return reply.send({ ok: true, analyticsEnabled: enabled }); + }, + ); } diff --git a/apps/api/src/routes/api-keys.ts b/apps/api/src/routes/api-keys.ts index 1cc43947..05a890e3 100644 --- a/apps/api/src/routes/api-keys.ts +++ b/apps/api/src/routes/api-keys.ts @@ -22,122 +22,131 @@ const createApiKeySchema = z.object({ export async function apiKeyRoutes(app: FastifyInstance): Promise { // POST /api/v1/api-keys — Generate a new API key - app.post("/api/v1/api-keys", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.post( + "/api/v1/api-keys", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const parsed = createApiKeySchema.safeParse(request.body ?? {}); - if (!parsed.success) { - return reply.status(400).send({ - error: parsed.error.issues.map((i) => i.message).join("; "), - code: "VALIDATION_ERROR", - }); - } - const body = parsed.data; - const name = body.name?.trim() || "Default API Key"; - - let scopedPermissions: string[] | null = null; - if (body.permissions && body.permissions.length > 0) { - const userPerms = await getPermissions(user.role); - const permSet = new Set(userPerms); - const invalid = body.permissions.filter((p) => !permSet.has(p)); - if (invalid.length > 0) { + const parsed = createApiKeySchema.safeParse(request.body ?? {}); + if (!parsed.success) { return reply.status(400).send({ - error: `Cannot scope key with permissions you don't have: ${invalid.join(", ")}`, + error: parsed.error.issues.map((i) => i.message).join("; "), code: "VALIDATION_ERROR", }); } - scopedPermissions = body.permissions; - } + const body = parsed.data; + const name = body.name?.trim() || "Default API Key"; - let expiresAt: Date | null = null; - if (body.expiresAt) { - const parsedDate = new Date(body.expiresAt); - if (Number.isNaN(parsedDate.getTime())) { - return reply - .status(400) - .send({ error: "Invalid expiresAt date", code: "VALIDATION_ERROR" }); + let scopedPermissions: string[] | null = null; + if (body.permissions && body.permissions.length > 0) { + const userPerms = await getPermissions(user.role); + const permSet = new Set(userPerms); + const invalid = body.permissions.filter((p) => !permSet.has(p)); + if (invalid.length > 0) { + return reply.status(400).send({ + error: `Cannot scope key with permissions you don't have: ${invalid.join(", ")}`, + code: "VALIDATION_ERROR", + }); + } + scopedPermissions = body.permissions; } - if (parsedDate <= new Date()) { - return reply - .status(400) - .send({ error: "expiresAt must be in the future", code: "VALIDATION_ERROR" }); + + let expiresAt: Date | null = null; + if (body.expiresAt) { + const parsedDate = new Date(body.expiresAt); + if (Number.isNaN(parsedDate.getTime())) { + return reply + .status(400) + .send({ error: "Invalid expiresAt date", code: "VALIDATION_ERROR" }); + } + if (parsedDate <= new Date()) { + return reply + .status(400) + .send({ error: "expiresAt must be in the future", code: "VALIDATION_ERROR" }); + } + expiresAt = parsedDate; } - expiresAt = parsedDate; - } - // Generate a raw API key: "si_" prefix + 48 random bytes as hex - const rawKey = `si_${randomBytes(48).toString("hex")}`; - const keyHash = await hashPassword(rawKey); - const keyPrefix = computeKeyPrefix(rawKey); - const id = randomUUID(); + // Generate a raw API key: "si_" prefix + 48 random bytes as hex + const rawKey = `si_${randomBytes(48).toString("hex")}`; + const keyHash = await hashPassword(rawKey); + const keyPrefix = computeKeyPrefix(rawKey); + const id = randomUUID(); - try { - await db.insert(schema.apiKeys).values({ - id, + try { + await db.insert(schema.apiKeys).values({ + id, + userId: user.id, + keyHash, + keyPrefix, + name, + permissions: scopedPermissions, + expiresAt, + }); + } catch { + return reply.status(409).send({ error: "Failed to create API key" }); + } + + await auditFromRequest(request)("API_KEY_CREATED", { userId: user.id, - keyHash, - keyPrefix, + keyId: id, + keyName: name, + }); + + // Return the raw key ONCE — it cannot be retrieved again + return reply.status(201).send({ + id, + key: rawKey, name, permissions: scopedPermissions, - expiresAt, + expiresAt: expiresAt?.toISOString() ?? null, + createdAt: new Date().toISOString(), }); - } catch { - return reply.status(409).send({ error: "Failed to create API key" }); - } - - await auditFromRequest(request)("API_KEY_CREATED", { - userId: user.id, - keyId: id, - keyName: name, - }); - - // Return the raw key ONCE — it cannot be retrieved again - return reply.status(201).send({ - id, - key: rawKey, - name, - permissions: scopedPermissions, - expiresAt: expiresAt?.toISOString() ?? null, - createdAt: new Date().toISOString(), - }); - }); + }, + ); // GET /api/v1/api-keys — List user's API keys (never returns the key itself) - app.get("/api/v1/api-keys", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.get( + "/api/v1/api-keys", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const selectFields = { - id: schema.apiKeys.id, - name: schema.apiKeys.name, - permissions: schema.apiKeys.permissions, - createdAt: schema.apiKeys.createdAt, - lastUsedAt: schema.apiKeys.lastUsedAt, - expiresAt: schema.apiKeys.expiresAt, - }; - const keys = (await hasEffectivePermission(user, "apikeys:all")) - ? await db.select(selectFields).from(schema.apiKeys) - : await db - .select(selectFields) - .from(schema.apiKeys) - .where(eq(schema.apiKeys.userId, user.id)); + const selectFields = { + id: schema.apiKeys.id, + name: schema.apiKeys.name, + permissions: schema.apiKeys.permissions, + createdAt: schema.apiKeys.createdAt, + lastUsedAt: schema.apiKeys.lastUsedAt, + expiresAt: schema.apiKeys.expiresAt, + }; + const keys = (await hasEffectivePermission(user, "apikeys:all")) + ? await db.select(selectFields).from(schema.apiKeys) + : await db + .select(selectFields) + .from(schema.apiKeys) + .where(eq(schema.apiKeys.userId, user.id)); - return reply.send({ - apiKeys: keys.map((k) => ({ - id: k.id, - name: k.name, - permissions: k.permissions ?? null, - createdAt: k.createdAt.toISOString(), - lastUsedAt: k.lastUsedAt?.toISOString() ?? null, - expiresAt: k.expiresAt?.toISOString() ?? null, - })), - }); - }); + return reply.send({ + apiKeys: keys.map((k) => ({ + id: k.id, + name: k.name, + permissions: k.permissions ?? null, + createdAt: k.createdAt.toISOString(), + lastUsedAt: k.lastUsedAt?.toISOString() ?? null, + expiresAt: k.expiresAt?.toISOString() ?? null, + })), + }); + }, + ); // DELETE /api/v1/api-keys/:id — Delete an API key app.delete( "/api/v1/api-keys/:id", + { config: { rateLimit: { max: 30, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; diff --git a/apps/api/src/routes/enterprise/scim.ts b/apps/api/src/routes/enterprise/scim.ts index d0d720a3..66981bd0 100644 --- a/apps/api/src/routes/enterprise/scim.ts +++ b/apps/api/src/routes/enterprise/scim.ts @@ -305,84 +305,91 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // ── User Operations ──────────────────────────────────────────── // POST /api/v1/scim/v2/Users -- create user - app.post("/api/v1/scim/v2/Users", async (request: FastifyRequest, reply: FastifyReply) => { - if (!(await scimAuth(request, reply))) return; - if (!(await requireScimFeature(reply))) return; + app.post( + "/api/v1/scim/v2/Users", + { + config: { rateLimit: { max: 120, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + if (!(await scimAuth(request, reply))) return; + if (!(await requireScimFeature(reply))) return; - const body = request.body as Record; - const userName = body.userName as string | undefined; - const externalId = body.externalId as string | undefined; - const active = body.active !== false; // default true - const emails = body.emails as Array<{ value: string; primary?: boolean }> | undefined; - if (!userName) { - return reply.status(400).send(scimError(400, "userName is required")); - } + const body = request.body as Record; + const userName = body.userName as string | undefined; + const externalId = body.externalId as string | undefined; + const active = body.active !== false; // default true + const emails = body.emails as Array<{ value: string; primary?: boolean }> | undefined; + if (!userName) { + return reply.status(400).send(scimError(400, "userName is required")); + } - // Check for duplicate username - const [existing] = await db - .select() - .from(schema.users) - .where(eq(schema.users.username, userName)); + // Check for duplicate username + const [existing] = await db + .select() + .from(schema.users) + .where(eq(schema.users.username, userName)); - if (existing) { - return reply.status(409).send(scimError(409, "User already exists")); - } + if (existing) { + return reply.status(409).send(scimError(409, "User already exists")); + } - const id = randomUUID(); - const email = emails?.find((e) => e.primary)?.value ?? emails?.[0]?.value ?? null; + const id = randomUUID(); + const email = emails?.find((e) => e.primary)?.value ?? emails?.[0]?.value ?? null; - // Resolve default team - const [defaultTeam] = await db - .select() - .from(schema.teams) - .where(eq(schema.teams.name, "Default")); - const teamId = defaultTeam?.id ?? "default-team-00000000"; + // Resolve default team + const [defaultTeam] = await db + .select() + .from(schema.teams) + .where(eq(schema.teams.name, "Default")); + const teamId = defaultTeam?.id ?? "default-team-00000000"; - const now = new Date(); - await db.insert(schema.users).values({ - id, - username: userName, - email, - externalId: externalId ?? null, - role: active ? "user" : "disabled", - team: teamId, - authProvider: "scim", - mustChangePassword: false, - createdAt: now, - updatedAt: now, - }); - - await auditLog( - request.log, - "SCIM_USER_PROVISIONED", - { - userId: id, + const now = new Date(); + await db.insert(schema.users).values({ + id, username: userName, - externalId, - }, - request.ip, - request.id, - ); + email, + externalId: externalId ?? null, + role: active ? "user" : "disabled", + team: teamId, + authProvider: "scim", + mustChangePassword: false, + createdAt: now, + updatedAt: now, + }); - const user = { - id, - username: userName, - email, - externalId: externalId ?? null, - role: active ? "user" : "disabled", - team: teamId, - legalHold: false, - passwordHash: null, - createdAt: now, - updatedAt: now, - }; + await auditLog( + request.log, + "SCIM_USER_PROVISIONED", + { + userId: id, + username: userName, + externalId, + }, + request.ip, + request.id, + ); - return reply.status(201).send(toScimUser(user, defaultTeam?.name)); - }); + const user = { + id, + username: userName, + email, + externalId: externalId ?? null, + role: active ? "user" : "disabled", + team: teamId, + legalHold: false, + passwordHash: null, + createdAt: now, + updatedAt: now, + }; + + return reply.status(201).send(toScimUser(user, defaultTeam?.name)); + }, + ); // GET /api/v1/scim/v2/Users/:id -- get user app.get( "/api/v1/scim/v2/Users/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -406,6 +413,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // GET /api/v1/scim/v2/Users -- list users with filter app.get( "/api/v1/scim/v2/Users", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async ( request: FastifyRequest<{ Querystring: { filter?: string; startIndex?: string; count?: string }; @@ -477,6 +485,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // PUT /api/v1/scim/v2/Users/:id -- replace user app.put( "/api/v1/scim/v2/Users/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -555,6 +564,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // PATCH /api/v1/scim/v2/Users/:id -- partial update app.patch( "/api/v1/scim/v2/Users/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -661,6 +671,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // DELETE /api/v1/scim/v2/Users/:id -- deactivate user (soft delete) app.delete( "/api/v1/scim/v2/Users/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -706,74 +717,81 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // ── Group Operations ─────────────────────────────────────────── // POST /api/v1/scim/v2/Groups -- create team - app.post("/api/v1/scim/v2/Groups", async (request: FastifyRequest, reply: FastifyReply) => { - if (!(await scimAuth(request, reply))) return; - if (!(await requireScimFeature(reply))) return; + app.post( + "/api/v1/scim/v2/Groups", + { + config: { rateLimit: { max: 120, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + if (!(await scimAuth(request, reply))) return; + if (!(await requireScimFeature(reply))) return; - const body = request.body as Record; - const displayName = body.displayName as string | undefined; - const members = body.members as Array<{ value: string }> | undefined; + const body = request.body as Record; + const displayName = body.displayName as string | undefined; + const members = body.members as Array<{ value: string }> | undefined; - if (!displayName) { - return reply.status(400).send(scimError(400, "displayName is required")); - } - - // Check for duplicate team name - const [existing] = await db - .select() - .from(schema.teams) - .where(eq(schema.teams.name, displayName)); - - if (existing) { - return reply.status(409).send(scimError(409, "Group already exists")); - } - - const id = randomUUID(); - const now = new Date(); - - await db.insert(schema.teams).values({ - id, - name: displayName, - createdAt: now, - }); - - // Assign members to the team - if (members && members.length > 0) { - for (const member of members) { - await db - .update(schema.users) - .set({ team: id, updatedAt: new Date() }) - .where(eq(schema.users.id, member.value)); + if (!displayName) { + return reply.status(400).send(scimError(400, "displayName is required")); } - } - // Fetch actual members - const teamMembers = await db - .select({ id: schema.users.id, username: schema.users.username }) - .from(schema.users) - .where(eq(schema.users.team, id)); + // Check for duplicate team name + const [existing] = await db + .select() + .from(schema.teams) + .where(eq(schema.teams.name, displayName)); - await auditLog( - request.log, - "SCIM_GROUP_SYNCED", - { - teamId: id, - teamName: displayName, - action: "created", - memberCount: teamMembers.length, - }, - request.ip, - request.id, - ); + if (existing) { + return reply.status(409).send(scimError(409, "Group already exists")); + } - return reply - .status(201) - .send(toScimGroup({ id, name: displayName, createdAt: now }, teamMembers)); - }); + const id = randomUUID(); + const now = new Date(); + + await db.insert(schema.teams).values({ + id, + name: displayName, + createdAt: now, + }); + + // Assign members to the team + if (members && members.length > 0) { + for (const member of members) { + await db + .update(schema.users) + .set({ team: id, updatedAt: new Date() }) + .where(eq(schema.users.id, member.value)); + } + } + + // Fetch actual members + const teamMembers = await db + .select({ id: schema.users.id, username: schema.users.username }) + .from(schema.users) + .where(eq(schema.users.team, id)); + + await auditLog( + request.log, + "SCIM_GROUP_SYNCED", + { + teamId: id, + teamName: displayName, + action: "created", + memberCount: teamMembers.length, + }, + request.ip, + request.id, + ); + + return reply + .status(201) + .send(toScimGroup({ id, name: displayName, createdAt: now }, teamMembers)); + }, + ); // GET /api/v1/scim/v2/Groups/:id -- get team app.get( "/api/v1/scim/v2/Groups/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -797,6 +815,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // GET /api/v1/scim/v2/Groups -- list teams app.get( "/api/v1/scim/v2/Groups", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async ( request: FastifyRequest<{ Querystring: { filter?: string; startIndex?: string; count?: string }; @@ -862,6 +881,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // PUT /api/v1/scim/v2/Groups/:id -- replace team app.put( "/api/v1/scim/v2/Groups/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -939,6 +959,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // PATCH /api/v1/scim/v2/Groups/:id -- update members app.patch( "/api/v1/scim/v2/Groups/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; @@ -1048,6 +1069,7 @@ export async function registerScimRoutes(app: FastifyInstance): Promise { // DELETE /api/v1/scim/v2/Groups/:id -- delete team app.delete( "/api/v1/scim/v2/Groups/:id", + { config: { rateLimit: { max: 120, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { if (!(await scimAuth(request, reply))) return; if (!(await requireScimFeature(reply))) return; diff --git a/apps/api/src/routes/enterprise/upgrade.ts b/apps/api/src/routes/enterprise/upgrade.ts index be6b85bb..36827b07 100644 --- a/apps/api/src/routes/enterprise/upgrade.ts +++ b/apps/api/src/routes/enterprise/upgrade.ts @@ -102,70 +102,76 @@ export async function registerUpgradeRoutes(app: FastifyInstance): Promise ); // GET /api/v1/admin/upgrade-check - app.get("/api/v1/admin/upgrade-check", async (request: FastifyRequest, reply: FastifyReply) => { - const user = await requirePermission("system:health")(request, reply); - if (!user) return; - if (!(await requireUpgradeFeature(reply))) return; + app.get( + "/api/v1/admin/upgrade-check", + { + config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = await requirePermission("system:health")(request, reply); + if (!user) return; + if (!(await requireUpgradeFeature(reply))) return; - // Check database connectivity - let dbOk = false; - try { - await db.select().from(schema.settings).limit(1); - dbOk = true; - } catch { - /* db unreachable */ - } - - // Check Redis connectivity - let redisOk = false; - try { - redisOk = await pingRedis(); - } catch { - /* redis unreachable */ - } - - // Check in-flight jobs - let activeCount = 0; - let noInFlightJobs = true; - try { - const [row] = await db - .select({ count: sql`count(*)::int` }) - .from(schema.jobs) - .where(sql`${schema.jobs.status} IN ('queued', 'processing')`); - activeCount = row?.count ?? 0; - noInFlightJobs = activeCount === 0; - } catch { - // If we can't query, assume there may be jobs - noInFlightJobs = false; - } - - // Check disk space (> 1 GB free) - const MIN_FREE_BYTES = 1024 * 1024 * 1024; // 1 GB - let diskOk = true; - let freeGb = 0; - if (env.STORAGE_MODE !== "s3") { + // Check database connectivity + let dbOk = false; try { - const stats = await statfs(env.WORKSPACE_PATH); - const freeBytes = stats.bfree * stats.bsize; - freeGb = Math.round((freeBytes / (1024 * 1024 * 1024)) * 100) / 100; - diskOk = freeBytes > MIN_FREE_BYTES; + await db.select().from(schema.settings).limit(1); + dbOk = true; } catch { - // Path doesn't exist -- skip check + /* db unreachable */ } - } - const ready = diskOk && noInFlightJobs && dbOk && redisOk; + // Check Redis connectivity + let redisOk = false; + try { + redisOk = await pingRedis(); + } catch { + /* redis unreachable */ + } - return reply.send({ - ready, - checks: { - diskSpace: { ok: diskOk, freeGb }, - inFlightJobs: { ok: noInFlightJobs, activeCount }, - databaseConnected: { ok: dbOk }, - redisConnected: { ok: redisOk }, - }, - }); - }); + // Check in-flight jobs + let activeCount = 0; + let noInFlightJobs = true; + try { + const [row] = await db + .select({ count: sql`count(*)::int` }) + .from(schema.jobs) + .where(sql`${schema.jobs.status} IN ('queued', 'processing')`); + activeCount = row?.count ?? 0; + noInFlightJobs = activeCount === 0; + } catch { + // If we can't query, assume there may be jobs + noInFlightJobs = false; + } + + // Check disk space (> 1 GB free) + const MIN_FREE_BYTES = 1024 * 1024 * 1024; // 1 GB + let diskOk = true; + let freeGb = 0; + if (env.STORAGE_MODE !== "s3") { + try { + const stats = await statfs(env.WORKSPACE_PATH); + const freeBytes = stats.bfree * stats.bsize; + freeGb = Math.round((freeBytes / (1024 * 1024 * 1024)) * 100) / 100; + diskOk = freeBytes > MIN_FREE_BYTES; + } catch { + // Path doesn't exist -- skip check + } + } + + const ready = diskOk && noInFlightJobs && dbOk && redisOk; + + return reply.send({ + ready, + checks: { + diskSpace: { ok: diskOk, freeGb }, + inFlightJobs: { ok: noInFlightJobs, activeCount }, + databaseConnected: { ok: dbOk }, + redisConnected: { ok: redisOk }, + }, + }); + }, + ); app.log.info("Enterprise upgrade management routes registered"); } diff --git a/apps/api/src/routes/features.ts b/apps/api/src/routes/features.ts index 9f5b82ea..e4cfb95c 100644 --- a/apps/api/src/routes/features.ts +++ b/apps/api/src/routes/features.ts @@ -106,32 +106,37 @@ function getDirSize(dirPath: string): number { export async function registerFeatureRoutes(app: FastifyInstance): Promise { // GET /api/v1/features - List feature bundles and their statuses - app.get("/api/v1/features", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.get( + "/api/v1/features", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - // In non-Docker environments, all bundles are available natively - if (!isDockerEnvironment()) { - const bundles = Object.values(FEATURE_BUNDLES).map((bundle) => ({ - id: bundle.id, - name: bundle.name, - description: bundle.description, - status: "installed" as const, - installedVersion: null, - estimatedSize: bundle.estimatedSize, - enablesTools: bundle.enablesTools, - progress: null, - error: null, - })); - return reply.send({ bundles }); - } + // In non-Docker environments, all bundles are available natively + if (!isDockerEnvironment()) { + const bundles = Object.values(FEATURE_BUNDLES).map((bundle) => ({ + id: bundle.id, + name: bundle.name, + description: bundle.description, + status: "installed" as const, + installedVersion: null, + estimatedSize: bundle.estimatedSize, + enablesTools: bundle.enablesTools, + progress: null, + error: null, + })); + return reply.send({ bundles }); + } - return reply.send({ bundles: getFeatureStates() }); - }); + return reply.send({ bundles: getFeatureStates() }); + }, + ); // POST /api/v1/admin/features/:bundleId/install - Install a feature bundle app.post( "/api/v1/admin/features/:bundleId/install", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: BundleIdParams }>, reply: FastifyReply) => { const admin = await requirePermission("features:manage")(request, reply); if (!admin) return; @@ -283,6 +288,7 @@ export async function registerFeatureRoutes(app: FastifyInstance): Promise // POST /api/v1/admin/features/:bundleId/uninstall - Uninstall a feature bundle app.post( "/api/v1/admin/features/:bundleId/uninstall", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: BundleIdParams }>, reply: FastifyReply) => { const admin = await requirePermission("features:manage")(request, reply); if (!admin) return; diff --git a/apps/api/src/routes/file-preview.ts b/apps/api/src/routes/file-preview.ts index 46003668..f5faf846 100644 --- a/apps/api/src/routes/file-preview.ts +++ b/apps/api/src/routes/file-preview.ts @@ -61,6 +61,7 @@ const OFFICE_MIMES = new Set([ export async function filePreviewRoutes(app: FastifyInstance): Promise { app.get( "/api/v1/files/:id/preview", + { config: { rateLimit: { max: 60, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; @@ -219,141 +220,147 @@ export async function filePreviewRoutes(app: FastifyInstance): Promise { ); // ── On-demand preview for uploaded (non-stored) media files ───── - app.post("/api/v1/preview/generate", async (request: FastifyRequest, reply: FastifyReply) => { - // Optional auth -- the preview is for the user's own uploaded file - getAuthUser(request); + app.post( + "/api/v1/preview/generate", + { + config: { rateLimit: { max: 60, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + // Optional auth -- the preview is for the user's own uploaded file + getAuthUser(request); - const parts = request.parts(); - let fileBuffer: Buffer | null = null; - let filename = "input"; + const parts = request.parts(); + let fileBuffer: Buffer | null = null; + let filename = "input"; - for await (const part of parts) { - if (part.type !== "file") continue; - const chunks: Buffer[] = []; - for await (const chunk of part.file) { - chunks.push(chunk); - } - fileBuffer = Buffer.concat(chunks); - filename = part.filename ?? "input"; - break; // only process the first file - } - - if (!fileBuffer || fileBuffer.length === 0) { - return reply.status(400).send({ error: "No file provided" }); - } - - const ext = filename.split(".").pop()?.toLowerCase() ?? ""; - const videoExts = new Set([ - "avi", - "mkv", - "wmv", - "flv", - "mov", - "mpg", - "mpeg", - "m4v", - "3gp", - "3g2", - "ts", - "mts", - "m2ts", - "vob", - "divx", - "asf", - "rm", - "rmvb", - "f4v", - "ogv", - "mp4", - "webm", - "ogg", - ]); - const audioExts = new Set([ - "wav", - "flac", - "aac", - "wma", - "ogg", - "oga", - "opus", - "m4a", - "aiff", - "aif", - "amr", - "ape", - "ac3", - "dts", - "mp3", - ]); - - const isVideo = videoExts.has(ext); - const isAudio = audioExts.has(ext); - - if (!isVideo && !isAudio) { - return reply.status(400).send({ error: "Unsupported file type for preview" }); - } - - const id = randomUUID(); - const inputPath = join(tmpdir(), `snapotter-preview-${id}.${ext}`); - const outputExt = isVideo ? "mp4" : "mp3"; - const outputPath = join(tmpdir(), `snapotter-preview-${id}-out.${outputExt}`); - - try { - await writeFile(inputPath, fileBuffer); - - if (isVideo) { - await runFfmpeg([ - "-i", - inputPath, - "-t", - "30", - "-vf", - "scale='min(720,iw)':-2", - "-c:v", - "libx264", - "-preset", - "ultrafast", - "-crf", - "28", - "-c:a", - "aac", - "-b:a", - "128k", - "-movflags", - "+faststart", - "-y", - outputPath, - ]); - } else { - await runFfmpeg([ - "-i", - inputPath, - "-t", - "60", - "-c:a", - "libmp3lame", - "-b:a", - "128k", - "-y", - outputPath, - ]); + for await (const part of parts) { + if (part.type !== "file") continue; + const chunks: Buffer[] = []; + for await (const chunk of part.file) { + chunks.push(chunk); + } + fileBuffer = Buffer.concat(chunks); + filename = part.filename ?? "input"; + break; // only process the first file } - const outputBuffer = await readFile(outputPath); - const contentType = isVideo ? "video/mp4" : "audio/mpeg"; + if (!fileBuffer || fileBuffer.length === 0) { + return reply.status(400).send({ error: "No file provided" }); + } - return reply - .header("Content-Type", contentType) - .header("Content-Length", outputBuffer.length) - .send(outputBuffer); - } catch (err) { - request.log.error({ err, filename }, "On-demand preview generation failed"); - return reply.status(422).send({ error: "Could not generate preview" }); - } finally { - await rm(inputPath, { force: true }).catch(() => {}); - await rm(outputPath, { force: true }).catch(() => {}); - } - }); + const ext = filename.split(".").pop()?.toLowerCase() ?? ""; + const videoExts = new Set([ + "avi", + "mkv", + "wmv", + "flv", + "mov", + "mpg", + "mpeg", + "m4v", + "3gp", + "3g2", + "ts", + "mts", + "m2ts", + "vob", + "divx", + "asf", + "rm", + "rmvb", + "f4v", + "ogv", + "mp4", + "webm", + "ogg", + ]); + const audioExts = new Set([ + "wav", + "flac", + "aac", + "wma", + "ogg", + "oga", + "opus", + "m4a", + "aiff", + "aif", + "amr", + "ape", + "ac3", + "dts", + "mp3", + ]); + + const isVideo = videoExts.has(ext); + const isAudio = audioExts.has(ext); + + if (!isVideo && !isAudio) { + return reply.status(400).send({ error: "Unsupported file type for preview" }); + } + + const id = randomUUID(); + const inputPath = join(tmpdir(), `snapotter-preview-${id}.${ext}`); + const outputExt = isVideo ? "mp4" : "mp3"; + const outputPath = join(tmpdir(), `snapotter-preview-${id}-out.${outputExt}`); + + try { + await writeFile(inputPath, fileBuffer); + + if (isVideo) { + await runFfmpeg([ + "-i", + inputPath, + "-t", + "30", + "-vf", + "scale='min(720,iw)':-2", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-crf", + "28", + "-c:a", + "aac", + "-b:a", + "128k", + "-movflags", + "+faststart", + "-y", + outputPath, + ]); + } else { + await runFfmpeg([ + "-i", + inputPath, + "-t", + "60", + "-c:a", + "libmp3lame", + "-b:a", + "128k", + "-y", + outputPath, + ]); + } + + const outputBuffer = await readFile(outputPath); + const contentType = isVideo ? "video/mp4" : "audio/mpeg"; + + return reply + .header("Content-Type", contentType) + .header("Content-Length", outputBuffer.length) + .send(outputBuffer); + } catch (err) { + request.log.error({ err, filename }, "On-demand preview generation failed"); + return reply.status(422).send({ error: "Could not generate preview" }); + } finally { + await rm(inputPath, { force: true }).catch(() => {}); + await rm(outputPath, { force: true }).catch(() => {}); + } + }, + ); app.log.info("File preview routes registered"); } diff --git a/apps/api/src/routes/pipeline.ts b/apps/api/src/routes/pipeline.ts index e7af3a21..31988cbb 100644 --- a/apps/api/src/routes/pipeline.ts +++ b/apps/api/src/routes/pipeline.ts @@ -203,425 +203,437 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise { - let fileBuffer: Buffer | null = null; - let filename = "file"; - let pipelineRaw: string | null = null; - let clientJobId: string | null = null; + app.post( + "/api/v1/pipeline/execute", + { config: { rateLimit: { max: 60, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + let fileBuffer: Buffer | null = null; + let filename = "file"; + let pipelineRaw: string | null = null; + let clientJobId: string | null = null; - // Parse multipart - try { - const parts = request.parts(); - for await (const part of parts) { - if (part.type === "file") { - const chunks: Buffer[] = []; - for await (const chunk of part.file) { - chunks.push(chunk); - } - fileBuffer = Buffer.concat(chunks); - filename = sanitizeFilename(part.filename ?? "file"); - } else if (part.fieldname === "pipeline") { - pipelineRaw = part.value as string; - } else if (part.fieldname === "clientJobId") { - const raw = part.value as string; - if (typeof raw === "string" && raw.length > 0 && raw.length <= 128) { - clientJobId = raw; - } - } - } - } catch (err) { - return reply.status(400).send({ - error: "Failed to parse multipart request", - details: err instanceof Error ? err.message : String(err), - }); - } - - if (!fileBuffer || fileBuffer.length === 0) { - return reply.status(400).send({ error: "No file provided" }); - } - - // The first pipeline step determines the input modality, so non-image - // inputs (audio/video/document) get validated by the right handler instead - // of always being forced through image validation/decoding. - let firstToolId: string | undefined; - try { - firstToolId = (JSON.parse(pipelineRaw ?? "{}") as { steps?: Array<{ toolId?: string }> }) - ?.steps?.[0]?.toolId; - } catch { - // Malformed pipeline JSON is reported when the definition is parsed below. - } - const inputModality = TOOLS.find((t) => t.id === firstToolId)?.modality ?? "image"; - const pipelineScratch = join(tmpdir(), "snapotter-scratch", `pipeline-${randomUUID()}`); - await mkdir(pipelineScratch, { recursive: true }); - - if (inputModality === "image") { - // Validate the initial image - const validation = await validateImageBuffer(fileBuffer, filename); - if (!validation.valid) { - return reply.status(400).send({ - error: `Invalid image: ${validation.reason}`, - }); - } - - // Decode HEIC/HEIF input via system heif-dec - if (validation.format === "heif") { - try { - fileBuffer = await decodeHeic(fileBuffer); - const ext = filename.match(/\.[^.]+$/)?.[0]; - if (ext) filename = `${filename.slice(0, -ext.length)}.png`; - } catch (err) { - return reply.status(422).send({ - error: "Failed to decode HEIC file. Ensure libheif-examples is installed.", - details: err instanceof Error ? err.message : String(err), - }); - } - } - - // Decode CLI-decoded formats (RAW, TGA, PSD, EXR, HDR) - if (needsCliDecode(validation.format)) { - try { - const fileExt = filename.split(".").pop()?.toLowerCase(); - fileBuffer = await decodeToSharpCompat(fileBuffer, validation.format, fileExt); - const ext = filename.match(/\.[^.]+$/)?.[0]; - if (ext) filename = `${filename.slice(0, -ext.length)}.png`; - } catch (err) { - return reply.status(422).send({ - error: `Failed to decode ${validation.format} file`, - details: err instanceof Error ? err.message : String(err), - }); - } - } - - // Sanitize SVG input and normalize EXIF orientation - const isSvg = isSvgBuffer(fileBuffer); - if (isSvg) { - fileBuffer = sanitizeSvg(fileBuffer); - } else { - fileBuffer = await autoOrient(fileBuffer); - } - } else { - // Non-image input: validate/decode via the tool's modality handler. + // Parse multipart try { - const prepared = await inputHandlerFor(inputModality).prepare(fileBuffer, filename, { - scratchDir: pipelineScratch, - lenient: getToolConfig(firstToolId ?? "")?.skipStructuralValidation, - }); - fileBuffer = prepared.buffer; - filename = prepared.filename; - } catch (err) { - if (err instanceof InputValidationError) { - const body: Record = { error: err.message }; - if (err.details) body.details = err.details; - return reply.status(err.statusCode).send(body); + const parts = request.parts(); + for await (const part of parts) { + if (part.type === "file") { + const chunks: Buffer[] = []; + for await (const chunk of part.file) { + chunks.push(chunk); + } + fileBuffer = Buffer.concat(chunks); + filename = sanitizeFilename(part.filename ?? "file"); + } else if (part.fieldname === "pipeline") { + pipelineRaw = part.value as string; + } else if (part.fieldname === "clientJobId") { + const raw = part.value as string; + if (typeof raw === "string" && raw.length > 0 && raw.length <= 128) { + clientJobId = raw; + } + } } - throw err; - } - } - - // Parse and validate the pipeline definition - if (!pipelineRaw) { - return reply.status(400).send({ error: "No pipeline definition provided" }); - } - - let pipeline: z.infer; - try { - const parsed = JSON.parse(pipelineRaw); - const result = pipelineDefinitionSchema.safeParse(parsed); - if (!result.success) { + } catch (err) { return reply.status(400).send({ - error: "Invalid pipeline definition", - details: formatZodErrors(result.error.issues), - }); - } - pipeline = result.data; - } catch { - return reply.status(400).send({ error: "Pipeline must be valid JSON" }); - } - - // Validate all tool IDs and settings; collect parsed steps - const parsedSteps: ParsedStep[] = []; - - for (let i = 0; i < pipeline.steps.length; i++) { - const step = pipeline.steps[i]; - - // Route content-aware resize to its dedicated tool - const resolvedToolId = - step.toolId === "resize" && step.settings?.contentAware - ? "content-aware-resize" - : step.toolId; - - const toolConfig = getToolConfig(resolvedToolId); - if (!toolConfig) { - return reply.status(400).send({ - error: `Step ${i + 1} (${step.toolId}): Tool not found or not available`, + error: "Failed to parse multipart request", + details: err instanceof Error ? err.message : String(err), }); } - // Guard: check if the tool's AI feature bundle is installed - if (!isToolInstalled(resolvedToolId)) { - const bundle = getBundleForTool(resolvedToolId); - return reply.status(501).send({ - error: `Step ${i + 1} (${step.toolId}): Feature "${bundle?.name}" is not installed`, - code: "FEATURE_NOT_INSTALLED", - feature: TOOL_BUNDLE_MAP[resolvedToolId], - featureName: bundle?.name ?? resolvedToolId, - }); + if (!fileBuffer || fileBuffer.length === 0) { + return reply.status(400).send({ error: "No file provided" }); } - if (PASSWORD_TOOLS.has(step.toolId)) { - return reply.status(400).send({ - error: `Step ${i + 1}: This tool cannot be used in pipelines because it requires a password`, - }); + // The first pipeline step determines the input modality, so non-image + // inputs (audio/video/document) get validated by the right handler instead + // of always being forced through image validation/decoding. + let firstToolId: string | undefined; + try { + firstToolId = (JSON.parse(pipelineRaw ?? "{}") as { steps?: Array<{ toolId?: string }> }) + ?.steps?.[0]?.toolId; + } catch { + // Malformed pipeline JSON is reported when the definition is parsed below. } + const inputModality = TOOLS.find((t) => t.id === firstToolId)?.modality ?? "image"; + const pipelineScratch = join(tmpdir(), "snapotter-scratch", `pipeline-${randomUUID()}`); + await mkdir(pipelineScratch, { recursive: true }); - const settingsResult = toolConfig.settingsSchema.safeParse(step.settings); - if (!settingsResult.success) { - return reply.status(400).send({ - error: `Step ${i + 1} (${step.toolId}): Invalid settings`, - details: settingsResult.error.issues.map( - (iss: { path: (string | number)[]; message: string }) => ({ - path: iss.path.join("."), - message: iss.message, - }), - ), - }); - } - - if (env.MAX_PIPELINE_STEP_PIXELS > 0) { - const s = settingsResult.data as Record; - const w = Number(s.width) || 0; - const h = Number(s.height) || 0; - if (w > 0 && h > 0 && w * h > env.MAX_PIPELINE_STEP_PIXELS) { + if (inputModality === "image") { + // Validate the initial image + const validation = await validateImageBuffer(fileBuffer, filename); + if (!validation.valid) { return reply.status(400).send({ - error: `Step ${i + 1} (${step.toolId}): Output dimensions ${w}x${h} exceed per-step pixel limit`, + error: `Invalid image: ${validation.reason}`, }); } + + // Decode HEIC/HEIF input via system heif-dec + if (validation.format === "heif") { + try { + fileBuffer = await decodeHeic(fileBuffer); + const ext = filename.match(/\.[^.]+$/)?.[0]; + if (ext) filename = `${filename.slice(0, -ext.length)}.png`; + } catch (err) { + return reply.status(422).send({ + error: "Failed to decode HEIC file. Ensure libheif-examples is installed.", + details: err instanceof Error ? err.message : String(err), + }); + } + } + + // Decode CLI-decoded formats (RAW, TGA, PSD, EXR, HDR) + if (needsCliDecode(validation.format)) { + try { + const fileExt = filename.split(".").pop()?.toLowerCase(); + fileBuffer = await decodeToSharpCompat(fileBuffer, validation.format, fileExt); + const ext = filename.match(/\.[^.]+$/)?.[0]; + if (ext) filename = `${filename.slice(0, -ext.length)}.png`; + } catch (err) { + return reply.status(422).send({ + error: `Failed to decode ${validation.format} file`, + details: err instanceof Error ? err.message : String(err), + }); + } + } + + // Sanitize SVG input and normalize EXIF orientation + const isSvg = isSvgBuffer(fileBuffer); + if (isSvg) { + fileBuffer = sanitizeSvg(fileBuffer); + } else { + fileBuffer = await autoOrient(fileBuffer); + } + } else { + // Non-image input: validate/decode via the tool's modality handler. + try { + const prepared = await inputHandlerFor(inputModality).prepare(fileBuffer, filename, { + scratchDir: pipelineScratch, + lenient: getToolConfig(firstToolId ?? "")?.skipStructuralValidation, + }); + fileBuffer = prepared.buffer; + filename = prepared.filename; + } catch (err) { + if (err instanceof InputValidationError) { + const body: Record = { error: err.message }; + if (err.details) body.details = err.details; + return reply.status(err.statusCode).send(body); + } + throw err; + } } - parsedSteps.push({ - toolId: step.toolId, - resolvedToolId, - parsedSettings: settingsResult.data, - pool: resolveToolPool(resolvedToolId), - }); - } + // Parse and validate the pipeline definition + if (!pipelineRaw) { + return reply.status(400).send({ error: "No pipeline definition provided" }); + } - // ── Enqueue as a BullMQ flow ──────────────────────────────── + let pipeline: z.infer; + try { + const parsed = JSON.parse(pipelineRaw); + const result = pipelineDefinitionSchema.safeParse(parsed); + if (!result.success) { + return reply.status(400).send({ + error: "Invalid pipeline definition", + details: formatZodErrors(result.error.issues), + }); + } + pipeline = result.data; + } catch { + return reply.status(400).send({ error: "Pipeline must be valid JSON" }); + } - const startTime = Date.now(); - const jobId = randomUUID(); - const userId = getAuthUser(request)?.id ?? null; - const originalSize = fileBuffer.length; + // Validate all tool IDs and settings; collect parsed steps + const parsedSteps: ParsedStep[] = []; - // Upload decoded file to object storage - const uploadKey = `uploads/${jobId}/${filename}`; - await putObject(uploadKey, fileBuffer); + for (let i = 0; i < pipeline.steps.length; i++) { + const step = pipeline.steps[i]; - // Report initial progress - if (clientJobId) { - updateSingleFileProgress({ - jobId: clientJobId, - phase: "processing", - percent: 0, - stage: "Preparing pipeline...", - }); - } + // Route content-aware resize to its dedicated tool + const resolvedToolId = + step.toolId === "resize" && step.settings?.contentAware + ? "content-aware-resize" + : step.toolId; - // Build the nested FlowJob tree - const { tree, stepJobIds } = buildPipelineFlowTree({ - jobId, - userId, - parsedSteps, - uploadKey, - filename, - clientJobId: clientJobId ?? jobId, - }); + const toolConfig = getToolConfig(resolvedToolId); + if (!toolConfig) { + return reply.status(400).send({ + error: `Step ${i + 1} (${step.toolId}): Tool not found or not available`, + }); + } - // Insert all durable rows before adding the flow. enqueueToolJob - // inserts row-then-add; for flows we insert ALL rows first, then - // one flow.add. - for (let i = 0; i < parsedSteps.length; i++) { - await db.insert(schema.jobs).values({ - id: stepJobIds[i], - userId, - toolId: parsedSteps[i].resolvedToolId, - pool: parsedSteps[i].pool, - type: "pipeline-step", - status: "queued", - inputRefs: i === 0 ? [uploadKey] : [], - settings: parsedSteps[i].parsedSettings as Record, - }); - } + // Guard: check if the tool's AI feature bundle is installed + if (!isToolInstalled(resolvedToolId)) { + const bundle = getBundleForTool(resolvedToolId); + return reply.status(501).send({ + error: `Step ${i + 1} (${step.toolId}): Feature "${bundle?.name}" is not installed`, + code: "FEATURE_NOT_INSTALLED", + feature: TOOL_BUNDLE_MAP[resolvedToolId], + featureName: bundle?.name ?? resolvedToolId, + }); + } - await db.insert(schema.jobs).values({ - id: jobId, - userId, - toolId: "pipeline", - pool: "image", - type: "pipeline", - status: "queued", - inputRefs: [], - settings: {}, - }); + if (PASSWORD_TOOLS.has(step.toolId)) { + return reply.status(400).send({ + error: `Step ${i + 1}: This tool cannot be used in pipelines because it requires a password`, + }); + } - // Inject OTel trace context into every node of the flow tree - injectTraceContextIntoFlow(tree); + const settingsResult = toolConfig.settingsSchema.safeParse(step.settings); + if (!settingsResult.success) { + return reply.status(400).send({ + error: `Step ${i + 1} (${step.toolId}): Invalid settings`, + details: settingsResult.error.issues.map( + (iss: { path: (string | number)[]; message: string }) => ({ + path: iss.path.join("."), + message: iss.message, + }), + ), + }); + } - // Add the flow to BullMQ - await getFlowProducer().add(tree); + if (env.MAX_PIPELINE_STEP_PIXELS > 0) { + const s = settingsResult.data as Record; + const w = Number(s.width) || 0; + const h = Number(s.height) || 0; + if (w > 0 && h > 0 && w * h > env.MAX_PIPELINE_STEP_PIXELS) { + return reply.status(400).send({ + error: `Step ${i + 1} (${step.toolId}): Output dimensions ${w}x${h} exceed per-step pixel limit`, + }); + } + } - // Wait for the finalize job (pipelines block to completion) - try { - const result = await waitForJob("image", jobId, 10 * 60_000); - - if (!result) { - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: false, - duration_ms: Date.now() - startTime, - status: "failed", - }); - return reply.status(422).send({ - error: "Pipeline processing timed out", + parsedSteps.push({ + toolId: step.toolId, + resolvedToolId, + parsedSettings: settingsResult.data, + pool: resolveToolPool(resolvedToolId), }); } - // Check for step failure reported by the finalize handler - if (result.resultPayload?.error) { - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: false, - duration_ms: Date.now() - startTime, - status: "failed", - }); - return reply.status(422).send({ - error: result.resultPayload.error as string, - completedSteps: result.resultPayload.steps, + // ── Enqueue as a BullMQ flow ──────────────────────────────── + + const startTime = Date.now(); + const jobId = randomUUID(); + const userId = getAuthUser(request)?.id ?? null; + const originalSize = fileBuffer.length; + + // Upload decoded file to object storage + const uploadKey = `uploads/${jobId}/${filename}`; + await putObject(uploadKey, fileBuffer); + + // Report initial progress + if (clientJobId) { + updateSingleFileProgress({ + jobId: clientJobId, + phase: "processing", + percent: 0, + stage: "Preparing pipeline...", }); } - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: false, - duration_ms: Date.now() - startTime, - status: "completed", - }); - - return reply.send({ + // Build the nested FlowJob tree + const { tree, stepJobIds } = buildPipelineFlowTree({ jobId, - downloadUrl: `/api/v1/download/${jobId}/${encodeURIComponent(result.filename)}`, - previewUrl: result.previewRef - ? `/api/v1/download/${jobId}/${result.previewRef.split("/").pop()}` - : undefined, - originalSize, - processedSize: result.processedSize, - stepsCompleted: result.resultPayload?.stepsCompleted ?? parsedSteps.length, - steps: result.resultPayload?.steps ?? [], + userId, + parsedSteps, + uploadKey, + filename, + clientJobId: clientJobId ?? jobId, }); - } catch (err) { - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: false, - duration_ms: Date.now() - startTime, - status: "failed", + + // Insert all durable rows before adding the flow. enqueueToolJob + // inserts row-then-add; for flows we insert ALL rows first, then + // one flow.add. + for (let i = 0; i < parsedSteps.length; i++) { + await db.insert(schema.jobs).values({ + id: stepJobIds[i], + userId, + toolId: parsedSteps[i].resolvedToolId, + pool: parsedSteps[i].pool, + type: "pipeline-step", + status: "queued", + inputRefs: i === 0 ? [uploadKey] : [], + settings: parsedSteps[i].parsedSettings as Record, + }); + } + + await db.insert(schema.jobs).values({ + id: jobId, + userId, + toolId: "pipeline", + pool: "image", + type: "pipeline", + status: "queued", + inputRefs: [], + settings: {}, }); - return reply.status(422).send({ - error: err instanceof Error ? err.message : "Pipeline processing failed", - }); - } - }); + + // Inject OTel trace context into every node of the flow tree + injectTraceContextIntoFlow(tree); + + // Add the flow to BullMQ + await getFlowProducer().add(tree); + + // Wait for the finalize job (pipelines block to completion) + try { + const result = await waitForJob("image", jobId, 10 * 60_000); + + if (!result) { + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: false, + duration_ms: Date.now() - startTime, + status: "failed", + }); + return reply.status(422).send({ + error: "Pipeline processing timed out", + }); + } + + // Check for step failure reported by the finalize handler + if (result.resultPayload?.error) { + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: false, + duration_ms: Date.now() - startTime, + status: "failed", + }); + return reply.status(422).send({ + error: result.resultPayload.error as string, + completedSteps: result.resultPayload.steps, + }); + } + + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: false, + duration_ms: Date.now() - startTime, + status: "completed", + }); + + return reply.send({ + jobId, + downloadUrl: `/api/v1/download/${jobId}/${encodeURIComponent(result.filename)}`, + previewUrl: result.previewRef + ? `/api/v1/download/${jobId}/${result.previewRef.split("/").pop()}` + : undefined, + originalSize, + processedSize: result.processedSize, + stepsCompleted: result.resultPayload?.stepsCompleted ?? parsedSteps.length, + steps: result.resultPayload?.steps ?? [], + }); + } catch (err) { + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: false, + duration_ms: Date.now() - startTime, + status: "failed", + }); + return reply.status(422).send({ + error: err instanceof Error ? err.message : "Pipeline processing failed", + }); + } + }, + ); /** * POST /api/v1/pipeline/save * * Save a named pipeline definition for later reuse. */ - app.post("/api/v1/pipeline/save", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.post( + "/api/v1/pipeline/save", + { config: { rateLimit: { max: 60, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const body = request.body as unknown; - const result = savePipelineSchema.safeParse(body); + const body = request.body as unknown; + const result = savePipelineSchema.safeParse(body); - if (!result.success) { - return reply.status(400).send({ - error: "Invalid pipeline definition", - details: result.error.issues.map((i) => ({ - path: i.path.join("."), - message: i.message, - })), - }); - } - - const { name, description, steps } = result.data; - - // Validate all tool IDs exist - for (let i = 0; i < steps.length; i++) { - if (PASSWORD_TOOLS.has(steps[i].toolId)) { + if (!result.success) { return reply.status(400).send({ - error: `This tool cannot be used in saved pipelines because it requires a password`, + error: "Invalid pipeline definition", + details: result.error.issues.map((i) => ({ + path: i.path.join("."), + message: i.message, + })), }); } - const toolConfig = getToolConfig(steps[i].toolId); - if (!toolConfig) { - return reply.status(400).send({ - error: `Step ${i + 1}: Tool "${steps[i].toolId}" not found`, - }); + + const { name, description, steps } = result.data; + + // Validate all tool IDs exist + for (let i = 0; i < steps.length; i++) { + if (PASSWORD_TOOLS.has(steps[i].toolId)) { + return reply.status(400).send({ + error: `This tool cannot be used in saved pipelines because it requires a password`, + }); + } + const toolConfig = getToolConfig(steps[i].toolId); + if (!toolConfig) { + return reply.status(400).send({ + error: `Step ${i + 1}: Tool "${steps[i].toolId}" not found`, + }); + } } - } - const id = randomUUID(); + const id = randomUUID(); - try { - await db.insert(schema.pipelines).values({ + try { + await db.insert(schema.pipelines).values({ + id, + userId: user.id, + name, + description: description ?? null, + steps, + }); + } catch { + return reply.status(409).send({ error: "Failed to save pipeline" }); + } + + return reply.status(201).send({ id, - userId: user.id, name, description: description ?? null, steps, + createdAt: new Date().toISOString(), }); - } catch { - return reply.status(409).send({ error: "Failed to save pipeline" }); - } - - return reply.status(201).send({ - id, - name, - description: description ?? null, - steps, - createdAt: new Date().toISOString(), - }); - }); + }, + ); /** * GET /api/v1/pipeline/list * * List all saved pipelines. */ - app.get("/api/v1/pipeline/list", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.get( + "/api/v1/pipeline/list", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - // Admins see all pipelines; regular users see their own + legacy (no owner) - const allRows = await db.select().from(schema.pipelines); - const rows = (await hasEffectivePermission(user, "pipelines:all")) - ? allRows - : allRows.filter((row) => !row.userId || row.userId === user.id); + // Admins see all pipelines; regular users see their own + legacy (no owner) + const allRows = await db.select().from(schema.pipelines); + const rows = (await hasEffectivePermission(user, "pipelines:all")) + ? allRows + : allRows.filter((row) => !row.userId || row.userId === user.id); - const pipelines = rows.map((row) => ({ - id: row.id, - name: row.name, - description: row.description, - steps: row.steps, - createdAt: row.createdAt.toISOString(), - })); + const pipelines = rows.map((row) => ({ + id: row.id, + name: row.name, + description: row.description, + steps: row.steps, + createdAt: row.createdAt.toISOString(), + })); - return reply.send({ pipelines }); - }); + return reply.send({ pipelines }); + }, + ); /** * DELETE /api/v1/pipeline/:id @@ -630,6 +642,7 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; @@ -678,503 +691,513 @@ export async function registerPipelineRoutes(app: FastifyInstance): Promise { - // ── Parse multipart ────────────────────────────────────────────── - interface ParsedFile { - buffer: Buffer; - filename: string; - } + app.post( + "/api/v1/pipeline/batch", + { + config: { rateLimit: { max: 20, timeWindow: "1 minute" } }, + }, + async (request: FastifyRequest, reply: FastifyReply) => { + // ── Parse multipart ────────────────────────────────────────────── + interface ParsedFile { + buffer: Buffer; + filename: string; + } - const files: ParsedFile[] = []; - let pipelineRaw: string | null = null; - let clientJobId: string | null = null; + const files: ParsedFile[] = []; + let pipelineRaw: string | null = null; + let clientJobId: string | null = null; - try { - const parts = request.parts(); - for await (const part of parts) { - if (part.type === "file") { - const chunks: Buffer[] = []; - for await (const chunk of part.file) { - chunks.push(chunk); + try { + const parts = request.parts(); + for await (const part of parts) { + if (part.type === "file") { + const chunks: Buffer[] = []; + for await (const chunk of part.file) { + chunks.push(chunk); + } + const buffer = Buffer.concat(chunks); + if (buffer.length > 0) { + files.push({ + buffer, + filename: sanitizeFilename(part.filename ?? "file"), + }); + } + } else if (part.fieldname === "pipeline") { + pipelineRaw = part.value as string; + } else if (part.fieldname === "clientJobId") { + const raw = part.value as string; + if (typeof raw === "string" && raw.length > 0 && raw.length <= 128) { + clientJobId = raw; + } } - const buffer = Buffer.concat(chunks); - if (buffer.length > 0) { - files.push({ - buffer, - filename: sanitizeFilename(part.filename ?? "file"), + } + } catch (err) { + return reply.status(400).send({ + error: "Failed to parse multipart request", + details: err instanceof Error ? err.message : String(err), + }); + } + + if (files.length === 0) { + return reply.status(400).send({ error: "No files provided" }); + } + + // Enforce batch size limit + if (env.MAX_BATCH_SIZE > 0 && files.length > env.MAX_BATCH_SIZE) { + return reply.status(400).send({ + error: `Too many files. Maximum batch size is ${env.MAX_BATCH_SIZE}`, + }); + } + + // ── Parse and validate pipeline definition ─────────────────────── + if (!pipelineRaw) { + return reply.status(400).send({ error: "No pipeline definition provided" }); + } + + let pipeline: z.infer; + try { + const parsed = JSON.parse(pipelineRaw); + const result = pipelineDefinitionSchema.safeParse(parsed); + if (!result.success) { + return reply.status(400).send({ + error: "Invalid pipeline definition", + details: formatZodErrors(result.error.issues), + }); + } + pipeline = result.data; + } catch { + return reply.status(400).send({ error: "Pipeline must be valid JSON" }); + } + + // Validate all tool IDs and settings + const parsedSteps: ParsedStep[] = []; + + for (let i = 0; i < pipeline.steps.length; i++) { + const step = pipeline.steps[i]; + + const resolvedToolId = + step.toolId === "resize" && step.settings?.contentAware + ? "content-aware-resize" + : step.toolId; + + const toolConfig = getToolConfig(resolvedToolId); + if (!toolConfig) { + return reply.status(400).send({ + error: `Step ${i + 1}: Tool "${step.toolId}" not found`, + }); + } + + if (!isToolInstalled(resolvedToolId)) { + const bundle = getBundleForTool(resolvedToolId); + return reply.status(501).send({ + error: `Step ${i + 1} (${step.toolId}): Feature "${bundle?.name}" is not installed`, + code: "FEATURE_NOT_INSTALLED", + feature: TOOL_BUNDLE_MAP[resolvedToolId], + featureName: bundle?.name ?? resolvedToolId, + }); + } + + if (PASSWORD_TOOLS.has(step.toolId)) { + return reply.status(400).send({ + error: `Step ${i + 1}: This tool cannot be used in pipelines because it requires a password`, + }); + } + + const settingsResult = toolConfig.settingsSchema.safeParse(step.settings); + if (!settingsResult.success) { + return reply.status(400).send({ + error: `Step ${i + 1} (${step.toolId}): Invalid settings`, + details: settingsResult.error.issues.map( + (iss: { path: (string | number)[]; message: string }) => ({ + path: iss.path.join("."), + message: iss.message, + }), + ), + }); + } + + if (env.MAX_PIPELINE_STEP_PIXELS > 0) { + const s = settingsResult.data as Record; + const w = Number(s.width) || 0; + const h = Number(s.height) || 0; + if (w > 0 && h > 0 && w * h > env.MAX_PIPELINE_STEP_PIXELS) { + return reply.status(400).send({ + error: `Step ${i + 1} (${step.toolId}): Output dimensions ${w}x${h} exceed per-step pixel limit`, }); } - } else if (part.fieldname === "pipeline") { - pipelineRaw = part.value as string; - } else if (part.fieldname === "clientJobId") { - const raw = part.value as string; - if (typeof raw === "string" && raw.length > 0 && raw.length <= 128) { - clientJobId = raw; - } - } - } - } catch (err) { - return reply.status(400).send({ - error: "Failed to parse multipart request", - details: err instanceof Error ? err.message : String(err), - }); - } - - if (files.length === 0) { - return reply.status(400).send({ error: "No files provided" }); - } - - // Enforce batch size limit - if (env.MAX_BATCH_SIZE > 0 && files.length > env.MAX_BATCH_SIZE) { - return reply.status(400).send({ - error: `Too many files. Maximum batch size is ${env.MAX_BATCH_SIZE}`, - }); - } - - // ── Parse and validate pipeline definition ─────────────────────── - if (!pipelineRaw) { - return reply.status(400).send({ error: "No pipeline definition provided" }); - } - - let pipeline: z.infer; - try { - const parsed = JSON.parse(pipelineRaw); - const result = pipelineDefinitionSchema.safeParse(parsed); - if (!result.success) { - return reply.status(400).send({ - error: "Invalid pipeline definition", - details: formatZodErrors(result.error.issues), - }); - } - pipeline = result.data; - } catch { - return reply.status(400).send({ error: "Pipeline must be valid JSON" }); - } - - // Validate all tool IDs and settings - const parsedSteps: ParsedStep[] = []; - - for (let i = 0; i < pipeline.steps.length; i++) { - const step = pipeline.steps[i]; - - const resolvedToolId = - step.toolId === "resize" && step.settings?.contentAware - ? "content-aware-resize" - : step.toolId; - - const toolConfig = getToolConfig(resolvedToolId); - if (!toolConfig) { - return reply.status(400).send({ - error: `Step ${i + 1}: Tool "${step.toolId}" not found`, - }); - } - - if (!isToolInstalled(resolvedToolId)) { - const bundle = getBundleForTool(resolvedToolId); - return reply.status(501).send({ - error: `Step ${i + 1} (${step.toolId}): Feature "${bundle?.name}" is not installed`, - code: "FEATURE_NOT_INSTALLED", - feature: TOOL_BUNDLE_MAP[resolvedToolId], - featureName: bundle?.name ?? resolvedToolId, - }); - } - - if (PASSWORD_TOOLS.has(step.toolId)) { - return reply.status(400).send({ - error: `Step ${i + 1}: This tool cannot be used in pipelines because it requires a password`, - }); - } - - const settingsResult = toolConfig.settingsSchema.safeParse(step.settings); - if (!settingsResult.success) { - return reply.status(400).send({ - error: `Step ${i + 1} (${step.toolId}): Invalid settings`, - details: settingsResult.error.issues.map( - (iss: { path: (string | number)[]; message: string }) => ({ - path: iss.path.join("."), - message: iss.message, - }), - ), - }); - } - - if (env.MAX_PIPELINE_STEP_PIXELS > 0) { - const s = settingsResult.data as Record; - const w = Number(s.width) || 0; - const h = Number(s.height) || 0; - if (w > 0 && h > 0 && w * h > env.MAX_PIPELINE_STEP_PIXELS) { - return reply.status(400).send({ - error: `Step ${i + 1} (${step.toolId}): Output dimensions ${w}x${h} exceed per-step pixel limit`, - }); - } - } - - parsedSteps.push({ - toolId: step.toolId, - resolvedToolId, - parsedSettings: settingsResult.data, - pool: resolveToolPool(resolvedToolId), - }); - } - - // ── Prepare files and build flow ───────────────────────────────── - const batchStartTime = Date.now(); - const parentId = clientJobId || randomUUID(); - const userId = getAuthUser(request)?.id ?? null; - - // Insert batch-finalize row BEFORE updateJobProgress to avoid - // a duplicate-key race with the progress persist layer. - await db.insert(schema.jobs).values({ - id: parentId, - userId, - toolId: "pipeline-batch", - pool: "system", - type: "batch", - status: "queued", - inputRefs: [], - settings: { flowChildCount: 0 }, - }); - - // Emit initial batch progress - updateJobProgress({ - jobId: parentId, - status: "processing", - totalFiles: files.length, - completedFiles: 0, - failedFiles: 0, - errors: [], - }); - - // Validate, decode, and upload each file; build per-file pipeline chains - const perFileChildren: FlowJob[] = []; - const preFailures: Array<{ originalIndex: number; filename: string; error: string }> = []; - let flowChildIndex = 0; - - // The first step's modality drives input validation for every file, so - // audio, video, and document pipelines are not rejected by the image - // validator. - const batchModality = - TOOLS.find((t) => t.id === pipeline.steps[0]?.toolId)?.modality ?? "image"; - const pipelineBatchScratch = join(tmpdir(), "snapotter-scratch", `pipeline-batch-${parentId}`); - await mkdir(pipelineBatchScratch, { recursive: true }); - - for (let fi = 0; fi < files.length; fi++) { - const file = files[fi]; - let processBuffer = file.buffer; - let processFilename = file.filename; - - if (batchModality === "image") { - const fileValidation = await validateImageBuffer(processBuffer, processFilename); - if (!fileValidation.valid) { - preFailures.push({ - originalIndex: fi, - filename: file.filename, - error: `Invalid image: ${fileValidation.reason}`, - }); - continue; } - // Decode chain - if (fileValidation.format === "heif") { - try { - processBuffer = await decodeHeic(processBuffer); - const ext = processFilename.match(/\.[^.]+$/)?.[0]; - if (ext) processFilename = `${processFilename.slice(0, -ext.length)}.png`; - } catch { + parsedSteps.push({ + toolId: step.toolId, + resolvedToolId, + parsedSettings: settingsResult.data, + pool: resolveToolPool(resolvedToolId), + }); + } + + // ── Prepare files and build flow ───────────────────────────────── + const batchStartTime = Date.now(); + const parentId = clientJobId || randomUUID(); + const userId = getAuthUser(request)?.id ?? null; + + // Insert batch-finalize row BEFORE updateJobProgress to avoid + // a duplicate-key race with the progress persist layer. + await db.insert(schema.jobs).values({ + id: parentId, + userId, + toolId: "pipeline-batch", + pool: "system", + type: "batch", + status: "queued", + inputRefs: [], + settings: { flowChildCount: 0 }, + }); + + // Emit initial batch progress + updateJobProgress({ + jobId: parentId, + status: "processing", + totalFiles: files.length, + completedFiles: 0, + failedFiles: 0, + errors: [], + }); + + // Validate, decode, and upload each file; build per-file pipeline chains + const perFileChildren: FlowJob[] = []; + const preFailures: Array<{ originalIndex: number; filename: string; error: string }> = []; + let flowChildIndex = 0; + + // The first step's modality drives input validation for every file, so + // audio, video, and document pipelines are not rejected by the image + // validator. + const batchModality = + TOOLS.find((t) => t.id === pipeline.steps[0]?.toolId)?.modality ?? "image"; + const pipelineBatchScratch = join( + tmpdir(), + "snapotter-scratch", + `pipeline-batch-${parentId}`, + ); + await mkdir(pipelineBatchScratch, { recursive: true }); + + for (let fi = 0; fi < files.length; fi++) { + const file = files[fi]; + let processBuffer = file.buffer; + let processFilename = file.filename; + + if (batchModality === "image") { + const fileValidation = await validateImageBuffer(processBuffer, processFilename); + if (!fileValidation.valid) { preFailures.push({ originalIndex: fi, filename: file.filename, - error: "Failed to decode HEIC file", + error: `Invalid image: ${fileValidation.reason}`, }); continue; } - } - if (needsCliDecode(fileValidation.format)) { - try { - const fileExt = processFilename.split(".").pop()?.toLowerCase(); - processBuffer = await decodeToSharpCompat( - processBuffer, - fileValidation.format, - fileExt, - ); - const ext = processFilename.match(/\.[^.]+$/)?.[0]; - if (ext) processFilename = `${processFilename.slice(0, -ext.length)}.png`; - } catch { - // Fall through -- tool might handle it + // Decode chain + if (fileValidation.format === "heif") { + try { + processBuffer = await decodeHeic(processBuffer); + const ext = processFilename.match(/\.[^.]+$/)?.[0]; + if (ext) processFilename = `${processFilename.slice(0, -ext.length)}.png`; + } catch { + preFailures.push({ + originalIndex: fi, + filename: file.filename, + error: "Failed to decode HEIC file", + }); + continue; + } } - } - if (isSvgBuffer(processBuffer)) { - processBuffer = sanitizeSvg(processBuffer); + if (needsCliDecode(fileValidation.format)) { + try { + const fileExt = processFilename.split(".").pop()?.toLowerCase(); + processBuffer = await decodeToSharpCompat( + processBuffer, + fileValidation.format, + fileExt, + ); + const ext = processFilename.match(/\.[^.]+$/)?.[0]; + if (ext) processFilename = `${processFilename.slice(0, -ext.length)}.png`; + } catch { + // Fall through -- tool might handle it + } + } + + if (isSvgBuffer(processBuffer)) { + processBuffer = sanitizeSvg(processBuffer); + } else { + processBuffer = await autoOrient(processBuffer); + } } else { - processBuffer = await autoOrient(processBuffer); - } - } else { - // Non-image input: validate/decode via the tool's modality handler. - try { - const prepared = await inputHandlerFor(batchModality).prepare( - processBuffer, - processFilename, - { - scratchDir: pipelineBatchScratch, - lenient: getToolConfig(pipeline.steps[0]?.toolId ?? "")?.skipStructuralValidation, - }, - ); - processBuffer = prepared.buffer; - processFilename = prepared.filename; - } catch (err) { - if (err instanceof InputValidationError) { - preFailures.push({ originalIndex: fi, filename: file.filename, error: err.message }); - continue; + // Non-image input: validate/decode via the tool's modality handler. + try { + const prepared = await inputHandlerFor(batchModality).prepare( + processBuffer, + processFilename, + { + scratchDir: pipelineBatchScratch, + lenient: getToolConfig(pipeline.steps[0]?.toolId ?? "")?.skipStructuralValidation, + }, + ); + processBuffer = prepared.buffer; + processFilename = prepared.filename; + } catch (err) { + if (err instanceof InputValidationError) { + preFailures.push({ originalIndex: fi, filename: file.filename, error: err.message }); + continue; + } + throw err; } - throw err; } - } - // Upload decoded file - const perFileJobId = `${parentId}-f${flowChildIndex}`; - const uploadKey = `uploads/${perFileJobId}-s0/${processFilename}`; - await putObject(uploadKey, processBuffer); + // Upload decoded file + const perFileJobId = `${parentId}-f${flowChildIndex}`; + const uploadKey = `uploads/${perFileJobId}-s0/${processFilename}`; + await putObject(uploadKey, processBuffer); - // Build per-file pipeline chain - const { tree: perFileTree, stepJobIds } = buildPipelineFlowTree({ - jobId: perFileJobId, - userId, - parsedSteps, - uploadKey, - filename: processFilename, - parentId, - totalFiles: files.length, - }); - - // Insert step + finalize rows for this file - for (let si = 0; si < parsedSteps.length; si++) { - await db.insert(schema.jobs).values({ - id: stepJobIds[si], + // Build per-file pipeline chain + const { tree: perFileTree, stepJobIds } = buildPipelineFlowTree({ + jobId: perFileJobId, userId, - toolId: parsedSteps[si].resolvedToolId, - pool: parsedSteps[si].pool, - type: "pipeline-step", - status: "queued", - inputRefs: si === 0 ? [uploadKey] : [], - settings: parsedSteps[si].parsedSettings as Record, + parsedSteps, + uploadKey, + filename: processFilename, + parentId, + totalFiles: files.length, }); - } - await db.insert(schema.jobs).values({ - id: perFileJobId, - userId, - toolId: "pipeline", - pool: "image", - type: "pipeline-finalize", - status: "queued", - inputRefs: [], - settings: {}, - }); - - perFileChildren.push(perFileTree); - flowChildIndex++; - } - - // Record pre-failures in batch progress - for (const pf of preFailures) { - await recordChildOutcome(parentId, files.length, pf.filename, pf.error); - } - - if (perFileChildren.length === 0) { - // All files failed validation - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: true, - file_count: files.length, - duration_ms: Date.now() - batchStartTime, - status: "failed", - }); - return reply.status(422).send({ - error: "All files failed processing", - errors: preFailures.map((f) => ({ filename: f.filename, error: f.error })), - }); - } - - // Build batch-finalize parent - const batchTree: FlowJob = { - name: "batch-finalize", - queueName: queueName("system"), - data: { - kind: "batch-finalize", - jobId: parentId, - toolId: "pipeline-batch", - userId, - pool: "system" as Pool, - totalFiles: files.length, - inputRefs: [], - filename: "", - settings: { flowChildCount: perFileChildren.length }, - } satisfies ToolJobData, - opts: { jobId: parentId, attempts: 1 }, - children: perFileChildren, - }; - - // Update the parent row with the final flow child count - await db - .update(schema.jobs) - .set({ settings: { flowChildCount: perFileChildren.length } }) - .where(eq(schema.jobs.id, parentId)); - - // Inject OTel trace context into every node of the batch flow tree - injectTraceContextIntoFlow(batchTree); - - await getFlowProducer().add(batchTree); - - // Wait for batch completion - const batchResult = await waitForJob("system", parentId, 30 * 60_000); - - if (!batchResult) { - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: true, - file_count: files.length, - duration_ms: Date.now() - batchStartTime, - status: "failed", - }); - return reply.status(422).send({ error: "Pipeline batch processing timed out" }); - } - - const manifest = (batchResult.resultPayload?.manifest ?? []) as Array<{ - index: number; - filename: string; - outputRef?: string; - error?: string; - }>; - - // Combine manifest with pre-failures - const allResults: Array<{ - originalIndex: number; - filename: string; - outputRef?: string; - error?: string; - }> = []; - - // Map flow indices back to original file indices - let fci = 0; - for (let fi = 0; fi < files.length; fi++) { - const pf = preFailures.find((p) => p.originalIndex === fi); - if (pf) { - allResults.push({ - originalIndex: fi, - filename: pf.filename, - error: pf.error, - }); - } else { - const entry = manifest.find((m) => m.index === fci); - if (entry) { - allResults.push({ - originalIndex: fi, - filename: entry.filename, - outputRef: entry.outputRef, - error: entry.error, + // Insert step + finalize rows for this file + for (let si = 0; si < parsedSteps.length; si++) { + await db.insert(schema.jobs).values({ + id: stepJobIds[si], + userId, + toolId: parsedSteps[si].resolvedToolId, + pool: parsedSteps[si].pool, + type: "pipeline-step", + status: "queued", + inputRefs: si === 0 ? [uploadKey] : [], + settings: parsedSteps[si].parsedSettings as Record, }); } - fci++; - } - } - // Deduplicate output filenames - const usedNames = new Set(); - function getUniqueName(name: string): string { - if (!usedNames.has(name)) { - usedNames.add(name); - return name; - } - const dotIdx = name.lastIndexOf("."); - const base = dotIdx > 0 ? name.slice(0, dotIdx) : name; - const ext = dotIdx > 0 ? name.slice(dotIdx) : ""; - let counter = 1; - let candidate = `${base}_${counter}${ext}`; - while (usedNames.has(candidate)) { - counter++; - candidate = `${base}_${counter}${ext}`; - } - usedNames.add(candidate); - return candidate; - } + await db.insert(schema.jobs).values({ + id: perFileJobId, + userId, + toolId: "pipeline", + pool: "image", + type: "pipeline-finalize", + status: "queued", + inputRefs: [], + settings: {}, + }); - const successEntries = allResults.filter((r) => r.outputRef); - const failedEntries = allResults.filter((r) => !r.outputRef); + perFileChildren.push(perFileTree); + flowChildIndex++; + } + + // Record pre-failures in batch progress + for (const pf of preFailures) { + await recordChildOutcome(parentId, files.length, pf.filename, pf.error); + } + + if (perFileChildren.length === 0) { + // All files failed validation + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: true, + file_count: files.length, + duration_ms: Date.now() - batchStartTime, + status: "failed", + }); + return reply.status(422).send({ + error: "All files failed processing", + errors: preFailures.map((f) => ({ filename: f.filename, error: f.error })), + }); + } + + // Build batch-finalize parent + const batchTree: FlowJob = { + name: "batch-finalize", + queueName: queueName("system"), + data: { + kind: "batch-finalize", + jobId: parentId, + toolId: "pipeline-batch", + userId, + pool: "system" as Pool, + totalFiles: files.length, + inputRefs: [], + filename: "", + settings: { flowChildCount: perFileChildren.length }, + } satisfies ToolJobData, + opts: { jobId: parentId, attempts: 1 }, + children: perFileChildren, + }; + + // Update the parent row with the final flow child count + await db + .update(schema.jobs) + .set({ settings: { flowChildCount: perFileChildren.length } }) + .where(eq(schema.jobs.id, parentId)); + + // Inject OTel trace context into every node of the batch flow tree + injectTraceContextIntoFlow(batchTree); + + await getFlowProducer().add(batchTree); + + // Wait for batch completion + const batchResult = await waitForJob("system", parentId, 30 * 60_000); + + if (!batchResult) { + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: true, + file_count: files.length, + duration_ms: Date.now() - batchStartTime, + status: "failed", + }); + return reply.status(422).send({ error: "Pipeline batch processing timed out" }); + } + + const manifest = (batchResult.resultPayload?.manifest ?? []) as Array<{ + index: number; + filename: string; + outputRef?: string; + error?: string; + }>; + + // Combine manifest with pre-failures + const allResults: Array<{ + originalIndex: number; + filename: string; + outputRef?: string; + error?: string; + }> = []; + + // Map flow indices back to original file indices + let fci = 0; + for (let fi = 0; fi < files.length; fi++) { + const pf = preFailures.find((p) => p.originalIndex === fi); + if (pf) { + allResults.push({ + originalIndex: fi, + filename: pf.filename, + error: pf.error, + }); + } else { + const entry = manifest.find((m) => m.index === fci); + if (entry) { + allResults.push({ + originalIndex: fi, + filename: entry.filename, + outputRef: entry.outputRef, + error: entry.error, + }); + } + fci++; + } + } + + // Deduplicate output filenames + const usedNames = new Set(); + function getUniqueName(name: string): string { + if (!usedNames.has(name)) { + usedNames.add(name); + return name; + } + const dotIdx = name.lastIndexOf("."); + const base = dotIdx > 0 ? name.slice(0, dotIdx) : name; + const ext = dotIdx > 0 ? name.slice(dotIdx) : ""; + let counter = 1; + let candidate = `${base}_${counter}${ext}`; + while (usedNames.has(candidate)) { + counter++; + candidate = `${base}_${counter}${ext}`; + } + usedNames.add(candidate); + return candidate; + } + + const successEntries = allResults.filter((r) => r.outputRef); + const failedEntries = allResults.filter((r) => !r.outputRef); + + if (successEntries.length === 0) { + trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { + step_count: pipeline.steps.length, + tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), + is_batch: true, + file_count: files.length, + duration_ms: Date.now() - batchStartTime, + status: "failed", + }); + return reply.status(422).send({ + error: "All files failed processing", + errors: failedEntries.map((f) => ({ filename: f.filename, error: f.error ?? "Failed" })), + }); + } - if (successEntries.length === 0) { trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { step_count: pipeline.steps.length, tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), is_batch: true, file_count: files.length, duration_ms: Date.now() - batchStartTime, - status: "failed", + status: "completed", }); - return reply.status(422).send({ - error: "All files failed processing", - errors: failedEntries.map((f) => ({ filename: f.filename, error: f.error ?? "Failed" })), - }); - } - trackEvent(request, ANALYTICS_EVENTS.PIPELINE_EXECUTED, { - step_count: pipeline.steps.length, - tool_ids: pipeline.steps.map((s: { toolId: string }) => s.toolId), - is_batch: true, - file_count: files.length, - duration_ms: Date.now() - batchStartTime, - status: "completed", - }); - - const fileResultsMap: Record = {}; - for (const entry of successEntries) { - const uniqueName = getUniqueName(entry.filename); - entry.filename = uniqueName; - fileResultsMap[String(entry.originalIndex)] = uniqueName; - } - - // ── Stream ZIP response ────────────────────────────────────────── - reply.hijack(); - reply.raw.writeHead(200, { - "Content-Type": "application/zip", - "Content-Disposition": `attachment; filename="pipeline-batch-${parentId.slice(0, 8)}.zip"`, - "Transfer-Encoding": "chunked", - "X-Job-Id": parentId, - "X-File-Results": encodeURIComponent(JSON.stringify(fileResultsMap)), - ...getSecurityHeaders(), - }); - - const archive = archiver("zip", { zlib: { level: 5 } }); - - archive.on("error", (err) => { - request.log.error({ err }, "Archiver error during pipeline batch processing"); - if (!reply.raw.writableEnded) { - reply.raw.end(); - } - }); - - archive.pipe(reply.raw); - - // Append results from object storage in original upload order - try { + const fileResultsMap: Record = {}; for (const entry of successEntries) { - if (!entry.outputRef) continue; - const stream = await getObjectStream(entry.outputRef); - archive.append(stream, { name: entry.filename }); + const uniqueName = getUniqueName(entry.filename); + entry.filename = uniqueName; + fileResultsMap[String(entry.originalIndex)] = uniqueName; } - await archive.finalize(); - } catch (err) { - request.log.error({ err }, "Failed to stream ZIP entries during pipeline batch processing"); - archive.abort(); - if (!reply.raw.writableEnded) { - reply.raw.end(); + // ── Stream ZIP response ────────────────────────────────────────── + reply.hijack(); + reply.raw.writeHead(200, { + "Content-Type": "application/zip", + "Content-Disposition": `attachment; filename="pipeline-batch-${parentId.slice(0, 8)}.zip"`, + "Transfer-Encoding": "chunked", + "X-Job-Id": parentId, + "X-File-Results": encodeURIComponent(JSON.stringify(fileResultsMap)), + ...getSecurityHeaders(), + }); + + const archive = archiver("zip", { zlib: { level: 5 } }); + + archive.on("error", (err) => { + request.log.error({ err }, "Archiver error during pipeline batch processing"); + if (!reply.raw.writableEnded) { + reply.raw.end(); + } + }); + + archive.pipe(reply.raw); + + // Append results from object storage in original upload order + try { + for (const entry of successEntries) { + if (!entry.outputRef) continue; + const stream = await getObjectStream(entry.outputRef); + archive.append(stream, { name: entry.filename }); + } + + await archive.finalize(); + } catch (err) { + request.log.error({ err }, "Failed to stream ZIP entries during pipeline batch processing"); + archive.abort(); + if (!reply.raw.writableEnded) { + reply.raw.end(); + } } - } - }); + }, + ); app.log.info("Pipeline routes registered"); } diff --git a/apps/api/src/routes/progress.ts b/apps/api/src/routes/progress.ts index 78415b21..04e3e1ae 100644 --- a/apps/api/src/routes/progress.ts +++ b/apps/api/src/routes/progress.ts @@ -263,6 +263,7 @@ export async function registerProgressRoutes(app: FastifyInstance): Promise, reply: FastifyReply) => { const { jobId } = request.params; diff --git a/apps/api/src/routes/settings.ts b/apps/api/src/routes/settings.ts index b4188958..5cb869ff 100644 --- a/apps/api/src/routes/settings.ts +++ b/apps/api/src/routes/settings.ts @@ -51,25 +51,29 @@ async function decryptIfNeeded(value: string): Promise { export async function settingsRoutes(app: FastifyInstance): Promise { // GET /api/v1/settings — Get all settings as a key-value object - app.get("/api/v1/settings", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.get( + "/api/v1/settings", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const isAdmin = user.role === "admin"; - const rows = await db.select().from(schema.settings); + const isAdmin = user.role === "admin"; + const rows = await db.select().from(schema.settings); - const settings: Record = {}; - for (const row of rows) { - if (!isAdmin && SENSITIVE_KEYS.has(row.key)) continue; - if (REDACTED_KEYS.has(row.key)) { - settings[row.key] = "********"; - continue; + const settings: Record = {}; + for (const row of rows) { + if (!isAdmin && SENSITIVE_KEYS.has(row.key)) continue; + if (REDACTED_KEYS.has(row.key)) { + settings[row.key] = "********"; + continue; + } + settings[row.key] = await decryptIfNeeded(row.value); } - settings[row.key] = await decryptIfNeeded(row.value); - } - return reply.send({ settings }); - }); + return reply.send({ settings }); + }, + ); // PUT /api/v1/settings — Save settings (admin only) app.put("/api/v1/settings", async (request: FastifyRequest, reply: FastifyReply) => { @@ -146,6 +150,7 @@ export async function settingsRoutes(app: FastifyInstance): Promise { // GET /api/v1/settings/:key — Get a specific setting app.get( "/api/v1/settings/:key", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { key: string } }>, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; diff --git a/apps/api/src/routes/user-files.ts b/apps/api/src/routes/user-files.ts index 46e6fc9b..b1d7d628 100644 --- a/apps/api/src/routes/user-files.ts +++ b/apps/api/src/routes/user-files.ts @@ -173,6 +173,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise { */ app.get( "/api/v1/files", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async ( request: FastifyRequest<{ Querystring: { search?: string; limit?: string; offset?: string }; @@ -344,6 +345,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise { */ app.get( "/api/v1/files/:id", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; @@ -430,6 +432,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise { */ app.get( "/api/v1/files/:id/download", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; @@ -469,6 +472,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise { */ app.get( "/api/v1/files/:id/thumbnail", + { config: { rateLimit: { max: 300, timeWindow: "1 minute" } } }, async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => { const user = requireAuth(request, reply); if (!user) return; @@ -567,52 +571,55 @@ export async function userFileRoutes(app: FastifyInstance): Promise { * Bulk delete. Body: { ids: string[] } * For each id, deletes the entire version chain (all ancestors and descendants). */ - app.delete("/api/v1/files", async (request: FastifyRequest, reply: FastifyReply) => { - const user = requireAuth(request, reply); - if (!user) return; + app.delete( + "/api/v1/files", + { config: { rateLimit: { max: 60, timeWindow: "1 minute" } } }, + async (request: FastifyRequest, reply: FastifyReply) => { + const user = requireAuth(request, reply); + if (!user) return; - const deleteSchema = z.object({ - ids: z.array(z.string()).min(1, "ids must be a non-empty array of strings"), - }); - const parsed = deleteSchema.safeParse(request.body); - if (!parsed.success) { - return reply.status(400).send({ - error: parsed.error.issues.map((i) => i.message).join("; "), + const deleteSchema = z.object({ + ids: z.array(z.string()).min(1, "ids must be a non-empty array of strings"), }); - } - const { ids } = parsed.data; + const parsed = deleteSchema.safeParse(request.body); + if (!parsed.success) { + return reply.status(400).send({ + error: parsed.error.issues.map((i) => i.message).join("; "), + }); + } + const { ids } = parsed.data; - // Check files:all permission once upfront - const canDeleteAll = await hasEffectivePermission(user, "files:all"); + // Check files:all permission once upfront + const canDeleteAll = await hasEffectivePermission(user, "files:all"); - // Batch ownership check: single SELECT for all requested IDs - const candidates = await db - .select({ id: schema.userFiles.id, userId: schema.userFiles.userId }) - .from(schema.userFiles) - .where(inArray(schema.userFiles.id, ids)); + // Batch ownership check: single SELECT for all requested IDs + const candidates = await db + .select({ id: schema.userFiles.id, userId: schema.userFiles.userId }) + .from(schema.userFiles) + .where(inArray(schema.userFiles.id, ids)); - const validIds = candidates - .filter((f) => f.userId === user.id || canDeleteAll) - .map((f) => f.id); + const validIds = candidates + .filter((f) => f.userId === user.id || canDeleteAll) + .map((f) => f.id); - if (validIds.length === 0) { - await auditFromRequest(request)("FILE_DELETED", { userId: user.id, count: 0, ids }); - return reply.send({ deleted: 0 }); - } + if (validIds.length === 0) { + await auditFromRequest(request)("FILE_DELETED", { userId: user.id, count: 0, ids }); + return reply.send({ deleted: 0 }); + } - type DeleteChainRow = { - id: string; - stored_name: string; - size: number | null; - user_id: string | null; - }; + type DeleteChainRow = { + id: string; + stored_name: string; + size: number | null; + user_id: string | null; + }; - // Single recursive CTE to collect all chain members for every valid ID - const seedIds = sql.join( - validIds.map((id) => sql`${id}`), - sql`, `, - ); - const cteResult = await db.execute(sql` + // Single recursive CTE to collect all chain members for every valid ID + const seedIds = sql.join( + validIds.map((id) => sql`${id}`), + sql`, `, + ); + const cteResult = await db.execute(sql` WITH RECURSIVE ancestors(id, parent_id) AS ( SELECT id, parent_id FROM user_files @@ -631,44 +638,45 @@ export async function userFileRoutes(app: FastifyInstance): Promise { ) SELECT DISTINCT id, stored_name, size, user_id FROM chain `); - const chainRows = cteResult.rows; + const chainRows = cteResult.rows; - // Filesystem deletes (must loop; cannot batch across the OS) - for (const row of chainRows) { - await deleteStoredFile(row.stored_name); - await deleteThumbnail(row.stored_name); - } - - // Batch DB delete - const chainIds = chainRows.map((r) => r.id); - if (chainIds.length > 0) { - await db.delete(schema.userFiles).where(inArray(schema.userFiles.id, chainIds)); - } - - // Decrement storageUsed per user (group by userId for files:all scenarios) - const perUserSizes = new Map(); - for (const row of chainRows) { - if (row.user_id && row.size) { - perUserSizes.set(row.user_id, (perUserSizes.get(row.user_id) ?? 0) + row.size); + // Filesystem deletes (must loop; cannot batch across the OS) + for (const row of chainRows) { + await deleteStoredFile(row.stored_name); + await deleteThumbnail(row.stored_name); } - } - for (const [uid, totalSize] of perUserSizes) { - await db - .update(schema.users) - .set({ - storageUsed: sql`GREATEST(0, ${schema.users.storageUsed} - ${totalSize})`, - }) - .where(eq(schema.users.id, uid)); - } - await auditFromRequest(request)("FILE_DELETED", { - userId: user.id, - count: chainRows.length, - ids, - }); + // Batch DB delete + const chainIds = chainRows.map((r) => r.id); + if (chainIds.length > 0) { + await db.delete(schema.userFiles).where(inArray(schema.userFiles.id, chainIds)); + } - return reply.send({ deleted: chainRows.length }); - }); + // Decrement storageUsed per user (group by userId for files:all scenarios) + const perUserSizes = new Map(); + for (const row of chainRows) { + if (row.user_id && row.size) { + perUserSizes.set(row.user_id, (perUserSizes.get(row.user_id) ?? 0) + row.size); + } + } + for (const [uid, totalSize] of perUserSizes) { + await db + .update(schema.users) + .set({ + storageUsed: sql`GREATEST(0, ${schema.users.storageUsed} - ${totalSize})`, + }) + .where(eq(schema.users.id, uid)); + } + + await auditFromRequest(request)("FILE_DELETED", { + userId: user.id, + count: chainRows.length, + ids, + }); + + return reply.send({ deleted: chainRows.length }); + }, + ); /** * POST /api/v1/files/save-result