From 5fff6c3a697cd91db5a8c4a1debd151a30dc3524 Mon Sep 17 00:00:00 2001 From: aXenDeveloper Date: Sat, 4 Jul 2026 21:37:14 +0200 Subject: [PATCH 1/2] feat: Add queue tasks --- apps/api/src/locales/@vitnode/core/en.json | 48 +- apps/docs/content/docs/dev/advanced/queue.mdx | 180 ++ .../migrations/0006_greedy_apocalypse.sql | 20 + apps/docs/migrations/meta/0006_snapshot.json | 1761 +++++++++++++++++ apps/docs/migrations/meta/_journal.json | 7 + .../core/advanced/queue/page.tsx | 51 + .../(vitnode-core)/core/debug/page.tsx | 13 +- apps/docs/src/locales/@vitnode/core/en.json | 48 +- packages/vitnode/src/api/lib/cache.test.ts | 55 + packages/vitnode/src/api/lib/cache.ts | 31 + packages/vitnode/src/api/lib/module.ts | 15 +- packages/vitnode/src/api/lib/plugin.ts | 8 + packages/vitnode/src/api/lib/queue.ts | 27 + .../src/api/middlewares/global.middleware.ts | 23 +- packages/vitnode/src/api/models/email.test.ts | 113 ++ packages/vitnode/src/api/models/email.ts | 91 +- packages/vitnode/src/api/models/queue.ts | 53 + .../admin/advanced/advanced.admin.module.ts | 3 +- .../advanced/queue/queue.admin.module.ts | 10 + .../admin/advanced/queue/routes/get.route.ts | 95 + .../modules/admin/debug/debug.admin.module.ts | 2 + .../admin/debug/routes/integrations.route.ts | 32 + .../modules/admin/debug/routes/queue.route.ts | 91 + .../modules/queue/cron/process-queue.cron.ts | 12 + .../queue/helpers/process-queue-tasks.ts | 122 ++ .../src/api/modules/queue/queue.module.ts | 13 + .../modules/queue/tasks/send-email.task.ts | 22 + packages/vitnode/src/api/plugin.ts | 10 +- packages/vitnode/src/database/queue.ts | 34 + .../lib/api/get-queue-backoff-date.test.ts | 41 + .../src/lib/api/get-queue-backoff-date.ts | 20 + .../api/resolve-queue-task-outcome.test.ts | 78 + .../src/lib/api/resolve-queue-task-outcome.ts | 42 + packages/vitnode/src/locales/en.json | 48 +- .../routes/admin/core/advanced/queue/page.tsx | 51 + .../src/routes/admin/core/debug/page.tsx | 13 +- .../layouts/sidebar/nav/get-admin-nav.tsx | 9 + .../advanced/queue/badges/status-badge.tsx | 45 + .../core/advanced/queue/queue-table-view.tsx | 111 ++ .../views/core/debug/queue/queue-view.tsx | 91 + .../system/integrations/integrations-view.tsx | 23 +- 41 files changed, 3518 insertions(+), 44 deletions(-) create mode 100644 apps/docs/content/docs/dev/advanced/queue.mdx create mode 100644 apps/docs/migrations/0006_greedy_apocalypse.sql create mode 100644 apps/docs/migrations/meta/0006_snapshot.json create mode 100644 apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/advanced/queue/page.tsx create mode 100644 packages/vitnode/src/api/lib/cache.test.ts create mode 100644 packages/vitnode/src/api/lib/queue.ts create mode 100644 packages/vitnode/src/api/models/email.test.ts create mode 100644 packages/vitnode/src/api/models/queue.ts create mode 100644 packages/vitnode/src/api/modules/admin/advanced/queue/queue.admin.module.ts create mode 100644 packages/vitnode/src/api/modules/admin/advanced/queue/routes/get.route.ts create mode 100644 packages/vitnode/src/api/modules/admin/debug/routes/queue.route.ts create mode 100644 packages/vitnode/src/api/modules/queue/cron/process-queue.cron.ts create mode 100644 packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts create mode 100644 packages/vitnode/src/api/modules/queue/queue.module.ts create mode 100644 packages/vitnode/src/api/modules/queue/tasks/send-email.task.ts create mode 100644 packages/vitnode/src/database/queue.ts create mode 100644 packages/vitnode/src/lib/api/get-queue-backoff-date.test.ts create mode 100644 packages/vitnode/src/lib/api/get-queue-backoff-date.ts create mode 100644 packages/vitnode/src/lib/api/resolve-queue-task-outcome.test.ts create mode 100644 packages/vitnode/src/lib/api/resolve-queue-task-outcome.ts create mode 100644 packages/vitnode/src/routes/admin/core/advanced/queue/page.tsx create mode 100644 packages/vitnode/src/views/admin/views/core/advanced/queue/badges/status-badge.tsx create mode 100644 packages/vitnode/src/views/admin/views/core/advanced/queue/queue-table-view.tsx create mode 100644 packages/vitnode/src/views/admin/views/core/debug/queue/queue-view.tsx diff --git a/apps/api/src/locales/@vitnode/core/en.json b/apps/api/src/locales/@vitnode/core/en.json index 27fef00d9..b591e6eb4 100644 --- a/apps/api/src/locales/@vitnode/core/en.json +++ b/apps/api/src/locales/@vitnode/core/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/apps/docs/content/docs/dev/advanced/queue.mdx b/apps/docs/content/docs/dev/advanced/queue.mdx new file mode 100644 index 000000000..dc3a11399 --- /dev/null +++ b/apps/docs/content/docs/dev/advanced/queue.mdx @@ -0,0 +1,180 @@ +--- +title: Queue Tasks +description: Run one-off background work through a database-backed task queue, drained by cron, with retries and optional Redis coordination. +--- + +Cron jobs handle **recurring** work on a schedule. Queue tasks handle **one-off** +background work you enqueue on demand — "send this email later", "process this +import", "retry this webhook". Tasks are stored in the database (`core_queue`) +and drained every minute by a built-in cron job, so a slow task never blocks the +request that scheduled it. + +Queue tasks build on the [cron system](/docs/dev/cron): the `process-queue` job +runs on the same per-minute tick, so **you need a cron adapter configured** (for +example [Node CRON](/docs/dev/cron/node-cron)) for tasks to be processed +automatically. + +## How it works + +```text +c.get("queue").dispatch({ name, payload }) ─► INSERT core_queue (status = pending) + +cron tick "* * * * *" ─► process-queue job + 1. optional Redis lock — one instance drains per tick when Redis is configured + 2. claim a batch (UPDATE … status = processing … FOR UPDATE SKIP LOCKED) + 3. run each task's handler, matched by pluginId:name + 4. success → completed · error → retry with backoff → failed after maxAttempts + 5. prune completed/failed rows older than 7 days +``` + +Concurrency is safe by default: batches are claimed with Postgres +`FOR UPDATE SKIP LOCKED`, so many instances (or many ticks) can drain the queue +at once without ever running the same task twice. + +## Defining a task + +A task is a named handler, registered like a cron job. The handler receives the +request context and the JSON `payload` the task was dispatched with. + + + +### Create the task file + +```ts title="modules/newsletter/tasks/send-digest.task.ts" +import { buildQueueTask } from "@vitnode/core/api/lib/queue"; + +export const sendDigestTask = buildQueueTask({ + name: "send-digest", + description: "Send the weekly digest email to a user", + // Optional — defaults to 3 + maxAttempts: 5, + handler: async (c, payload) => { + const { userId } = payload as { userId: number }; + + // ...do the work; throwing schedules a retry with backoff + await c.get("log").debug(`Sending digest to user ${userId}`); + }, +}); +``` + + + `payload` is typed as `Record` because it round-trips through + the database as JSON. Narrow it inside the handler (a cast, or validate it with + zod) before use. + + + + +### Register it in a module + +Pass your tasks to `queueTasks` — the same module builder that carries +`cronJobs`. + +```ts title="modules/newsletter/newsletter.module.ts" +import { buildModule } from "@vitnode/core/api/lib/module"; +import { CONFIG_PLUGIN } from "@/config"; + +// [!code ++] +import { sendDigestTask } from "./tasks/send-digest.task"; + +export const newsletterModule = buildModule({ + pluginId: CONFIG_PLUGIN.pluginId, + name: "newsletter", + routes: [], + // [!code ++] + queueTasks: [sendDigestTask], +}); +``` + + + + +## Enqueuing work + +Dispatch a task from any route or model with `c.get("queue")`. The `name` must +match a registered task; the row is picked up on the next cron tick. + +```ts title="modules/newsletter/routes/subscribe.route.ts" +export const subscribeRoute = buildRoute({ + pluginId: CONFIG_PLUGIN.pluginId, + route: {}, + handler: async c => { + await c.get("queue").dispatch({ + name: "send-digest", + payload: { userId: c.get("user")?.id }, + }); + + return c.json({ queued: true }); + }, +}); +``` + +### dispatch options + +import { TypeTable } from "fumadocs-ui/components/type-table"; + +", + default: "{}", + }, + queue: { + description: "Logical queue/channel name, for grouping and filtering.", + type: "string", + default: '"default"', + }, + priority: { + description: "Higher values are claimed first within a batch.", + type: "number", + default: "0", + }, + maxAttempts: { + description: "How many times to try before marking the task failed.", + type: "number", + default: "3", + }, + availableAt: { + description: + "Earliest time the task may run — set it in the future to delay work.", + type: "Date", + default: "now", + }, + }} +/> + +## Retries and backoff + +If a handler throws, the task is retried with an exponential backoff +(`10s`, `20s`, `40s`, … capped at one hour) until `maxAttempts` is reached, after +which its status becomes `failed` and the error is stored in `lastError`. A task +with no registered handler fails immediately. + +Completed and failed rows are pruned automatically after 7 days, so the table +does not grow without bound. + +## Optional: Redis coordination + +The queue works with just Postgres. When you run **multiple instances**, adding +[Redis](/docs/dev/advanced/redis) lets a single instance drain the queue per +tick (an advisory lock), which reduces database contention. Correctness never +depends on it — `FOR UPDATE SKIP LOCKED` already prevents double processing. + +There is nothing extra to enable: set `REDIS_URL` (see the +[Redis guide](/docs/dev/advanced/redis)) and the queue uses it automatically. +Without Redis, every instance simply claims its own disjoint batch. + +## Monitoring in the AdminCP + +- **Core → Advanced → Queue Tasks** — the full, paginated list of tasks with a + status filter, attempt counts, and the last error. Requires the + `queue:can_view` staff permission. +- **Debug Panel** — a live view of currently active (pending/processing) tasks + plus per-status counts. +- **Core → System → Integrations** — a card summarizing registered task handlers + and the number of pending/running tasks. diff --git a/apps/docs/migrations/0006_greedy_apocalypse.sql b/apps/docs/migrations/0006_greedy_apocalypse.sql new file mode 100644 index 000000000..2acb9dc88 --- /dev/null +++ b/apps/docs/migrations/0006_greedy_apocalypse.sql @@ -0,0 +1,20 @@ +CREATE TABLE "core_queue" ( + "id" serial PRIMARY KEY NOT NULL, + "pluginId" varchar(100) NOT NULL, + "name" varchar(100) NOT NULL, + "queue" varchar(100) DEFAULT 'default' NOT NULL, + "status" varchar(20) DEFAULT 'pending' NOT NULL, + "payload" jsonb DEFAULT '{}'::jsonb NOT NULL, + "priority" integer DEFAULT 0 NOT NULL, + "attempts" integer DEFAULT 0 NOT NULL, + "maxAttempts" integer DEFAULT 3 NOT NULL, + "availableAt" timestamp DEFAULT now() NOT NULL, + "reservedAt" timestamp, + "lastError" text, + "createdAt" timestamp DEFAULT now() NOT NULL, + "updatedAt" timestamp NOT NULL, + "completedAt" timestamp +); +--> statement-breakpoint +ALTER TABLE "core_queue" ENABLE ROW LEVEL SECURITY;--> statement-breakpoint +CREATE INDEX "core_queue_status_available_at_idx" ON "core_queue" USING btree ("status","availableAt"); \ No newline at end of file diff --git a/apps/docs/migrations/meta/0006_snapshot.json b/apps/docs/migrations/meta/0006_snapshot.json new file mode 100644 index 000000000..acdd9e113 --- /dev/null +++ b/apps/docs/migrations/meta/0006_snapshot.json @@ -0,0 +1,1761 @@ +{ + "id": "34389650-950c-49e6-8f6f-6ffa68c26fd9", + "prevId": "755b9c3b-3019-488c-b361-448bcdeedb47", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.core_admin_permissions": { + "name": "core_admin_permissions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "roleId": { + "name": "roleId", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "protected": { + "name": "protected", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "data": { + "name": "data", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{\"unrestricted\":false,\"permissions\":[]}'::jsonb" + } + }, + "indexes": { + "core_admin_permissions_role_id_idx": { + "name": "core_admin_permissions_role_id_idx", + "columns": [ + { + "expression": "roleId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_admin_permissions_user_id_idx": { + "name": "core_admin_permissions_user_id_idx", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_admin_permissions_roleId_core_roles_id_fk": { + "name": "core_admin_permissions_roleId_core_roles_id_fk", + "tableFrom": "core_admin_permissions", + "tableTo": "core_roles", + "columnsFrom": [ + "roleId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "core_admin_permissions_userId_core_users_id_fk": { + "name": "core_admin_permissions_userId_core_users_id_fk", + "tableFrom": "core_admin_permissions", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_admin_sessions": { + "name": "core_admin_sessions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "token": { + "name": "token", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "lastSeen": { + "name": "lastSeen", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expiresAt": { + "name": "expiresAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "deviceId": { + "name": "deviceId", + "type": "integer", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "core_admin_sessions_token_idx": { + "name": "core_admin_sessions_token_idx", + "columns": [ + { + "expression": "token", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_admin_sessions_user_id_idx": { + "name": "core_admin_sessions_user_id_idx", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_admin_sessions_userId_core_users_id_fk": { + "name": "core_admin_sessions_userId_core_users_id_fk", + "tableFrom": "core_admin_sessions", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "core_admin_sessions_deviceId_core_sessions_known_devices_id_fk": { + "name": "core_admin_sessions_deviceId_core_sessions_known_devices_id_fk", + "tableFrom": "core_admin_sessions", + "tableTo": "core_sessions_known_devices", + "columnsFrom": [ + "deviceId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_admin_sessions_token_unique": { + "name": "core_admin_sessions_token_unique", + "nullsNotDistinct": false, + "columns": [ + "token" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_cron": { + "name": "core_cron", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "varchar(255)", + "primaryKey": false, + "notNull": false + }, + "lastRun": { + "name": "lastRun", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "pluginId": { + "name": "pluginId", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "module": { + "name": "module", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "nextRun": { + "name": "nextRun", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "schedule": { + "name": "schedule", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_languages": { + "name": "core_languages", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "code": { + "name": "code", + "type": "varchar(32)", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "timezone": { + "name": "timezone", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true, + "default": "'UTC'" + }, + "protected": { + "name": "protected", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "default": { + "name": "default", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "enabled": { + "name": "enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "time24": { + "name": "time24", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + } + }, + "indexes": { + "core_languages_code_idx": { + "name": "core_languages_code_idx", + "columns": [ + { + "expression": "code", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_languages_name_idx": { + "name": "core_languages_name_idx", + "columns": [ + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_languages_code_unique": { + "name": "core_languages_code_unique", + "nullsNotDistinct": false, + "columns": [ + "code" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_languages_words": { + "name": "core_languages_words", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "languageCode": { + "name": "languageCode", + "type": "varchar", + "primaryKey": false, + "notNull": true + }, + "pluginCode": { + "name": "pluginCode", + "type": "varchar(50)", + "primaryKey": false, + "notNull": true + }, + "itemId": { + "name": "itemId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "tableName": { + "name": "tableName", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "variable": { + "name": "variable", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "core_languages_words_lang_code_idx": { + "name": "core_languages_words_lang_code_idx", + "columns": [ + { + "expression": "languageCode", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_languages_words_languageCode_core_languages_code_fk": { + "name": "core_languages_words_languageCode_core_languages_code_fk", + "tableFrom": "core_languages_words", + "tableTo": "core_languages", + "columnsFrom": [ + "languageCode" + ], + "columnsTo": [ + "code" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_logs": { + "name": "core_logs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "pluginId": { + "name": "pluginId", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "type": { + "name": "type", + "type": "varchar(10)", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "ipAddress": { + "name": "ipAddress", + "type": "varchar(45)", + "primaryKey": false, + "notNull": true + }, + "method": { + "name": "method", + "type": "varchar(10)", + "primaryKey": false, + "notNull": true, + "default": "'GET'" + }, + "path": { + "name": "path", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'localhost'" + }, + "userAgent": { + "name": "userAgent", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "statusCode": { + "name": "statusCode", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 500 + }, + "userId": { + "name": "userId", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "test123": { + "name": "test123", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + } + }, + "indexes": {}, + "foreignKeys": { + "core_logs_userId_core_users_id_fk": { + "name": "core_logs_userId_core_users_id_fk", + "tableFrom": "core_logs", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "cascade" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_moderators_permissions": { + "name": "core_moderators_permissions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "roleId": { + "name": "roleId", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "protected": { + "name": "protected", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "data": { + "name": "data", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{\"unrestricted\":false,\"permissions\":[]}'::jsonb" + } + }, + "indexes": { + "core_moderators_permissions_role_id_idx": { + "name": "core_moderators_permissions_role_id_idx", + "columns": [ + { + "expression": "roleId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_moderators_permissions_user_id_idx": { + "name": "core_moderators_permissions_user_id_idx", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_moderators_permissions_roleId_core_roles_id_fk": { + "name": "core_moderators_permissions_roleId_core_roles_id_fk", + "tableFrom": "core_moderators_permissions", + "tableTo": "core_roles", + "columnsFrom": [ + "roleId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "core_moderators_permissions_userId_core_users_id_fk": { + "name": "core_moderators_permissions_userId_core_users_id_fk", + "tableFrom": "core_moderators_permissions", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_queue": { + "name": "core_queue", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "pluginId": { + "name": "pluginId", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "queue": { + "name": "queue", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true, + "default": "'default'" + }, + "status": { + "name": "status", + "type": "varchar(20)", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + }, + "priority": { + "name": "priority", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "attempts": { + "name": "attempts", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "maxAttempts": { + "name": "maxAttempts", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 3 + }, + "availableAt": { + "name": "availableAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "reservedAt": { + "name": "reservedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "lastError": { + "name": "lastError", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "completedAt": { + "name": "completedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "core_queue_status_available_at_idx": { + "name": "core_queue_status_available_at_idx", + "columns": [ + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "availableAt", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_roles": { + "name": "core_roles", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "protected": { + "name": "protected", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "default": { + "name": "default", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "root": { + "name": "root", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "guest": { + "name": "guest", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "color": { + "name": "color", + "type": "varchar(19)", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_sessions": { + "name": "core_sessions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "token": { + "name": "token", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expiresAt": { + "name": "expiresAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "deviceId": { + "name": "deviceId", + "type": "integer", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "core_sessions_user_id_idx": { + "name": "core_sessions_user_id_idx", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_sessions_userId_core_users_id_fk": { + "name": "core_sessions_userId_core_users_id_fk", + "tableFrom": "core_sessions", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "core_sessions_deviceId_core_sessions_known_devices_id_fk": { + "name": "core_sessions_deviceId_core_sessions_known_devices_id_fk", + "tableFrom": "core_sessions", + "tableTo": "core_sessions_known_devices", + "columnsFrom": [ + "deviceId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_sessions_token_unique": { + "name": "core_sessions_token_unique", + "nullsNotDistinct": false, + "columns": [ + "token" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_sessions_known_devices": { + "name": "core_sessions_known_devices", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "publicId": { + "name": "publicId", + "type": "varchar(32)", + "primaryKey": false, + "notNull": true + }, + "ipAddress": { + "name": "ipAddress", + "type": "varchar(40)", + "primaryKey": false, + "notNull": true + }, + "userAgent": { + "name": "userAgent", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "lastSeen": { + "name": "lastSeen", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "core_sessions_known_devices_ip_address_idx": { + "name": "core_sessions_known_devices_ip_address_idx", + "columns": [ + { + "expression": "ipAddress", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_sessions_known_devices_publicId_unique": { + "name": "core_sessions_known_devices_publicId_unique", + "nullsNotDistinct": false, + "columns": [ + "publicId" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_users": { + "name": "core_users", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "nameCode": { + "name": "nameCode", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "email": { + "name": "email", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "password": { + "name": "password", + "type": "varchar", + "primaryKey": false, + "notNull": false + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "newsletter": { + "name": "newsletter", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "avatarColor": { + "name": "avatarColor", + "type": "varchar(6)", + "primaryKey": false, + "notNull": true + }, + "emailVerified": { + "name": "emailVerified", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "roleId": { + "name": "roleId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "birthday": { + "name": "birthday", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "ipAddress": { + "name": "ipAddress", + "type": "varchar(40)", + "primaryKey": false, + "notNull": true + }, + "language": { + "name": "language", + "type": "varchar(32)", + "primaryKey": false, + "notNull": true, + "default": "'en'" + } + }, + "indexes": { + "core_users_name_code_idx": { + "name": "core_users_name_code_idx", + "columns": [ + { + "expression": "nameCode", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_users_name_idx": { + "name": "core_users_name_idx", + "columns": [ + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_users_email_idx": { + "name": "core_users_email_idx", + "columns": [ + { + "expression": "email", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_users_roleId_core_roles_id_fk": { + "name": "core_users_roleId_core_roles_id_fk", + "tableFrom": "core_users", + "tableTo": "core_roles", + "columnsFrom": [ + "roleId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "core_users_language_core_languages_code_fk": { + "name": "core_users_language_core_languages_code_fk", + "tableFrom": "core_users", + "tableTo": "core_languages", + "columnsFrom": [ + "language" + ], + "columnsTo": [ + "code" + ], + "onDelete": "set default", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_users_nameCode_unique": { + "name": "core_users_nameCode_unique", + "nullsNotDistinct": false, + "columns": [ + "nameCode" + ] + }, + "core_users_name_unique": { + "name": "core_users_name_unique", + "nullsNotDistinct": false, + "columns": [ + "name" + ] + }, + "core_users_email_unique": { + "name": "core_users_email_unique", + "nullsNotDistinct": false, + "columns": [ + "email" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_users_confirm_emails": { + "name": "core_users_confirm_emails", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "token": { + "name": "token", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expiresAt": { + "name": "expiresAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "ipAddress": { + "name": "ipAddress", + "type": "varchar(40)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "core_users_confirm_emails_userId_core_users_id_fk": { + "name": "core_users_confirm_emails_userId_core_users_id_fk", + "tableFrom": "core_users_confirm_emails", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_users_confirm_emails_token_unique": { + "name": "core_users_confirm_emails_token_unique", + "nullsNotDistinct": false, + "columns": [ + "token" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_users_forgot_password": { + "name": "core_users_forgot_password", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "token": { + "name": "token", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "ipAddress": { + "name": "ipAddress", + "type": "varchar(40)", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expiresAt": { + "name": "expiresAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "core_users_forgot_password_userId_core_users_id_fk": { + "name": "core_users_forgot_password_userId_core_users_id_fk", + "tableFrom": "core_users_forgot_password", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "core_users_forgot_password_userId_unique": { + "name": "core_users_forgot_password_userId_unique", + "nullsNotDistinct": false, + "columns": [ + "userId" + ] + }, + "core_users_forgot_password_token_unique": { + "name": "core_users_forgot_password_token_unique", + "nullsNotDistinct": false, + "columns": [ + "token" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_users_secondary_roles": { + "name": "core_users_secondary_roles", + "schema": "", + "columns": { + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "roleId": { + "name": "roleId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "core_users_secondary_roles_user_id_idx": { + "name": "core_users_secondary_roles_user_id_idx", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "core_users_secondary_roles_role_id_idx": { + "name": "core_users_secondary_roles_role_id_idx", + "columns": [ + { + "expression": "roleId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_users_secondary_roles_userId_core_users_id_fk": { + "name": "core_users_secondary_roles_userId_core_users_id_fk", + "tableFrom": "core_users_secondary_roles", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "core_users_secondary_roles_roleId_core_roles_id_fk": { + "name": "core_users_secondary_roles_roleId_core_roles_id_fk", + "tableFrom": "core_users_secondary_roles", + "tableTo": "core_roles", + "columnsFrom": [ + "roleId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "core_users_secondary_roles_userId_roleId_pk": { + "name": "core_users_secondary_roles_userId_roleId_pk", + "columns": [ + "userId", + "roleId" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.core_users_sso": { + "name": "core_users_sso", + "schema": "", + "columns": { + "userId": { + "name": "userId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "providerId": { + "name": "providerId", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "providerAccountId": { + "name": "providerAccountId", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "core_users_sso_user_id_idx": { + "name": "core_users_sso_user_id_idx", + "columns": [ + { + "expression": "userId", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "core_users_sso_userId_core_users_id_fk": { + "name": "core_users_sso_userId_core_users_id_fk", + "tableFrom": "core_users_sso", + "tableTo": "core_users", + "columnsFrom": [ + "userId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.blog_categories": { + "name": "blog_categories", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "title": { + "name": "title", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "titleSeo": { + "name": "titleSeo", + "type": "varchar(100)", + "primaryKey": false, + "notNull": true, + "default": "''" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "blog_categories_titleSeo_unique": { + "name": "blog_categories_titleSeo_unique", + "nullsNotDistinct": false, + "columns": [ + "titleSeo" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + }, + "public.blog_posts": { + "name": "blog_posts", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "serial", + "primaryKey": true, + "notNull": true + }, + "title": { + "name": "title", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "titleSeo": { + "name": "titleSeo", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "content": { + "name": "content", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "categoryId": { + "name": "categoryId", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "createdAt": { + "name": "createdAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updatedAt": { + "name": "updatedAt", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "blog_posts_categoryId_blog_categories_id_fk": { + "name": "blog_posts_categoryId_blog_categories_id_fk", + "tableFrom": "blog_posts", + "tableTo": "blog_categories", + "columnsFrom": [ + "categoryId" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "blog_posts_titleSeo_unique": { + "name": "blog_posts_titleSeo_unique", + "nullsNotDistinct": false, + "columns": [ + "titleSeo" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": true + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/apps/docs/migrations/meta/_journal.json b/apps/docs/migrations/meta/_journal.json index e61d4ed6a..68e489a74 100644 --- a/apps/docs/migrations/meta/_journal.json +++ b/apps/docs/migrations/meta/_journal.json @@ -43,6 +43,13 @@ "when": 1782638542375, "tag": "0005_stale_firebird", "breakpoints": true + }, + { + "idx": 6, + "version": "7", + "when": 1783191895889, + "tag": "0006_greedy_apocalypse", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/advanced/queue/page.tsx b/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/advanced/queue/page.tsx new file mode 100644 index 000000000..c9bc0b9d4 --- /dev/null +++ b/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/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 "@vitnode/core/components/i18n-provider"; +import { DataTableSkeleton } from "@vitnode/core/components/table/data-table"; +import { HeaderContent } from "@vitnode/core/components/ui/header-content"; +import { checkAdminPermissionApi } from "@vitnode/core/lib/api/get-session-admin-api"; + +const QueueTableView = dynamic(async () => + import("@vitnode/core/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/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/debug/page.tsx b/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/debug/page.tsx index 76bd11f1a..644205650 100644 --- a/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/debug/page.tsx +++ b/apps/docs/src/app/[locale]/admin/(auth)/(plugins)/(vitnode-core)/core/debug/page.tsx @@ -17,6 +17,12 @@ const SystemLogsView = dynamic(async () => ), ); +const QueueView = dynamic(async () => + import("@vitnode/core/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/apps/docs/src/locales/@vitnode/core/en.json b/apps/docs/src/locales/@vitnode/core/en.json index 27fef00d9..b591e6eb4 100644 --- a/apps/docs/src/locales/@vitnode/core/en.json +++ b/apps/docs/src/locales/@vitnode/core/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/api/lib/cache.test.ts b/packages/vitnode/src/api/lib/cache.test.ts new file mode 100644 index 000000000..e57cf4ed1 --- /dev/null +++ b/packages/vitnode/src/api/lib/cache.test.ts @@ -0,0 +1,55 @@ +import type { Context } from "hono"; +import type { Redis } from "ioredis"; + +import { describe, expect, it, vi } from "vitest"; + +import { CacheModel } from "./cache"; + +const fakeContext = { get: () => undefined } as unknown as Context; +const LOCK_KEY = "vitnode:cache:__system__:lock:queue:process"; + +describe("CacheModel locks", () => { + describe("without Redis", () => { + const cache = new CacheModel(null, fakeContext); + + it("acquireLock returns true so cache-less deployments still proceed", async () => { + await expect(cache.acquireLock("queue:process", 55)).resolves.toBe(true); + }); + + it("releaseLock is a no-op", async () => { + await expect(cache.releaseLock("queue:process")).resolves.toBeUndefined(); + }); + }); + + describe("with Redis", () => { + it("acquires the lock with SET NX EX in the system namespace", async () => { + const set = vi.fn().mockResolvedValue("OK"); + const cache = new CacheModel({ set } as unknown as Redis, fakeContext); + + await expect(cache.acquireLock("queue:process", 55)).resolves.toBe(true); + expect(set).toHaveBeenCalledWith(LOCK_KEY, "1", "EX", 55, "NX"); + }); + + it("returns false when the lock is already held", async () => { + const set = vi.fn().mockResolvedValue(null); + const cache = new CacheModel({ set } as unknown as Redis, fakeContext); + + await expect(cache.acquireLock("queue:process", 55)).resolves.toBe(false); + }); + + it("returns false when Redis errors, to skip rather than double-run", async () => { + const set = vi.fn().mockRejectedValue(new Error("down")); + const cache = new CacheModel({ set } as unknown as Redis, fakeContext); + + await expect(cache.acquireLock("queue:process", 55)).resolves.toBe(false); + }); + + it("releaseLock deletes the lock key", async () => { + const del = vi.fn().mockResolvedValue(1); + const cache = new CacheModel({ del } as unknown as Redis, fakeContext); + + await cache.releaseLock("queue:process"); + expect(del).toHaveBeenCalledWith(LOCK_KEY); + }); + }); +}); diff --git a/packages/vitnode/src/api/lib/cache.ts b/packages/vitnode/src/api/lib/cache.ts index f49bb8864..72a104816 100644 --- a/packages/vitnode/src/api/lib/cache.ts +++ b/packages/vitnode/src/api/lib/cache.ts @@ -90,6 +90,32 @@ export class CacheModel { } } + /** + * Acquire a short-lived distributed lock (`SET key val NX EX ttl`) in the + * system namespace. Returns `true` when the lock is held by this caller. + * **Without Redis it returns `true`** so cache-less / single-instance + * deployments still make progress — callers must guard correctness some other + * way (e.g. Postgres `FOR UPDATE SKIP LOCKED`). Returns `false` on a Redis + * error so a flaky connection skips rather than double-runs work. + */ + async acquireLock(key: string, ttlSeconds: number): Promise { + if (!this.client) return true; + + try { + const result = await this.client.set( + this.systemKey(`lock:${key}`), + "1", + "EX", + ttlSeconds, + "NX", + ); + + return result === "OK"; + } catch { + return false; + } + } + /** Remove one or more keys. No-op without Redis. */ async delete(key: string | string[]): Promise { const keys = (Array.isArray(key) ? key : [key]).map(k => this.key(k)); @@ -148,6 +174,11 @@ export class CacheModel { } } + /** Release a lock taken with {@link acquireLock}. No-op without Redis. */ + async releaseLock(key: string): Promise { + await this.removeKeys([this.systemKey(`lock:${key}`)]); + } + async remember( 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.ts b/packages/vitnode/src/api/models/queue.ts new file mode 100644 index 000000000..bea7513f3 --- /dev/null +++ b/packages/vitnode/src/api/models/queue.ts @@ -0,0 +1,53 @@ +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 = 3, + availableAt, + }: QueueDispatchArgs): Promise<{ id: number }> { + const pluginId = this.c.get("plugin")?.id ?? "@vitnode/core"; + + const [row] = await this.c + .get("db") + .insert(core_queue) + .values({ + pluginId, + name, + queue, + payload, + priority, + maxAttempts, + 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..f2c9a45ea --- /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: outcome.status === "pending" ? null : task.reservedAt, + }) + .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/queue.ts b/packages/vitnode/src/database/queue.ts new file mode 100644 index 000000000..a2d62dbec --- /dev/null +++ b/packages/vitnode/src/database/queue.ts @@ -0,0 +1,34 @@ +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() + .$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/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}

+
+ ), + }, + { id: "queue", label: t("list.queue") }, + { + id: "status", + label: t("list.status"), + cell: ({ row }) => , + }, + { + id: "attempts", + label: t("list.attempts"), + cell: ({ row }) => `${row.attempts}/${row.maxAttempts}`, + }, + { + id: "availableAt", + label: t("list.availableAt"), + cell: ({ row }) => , + }, + { + id: "createdAt", + label: t("list.createdAt"), + cell: ({ row }) => , + }, + { + id: "lastError", + label: t("list.lastError"), + cell: ({ row }) => + row.lastError ? ( + + {row.lastError} + + ) : ( + — + ), + }, + ]} + edges={data.edges} + filters={[ + { + id: "status", + label: t("list.statusFilter"), + options: QUEUE_STATUSES.map(status => ({ + value: status, + label: t(`status.${status}`), + })), + }, + ]} + id="queue-table" + order={{ + columns: ["createdAt", "availableAt", "status"], + defaultOrder: { + column: "createdAt", + order: "desc", + }, + }} + pageInfo={data.pageInfo} + /> + ); +}; diff --git a/packages/vitnode/src/views/admin/views/core/debug/queue/queue-view.tsx b/packages/vitnode/src/views/admin/views/core/debug/queue/queue-view.tsx new file mode 100644 index 000000000..4a2905c76 --- /dev/null +++ b/packages/vitnode/src/views/admin/views/core/debug/queue/queue-view.tsx @@ -0,0 +1,91 @@ +import { getTranslations } from "next-intl/server"; + +import { debugAdminModule } from "@/api/modules/admin/debug/debug.admin.module"; +import { DateFormat } from "@/components/date-format"; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from "@/components/ui/table"; +import { fetcher } from "@/lib/fetcher"; +import { QueueStatusBadge } from "@/views/admin/views/core/advanced/queue/badges/status-badge"; + +const COUNT_STATUSES = [ + "pending", + "processing", + "completed", + "failed", +] as const; + +export const QueueView = async () => { + const [t, res] = await Promise.all([ + getTranslations("admin.debug.queue"), + fetcher(debugAdminModule, { + prefixPath: "/admin", + path: "/queue", + method: "get", + module: "debug", + }), + ]); + const data = await res.json(); + + return ( +
+
+ {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} + + + + + + ))} + +
+
+ )} +
+ ); +}; diff --git a/packages/vitnode/src/views/admin/views/core/system/integrations/integrations-view.tsx b/packages/vitnode/src/views/admin/views/core/system/integrations/integrations-view.tsx index 7412a577a..45931c2d6 100644 --- a/packages/vitnode/src/views/admin/views/core/system/integrations/integrations-view.tsx +++ b/packages/vitnode/src/views/admin/views/core/system/integrations/integrations-view.tsx @@ -1,6 +1,7 @@ import { ClockIcon, DatabaseIcon, + ListTodoIcon, MailIcon, RadioTowerIcon, ShieldCheckIcon, @@ -19,6 +20,7 @@ const DOCS_URLS = { captcha: "https://vitnode.com/docs/dev/captcha", cron: "https://vitnode.com/docs/dev/cron", email: "https://vitnode.com/docs/dev/email", + queue: "https://vitnode.com/docs/dev/advanced/queue", redis: "https://vitnode.com/docs/dev/advanced/redis", websocket: "https://vitnode.com/docs/dev/websocket", }; @@ -128,6 +130,25 @@ export const IntegrationsView = async () => { title={t("cron.title")} /> + + {t("queue.tasks", { count: data.queue.tasks })} ·{" "} + {t("queue.queued", { + pending: data.queue.pending, + processing: data.queue.processing, + })} + + } + readMoreLabel={t("read_more")} + status={toStatus(data.queue.active)} + statusLabel={statusLabel(toStatus(data.queue.active))} + title={t("queue.title")} + /> + { export const IntegrationsViewSkeleton = () => (
- {["websocket", "redis", "email", "cron", "captcha"].map(id => ( + {["websocket", "redis", "email", "cron", "queue", "captcha"].map(id => ( ))}
From 90140cd37c7a9cd4e61b08fe1bda68531d40bc7e Mon Sep 17 00:00:00 2001 From: aXenDeveloper Date: Sat, 4 Jul 2026 21:47:30 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat(queue):=20=E2=9C=A8=20Enhance=20queue?= =?UTF-8?q?=20dispatch=20with=20dynamic=20maxAttempts=20handling?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/vitnode/scripts/shared/file-utils.ts | 4 +- packages/vitnode/src/api/models/queue.test.ts | 86 +++++++++++++++++++ packages/vitnode/src/api/models/queue.ts | 8 +- .../queue/helpers/process-queue-tasks.ts | 2 +- packages/vitnode/src/database/admins.ts | 1 + packages/vitnode/src/database/languages.ts | 1 + packages/vitnode/src/database/moderators.ts | 1 + packages/vitnode/src/database/queue.ts | 1 + packages/vitnode/src/database/users.ts | 1 + 9 files changed, 101 insertions(+), 4 deletions(-) create mode 100644 packages/vitnode/src/api/models/queue.test.ts diff --git a/packages/vitnode/scripts/shared/file-utils.ts b/packages/vitnode/scripts/shared/file-utils.ts index f68fd9fa4..a88763eaa 100644 --- a/packages/vitnode/scripts/shared/file-utils.ts +++ b/packages/vitnode/scripts/shared/file-utils.ts @@ -340,7 +340,9 @@ export const cleanupDeletedFiles = ( if (!existsSync(sourceDir)) return; if (!existsSync(destinationDir)) return; - const isLocaleDir = destinationDir.includes(join("src", "locales")); + const isLocaleDir = destinationDir + .replace(/\\/g, "/") + .includes("src/locales"); if (isLocaleDir) return; const destFiles = getAllFiles(destinationDir); 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 index bea7513f3..38f66309f 100644 --- a/packages/vitnode/src/api/models/queue.ts +++ b/packages/vitnode/src/api/models/queue.ts @@ -29,11 +29,15 @@ export class QueueModel { payload = {}, queue = "default", priority = 0, - maxAttempts = 3, + 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) @@ -43,7 +47,7 @@ export class QueueModel { queue, payload, priority, - maxAttempts, + maxAttempts: maxAttempts ?? registeredTask?.maxAttempts ?? 3, availableAt: availableAt ?? new Date(), }) .returning({ id: core_queue.id }); 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 index f2c9a45ea..2bd9fca7e 100644 --- a/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts +++ b/packages/vitnode/src/api/modules/queue/helpers/process-queue-tasks.ts @@ -99,7 +99,7 @@ export const processQueueTasks = async ( lastError: outcome.lastError, availableAt: outcome.availableAt ?? task.availableAt, completedAt: outcome.completedAt ?? null, - reservedAt: outcome.status === "pending" ? null : task.reservedAt, + reservedAt: null, }) .where(eq(core_queue.id, task.id)); } 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 index a2d62dbec..f1a440c5d 100644 --- a/packages/vitnode/src/database/queue.ts +++ b/packages/vitnode/src/database/queue.ts @@ -25,6 +25,7 @@ export const core_queue = pgTable( updatedAt: t .timestamp() .notNull() + .defaultNow() .$onUpdate(() => new Date()), completedAt: t.timestamp(), }), 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)],