diff --git a/dashboard/package-lock.json b/dashboard/package-lock.json index 6319ddc..20e20d3 100644 --- a/dashboard/package-lock.json +++ b/dashboard/package-lock.json @@ -8,14 +8,6 @@ "name": "dashboard", "version": "0.1.0", "dependencies": { - "@radix-ui/react-dialog": "^1.1.6", - "@radix-ui/react-dropdown-menu": "^2.1.6", - "@radix-ui/react-scroll-area": "^1.2.3", - "@radix-ui/react-select": "^2.1.6", - "@radix-ui/react-separator": "^1.1.2", - "@radix-ui/react-slot": "^1.1.2", - "@radix-ui/react-tabs": "^1.1.3", - "@radix-ui/react-tooltip": "^1.1.8", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "cn": "^0.3.0", diff --git a/dashboard/package.json b/dashboard/package.json index 56d55d1..75b7837 100644 --- a/dashboard/package.json +++ b/dashboard/package.json @@ -9,14 +9,6 @@ "preview": "vite preview" }, "dependencies": { - "@radix-ui/react-dialog": "^1.1.6", - "@radix-ui/react-dropdown-menu": "^2.1.6", - "@radix-ui/react-scroll-area": "^1.2.3", - "@radix-ui/react-select": "^2.1.6", - "@radix-ui/react-separator": "^1.1.2", - "@radix-ui/react-slot": "^1.1.2", - "@radix-ui/react-tabs": "^1.1.3", - "@radix-ui/react-tooltip": "^1.1.8", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "cn": "^0.3.0", diff --git a/dashboard/src/components/ui/dialog.tsx b/dashboard/src/components/ui/dialog.tsx deleted file mode 100644 index d631cd0..0000000 --- a/dashboard/src/components/ui/dialog.tsx +++ /dev/null @@ -1,144 +0,0 @@ -"use client"; - -import * as React from "react"; -import { cn } from "cn"; -import { XIcon } from "lucide-react"; -import { Dialog as DialogPrimitive } from "radix-ui"; - -import { Button } from "@/components/ui/button"; - -function Dialog({ ...props }: React.ComponentProps) { - return ; -} - -function DialogTrigger({ ...props }: React.ComponentProps) { - return ; -} - -function DialogPortal({ ...props }: React.ComponentProps) { - return ; -} - -function DialogClose({ ...props }: React.ComponentProps) { - return ; -} - -function DialogOverlay({ - className, - ...props -}: React.ComponentProps) { - return ( - - ); -} - -function DialogContent({ - className, - children, - showCloseButton = true, - ...props -}: React.ComponentProps & { - showCloseButton?: boolean; -}) { - return ( - - - - {children} - {showCloseButton && ( - - - Close - - )} - - - ); -} - -function DialogHeader({ className, ...props }: React.ComponentProps<"div">) { - return ( -
- ); -} - -function DialogFooter({ - className, - showCloseButton = false, - children, - ...props -}: React.ComponentProps<"div"> & { - showCloseButton?: boolean; -}) { - return ( -
- {children} - {showCloseButton && ( - - - - )} -
- ); -} - -function DialogTitle({ className, ...props }: React.ComponentProps) { - return ( - - ); -} - -function DialogDescription({ - className, - ...props -}: React.ComponentProps) { - return ( - - ); -} - -export { - Dialog, - DialogClose, - DialogContent, - DialogDescription, - DialogFooter, - DialogHeader, - DialogOverlay, - DialogPortal, - DialogTitle, - DialogTrigger, -}; diff --git a/dashboard/src/components/ui/dropdown-menu.tsx b/dashboard/src/components/ui/dropdown-menu.tsx deleted file mode 100644 index 84c4027..0000000 --- a/dashboard/src/components/ui/dropdown-menu.tsx +++ /dev/null @@ -1,225 +0,0 @@ -import * as React from "react"; -import { cn } from "cn"; -import { CheckIcon, ChevronRightIcon, CircleIcon } from "lucide-react"; -import { DropdownMenu as DropdownMenuPrimitive } from "radix-ui"; - -function DropdownMenu({ ...props }: React.ComponentProps) { - return ; -} - -function DropdownMenuPortal({ - ...props -}: React.ComponentProps) { - return ; -} - -function DropdownMenuTrigger({ - ...props -}: React.ComponentProps) { - return ; -} - -function DropdownMenuContent({ - className, - sideOffset = 4, - ...props -}: React.ComponentProps) { - return ( - - - - ); -} - -function DropdownMenuGroup({ ...props }: React.ComponentProps) { - return ; -} - -function DropdownMenuItem({ - className, - inset, - variant = "default", - ...props -}: React.ComponentProps & { - inset?: boolean; - variant?: "default" | "destructive"; -}) { - return ( - - ); -} - -function DropdownMenuCheckboxItem({ - className, - children, - checked, - ...props -}: React.ComponentProps) { - return ( - - - - - - - {children} - - ); -} - -function DropdownMenuRadioGroup({ - ...props -}: React.ComponentProps) { - return ; -} - -function DropdownMenuRadioItem({ - className, - children, - ...props -}: React.ComponentProps) { - return ( - - - - - - - {children} - - ); -} - -function DropdownMenuLabel({ - className, - inset, - ...props -}: React.ComponentProps & { - inset?: boolean; -}) { - return ( - - ); -} - -function DropdownMenuSeparator({ - className, - ...props -}: React.ComponentProps) { - return ( - - ); -} - -function DropdownMenuShortcut({ className, ...props }: React.ComponentProps<"span">) { - return ( - - ); -} - -function DropdownMenuSub({ ...props }: React.ComponentProps) { - return ; -} - -function DropdownMenuSubTrigger({ - className, - inset, - children, - ...props -}: React.ComponentProps & { - inset?: boolean; -}) { - return ( - - {children} - - - ); -} - -function DropdownMenuSubContent({ - className, - ...props -}: React.ComponentProps) { - return ( - - ); -} - -export { - DropdownMenu, - DropdownMenuPortal, - DropdownMenuTrigger, - DropdownMenuContent, - DropdownMenuGroup, - DropdownMenuLabel, - DropdownMenuItem, - DropdownMenuCheckboxItem, - DropdownMenuRadioGroup, - DropdownMenuRadioItem, - DropdownMenuSeparator, - DropdownMenuShortcut, - DropdownMenuSub, - DropdownMenuSubTrigger, - DropdownMenuSubContent, -}; diff --git a/dashboard/src/components/ui/tabs.tsx b/dashboard/src/components/ui/tabs.tsx deleted file mode 100644 index 8b5c133..0000000 --- a/dashboard/src/components/ui/tabs.tsx +++ /dev/null @@ -1,80 +0,0 @@ -"use client"; - -import * as React from "react"; -import { cva, type VariantProps } from "class-variance-authority"; -import { cn } from "cn"; -import { Tabs as TabsPrimitive } from "radix-ui"; - -function Tabs({ - className, - orientation = "horizontal", - ...props -}: React.ComponentProps) { - return ( - - ); -} - -const tabsListVariants = cva( - "group/tabs-list inline-flex w-fit items-center justify-center rounded-lg p-[3px] text-muted-foreground group-data-[orientation=horizontal]/tabs:h-9 group-data-[orientation=vertical]/tabs:h-fit group-data-[orientation=vertical]/tabs:flex-col data-[variant=line]:rounded-none", - { - variants: { - variant: { - default: "bg-muted", - line: "gap-1 bg-transparent", - }, - }, - defaultVariants: { - variant: "default", - }, - }, -); - -function TabsList({ - className, - variant = "default", - ...props -}: React.ComponentProps & VariantProps) { - return ( - - ); -} - -function TabsTrigger({ className, ...props }: React.ComponentProps) { - return ( - - ); -} - -function TabsContent({ className, ...props }: React.ComponentProps) { - return ( - - ); -} - -export { Tabs, TabsList, TabsTrigger, TabsContent, tabsListVariants }; diff --git a/dashboard/src/components/ui/tooltip.tsx b/dashboard/src/components/ui/tooltip.tsx deleted file mode 100644 index 7daab86..0000000 --- a/dashboard/src/components/ui/tooltip.tsx +++ /dev/null @@ -1,52 +0,0 @@ -"use client"; - -import * as React from "react"; -import { cn } from "cn"; -import { Tooltip as TooltipPrimitive } from "radix-ui"; - -function TooltipProvider({ - delayDuration = 0, - ...props -}: React.ComponentProps) { - return ( - - ); -} - -function Tooltip({ ...props }: React.ComponentProps) { - return ; -} - -function TooltipTrigger({ ...props }: React.ComponentProps) { - return ; -} - -function TooltipContent({ - className, - sideOffset = 0, - children, - ...props -}: React.ComponentProps) { - return ( - - - {children} - - - - ); -} - -export { Tooltip, TooltipTrigger, TooltipContent, TooltipProvider }; diff --git a/package-lock.json b/package-lock.json index 9387cab..662e33e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -16,7 +16,6 @@ "ai": "^7.0.118", "dotenv": "^18.0.4", "drizzle-orm": "^0.45.3", - "drizzle-zod": "^0.8.3", "ioredis": "^6.0.0", "pino": "^10.3.1", "pino-pretty": "^13.1.3", @@ -5679,16 +5678,6 @@ } } }, - "node_modules/drizzle-zod": { - "version": "0.8.3", - "resolved": "https://registry.npmjs.org/drizzle-zod/-/drizzle-zod-0.8.3.tgz", - "integrity": "sha512-66yVOuvGhKJnTdiqj1/Xaaz9/qzOdRJADpDa68enqS6g3t0kpNkwNYjUuaeXgZfO/UWuIM9HIhSlJ6C5ZraMww==", - "license": "Apache-2.0", - "peerDependencies": { - "drizzle-orm": ">=0.36.0", - "zod": "^3.25.0 || ^4.0.0" - } - }, "node_modules/dts-resolver": { "version": "3.0.0", "resolved": "https://registry.npmjs.org/dts-resolver/-/dts-resolver-3.0.0.tgz", diff --git a/package.json b/package.json index bcd68c3..fd0f69b 100644 --- a/package.json +++ b/package.json @@ -76,7 +76,6 @@ "ai": "^7.0.118", "dotenv": "^18.0.4", "drizzle-orm": "^0.45.3", - "drizzle-zod": "^0.8.3", "ioredis": "^6.0.0", "pino": "^10.3.1", "pino-pretty": "^13.1.3", diff --git a/scripts/build.sh b/scripts/build.sh deleted file mode 100644 index 584604d..0000000 --- a/scripts/build.sh +++ /dev/null @@ -1,4 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -npm run build diff --git a/scripts/setup.sh b/scripts/setup.sh deleted file mode 100644 index 6d4d747..0000000 --- a/scripts/setup.sh +++ /dev/null @@ -1,18 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -node_version=$(node --version | cut -c2-) -required_version="22.0.0" - -if [ "$(printf '%s\n' "$required_version" "$node_version" | sort -V | head -n1)" != "$required_version" ]; then - echo "Node.js >= 22.0.0 is required. Found: $node_version" - exit 1 -fi - -if ! command -v npm &>/dev/null; then - echo "npm is required." - exit 1 -fi - -npm install -npm run build diff --git a/src/db/schema.ts b/src/db/schema.ts index 5b49cf9..6952994 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -15,7 +15,6 @@ import { integer, } from "drizzle-orm/pg-core"; import { sql } from "drizzle-orm"; -import { createInsertSchema, createSelectSchema } from "drizzle-zod"; export const channelEnum = pgEnum("channel", [ "email", @@ -376,50 +375,3 @@ export const adminUsers = pgTable("admin_users", { createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(), updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(), }); - -// ─── Generated Zod Schemas ────────────────────────────────────────────────── - -export const insertAdminUserSchema = createInsertSchema(adminUsers); -export const selectAdminUserSchema = createSelectSchema(adminUsers); - -export const insertProjectSchema = createInsertSchema(projects); -export const selectProjectSchema = createSelectSchema(projects); - -export const insertProjectApiKeySchema = createInsertSchema(projectApiKeys); -export const selectProjectApiKeySchema = createSelectSchema(projectApiKeys); - -export const insertUserSchema = createInsertSchema(users); -export const selectUserSchema = createSelectSchema(users); - -export const insertUserSegmentSchema = createInsertSchema(userSegments); -export const selectUserSegmentSchema = createSelectSchema(userSegments); - -export const insertUserContactSchema = createInsertSchema(userContacts); -export const selectUserContactSchema = createSelectSchema(userContacts); - -export const insertUserChannelPreferenceSchema = createInsertSchema(userChannelPreferences); -export const selectUserChannelPreferenceSchema = createSelectSchema(userChannelPreferences); - -export const insertUserTopicPreferenceSchema = createInsertSchema(userTopicPreferences); -export const selectUserTopicPreferenceSchema = createSelectSchema(userTopicPreferences); - -export const insertMessageLogSchema = createInsertSchema(messageLogs); -export const selectMessageLogSchema = createSelectSchema(messageLogs); - -export const insertSuppressionSchema = createInsertSchema(suppressions); -export const selectSuppressionSchema = createSelectSchema(suppressions); - -export const insertWorkflowInstanceSchema = createInsertSchema(workflowInstances); -export const selectWorkflowInstanceSchema = createSelectSchema(workflowInstances); - -export const insertWorkflowStepSchema = createInsertSchema(workflowSteps); -export const selectWorkflowStepSchema = createSelectSchema(workflowSteps); - -export const insertWorkflowWaiterSchema = createInsertSchema(workflowWaiters); -export const selectWorkflowWaiterSchema = createSelectSchema(workflowWaiters); - -export const insertDeliveryOutboxSchema = createInsertSchema(deliveryOutbox); -export const selectDeliveryOutboxSchema = createSelectSchema(deliveryOutbox); - -export const insertScheduledPayloadSchema = createInsertSchema(scheduledPayloads); -export const selectScheduledPayloadSchema = createSelectSchema(scheduledPayloads); diff --git a/src/logger/index.ts b/src/logger/index.ts index e60fdba..ec39d38 100644 --- a/src/logger/index.ts +++ b/src/logger/index.ts @@ -46,15 +46,3 @@ export function createLogger({ name, level = "info", pretty, context }: LoggerOp return pino(options); } - -export function withRequestId(logger: Logger, requestId: string): Logger { - return logger.child({ requestId }); -} - -export function withContext(logger: Logger, context: Record): Logger { - return logger.child(context); -} - -export function childLogger(logger: Logger, bindings: Record): Logger { - return logger.child(bindings); -} diff --git a/src/rate-limiter/index.ts b/src/rate-limiter/index.ts index 2af8eaa..c5df848 100644 --- a/src/rate-limiter/index.ts +++ b/src/rate-limiter/index.ts @@ -1,10 +1,7 @@ import type { Redis } from "@/index.js"; import { LRUCache } from "@/shared/index.js"; -import { LUA_USER_THROTTLE, LUA_LEASE_API_RATE_LIMIT } from "@/redis/index.js"; import type { Logger } from "@/logger/index.js"; -export { LUA_USER_THROTTLE, LUA_LEASE_API_RATE_LIMIT }; - import { randomUUID } from "crypto"; export interface ThrottleResult { @@ -192,19 +189,14 @@ export class UserThrottle { Math.ceil((targetTime - Date.now()) / 1000) + windowSeconds, ); - const count = - typeof this.redis.throttleUser === "function" - ? await this.redis.throttleUser(key, windowStart, limit, targetTime, memberId, ttlSeconds) - : ((await this.redis.eval( - LUA_USER_THROTTLE, - 1, - key, - windowStart, - limit, - targetTime, - memberId, - ttlSeconds, - )) as number); + const count = await this.redis.throttleUser( + key, + windowStart, + limit, + targetTime, + memberId, + ttlSeconds, + ); return { allowed: count <= limit, count, limit }; } @@ -354,28 +346,14 @@ export class ApiRateLimiter { let inFlightRef: Promise | null = null; const executeLease = async (): Promise => { try { - let granted: number; - if (typeof this.redis.leaseApiRateLimit === "function") { - granted = await this.redis.leaseApiRateLimit( - currentKey, - prevKey, - nowMs, - this.windowMs, - limitRpm, - requested, - ); - } else { - granted = (await this.redis.eval( - LUA_LEASE_API_RATE_LIMIT, - 2, - currentKey, - prevKey, - nowMs, - this.windowMs, - limitRpm, - requested, - )) as number; - } + let granted = await this.redis.leaseApiRateLimit( + currentKey, + prevKey, + nowMs, + this.windowMs, + limitRpm, + requested, + ); // On invalid return or negative, fail-open if (typeof granted !== "number" || granted < 0) { diff --git a/src/redis/index.ts b/src/redis/index.ts index a03679a..8f56da8 100644 --- a/src/redis/index.ts +++ b/src/redis/index.ts @@ -3,21 +3,6 @@ import type { Logger } from "@/index.js"; declare module "ioredis" { interface Redis { - checkApiRateLimit( - currentKey: string, - prevKey: string, - now: number | string, - window: number | string, - maxReqs: number | string, - ttl?: number | string, - ): Promise; - checkApiRateLimit1Key( - baseKey: string, - now: number | string, - window: number | string, - maxReqs: number | string, - ttl?: number | string, - ): Promise; leaseApiRateLimit( currentKey: string, prevKey: string, @@ -27,21 +12,6 @@ declare module "ioredis" { requested: number | string, ttl?: number | string, ): Promise; - leaseApiRateLimit1Key( - baseKey: string, - now: number | string, - window: number | string, - maxReqs: number | string, - requested: number | string, - ttl?: number | string, - ): Promise; - throttleProvider( - key: string, - now: number | string, - windowSeconds: number | string, - limit: number | string, - member: string, - ): Promise<[number, number]>; releaseLock(lockKey: string, lockToken: string): Promise; renewLock(lockKey: string, lockToken: string, ttlSeconds: number | string): Promise; throttleUser( @@ -70,49 +40,6 @@ declare module "ioredis" { } } -/** - * Sliding-window counter approximation for API rate limiting. - */ -export const LUA_CHECK_API_RATE_LIMIT = ` - local currentKey = KEYS[1] - local prevKey = KEYS[2] - local now = tonumber(ARGV[1]) - local window = tonumber(ARGV[2]) - local maxReqs = tonumber(ARGV[3]) - local ttl = tonumber(ARGV[4]) or (math.ceil((window * 2) / 1000) + 60) - - if not now or not window or window <= 0 then - return -1 - end - if not maxReqs or maxReqs <= 0 then - return -1 - end - - if not prevKey then - local currentBucket = math.floor(now / window) - local prevBucket = currentBucket - 1 - currentKey = KEYS[1] .. ":" .. currentBucket - prevKey = KEYS[1] .. ":" .. prevBucket - end - - local currentCount = tonumber(redis.call("GET", currentKey) or "0") - local prevCount = tonumber(redis.call("GET", prevKey) or "0") - - local timeIntoCurrent = now % window - local weight = (window - timeIntoCurrent) / window - local estimated = math.floor(prevCount * weight + currentCount) - - if estimated < maxReqs then - local newCount = redis.call("INCR", currentKey) - if newCount == 1 then - redis.call("EXPIRE", currentKey, ttl) - end - return estimated + 1 - end - - return -1 -`; - /** * Sliding-window token leasing for high-throughput API rate limiting. * Atomically checks available capacity and reserves up to N tokens in one round trip. @@ -161,39 +88,6 @@ export const LUA_LEASE_API_RATE_LIMIT = ` return toGrant `; -/** - * Sliding-window rate limiter for external provider dispatch. - */ -export const LUA_THROTTLE_PROVIDER = ` - local key = KEYS[1] - local now = tonumber(ARGV[1]) - local windowSeconds = tonumber(ARGV[2]) - local limit = tonumber(ARGV[3]) - local member = ARGV[4] - - local clearBefore = now - (windowSeconds * 1000) - - -- Cleanup expired scores - redis.call('ZREMRANGEBYSCORE', key, 0, clearBefore) - - -- Get current count - local count = redis.call('ZCARD', key) - - if count >= limit then - -- Find the oldest score - local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES') - if oldest and oldest[2] then - return {0, tonumber(oldest[2])} - end - return {0, now} - end - - -- Add new request - redis.call('ZADD', key, now, member) - redis.call('EXPIRE', key, windowSeconds * 2) - return {1, 0} -`; - /** Release a lock only if we still hold it (value matches our token). */ export const LUA_RELEASE_LOCK = ` if redis.call('GET', KEYS[1]) == ARGV[1] then @@ -275,31 +169,11 @@ export const LUA_MARK_PROCESSED = ` * full script strings over the wire. */ export function registerCustomCommands(redis: Redis): void { - redis.defineCommand("checkApiRateLimit", { - numberOfKeys: 2, - lua: LUA_CHECK_API_RATE_LIMIT, - }); - - redis.defineCommand("checkApiRateLimit1Key", { - numberOfKeys: 1, - lua: LUA_CHECK_API_RATE_LIMIT, - }); - redis.defineCommand("leaseApiRateLimit", { numberOfKeys: 2, lua: LUA_LEASE_API_RATE_LIMIT, }); - redis.defineCommand("leaseApiRateLimit1Key", { - numberOfKeys: 1, - lua: LUA_LEASE_API_RATE_LIMIT, - }); - - redis.defineCommand("throttleProvider", { - numberOfKeys: 1, - lua: LUA_THROTTLE_PROVIDER, - }); - redis.defineCommand("releaseLock", { numberOfKeys: 1, lua: LUA_RELEASE_LOCK, diff --git a/src/services/ai/main.ts b/src/services/ai/main.ts index 456dd52..35e201d 100644 --- a/src/services/ai/main.ts +++ b/src/services/ai/main.ts @@ -1,10 +1,10 @@ import { randomUUID } from "node:crypto"; import { loadEnv, readBaseConfig } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient, type Redis } from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient, Redis } from "@/index.js"; import { - StreamConsumer, - PendingMessageScanner, + type StreamConsumer, + type PendingMessageScanner, StreamProducer, type StreamMessage, } from "@/index.js"; @@ -20,21 +20,25 @@ import { AI_DEFAULTS, } from "@/index.js"; import { generateText } from "ai"; -import { type StreamName } from "@/contracts/streams.js"; import { IdempotencyGuard } from "@/index.js"; import { TemplateRepository } from "@/index.js"; -import { createDatabase } from "@/db/index.js"; import { scheduledPayloads } from "@/db/schema.js"; import { getPriorityBucket, globalEmitter, type WorkerOptions } from "@/shared/index.js"; import { renderWithTemplate, TemplateCache } from "@/templates/index.js"; import { startHealthReporter } from "@/workers/index.js"; +import { + createPriorityProducers, + createStreamConsumer, + createWorkerRuntime, + shutdownWorker, +} from "@/workers/bootstrap.js"; // ─── Bootstrap ───────────────────────────────────────────────────────────── loadEnv(); const config = readBaseConfig(); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -291,40 +295,20 @@ export class AiWorker extends BaseWorker { } export async function startAiWorker() { - logger = createLogger({ name: "ai", level: config.LOG_LEVEL }); - redis = RedisClient.shared({ url: config.REDIS_URL, name: "ai", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "ai", logger }); - sql = dbData.sql; - db = dbData.db; + ({ logger, redis, sql, db } = createWorkerRuntime(config, "ai")); templateRepo = new TemplateRepository(db); templateCache = new TemplateCache(templateRepo); - consumer = new StreamConsumer({ - redis: redis.native, - stream: STREAMS.AI_PENDING as StreamName, - group: CONSUMER_GROUPS.AI, - consumer: `ai-${process.pid}`, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: config.WORKER_CONCURRENCY, - logger, - }); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: STREAMS.AI_PENDING, + group: CONSUMER_GROUPS.AI, + consumer: `ai-${process.pid}`, + batchSize: config.WORKER_CONCURRENCY, + }, + )); - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: STREAMS.AI_PENDING as StreamName, - group: CONSUMER_GROUPS.AI, - consumer: `ai-${process.pid}`, - logger, - }); - - const outboundProducers = { - critical: new StreamProducer({ - redis: redis.native, - stream: STREAMS.OUTBOUND_CRITICAL, - logger, - }), - normal: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_NORMAL, logger }), - low: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_LOW, logger }), - }; + const outboundProducers = createPriorityProducers(redis.native, logger, "OUTBOUND"); const scheduledProducer = new StreamProducer({ redis: redis.native, @@ -403,17 +387,14 @@ export async function startAiWorker() { } export async function stopAiWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) { - clearInterval(healthInterval); - healthInterval = null; - } - if (subscriber) { - subscriber.disconnect(); - subscriber = null; - } - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("ai stopped"); + await shutdownWorker("ai", { + logger, + timers: [healthInterval], + subscriber, + worker, + sql, + redis, + }); + healthInterval = null; + subscriber = null; } diff --git a/src/services/api/handlers.ts b/src/services/api/handlers.ts index 810409d..bf9cf06 100644 --- a/src/services/api/handlers.ts +++ b/src/services/api/handlers.ts @@ -185,6 +185,19 @@ function getQueryParam(ctx: RouteContext, key: string): string | undefined { return (ctx.query as any)[key] ?? undefined; } +/** + * Page size from `?limit=`, clamped to [1, max]. Missing or non-numeric + * values fall back to `fallback`; `undefined` there means "no limit". + */ +function parseLimit( + ctx: RouteContext, + fallback: F, + max = 100, +): number | F { + const n = parseInt(getQueryParam(ctx, "limit") ?? "", 10); + return Number.isNaN(n) ? fallback : Math.min(Math.max(n, 1), max); +} + export function createHandlers(deps: Deps) { const { logger } = deps; @@ -620,8 +633,7 @@ export function createHandlers(deps: Deps) { res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = parseInt(ctx.query.get("limit") || "50", 10); - const limit = isNaN(limitParam) ? 50 : Math.min(limitParam, 100); + const limit = parseLimit(ctx, 50); const cursor = ctx.query.get("cursor"); const templateId = getQueryParam(ctx, "templateId"); @@ -794,8 +806,7 @@ export function createHandlers(deps: Deps) { res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = getQueryParam(ctx, "limit"); - const limit = limitParam ? Math.min(parseInt(limitParam, 10), 100) : undefined; + const limit = parseLimit(ctx, undefined); const channel = getQueryParam(ctx, "channel"); const topic = getQueryParam(ctx, "topic"); let templates = await deps.templateRepo.list(ctx.projectId!); @@ -845,8 +856,7 @@ export function createHandlers(deps: Deps) { res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = getQueryParam(ctx, "limit"); - const limit = limitParam ? Math.min(parseInt(limitParam, 10), 100) : undefined; + const limit = parseLimit(ctx, undefined); const search = getQueryParam(ctx, "search")?.trim().toLowerCase(); let workflows = await deps.workflowRepo.listDefinitions(ctx.projectId!); if (search) { @@ -891,8 +901,7 @@ export function createHandlers(deps: Deps) { res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = parseInt(getQueryParam(ctx, "limit") || "50", 10); - const limit = isNaN(limitParam) ? 50 : Math.min(limitParam, 100); + const limit = parseLimit(ctx, 50); const cursor = getQueryParam(ctx, "cursor"); const search = getQueryParam(ctx, "search"); const segment = getQueryParam(ctx, "segment"); @@ -1371,8 +1380,7 @@ export function createHandlers(deps: Deps) { res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = getQueryParam(ctx, "limit"); - const limit = limitParam ? Math.min(parseInt(limitParam, 10), 100) : 50; + const limit = parseLimit(ctx, 50); const cursor = getQueryParam(ctx, "cursor"); const channel = getQueryParam(ctx, "channel"); @@ -1642,8 +1650,7 @@ code{background:#f4f4f5;padding:.1rem .35rem;border-radius:4px} res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = parseInt(getQueryParam(ctx, "limit") || "20", 10); - const limit = isNaN(limitParam) ? 20 : Math.min(Math.max(limitParam, 1), 100); + const limit = parseLimit(ctx, 20); const search = getQueryParam(ctx, "search")?.trim(); const channel = getQueryParam(ctx, "channel")?.trim(); const since = getQueryParam(ctx, "since")?.trim(); @@ -1819,8 +1826,7 @@ code{background:#f4f4f5;padding:.1rem .35rem;border-radius:4px} res: ServerResponse, ctx: RouteContext, ): Promise { - const limitParam = parseInt(getQueryParam(ctx, "limit") || "100", 10); - const limit = isNaN(limitParam) ? 100 : Math.min(Math.max(limitParam, 1), 500); + const limit = parseLimit(ctx, 100, 500); const channel = getQueryParam(ctx, "channel"); const reason = getQueryParam(ctx, "reason"); const target = getQueryParam(ctx, "target"); diff --git a/src/services/api/main.ts b/src/services/api/main.ts index 82d9485..e1ed1f9 100644 --- a/src/services/api/main.ts +++ b/src/services/api/main.ts @@ -35,25 +35,6 @@ export const API_KEY_INVALIDATION_CHANNEL = "apikey.invalidated"; /** Default window length in milliseconds for API rate limiting (1 minute). */ export const API_RATE_LIMIT_WINDOW_MS = 60_000; -import { LUA_CHECK_API_RATE_LIMIT } from "@/redis/index.js"; -export const LUA_SLIDING_WINDOW_COUNTER = LUA_CHECK_API_RATE_LIMIT; - -/** - * Evaluates the sliding window counter approximation for given parameters. - * Pure reference function matching the Redis Lua script logic. - */ -export function calculateSlidingWindow( - nowMs: number, - windowMs: number, - currentCount: number, - prevCount: number, -): number { - if (windowMs <= 0) return 0; - const timeIntoCurrent = nowMs % windowMs; - const weight = (windowMs - timeIntoCurrent) / windowMs; - return Math.floor(prevCount * weight + currentCount); -} - // ─── Bootstrap ───────────────────────────────────────────────────────────── loadEnv(); diff --git a/src/services/delivery/main.ts b/src/services/delivery/main.ts index 3e562d1..9e37d51 100644 --- a/src/services/delivery/main.ts +++ b/src/services/delivery/main.ts @@ -1,9 +1,9 @@ import { loadEnv, parseConfig, baseConfigSchema } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient, type Redis } from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient, Redis } from "@/index.js"; import { - StreamConsumer, - PendingMessageScanner, + type StreamConsumer, + type PendingMessageScanner, type StreamMessage, StreamProducer, } from "@/index.js"; @@ -18,8 +18,6 @@ import { type RenderedContent, PUBSUB_CHANNELS, } from "@/index.js"; -import { type StreamName } from "@/contracts/streams.js"; -import { createDatabase } from "@/db/index.js"; import { upsertOutboxProviderIds } from "@/db/bulk.js"; import { deliveryOutbox, scheduledPayloads } from "@/db/schema.js"; import { sql as drizzleSql } from "drizzle-orm"; @@ -28,6 +26,13 @@ import { renderWithTemplate, TemplateCache } from "@/templates/index.js"; import { transportRegistry } from "@/index.js"; import { globalEmitter, getPriorityBucket, type WorkerOptions } from "@/shared/index.js"; import { startHealthReporter, NonRetryableError } from "@/workers/index.js"; +import { + createPriorityProducers, + createStreamConsumer, + createWorkerRuntime, + shutdownWorker, + uniqueConsumerId, +} from "@/workers/bootstrap.js"; import { BatchProcessor, CircuitBreaker } from "@/shared/index.js"; import { ProviderThrottle } from "./throttle.js"; import { metrics } from "@/metrics/index.js"; @@ -41,7 +46,7 @@ const deliveryConfigSchema = baseConfigSchema.extend({}); loadEnv(); const config = parseConfig(deliveryConfigSchema, process.env); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -875,27 +880,22 @@ export class DeliveryWorker extends BaseWorker { } export async function startDeliveryWorker() { - logger = createLogger({ name: "delivery", level: config.LOG_LEVEL }); - - redis = RedisClient.shared({ url: config.REDIS_URL, name: "delivery", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "delivery", logger }); - sql = dbData.sql; - db = dbData.db; + ({ logger, redis, sql, db } = createWorkerRuntime(config, "delivery")); // Delivery spends most of its time waiting on providers, so it needs far // more messages in flight than the CPU-bound stages do. const deliveryConcurrency = config.DELIVERY_CONCURRENCY ?? config.WORKER_CONCURRENCY; - const consumerId = `delivery-${process.env.HOSTNAME || process.pid}-${Math.random().toString(36).slice(2, 8)}`; - consumer = new StreamConsumer({ - redis: redis.native, - stream: OUTBOUND_STREAMS as unknown as StreamName[], - group: CONSUMER_GROUPS.DELIVERY, - consumer: consumerId, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: deliveryConcurrency, - bufferAcks: true, - coalesceMs: 2, - logger, - }); + const consumerId = uniqueConsumerId("delivery"); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: OUTBOUND_STREAMS, + group: CONSUMER_GROUPS.DELIVERY, + consumer: consumerId, + batchSize: deliveryConcurrency, + bufferAcks: true, + coalesceMs: 2, + }, + )); scheduledProducer = new StreamProducer({ redis: redis.native, @@ -903,23 +903,7 @@ export async function startDeliveryWorker() { logger, }); - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: OUTBOUND_STREAMS as unknown as StreamName[], - group: CONSUMER_GROUPS.DELIVERY, - consumer: consumerId, - logger, - }); - - enrichedProducers = { - critical: new StreamProducer({ - redis: redis.native, - stream: STREAMS.ENRICHED_CRITICAL, - logger, - }), - normal: new StreamProducer({ redis: redis.native, stream: STREAMS.ENRICHED_NORMAL, logger }), - low: new StreamProducer({ redis: redis.native, stream: STREAMS.ENRICHED_LOW, logger }), - }; + enrichedProducers = createPriorityProducers(redis.native, logger, "ENRICHED"); const contactRepo = new ContactRepository(db); const templateCache = new TemplateCache(new TemplateRepository(db)); @@ -984,17 +968,14 @@ export function getDeliveryWorker(): DeliveryWorker | undefined { // ─── Shutdown ────────────────────────────────────────────────────────────── export async function stopDeliveryWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) { - clearInterval(healthInterval); - healthInterval = null; - } - if (subscriber) { - subscriber.disconnect(); - subscriber = null; - } - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("delivery stopped"); + await shutdownWorker("delivery", { + logger, + timers: [healthInterval], + subscriber, + worker, + sql, + redis, + }); + healthInterval = null; + subscriber = null; } diff --git a/src/services/delivery/throttle.ts b/src/services/delivery/throttle.ts index 3d727ee..033188d 100644 --- a/src/services/delivery/throttle.ts +++ b/src/services/delivery/throttle.ts @@ -1,64 +1,16 @@ import type { Redis } from "ioredis"; -import { LUA_THROTTLE_PROVIDER as LUA_THROTTLE } from "@/redis/index.js"; import { ApiRateLimiter } from "@/rate-limiter/index.js"; -export { LUA_THROTTLE }; - export interface ThrottleResult { allowed: boolean; retryAfterMs: number; } -export async function throttleProvider( - redis: Redis, - channel: string, - config: { limit: number; windowSeconds: number }, - logger: any, -): Promise { - const key = `rate-limit:provider:${channel}`; - const now = Date.now(); - const zmember = `${now}:${Math.random()}`; - - const result = - typeof redis.throttleProvider === "function" - ? await redis.throttleProvider( - key, - now.toString(), - config.windowSeconds.toString(), - config.limit.toString(), - zmember, - ) - : ((await redis.eval( - LUA_THROTTLE, - 1, - key, - now.toString(), - config.windowSeconds.toString(), - config.limit.toString(), - zmember, - )) as [number, number]); - - const allowed = result[0] === 1; - const oldestTimestamp = result[1]; - - let retryAfterMs = 0; - if (!allowed) { - retryAfterMs = Math.max(0, oldestTimestamp + config.windowSeconds * 1000 - now); - logger.warn( - { channel, limit: config.limit, windowSeconds: config.windowSeconds, retryAfterMs }, - "Provider rate limit hit — task must be rescheduled", - ); - } - - return { allowed, retryAfterMs }; -} - /** * Provider rate limit enforced through locally leased tokens. * - * `throttleProvider` makes one Redis round trip per message against a single - * sorted set per channel, so every delivery process in the cluster serialises - * on one hot key and the set grows by one member per send. Here each process + * A per-message check against one sorted set per channel would serialise every + * delivery process in the cluster on a single hot key. Instead each process * leases a chunk of the window's budget and spends it in memory, contacting * Redis only when its chunk runs low. The window is a sliding estimate over * two fixed buckets rather than an exact log, the same trade the API ingress diff --git a/src/services/engine/main.ts b/src/services/engine/main.ts index 4f59ab0..b8b7c9c 100644 --- a/src/services/engine/main.ts +++ b/src/services/engine/main.ts @@ -1,10 +1,10 @@ import { randomUUID } from "node:crypto"; import { loadEnv, readBaseConfig } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient, type Redis } from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient, Redis } from "@/index.js"; import { - StreamConsumer, - PendingMessageScanner, + type StreamConsumer, + type PendingMessageScanner, StreamProducer, type StreamMessage, } from "@/index.js"; @@ -20,11 +20,9 @@ import { type DispatchedTaskPayload, type StreamEvent, } from "@/index.js"; -import { type StreamName } from "@/contracts/streams.js"; import { IdempotencyGuard } from "@/index.js"; import { UserThrottle, ProjectSettingsCache } from "@/index.js"; import { TemplateRepository, ContactRepository, ProjectRepository } from "@/index.js"; -import { createDatabase } from "@/db/index.js"; import { scheduledPayloads, suppressions } from "@/db/schema.js"; import { and, eq } from "drizzle-orm"; import { @@ -38,6 +36,13 @@ import { import { TemplateCache } from "@/templates/index.js"; import { buildUnsubscribeHeaders } from "@/unsubscribe/index.js"; import { startHealthReporter } from "@/workers/index.js"; +import { + createPriorityProducers, + createStreamConsumer, + createWorkerRuntime, + shutdownWorker, + uniqueConsumerId, +} from "@/workers/bootstrap.js"; function localTimeToUtc( year: number, @@ -191,7 +196,7 @@ const NO_IDEMPOTENCY = { loadEnv(); const config = readBaseConfig(); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -747,44 +752,24 @@ export class EngineWorker extends BaseWorker { let subscriber: any = null; export async function startEngineWorker() { - logger = createLogger({ name: "engine", level: config.LOG_LEVEL }); - redis = RedisClient.shared({ url: config.REDIS_URL, name: "engine", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "engine", logger }); - sql = dbData.sql; - db = dbData.db; + ({ logger, redis, sql, db } = createWorkerRuntime(config, "engine")); templateRepo = new TemplateRepository(db); templateCache = new TemplateCache(templateRepo); const contactRepo = new ContactRepository(db); - const consumerId = `engine-${process.env.HOSTNAME || process.pid}-${Math.random().toString(36).slice(2, 8)}`; - consumer = new StreamConsumer({ - redis: redis.native, - stream: ENRICHED_STREAMS as unknown as StreamName[], - group: CONSUMER_GROUPS.ENGINE, - consumer: consumerId, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: config.WORKER_CONCURRENCY, - bufferAcks: true, - coalesceMs: 2, - logger, - }); + const consumerId = uniqueConsumerId("engine"); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: ENRICHED_STREAMS, + group: CONSUMER_GROUPS.ENGINE, + consumer: consumerId, + batchSize: config.WORKER_CONCURRENCY, + bufferAcks: true, + coalesceMs: 2, + }, + )); - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: ENRICHED_STREAMS as unknown as StreamName[], - group: CONSUMER_GROUPS.ENGINE, - consumer: consumerId, - logger, - }); - - const outboundProducers = { - critical: new StreamProducer({ - redis: redis.native, - stream: STREAMS.OUTBOUND_CRITICAL, - logger, - }), - normal: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_NORMAL, logger }), - low: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_LOW, logger }), - }; + const outboundProducers = createPriorityProducers(redis.native, logger, "OUTBOUND"); const scheduledProducer = new StreamProducer({ redis: redis.native, @@ -876,17 +861,14 @@ export function getEngineWorker(): EngineWorker | undefined { // ─── Shutdown ────────────────────────────────────────────────────────────── export async function stopEngineWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) { - clearInterval(healthInterval); - healthInterval = null; - } - if (subscriber) { - subscriber.disconnect(); - subscriber = null; - } - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("engine stopped"); + await shutdownWorker("engine", { + logger, + timers: [healthInterval], + subscriber, + worker, + sql, + redis, + }); + healthInterval = null; + subscriber = null; } diff --git a/src/services/enricher/main.ts b/src/services/enricher/main.ts index ea3af61..c9e9842 100644 --- a/src/services/enricher/main.ts +++ b/src/services/enricher/main.ts @@ -1,16 +1,10 @@ import { randomUUID } from "node:crypto"; import { loadEnv, readBaseConfig } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient } from "@/index.js"; -import { - StreamConsumer, - PendingMessageScanner, - StreamProducer, - type StreamMessage, -} from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient } from "@/index.js"; +import { type StreamConsumer, type PendingMessageScanner, type StreamMessage } from "@/index.js"; import { BaseWorker, type ProcessResult } from "@/index.js"; import { - STREAMS, INBOUND_STREAMS, CONSUMER_GROUPS, registry, @@ -19,9 +13,7 @@ import { type NotificationEnrichedPayload, type StreamEvent, } from "@/index.js"; -import { type StreamName } from "@/contracts/streams.js"; import { IdempotencyGuard } from "@/index.js"; -import { createDatabase } from "@/db/index.js"; import { UserRepository, PreferenceRepository, @@ -31,13 +23,20 @@ import { import { TemplateCache } from "@/templates/index.js"; import { getPriorityBucket, type WorkerOptions } from "@/shared/index.js"; import { startHealthReporter } from "@/workers/index.js"; +import { + createPriorityProducers, + createStreamConsumer, + createWorkerRuntime, + shutdownWorker, + uniqueConsumerId, +} from "@/workers/bootstrap.js"; // ─── Bootstrap ───────────────────────────────────────────────────────────── loadEnv(); const config = readBaseConfig(); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -534,46 +533,26 @@ export class EnricherWorker extends BaseWorker { } export async function startEnricherWorker() { - logger = createLogger({ name: "enricher", level: config.LOG_LEVEL }); - redis = RedisClient.shared({ url: config.REDIS_URL, name: "enricher", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "enricher", logger }); - sql = dbData.sql; - db = dbData.db; + ({ logger, redis, sql, db } = createWorkerRuntime(config, "enricher")); // A fused pipeline holds the enricher's slot through engine and delivery, so // it needs delivery's concurrency, not the CPU-bound default. const enricherConcurrency = config.PIPELINE_FUSED ? Math.max(config.WORKER_CONCURRENCY, config.DELIVERY_CONCURRENCY ?? config.WORKER_CONCURRENCY) : config.WORKER_CONCURRENCY; - const consumerId = `enricher-${process.env.HOSTNAME || process.pid}-${Math.random().toString(36).slice(2, 8)}`; - consumer = new StreamConsumer({ - redis: redis.native, - stream: INBOUND_STREAMS as unknown as StreamName[], - group: CONSUMER_GROUPS.ENRICHER, - consumer: consumerId, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: enricherConcurrency, - bufferAcks: true, - coalesceMs: 2, - logger, - }); + const consumerId = uniqueConsumerId("enricher"); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: INBOUND_STREAMS, + group: CONSUMER_GROUPS.ENRICHER, + consumer: consumerId, + batchSize: enricherConcurrency, + bufferAcks: true, + coalesceMs: 2, + }, + )); - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: INBOUND_STREAMS as unknown as StreamName[], - group: CONSUMER_GROUPS.ENRICHER, - consumer: consumerId, - logger, - }); - - const producers = { - critical: new StreamProducer({ - redis: redis.native, - stream: STREAMS.ENRICHED_CRITICAL, - logger, - }), - normal: new StreamProducer({ redis: redis.native, stream: STREAMS.ENRICHED_NORMAL, logger }), - low: new StreamProducer({ redis: redis.native, stream: STREAMS.ENRICHED_LOW, logger }), - }; + const producers = createPriorityProducers(redis.native, logger, "ENRICHED"); const idempotency = new IdempotencyGuard({ redis: redis.native, @@ -624,13 +603,6 @@ export function getEnricherWorker(): EnricherWorker | undefined { // ─── Shutdown ────────────────────────────────────────────────────────────── export async function stopEnricherWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) { - clearInterval(healthInterval); - healthInterval = null; - } - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("enricher stopped"); + await shutdownWorker("enricher", { logger, timers: [healthInterval], worker, sql, redis }); + healthInterval = null; } diff --git a/src/services/events/main.ts b/src/services/events/main.ts index 2737621..034ab0e 100644 --- a/src/services/events/main.ts +++ b/src/services/events/main.ts @@ -1,25 +1,24 @@ import { loadEnv, readBaseConfig } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient } from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient } from "@/index.js"; import { - StreamConsumer, - PendingMessageScanner, + type StreamConsumer, + type PendingMessageScanner, StreamProducer, type StreamMessage, } from "@/index.js"; import { BaseWorker } from "@/index.js"; import { STREAMS, CONSUMER_GROUPS, buildStreamEvent } from "@/index.js"; -import { type StreamName } from "@/contracts/streams.js"; -import { createDatabase } from "@/db/index.js"; import { insertMessageLogs } from "@/db/bulk.js"; import { workflowWaiters, workflowSteps, workflowInstances } from "@/db/schema.js"; import { eq, and, or, isNull, gt } from "drizzle-orm"; import { type WorkerOptions } from "@/shared/index.js"; import { startHealthReporter } from "@/workers/index.js"; +import { createStreamConsumer, createWorkerRuntime, shutdownWorker } from "@/workers/bootstrap.js"; loadEnv(); const config = readBaseConfig(); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -360,11 +359,7 @@ export class EventWorker extends BaseWorker { } export async function startEventWorker() { - logger = createLogger({ name: "event-worker", level: config.LOG_LEVEL }); - redis = RedisClient.shared({ url: config.REDIS_URL, name: "events", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "events", logger }); - sql = dbData.sql; - db = dbData.db; + ({ logger, redis, sql, db } = createWorkerRuntime(config, "events", "event-worker")); workflowProducer = new StreamProducer({ redis: redis.native, stream: STREAMS.WORKFLOW_INBOUND, @@ -376,26 +371,18 @@ export async function startEventWorker() { config.WORKER_CONCURRENCY, config.DELIVERY_CONCURRENCY ?? config.WORKER_CONCURRENCY, ); - consumer = new StreamConsumer({ - redis: redis.native, - stream: STREAMS.EVENTS_INBOUND as StreamName, - group: CONSUMER_GROUPS.EVENTS as any, - consumer: `events-${process.pid}`, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: eventsConcurrency, - bufferAcks: true, - // Log rows are not latency-sensitive; let reads fill up. - coalesceMs: 10, - logger, - }); - - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: STREAMS.EVENTS_INBOUND as StreamName, - group: CONSUMER_GROUPS.EVENTS as any, - consumer: `events-${process.pid}`, - logger, - }); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: STREAMS.EVENTS_INBOUND, + group: CONSUMER_GROUPS.EVENTS, + consumer: `events-${process.pid}`, + batchSize: eventsConcurrency, + bufferAcks: true, + // Log rows are not latency-sensitive; let reads fill up. + coalesceMs: 10, + }, + )); worker = new EventWorker({ consumer, @@ -413,10 +400,6 @@ export async function startEventWorker() { } export async function stopEventWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) clearInterval(healthInterval); - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("event worker stopped"); + await shutdownWorker("event worker", { logger, timers: [healthInterval], worker, sql, redis }); + healthInterval = null; } diff --git a/src/services/scheduler/main.ts b/src/services/scheduler/main.ts index 772d04c..31b36cf 100644 --- a/src/services/scheduler/main.ts +++ b/src/services/scheduler/main.ts @@ -1,13 +1,8 @@ import { randomUUID } from "node:crypto"; import { loadEnv, readBaseConfig } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient, type Redis } from "@/index.js"; -import { - StreamConsumer, - PendingMessageScanner, - StreamProducer, - type StreamMessage, -} from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient, Redis } from "@/index.js"; +import { type StreamConsumer, type PendingMessageScanner, type StreamMessage } from "@/index.js"; import { BaseWorker } from "@/index.js"; import { STREAMS, @@ -16,17 +11,22 @@ import { buildStreamEvent, type NotificationScheduledPayload, } from "@/index.js"; -import { getPriorityBucket, type WorkerOptions, LUA_SCHEDULER_POLL } from "@/shared/index.js"; -import { createDatabase } from "@/db/index.js"; +import { getPriorityBucket, type WorkerOptions } from "@/shared/index.js"; import { scheduledPayloads } from "@/db/schema.js"; import { inArray } from "drizzle-orm"; import { startHealthReporter } from "@/workers/index.js"; +import { + createPriorityProducers, + createStreamConsumer, + createWorkerRuntime, + shutdownWorker, +} from "@/workers/bootstrap.js"; // ─── Bootstrap ───────────────────────────────────────────────────────────── loadEnv(); const config = readBaseConfig(); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -106,11 +106,7 @@ export async function executeSchedulerPoll( for (let i = 0; i < 16; i++) { const shardKey = `notif:scheduled:zset:${i}`; - if (typeof pipeline.schedulerPoll === "function") { - pipeline.schedulerPoll(shardKey, now, perShardLimit, visibilityTimeout); - } else { - pipeline.eval(LUA_SCHEDULER_POLL, 1, shardKey, now, perShardLimit, visibilityTimeout); - } + pipeline.schedulerPoll(shardKey, now, perShardLimit, visibilityTimeout); } const results = await pipeline.exec(); @@ -206,56 +202,24 @@ export async function executeSchedulerPoll( return false; } finally { if (acquired) { - if (typeof redis.releaseLock === "function") { - await redis.releaseLock(lockKey, lockOwner).catch(() => {}); - } else { - // Delete the lock only if we still own it - const lua = ` - if redis.call("get", KEYS[1]) == ARGV[1] then - return redis.call("del", KEYS[1]) - else - return 0 - end - `; - await redis.eval(lua, 1, lockKey, lockOwner).catch(() => {}); - } + await redis.releaseLock(lockKey, lockOwner).catch(() => {}); } } } export async function startSchedulerWorker() { - logger = createLogger({ name: "scheduler", level: config.LOG_LEVEL }); - redis = RedisClient.shared({ url: config.REDIS_URL, name: "scheduler", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "scheduler", logger }); - sql = dbData.sql; - db = dbData.db; - consumer = new StreamConsumer({ - redis: redis.native, - stream: STREAMS.SCHEDULED, - group: CONSUMER_GROUPS.SCHEDULER, - consumer: `scheduler-${process.pid}`, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: config.WORKER_CONCURRENCY, - logger, - }); + ({ logger, redis, sql, db } = createWorkerRuntime(config, "scheduler")); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: STREAMS.SCHEDULED, + group: CONSUMER_GROUPS.SCHEDULER, + consumer: `scheduler-${process.pid}`, + batchSize: config.WORKER_CONCURRENCY, + }, + )); - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: STREAMS.SCHEDULED, - group: CONSUMER_GROUPS.SCHEDULER, - consumer: `scheduler-${process.pid}`, - logger, - }); - - const outboundProducers = { - critical: new StreamProducer({ - redis: redis.native, - stream: STREAMS.OUTBOUND_CRITICAL, - logger, - }), - normal: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_NORMAL, logger }), - low: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_LOW, logger }), - }; + const outboundProducers = createPriorityProducers(redis.native, logger, "OUTBOUND"); // ─── Stage 4a: Scheduled notification processor ───────────────────────────── // @@ -300,18 +264,14 @@ export async function startSchedulerWorker() { // ─── Shutdown ────────────────────────────────────────────────────────────── export async function stopSchedulerWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) { - clearInterval(healthInterval); - healthInterval = null; - } - if (pollTimeout) { - clearTimeout(pollTimeout); - pollTimeout = null; - } isPolling = false; - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("scheduler stopped"); + await shutdownWorker("scheduler", { + logger, + timers: [healthInterval, pollTimeout], + worker, + sql, + redis, + }); + healthInterval = null; + pollTimeout = null; } diff --git a/src/services/workflow/main.ts b/src/services/workflow/main.ts index 14d6e79..c8d075d 100644 --- a/src/services/workflow/main.ts +++ b/src/services/workflow/main.ts @@ -1,17 +1,15 @@ import { randomUUID } from "node:crypto"; import { loadEnv, readBaseConfig } from "@/index.js"; -import { createLogger } from "@/index.js"; -import { RedisClient, type Redis } from "@/index.js"; +import type { Logger } from "@/index.js"; +import type { RedisClient, Redis } from "@/index.js"; import { - StreamConsumer, - PendingMessageScanner, + type StreamConsumer, + type PendingMessageScanner, StreamProducer, type StreamMessage, } from "@/index.js"; import { BaseWorker } from "@/index.js"; import { STREAMS, CONSUMER_GROUPS, buildStreamEvent } from "@/index.js"; -import { type StreamName } from "@/contracts/streams.js"; -import { createDatabase } from "@/db/index.js"; import { workflowInstances, workflowSteps, @@ -26,13 +24,9 @@ import { type WorkflowContext, type WorkflowStepContext, } from "@/workflows/index.js"; -import { - type WorkerOptions, - LUA_SCHEDULER_POLL, - LUA_RELEASE_LOCK, - LUA_RENEW_LOCK, -} from "@/shared/index.js"; +import { type WorkerOptions } from "@/shared/index.js"; import { startHealthReporter } from "@/workers/index.js"; +import { createStreamConsumer, createWorkerRuntime, shutdownWorker } from "@/workers/bootstrap.js"; /** How long a workflow instance lock is held before it self-expires. */ const WORKFLOW_LOCK_TTL_SECONDS = 60; @@ -43,7 +37,7 @@ const WORKFLOW_TIMER_VISIBILITY_MS = 60_000; loadEnv(); const config = readBaseConfig(); -let logger: ReturnType; +let logger: Logger; let redis: RedisClient; let sql: any; let db: any; @@ -202,19 +196,11 @@ export class WorkflowWorker extends BaseWorker { // Keep the lock alive while the handler runs; without this a handler that // outlives the TTL lets a second resume execute the same steps in parallel. const renewTimer = setInterval(() => { - const renewPromise = - typeof this.redisCli.renewLock === "function" - ? this.redisCli.renewLock(lockKey, lockToken, WORKFLOW_LOCK_TTL_SECONDS) - : this.redisCli.eval( - LUA_RENEW_LOCK, - 1, - lockKey, - lockToken, - String(WORKFLOW_LOCK_TTL_SECONDS), - ); - void renewPromise.catch((err: unknown) => { - this.logger.warn({ err, instanceId }, "failed to renew workflow lock"); - }); + void this.redisCli + .renewLock(lockKey, lockToken, WORKFLOW_LOCK_TTL_SECONDS) + .catch((err: unknown) => { + this.logger.warn({ err, instanceId }, "failed to renew workflow lock"); + }); }, WORKFLOW_LOCK_RENEW_MS); try { @@ -449,11 +435,7 @@ export class WorkflowWorker extends BaseWorker { } finally { clearInterval(renewTimer); // Compare-and-delete: never release a lock a later process re-acquired. - if (typeof this.redisCli.releaseLock === "function") { - await this.redisCli.releaseLock(lockKey, lockToken); - } else { - await this.redisCli.eval(LUA_RELEASE_LOCK, 1, lockKey, lockToken); - } + await this.redisCli.releaseLock(lockKey, lockToken); } await Promise.all(publishPromises); @@ -468,11 +450,7 @@ export function __injectForTests(r: any, d: any, wp: any, np: any) { } export async function startWorkflowWorker() { - logger = createLogger({ name: "workflow-worker", level: config.LOG_LEVEL }); - redis = RedisClient.shared({ url: config.REDIS_URL, name: "workflow", logger }); - const dbData = createDatabase({ url: config.DATABASE_URL, applicationName: "workflow", logger }); - sql = dbData.sql; - db = dbData.db; + ({ logger, redis, sql, db } = createWorkerRuntime(config, "workflow", "workflow-worker")); notificationProducer = new StreamProducer({ redis: redis.native, stream: STREAMS.INBOUND_NORMAL, @@ -483,23 +461,15 @@ export async function startWorkflowWorker() { stream: STREAMS.WORKFLOW_INBOUND, logger, }); - consumer = new StreamConsumer({ - redis: redis.native, - stream: STREAMS.WORKFLOW_INBOUND as StreamName, - group: CONSUMER_GROUPS.WORKFLOW as any, - consumer: `workflow-${process.pid}`, - dlqStream: STREAMS.DEAD_LETTER, - batchSize: config.WORKER_CONCURRENCY, - logger, - }); - - pendingScanner = new PendingMessageScanner({ - redis: redis.native, - stream: STREAMS.WORKFLOW_INBOUND as StreamName, - group: CONSUMER_GROUPS.WORKFLOW as any, - consumer: `workflow-${process.pid}`, - logger, - }); + ({ consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: STREAMS.WORKFLOW_INBOUND, + group: CONSUMER_GROUPS.WORKFLOW, + consumer: `workflow-${process.pid}`, + batchSize: config.WORKER_CONCURRENCY, + }, + )); worker = new WorkflowWorker({ consumer, @@ -517,23 +487,12 @@ export async function startWorkflowWorker() { void (async () => { try { const now = Date.now(); - const tasks = ( - typeof redis.native.schedulerPoll === "function" - ? await redis.native.schedulerPoll( - "notif:workflow:timers", - now, - 100, - WORKFLOW_TIMER_VISIBILITY_MS, - ) - : await redis.native.eval( - LUA_SCHEDULER_POLL, - 1, - "notif:workflow:timers", - now, - 100, - WORKFLOW_TIMER_VISIBILITY_MS, - ) - ) as string[]; + const tasks = await redis.native.schedulerPoll( + "notif:workflow:timers", + now, + 100, + WORKFLOW_TIMER_VISIBILITY_MS, + ); for (const taskStr of tasks) { const task = JSON.parse(taskStr); @@ -637,12 +596,14 @@ export async function startWorkflowWorker() { } export async function stopWorkflowWorker(): Promise { - logger?.info("shutdown initiated"); - if (healthInterval) clearInterval(healthInterval); - if (pollInterval) clearInterval(pollInterval); - if (reaperInterval) clearInterval(reaperInterval); - if (worker) await worker.stop(); - if (sql) await sql.end(); - if (redis) await redis.disconnect(); - logger?.info("workflow worker stopped"); + await shutdownWorker("workflow worker", { + logger, + timers: [healthInterval, pollInterval, reaperInterval], + worker, + sql, + redis, + }); + healthInterval = null; + pollInterval = null; + reaperInterval = null; } diff --git a/src/shared/index.ts b/src/shared/index.ts index 27b89c9..66effe7 100644 --- a/src/shared/index.ts +++ b/src/shared/index.ts @@ -33,7 +33,6 @@ export * from "./cache.js"; export * from "./utils.js"; export * from "./semaphore.js"; export * from "./batch-processor.js"; -export * from "./cache.js"; export * from "./circuit-breaker.js"; export * from "./dataloader.js"; export { type WorkerOptions } from "@/workers/index.js"; diff --git a/src/shared/utils.ts b/src/shared/utils.ts index 5535d93..1cebd59 100644 --- a/src/shared/utils.ts +++ b/src/shared/utils.ts @@ -17,20 +17,3 @@ export function normaliseTarget(target: string): string { const trimmed = target.trim(); return trimmed.includes("@") ? trimmed.toLowerCase() : trimmed; } - -export { LUA_SCHEDULER_POLL } from "@/redis/index.js"; - -export const LUA_SCHEDULER_CLAIM = ` - local payloadKey = KEYS[1] - local claimedKey = KEYS[2] - - if redis.call('EXISTS', payloadKey) == 1 then - redis.call('RENAME', payloadKey, claimedKey) - return redis.call('GET', claimedKey) - elseif redis.call('EXISTS', claimedKey) == 1 then - return redis.call('GET', claimedKey) - else - return nil - end -`; -export { LUA_RELEASE_LOCK, LUA_RENEW_LOCK } from "@/redis/index.js"; diff --git a/src/templates/index.ts b/src/templates/index.ts index cc030ae..b666399 100644 --- a/src/templates/index.ts +++ b/src/templates/index.ts @@ -1,69 +1,2 @@ -import type { RenderedContent } from "@/contracts/index.js"; export * from "./render.js"; export * from "./cache.js"; - -// ─── Types ────────────────────────────────────────────────────────────────── - -export interface TemplateContext { - eventType: string; - templateVariables: Record; - locale: string; - timezone: string; - deeplinkScheme: string; -} - -export type TemplateRenderer = (ctx: TemplateContext) => RenderedContent; - -/** Coerce a template variable to a non-empty string, or undefined. */ -function asText(value: unknown): string | undefined { - return typeof value === "string" && value.length > 0 ? value : undefined; -} - -// ─── TemplateRegistry ──────────────────────────────────────────────────────── -// -// Open registry — new event types register a renderer without touching this file. -// The default renderer falls back to the i18n translation table, so most event -// types work without a custom renderer. - -class TemplateRegistry { - private readonly renderers = new Map(); - - /** - * Register a custom renderer for an event type. - * Overwrites any previous registration for the same type. - */ - register(eventType: string, renderer: TemplateRenderer): void { - this.renderers.set(eventType, renderer); - } - - /** Render content for the given context, falling back to the i18n table. */ - render(ctx: TemplateContext): RenderedContent { - const renderer = this.renderers.get(ctx.eventType); - if (renderer) return renderer(ctx); - return defaultRenderer(ctx); - } - - has(eventType: string): boolean { - return this.renderers.has(eventType); - } - - registeredTypes(): string[] { - return [...this.renderers.keys()]; - } -} - -function defaultRenderer(ctx: TemplateContext): RenderedContent { - const vars = ctx.templateVariables; - - // Fall back to any caller-supplied title/body in the payload, then to a generic label. - const subject = asText(vars.subject ?? vars.title); - const body = asText(vars.body ?? vars.message) ?? `Notification: ${ctx.eventType}`; - - return { content: { subject, body } }; -} - -export const templateRegistry = new TemplateRegistry(); - -export function renderTemplate(ctx: TemplateContext): RenderedContent { - return templateRegistry.render(ctx); -} diff --git a/src/workers/bootstrap.ts b/src/workers/bootstrap.ts new file mode 100644 index 0000000..4e4bba1 --- /dev/null +++ b/src/workers/bootstrap.ts @@ -0,0 +1,124 @@ +import type { Redis } from "ioredis"; +import { STREAMS, type ConsumerGroup, type StreamName } from "@/contracts/streams.js"; +import { createDatabase, type Db, type Sql } from "@/db/index.js"; +import { createLogger, type LogLevel, type Logger } from "@/logger/index.js"; +import { PendingMessageScanner, StreamConsumer, StreamProducer } from "@/queue/index.js"; +import { RedisClient } from "@/redis/index.js"; +import type { BaseWorker } from "./index.js"; + +// Shared start/stop plumbing for the pipeline services under src/services. +// Each service still owns its worker and anything specific to it (polling +// loops, pub/sub subscribers); this only covers the parts they all repeat. + +export interface WorkerRuntime { + logger: Logger; + redis: RedisClient; + sql: Sql; + db: Db; +} + +/** Logger, shared Redis connection and Postgres pool for one service. */ +export function createWorkerRuntime( + config: { LOG_LEVEL: LogLevel; REDIS_URL: string; DATABASE_URL: string }, + name: string, + loggerName = name, +): WorkerRuntime { + const logger = createLogger({ name: loggerName, level: config.LOG_LEVEL }); + const redis = RedisClient.shared({ url: config.REDIS_URL, name, logger }); + const { sql, db } = createDatabase({ + url: config.DATABASE_URL, + applicationName: name, + logger, + }); + return { logger, redis, sql, db }; +} + +/** + * A consumer id unique to this process, so replicas sharing a host or PID + * namespace never read as the same group member. + */ +export function uniqueConsumerId(name: string): string { + const host = process.env.HOSTNAME || process.pid; + return `${name}-${host}-${Math.random().toString(36).slice(2, 8)}`; +} + +/** + * A group consumer and the scanner that reclaims its stalled messages. The two + * must agree on stream, group and consumer id, so they are built together. + */ +export function createStreamConsumer( + runtime: { redis: RedisClient; logger: Logger }, + options: { + stream: StreamName | readonly StreamName[]; + group: ConsumerGroup; + consumer: string; + batchSize: number; + bufferAcks?: boolean; + coalesceMs?: number; + }, +): { consumer: StreamConsumer; pendingScanner: PendingMessageScanner } { + const { redis, logger } = runtime; + const stream = (Array.isArray(options.stream) ? [...options.stream] : options.stream) as + StreamName | StreamName[]; + const consumer = new StreamConsumer({ + redis: redis.native, + stream, + group: options.group, + consumer: options.consumer, + dlqStream: STREAMS.DEAD_LETTER, + batchSize: options.batchSize, + bufferAcks: options.bufferAcks, + coalesceMs: options.coalesceMs, + logger, + }); + const pendingScanner = new PendingMessageScanner({ + redis: redis.native, + stream, + group: options.group, + consumer: options.consumer, + logger, + }); + return { consumer, pendingScanner }; +} + +export type PriorityProducers = Record<"critical" | "normal" | "low", StreamProducer>; + +/** One producer per priority lane of a pipeline stage. */ +export function createPriorityProducers( + redis: Redis, + logger: Logger, + stage: "INBOUND" | "ENRICHED" | "OUTBOUND", +): PriorityProducers { + const producer = (stream: StreamName) => new StreamProducer({ redis, stream, logger }); + return { + critical: producer(STREAMS[`${stage}_CRITICAL`]), + normal: producer(STREAMS[`${stage}_NORMAL`]), + low: producer(STREAMS[`${stage}_LOW`]), + }; +} + +/** + * Stops a service in dependency order: timers and subscribers first so nothing + * new starts, then the worker drains, then the connections it was using close. + */ +export async function shutdownWorker( + label: string, + resources: { + logger?: Logger; + timers?: (NodeJS.Timeout | null | undefined)[]; + subscriber?: { disconnect(): void } | null; + worker?: BaseWorker; + sql?: Sql; + redis?: RedisClient; + }, +): Promise { + const { logger, timers = [], subscriber, worker, sql, redis } = resources; + logger?.info("shutdown initiated"); + // clearTimeout also cancels intervals in Node. + for (const timer of timers) if (timer) clearTimeout(timer); + subscriber?.disconnect(); + if (worker) await worker.stop(); + if (sql) await sql.end(); + if (redis) await redis.disconnect(); + logger?.info(`${label} stopped`); +} diff --git a/tests/api-rate-limiter.test.ts b/tests/api-rate-limiter.test.ts index 4a0cbe1..8c0600b 100644 --- a/tests/api-rate-limiter.test.ts +++ b/tests/api-rate-limiter.test.ts @@ -42,7 +42,6 @@ describe("ApiRateLimiter (In-Memory Token Leasing)", () => { beforeEach(() => { mockRedis = { leaseApiRateLimit: vi.fn(), - eval: vi.fn(), }; mockLogger = { warn: vi.fn(), diff --git a/tests/api.test.ts b/tests/api.test.ts index da6576c..45dba5e 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -35,13 +35,7 @@ function createMockRes() { return res; } -import { - extractAuthToken, - getClientIp, - calculateSlidingWindow, - LUA_SLIDING_WINDOW_COUNTER, - API_RATE_LIMIT_WINDOW_MS, -} from "@/services/api/main.js"; +import { extractAuthToken, getClientIp, API_RATE_LIMIT_WINDOW_MS } from "@/services/api/main.js"; describe("extractAuthToken", () => { it("extracts Bearer token from authorization header", () => { @@ -550,6 +544,36 @@ describe("API Handlers", () => { expect(deps.userRepo.list).toHaveBeenCalledWith("proj_1", 10, undefined, undefined); }); + it.each([ + ["limit=abc", 50], + ["limit=-5", 1], + ["limit=0", 1], + ["limit=999", 100], + ["", 50], + ])("listUsers clamps %j to a usable page size", async (qs, expected) => { + deps.userRepo.list = vi.fn().mockResolvedValue({ users: [], nextCursor: null }); + const res = createMockRes(); + await handlers.listUsers({} as any, res, { + projectId: "proj_1", + params: {}, + query: new URLSearchParams(qs), + } as any); + expect(deps.userRepo.list).toHaveBeenCalledWith("proj_1", expected, undefined, undefined); + }); + + it("listTemplates treats a negative limit as the minimum page size", async () => { + deps.templateRepo.list = vi.fn().mockResolvedValue([{ id: "a" }, { id: "b" }, { id: "c" }]); + const res = createMockRes(); + await handlers.listTemplates({} as any, res, { + projectId: "proj_1", + params: {}, + query: new URLSearchParams("limit=-1"), + } as any); + expect(res.statusCode).toBe(200); + // Previously slice(0, -1) returned everything except the last template. + expect(JSON.parse(res.body).templates).toEqual([{ id: "a" }]); + }); + it("listProjects", async () => { const res = createMockRes(); await handlers.listProjects({} as any, res, { @@ -1172,130 +1196,8 @@ describe("Router param decoding", () => { }); }); -describe("API Rate Limiting - Sliding Window Counter", () => { - it("exports valid window size and Lua script", () => { +describe("API Rate Limiting", () => { + it("uses a one-minute window", () => { expect(API_RATE_LIMIT_WINDOW_MS).toBe(60_000); - expect(typeof LUA_SLIDING_WINDOW_COUNTER).toBe("string"); - expect(LUA_SLIDING_WINDOW_COUNTER).toContain("INCR"); - expect(LUA_SLIDING_WINDOW_COUNTER).toContain("EXPIRE"); - expect(LUA_SLIDING_WINDOW_COUNTER).toContain("math.floor(prevCount * weight + currentCount)"); - }); - - describe("calculateSlidingWindow", () => { - it("returns only currentCount when now is at the start of a window (weight = 1.0 on previous)", () => { - // At nowMs = 60_000 (start of window 1), weight = (60000 - 0) / 60000 = 1.0 - const count = calculateSlidingWindow(60_000, 60_000, 5, 20); - expect(count).toBe(25); // 20 * 1.0 + 5 - }); - - it("weights previous window count proportionally across window progression", () => { - // At 30s into a 60s window, previous count weight is 50% - const halfwayCount = calculateSlidingWindow(90_000, 60_000, 10, 100); - expect(halfwayCount).toBe(60); // floor(100 * 0.5) + 10 = 60 - - // At 45s into a 60s window, previous count weight is 25% - const threeQuarterCount = calculateSlidingWindow(105_000, 60_000, 10, 100); - expect(threeQuarterCount).toBe(35); // floor(100 * 0.25) + 10 = 35 - - // At 59s into a 60s window, previous count weight is ~1.67% - const endCount = calculateSlidingWindow(119_000, 60_000, 10, 100); - expect(endCount).toBe(11); // floor(100 * (1/60)) + 10 = 1 + 10 = 11 - }); - - it("handles zero previous count correctly", () => { - const count = calculateSlidingWindow(15_000, 60_000, 8, 0); - expect(count).toBe(8); - }); - - it("handles invalid windowMs safely", () => { - expect(calculateSlidingWindow(15_000, 0, 8, 10)).toBe(0); - expect(calculateSlidingWindow(15_000, -100, 8, 10)).toBe(0); - }); - }); - - describe("Lua script behavior simulation", () => { - function simulateLua( - store: Map, - keys: string[], - argv: (string | number)[], - ): number { - const currentKey = keys[0]!; - const prevKey = keys[1]; - const now = Number(argv[0]); - const window = Number(argv[1]); - const maxReqs = Number(argv[2]); - const ttl = argv[3] ? Number(argv[3]) : Math.ceil((window * 2) / 1000) + 60; - - if (!now || !window || window <= 0 || !maxReqs || maxReqs <= 0) { - return -1; - } - - let cKey = currentKey; - let pKey = prevKey; - if (!pKey) { - const currentBucket = Math.floor(now / window); - const prevBucket = currentBucket - 1; - cKey = `${keys[0]}:${currentBucket}`; - pKey = `${keys[0]}:${prevBucket}`; - } - - const currentCount = store.get(cKey)?.value ?? 0; - const prevCount = store.get(pKey)?.value ?? 0; - - const timeIntoCurrent = now % window; - const weight = (window - timeIntoCurrent) / window; - const estimated = Math.floor(prevCount * weight + currentCount); - - if (estimated < maxReqs) { - const item = store.get(cKey) ?? { value: 0, ttl: 0 }; - item.value += 1; - if (item.value === 1) { - item.ttl = ttl; - } - store.set(cKey, item); - return estimated + 1; - } - - return -1; - } - - it("increments within limit and sets TTL on first entry", () => { - const store = new Map(); - const currentKey = "{rate-limit:api:req:proj1}:10"; - const prevKey = "{rate-limit:api:req:proj1}:9"; - - const res1 = simulateLua(store, [currentKey, prevKey], [600_000, 60_000, 5]); - expect(res1).toBe(1); - expect(store.get(currentKey)?.value).toBe(1); - expect(store.get(currentKey)?.ttl).toBe(180); - - const res2 = simulateLua(store, [currentKey, prevKey], [601_000, 60_000, 5]); - expect(res2).toBe(2); - expect(store.get(currentKey)?.value).toBe(2); - }); - - it("rejects when limit is reached", () => { - const store = new Map(); - const currentKey = "{rate-limit:api:req:proj1}:10"; - const prevKey = "{rate-limit:api:req:proj1}:9"; - - for (let i = 0; i < 3; i++) { - const res = simulateLua(store, [currentKey, prevKey], [600_000, 60_000, 3]); - expect(res).toBe(i + 1); - } - - const rejected = simulateLua(store, [currentKey, prevKey], [600_500, 60_000, 3]); - expect(rejected).toBe(-1); - expect(store.get(currentKey)?.value).toBe(3); - }); - - it("computes keys automatically if only base key is provided", () => { - const store = new Map(); - const baseKey = "rate-limit:api:req:proj1"; - - const res = simulateLua(store, [baseKey], [120_000, 60_000, 5]); - expect(res).toBe(1); - expect(store.has("rate-limit:api:req:proj1:2")).toBe(true); - }); }); }); diff --git a/tests/redis.test.ts b/tests/redis.test.ts index 2ec17d5..dfba3be 100644 --- a/tests/redis.test.ts +++ b/tests/redis.test.ts @@ -2,9 +2,7 @@ import { describe, it, expect, vi } from "vitest"; import { registerCustomCommands, RedisClient, - LUA_CHECK_API_RATE_LIMIT, LUA_LEASE_API_RATE_LIMIT, - LUA_THROTTLE_PROVIDER, LUA_RELEASE_LOCK, LUA_RENEW_LOCK, LUA_USER_THROTTLE, @@ -12,7 +10,6 @@ import { LUA_ACQUIRE_LEASE, LUA_MARK_PROCESSED, } from "@/redis/index.js"; -import { throttleProvider } from "@/services/delivery/throttle.js"; import { UserThrottle } from "@/rate-limiter/index.js"; import { executeSchedulerPoll } from "@/services/scheduler/main.js"; import { WorkflowWorker } from "@/services/workflow/main.js"; @@ -29,26 +26,10 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { registerCustomCommands(mockRedis); - expect(definedCommands.has("checkApiRateLimit")).toBe(true); - expect(definedCommands.get("checkApiRateLimit")?.numberOfKeys).toBe(2); - expect(definedCommands.get("checkApiRateLimit")?.lua).toBe(LUA_CHECK_API_RATE_LIMIT); - - expect(definedCommands.has("checkApiRateLimit1Key")).toBe(true); - expect(definedCommands.get("checkApiRateLimit1Key")?.numberOfKeys).toBe(1); - expect(definedCommands.get("checkApiRateLimit1Key")?.lua).toBe(LUA_CHECK_API_RATE_LIMIT); - expect(definedCommands.has("leaseApiRateLimit")).toBe(true); expect(definedCommands.get("leaseApiRateLimit")?.numberOfKeys).toBe(2); expect(definedCommands.get("leaseApiRateLimit")?.lua).toBe(LUA_LEASE_API_RATE_LIMIT); - expect(definedCommands.has("leaseApiRateLimit1Key")).toBe(true); - expect(definedCommands.get("leaseApiRateLimit1Key")?.numberOfKeys).toBe(1); - expect(definedCommands.get("leaseApiRateLimit1Key")?.lua).toBe(LUA_LEASE_API_RATE_LIMIT); - - expect(definedCommands.has("throttleProvider")).toBe(true); - expect(definedCommands.get("throttleProvider")?.numberOfKeys).toBe(1); - expect(definedCommands.get("throttleProvider")?.lua).toBe(LUA_THROTTLE_PROVIDER); - expect(definedCommands.has("releaseLock")).toBe(true); expect(definedCommands.get("releaseLock")?.numberOfKeys).toBe(1); expect(definedCommands.get("releaseLock")?.lua).toBe(LUA_RELEASE_LOCK); @@ -80,11 +61,7 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { redisOptions: { lazyConnect: true }, }); try { - expect(typeof client.native.checkApiRateLimit).toBe("function"); - expect(typeof client.native.checkApiRateLimit1Key).toBe("function"); expect(typeof client.native.leaseApiRateLimit).toBe("function"); - expect(typeof client.native.leaseApiRateLimit1Key).toBe("function"); - expect(typeof client.native.throttleProvider).toBe("function"); expect(typeof client.native.releaseLock).toBe("function"); expect(typeof client.native.renewLock).toBe("function"); expect(typeof client.native.throttleUser).toBe("function"); @@ -97,18 +74,10 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { }); it("validates Lua scripts contain required Redis commands", () => { - expect(LUA_CHECK_API_RATE_LIMIT).toContain('redis.call("INCR"'); - expect(LUA_CHECK_API_RATE_LIMIT).toContain('redis.call("EXPIRE"'); - expect(LUA_CHECK_API_RATE_LIMIT).toContain("math.floor(prevCount * weight + currentCount)"); - expect(LUA_LEASE_API_RATE_LIMIT).toContain('redis.call("INCRBY"'); expect(LUA_LEASE_API_RATE_LIMIT).toContain('redis.call("EXPIRE"'); expect(LUA_LEASE_API_RATE_LIMIT).toContain("math.floor(prevCount * weight + currentCount)"); - expect(LUA_THROTTLE_PROVIDER).toContain("redis.call('ZREMRANGEBYSCORE'"); - expect(LUA_THROTTLE_PROVIDER).toContain("redis.call('ZCARD'"); - expect(LUA_THROTTLE_PROVIDER).toContain("redis.call('ZADD'"); - expect(LUA_USER_THROTTLE).toContain('redis.call("ZREMRANGEBYSCORE"'); expect(LUA_USER_THROTTLE).toContain('redis.call("ZCARD"'); expect(LUA_USER_THROTTLE).toContain('redis.call("ZADD"'); @@ -130,64 +99,6 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { }); }); - describe("throttleProvider", () => { - it("uses pre-compiled command when available on Redis", async () => { - const mockRedis: any = { - throttleProvider: vi.fn().mockResolvedValue([1, 0]), - eval: vi.fn(), - }; - const mockLogger = { warn: vi.fn(), info: vi.fn() }; - - const result = await throttleProvider( - mockRedis, - "email", - { limit: 10, windowSeconds: 60 }, - mockLogger, - ); - - expect(mockRedis.throttleProvider).toHaveBeenCalled(); - expect(mockRedis.eval).not.toHaveBeenCalled(); - expect(result.allowed).toBe(true); - }); - - it("calculates retryAfterMs when pre-compiled command returns throttled", async () => { - const now = Date.now(); - const mockRedis: any = { - throttleProvider: vi.fn().mockResolvedValue([0, now - 500]), - eval: vi.fn(), - }; - const mockLogger = { warn: vi.fn(), info: vi.fn() }; - - const result = await throttleProvider( - mockRedis, - "sms", - { limit: 5, windowSeconds: 10 }, - mockLogger, - ); - - expect(result.allowed).toBe(false); - expect(result.retryAfterMs).toBeGreaterThanOrEqual(9000); - expect(mockLogger.warn).toHaveBeenCalled(); - }); - - it("falls back to redis.eval when pre-compiled command is absent", async () => { - const mockRedis: any = { - eval: vi.fn().mockResolvedValue([1, 0]), - }; - const mockLogger = { warn: vi.fn(), info: vi.fn() }; - - const result = await throttleProvider( - mockRedis, - "email", - { limit: 10, windowSeconds: 60 }, - mockLogger, - ); - - expect(mockRedis.eval).toHaveBeenCalled(); - expect(result.allowed).toBe(true); - }); - }); - describe("UserThrottle", () => { it("uses pre-compiled throttleUser when available", async () => { const mockRedis: any = { @@ -215,17 +126,6 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { expect(result.allowed).toBe(false); expect(result.count).toBe(6); }); - - it("falls back to redis.eval when throttleUser is absent", async () => { - const mockRedis: any = { - eval: vi.fn().mockResolvedValue(1), - }; - const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 5 }); - const result = await throttle.check("proj_1", "user_1"); - - expect(mockRedis.eval).toHaveBeenCalled(); - expect(result.allowed).toBe(true); - }); }); describe("executeSchedulerPoll (schedulerPoll & releaseLock)", () => { @@ -261,34 +161,6 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { ); expect(mockRedis.eval).not.toHaveBeenCalled(); }); - - it("falls back to pipeline.eval and redis.eval when pre-compiled methods are absent", async () => { - const pollPipeline = { - eval: vi.fn(), - exec: vi.fn().mockResolvedValue(Array.from({ length: 16 }, () => [null, []])), - }; - const cleanupPipeline = { zrem: vi.fn(), exec: vi.fn().mockResolvedValue([]) }; - - let pipelineCount = 0; - const mockRedis: any = { - set: vi.fn().mockResolvedValue("OK"), - pipeline: vi.fn(() => (pipelineCount++ === 0 ? pollPipeline : cleanupPipeline)), - eval: vi.fn().mockResolvedValue(1), - }; - const mockProducers = { - critical: { publishBatch: vi.fn() }, - normal: { publishBatch: vi.fn() }, - low: { publishBatch: vi.fn() }, - }; - const mockLogger = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }; - const mockDb: any = {}; - - const polled = await executeSchedulerPoll(mockRedis, mockProducers, mockLogger, mockDb); - - expect(polled).toBe(false); - expect(pollPipeline.eval).toHaveBeenCalledTimes(16); - expect(mockRedis.eval).toHaveBeenCalled(); - }); }); describe("WorkflowWorker (releaseLock & renewLock)", () => { @@ -358,72 +230,6 @@ describe("Redis Pre-compiled Commands (defineCommand)", () => { ); expect(mockRedis.eval).not.toHaveBeenCalled(); }); - - it("falls back to eval for releaseLock when releaseLock is absent", async () => { - const { workflowRegistry } = await import("@/workflows/index.js"); - workflowRegistry.register("flow2", async () => {}); - - const mockRedis: any = { - set: vi.fn().mockResolvedValue("OK"), - eval: vi.fn().mockResolvedValue(1), - zadd: vi.fn().mockResolvedValue(1), - }; - - const qb: any = {}; - qb.select = vi.fn().mockReturnValue(qb); - qb.from = vi.fn().mockReturnValue(qb); - qb.where = vi.fn().mockReturnValue(qb); - qb.limit = vi - .fn() - .mockResolvedValue([{ id: "inst-2", status: "pending", name: "flow2", input: {} }]); - qb.update = vi.fn().mockReturnValue(qb); - qb.set = vi.fn().mockReturnValue(qb); - qb.then = function (resolve: any) { - resolve([{ id: "inst-2", status: "pending", name: "flow2", input: {} }]); - }; - - const worker = new WorkflowWorker({ - consumer: { ack: vi.fn(), nack: vi.fn() } as any, - pendingScanner: {} as any, - logger: { - child: vi.fn().mockReturnThis(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - debug: vi.fn(), - } as any, - concurrency: 1, - redis: mockRedis, - db: qb, - workflowProducer: { - publishBatch: vi.fn().mockResolvedValue({ messageIds: [], eventIds: [] }), - }, - notificationProducer: { - publishBatch: vi.fn().mockResolvedValue({ messageIds: [], eventIds: [] }), - }, - }); - - const msg = { - id: "msg-2", - stream: "workflow" as any, - event: { - id: "evt-2", - type: "workflow.triggered", - name: "workflow.triggered", - payload: { instanceId: "inst-2", name: "flow2", projectId: "proj-1" }, - metadata: { timestamp: new Date().toISOString() }, - }, - }; - - await worker.process(msg as any); - - expect(mockRedis.eval).toHaveBeenCalledWith( - expect.stringContaining("redis.call('GET'"), - 1, - "lock:workflow:inst-2", - expect.any(String), - ); - }); }); }); diff --git a/tests/regressions.test.ts b/tests/regressions.test.ts index 03fff57..2d0b630 100644 --- a/tests/regressions.test.ts +++ b/tests/regressions.test.ts @@ -3,7 +3,7 @@ import { renderWithTemplate } from "@/templates/render.js"; import { UserThrottle, ProjectSettingsCache } from "@/rate-limiter/index.js"; import { PendingMessageScanner } from "@/queue/index.js"; import { isRetryableAiError, PermanentAiError } from "@/services/ai/main.js"; -import { LUA_SCHEDULER_POLL } from "@/shared/index.js"; +import { LUA_SCHEDULER_POLL } from "@/redis/index.js"; import { EngineWorker } from "@/services/engine/main.js"; import { SchedulerWorker } from "@/services/scheduler/main.js"; import { registry } from "@/index.js"; @@ -58,14 +58,14 @@ describe("renderWithTemplate (P1-5: template injection / JSON break)", () => { describe("UserThrottle (P1-8: tenant isolation)", () => { it("namespaces the throttle key by project", async () => { - const redis: any = { eval: vi.fn().mockResolvedValue(1) }; + const redis: any = { throttleUser: vi.fn().mockResolvedValue(1) }; const throttle = new UserThrottle({ redis, maxPerHour: 5 }); await throttle.check("proj-a", "user-1", "normal"); await throttle.check("proj-b", "user-1", "normal"); - const keyA = redis.eval.mock.calls[0][2]; - const keyB = redis.eval.mock.calls[1][2]; + const keyA = redis.throttleUser.mock.calls[0][0]; + const keyB = redis.throttleUser.mock.calls[1][0]; expect(keyA).toBe("throttle:proj-a:user:user-1"); expect(keyB).toBe("throttle:proj-b:user:user-1"); diff --git a/tests/scheduler-poll.test.ts b/tests/scheduler-poll.test.ts index 00d32bd..2bf27f5 100644 --- a/tests/scheduler-poll.test.ts +++ b/tests/scheduler-poll.test.ts @@ -40,7 +40,7 @@ interface Harness { function harness(options: { shards?: any[]; lock?: string | null } = {}): Harness { const pollPipeline = { - eval: vi.fn(), + schedulerPoll: vi.fn(), exec: vi.fn().mockResolvedValue(options.shards ?? shardReplies()), }; const cleanupPipeline = { zrem: vi.fn(), exec: vi.fn().mockResolvedValue([]) }; @@ -48,7 +48,7 @@ function harness(options: { shards?: any[]; lock?: string | null } = {}): Harnes let pipelineCall = 0; const redis = { set: vi.fn().mockResolvedValue(options.lock === undefined ? "OK" : options.lock), - eval: vi.fn().mockResolvedValue(1), + releaseLock: vi.fn().mockResolvedValue(1), pipeline: vi.fn(() => (pipelineCall++ === 0 ? pollPipeline : cleanupPipeline)), }; @@ -95,12 +95,9 @@ describe("executeSchedulerPoll", () => { const h = harness(); await run(h); - // A compare-and-delete: an expired lock now held by another poller must - // not be deleted by this one. - const [script, keyCount, key, owner] = h.redis.eval.mock.calls.at(-1)!; - expect(script).toContain('redis.call("get", KEYS[1])'); - expect(script).toContain('redis.call("del", KEYS[1])'); - expect(keyCount).toBe(1); + // releaseLock is a compare-and-delete: an expired lock now held by + // another poller must not be deleted by this one. + const [key, owner] = h.redis.releaseLock.mock.calls.at(-1)!; expect(key).toBe("notif:lock:scheduler:poll"); expect(owner).toBe(h.redis.set.mock.calls[0]![1]); }); @@ -109,7 +106,7 @@ describe("executeSchedulerPoll", () => { const h = harness({ lock: null }); await run(h); - expect(h.redis.eval).not.toHaveBeenCalled(); + expect(h.redis.releaseLock).not.toHaveBeenCalled(); }); it("releases the lock even when the poll throws", async () => { @@ -118,7 +115,7 @@ describe("executeSchedulerPoll", () => { await expect(run(h)).resolves.toBe(false); expect(h.logger.error).toHaveBeenCalled(); - expect(h.redis.eval).toHaveBeenCalledTimes(1); + expect(h.redis.releaseLock).toHaveBeenCalledTimes(1); }); }); @@ -127,11 +124,9 @@ describe("executeSchedulerPoll", () => { const h = harness(); await run(h); - expect(h.pollPipeline.eval).toHaveBeenCalledTimes(SHARDS); + expect(h.pollPipeline.schedulerPoll).toHaveBeenCalledTimes(SHARDS); for (let i = 0; i < SHARDS; i++) { - expect(h.pollPipeline.eval).toHaveBeenCalledWith( - expect.any(String), - 1, + expect(h.pollPipeline.schedulerPoll).toHaveBeenCalledWith( `notif:scheduled:zset:${i}`, expect.any(Number), 100, diff --git a/tests/templates.test.ts b/tests/templates.test.ts deleted file mode 100644 index 3629f98..0000000 --- a/tests/templates.test.ts +++ /dev/null @@ -1,147 +0,0 @@ -import { describe, it, expect } from "vitest"; -import { templateRegistry, renderTemplate, type TemplateContext } from "@/templates/index.js"; - -describe("TemplateRegistry", () => { - it("should register and use custom renderers", () => { - const customRenderer = (ctx: TemplateContext) => ({ - content: { subject: "Custom", body: ctx.eventType }, - }); - templateRegistry.register("test.event", customRenderer); - - expect(templateRegistry.has("test.event")).toBe(true); - expect(templateRegistry.registeredTypes()).toContain("test.event"); - - const ctx: TemplateContext = { - eventType: "test.event", - templateVariables: {}, - locale: "en", - timezone: "UTC", - deeplinkScheme: "app://", - }; - - const result = renderTemplate(ctx); - expect(result.content.subject).toBe("Custom"); - expect(result.content.body).toBe("test.event"); - }); - - it("should fallback to default renderer when no custom renderer exists", () => { - const ctx: TemplateContext = { - eventType: "unknown.event", - templateVariables: { - subject: "Hello", - message: "World", - }, - locale: "en", - timezone: "UTC", - deeplinkScheme: "app://", - }; - - const result = renderTemplate(ctx); - expect(result.content.subject).toBe("Hello"); - expect(result.content.body).toBe("World"); - }); - - it("should use generic default when no vars provided", () => { - const ctx: TemplateContext = { - eventType: "empty.event", - templateVariables: {}, - locale: "en", - timezone: "UTC", - deeplinkScheme: "app://", - }; - - const result = renderTemplate(ctx); - expect(result.content.subject).toBeUndefined(); - expect(result.content.body).toBe("Notification: empty.event"); - }); -}); - -import { renderWithTemplate } from "@/templates/render.js"; - -describe("renderWithTemplate Edge Cases", () => { - it("interpolates falsy values (false, 0) as string literals instead of blank strings", () => { - const template = { - content: { - body: "Count: {{count}}, Active: {{active}}, NullVal: {{nullVal}}, UndefVal: {{undefVal}}", - }, - }; - - const variables = { - count: 0, - active: false, - nullVal: null, - undefVal: undefined, - }; - - const rendered = renderWithTemplate(template, variables); - expect((rendered.content as any).body).toBe("Count: 0, Active: false, NullVal: , UndefVal: "); - }); - - it("applies HTML escaping to HTML fields while preserving text fields", () => { - const template = { - content: { - htmlBody: "Hello, {{name}} & welcome to {{company}}!", - textBody: "Hello, {{name}} & welcome to {{company}}!", - }, - }; - - const variables = { - name: "", - company: 'Acme "Best" & Co.', - }; - - const rendered = renderWithTemplate(template, variables); - expect((rendered.content as any).htmlBody).toBe( - "Hello, <script>alert('xss')</script> & welcome to Acme "Best" & Co.!", - ); - expect((rendered.content as any).textBody).toBe( - "Hello, & welcome to Acme \"Best\" & Co.!", - ); - }); - - it("strips CRLF from header fields to prevent email header injection", () => { - const template = { - content: { - subject: "Welcome {{user}} to Notifkit", - from: "notifications+{{tenant}}@notifkit.internal", - }, - }; - - const variables = { - user: "Alice\r\nBcc: victim@example.com", - tenant: "corp\nInjection: dangerous", - }; - - const rendered = renderWithTemplate(template, variables); - expect((rendered.content as any).subject).toBe( - "Welcome Alice Bcc: victim@example.com to Notifkit", - ); - expect((rendered.content as any).from).toBe( - "notifications+corp Injection: dangerous@notifkit.internal", - ); - }); - - it("interpolates objects and arrays as JSON stringified values", () => { - const template = { - content: { - body: "Payload: {{data}}, Items: {{items}}", - }, - }; - - const variables = { - data: { score: 100, role: "admin" }, - items: [1, 2, 3], - }; - - const rendered = renderWithTemplate(template, variables); - expect((rendered.content as any).body).toBe( - 'Payload: {"score":100,"role":"admin"}, Items: [1,2,3]', - ); - }); - - it("renders default fallback payload when dbTemplate is missing", () => { - const rendered = renderWithTemplate(null, { orderId: "ord_123" }); - expect(rendered.content.subject).toBe("Notification"); - expect(rendered.content.body).toContain('"orderId": "ord_123"'); - }); -}); diff --git a/tests/workers-bootstrap.test.ts b/tests/workers-bootstrap.test.ts new file mode 100644 index 0000000..c3fdb54 --- /dev/null +++ b/tests/workers-bootstrap.test.ts @@ -0,0 +1,96 @@ +import { describe, it, expect, vi } from "vitest"; + +vi.mock("@/queue/index.js", () => { + const record = (kind: string) => + vi.fn(function (this: any, options: any) { + this.kind = kind; + this.options = options; + }); + return { + StreamConsumer: record("consumer"), + PendingMessageScanner: record("scanner"), + StreamProducer: record("producer"), + }; +}); + +import { + createPriorityProducers, + createStreamConsumer, + shutdownWorker, + uniqueConsumerId, +} from "@/workers/bootstrap.js"; +import { CONSUMER_GROUPS, OUTBOUND_STREAMS, STREAMS } from "@/contracts/streams.js"; + +const logger: any = { info: vi.fn() }; + +describe("createPriorityProducers", () => { + it("maps each priority lane to its stage's stream", () => { + const producers = createPriorityProducers({} as any, logger, "ENRICHED") as any; + + expect(producers.critical.options.stream).toBe(STREAMS.ENRICHED_CRITICAL); + expect(producers.normal.options.stream).toBe(STREAMS.ENRICHED_NORMAL); + expect(producers.low.options.stream).toBe(STREAMS.ENRICHED_LOW); + }); +}); + +describe("createStreamConsumer", () => { + it("gives the consumer and its pending scanner the same stream, group and id", () => { + const redis: any = { native: {} }; + const { consumer, pendingScanner } = createStreamConsumer( + { redis, logger }, + { + stream: OUTBOUND_STREAMS, + group: CONSUMER_GROUPS.DELIVERY, + consumer: "delivery-1", + batchSize: 8, + bufferAcks: true, + }, + ) as any; + + for (const built of [consumer, pendingScanner]) { + expect(built.options).toMatchObject({ + redis: redis.native, + stream: [...OUTBOUND_STREAMS], + group: CONSUMER_GROUPS.DELIVERY, + consumer: "delivery-1", + }); + } + expect(consumer.options).toMatchObject({ + dlqStream: STREAMS.DEAD_LETTER, + batchSize: 8, + bufferAcks: true, + }); + }); +}); + +describe("uniqueConsumerId", () => { + it("differs between two consumers in the same process", () => { + expect(uniqueConsumerId("engine")).toMatch(/^engine-/); + expect(uniqueConsumerId("engine")).not.toBe(uniqueConsumerId("engine")); + }); +}); + +describe("shutdownWorker", () => { + it("stops timers and subscribers before draining, and closes connections last", async () => { + const order: string[] = []; + let fired = false; + const timer = setTimeout(() => (fired = true), 0); + + await shutdownWorker("test", { + logger, + timers: [timer, null], + subscriber: { disconnect: () => order.push("subscriber") }, + worker: { stop: async () => void order.push("worker") } as any, + sql: { end: async () => void order.push("sql") } as any, + redis: { disconnect: async () => void order.push("redis") } as any, + }); + await new Promise((r) => setTimeout(r, 5)); + + expect(fired).toBe(false); + expect(order).toEqual(["subscriber", "worker", "sql", "redis"]); + }); + + it("tolerates a service that never started", async () => { + await expect(shutdownWorker("test", {})).resolves.toBeUndefined(); + }); +}); diff --git a/tests/workers-workflow.test.ts b/tests/workers-workflow.test.ts index 0931579..69cb937 100644 --- a/tests/workers-workflow.test.ts +++ b/tests/workers-workflow.test.ts @@ -14,7 +14,8 @@ describe("WorkflowWorker Edge Cases", () => { beforeEach(() => { mockRedis = { set: vi.fn().mockResolvedValue("OK"), - eval: vi.fn().mockResolvedValue(null), + releaseLock: vi.fn().mockResolvedValue(1), + renewLock: vi.fn().mockResolvedValue(1), zadd: vi.fn().mockResolvedValue(1), }; @@ -599,7 +600,7 @@ describe("WorkflowWorker Edge Cases", () => { expect(mockDb.set).toHaveBeenCalledWith({ status: "pending" }); }); - it("always releases lock via LUA_RELEASE_LOCK in finally block", async () => { + it("always releases lock in finally block", async () => { workflowRegistry.register("test_lock_release", async (ctx) => { await ctx.step.notify({ template: "tpl-1", user: "u-1" }); }); @@ -623,10 +624,7 @@ describe("WorkflowWorker Edge Cases", () => { await worker.process(msg); - // Expect eval called with LUA_RELEASE_LOCK - expect(mockRedis.eval).toHaveBeenCalledWith( - expect.stringContaining("redis.call('GET'"), - 1, + expect(mockRedis.releaseLock).toHaveBeenCalledWith( "lock:workflow:inst-lock-rel", expect.any(String), ); diff --git a/tests/workers.test.ts b/tests/workers.test.ts index 5ecfc49..3d5ffa7 100644 --- a/tests/workers.test.ts +++ b/tests/workers.test.ts @@ -454,9 +454,7 @@ describe("UserThrottle (Priority-Aware Throttling)", () => { beforeEach(() => { mockRedis = { - incr: vi.fn(), - expire: vi.fn(), - eval: vi.fn(), + throttleUser: vi.fn(), }; }); @@ -465,80 +463,80 @@ describe("UserThrottle (Priority-Aware Throttling)", () => { const result = await throttle.check("proj-1", "usr-1", "critical"); expect(result.allowed).toBe(true); - expect(mockRedis.eval).not.toHaveBeenCalled(); + expect(mockRedis.throttleUser).not.toHaveBeenCalled(); }); it("throttles normal notifications when limit is exceeded", async () => { const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 3 }); - mockRedis.eval.mockResolvedValueOnce(4); + mockRedis.throttleUser.mockResolvedValueOnce(4); const result = await throttle.check("proj-1", "usr-1", "normal"); expect(result.allowed).toBe(false); - expect(mockRedis.eval).toHaveBeenCalled(); + expect(mockRedis.throttleUser).toHaveBeenCalled(); }); - // eval args: [script, numkeys, key, windowStart, limit, targetTime, memberId, ttl] - const argWindowStart = (call: any[]) => call[3]; - const argLimit = (call: any[]) => call[4]; - const argTargetTime = (call: any[]) => call[5]; + // throttleUser args: [key, windowStart, limit, targetTime, memberId, ttl] + const argWindowStart = (call: any[]) => call[1]; + const argLimit = (call: any[]) => call[2]; + const argTargetTime = (call: any[]) => call[3]; /** Window length the script was actually asked to enforce. */ const windowMsOf = (call: any[]) => argTargetTime(call) - argWindowStart(call); it("counts a message under its own id, so a retry is not counted twice", async () => { const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 3 }); - mockRedis.eval.mockResolvedValue(1); + mockRedis.throttleUser.mockResolvedValue(1); await throttle.check("proj-1", "usr-1", "normal", { messageId: "evt-1:usr-1:email" }); await throttle.check("proj-1", "usr-1", "normal", { messageId: "evt-1:usr-1:email" }); - const members = mockRedis.eval.mock.calls.map((c: any[]) => c[6]); + const members = mockRedis.throttleUser.mock.calls.map((c: any[]) => c[4]); expect(members).toEqual(["evt-1:usr-1:email", "evt-1:usr-1:email"]); }); it("applies a per-project limit override in place of the global default", async () => { const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 3 }); - mockRedis.eval.mockResolvedValueOnce(40); + mockRedis.throttleUser.mockResolvedValueOnce(40); const result = await throttle.check("proj-1", "usr-1", "normal", { limit: 50 }); expect(result.allowed).toBe(true); expect(result.limit).toBe(50); - expect(argLimit(mockRedis.eval.mock.calls[0])).toBe(50); + expect(argLimit(mockRedis.throttleUser.mock.calls[0])).toBe(50); }); it("applies a per-project window override", async () => { const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 3 }); - mockRedis.eval.mockResolvedValueOnce(1); + mockRedis.throttleUser.mockResolvedValueOnce(1); await throttle.check("proj-1", "usr-1", "normal", { windowHours: 24 }); // The window must span a full 24h, not the default hour. - expect(windowMsOf(mockRedis.eval.mock.calls[0])).toBe(24 * 3600_000); + expect(windowMsOf(mockRedis.throttleUser.mock.calls[0])).toBe(24 * 3600_000); }); it("falls back to the default when overrides are null or nonsensical", async () => { const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 7, windowHours: 1 }); - mockRedis.eval.mockResolvedValue(1); + mockRedis.throttleUser.mockResolvedValue(1); await throttle.check("proj-1", "usr-1", "normal", { limit: null, windowHours: null }); - expect(argLimit(mockRedis.eval.mock.calls[0])).toBe(7); + expect(argLimit(mockRedis.throttleUser.mock.calls[0])).toBe(7); // A stored zero or negative window would make the window meaningless. await throttle.check("proj-1", "usr-1", "normal", { windowHours: 0 }); - expect(windowMsOf(mockRedis.eval.mock.calls[1])).toBe(3600_000); + expect(windowMsOf(mockRedis.throttleUser.mock.calls[1])).toBe(3600_000); await throttle.check("proj-1", "usr-1", "normal", { windowHours: -5 }); - expect(windowMsOf(mockRedis.eval.mock.calls[2])).toBe(3600_000); + expect(windowMsOf(mockRedis.throttleUser.mock.calls[2])).toBe(3600_000); }); it("treats a limit of 0 as a deliberate kill switch, not a missing value", async () => { const throttle = new UserThrottle({ redis: mockRedis, maxPerHour: 100 }); - mockRedis.eval.mockResolvedValueOnce(1); + mockRedis.throttleUser.mockResolvedValueOnce(1); const result = await throttle.check("proj-1", "usr-1", "normal", { limit: 0 }); - expect(argLimit(mockRedis.eval.mock.calls[0])).toBe(0); + expect(argLimit(mockRedis.throttleUser.mock.calls[0])).toBe(0); expect(result.allowed).toBe(false); }); @@ -548,7 +546,7 @@ describe("UserThrottle (Priority-Aware Throttling)", () => { const result = await throttle.check("proj-1", "usr-1", "critical", { limit: 0 }); expect(result.allowed).toBe(true); - expect(mockRedis.eval).not.toHaveBeenCalled(); + expect(mockRedis.throttleUser).not.toHaveBeenCalled(); }); }); @@ -1053,15 +1051,6 @@ describe("DeliveryWorker", () => { }; mockConsumer = { ack: vi.fn(), nack: vi.fn(), redis: { incr: vi.fn(), expire: vi.fn() } }; - // Set rate limits to undefined to skip throttling logic for these tests - vi.mock("../src/rate-limiter/index.js", async (importOriginal) => { - const actual = await importOriginal(); - return { - ...actual, - throttleProvider: vi.fn(), - }; - }); - worker = new DeliveryWorker({ consumer: mockConsumer as any, pendingScanner: {} as any, @@ -1503,7 +1492,7 @@ describe("DeliveryWorker", () => { const send = vi.fn(); mockTransportRegistry.getAll.mockReturnValue([limitedTransport(send)]); // The lease script grants no tokens: the window's budget is spent. - mockRedis.eval = vi.fn().mockResolvedValue(0); + mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(0); await worker.process(dispatched() as any); @@ -1517,7 +1506,7 @@ describe("DeliveryWorker", () => { it("counts each throttled attempt so the retries cannot loop forever", async () => { mockTransportRegistry.getAll.mockReturnValue([limitedTransport(vi.fn())]); - mockRedis.eval = vi.fn().mockResolvedValue(0); + mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(0); await worker.process(dispatched({ throttleAttemptCount: 2 }) as any); @@ -1527,7 +1516,7 @@ describe("DeliveryWorker", () => { it("gives up once the throttled retries exceed maxAttempts", async () => { const send = vi.fn(); mockTransportRegistry.getAll.mockReturnValue([limitedTransport(send)]); - mockRedis.eval = vi.fn().mockResolvedValue(0); + mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(0); await worker.process( dispatched({ throttleAttemptCount: 3, deliveryOptions: { maxAttempts: 3 } }) as any, @@ -1546,7 +1535,7 @@ describe("DeliveryWorker", () => { it("falls back to another channel instead of dropping an exhausted task", async () => { mockTransportRegistry.getAll.mockReturnValue([limitedTransport(vi.fn())]); - mockRedis.eval = vi.fn().mockResolvedValue(0); + mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(0); await worker.process( dispatched({ @@ -1564,7 +1553,7 @@ describe("DeliveryWorker", () => { it("sends normally when the provider limit has room", async () => { const send = vi.fn().mockResolvedValue({ success: true, providerMessageId: "p-1" }); mockTransportRegistry.getAll.mockReturnValue([limitedTransport(send)]); - mockRedis.eval = vi.fn().mockResolvedValue(1); + mockRedis.leaseApiRateLimit = vi.fn().mockResolvedValue(1); await worker.process(dispatched() as any); @@ -1926,80 +1915,6 @@ describe("NotifkitClient API Key Support", () => { }); }); -import { throttleProvider } from "@/services/delivery/throttle.js"; - -describe("throttleProvider (Provider-Level Rate Limiting)", () => { - let mockRedis: any; - let mockLogger: any; - - beforeEach(() => { - mockRedis = { - eval: vi.fn(), - }; - mockLogger = { - warn: vi.fn(), - info: vi.fn(), - }; - }); - - it("allows requests and records them if under the limit", async () => { - mockRedis.eval.mockResolvedValueOnce([1, 0]); - - const result = await throttleProvider( - mockRedis, - "email", - { limit: 2, windowSeconds: 10 }, - mockLogger, - ); - - expect(mockRedis.eval).toHaveBeenCalled(); - expect(result.allowed).toBe(true); - expect(mockLogger.warn).not.toHaveBeenCalled(); - }); - - it("returns retryAfterMs and logs warning if over the limit", async () => { - const oldestTimestamp = Date.now() - 200; - mockRedis.eval.mockResolvedValueOnce([0, oldestTimestamp]); - - const result = await throttleProvider( - mockRedis, - "email", - { limit: 2, windowSeconds: 0.5 }, - mockLogger, - ); - - expect(mockRedis.eval).toHaveBeenCalled(); - expect(result.allowed).toBe(false); - expect(result.retryAfterMs).toBeGreaterThanOrEqual(100); - expect(mockLogger.warn).toHaveBeenCalled(); - }); - - it("caps retryAfterMs at 0 when oldestTimestamp exceeds window boundary", async () => { - // Oldest timestamp is way in the past (e.g. 5 seconds ago on a 1s window) - const oldestTimestamp = Date.now() - 5000; - mockRedis.eval.mockResolvedValueOnce([0, oldestTimestamp]); - - const result = await throttleProvider( - mockRedis, - "sms", - { limit: 5, windowSeconds: 1 }, - mockLogger, - ); - - expect(result.allowed).toBe(false); - expect(result.retryAfterMs).toBe(0); - }); - - it("uses separate rate limit keys for different channels", async () => { - mockRedis.eval.mockResolvedValue([1, 0]); - - await throttleProvider(mockRedis, "push", { limit: 10, windowSeconds: 60 }, mockLogger); - - const callArgs = mockRedis.eval.mock.calls[0]; - expect(callArgs[2]).toBe("rate-limit:provider:push"); - }); -}); - import { ProviderThrottle } from "@/services/delivery/throttle.js"; describe("ProviderThrottle (leased provider rate limiting)", () => { @@ -2761,7 +2676,6 @@ describe("SchedulerWorker", () => { }; mockRedis = { zadd: vi.fn(), - eval: vi.fn(), get: vi.fn(), del: vi.fn(), set: vi.fn().mockResolvedValue("OK"), @@ -2808,12 +2722,12 @@ describe("SchedulerWorker", () => { get: vi.fn(), del: vi.fn(), zrem: vi.fn(), - eval: vi.fn(), + schedulerPoll: vi.fn(), exec: vi.fn(), }; mockRedis.pipeline = vi.fn().mockReturnValue(mockPipeline); mockRedis.set.mockResolvedValue("OK"); - mockRedis.eval = vi.fn().mockResolvedValue(1); + mockRedis.releaseLock = vi.fn().mockResolvedValue(1); mockPipeline.exec .mockResolvedValueOnce([ @@ -2832,9 +2746,6 @@ describe("SchedulerWorker", () => { ]) .mockResolvedValueOnce([[null, JSON.stringify({ priority: "normal", someField: "x" })]]); - // Instead of mockRedis.eval - // mockRedis.eval.mockResolvedValue([ ... ]); - const mockOutboundProducers = { normal: { publishBatch: vi.fn() }, }; @@ -2854,9 +2765,7 @@ describe("SchedulerWorker", () => { mockDb as any, ); - expect(mockPipeline.eval).toHaveBeenCalledWith( - expect.any(String), - 1, + expect(mockPipeline.schedulerPoll).toHaveBeenCalledWith( expect.stringMatching(/^notif:scheduled:zset:\d+$/), expect.any(Number), 100, @@ -2877,7 +2786,6 @@ describe("Integration: Engine -> Scheduler Handoff with real registry", () => { const mockRedis: any = { set: vi.fn(), zadd: vi.fn(), - eval: vi.fn(), get: vi.fn(), del: vi.fn(), pipeline: vi.fn().mockReturnValue({ @@ -3288,7 +3196,7 @@ describe("DeliveryWorker", () => { unmark: vi.fn().mockResolvedValue(undefined), }; mockRedis = { - eval: vi.fn().mockResolvedValue([1, 1000]), // Default to allowed + leaseApiRateLimit: vi.fn().mockResolvedValue(1), // Default to allowed }; mockScheduledProducer = { publish: vi.fn() }; mockEnrichedProducers = { normal: { publish: vi.fn() } }; @@ -3340,32 +3248,6 @@ describe("DeliveryWorker", () => { if (worker) await worker.stop(); }); - it("handles provider throttling and schedules retry", async () => { - // Mock redis.eval to return rate limit exceeded - mockRedis.eval.mockResolvedValue([0, 5000]); // 0 = denied, 5000 = retry after - - const msg = { - id: "msg-del-1", - event: { - payload: { - taskId: "task-1", - projectId: "proj-1", - enrichedEventId: "evt-1", - recipientId: "usr-1", - channel: "email", - destination: "test@example.com", - priority: "normal", - }, - metadata: { traceId: "trace-1" }, - }, - }; - - await worker.process(msg as any); - - // Bypassing throttling test because getProviderRateLimits is empty config. - // Just expect it to drop/fail or process normally without throttling - }); - it("falls back to next channel on transport failure", async () => { const mockTransport = { send: vi.fn().mockResolvedValue({ success: false, error: "Mock failure" }), diff --git a/tests/workflows.test.ts b/tests/workflows.test.ts index 80acb72..cdb85cb 100644 --- a/tests/workflows.test.ts +++ b/tests/workflows.test.ts @@ -24,7 +24,8 @@ const { mockDb, mockRedisNative, mockNotificationProducer, mockWorkflowProducer // The ioredis client the worker actually receives. mockRedisNative: { zadd: vi.fn(), - eval: vi.fn(), + releaseLock: vi.fn().mockResolvedValue(1), + renewLock: vi.fn().mockResolvedValue(1), set: vi.fn().mockResolvedValue("OK"), del: vi.fn(), duplicate: vi.fn(),