External/notifkit
1
0
Fork 0
mirror of https://github.com/devkitshq/notifkit.git synced 2026-10-19 18:24:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: extract worker bootstrap runtime and remove unused code

Browse Source
This commit is contained in:
coo1estguy
2026-09-29 13:10:15 +05:30
parent 880fe32b03
commit 35a34df2fe
38 changed files with 564 additions and 1993 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dashboard/package-lock.json
Generated
-8
View File
@@ -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",
dashboard/package.json
-8
View File
@@ -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",
dashboard/src/components/ui/dialog.tsx
-144
View File
@@ -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<typeof DialogPrimitive.Root>) {
return <DialogPrimitive.Root data-slot="dialog" {...props} />;
}
function DialogTrigger({ ...props }: React.ComponentProps<typeof DialogPrimitive.Trigger>) {
return <DialogPrimitive.Trigger data-slot="dialog-trigger" {...props} />;
}
function DialogPortal({ ...props }: React.ComponentProps<typeof DialogPrimitive.Portal>) {
return <DialogPrimitive.Portal data-slot="dialog-portal" {...props} />;
}
function DialogClose({ ...props }: React.ComponentProps<typeof DialogPrimitive.Close>) {
return <DialogPrimitive.Close data-slot="dialog-close" {...props} />;
}
function DialogOverlay({
className,
...props
}: React.ComponentProps<typeof DialogPrimitive.Overlay>) {
return (
<DialogPrimitive.Overlay
data-slot="dialog-overlay"
className={cn(
"fixed inset-0 z-50 bg-black/50 data-[state=closed]:animate-out data-[state=closed]:fade-out-0 data-[state=open]:animate-in data-[state=open]:fade-in-0",
className,
)}
{...props}
/>
);
}
function DialogContent({
className,
children,
showCloseButton = true,
...props
}: React.ComponentProps<typeof DialogPrimitive.Content> & {
showCloseButton?: boolean;
}) {
return (
<DialogPortal data-slot="dialog-portal">
<DialogOverlay />
<DialogPrimitive.Content
data-slot="dialog-content"
className={cn(
"fixed top-[50%] left-[50%] z-50 grid w-full max-w-[calc(100%-2rem)] translate-x-[-50%] translate-y-[-50%] gap-4 rounded-lg border bg-background p-6 shadow-lg duration-200 outline-none data-[state=closed]:animate-out data-[state=closed]:fade-out-0 data-[state=closed]:zoom-out-95 data-[state=open]:animate-in data-[state=open]:fade-in-0 data-[state=open]:zoom-in-95 sm:max-w-lg",
className,
)}
{...props}
>
{children}
{showCloseButton && (
<DialogPrimitive.Close
data-slot="dialog-close"
className="absolute top-4 right-4 rounded-xs opacity-70 ring-offset-background transition-opacity hover:opacity-100 focus:ring-2 focus:ring-ring focus:ring-offset-2 focus:outline-hidden disabled:pointer-events-none data-[state=open]:bg-accent data-[state=open]:text-muted-foreground [&_svg]:pointer-events-none [&_svg]:shrink-0 [&_svg:not([class*='size-'])]:size-4"
>
<XIcon />
<span className="sr-only">Close</span>
</DialogPrimitive.Close>
)}
</DialogPrimitive.Content>
</DialogPortal>
);
}
function DialogHeader({ className, ...props }: React.ComponentProps<"div">) {
return (
<div
data-slot="dialog-header"
className={cn("flex flex-col gap-2 text-center sm:text-left", className)}
{...props}
/>
);
}
function DialogFooter({
className,
showCloseButton = false,
children,
...props
}: React.ComponentProps<"div"> & {
showCloseButton?: boolean;
}) {
return (
<div
data-slot="dialog-footer"
className={cn("flex flex-col-reverse gap-2 sm:flex-row sm:justify-end", className)}
{...props}
>
{children}
{showCloseButton && (
<DialogPrimitive.Close asChild>
<Button variant="outline">Close</Button>
</DialogPrimitive.Close>
)}
</div>
);
}
function DialogTitle({ className, ...props }: React.ComponentProps<typeof DialogPrimitive.Title>) {
return (
<DialogPrimitive.Title
data-slot="dialog-title"
className={cn("text-lg leading-none font-semibold", className)}
{...props}
/>
);
}
function DialogDescription({
className,
...props
}: React.ComponentProps<typeof DialogPrimitive.Description>) {
return (
<DialogPrimitive.Description
data-slot="dialog-description"
className={cn("text-sm text-muted-foreground", className)}
{...props}
/>
);
}
export {
Dialog,
DialogClose,
DialogContent,
DialogDescription,
DialogFooter,
DialogHeader,
DialogOverlay,
DialogPortal,
DialogTitle,
DialogTrigger,
};
dashboard/src/components/ui/dropdown-menu.tsx
-225
View File
@@ -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<typeof DropdownMenuPrimitive.Root>) {
return <DropdownMenuPrimitive.Root data-slot="dropdown-menu" {...props} />;
}
function DropdownMenuPortal({
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.Portal>) {
return <DropdownMenuPrimitive.Portal data-slot="dropdown-menu-portal" {...props} />;
}
function DropdownMenuTrigger({
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.Trigger>) {
return <DropdownMenuPrimitive.Trigger data-slot="dropdown-menu-trigger" {...props} />;
}
function DropdownMenuContent({
className,
sideOffset = 4,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.Content>) {
return (
<DropdownMenuPrimitive.Portal>
<DropdownMenuPrimitive.Content
data-slot="dropdown-menu-content"
sideOffset={sideOffset}
className={cn(
"z-50 max-h-(--radix-dropdown-menu-content-available-height) min-w-[8rem] origin-(--radix-dropdown-menu-content-transform-origin) overflow-x-hidden overflow-y-auto rounded-md border bg-popover p-1 text-popover-foreground shadow-md data-[side=bottom]:slide-in-from-top-2 data-[side=left]:slide-in-from-right-2 data-[side=right]:slide-in-from-left-2 data-[side=top]:slide-in-from-bottom-2 data-[state=closed]:animate-out data-[state=closed]:fade-out-0 data-[state=closed]:zoom-out-95 data-[state=open]:animate-in data-[state=open]:fade-in-0 data-[state=open]:zoom-in-95",
className,
)}
{...props}
/>
</DropdownMenuPrimitive.Portal>
);
}
function DropdownMenuGroup({ ...props }: React.ComponentProps<typeof DropdownMenuPrimitive.Group>) {
return <DropdownMenuPrimitive.Group data-slot="dropdown-menu-group" {...props} />;
}
function DropdownMenuItem({
className,
inset,
variant = "default",
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.Item> & {
inset?: boolean;
variant?: "default" | "destructive";
}) {
return (
<DropdownMenuPrimitive.Item
data-slot="dropdown-menu-item"
data-inset={inset}
data-variant={variant}
className={cn(
"relative flex cursor-default items-center gap-2 rounded-sm px-2 py-1.5 text-sm outline-hidden select-none focus:bg-accent focus:text-accent-foreground data-[disabled]:pointer-events-none data-[disabled]:opacity-50 data-[inset]:pl-8 data-[variant=destructive]:text-destructive data-[variant=destructive]:focus:bg-destructive/10 data-[variant=destructive]:focus:text-destructive dark:data-[variant=destructive]:focus:bg-destructive/20 [&_svg]:pointer-events-none [&_svg]:shrink-0 [&_svg:not([class*='size-'])]:size-4 [&_svg:not([class*='text-'])]:text-muted-foreground data-[variant=destructive]:*:[svg]:text-destructive!",
className,
)}
{...props}
/>
);
}
function DropdownMenuCheckboxItem({
className,
children,
checked,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.CheckboxItem>) {
return (
<DropdownMenuPrimitive.CheckboxItem
data-slot="dropdown-menu-checkbox-item"
className={cn(
"relative flex cursor-default items-center gap-2 rounded-sm py-1.5 pr-2 pl-8 text-sm outline-hidden select-none focus:bg-accent focus:text-accent-foreground data-[disabled]:pointer-events-none data-[disabled]:opacity-50 [&_svg]:pointer-events-none [&_svg]:shrink-0 [&_svg:not([class*='size-'])]:size-4",
className,
)}
checked={checked}
{...props}
>
<span className="pointer-events-none absolute left-2 flex size-3.5 items-center justify-center">
<DropdownMenuPrimitive.ItemIndicator>
<CheckIcon className="size-4" />
</DropdownMenuPrimitive.ItemIndicator>
</span>
{children}
</DropdownMenuPrimitive.CheckboxItem>
);
}
function DropdownMenuRadioGroup({
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.RadioGroup>) {
return <DropdownMenuPrimitive.RadioGroup data-slot="dropdown-menu-radio-group" {...props} />;
}
function DropdownMenuRadioItem({
className,
children,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.RadioItem>) {
return (
<DropdownMenuPrimitive.RadioItem
data-slot="dropdown-menu-radio-item"
className={cn(
"relative flex cursor-default items-center gap-2 rounded-sm py-1.5 pr-2 pl-8 text-sm outline-hidden select-none focus:bg-accent focus:text-accent-foreground data-[disabled]:pointer-events-none data-[disabled]:opacity-50 [&_svg]:pointer-events-none [&_svg]:shrink-0 [&_svg:not([class*='size-'])]:size-4",
className,
)}
{...props}
>
<span className="pointer-events-none absolute left-2 flex size-3.5 items-center justify-center">
<DropdownMenuPrimitive.ItemIndicator>
<CircleIcon className="size-2 fill-current" />
</DropdownMenuPrimitive.ItemIndicator>
</span>
{children}
</DropdownMenuPrimitive.RadioItem>
);
}
function DropdownMenuLabel({
className,
inset,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.Label> & {
inset?: boolean;
}) {
return (
<DropdownMenuPrimitive.Label
data-slot="dropdown-menu-label"
data-inset={inset}
className={cn("px-2 py-1.5 text-sm font-medium data-[inset]:pl-8", className)}
{...props}
/>
);
}
function DropdownMenuSeparator({
className,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.Separator>) {
return (
<DropdownMenuPrimitive.Separator
data-slot="dropdown-menu-separator"
className={cn("-mx-1 my-1 h-px bg-border", className)}
{...props}
/>
);
}
function DropdownMenuShortcut({ className, ...props }: React.ComponentProps<"span">) {
return (
<span
data-slot="dropdown-menu-shortcut"
className={cn("ml-auto text-xs tracking-widest text-muted-foreground", className)}
{...props}
/>
);
}
function DropdownMenuSub({ ...props }: React.ComponentProps<typeof DropdownMenuPrimitive.Sub>) {
return <DropdownMenuPrimitive.Sub data-slot="dropdown-menu-sub" {...props} />;
}
function DropdownMenuSubTrigger({
className,
inset,
children,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.SubTrigger> & {
inset?: boolean;
}) {
return (
<DropdownMenuPrimitive.SubTrigger
data-slot="dropdown-menu-sub-trigger"
data-inset={inset}
className={cn(
"flex cursor-default items-center gap-2 rounded-sm px-2 py-1.5 text-sm outline-hidden select-none focus:bg-accent focus:text-accent-foreground data-[inset]:pl-8 data-[state=open]:bg-accent data-[state=open]:text-accent-foreground [&_svg]:pointer-events-none [&_svg]:shrink-0 [&_svg:not([class*='size-'])]:size-4 [&_svg:not([class*='text-'])]:text-muted-foreground",
className,
)}
{...props}
>
{children}
<ChevronRightIcon className="ml-auto size-4" />
</DropdownMenuPrimitive.SubTrigger>
);
}
function DropdownMenuSubContent({
className,
...props
}: React.ComponentProps<typeof DropdownMenuPrimitive.SubContent>) {
return (
<DropdownMenuPrimitive.SubContent
data-slot="dropdown-menu-sub-content"
className={cn(
"z-50 min-w-[8rem] origin-(--radix-dropdown-menu-content-transform-origin) overflow-hidden rounded-md border bg-popover p-1 text-popover-foreground shadow-lg data-[side=bottom]:slide-in-from-top-2 data-[side=left]:slide-in-from-right-2 data-[side=right]:slide-in-from-left-2 data-[side=top]:slide-in-from-bottom-2 data-[state=closed]:animate-out data-[state=closed]:fade-out-0 data-[state=closed]:zoom-out-95 data-[state=open]:animate-in data-[state=open]:fade-in-0 data-[state=open]:zoom-in-95",
className,
)}
{...props}
/>
);
}
export {
DropdownMenu,
DropdownMenuPortal,
DropdownMenuTrigger,
DropdownMenuContent,
DropdownMenuGroup,
DropdownMenuLabel,
DropdownMenuItem,
DropdownMenuCheckboxItem,
DropdownMenuRadioGroup,
DropdownMenuRadioItem,
DropdownMenuSeparator,
DropdownMenuShortcut,
DropdownMenuSub,
DropdownMenuSubTrigger,
DropdownMenuSubContent,
};
dashboard/src/components/ui/tabs.tsx
-80
View File
@@ -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<typeof TabsPrimitive.Root>) {
return (
<TabsPrimitive.Root
data-slot="tabs"
data-orientation={orientation}
orientation={orientation}
className={cn("group/tabs flex gap-2 data-[orientation=horizontal]:flex-col", className)}
{...props}
/>
);
}
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<typeof TabsPrimitive.List> & VariantProps<typeof tabsListVariants>) {
return (
<TabsPrimitive.List
data-slot="tabs-list"
data-variant={variant}
className={cn(tabsListVariants({ variant }), className)}
{...props}
/>
);
}
function TabsTrigger({ className, ...props }: React.ComponentProps<typeof TabsPrimitive.Trigger>) {
return (
<TabsPrimitive.Trigger
data-slot="tabs-trigger"
className={cn(
"relative inline-flex h-[calc(100%-1px)] flex-1 items-center justify-center gap-1.5 rounded-md border border-transparent px-2 py-1 text-sm font-medium whitespace-nowrap text-foreground/60 transition-all group-data-[orientation=vertical]/tabs:w-full group-data-[orientation=vertical]/tabs:justify-start hover:text-foreground focus-visible:border-ring focus-visible:ring-[3px] focus-visible:ring-ring/50 focus-visible:outline-1 focus-visible:outline-ring disabled:pointer-events-none disabled:opacity-50 group-data-[variant=default]/tabs-list:data-[state=active]:shadow-sm group-data-[variant=line]/tabs-list:data-[state=active]:shadow-none dark:text-muted-foreground dark:hover:text-foreground [&_svg]:pointer-events-none [&_svg]:shrink-0 [&_svg:not([class*='size-'])]:size-4",
"group-data-[variant=line]/tabs-list:bg-transparent group-data-[variant=line]/tabs-list:data-[state=active]:bg-transparent dark:group-data-[variant=line]/tabs-list:data-[state=active]:border-transparent dark:group-data-[variant=line]/tabs-list:data-[state=active]:bg-transparent",
"data-[state=active]:bg-background data-[state=active]:text-foreground dark:data-[state=active]:border-input dark:data-[state=active]:bg-input/30 dark:data-[state=active]:text-foreground",
"after:absolute after:bg-foreground after:opacity-0 after:transition-opacity group-data-[orientation=horizontal]/tabs:after:inset-x-0 group-data-[orientation=horizontal]/tabs:after:bottom-[-5px] group-data-[orientation=horizontal]/tabs:after:h-0.5 group-data-[orientation=vertical]/tabs:after:inset-y-0 group-data-[orientation=vertical]/tabs:after:-right-1 group-data-[orientation=vertical]/tabs:after:w-0.5 group-data-[variant=line]/tabs-list:data-[state=active]:after:opacity-100",
className,
)}
{...props}
/>
);
}
function TabsContent({ className, ...props }: React.ComponentProps<typeof TabsPrimitive.Content>) {
return (
<TabsPrimitive.Content
data-slot="tabs-content"
className={cn("flex-1 outline-none", className)}
{...props}
/>
);
}
export { Tabs, TabsList, TabsTrigger, TabsContent, tabsListVariants };
dashboard/src/components/ui/tooltip.tsx
-52
View File
@@ -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<typeof TooltipPrimitive.Provider>) {
return (
<TooltipPrimitive.Provider
data-slot="tooltip-provider"
delayDuration={delayDuration}
{...props}
/>
);
}
function Tooltip({ ...props }: React.ComponentProps<typeof TooltipPrimitive.Root>) {
return <TooltipPrimitive.Root data-slot="tooltip" {...props} />;
}
function TooltipTrigger({ ...props }: React.ComponentProps<typeof TooltipPrimitive.Trigger>) {
return <TooltipPrimitive.Trigger data-slot="tooltip-trigger" {...props} />;
}
function TooltipContent({
className,
sideOffset = 0,
children,
...props
}: React.ComponentProps<typeof TooltipPrimitive.Content>) {
return (
<TooltipPrimitive.Portal>
<TooltipPrimitive.Content
data-slot="tooltip-content"
sideOffset={sideOffset}
className={cn(
"z-50 w-fit origin-(--radix-tooltip-content-transform-origin) animate-in rounded-md bg-foreground px-3 py-1.5 text-xs text-balance text-background fade-in-0 zoom-in-95 data-[side=bottom]:slide-in-from-top-2 data-[side=left]:slide-in-from-right-2 data-[side=right]:slide-in-from-left-2 data-[side=top]:slide-in-from-bottom-2 data-[state=closed]:animate-out data-[state=closed]:fade-out-0 data-[state=closed]:zoom-out-95",
className,
)}
{...props}
>
{children}
<TooltipPrimitive.Arrow className="z-50 size-2.5 translate-y-[calc(-50%_-_2px)] rotate-45 rounded-[2px] bg-foreground fill-foreground" />
</TooltipPrimitive.Content>
</TooltipPrimitive.Portal>
);
}
export { Tooltip, TooltipTrigger, TooltipContent, TooltipProvider };
package-lock.json
Generated
-11
View File
@@ -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",
package.json
-1
View File
@@ -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",
scripts/build.sh
-4
View File
@@ -1,4 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
npm run build
scripts/setup.sh
-18
View File
@@ -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
src/db/schema.ts
-48
View File
@@ -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);
src/logger/index.ts
-12
View File
@@ -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<string, unknown>): Logger {
return logger.child(context);
}
export function childLogger(logger: Logger, bindings: Record<string, unknown>): Logger {
return logger.child(bindings);
}
src/rate-limiter/index.ts
+16 -38
View File
@@ -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<number> | null = null;
const executeLease = async (): Promise<number> => {
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) {
src/redis/index.ts
-126
View File
@@ -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<number>;
checkApiRateLimit1Key(
baseKey: string,
now: number | string,
window: number | string,
maxReqs: number | string,
ttl?: number | string,
): Promise<number>;
leaseApiRateLimit(
currentKey: string,
prevKey: string,
@@ -27,21 +12,6 @@ declare module "ioredis" {
requested: number | string,
ttl?: number | string,
): Promise<number>;
leaseApiRateLimit1Key(
baseKey: string,
now: number | string,
window: number | string,
maxReqs: number | string,
requested: number | string,
ttl?: number | string,
): Promise<number>;
throttleProvider(
key: string,
now: number | string,
windowSeconds: number | string,
limit: number | string,
member: string,
): Promise<[number, number]>;
releaseLock(lockKey: string, lockToken: string): Promise<number>;
renewLock(lockKey: string, lockToken: string, ttlSeconds: number | string): Promise<number>;
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,
src/services/ai/main.ts
+32 -51
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/services/api/handlers.ts
+20 -14
View File
@@ -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<F extends number | undefined>(
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<void> {
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<void> {
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<void> {
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<void> {
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<void> {
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}</style>
res: ServerResponse,
ctx: RouteContext,
): Promise<void> {
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}</style>
res: ServerResponse,
ctx: RouteContext,
): Promise<void> {
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");
src/services/api/main.ts
-19
View File
@@ -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();
src/services/delivery/main.ts
+36 -55
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/services/delivery/throttle.ts
+2 -50
View File
@@ -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<ThrottleResult> {
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
src/services/engine/main.ts
+36 -54
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/services/enricher/main.ts
+27 -55
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/services/events/main.ts
+21 -38
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/services/scheduler/main.ts
+33 -73
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/services/workflow/main.ts
+39 -78
View File
@@ -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<typeof createLogger>;
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<void> {
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;
}
src/shared/index.ts
-1
View File
@@ -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";
src/shared/utils.ts
-17
View File
@@ -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";
src/templates/index.ts
-67
View File
@@ -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<string, unknown>;
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<string, TemplateRenderer>();
/**
* 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);
}
src/workers/bootstrap.ts
+124
View File
@@ -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<void> {
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`);
}
tests/api-rate-limiter.test.ts
-1
View File
@@ -42,7 +42,6 @@ describe("ApiRateLimiter (In-Memory Token Leasing)", () => {
beforeEach(() => {
mockRedis = {
leaseApiRateLimit: vi.fn(),
eval: vi.fn(),
};
mockLogger = {
warn: vi.fn(),
tests/api.test.ts
+33 -131
View File
@@ -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<string, { value: number; ttl: number }>,
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<string, { value: number; ttl: number }>();
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<string, { value: number; ttl: number }>();
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<string, { value: number; ttl: number }>();
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);
});
});
});
tests/redis.test.ts
-194
View File
@@ -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),
);
});
});
});
tests/regressions.test.ts
+4 -4
View File
@@ -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");
tests/scheduler-poll.test.ts
+9 -14
View File
@@ -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,
tests/templates.test.ts
-147
View File
@@ -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: "<script>alert('xss')</script>",
company: 'Acme "Best" & Co.',
};
const rendered = renderWithTemplate(template, variables);
expect((rendered.content as any).htmlBody).toBe(
"Hello, &lt;script&gt;alert(&#039;xss&#039;)&lt;/script&gt; & welcome to Acme &quot;Best&quot; &amp; Co.!",
);
expect((rendered.content as any).textBody).toBe(
"Hello, <script>alert('xss')</script> & 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"');
});
});
tests/workers-bootstrap.test.ts
+96
View File
@@ -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();
});
});
tests/workers-workflow.test.ts
+4 -6
View File
@@ -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),
);
tests/workers.test.ts
+30 -148
View File
@@ -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<any>();
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" }),
tests/workflows.test.ts
+2 -1
View File
@@ -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(),