(
key: string,
ttlSeconds: number,
diff --git a/packages/vitnode/src/api/lib/module.ts b/packages/vitnode/src/api/lib/module.ts
index 30af2f7af..3eeb34cbd 100644
--- a/packages/vitnode/src/api/lib/module.ts
+++ b/packages/vitnode/src/api/lib/module.ts
@@ -1,6 +1,7 @@
import { OpenAPIHono } from "@hono/zod-openapi";
import type { BuildCronReturn } from "./cron";
+import type { BuildQueueTaskReturn } from "./queue";
import type { Route } from "./route";
import type { BuildWebSocketReturn } from "./websocket";
@@ -19,6 +20,7 @@ export interface BaseBuildModuleReturn<
modules?: BaseBuildModuleReturn[];
name: M;
pluginId: P;
+ queueTasks: BuildQueueTaskReturn[];
routes: Routes;
webSockets: BuildWebSocketReturn[];
}
@@ -43,12 +45,14 @@ export function buildModule<
name,
modules,
cronJobs = [],
+ queueTasks = [],
webSockets = [],
}: {
cronJobs?: BuildCronReturn[];
modules?: Modules;
name: M;
pluginId: P;
+ queueTasks?: BuildQueueTaskReturn[];
routes: Routes;
webSockets?: BuildWebSocketReturn[];
}): BuildModuleReturn {
@@ -66,5 +70,14 @@ export function buildModule<
});
}
- return { routes, pluginId, hono, name, modules, cronJobs, webSockets };
+ return {
+ routes,
+ pluginId,
+ hono,
+ name,
+ modules,
+ cronJobs,
+ queueTasks,
+ webSockets,
+ };
}
diff --git a/packages/vitnode/src/api/lib/plugin.ts b/packages/vitnode/src/api/lib/plugin.ts
index eac1d7bc4..991724282 100644
--- a/packages/vitnode/src/api/lib/plugin.ts
+++ b/packages/vitnode/src/api/lib/plugin.ts
@@ -3,6 +3,7 @@ import { OpenAPIHono } from "@hono/zod-openapi";
import type { CronJobConfig } from "./cron";
import type { BuildModuleReturn } from "./module";
import type { PermissionStaffConfig } from "./permission-staff";
+import type { QueueTaskConfig } from "./queue";
import type { WebSocketConfig } from "./websocket";
import { checkPluginId } from "./check-plugin-id";
@@ -12,6 +13,7 @@ export interface BuildPluginApiReturn {
hono: OpenAPIHono;
permissionStaff?: PermissionStaffConfig;
pluginId: string;
+ queueTasks?: Omit[];
webSockets?: Omit[];
}
@@ -29,6 +31,7 @@ export function buildApiPlugin({
const hono = new OpenAPIHono();
const cronJobs: BuildPluginApiReturn["cronJobs"] = [];
+ const queueTasks: BuildPluginApiReturn["queueTasks"] = [];
const webSockets: BuildPluginApiReturn["webSockets"] = [];
modules.forEach(handler => {
hono.route(`/${handler.name}`, handler.hono);
@@ -37,6 +40,10 @@ export function buildApiPlugin
({
cronJobs.push({ ...cron, module: handler.name });
});
+ handler.queueTasks?.forEach(task => {
+ queueTasks.push({ ...task, module: handler.name });
+ });
+
handler.webSockets?.forEach(webSocket => {
webSockets.push({ ...webSocket, module: handler.name });
});
@@ -46,6 +53,7 @@ export function buildApiPlugin
({
pluginId,
hono,
cronJobs,
+ queueTasks,
webSockets,
permissionStaff,
};
diff --git a/packages/vitnode/src/api/lib/queue.ts b/packages/vitnode/src/api/lib/queue.ts
new file mode 100644
index 000000000..8ef871bbb
--- /dev/null
+++ b/packages/vitnode/src/api/lib/queue.ts
@@ -0,0 +1,27 @@
+import type { Context } from "hono";
+
+import type { EnvVitNode } from "../middlewares/global.middleware";
+
+export interface BuildQueueTaskReturn {
+ description?: string;
+ handler: (
+ c: Context,
+ payload: Record,
+ ) => Promise | void;
+ maxAttempts?: number;
+ name: string;
+}
+
+export interface QueueTaskConfig extends BuildQueueTaskReturn {
+ module: string;
+ pluginId: string;
+}
+
+export function buildQueueTask({
+ name,
+ handler,
+ description,
+ maxAttempts,
+}: BuildQueueTaskReturn): BuildQueueTaskReturn {
+ return { name, handler, description, maxAttempts };
+}
diff --git a/packages/vitnode/src/api/middlewares/global.middleware.ts b/packages/vitnode/src/api/middlewares/global.middleware.ts
index e93f130e3..046828ad8 100644
--- a/packages/vitnode/src/api/middlewares/global.middleware.ts
+++ b/packages/vitnode/src/api/middlewares/global.middleware.ts
@@ -7,7 +7,8 @@ import type { VitNodeApiConfig, VitNodeConfig } from "@/vitnode.config";
import type { VitNodeRealtime } from "@/ws/registry";
import { CacheModel } from "@/api/lib/cache";
-import { EmailModel, type EmailModelSendArgs } from "@/api/models/email";
+import { EmailModel } from "@/api/models/email";
+import { QueueModel } from "@/api/models/queue";
import { SessionModel } from "@/api/models/session";
import { SessionAdminModel } from "@/api/models/session-admin";
import { CONFIG } from "@/lib/config";
@@ -15,6 +16,7 @@ import { realtime } from "@/ws/registry";
import type { BuildCronReturn } from "../lib/cron";
import type { PermissionStaffCatalogEntry } from "../lib/permission-staff";
+import type { BuildQueueTaskReturn } from "../lib/queue";
import type { WebSocketConfig } from "../lib/websocket";
import type { SSOApiPlugin } from "../models/sso";
@@ -75,17 +77,17 @@ export interface EnvVariablesVitNode {
pathToMessages: (path: string) => Promise<{ default: object }>;
permissionStaff: PermissionStaffCatalogEntry[];
plugins: { id: string }[];
+ queue: (BuildQueueTaskReturn & { module: string; pluginId: string })[];
webSockets: WebSocketConfig[];
};
db: Pick["dbProvider"];
- email: {
- send: (args: EmailModelSendArgs) => Promise;
- };
+ email: EmailModel;
ipAddress: string;
log: LoggerMiddlewareType;
plugin: {
id: string;
};
+ queue: QueueModel;
realtime: VitNodeRealtime;
user: null | {
avatarColor: string;
@@ -137,6 +139,17 @@ export const globalMiddleware = ({
})),
);
+ const queueMetadata = plugins.flatMap(plugin =>
+ (plugin.queueTasks ?? []).map(task => ({
+ pluginId: plugin.pluginId,
+ module: task.module,
+ name: task.name,
+ handler: task.handler,
+ description: task.description,
+ maxAttempts: task.maxAttempts,
+ })),
+ );
+
const webSocketsMetadata: WebSocketConfig[] = plugins.flatMap(plugin =>
(plugin.webSockets ?? []).map(webSocket => ({
...webSocket,
@@ -190,6 +203,7 @@ export const globalMiddleware = ({
c.set("db", dbProvider);
c.set("cache", new CacheModel(cacheClient, c));
c.set("email", new EmailModel(c));
+ c.set("queue", new QueueModel(c));
c.set("realtime", realtime);
c.set("core", {
@@ -214,6 +228,7 @@ export const globalMiddleware = ({
hasCronAdapter: !!cron,
plugins: pluginsMetadata,
cron: cronMetadata,
+ queue: queueMetadata,
webSockets: webSocketsMetadata,
permissionStaff: permissionStaffMetadata,
});
diff --git a/packages/vitnode/src/api/models/email.test.ts b/packages/vitnode/src/api/models/email.test.ts
new file mode 100644
index 000000000..b3649df94
--- /dev/null
+++ b/packages/vitnode/src/api/models/email.test.ts
@@ -0,0 +1,113 @@
+import type { Context } from "hono";
+
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("react-email", () => ({
+ render: vi.fn().mockResolvedValue("RENDERED"),
+}));
+
+import { EmailModel } from "./email";
+
+const makeCtx = (
+ overrides: { email?: unknown } = {},
+): {
+ ctx: Context;
+ dispatch: ReturnType;
+ error: ReturnType;
+ sendEmail: ReturnType;
+} => {
+ const dispatch = vi.fn().mockResolvedValue({ id: 1 });
+ const sendEmail = vi.fn().mockResolvedValue(undefined);
+ const error = vi.fn().mockResolvedValue(undefined);
+ const store: Record = {
+ core: {
+ email:
+ "email" in overrides ? overrides.email : { adapter: { sendEmail } },
+ plugins: [],
+ metadata: { title: "Test" },
+ pathToMessages: vi.fn().mockResolvedValue({ default: {} }),
+ },
+ queue: { dispatch },
+ log: { error, warn: vi.fn(), debug: vi.fn() },
+ };
+
+ return {
+ ctx: { get: (k: string) => store[k] } as unknown as Context,
+ dispatch,
+ sendEmail,
+ error,
+ };
+};
+
+describe("EmailModel.queue", () => {
+ it("renders the email and enqueues a send-email task", async () => {
+ const { ctx, dispatch } = makeCtx();
+
+ await new EmailModel(ctx).send({
+ to: "a@b.com",
+ locale: "en",
+ subject: "Hi",
+ content: () => null,
+ });
+
+ expect(dispatch).toHaveBeenCalledWith({
+ name: "send-email",
+ payload: {
+ to: "a@b.com",
+ subject: "Hi",
+ html: "RENDERED",
+ text: "RENDERED",
+ },
+ });
+ });
+
+ it("resolves a function subject before enqueueing", async () => {
+ const { ctx, dispatch } = makeCtx();
+
+ await new EmailModel(ctx).send({
+ to: "a@b.com",
+ locale: "en",
+ subject: () => "Computed",
+ content: () => null,
+ });
+
+ expect(dispatch.mock.calls[0][0].payload.subject).toBe("Computed");
+ });
+
+ it("throws when no email provider is configured", async () => {
+ const { ctx, dispatch } = makeCtx({ email: undefined });
+
+ await expect(
+ new EmailModel(ctx).send({
+ to: "a@b.com",
+ locale: "en",
+ subject: "Hi",
+ content: () => null,
+ }),
+ ).rejects.toThrow();
+ expect(dispatch).not.toHaveBeenCalled();
+ });
+});
+
+describe("EmailModel.deliver", () => {
+ it("sends a prebuilt email through the provider", async () => {
+ const { ctx, sendEmail } = makeCtx();
+
+ await new EmailModel(ctx).deliver({
+ to: "a@b.com",
+ subject: "Hi",
+ html: "hi
",
+ text: "hi",
+ });
+
+ expect(sendEmail).toHaveBeenCalledWith(
+ expect.objectContaining({
+ to: "a@b.com",
+ subject: "Hi",
+ html: "hi
",
+ text: "hi",
+ metadata: { title: "Test" },
+ }),
+ );
+ });
+});
diff --git a/packages/vitnode/src/api/models/email.ts b/packages/vitnode/src/api/models/email.ts
index 4737cd56e..e6e009929 100644
--- a/packages/vitnode/src/api/models/email.ts
+++ b/packages/vitnode/src/api/models/email.ts
@@ -50,6 +50,14 @@ export type EmailModelSendArgs = {
// eslint-disable-next-line perfectionist/sort-intersection-types
} & (EmailModelSendArgsWithEmail | EmailModelSendArgsWithUser);
+export interface BuiltEmail {
+ html: string;
+ replyTo?: string;
+ subject: string;
+ text: string;
+ to: string;
+}
+
export class EmailModel {
constructor(c: Context) {
this.c = c;
@@ -57,7 +65,18 @@ export class EmailModel {
protected readonly c: Context;
- async send({
+ private requireProvider() {
+ const provider = this.c.get("core").email?.adapter;
+ if (!provider) {
+ throw new HTTPException(500, {
+ message: "Email provider not found",
+ });
+ }
+
+ return provider;
+ }
+
+ async build({
html,
replyTo,
subject,
@@ -65,19 +84,12 @@ export class EmailModel {
user,
content,
locale: localeFromArgs,
- }: EmailModelSendArgs) {
+ }: EmailModelSendArgs): Promise {
const core = this.c.get("core");
- const provider = core.email?.adapter;
- if (!provider) {
- throw new HTTPException(500, {
- message: "Email provider not found",
- });
- }
-
const locale = localeFromArgs ?? user?.language ?? "en";
const pluginIds: string[] = [
"@vitnode/core",
- ...this.c.get("core").plugins.map(plugin => plugin.id),
+ ...core.plugins.map(plugin => plugin.id),
];
const messagesPromises = pluginIds.map(async pluginId => {
@@ -124,27 +136,44 @@ export class EmailModel {
});
}
- try {
- await provider.sendEmail({
- html: await render(htmlContent),
- to: emailTo,
- subject:
- typeof subject === "function"
- ? subject({ i18n: { locale, messages } })
- : subject,
- replyTo,
- metadata: core.metadata,
- text: await render(htmlContent, {
- plainText: true,
- }),
- });
- } catch (err) {
- const error =
- err instanceof Error
- ? err
- : new Error("Unknown error from email provider");
+ return {
+ to: emailTo,
+ subject:
+ typeof subject === "function"
+ ? subject({ i18n: { locale, messages } })
+ : subject,
+ html: await render(htmlContent),
+ text: await render(htmlContent, { plainText: true }),
+ replyTo,
+ };
+ }
- await this.c.get("log").error(`Failed to send email: ${error.message}`);
- }
+ async deliver(email: BuiltEmail): Promise {
+ const provider = this.requireProvider();
+
+ await provider.sendEmail({
+ html: email.html,
+ to: email.to,
+ subject: email.subject,
+ replyTo: email.replyTo,
+ metadata: this.c.get("core").metadata,
+ text: email.text,
+ });
+ }
+
+ async send(args: EmailModelSendArgs): Promise {
+ this.requireProvider();
+ const email = await this.build(args);
+
+ await this.c.get("queue").dispatch({
+ name: "send-email",
+ payload: {
+ to: email.to,
+ subject: email.subject,
+ html: email.html,
+ text: email.text,
+ ...(email.replyTo ? { replyTo: email.replyTo } : {}),
+ },
+ });
}
}
diff --git a/packages/vitnode/src/api/models/queue.test.ts b/packages/vitnode/src/api/models/queue.test.ts
new file mode 100644
index 000000000..c0cd10117
--- /dev/null
+++ b/packages/vitnode/src/api/models/queue.test.ts
@@ -0,0 +1,86 @@
+import type { Context } from "hono";
+
+import { describe, expect, it, vi } from "vitest";
+
+import { QueueModel } from "./queue";
+
+const makeCtx = (
+ overrides: {
+ plugin?: { id: string };
+ queue?: { maxAttempts?: number; name: string; pluginId: string }[];
+ } = {},
+): {
+ ctx: Context;
+ values: ReturnType;
+} => {
+ const values = vi.fn().mockReturnValue({
+ returning: vi.fn().mockResolvedValue([{ id: 1 }]),
+ });
+ const store: Record = {
+ db: { insert: vi.fn().mockReturnValue({ values }) },
+ core: { queue: overrides.queue ?? [] },
+ plugin: overrides.plugin,
+ };
+
+ return {
+ ctx: { get: (k: string) => store[k] } as unknown as Context,
+ values,
+ };
+};
+
+describe("QueueModel.dispatch", () => {
+ it("uses the explicit maxAttempts when provided", async () => {
+ const { ctx, values } = makeCtx({
+ queue: [{ name: "job", pluginId: "@vitnode/core", maxAttempts: 5 }],
+ });
+
+ await new QueueModel(ctx).dispatch({ name: "job", maxAttempts: 7 });
+
+ expect(values.mock.calls[0][0]).toMatchObject({ maxAttempts: 7 });
+ });
+
+ it("falls back to the registered task maxAttempts", async () => {
+ const { ctx, values } = makeCtx({
+ queue: [{ name: "job", pluginId: "@vitnode/core", maxAttempts: 5 }],
+ });
+
+ await new QueueModel(ctx).dispatch({ name: "job" });
+
+ expect(values.mock.calls[0][0]).toMatchObject({ maxAttempts: 5 });
+ });
+
+ it("defaults to 3 when the registered task has no maxAttempts", async () => {
+ const { ctx, values } = makeCtx({
+ queue: [{ name: "job", pluginId: "@vitnode/core" }],
+ });
+
+ await new QueueModel(ctx).dispatch({ name: "job" });
+
+ expect(values.mock.calls[0][0]).toMatchObject({ maxAttempts: 3 });
+ });
+
+ it("defaults to 3 when the task is not registered", async () => {
+ const { ctx, values } = makeCtx({ queue: [] });
+
+ await new QueueModel(ctx).dispatch({ name: "job" });
+
+ expect(values.mock.calls[0][0]).toMatchObject({ maxAttempts: 3 });
+ });
+
+ it("scopes the task lookup by pluginId", async () => {
+ const { ctx, values } = makeCtx({
+ plugin: { id: "@vitnode/blog" },
+ queue: [
+ { name: "job", pluginId: "@vitnode/core", maxAttempts: 5 },
+ { name: "job", pluginId: "@vitnode/blog", maxAttempts: 9 },
+ ],
+ });
+
+ await new QueueModel(ctx).dispatch({ name: "job" });
+
+ expect(values.mock.calls[0][0]).toMatchObject({
+ pluginId: "@vitnode/blog",
+ maxAttempts: 9,
+ });
+ });
+});
diff --git a/packages/vitnode/src/api/models/queue.ts b/packages/vitnode/src/api/models/queue.ts
new file mode 100644
index 000000000..38f66309f
--- /dev/null
+++ b/packages/vitnode/src/api/models/queue.ts
@@ -0,0 +1,57 @@
+import type { Context } from "hono";
+
+import { core_queue } from "@/database/queue";
+
+export interface QueueDispatchArgs {
+ availableAt?: Date;
+ maxAttempts?: number;
+ name: string;
+ payload?: Record;
+ priority?: number;
+ queue?: string;
+}
+
+/**
+ * Enqueue background work into the database-backed task queue, exposed on the
+ * request context as `c.get("queue")`. Rows are drained by the `process-queue`
+ * cron job. The task `name` must match a handler registered via
+ * {@link buildQueueTask} in a module's `queueTasks`.
+ */
+export class QueueModel {
+ constructor(c: Context) {
+ this.c = c;
+ }
+
+ protected readonly c: Context;
+
+ async dispatch({
+ name,
+ payload = {},
+ queue = "default",
+ priority = 0,
+ maxAttempts,
+ availableAt,
+ }: QueueDispatchArgs): Promise<{ id: number }> {
+ const pluginId = this.c.get("plugin")?.id ?? "@vitnode/core";
+
+ const registeredTask = this.c
+ .get("core")
+ .queue.find(task => task.pluginId === pluginId && task.name === name);
+
+ const [row] = await this.c
+ .get("db")
+ .insert(core_queue)
+ .values({
+ pluginId,
+ name,
+ queue,
+ payload,
+ priority,
+ maxAttempts: maxAttempts ?? registeredTask?.maxAttempts ?? 3,
+ availableAt: availableAt ?? new Date(),
+ })
+ .returning({ id: core_queue.id });
+
+ return row;
+ }
+}
diff --git a/packages/vitnode/src/api/modules/admin/advanced/advanced.admin.module.ts b/packages/vitnode/src/api/modules/admin/advanced/advanced.admin.module.ts
index a9c616501..281b67980 100644
--- a/packages/vitnode/src/api/modules/admin/advanced/advanced.admin.module.ts
+++ b/packages/vitnode/src/api/modules/admin/advanced/advanced.admin.module.ts
@@ -2,10 +2,11 @@ import { buildModule } from "@/api/lib/module";
import { CONFIG_PLUGIN } from "@/config";
import { cronAdminModule } from "./cron/cron.admin.module";
+import { queueAdminModule } from "./queue/queue.admin.module";
export const advancedAdminModule = buildModule({
pluginId: CONFIG_PLUGIN.pluginId,
name: "advanced",
routes: [],
- modules: [cronAdminModule],
+ modules: [cronAdminModule, queueAdminModule],
});
diff --git a/packages/vitnode/src/api/modules/admin/advanced/queue/queue.admin.module.ts b/packages/vitnode/src/api/modules/admin/advanced/queue/queue.admin.module.ts
new file mode 100644
index 000000000..3a09e7692
--- /dev/null
+++ b/packages/vitnode/src/api/modules/admin/advanced/queue/queue.admin.module.ts
@@ -0,0 +1,10 @@
+import { buildModule } from "@/api/lib/module";
+import { CONFIG_PLUGIN } from "@/config";
+
+import { getQueueTasksRoute } from "./routes/get.route";
+
+export const queueAdminModule = buildModule({
+ pluginId: CONFIG_PLUGIN.pluginId,
+ name: "queue",
+ routes: [getQueueTasksRoute],
+});
diff --git a/packages/vitnode/src/api/modules/admin/advanced/queue/routes/get.route.ts b/packages/vitnode/src/api/modules/admin/advanced/queue/routes/get.route.ts
new file mode 100644
index 000000000..40181afa9
--- /dev/null
+++ b/packages/vitnode/src/api/modules/admin/advanced/queue/routes/get.route.ts
@@ -0,0 +1,95 @@
+import { inArray } from "drizzle-orm";
+import z from "zod";
+
+import { buildRoute } from "@/api/lib/route";
+import {
+ withPagination,
+ zodPaginationPageInfo,
+ zodPaginationQuery,
+} from "@/api/lib/with-pagination";
+import { CONFIG_PLUGIN } from "@/config";
+import { core_queue } from "@/database/queue";
+
+const QUEUE_STATUSES = [
+ "pending",
+ "processing",
+ "completed",
+ "failed",
+] as const;
+
+export const getQueueTasksRoute = buildRoute({
+ pluginId: CONFIG_PLUGIN.pluginId,
+ adminStaffPermission: { module: "queue", permission: "can_view" },
+ route: {
+ method: "get",
+ description: "Get Admin Queue Tasks",
+ path: "/",
+ request: {
+ query: zodPaginationQuery.extend({
+ order: z.enum(["asc", "desc"]).optional(),
+ orderBy: z.enum(["createdAt", "availableAt", "status"]).optional(),
+ status: z.string().optional(),
+ }),
+ },
+ responses: {
+ 200: {
+ content: {
+ "application/json": {
+ schema: z.object({
+ edges: z.array(
+ z.object({
+ id: z.number(),
+ pluginId: z.string(),
+ name: z.string(),
+ queue: z.string(),
+ status: z.enum(QUEUE_STATUSES),
+ priority: z.number(),
+ attempts: z.number(),
+ maxAttempts: z.number(),
+ availableAt: z.date(),
+ reservedAt: z.date().nullable(),
+ lastError: z.string().nullable(),
+ createdAt: z.date(),
+ completedAt: z.date().nullable(),
+ }),
+ ),
+ pageInfo: zodPaginationPageInfo,
+ }),
+ },
+ },
+ description: "List of queue tasks",
+ },
+ },
+ },
+ handler: async c => {
+ const query = c.req.valid("query");
+ const statuses = (query.status?.split(",") ?? []).filter(
+ (status): status is (typeof QUEUE_STATUSES)[number] =>
+ (QUEUE_STATUSES as readonly string[]).includes(status),
+ );
+
+ const data = await withPagination({
+ params: { query },
+ c,
+ primaryCursor: core_queue.id,
+ where: statuses.length ? inArray(core_queue.status, statuses) : undefined,
+ query: async ({ limit, where, orderBy }) =>
+ await c
+ .get("db")
+ .select()
+ .from(core_queue)
+ .where(where)
+ .orderBy(orderBy)
+ .limit(limit),
+ table: core_queue,
+ orderBy: {
+ column: query.orderBy
+ ? core_queue[query.orderBy]
+ : core_queue.createdAt,
+ order: query.order ?? "desc",
+ },
+ });
+
+ return c.json(data);
+ },
+});
diff --git a/packages/vitnode/src/api/modules/admin/debug/debug.admin.module.ts b/packages/vitnode/src/api/modules/admin/debug/debug.admin.module.ts
index 7b39f0a43..7baa013e2 100644
--- a/packages/vitnode/src/api/modules/admin/debug/debug.admin.module.ts
+++ b/packages/vitnode/src/api/modules/admin/debug/debug.admin.module.ts
@@ -2,6 +2,7 @@ import { CONFIG_PLUGIN } from "../../../../config";
import { buildModule } from "../../../lib/module";
import { integrationsDebugAdminRoute } from "./routes/integrations.route";
import { logsDebugAdminRoute } from "./routes/logs.route";
+import { queueDebugAdminRoute } from "./routes/queue.route";
import { sendTestEmailDebugAdminRoute } from "./routes/send-test-email.route";
export const debugAdminModule = buildModule({
@@ -10,6 +11,7 @@ export const debugAdminModule = buildModule({
routes: [
logsDebugAdminRoute,
integrationsDebugAdminRoute,
+ queueDebugAdminRoute,
sendTestEmailDebugAdminRoute,
],
});
diff --git a/packages/vitnode/src/api/modules/admin/debug/routes/integrations.route.ts b/packages/vitnode/src/api/modules/admin/debug/routes/integrations.route.ts
index 5c89bf6a8..45fa5259e 100644
--- a/packages/vitnode/src/api/modules/admin/debug/routes/integrations.route.ts
+++ b/packages/vitnode/src/api/modules/admin/debug/routes/integrations.route.ts
@@ -1,6 +1,8 @@
+import { inArray, sql } from "drizzle-orm";
import { z } from "zod";
import { CONFIG_PLUGIN } from "@/config";
+import { core_queue } from "@/database/queue";
import { INSECURE_DEFAULT_CRON_SECRET } from "@/lib/config";
import { isRealtimePubSubEnabled, isWebSocketEnabled } from "@/ws/registry";
@@ -38,6 +40,16 @@ export const integrationsDebugAdminRoute = buildRoute({
email: z.object({
active: z.boolean(),
}),
+ queue: z.object({
+ // `true` when at least one queue task handler is registered.
+ active: z.boolean(),
+ // Number of pending tasks waiting to be processed.
+ pending: z.number(),
+ // Number of tasks currently being processed.
+ processing: z.number(),
+ // Registered queue task handlers across core + plugins.
+ tasks: z.number(),
+ }),
redis: z.object({
active: z.boolean(),
// `true` when Redis is configured but currently unreachable — a
@@ -60,6 +72,20 @@ export const integrationsDebugAdminRoute = buildRoute({
const captcha = core.captcha;
const redis = await c.get("cache").status();
+ const queueGrouped = await c
+ .get("db")
+ .select({
+ status: core_queue.status,
+ count: sql`count(*)::int`,
+ })
+ .from(core_queue)
+ .where(inArray(core_queue.status, ["pending", "processing"]))
+ .groupBy(core_queue.status);
+ const queuePending =
+ queueGrouped.find(row => row.status === "pending")?.count ?? 0;
+ const queueProcessing =
+ queueGrouped.find(row => row.status === "processing")?.count ?? 0;
+
return c.json(
{
captcha: {
@@ -76,6 +102,12 @@ export const integrationsDebugAdminRoute = buildRoute({
email: {
active: !!core.email?.adapter,
},
+ queue: {
+ active: core.queue.length > 0,
+ pending: queuePending,
+ processing: queueProcessing,
+ tasks: core.queue.length,
+ },
redis: {
active: redis.configured && redis.connected,
configuredButDown: redis.configured && !redis.connected,
diff --git a/packages/vitnode/src/api/modules/admin/debug/routes/queue.route.ts b/packages/vitnode/src/api/modules/admin/debug/routes/queue.route.ts
new file mode 100644
index 000000000..c6a30a18e
--- /dev/null
+++ b/packages/vitnode/src/api/modules/admin/debug/routes/queue.route.ts
@@ -0,0 +1,91 @@
+import { asc, desc, inArray, sql } from "drizzle-orm";
+import z from "zod";
+
+import { buildRoute } from "@/api/lib/route";
+import { CONFIG_PLUGIN } from "@/config";
+import { core_queue } from "@/database/queue";
+
+const QUEUE_STATUSES = [
+ "pending",
+ "processing",
+ "completed",
+ "failed",
+] as const;
+
+export const queueDebugAdminRoute = buildRoute({
+ pluginId: CONFIG_PLUGIN.pluginId,
+ adminStaffPermission: { module: "debug", permission: "can_view" },
+ route: {
+ method: "get",
+ description:
+ "Currently active (pending/processing) queue tasks and per-status counts.",
+ path: "/queue",
+ responses: {
+ 200: {
+ content: {
+ "application/json": {
+ schema: z.object({
+ counts: z.object({
+ pending: z.number(),
+ processing: z.number(),
+ completed: z.number(),
+ failed: z.number(),
+ }),
+ active: z.array(
+ z.object({
+ id: z.number(),
+ name: z.string(),
+ pluginId: z.string(),
+ queue: z.string(),
+ status: z.enum(QUEUE_STATUSES),
+ attempts: z.number(),
+ maxAttempts: z.number(),
+ availableAt: z.date(),
+ createdAt: z.date(),
+ }),
+ ),
+ }),
+ },
+ },
+ description: "Queue status",
+ },
+ },
+ },
+ handler: async c => {
+ const db = c.get("db");
+
+ const grouped = await db
+ .select({
+ status: core_queue.status,
+ count: sql`count(*)::int`,
+ })
+ .from(core_queue)
+ .groupBy(core_queue.status);
+
+ const counts = { pending: 0, processing: 0, completed: 0, failed: 0 };
+ for (const row of grouped) {
+ if (row.status in counts) {
+ counts[row.status] = row.count;
+ }
+ }
+
+ const active = await db
+ .select({
+ id: core_queue.id,
+ name: core_queue.name,
+ pluginId: core_queue.pluginId,
+ queue: core_queue.queue,
+ status: core_queue.status,
+ attempts: core_queue.attempts,
+ maxAttempts: core_queue.maxAttempts,
+ availableAt: core_queue.availableAt,
+ createdAt: core_queue.createdAt,
+ })
+ .from(core_queue)
+ .where(inArray(core_queue.status, ["pending", "processing"]))
+ .orderBy(desc(core_queue.priority), asc(core_queue.availableAt))
+ .limit(50);
+
+ return c.json({ counts, active }, 200);
+ },
+});
diff --git a/packages/vitnode/src/api/modules/queue/cron/process-queue.cron.ts b/packages/vitnode/src/api/modules/queue/cron/process-queue.cron.ts
new file mode 100644
index 000000000..baef34e6f
--- /dev/null
+++ b/packages/vitnode/src/api/modules/queue/cron/process-queue.cron.ts
@@ -0,0 +1,12 @@
+import { buildCron } from "@/api/lib/cron";
+
+import { processQueueTasks } from "../helpers/process-queue-tasks";
+
+export const processQueueCron = buildCron({
+ name: "process-queue",
+ description: "Process pending database queue tasks",
+ schedule: "* * * * *",
+ handler: async c => {
+ await processQueueTasks(c);
+ },
+});
diff --git a/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts b/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts
new file mode 100644
index 000000000..2bd9fca7e
--- /dev/null
+++ b/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts
@@ -0,0 +1,122 @@
+import type { Context } from "hono";
+
+import { and, asc, desc, eq, inArray, lt, lte, sql } from "drizzle-orm";
+
+import type { EnvVitNode } from "@/api/middlewares/global.middleware";
+
+import { core_queue } from "@/database/queue";
+import { resolveQueueTaskOutcome } from "@/lib/api/resolve-queue-task-outcome";
+
+const QUEUE_BATCH_SIZE = 25;
+const QUEUE_LOCK_KEY = "queue:process";
+const QUEUE_LOCK_TTL_SECONDS = 55;
+const QUEUE_RETENTION_DAYS = 7;
+
+/**
+ * Drain due queue tasks. Correctness comes from Postgres
+ * `FOR UPDATE SKIP LOCKED`, which lets many instances claim disjoint batches
+ * safely; the optional Redis lock (`c.get("cache").acquireLock`) is only an
+ * optimization so a single instance drains per tick when Redis is configured.
+ */
+export const processQueueTasks = async (
+ c: Context,
+): Promise => {
+ const gotLock = await c
+ .get("cache")
+ .acquireLock(QUEUE_LOCK_KEY, QUEUE_LOCK_TTL_SECONDS);
+ if (!gotLock) return;
+
+ try {
+ const db = c.get("db");
+ const now = new Date();
+
+ const claimed = await db.transaction(async tx => {
+ const rows = await tx
+ .select({ id: core_queue.id })
+ .from(core_queue)
+ .where(
+ and(
+ eq(core_queue.status, "pending"),
+ lte(core_queue.availableAt, now),
+ ),
+ )
+ .orderBy(desc(core_queue.priority), asc(core_queue.id))
+ .limit(QUEUE_BATCH_SIZE)
+ .for("update", { skipLocked: true });
+
+ if (rows.length === 0) return [];
+
+ return tx
+ .update(core_queue)
+ .set({
+ status: "processing",
+ reservedAt: now,
+ attempts: sql`${core_queue.attempts} + 1`,
+ })
+ .where(
+ inArray(
+ core_queue.id,
+ rows.map(row => row.id),
+ ),
+ )
+ .returning();
+ });
+
+ if (claimed.length > 0) {
+ const handlerMap = new Map(
+ c
+ .get("core")
+ .queue.map(task => [`${task.pluginId}:${task.name}`, task]),
+ );
+
+ for (const task of claimed) {
+ const key = `${task.pluginId}:${task.name}`;
+ const registered = handlerMap.get(key);
+ let error: null | string = null;
+
+ if (!registered) {
+ error = `No handler registered for queue task "${key}"`;
+ await c.get("log").warn(error);
+ } else {
+ try {
+ await registered.handler(c, task.payload);
+ } catch (err) {
+ error = err instanceof Error ? err.message : String(err);
+ await c.get("log").error(`Queue task "${key}" failed: ${error}`);
+ }
+ }
+
+ const outcome = resolveQueueTaskOutcome({
+ attempts: task.attempts,
+ maxAttempts: task.maxAttempts,
+ error,
+ });
+
+ await db
+ .update(core_queue)
+ .set({
+ status: outcome.status,
+ lastError: outcome.lastError,
+ availableAt: outcome.availableAt ?? task.availableAt,
+ completedAt: outcome.completedAt ?? null,
+ reservedAt: null,
+ })
+ .where(eq(core_queue.id, task.id));
+ }
+ }
+
+ const cutoff = new Date(
+ now.getTime() - QUEUE_RETENTION_DAYS * 24 * 60 * 60 * 1000,
+ );
+ await db
+ .delete(core_queue)
+ .where(
+ and(
+ inArray(core_queue.status, ["completed", "failed"]),
+ lt(core_queue.completedAt, cutoff),
+ ),
+ );
+ } finally {
+ await c.get("cache").releaseLock(QUEUE_LOCK_KEY);
+ }
+};
diff --git a/packages/vitnode/src/api/modules/queue/queue.module.ts b/packages/vitnode/src/api/modules/queue/queue.module.ts
new file mode 100644
index 000000000..964f4cecb
--- /dev/null
+++ b/packages/vitnode/src/api/modules/queue/queue.module.ts
@@ -0,0 +1,13 @@
+import { buildModule } from "@/api/lib/module";
+import { CONFIG_PLUGIN } from "@/config";
+
+import { processQueueCron } from "./cron/process-queue.cron";
+import { sendEmailQueueTask } from "./tasks/send-email.task";
+
+export const queueModule = buildModule({
+ pluginId: CONFIG_PLUGIN.pluginId,
+ name: "queue",
+ routes: [],
+ cronJobs: [processQueueCron],
+ queueTasks: [sendEmailQueueTask],
+});
diff --git a/packages/vitnode/src/api/modules/queue/tasks/send-email.task.ts b/packages/vitnode/src/api/modules/queue/tasks/send-email.task.ts
new file mode 100644
index 000000000..baedc437b
--- /dev/null
+++ b/packages/vitnode/src/api/modules/queue/tasks/send-email.task.ts
@@ -0,0 +1,22 @@
+import { z } from "zod";
+
+import { buildQueueTask } from "@/api/lib/queue";
+import { EmailModel } from "@/api/models/email";
+
+export const sendEmailPayloadSchema = z.object({
+ to: z.string(),
+ subject: z.string(),
+ html: z.string(),
+ text: z.string(),
+ replyTo: z.string().optional(),
+});
+
+export const sendEmailQueueTask = buildQueueTask({
+ name: "send-email",
+ description: "Deliver a rendered email through the configured provider.",
+ handler: async (c, payload) => {
+ const email = sendEmailPayloadSchema.parse(payload);
+
+ await new EmailModel(c).deliver(email);
+ },
+});
diff --git a/packages/vitnode/src/api/plugin.ts b/packages/vitnode/src/api/plugin.ts
index f55b716a0..da9d833e9 100644
--- a/packages/vitnode/src/api/plugin.ts
+++ b/packages/vitnode/src/api/plugin.ts
@@ -4,11 +4,18 @@ import { buildApiPlugin } from "./lib/plugin";
import { adminModule } from "./modules/admin/admin.module";
import { cronModule } from "./modules/cron/cron.module";
import { middlewareModule } from "./modules/middleware/middleware.module";
+import { queueModule } from "./modules/queue/queue.module";
import { usersModule } from "./modules/users/users.module";
export const newBuildPluginApiCore = buildApiPlugin({
pluginId: CONFIG_PLUGIN.pluginId,
- modules: [middlewareModule, usersModule, adminModule, cronModule],
+ modules: [
+ middlewareModule,
+ usersModule,
+ adminModule,
+ cronModule,
+ queueModule,
+ ],
permissionStaff: {
moderator: {
users: ["can_edit"],
@@ -29,6 +36,7 @@ export const newBuildPluginApiCore = buildApiPlugin({
"can_view",
{ permission: "can_send_test_email", dependsOn: ["can_view"] },
],
+ queue: ["can_view"],
staff_moderators: [
"can_view",
{ permission: "can_create", dependsOn: ["can_view"] },
diff --git a/packages/vitnode/src/database/admins.ts b/packages/vitnode/src/database/admins.ts
index 1150cb08e..4b4d07754 100644
--- a/packages/vitnode/src/database/admins.ts
+++ b/packages/vitnode/src/database/admins.ts
@@ -21,6 +21,7 @@ export const core_admin_permissions = pgTable(
updatedAt: t
.timestamp()
.notNull()
+ .defaultNow()
.$onUpdate(() => new Date()),
protected: t.boolean().notNull().default(false),
data: t
diff --git a/packages/vitnode/src/database/languages.ts b/packages/vitnode/src/database/languages.ts
index 06b28dc57..06ac4c3a8 100644
--- a/packages/vitnode/src/database/languages.ts
+++ b/packages/vitnode/src/database/languages.ts
@@ -15,6 +15,7 @@ export const core_languages = pgTable(
updatedAt: t
.timestamp()
.notNull()
+ .defaultNow()
.$onUpdate(() => new Date()),
time24: t.boolean().notNull().default(false),
}),
diff --git a/packages/vitnode/src/database/moderators.ts b/packages/vitnode/src/database/moderators.ts
index cc85ef3cb..7cdfe9ac0 100644
--- a/packages/vitnode/src/database/moderators.ts
+++ b/packages/vitnode/src/database/moderators.ts
@@ -20,6 +20,7 @@ export const core_moderators_permissions = pgTable(
updatedAt: t
.timestamp()
.notNull()
+ .defaultNow()
.$onUpdate(() => new Date()),
protected: t.boolean().notNull().default(false),
data: t
diff --git a/packages/vitnode/src/database/queue.ts b/packages/vitnode/src/database/queue.ts
new file mode 100644
index 000000000..f1a440c5d
--- /dev/null
+++ b/packages/vitnode/src/database/queue.ts
@@ -0,0 +1,35 @@
+import { index, pgTable } from "drizzle-orm/pg-core";
+
+export const core_queue = pgTable(
+ "core_queue",
+ t => ({
+ id: t.serial().primaryKey(),
+ pluginId: t.varchar({ length: 100 }).notNull(),
+ name: t.varchar({ length: 100 }).notNull(),
+ queue: t.varchar({ length: 100 }).notNull().default("default"),
+ status: t
+ .varchar({
+ enum: ["pending", "processing", "completed", "failed"],
+ length: 20,
+ })
+ .notNull()
+ .default("pending"),
+ payload: t.jsonb().$type>().notNull().default({}),
+ priority: t.integer().notNull().default(0),
+ attempts: t.integer().notNull().default(0),
+ maxAttempts: t.integer().notNull().default(3),
+ availableAt: t.timestamp().notNull().defaultNow(),
+ reservedAt: t.timestamp(),
+ lastError: t.text(),
+ createdAt: t.timestamp().notNull().defaultNow(),
+ updatedAt: t
+ .timestamp()
+ .notNull()
+ .defaultNow()
+ .$onUpdate(() => new Date()),
+ completedAt: t.timestamp(),
+ }),
+ t => [
+ index("core_queue_status_available_at_idx").on(t.status, t.availableAt),
+ ],
+).enableRLS();
diff --git a/packages/vitnode/src/database/users.ts b/packages/vitnode/src/database/users.ts
index bf5ff4b18..1eeb639ba 100644
--- a/packages/vitnode/src/database/users.ts
+++ b/packages/vitnode/src/database/users.ts
@@ -111,6 +111,7 @@ export const core_users_sso = pgTable(
updatedAt: t
.timestamp()
.notNull()
+ .defaultNow()
.$onUpdate(() => new Date()),
}),
t => [index("core_users_sso_user_id_idx").on(t.userId)],
diff --git a/packages/vitnode/src/lib/api/get-queue-backoff-date.test.ts b/packages/vitnode/src/lib/api/get-queue-backoff-date.test.ts
new file mode 100644
index 000000000..b284d37ce
--- /dev/null
+++ b/packages/vitnode/src/lib/api/get-queue-backoff-date.test.ts
@@ -0,0 +1,41 @@
+import { describe, expect, it } from "vitest";
+
+import { getQueueBackoffDate } from "./get-queue-backoff-date";
+
+describe("getQueueBackoffDate", () => {
+ const from = new Date("2024-01-01T00:00:00Z");
+ const secondsFrom = (date: Date) => (date.getTime() - from.getTime()) / 1000;
+
+ it("delays by the base amount after the first attempt", () => {
+ expect(secondsFrom(getQueueBackoffDate(1, from))).toBe(10);
+ });
+
+ it("grows exponentially with attempts", () => {
+ expect(secondsFrom(getQueueBackoffDate(2, from))).toBe(20);
+ expect(secondsFrom(getQueueBackoffDate(3, from))).toBe(40);
+ expect(secondsFrom(getQueueBackoffDate(4, from))).toBe(80);
+ });
+
+ it("is monotonically increasing", () => {
+ const delays = [1, 2, 3, 4, 5].map(a =>
+ secondsFrom(getQueueBackoffDate(a, from)),
+ );
+ for (let i = 1; i < delays.length; i++) {
+ expect(delays[i]).toBeGreaterThan(delays[i - 1]);
+ }
+ });
+
+ it("caps the delay at one hour", () => {
+ expect(secondsFrom(getQueueBackoffDate(50, from))).toBe(3600);
+ });
+
+ it("treats attempts <= 1 as the base delay (no negative exponent)", () => {
+ expect(secondsFrom(getQueueBackoffDate(0, from))).toBe(10);
+ });
+
+ it("returns a date after the provided origin", () => {
+ expect(getQueueBackoffDate(1, from).getTime()).toBeGreaterThan(
+ from.getTime(),
+ );
+ });
+});
diff --git a/packages/vitnode/src/lib/api/get-queue-backoff-date.ts b/packages/vitnode/src/lib/api/get-queue-backoff-date.ts
new file mode 100644
index 000000000..ebb2d5a38
--- /dev/null
+++ b/packages/vitnode/src/lib/api/get-queue-backoff-date.ts
@@ -0,0 +1,20 @@
+const BASE_DELAY_SECONDS = 10;
+const MAX_DELAY_SECONDS = 60 * 60;
+
+/**
+ * Exponential backoff for a failed queue task. `attempts` is the number of
+ * attempts already made (>= 1); the delay grows as
+ * `BASE_DELAY_SECONDS * 2^(attempts - 1)`, capped at {@link MAX_DELAY_SECONDS}.
+ */
+export const getQueueBackoffDate = (
+ attempts: number,
+ from: Date = new Date(),
+): Date => {
+ const exponent = Math.max(0, attempts - 1);
+ const delaySeconds = Math.min(
+ BASE_DELAY_SECONDS * 2 ** exponent,
+ MAX_DELAY_SECONDS,
+ );
+
+ return new Date(from.getTime() + delaySeconds * 1000);
+};
diff --git a/packages/vitnode/src/lib/api/resolve-queue-task-outcome.test.ts b/packages/vitnode/src/lib/api/resolve-queue-task-outcome.test.ts
new file mode 100644
index 000000000..33de0ae5a
--- /dev/null
+++ b/packages/vitnode/src/lib/api/resolve-queue-task-outcome.test.ts
@@ -0,0 +1,78 @@
+import { describe, expect, it } from "vitest";
+
+import { resolveQueueTaskOutcome } from "./resolve-queue-task-outcome";
+
+describe("resolveQueueTaskOutcome", () => {
+ const now = new Date("2024-01-01T00:00:00Z");
+
+ it("completes when there is no error", () => {
+ const outcome = resolveQueueTaskOutcome({
+ attempts: 1,
+ maxAttempts: 3,
+ now,
+ });
+
+ expect(outcome.status).toBe("completed");
+ expect(outcome.completedAt).toEqual(now);
+ expect(outcome.lastError).toBeNull();
+ expect(outcome.availableAt).toBeUndefined();
+ });
+
+ it("retries with a backoff when the error is recoverable", () => {
+ const outcome = resolveQueueTaskOutcome({
+ attempts: 1,
+ maxAttempts: 3,
+ error: "boom",
+ now,
+ });
+
+ expect(outcome.status).toBe("pending");
+ expect(outcome.lastError).toBe("boom");
+ expect(outcome.availableAt?.getTime()).toBeGreaterThan(now.getTime());
+ expect(outcome.completedAt).toBeUndefined();
+ });
+
+ it("fails once attempts reach maxAttempts", () => {
+ const outcome = resolveQueueTaskOutcome({
+ attempts: 3,
+ maxAttempts: 3,
+ error: "boom",
+ now,
+ });
+
+ expect(outcome.status).toBe("failed");
+ expect(outcome.lastError).toBe("boom");
+ expect(outcome.completedAt).toEqual(now);
+ expect(outcome.availableAt).toBeUndefined();
+ });
+
+ it("fails when attempts exceed maxAttempts", () => {
+ const outcome = resolveQueueTaskOutcome({
+ attempts: 5,
+ maxAttempts: 3,
+ error: "boom",
+ now,
+ });
+
+ expect(outcome.status).toBe("failed");
+ });
+
+ it("backs off further on later retries", () => {
+ const first = resolveQueueTaskOutcome({
+ attempts: 1,
+ maxAttempts: 5,
+ error: "boom",
+ now,
+ });
+ const second = resolveQueueTaskOutcome({
+ attempts: 2,
+ maxAttempts: 5,
+ error: "boom",
+ now,
+ });
+
+ expect(second.availableAt?.getTime()).toBeGreaterThan(
+ first.availableAt?.getTime() ?? 0,
+ );
+ });
+});
diff --git a/packages/vitnode/src/lib/api/resolve-queue-task-outcome.ts b/packages/vitnode/src/lib/api/resolve-queue-task-outcome.ts
new file mode 100644
index 000000000..78fac74d7
--- /dev/null
+++ b/packages/vitnode/src/lib/api/resolve-queue-task-outcome.ts
@@ -0,0 +1,42 @@
+import { getQueueBackoffDate } from "./get-queue-backoff-date";
+
+export type QueueTaskStatus = "completed" | "failed" | "pending" | "processing";
+
+export interface QueueTaskOutcome {
+ availableAt?: Date;
+ completedAt?: Date;
+ lastError: null | string;
+ status: QueueTaskStatus;
+}
+
+/**
+ * Decide the next state of a task after its handler ran. `attempts` is the
+ * count including the run that just happened. Without an error the task is
+ * `completed`; with one it is retried (`pending` with a backoff `availableAt`)
+ * until `maxAttempts` is reached, after which it is `failed`.
+ */
+export const resolveQueueTaskOutcome = ({
+ attempts,
+ maxAttempts,
+ error,
+ now = new Date(),
+}: {
+ attempts: number;
+ error?: null | string;
+ maxAttempts: number;
+ now?: Date;
+}): QueueTaskOutcome => {
+ if (!error) {
+ return { status: "completed", completedAt: now, lastError: null };
+ }
+
+ if (attempts < maxAttempts) {
+ return {
+ status: "pending",
+ availableAt: getQueueBackoffDate(attempts, now),
+ lastError: error,
+ };
+ }
+
+ return { status: "failed", completedAt: now, lastError: error };
+};
diff --git a/packages/vitnode/src/locales/en.json b/packages/vitnode/src/locales/en.json
index 27fef00d9..b591e6eb4 100644
--- a/packages/vitnode/src/locales/en.json
+++ b/packages/vitnode/src/locales/en.json
@@ -15,6 +15,8 @@
"@vitnode/core:system": "System",
"@vitnode/core:system:can_view": "View integrations",
"@vitnode/core:system:can_send_test_email": "Send test email",
+ "@vitnode/core:queue": "Queue Tasks",
+ "@vitnode/core:queue:can_view": "View queue tasks",
"@vitnode/core:staff_moderators": "Staff: Moderators",
"@vitnode/core:staff_moderators:can_view": "View moderators list",
"@vitnode/core:staff_moderators:can_create": "Create moderators",
@@ -274,7 +276,8 @@
},
"advanced": {
"title": "Advanced",
- "cron": "Cron Jobs"
+ "cron": "Cron Jobs",
+ "queue": "Queue Tasks"
}
}
},
@@ -302,6 +305,26 @@
}
}
}
+ },
+ "queue": {
+ "title": "Queue Tasks",
+ "desc": "Monitor background tasks queued for processing by the cron worker.",
+ "list": {
+ "name": "Task",
+ "queue": "Queue",
+ "status": "Status",
+ "attempts": "Attempts",
+ "availableAt": "Available At",
+ "createdAt": "Created At",
+ "lastError": "Last Error",
+ "statusFilter": "Status"
+ },
+ "status": {
+ "pending": "Pending",
+ "processing": "Processing",
+ "completed": "Completed",
+ "failed": "Failed"
+ }
}
},
"user": {
@@ -544,12 +567,35 @@
"insecure": "Using the default CRON_SECRET — set a secure one in production.",
"not_configured": "No cron adapter configured — jobs won't run automatically.",
"jobs": "{count, plural, =0 {No jobs scheduled} one {# job scheduled} other {# jobs scheduled}}"
+ },
+ "queue": {
+ "title": "Queue Tasks",
+ "desc": "Background jobs enqueued in the database and drained by the cron worker.",
+ "tasks": "{count, plural, =0 {No tasks registered} one {# task registered} other {# tasks registered}}",
+ "queued": "{pending} pending · {processing} running"
}
}
},
"debug": {
"title": "Debug Panel",
"desc": "Check logs, errors, and other debug information.",
+ "queue": {
+ "title": "Queue Tasks",
+ "empty": "No active queue tasks.",
+ "counts": {
+ "pending": "Pending",
+ "processing": "Processing",
+ "completed": "Completed",
+ "failed": "Failed"
+ },
+ "list": {
+ "name": "Task",
+ "queue": "Queue",
+ "status": "Status",
+ "attempts": "Attempts",
+ "availableAt": "Available At"
+ }
+ },
"actions": {
"clear_cache": {
"label": "Clear Cache",
diff --git a/packages/vitnode/src/routes/admin/core/advanced/queue/page.tsx b/packages/vitnode/src/routes/admin/core/advanced/queue/page.tsx
new file mode 100644
index 000000000..1451b8db6
--- /dev/null
+++ b/packages/vitnode/src/routes/admin/core/advanced/queue/page.tsx
@@ -0,0 +1,51 @@
+import { getTranslations } from "next-intl/server";
+import dynamic from "next/dynamic";
+import { notFound } from "next/navigation";
+import React from "react";
+
+import { I18nProvider } from "@/components/i18n-provider";
+import { DataTableSkeleton } from "@/components/table/data-table";
+import { HeaderContent } from "@/components/ui/header-content";
+import { checkAdminPermissionApi } from "@/lib/api/get-session-admin-api";
+
+const QueueTableView = dynamic(async () =>
+ import("@/views/admin/views/core/advanced/queue/queue-table-view").then(
+ module => ({
+ default: module.QueueTableView,
+ }),
+ ),
+);
+
+export const generateMetadata = async () => {
+ const t = await getTranslations("admin.advanced.queue");
+
+ return {
+ title: t("title"),
+ description: t("desc"),
+ };
+};
+
+export default async function Page(
+ props: React.ComponentProps,
+) {
+ const [t, canView] = await Promise.all([
+ getTranslations("admin.advanced.queue"),
+ checkAdminPermissionApi({ module: "queue", permission: "can_view" }),
+ ]);
+
+ if (!canView) {
+ notFound();
+ }
+
+ return (
+
+
+
+
+ }>
+
+
+
+
+ );
+}
diff --git a/packages/vitnode/src/routes/admin/core/debug/page.tsx b/packages/vitnode/src/routes/admin/core/debug/page.tsx
index d3e73773f..e1462f388 100644
--- a/packages/vitnode/src/routes/admin/core/debug/page.tsx
+++ b/packages/vitnode/src/routes/admin/core/debug/page.tsx
@@ -17,6 +17,12 @@ const SystemLogsView = dynamic(async () =>
),
);
+const QueueView = dynamic(async () =>
+ import("@/views/admin/views/core/debug/queue/queue-view").then(module => ({
+ default: module.QueueView,
+ })),
+);
+
export const generateMetadata = async () => {
const t = await getTranslations("admin.debug");
@@ -40,12 +46,17 @@ export default async function Page(
}
return (
-
+
{canClearCache && }
+
+
}>
+
+
+
}>
diff --git a/packages/vitnode/src/views/admin/layouts/sidebar/nav/get-admin-nav.tsx b/packages/vitnode/src/views/admin/layouts/sidebar/nav/get-admin-nav.tsx
index b9b7c2c4e..7a0ef4e46 100644
--- a/packages/vitnode/src/views/admin/layouts/sidebar/nav/get-admin-nav.tsx
+++ b/packages/vitnode/src/views/admin/layouts/sidebar/nav/get-admin-nav.tsx
@@ -188,6 +188,15 @@ export const getAdminNav = async ({
title: t("admin.global.nav.advanced.cron"),
href: "/admin/core/advanced/cron",
},
+ {
+ title: t("admin.global.nav.advanced.queue"),
+ href: "/admin/core/advanced/queue",
+ permission: {
+ plugin: CONFIG_PLUGIN.pluginId,
+ module: "queue",
+ permission: "can_view",
+ },
+ },
],
},
],
diff --git a/packages/vitnode/src/views/admin/views/core/advanced/queue/badges/status-badge.tsx b/packages/vitnode/src/views/admin/views/core/advanced/queue/badges/status-badge.tsx
new file mode 100644
index 000000000..e7e8e3fdb
--- /dev/null
+++ b/packages/vitnode/src/views/admin/views/core/advanced/queue/badges/status-badge.tsx
@@ -0,0 +1,45 @@
+import {
+ CircleCheckIcon,
+ CircleXIcon,
+ ClockIcon,
+ LoaderIcon,
+} from "lucide-react";
+import { useTranslations } from "next-intl";
+
+import { Badge } from "@/components/ui/badge";
+
+export type QueueTaskStatus = "completed" | "failed" | "pending" | "processing";
+
+export const QueueStatusBadge = ({ status }: { status: QueueTaskStatus }) => {
+ const t = useTranslations("admin.advanced.queue.status");
+
+ if (status === "processing") {
+ return (
+
+ {t("processing")}
+
+ );
+ }
+
+ if (status === "completed") {
+ return (
+
+ {t("completed")}
+
+ );
+ }
+
+ if (status === "failed") {
+ return (
+
+ {t("failed")}
+
+ );
+ }
+
+ return (
+
+ {t("pending")}
+
+ );
+};
diff --git a/packages/vitnode/src/views/admin/views/core/advanced/queue/queue-table-view.tsx b/packages/vitnode/src/views/admin/views/core/advanced/queue/queue-table-view.tsx
new file mode 100644
index 000000000..8da4b52d3
--- /dev/null
+++ b/packages/vitnode/src/views/admin/views/core/advanced/queue/queue-table-view.tsx
@@ -0,0 +1,111 @@
+import { getTranslations } from "next-intl/server";
+
+import { queueAdminModule } from "@/api/modules/admin/advanced/queue/queue.admin.module";
+import { DateFormat } from "@/components/date-format";
+import {
+ DataTable,
+ type SearchParamsDataTable,
+} from "@/components/table/data-table";
+import { fetcher } from "@/lib/fetcher";
+
+import { QueueStatusBadge } from "./badges/status-badge";
+
+const QUEUE_STATUSES = [
+ "pending",
+ "processing",
+ "completed",
+ "failed",
+] as const;
+
+export const QueueTableView = async ({
+ searchParams,
+}: {
+ searchParams: Promise
;
+}) => {
+ const query = await searchParams;
+ const res = await fetcher(queueAdminModule, {
+ path: "/",
+ method: "get",
+ module: "queue",
+ prefixPath: "/admin/advanced",
+ args: {
+ query,
+ },
+ withPagination: true,
+ });
+
+ const [data, t] = await Promise.all([
+ res.json(),
+ getTranslations("admin.advanced.queue"),
+ ]);
+
+ return (
+ (
+
+
{row.name}
+
{row.pluginId}
+
+
+ {COUNT_STATUSES.map(status => (
+
+
+ {t(`counts.${status}`)}
+
+
{data.counts[status]}
+
+ ))}
+
+
+ {data.active.length === 0 ? (
+
{t("empty")}
+ ) : (
+
+
+
+
+ {t("list.name")}
+ {t("list.queue")}
+ {t("list.status")}
+ {t("list.attempts")}
+ {t("list.availableAt")}
+
+
+
+ {data.active.map(task => (
+
+
+
+ {task.name}
+
+ {task.pluginId}
+
+
+
+ {task.queue}
+
+
+
+
+ {task.attempts}/{task.maxAttempts}
+
+
+
+
+
+ ))}
+
+
+
+ )}
+
- {["websocket", "redis", "email", "cron", "captcha"].map(id => (
+ {["websocket", "redis", "email", "cron", "queue", "captcha"].map(id => (
))}