danvics/openreader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(tasks): introduce scheduled task engine and admin UI for background jobs

Browse source 
Add a general-purpose scheduled task system with a persistent registry and status tracking, supporting background maintenance jobs such as orphaned blob reaping, expired upload cleanup, job event pruning, and TTS usage retention. Implement a new `scheduled_tasks` table, task engine, and handlers for each maintenance operation. Integrate an admin UI panel for monitoring, manual runs, and configuration of tasks. Update document and user data cleanup flows to delegate shared blob and preview deletion to the scheduled reaper. Add Vercel cron integration for serverless environments.

BREAKING CHANGE: Document and user storage cleanup now relies on background scheduled tasks for shared blob and preview deletion; immediate inline deletion is no longer performed.
This commit is contained in:
Richard R 2026-06-07 05:29:29 -06:00
parent 16fb5dd09c
commit 16cada6129
37 changed files with 5056 additions and 510 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
drizzle/postgres/0011_scheduled-tasks.sql Normal file
View file

@ -0,0 +1,14 @@
CREATE TABLE "scheduled_tasks" (
"key" text PRIMARY KEY NOT NULL,
"enabled" boolean DEFAULT true NOT NULL,
"interval_ms" bigint NOT NULL,
"last_status" text DEFAULT 'idle' NOT NULL,
"last_run_at" bigint,
"last_duration_ms" bigint,
"last_error" text,
"last_result_json" text,
"next_run_at" bigint,
"run_requested" boolean DEFAULT false NOT NULL,
"running_since" bigint,
"updated_at" bigint DEFAULT (extract(epoch from now()) * 1000)::bigint NOT NULL
);

1997
drizzle/postgres/meta/0011_snapshot.json Normal file
View file

File diff suppressed because it is too large Load diff

7
drizzle/postgres/meta/_journal.json
View file

@ -78,6 +78,13 @@
"when": 1780794520924,
"tag": "0010_user-data-cleanup-cascades",
"breakpoints": true
},
{
"idx": 11,
"version": "7",
"when": 1780825886154,
"tag": "0011_scheduled-tasks",
"breakpoints": true
}
]
}

14
drizzle/sqlite/0011_scheduled-tasks.sql Normal file
View file

@ -0,0 +1,14 @@
CREATE TABLE `scheduled_tasks` (
`key` text PRIMARY KEY NOT NULL,
`enabled` integer DEFAULT true NOT NULL,
`interval_ms` integer NOT NULL,
`last_status` text DEFAULT 'idle' NOT NULL,
`last_run_at` integer,
`last_duration_ms` integer,
`last_error` text,
`last_result_json` text,
`next_run_at` integer,
`run_requested` integer DEFAULT false NOT NULL,
`running_since` integer,
`updated_at` integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL
);

1849
drizzle/sqlite/meta/0011_snapshot.json Normal file
View file

File diff suppressed because it is too large Load diff

7
drizzle/sqlite/meta/_journal.json
View file

@ -78,6 +78,13 @@
"when": 1780794515094,
"tag": "0010_user-data-cleanup-cascades",
"breakpoints": true
},
{
"idx": 11,
"version": "6",
"when": 1780825885684,
"tag": "0011_scheduled-tasks",
"breakpoints": true
}
]
}

39
src/app/api/admin/tasks/[key]/route.ts Normal file
View file

@ -0,0 +1,39 @@
import { NextRequest, NextResponse } from 'next/server';
import { requireAdminContext } from '@/lib/server/auth/admin';
import { updateTask } from '@/lib/server/tasks/engine';
import { TASK_REGISTRY } from '@/lib/server/tasks/registry';
export const dynamic = 'force-dynamic';
export async function PATCH(
req: NextRequest,
{ params }: { params: Promise<{ key: string }> },
): Promise<Response> {
const ctx = await requireAdminContext(req);
if (ctx instanceof Response) return ctx;
const { key } = await params;
if (!(key in TASK_REGISTRY)) {
return NextResponse.json({ error: 'Unknown task' }, { status: 404 });
}
let body: unknown;
try {
body = await req.json();
} catch {
return NextResponse.json({ error: 'Invalid JSON' }, { status: 400 });
}
if (!body || typeof body !== 'object') {
return NextResponse.json({ error: 'Expected JSON object' }, { status: 400 });
}
const { enabled, intervalMs } = body as { enabled?: unknown; intervalMs?: unknown };
const patch: { enabled?: boolean; intervalMs?: number } = {};
if (typeof enabled === 'boolean') patch.enabled = enabled;
if (typeof intervalMs === 'number' && Number.isFinite(intervalMs) && intervalMs > 0) {
patch.intervalMs = Math.floor(intervalMs);
}
await updateTask(key, patch);
return NextResponse.json({ ok: true });
}

22
src/app/api/admin/tasks/[key]/run/route.ts Normal file
View file

@ -0,0 +1,22 @@
import { NextRequest, NextResponse } from 'next/server';
import { requireAdminContext } from '@/lib/server/auth/admin';
import { runTaskNow } from '@/lib/server/tasks/engine';
import { TASK_REGISTRY } from '@/lib/server/tasks/registry';
export const dynamic = 'force-dynamic';
export async function POST(
req: NextRequest,
{ params }: { params: Promise<{ key: string }> },
): Promise<Response> {
const ctx = await requireAdminContext(req);
if (ctx instanceof Response) return ctx;
const { key } = await params;
if (!(key in TASK_REGISTRY)) {
return NextResponse.json({ error: 'Unknown task' }, { status: 404 });
}
const ran = await runTaskNow(key);
return NextResponse.json({ ran });
}

13
src/app/api/admin/tasks/route.ts Normal file
View file

@ -0,0 +1,13 @@
import { NextRequest, NextResponse } from 'next/server';
import { requireAdminContext } from '@/lib/server/auth/admin';
import { listTasks } from '@/lib/server/tasks/engine';
export const dynamic = 'force-dynamic';
export async function GET(req: NextRequest): Promise<Response> {
const ctx = await requireAdminContext(req);
if (ctx instanceof Response) return ctx;
const tasks = await listTasks();
return NextResponse.json({ tasks });
}

23
src/app/api/admin/tasks/tick/route.ts Normal file
View file

@ -0,0 +1,23 @@
import { NextRequest, NextResponse } from 'next/server';
import { runDueTasks } from '@/lib/server/tasks/engine';
export const dynamic = 'force-dynamic';
/**
* Cron-driven tick for serverless (Vercel) deployments. Vercel automatically
* sends `Authorization: Bearer ${CRON_SECRET}` on scheduled invocations, so we
* require that secret. Self-hosted deployments drive ticks via the in-process
* scheduler and don't need this route.
*/
export async function GET(req: NextRequest): Promise<Response> {
const secret = process.env.CRON_SECRET;
if (!secret) {
return NextResponse.json({ error: 'Cron not configured' }, { status: 503 });
}
if (req.headers.get('authorization') !== `Bearer ${secret}`) {
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
}
await runDueTasks();
return NextResponse.json({ ok: true });
}

83
src/app/api/documents/blob/upload/finalize/route.ts
View file

@ -24,7 +24,6 @@ import { errorToLog, serverLogger } from '@/lib/server/logger';
import { getOpenReaderTestNamespace } from '@/lib/server/testing/test-namespace';
import { isS3Configured } from '@/lib/server/storage/s3';
import type { BaseDocument, DocumentType } from '@/types/documents';
import { withDocumentLock } from '@/lib/server/documents/document-lock';
export const dynamic = 'force-dynamic';
@ -150,49 +149,51 @@ async function finalizeOne(input: {
: input.upload.name;
const documentId = createHash('sha256').update(finalizedBody).digest('hex');
const stored = await withDocumentLock(documentId, async () => {
try {
await headDocumentBlob(documentId, input.namespace);
} catch (error) {
if (!isMissingBlobError(error)) throw error;
if (!isDocxUpload) {
try {
await copyTempDocumentBlobToDocument(
input.upload.token,
input.userId,
documentId,
input.namespace,
finalizedContentType,
{ ifNoneMatch: true },
);
} catch (copyError) {
if (!isPreconditionFailed(copyError)) throw copyError;
}
} else {
try {
await putDocumentBlob(
documentId,
finalizedBody,
finalizedContentType,
input.namespace,
{ ifNoneMatch: true },
);
} catch (putError) {
if (!isPreconditionFailed(putError)) throw putError;
}
// Ensure the content-addressed blob exists. This is idempotent and race-safe
// on its own (ifNoneMatch + PreconditionFailed tolerates a concurrent finalize
// of the same content), so no lock is needed. The reaper's grace window
// protects a freshly written blob from being reaped before its row commits.
try {
await headDocumentBlob(documentId, input.namespace);
} catch (error) {
if (!isMissingBlobError(error)) throw error;
if (!isDocxUpload) {
try {
await copyTempDocumentBlobToDocument(
input.upload.token,
input.userId,
documentId,
input.namespace,
finalizedContentType,
{ ifNoneMatch: true },
);
} catch (copyError) {
if (!isPreconditionFailed(copyError)) throw copyError;
}
} else {
try {
await putDocumentBlob(
documentId,
finalizedBody,
finalizedContentType,
input.namespace,
{ ifNoneMatch: true },
);
} catch (putError) {
if (!isPreconditionFailed(putError)) throw putError;
}
}
}
const canonicalHead = await headDocumentBlob(documentId, input.namespace);
return registerUploadedDocument({
documentId,
userId: input.userId,
namespace: input.namespace,
name: finalizedName,
type: finalizedType,
size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : finalizedBody.byteLength,
lastModified: input.upload.lastModified,
});
const canonicalHead = await headDocumentBlob(documentId, input.namespace);
const stored = await registerUploadedDocument({
documentId,
userId: input.userId,
namespace: input.namespace,
name: finalizedName,
type: finalizedType,
size: canonicalHead.contentLength > 0 ? canonicalHead.contentLength : finalizedBody.byteLength,
lastModified: input.upload.lastModified,
});
await putTempDocumentFinalizeReceipt(

5
src/app/api/documents/blob/upload/presign/route.ts
View file

@ -2,8 +2,6 @@ import { randomUUID } from 'node:crypto';
import { NextRequest, NextResponse } from 'next/server';
import { requireAuthContext } from '@/lib/server/auth/auth';
import {
TEMP_DOCUMENT_UPLOAD_TTL_MS,
deleteExpiredTempDocumentUploads,
presignTempPut,
} from '@/lib/server/documents/blobstore';
import { getResolvedRuntimeConfig } from '@/lib/server/runtime-config';
@ -72,8 +70,7 @@ export async function POST(req: NextRequest) {
}
const namespace = getOpenReaderTestNamespace(req.headers);
await deleteExpiredTempDocumentUploads(userId, namespace, Date.now() - TEMP_DOCUMENT_UPLOAD_TTL_MS)
.catch(() => undefined);
// Expired temp uploads are swept by the cleanup-temp-uploads scheduled task.
const signed = await Promise.all(
uploads.map(async (upload) => {
const token = randomUUID();

7
src/components/SettingsModal.tsx
View file

@ -36,6 +36,7 @@ import {
import { useRuntimeConfig } from '@/contexts/RuntimeConfigContext';
import { AdminProvidersPanel } from '@/components/admin/AdminProvidersPanel';
import { AdminFeaturesPanel } from '@/components/admin/AdminFeaturesPanel';
import { AdminTasksPanel } from '@/components/admin/AdminTasksPanel';
import { useSharedProviders } from '@/hooks/useSharedProviders';
import { flushUserPreferencesSync } from '@/lib/client/api/user-state';
import toast from 'react-hot-toast';
@ -169,7 +170,7 @@ const SIDEBAR_SECTIONS: SidebarSection[] = [
{ id: 'admin', label: 'Admin', icon: SettingsIcon, authOnly: true, adminOnly: true },
];
type AdminSubTab = 'providers' | 'features';
type AdminSubTab = 'providers' | 'features' | 'tasks';
export function SettingsTrigger({
className = '',
@ -1054,13 +1055,15 @@ export function SettingsModal({
options={[
{ value: 'providers', label: 'Shared providers' },
{ value: 'features', label: 'Site features' },
{ value: 'tasks', label: 'Tasks' },
]}
onChange={setAdminSubTab}
ariaLabel="Admin tab"
className="grid-cols-2"
className="grid-cols-3"
/>
{adminSubTab === 'providers' && <AdminProvidersPanel />}
{adminSubTab === 'features' && <AdminFeaturesPanel />}
{adminSubTab === 'tasks' && <AdminTasksPanel />}
</div>
)}

209
src/components/admin/AdminTasksPanel.tsx Normal file
View file

@ -0,0 +1,209 @@
'use client';
import { useEffect, useState } from 'react';
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
import toast from 'react-hot-toast';
import { Badge, Button, Input, Section, Switch } from '@/components/ui';
import type { BadgeTone } from '@/components/ui/badge';
type TaskRunStatus = 'idle' | 'running' | 'ok' | 'error';
interface TaskView {
key: string;
name: string;
description?: string;
enabled: boolean;
intervalMs: number;
lastStatus: TaskRunStatus;
lastRunAt: number | null;
lastDurationMs: number | null;
lastError: string | null;
lastResult: string | null;
nextRunAt: number | null;
running: boolean;
}
const TASKS_QUERY_KEY = ['admin-tasks'] as const;
async function fetchTasks(): Promise<TaskView[]> {
const res = await fetch('/api/admin/tasks');
if (!res.ok) throw new Error(`HTTP ${res.status}`);
return ((await res.json()) as { tasks: TaskView[] }).tasks;
}
const STATUS_TONE: Record<TaskRunStatus, BadgeTone> = {
idle: 'muted',
running: 'accent',
ok: 'foreground',
error: 'danger',
};
function formatRelative(ms: number | null): string {
if (ms == null) return 'never';
const diff = ms - Date.now();
const abs = Math.abs(diff);
const units: Array<[number, string]> = [
[86_400_000, 'd'],
[3_600_000, 'h'],
[60_000, 'm'],
[1_000, 's'],
];
let label = 'now';
for (const [unitMs, suffix] of units) {
if (abs >= unitMs) {
label = `${Math.round(abs / unitMs)}${suffix}`;
break;
}
}
if (label === 'now') return 'just now';
return diff < 0 ? `${label} ago` : `in ${label}`;
}
function formatDuration(ms: number | null): string {
if (ms == null) return '—';
if (ms < 1000) return `${ms}ms`;
return `${(ms / 1000).toFixed(1)}s`;
}
export function AdminTasksPanel() {
const queryClient = useQueryClient();
const { data, error } = useQuery({
queryKey: TASKS_QUERY_KEY,
queryFn: fetchTasks,
refetchInterval: 5000,
});
useEffect(() => {
if (!error) return;
console.error('[AdminTasksPanel] load failed:', error);
toast.error('Failed to load tasks');
}, [error]);
const invalidate = () => queryClient.invalidateQueries({ queryKey: TASKS_QUERY_KEY });
const patchTask = useMutation({
mutationFn: async ({ key, patch }: { key: string; patch: { enabled?: boolean; intervalMs?: number } }) => {
const res = await fetch(`/api/admin/tasks/${encodeURIComponent(key)}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(patch),
});
if (!res.ok) throw new Error(`HTTP ${res.status}`);
},
onSuccess: invalidate,
onError: () => toast.error('Update failed'),
});
const runTask = useMutation({
mutationFn: async (key: string) => {
const res = await fetch(`/api/admin/tasks/${encodeURIComponent(key)}/run`, { method: 'POST' });
if (!res.ok) throw new Error(`HTTP ${res.status}`);
return ((await res.json()) as { ran: boolean }).ran;
},
onSuccess: (ran) => {
toast.success(ran ? 'Task ran' : 'Task already running');
invalidate();
},
onError: () => toast.error('Run failed'),
});
return (
<Section title="Scheduled tasks" subtitle="Background maintenance jobs. Run them on demand or adjust their schedule.">
<div className="space-y-2">
{(data ?? []).map((task) => (
<TaskRow
key={task.key}
task={task}
busy={patchTask.isPending || runTask.isPending}
onToggle={(enabled) => patchTask.mutate({ key: task.key, patch: { enabled } })}
onSaveInterval={(intervalMs) => patchTask.mutate({ key: task.key, patch: { intervalMs } })}
onRun={() => runTask.mutate(task.key)}
/>
))}
{data && data.length === 0 && (
<p className="text-sm text-soft">No tasks registered.</p>
)}
</div>
</Section>
);
}
function TaskRow({
task,
busy,
onToggle,
onSaveInterval,
onRun,
}: {
task: TaskView;
busy: boolean;
onToggle: (enabled: boolean) => void;
onSaveInterval: (intervalMs: number) => void;
onRun: () => void;
}) {
const [minutes, setMinutes] = useState(String(Math.round(task.intervalMs / 60000)));
useEffect(() => {
setMinutes(String(Math.round(task.intervalMs / 60000)));
}, [task.intervalMs]);
const parsedMinutes = Number(minutes);
const intervalDirty =
Number.isFinite(parsedMinutes) && parsedMinutes > 0 && parsedMinutes * 60000 !== task.intervalMs;
return (
<div className="rounded-lg border border-line bg-surface px-3 py-2 space-y-2">
<div className="flex items-start justify-between gap-3">
<div className="min-w-0">
<div className="flex items-center gap-2">
<span className="text-sm font-medium text-foreground truncate">{task.name}</span>
<Badge tone={STATUS_TONE[task.lastStatus]}>{task.lastStatus}</Badge>
</div>
{task.description && <p className="text-xs text-soft mt-0.5">{task.description}</p>}
</div>
<div className="flex items-center gap-2 shrink-0">
<Switch checked={task.enabled} onChange={onToggle} ariaLabel={`Enable ${task.name}`} disabled={busy} />
<Button variant="outline" size="sm" onClick={onRun} disabled={busy || task.running}>
{task.running ? 'Running…' : 'Run now'}
</Button>
</div>
</div>
<div className="flex flex-wrap items-center gap-x-4 gap-y-1 text-xs text-soft">
<span>Last run: {formatRelative(task.lastRunAt)}</span>
<span>Duration: {formatDuration(task.lastDurationMs)}</span>
<span>Next run: {task.enabled ? formatRelative(task.nextRunAt) : 'disabled'}</span>
<span className="flex items-center gap-1">
Every
<Input
type="number"
min={1}
controlSize="sm"
className="w-16"
value={minutes}
onChange={(e) => setMinutes(e.target.value)}
aria-label={`${task.name} interval in minutes`}
/>
min
{intervalDirty && (
<Button
variant="primary"
size="xs"
onClick={() => onSaveInterval(Math.floor(parsedMinutes) * 60000)}
disabled={busy}
>
Save
</Button>
)}
</span>
</div>
{task.lastStatus === 'error' && task.lastError && (
<p className="text-xs text-danger truncate" title={task.lastError}>{task.lastError}</p>
)}
{task.lastStatus === 'ok' && task.lastResult && (
<p className="text-xs text-soft truncate" title={task.lastResult}>{task.lastResult}</p>
)}
</div>
);
}

1
src/db/schema.ts
View file

@ -19,3 +19,4 @@ export const ttsSegmentEntries = usePostgres ? postgresSchema.ttsSegmentEntries
export const ttsSegmentVariants = usePostgres ? postgresSchema.ttsSegmentVariants : sqliteSchema.ttsSegmentVariants;
export const adminProviders = usePostgres ? postgresSchema.adminProviders : sqliteSchema.adminProviders;
export const adminSettings = usePostgres ? postgresSchema.adminSettings : sqliteSchema.adminSettings;
export const scheduledTasks = usePostgres ? postgresSchema.scheduledTasks : sqliteSchema.scheduledTasks;

17
src/db/schema_postgres.ts
View file

@ -1,5 +1,5 @@
import { sql } from 'drizzle-orm';
import { pgTable, text, integer, real, date, bigint, primaryKey, index, jsonb, foreignKey } from 'drizzle-orm/pg-core';
import { pgTable, text, integer, real, date, bigint, boolean, primaryKey, index, jsonb, foreignKey } from 'drizzle-orm/pg-core';
import { user } from './schema_auth_postgres';
const PG_NOW_MS = sql`(extract(epoch from now()) * 1000)::bigint`;
@ -210,6 +210,21 @@ export const adminSettings = pgTable('admin_settings', {
updatedAt: bigint('updated_at', { mode: 'number' }).notNull().default(PG_NOW_MS),
});
export const scheduledTasks = pgTable('scheduled_tasks', {
key: text('key').primaryKey(),
enabled: boolean('enabled').notNull().default(true),
intervalMs: bigint('interval_ms', { mode: 'number' }).notNull(),
lastStatus: text('last_status').notNull().default('idle'),
lastRunAt: bigint('last_run_at', { mode: 'number' }),
lastDurationMs: bigint('last_duration_ms', { mode: 'number' }),
lastError: text('last_error'),
lastResultJson: text('last_result_json'),
nextRunAt: bigint('next_run_at', { mode: 'number' }),
runRequested: boolean('run_requested').notNull().default(false),
runningSince: bigint('running_since', { mode: 'number' }),
updatedAt: bigint('updated_at', { mode: 'number' }).notNull().default(PG_NOW_MS),
});
export const ttsSegmentVariants = pgTable('tts_segment_variants', {
segmentId: text('segment_id').notNull(),
userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }),

15
src/db/schema_sqlite.ts
View file

@ -210,6 +210,21 @@ export const adminSettings = sqliteTable('admin_settings', {
updatedAt: integer('updated_at').notNull().default(SQLITE_NOW_MS),
});
export const scheduledTasks = sqliteTable('scheduled_tasks', {
key: text('key').primaryKey(),
enabled: integer('enabled', { mode: 'boolean' }).notNull().default(true),
intervalMs: integer('interval_ms').notNull(),
lastStatus: text('last_status').notNull().default('idle'),
lastRunAt: integer('last_run_at'),
lastDurationMs: integer('last_duration_ms'),
lastError: text('last_error'),
lastResultJson: text('last_result_json'),
nextRunAt: integer('next_run_at'),
runRequested: integer('run_requested', { mode: 'boolean' }).notNull().default(false),
runningSince: integer('running_since'),
updatedAt: integer('updated_at').notNull().default(SQLITE_NOW_MS),
});
export const ttsSegmentVariants = sqliteTable('tts_segment_variants', {
segmentId: text('segment_id').notNull(),
userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }),

12
src/instrumentation.ts Normal file
View file

@ -0,0 +1,12 @@
/**
* Next.js startup hook. On a long-lived self-hosted server we run the task
* scheduler in-process. On Vercel (serverless, no persistent process) there is
* nothing to keep a loop alive, so a cron route drives the ticks instead.
*/
export async function register(): Promise<void> {
if (process.env.VERCEL) return;
if (process.env.NEXT_RUNTIME !== 'nodejs') return;
const { startTaskScheduler } = await import('@/lib/server/tasks/scheduler');
startTaskScheduler();
}

54
src/lib/server/documents/blobstore.ts
View file

@ -600,14 +600,59 @@ export async function deleteDocumentPrefix(prefix: string): Promise<number> {
return deleted;
}
export async function deleteExpiredTempDocumentUploads(
userId: string,
/**
* List the source document blobs under a namespace (content-addressed objects
* directly beneath `documents_v1/`, excluding the `parsed_v2/` and `ns/`
* subtrees via the `/` delimiter). Used by the orphaned-blob reaper.
*/
export async function listDocumentSourceBlobs(
namespace: string | null,
): Promise<Array<{ id: string; lastModifiedMs: number }>> {
const cfg = getS3Config();
const client = getS3ProxyClient();
const ns = sanitizeNamespace(namespace);
const nsSegment = ns ? `ns/${ns}/` : '';
const prefix = `${cfg.prefix}/documents_v1/${nsSegment}`;
const out: Array<{ id: string; lastModifiedMs: number }> = [];
let continuationToken: string | undefined;
do {
const listRes = await client.send(
new ListObjectsV2Command({
Bucket: cfg.bucket,
Prefix: prefix,
Delimiter: '/',
ContinuationToken: continuationToken,
}),
);
for (const item of listRes.Contents ?? []) {
const key = item.Key;
if (!key) continue;
const id = key.slice(prefix.length);
if (!isValidDocumentId(id)) continue;
out.push({ id, lastModifiedMs: item.LastModified?.getTime() ?? 0 });
}
continuationToken = listRes.IsTruncated ? listRes.NextContinuationToken : undefined;
} while (continuationToken);
return out;
}
/**
* Delete every temporary upload object (across all users) older than the given
* cutoff. Used by the cleanup-temp-uploads scheduled task.
*/
export async function deleteAllExpiredTempDocumentUploads(
namespace: string | null,
olderThanMs: number,
): Promise<number> {
const cfg = getS3Config();
const client = getS3ProxyClient();
const prefix = tempDocumentUploadPrefix(userId, namespace);
const ns = sanitizeNamespace(namespace);
const nsSegment = ns ? `ns/${ns}/` : '';
const prefix = `${cfg.prefix}/document_uploads_temp_v1/${nsSegment}`;
let continuationToken: string | undefined;
const keys: string[] = [];
@ -644,8 +689,9 @@ export async function deleteExpiredTempDocumentUploads(
},
}),
);
deleted += deleteRes.Deleted?.length ?? 0;
deleted += deleteRes.Deleted?.length ?? batch.length - (deleteRes.Errors?.length ?? 0);
}
return deleted;
}

104
src/lib/server/documents/delete-owned.ts
View file

@ -1,91 +1,41 @@
import { and, eq, ne } from 'drizzle-orm';
import { and, eq } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { deleteDocumentBlob } from '@/lib/server/documents/blobstore';
import {
cleanupDocumentPreviewArtifacts,
deleteDocumentPreviewRows,
} from '@/lib/server/documents/previews';
import { deleteDocumentTtsSegmentCache } from '@/lib/server/tts/segments-cache';
import { withDocumentLock } from '@/lib/server/documents/document-lock';
import { hashForLog, serverLogger } from '@/lib/server/logger';
import { logDegraded } from '@/lib/server/errors/logging';
type DocumentRow = typeof documents.$inferSelect;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
async function restoreDocumentOwnership(conn: any, row: DocumentRow): Promise<void> {
await conn.insert(documents).values(row).onConflictDoNothing();
}
async function removeOwnershipAndCheckLastOwner(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
conn: any,
input: { userId: string; documentId: string },
): Promise<{ removed: DocumentRow | null; isLastOwner: boolean }> {
const [removed] = await conn
/**
* Remove a user's ownership of a document.
*
* Only the per-user TTS segment cache is cleaned inline (it is keyed by userId
* and not reachable afterwards). The shared, content-addressed document blob and
* its previews are reclaimed by the `reap-orphaned-blobs` task once no owner
* remains, so there is no inline blob deletion, last-owner check, or lock.
*/
export async function deleteOwnedDocument(input: {
userId: string;
documentId: string;
namespace: string | null;
}): Promise<boolean> {
const [removed] = await db
.delete(documents)
.where(and(
eq(documents.id, input.documentId),
eq(documents.userId, input.userId),
))
.returning();
if (!removed) return { removed: null, isLastOwner: false };
if (!removed) return false;
const otherOwners = await conn
.select({ id: documents.id })
.from(documents)
.where(and(
eq(documents.id, input.documentId),
ne(documents.userId, input.userId),
))
.limit(1);
return { removed, isLastOwner: otherOwners.length === 0 };
}
export async function deleteOwnedDocument(input: {
userId: string;
documentId: string;
namespace: string | null;
}): Promise<boolean> {
return withDocumentLock(input.documentId, async (conn) => {
const { removed, isLastOwner } = await removeOwnershipAndCheckLastOwner(conn, input);
if (!removed) return false;
try {
await deleteDocumentTtsSegmentCache(input);
if (isLastOwner) {
await cleanupDocumentPreviewArtifacts(input.documentId, input.namespace);
await deleteDocumentPreviewRows(input.documentId, input.namespace);
const [newOwner] = await conn
.select({ id: documents.id })
.from(documents)
.where(eq(documents.id, input.documentId))
.limit(1);
if (!newOwner) {
await deleteDocumentBlob(input.documentId, input.namespace);
}
}
} catch (error) {
// Best-effort rollback; never let a restore failure mask the original error.
try {
await restoreDocumentOwnership(conn, removed);
} catch (restoreError) {
logDegraded(serverLogger, {
event: 'documents.delete_owned.restore_ownership.failed',
msg: 'Failed to restore document ownership after deletion failure',
step: 'restore_document_ownership',
context: {
documentId: input.documentId,
userIdHash: hashForLog(input.userId),
},
error: restoreError,
});
}
throw error;
}
return true;
await deleteDocumentTtsSegmentCache(input).catch((error) => {
logDegraded(serverLogger, {
event: 'documents.delete_owned.tts_cache_cleanup.failed',
msg: 'Failed to clean TTS segment cache after document deletion',
step: 'delete_document_tts_segment_cache',
context: { documentId: input.documentId, userIdHash: hashForLog(input.userId) },
error,
});
});
return true;
}

59
src/lib/server/documents/document-lock.ts
View file

@ -1,59 +0,0 @@
import { sql } from 'drizzle-orm';
import { runInDbTransaction } from '@/db/run-in-transaction';
// In-process serialization for the single-writer SQLite deployment. SQLite runs
// in one process, so an in-memory promise chain per document is sufficient (and
// is the only lock available — there is no advisory lock). On Postgres this is
// unused: the transaction-scoped advisory lock below provides exclusion that
// works across a stateless/serverless fleet.
const localTails = new Map<string, Promise<void>>();
async function withLocalDocumentLock<T>(documentId: string, fn: () => Promise<T>): Promise<T> {
const previous = localTails.get(documentId) ?? Promise.resolve();
let release!: () => void;
const current = new Promise<void>((resolve) => {
release = resolve;
});
localTails.set(documentId, current);
await previous;
try {
return await fn();
} finally {
release();
if (localTails.get(documentId) === current) {
localTails.delete(documentId);
}
}
}
/**
* Serialize all mutations to a single document and run `fn` with exclusive
* access, passing it the connection to do its work on.
*
* The SQLite/Postgres transaction bridge is delegated to `runInDbTransaction`;
* this helper only adds the exclusion on top:
* - Postgres: a transaction-scoped advisory lock keyed on the document id,
* acquired as the first statement of the shared transaction so it covers
* the whole read-modify-write. It releases on commit, so it is safe under
* transaction-mode poolers and gives fleet-wide exclusion in a stateless
* deployment.
* - SQLite: an in-process lock; the single WAL writer needs nothing more.
*
* Because this guarantees exclusivity for the document, `fn` does not need its
* own transaction or `SELECT ... FOR UPDATE` to read-modify-write safely.
*/
export async function withDocumentLock<T>(
documentId: string,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
fn: (conn: any) => Promise<T>,
): Promise<T> {
if (process.env.POSTGRES_URL) {
return runInDbTransaction(async (conn) => {
await conn.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${documentId}, 0))`);
return fn(conn);
});
}
return withLocalDocumentLock(documentId, () => runInDbTransaction(fn));
}

5
src/lib/server/documents/previews.ts
View file

@ -8,7 +8,6 @@ import {
DOCUMENT_PREVIEW_CONTENT_TYPE,
DOCUMENT_PREVIEW_VARIANT,
DOCUMENT_PREVIEW_WIDTH,
deleteDocumentPreviewArtifacts,
documentPreviewKey,
headDocumentPreview,
isMissingBlobError,
@ -419,7 +418,3 @@ export async function deleteDocumentPreviewRows(documentId: string, namespace: s
.delete(documentPreviews)
.where(and(eq(documentPreviews.documentId, documentId), eq(documentPreviews.namespace, namespaceKey)));
}
export async function cleanupDocumentPreviewArtifacts(documentId: string, namespace: string | null): Promise<void> {
await deleteDocumentPreviewArtifacts(documentId, namespace);
}

23
src/lib/server/rate-limit/job-rate-limiter.ts
View file

@ -1,4 +1,4 @@
import { and, eq, gte, lt, sql } from 'drizzle-orm';
import { and, eq, gte, sql } from 'drizzle-orm';
import { db } from '@/db';
import { userJobEvents } from '@/db/schema';
import { nowTimestampMs } from '@/lib/shared/timestamps';
@ -143,24 +143,5 @@ export async function recordJobEvent(
} catch {
// Recording is best-effort; never block op creation on ledger writes.
}
// Opportunistic prune of rows older than the largest configured window so
// the ledger stays small without a separate cron, but never deletes events
// that could still affect an in-window count.
if (Math.random() < 0.05) {
const largestWindowMs = config.windows.reduce(
(max, w) => (Number.isFinite(w.windowMs) && w.windowMs > max ? w.windowMs : max),
24 * 60 * 60 * 1000,
);
try {
await safeDb()
.delete(userJobEvents)
.where(and(
eq(userJobEvents.action, action),
lt(userJobEvents.createdAt, now - largestWindowMs),
));
} catch {
// ignore prune failures
}
}
// Old rows are removed by the prune-job-events scheduled task.
}

186
src/lib/server/tasks/engine.ts Normal file
View file

@ -0,0 +1,186 @@
import { and, eq, lt, ne, or, sql } from 'drizzle-orm';
import { db } from '@/db';
import { scheduledTasks } from '@/db/schema';
import { serverLogger } from '@/lib/server/logger';
import { logDegraded } from '@/lib/server/errors/logging';
import { TASK_REGISTRY } from './registry';
import type { TaskDef, TaskRegistry, TaskRunStatus } from './types';
// A task still marked 'running' after this long is assumed abandoned (process
// crashed mid-run) and may be reclaimed by the next tick.
const STALE_RUNNING_MS = 60 * 60 * 1000;
/**
* Seed a `scheduled_tasks` row for every registered task. Idempotent: existing
* rows (including user-edited interval/enabled) are left untouched.
*/
async function ensureTaskRows(registry: TaskRegistry = TASK_REGISTRY): Promise<void> {
const now = Date.now();
for (const [key, def] of Object.entries(registry)) {
await db
.insert(scheduledTasks)
.values({
key,
enabled: true,
intervalMs: def.defaultIntervalMs,
lastStatus: 'idle',
nextRunAt: now,
})
.onConflictDoNothing();
}
}
/**
* Atomically claim a task for execution. Returns true only if this caller won
* the claim, so the same task never runs concurrently (across instances too):
* the single UPDATE flips status to 'running' iff it is not already running
* (or its running marker is stale).
*/
async function claimTask(key: string, now: number): Promise<boolean> {
const claimed = await db
.update(scheduledTasks)
.set({ lastStatus: 'running', runningSince: now, runRequested: false, updatedAt: now })
.where(and(
eq(scheduledTasks.key, key),
or(
ne(scheduledTasks.lastStatus, 'running'),
lt(scheduledTasks.runningSince, now - STALE_RUNNING_MS),
),
))
.returning({ key: scheduledTasks.key });
return claimed.length > 0;
}
async function finishTask(
key: string,
outcome:
| { status: 'ok'; startedAt: number; summary: string | null }
| { status: 'error'; startedAt: number; error: unknown },
): Promise<void> {
const now = Date.now();
await db
.update(scheduledTasks)
.set({
lastStatus: outcome.status,
lastRunAt: now,
lastDurationMs: now - outcome.startedAt,
lastError: outcome.status === 'error' ? String(outcome.error) : null,
lastResultJson: outcome.status === 'ok' && outcome.summary ? outcome.summary : null,
// Schedule the next run off the row's (possibly user-edited) interval.
nextRunAt: sql`${now} + ${scheduledTasks.intervalMs}`,
runningSince: null,
runRequested: false,
updatedAt: now,
})
.where(eq(scheduledTasks.key, key));
}
async function executeTask(key: string, def: TaskDef): Promise<void> {
const startedAt = Date.now();
try {
const result = await def.run();
const summary = result && typeof result.summary === 'string' ? result.summary : null;
await finishTask(key, { status: 'ok', startedAt, summary });
} catch (error) {
logDegraded(serverLogger, {
event: 'tasks.run.failed',
msg: `Scheduled task "${key}" failed`,
step: key,
error,
});
await finishTask(key, { status: 'error', startedAt, error });
}
}
/**
* Run every task that is due (interval elapsed or a manual run was requested),
* claiming each first so concurrent ticks don't double-run. Safe to call from
* any trigger: the self-host interval, a Vercel cron route, or a manual run.
*/
export async function runDueTasks(options?: { registry?: TaskRegistry }): Promise<void> {
const registry = options?.registry ?? TASK_REGISTRY;
await ensureTaskRows(registry);
const now = Date.now();
const rows = await db.select().from(scheduledTasks);
for (const row of rows) {
const def = registry[row.key];
if (!def) continue; // orphaned row for a task no longer in the registry
const due =
row.runRequested ||
(row.enabled && row.nextRunAt != null && now >= Number(row.nextRunAt));
if (!due) continue;
if (!(await claimTask(row.key, now))) continue;
await executeTask(row.key, def);
}
}
/**
* Run a single task immediately, regardless of schedule. Returns false if the
* key is unknown or the task is already running.
*/
export async function runTaskNow(key: string, registry: TaskRegistry = TASK_REGISTRY): Promise<boolean> {
const def = registry[key];
if (!def) return false;
await ensureTaskRows(registry);
if (!(await claimTask(key, Date.now()))) return false;
await executeTask(key, def);
return true;
}
/** Update a task's user-editable fields (enable/disable, run interval). */
export async function updateTask(
key: string,
patch: { enabled?: boolean; intervalMs?: number },
): Promise<void> {
const set: Record<string, unknown> = { updatedAt: Date.now() };
if (typeof patch.enabled === 'boolean') set.enabled = patch.enabled;
if (typeof patch.intervalMs === 'number' && Number.isFinite(patch.intervalMs) && patch.intervalMs > 0) {
set.intervalMs = Math.floor(patch.intervalMs);
}
await db.update(scheduledTasks).set(set).where(eq(scheduledTasks.key, key));
}
export type TaskView = {
key: string;
name: string;
description?: string;
enabled: boolean;
intervalMs: number;
lastStatus: TaskRunStatus;
lastRunAt: number | null;
lastDurationMs: number | null;
lastError: string | null;
lastResult: string | null;
nextRunAt: number | null;
running: boolean;
};
/** Combined registry + stored-state view for the admin tasks UI. */
export async function listTasks(registry: TaskRegistry = TASK_REGISTRY): Promise<TaskView[]> {
await ensureTaskRows(registry);
const rows = await db.select().from(scheduledTasks);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const byKey = new Map<string, any>(rows.map((row: any) => [row.key, row]));
return Object.entries(registry).map(([key, def]) => {
const row = byKey.get(key);
return {
key,
name: def.name,
description: def.description,
enabled: !!row?.enabled,
intervalMs: Number(row?.intervalMs ?? def.defaultIntervalMs),
lastStatus: (row?.lastStatus ?? 'idle') as TaskRunStatus,
lastRunAt: row?.lastRunAt != null ? Number(row.lastRunAt) : null,
lastDurationMs: row?.lastDurationMs != null ? Number(row.lastDurationMs) : null,
lastError: row?.lastError ?? null,
lastResult: row?.lastResultJson ?? null,
nextRunAt: row?.nextRunAt != null ? Number(row.nextRunAt) : null,
running: row?.lastStatus === 'running',
};
});
}

17
src/lib/server/tasks/handlers/cleanup-temp-uploads.ts Normal file
View file

@ -0,0 +1,17 @@
import { isS3Configured } from '@/lib/server/storage/s3';
import {
TEMP_DOCUMENT_UPLOAD_TTL_MS,
deleteAllExpiredTempDocumentUploads,
} from '@/lib/server/documents/blobstore';
import type { TaskResult } from '../types';
/** Remove temporary upload objects past their TTL across all users. */
export async function cleanupTempUploads(): Promise<TaskResult> {
if (!isS3Configured()) {
return { summary: 'Skipped: object storage not configured', deleted: 0 };
}
const cutoff = Date.now() - TEMP_DOCUMENT_UPLOAD_TTL_MS;
const deleted = await deleteAllExpiredTempDocumentUploads(null, cutoff);
return { summary: `Deleted ${deleted} expired upload object(s)`, deleted };
}

25
src/lib/server/tasks/handlers/prune-job-events.ts Normal file
View file

@ -0,0 +1,25 @@
import { lt } from 'drizzle-orm';
import { db } from '@/db';
import { userJobEvents } from '@/db/schema';
import type { TaskResult } from '../types';
// Retention far exceeds the largest rate-limit window, so pruning never removes
// an event that could still affect an in-window count.
const RETENTION_MS = 7 * 24 * 60 * 60 * 1000;
function rowsAffected(result: unknown): number {
if (result && typeof result === 'object') {
const rec = result as Record<string, unknown>;
if (typeof rec.rowCount === 'number') return rec.rowCount;
if (typeof rec.changes === 'number') return rec.changes;
}
return 0;
}
/** Delete rate-limit job-event rows older than the retention window. */
export async function pruneJobEvents(): Promise<TaskResult> {
const cutoff = Date.now() - RETENTION_MS;
const result = await db.delete(userJobEvents).where(lt(userJobEvents.createdAt, cutoff));
const pruned = rowsAffected(result);
return { summary: `Pruned ${pruned} job event(s)`, pruned };
}

10
src/lib/server/tasks/handlers/prune-tts-usage.ts Normal file
View file

@ -0,0 +1,10 @@
import { rateLimiter } from '@/lib/server/rate-limit/rate-limiter';
import type { TaskResult } from '../types';
const RETENTION_DAYS = 30;
/** Delete TTS usage counter rows (user_tts_chars) older than the retention window. */
export async function pruneTtsUsage(): Promise<TaskResult> {
await rateLimiter.cleanupOldRecords(RETENTION_DAYS);
return { summary: `Pruned TTS usage older than ${RETENTION_DAYS}d` };
}

68
src/lib/server/tasks/handlers/reap-orphaned-blobs.ts Normal file
View file

@ -0,0 +1,68 @@
import { inArray } from 'drizzle-orm';
import { db } from '@/db';
import { documents } from '@/db/schema';
import { isS3Configured } from '@/lib/server/storage/s3';
import { deleteDocumentBlob, listDocumentSourceBlobs } from '@/lib/server/documents/blobstore';
import { deleteDocumentPreviewArtifacts } from '@/lib/server/documents/previews-blobstore';
import { deleteDocumentPreviewRows } from '@/lib/server/documents/previews';
import { serverLogger } from '@/lib/server/logger';
import { logDegraded } from '@/lib/server/errors/logging';
import type { TaskResult } from '../types';
// Don't reap a blob younger than this — it may belong to an in-flight finalize
// that has written the blob but not yet committed its ownership row.
const GRACE_MS = 60 * 60 * 1000;
const OWNERSHIP_CHECK_BATCH = 200;
/**
* Delete content-addressed document blobs that no longer have any owner.
*
* Reference count = ownership rows in `documents`. A blob with zero rows and
* age past the grace window is an orphan (e.g. left by a failed inline delete)
* and is safe to remove. Production data is non-namespaced.
*/
export async function reapOrphanedBlobs(): Promise<TaskResult> {
if (!isS3Configured()) {
return { summary: 'Skipped: object storage not configured', reaped: 0 };
}
const now = Date.now();
const blobs = await listDocumentSourceBlobs(null);
const candidates = blobs.filter((blob) => now - blob.lastModifiedMs > GRACE_MS);
let reaped = 0;
for (let i = 0; i < candidates.length; i += OWNERSHIP_CHECK_BATCH) {
const chunk = candidates.slice(i, i + OWNERSHIP_CHECK_BATCH);
const ids = chunk.map((c) => c.id);
const ownedRows = await db
.select({ id: documents.id })
.from(documents)
.where(inArray(documents.id, ids));
const owned = new Set(ownedRows.map((row: { id: string }) => row.id));
for (const candidate of chunk) {
if (owned.has(candidate.id)) continue;
try {
await deleteDocumentBlob(candidate.id, null);
await deleteDocumentPreviewArtifacts(candidate.id, null);
await deleteDocumentPreviewRows(candidate.id, null);
reaped += 1;
} catch (error) {
logDegraded(serverLogger, {
event: 'tasks.reap_orphaned_blobs.delete_failed',
msg: 'Failed to reap orphaned document storage',
step: 'reap_orphaned_blob',
context: { documentId: candidate.id },
error,
});
}
}
}
return {
summary: `Reaped ${reaped} orphaned blob(s)`,
scanned: blobs.length,
candidates: candidates.length,
reaped,
};
}

37
src/lib/server/tasks/registry.ts Normal file
View file

@ -0,0 +1,37 @@
import type { TaskRegistry } from './types';
import { reapOrphanedBlobs } from './handlers/reap-orphaned-blobs';
import { cleanupTempUploads } from './handlers/cleanup-temp-uploads';
import { pruneJobEvents } from './handlers/prune-job-events';
import { pruneTtsUsage } from './handlers/prune-tts-usage';
/**
* The catalog of scheduled tasks. Each key is the stable task id stored in the
* `scheduled_tasks` table; renaming a key orphans its row (the engine ignores
* rows with no matching definition).
*/
export const TASK_REGISTRY: TaskRegistry = {
'reap-orphaned-blobs': {
name: 'Reap orphaned document blobs',
description: 'Delete content-addressed document blobs that no longer have any owner.',
defaultIntervalMs: 6 * 60 * 60 * 1000,
run: reapOrphanedBlobs,
},
'cleanup-temp-uploads': {
name: 'Clean up expired uploads',
description: 'Remove temporary upload objects past their TTL.',
defaultIntervalMs: 60 * 60 * 1000,
run: cleanupTempUploads,
},
'prune-job-events': {
name: 'Prune job event ledger',
description: 'Delete rate-limit job events older than the retention window.',
defaultIntervalMs: 24 * 60 * 60 * 1000,
run: pruneJobEvents,
},
'prune-tts-usage': {
name: 'Prune TTS usage counters',
description: 'Delete TTS usage rows (user_tts_chars) older than 30 days.',
defaultIntervalMs: 24 * 60 * 60 * 1000,
run: pruneTtsUsage,
},
};

33
src/lib/server/tasks/scheduler.ts Normal file
View file

@ -0,0 +1,33 @@
import { runDueTasks } from './engine';
import { serverLogger } from '@/lib/server/logger';
const TICK_INTERVAL_MS = 60_000;
const INITIAL_DELAY_MS = 10_000;
let started = false;
/**
* Start the in-process scheduler loop. Intended for the long-lived self-hosted
* server only — on Vercel a cron route drives the ticks instead. Idempotent:
* repeated calls (e.g. dev HMR) start a single loop.
*/
export function startTaskScheduler(): void {
if (started) return;
started = true;
const tick = async () => {
try {
await runDueTasks();
} catch (error) {
serverLogger.warn(
{ event: 'tasks.scheduler.tick_failed', error: String(error) },
'Task scheduler tick failed',
);
}
};
setTimeout(tick, INITIAL_DELAY_MS);
const handle = setInterval(tick, TICK_INTERVAL_MS);
// Don't keep the process alive solely for the scheduler.
if (typeof handle.unref === 'function') handle.unref();
}

25
src/lib/server/tasks/types.ts Normal file
View file

@ -0,0 +1,25 @@
/** Outcome a task handler may return; surfaced in the admin UI as a summary. */
export type TaskResult = {
/** Short human-readable summary, e.g. "Reaped 3 orphaned blobs". */
summary?: string;
/** Arbitrary structured detail for debugging. */
[key: string]: unknown;
};
export type TaskHandler = () => Promise<TaskResult | void>;
/** Static definition of a task, kept in code (the registry). */
export type TaskDef = {
/** Display name shown in the admin tasks list. */
name: string;
/** Optional longer description of what the task does. */
description?: string;
/** Default run interval in ms; the per-task row may override it. */
defaultIntervalMs: number;
/** The work to perform. Must be idempotent and safe to re-run. */
run: TaskHandler;
};
export type TaskRegistry = Record<string, TaskDef>;
export type TaskRunStatus = 'idle' | 'running' | 'ok' | 'error';

200
src/lib/server/user/data-cleanup.ts
View file

@ -1,14 +1,20 @@
/**
* Cleans up all S3 storage artifacts belonging to a user.
* Called from Better Auth's `beforeDelete` hook so that blobs are removed
* before the DB cascade wipes the metadata rows we query against.
* Cleans up user-scoped storage that the orphaned-blob reaper cannot reach.
*
* Called from Better Auth's `beforeDelete` hook (canonical pass) and the
* account-delete route (test-namespaced pass). Shared, content-addressed
* document blobs + previews are NOT deleted here on the canonical pass — they
* are reclaimed by the `reap-orphaned-blobs` task once their ownership rows are
* gone. Per-user storage (audiobooks, TTS segments, temp uploads) is keyed by
* userId and would be unreachable after the cascade, so it is deleted inline
* and failures block the deletion.
*/
import { db } from '@/db';
import { documents, audiobooks, userJobEvents, userTtsChars } from '@/db/schema';
import * as authSchemaSqlite from '@/db/schema_auth_sqlite';
import * as authSchemaPostgres from '@/db/schema_auth_postgres';
import { and, eq, ne } from 'drizzle-orm';
import { eq } from 'drizzle-orm';
import { getS3Config, isS3Configured } from '@/lib/server/storage/s3';
import {
deleteDocumentBlob,
@ -21,21 +27,9 @@ import { audiobookPrefix, deleteAudiobookPrefix } from '@/lib/server/audiobooks/
import { deleteTtsSegmentPrefix } from '@/lib/server/tts/segments-blobstore';
import { hashForLog, serverLogger } from '@/lib/server/logger';
import { logDegraded } from '@/lib/server/errors/logging';
import { withDocumentLock } from '@/lib/server/documents/document-lock';
type DocumentRow = typeof documents.$inferSelect;
type AudiobookRow = { id: string };
/**
* Delete all S3 blobs owned by `userId`.
*
* This covers:
* - Document file blobs
* - Document preview images
* - Audiobook audio files (chapter mp3s, metadata json, etc.)
*
* Each item is cleaned up independently; a failure on one does not block the rest.
*/
export async function deleteUserStorageData(
userId: string,
namespace: string | null,
@ -47,130 +41,37 @@ export async function deleteUserStorageData(
const database = db as any;
const authSchema = process.env.POSTGRES_URL ? authSchemaPostgres : authSchemaSqlite;
// --- Documents & previews ---
const userDocs: DocumentRow[] = await database
.select()
.from(documents)
.where(eq(documents.userId, userId));
let docsDeleted = 0;
const removedDocs: DocumentRow[] = [];
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const restoreRemovedDocs = async (conn: any = database) => {
if (removedDocs.length === 0) return;
const docsToRestore = [...removedDocs];
await conn.insert(documents).values(docsToRestore).onConflictDoNothing();
removedDocs.length = 0;
};
// Restore before throwing the diagnostic AggregateError, but never let a
// restore failure replace the original cleanup failures we want to surface.
const tryRestoreRemovedDocs = async () => {
try {
await restoreRemovedDocs();
} catch (error) {
logDegraded(serverLogger, {
event: 'user.data_cleanup.restore_removed_docs.failed',
msg: 'Failed to restore document rows after cleanup failure',
step: 'restore_removed_docs',
context: { userIdHash: hashForLog(userId) },
error,
});
}
};
// The mutation lock already serializes all mutations for this document, so a
// plain read-modify-write is safe without an inner transaction or FOR UPDATE.
const removeOwnershipAndCheckLastOwner = async (
// eslint-disable-next-line @typescript-eslint/no-explicit-any
conn: any,
doc: DocumentRow,
) => {
const [removedDoc] = await conn
.delete(documents)
.where(and(eq(documents.id, doc.id), eq(documents.userId, userId)))
.returning();
if (!removedDoc) return { removedDoc: null, isLastOwner: false };
const otherOwners = await conn
// --- Document blobs & previews ---
// Canonical pass: deferred to the reap-orphaned-blobs task (the rows cascade
// away with the user, then the reaper reclaims the now-orphaned blobs). The
// reaper only runs for the canonical namespace, so test-namespaced storage is
// deleted inline here.
let docBlobsDeleted = 0;
if (s3Enabled && namespace !== null) {
const userDocs = (await database
.select({ id: documents.id })
.from(documents)
.where(and(
eq(documents.id, doc.id),
ne(documents.userId, userId),
))
.limit(1);
return { removedDoc, isLastOwner: otherOwners.length === 0 };
};
for (const doc of userDocs) {
await withDocumentLock(doc.id, async (conn) => {
const { removedDoc, isLastOwner } = await removeOwnershipAndCheckLastOwner(conn, doc);
if (!removedDoc) return;
removedDocs.push(removedDoc);
if (s3Enabled && isLastOwner) {
try {
await deleteDocumentPreviewArtifacts(doc.id, namespace);
} catch (error) {
failures.push(error);
logDegraded(serverLogger, {
event: 'user.data_cleanup.document_preview_delete.failed',
msg: 'Failed to delete preview artifacts',
step: 'delete_document_preview_artifacts',
context: {
documentId: doc.id,
userIdHash: hashForLog(userId),
},
error,
});
}
.where(eq(documents.userId, userId))) as Array<{ id: string }>;
for (const doc of userDocs) {
try {
await deleteDocumentBlob(doc.id, namespace);
await deleteDocumentPreviewArtifacts(doc.id, namespace);
await deleteDocumentPreviewRows(doc.id, namespace);
docBlobsDeleted++;
} catch (error) {
failures.push(error);
logDegraded(serverLogger, {
event: 'user.data_cleanup.document_storage_delete.failed',
msg: 'Failed to delete namespaced document storage',
step: 'delete_namespaced_document_storage',
context: { documentId: doc.id, userIdHash: hashForLog(userId) },
error,
});
}
// Preview metadata is global, so only the canonical final-owner pass may
// remove it.
if (namespace === null && isLastOwner) {
try {
await deleteDocumentPreviewRows(doc.id, namespace);
} catch (error) {
failures.push(error);
logDegraded(serverLogger, {
event: 'user.data_cleanup.document_preview_rows_delete.failed',
msg: 'Failed to delete preview rows',
step: 'delete_document_preview_rows',
context: {
documentId: doc.id,
userIdHash: hashForLog(userId),
},
error,
});
}
}
if (s3Enabled && isLastOwner) {
try {
await deleteDocumentBlob(doc.id, namespace);
docsDeleted++;
} catch (error) {
failures.push(error);
logDegraded(serverLogger, {
event: 'user.data_cleanup.document_blob_delete.failed',
msg: 'Failed to delete document blob',
step: 'delete_document_blob',
context: {
documentId: doc.id,
userIdHash: hashForLog(userId),
},
error,
});
}
}
if (namespace !== null) {
await restoreRemovedDocs(conn);
}
});
}
}
// --- Audiobooks ---
// --- Audiobooks (per-user object storage; not reaped) ---
const userBooks: AudiobookRow[] = s3Enabled
? await database
.select({ id: audiobooks.id })
@ -181,8 +82,7 @@ export async function deleteUserStorageData(
let booksDeleted = 0;
for (const book of userBooks) {
try {
const prefix = audiobookPrefix(book.id, userId, namespace);
await deleteAudiobookPrefix(prefix);
await deleteAudiobookPrefix(audiobookPrefix(book.id, userId, namespace));
booksDeleted++;
} catch (error) {
failures.push(error);
@ -190,16 +90,13 @@ export async function deleteUserStorageData(
event: 'user.data_cleanup.audiobook_blobs_delete.failed',
msg: 'Failed to delete audiobook blobs',
step: 'delete_audiobook_prefix',
context: {
bookId: book.id,
userIdHash: hashForLog(userId),
},
context: { bookId: book.id, userIdHash: hashForLog(userId) },
error,
});
}
}
// --- TTS segments ---
// --- Temp uploads + TTS segments (per-user object storage; not reaped) ---
let segmentsDeleted = 0;
if (s3Enabled) {
try {
@ -234,17 +131,18 @@ export async function deleteUserStorageData(
}
}
// Block deletion if any non-reapable storage cleanup failed — proceeding
// would permanently orphan it. Nothing was removed from the database yet, so
// there is nothing to roll back.
if (failures.length > 0) {
await tryRestoreRemovedDocs();
throw new AggregateError(failures, `User storage cleanup failed in ${failures.length} operation(s)`);
}
// Namespaced cleanup is an additional storage pass made by the account
// route. Database rows are global and must only be removed during the
// canonical non-namespaced beforeDelete pass.
// Namespaced cleanup is a storage-only pass; database rows are global and are
// only removed on the canonical (non-namespaced) pass.
if (namespace === null) {
// Delete explicitly for compatibility with pre-cascade installations and
// to remove auth verification tokens, which cannot carry a user FK.
// Explicit for compatibility with pre-cascade installations and to remove
// auth verification tokens, which cannot carry a user FK.
for (const { table, userColumn, step } of [
{ table: userTtsChars, userColumn: userTtsChars.userId, step: 'delete_user_tts_usage_rows' },
{ table: userJobEvents, userColumn: userJobEvents.userId, step: 'delete_user_job_event_rows' },
@ -263,12 +161,11 @@ export async function deleteUserStorageData(
}
}
if (docsDeleted > 0 || booksDeleted > 0 || segmentsDeleted > 0) {
if (docBlobsDeleted > 0 || booksDeleted > 0 || segmentsDeleted > 0) {
serverLogger.info({
event: 'user.data_cleanup.completed',
userIdHash: hashForLog(userId),
docsDeleted,
totalDocs: userDocs.length,
docBlobsDeleted,
booksDeleted,
totalBooks: userBooks.length,
segmentsDeleted,
@ -276,7 +173,6 @@ export async function deleteUserStorageData(
}
if (failures.length > 0) {
await tryRestoreRemovedDocs();
throw new AggregateError(failures, `User database cleanup failed in ${failures.length} operation(s)`);
}
}

87
tests/unit/document-delete-cleanup.vitest.spec.ts
View file

@ -1,24 +1,12 @@
import { beforeEach, describe, expect, test, vi } from 'vitest';
const mocks = vi.hoisted(() => ({
selectResults: [] as unknown[][],
deleteResults: [] as unknown[][],
insertValues: vi.fn(() => ({ onConflictDoNothing: vi.fn(async () => undefined) })),
deleteDocumentBlob: vi.fn(async () => undefined),
cleanupDocumentPreviewArtifacts: vi.fn(async () => undefined),
deleteDocumentPreviewRows: vi.fn(async () => undefined),
deleteDocumentTtsSegmentCache: vi.fn(async () => undefined),
}));
function resultBuilder(result: unknown[]) {
const limited = {
all: () => result,
then: (resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) =>
Promise.resolve(result).then(resolve, reject),
};
return {
all: () => result,
limit: () => limited,
then: (resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) =>
Promise.resolve(result).then(resolve, reject),
};
@ -26,58 +14,37 @@ function resultBuilder(result: unknown[]) {
vi.mock('@/db', () => {
const database = {
select: vi.fn(() => ({
from: vi.fn(() => ({
where: vi.fn(() => resultBuilder(mocks.selectResults.shift() ?? [])),
})),
})),
delete: vi.fn(() => ({
where: vi.fn(() => ({
returning: () => resultBuilder(mocks.deleteResults.shift() ?? []),
})),
})),
insert: vi.fn(() => ({
values: mocks.insertValues,
})),
transaction: vi.fn((callback: (tx: unknown) => unknown) => callback(database)),
};
return { db: database };
});
vi.mock('@/lib/server/documents/blobstore', () => ({
deleteDocumentBlob: mocks.deleteDocumentBlob,
}));
vi.mock('@/lib/server/documents/previews', () => ({
cleanupDocumentPreviewArtifacts: mocks.cleanupDocumentPreviewArtifacts,
deleteDocumentPreviewRows: mocks.deleteDocumentPreviewRows,
}));
vi.mock('@/lib/server/tts/segments-cache', () => ({
deleteDocumentTtsSegmentCache: mocks.deleteDocumentTtsSegmentCache,
}));
vi.mock('@/lib/server/logger', () => ({
hashForLog: () => 'hash',
serverLogger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
}));
vi.mock('@/lib/server/errors/logging', () => ({ logDegraded: vi.fn() }));
import { deleteOwnedDocument } from '../../src/lib/server/documents/delete-owned';
describe('owned document cleanup', () => {
beforeEach(() => {
mocks.selectResults = [];
mocks.deleteResults = [];
mocks.insertValues.mockReset();
mocks.insertValues.mockReturnValue({ onConflictDoNothing: vi.fn(async () => undefined) });
mocks.deleteDocumentBlob.mockReset();
mocks.deleteDocumentBlob.mockResolvedValue(undefined);
mocks.cleanupDocumentPreviewArtifacts.mockReset();
mocks.cleanupDocumentPreviewArtifacts.mockResolvedValue(undefined);
mocks.deleteDocumentPreviewRows.mockReset();
mocks.deleteDocumentPreviewRows.mockResolvedValue(undefined);
mocks.deleteDocumentTtsSegmentCache.mockReset();
mocks.deleteDocumentTtsSegmentCache.mockResolvedValue(undefined);
});
test('deletes only user-scoped TTS when another owner remains', async () => {
test('removes the ownership row and cleans the per-user TTS cache', async () => {
mocks.deleteResults = [[{ id: 'doc-1', userId: 'user-1' }]];
mocks.selectResults = [[{ id: 'doc-1' }]];
await expect(deleteOwnedDocument({
userId: 'user-1',
@ -86,50 +53,28 @@ describe('owned document cleanup', () => {
})).resolves.toBe(true);
expect(mocks.deleteDocumentTtsSegmentCache).toHaveBeenCalledOnce();
expect(mocks.deleteDocumentBlob).not.toHaveBeenCalled();
expect(mocks.cleanupDocumentPreviewArtifacts).not.toHaveBeenCalled();
});
test('deletes shared artifacts only for the final owner', async () => {
mocks.deleteResults = [[{ id: 'doc-1', userId: 'user-1' }]];
mocks.selectResults = [[], []];
await deleteOwnedDocument({
userId: 'user-1',
documentId: 'doc-1',
namespace: null,
});
expect(mocks.cleanupDocumentPreviewArtifacts).toHaveBeenCalledWith('doc-1', null);
expect(mocks.deleteDocumentPreviewRows).toHaveBeenCalledWith('doc-1', null);
expect(mocks.deleteDocumentBlob).toHaveBeenCalledWith('doc-1', null);
});
test('restores ownership when cleanup fails', async () => {
const removed = { id: 'doc-1', userId: 'user-1' };
mocks.deleteResults = [[removed]];
mocks.selectResults = [[], []];
mocks.deleteDocumentBlob.mockRejectedValueOnce(new Error('storage unavailable'));
test('returns false and skips cleanup when no row was owned', async () => {
mocks.deleteResults = [[]];
await expect(deleteOwnedDocument({
userId: 'user-1',
documentId: 'doc-1',
namespace: null,
})).rejects.toThrow('storage unavailable');
})).resolves.toBe(false);
expect(mocks.insertValues).toHaveBeenCalledWith(removed);
expect(mocks.deleteDocumentTtsSegmentCache).not.toHaveBeenCalled();
});
test('keeps shared artifacts when a new owner appears before deletion', async () => {
test('still succeeds if TTS cache cleanup fails (best effort)', async () => {
mocks.deleteResults = [[{ id: 'doc-1', userId: 'user-1' }]];
mocks.selectResults = [[], [{ id: 'doc-1' }]];
mocks.deleteDocumentTtsSegmentCache.mockRejectedValueOnce(new Error('storage unavailable'));
await deleteOwnedDocument({
await expect(deleteOwnedDocument({
userId: 'user-1',
documentId: 'doc-1',
namespace: null,
});
expect(mocks.deleteDocumentBlob).not.toHaveBeenCalled();
})).resolves.toBe(true);
});
});

66
tests/unit/reap-orphaned-blobs.vitest.spec.ts Normal file
View file

@ -0,0 +1,66 @@
import { beforeEach, describe, expect, test, vi } from 'vitest';
const mocks = vi.hoisted(() => ({
list: vi.fn(),
del: vi.fn(async () => undefined),
owned: [] as Array<{ id: string }>,
}));
vi.mock('@/lib/server/storage/s3', () => ({ isS3Configured: () => true }));
vi.mock('@/lib/server/documents/blobstore', () => ({
listDocumentSourceBlobs: mocks.list,
deleteDocumentBlob: mocks.del,
}));
vi.mock('@/lib/server/documents/previews-blobstore', () => ({
deleteDocumentPreviewArtifacts: vi.fn(async () => 0),
}));
vi.mock('@/lib/server/documents/previews', () => ({
deleteDocumentPreviewRows: vi.fn(async () => undefined),
}));
vi.mock('@/db', () => ({
db: {
select: () => ({ from: () => ({ where: () => Promise.resolve(mocks.owned) }) }),
},
}));
import { reapOrphanedBlobs } from '../../src/lib/server/tasks/handlers/reap-orphaned-blobs';
const TWO_HOURS = 2 * 60 * 60 * 1000;
beforeEach(() => {
mocks.list.mockReset();
mocks.del.mockReset();
mocks.del.mockResolvedValue(undefined);
mocks.owned = [];
});
describe('reap-orphaned-blobs', () => {
test('reaps only old blobs with no owner', async () => {
const now = Date.now();
mocks.list.mockResolvedValue([
{ id: 'orphan-old', lastModifiedMs: now - TWO_HOURS },
{ id: 'owned-old', lastModifiedMs: now - TWO_HOURS },
{ id: 'orphan-young', lastModifiedMs: now - 60_000 },
]);
mocks.owned = [{ id: 'owned-old' }];
const result = await reapOrphanedBlobs();
expect(mocks.del).toHaveBeenCalledTimes(1);
expect(mocks.del).toHaveBeenCalledWith('orphan-old', null);
expect(result.reaped).toBe(1);
expect(result.scanned).toBe(3);
expect(result.candidates).toBe(2);
});
test('reaps nothing when every old blob still has an owner', async () => {
const now = Date.now();
mocks.list.mockResolvedValue([{ id: 'kept', lastModifiedMs: now - TWO_HOURS }]);
mocks.owned = [{ id: 'kept' }];
const result = await reapOrphanedBlobs();
expect(mocks.del).not.toHaveBeenCalled();
expect(result.reaped).toBe(0);
});
});

137
tests/unit/scheduled-tasks-engine.vitest.spec.ts Normal file
View file

@ -0,0 +1,137 @@
import { beforeEach, describe, expect, test, vi } from 'vitest';
import Database from 'better-sqlite3';
import { drizzle } from 'drizzle-orm/better-sqlite3';
import { eq } from 'drizzle-orm';
import * as sqliteSchema from '../../src/db/schema_sqlite';
// Back the engine with a real in-memory SQLite so the CAS claim, due-detection,
// and nextRunAt arithmetic are exercised against real SQL rather than mocks.
const holder = vi.hoisted(() => ({ db: null as unknown as ReturnType<typeof drizzle> }));
vi.mock('@/db', () => ({
get db() {
return holder.db;
},
}));
// Keep the error-path test from printing the expected failure to the console.
vi.mock('@/lib/server/errors/logging', () => ({ logDegraded: vi.fn() }));
import { runDueTasks } from '../../src/lib/server/tasks/engine';
import type { TaskRegistry } from '../../src/lib/server/tasks/types';
const tasks = sqliteSchema.scheduledTasks;
const CREATE_TABLE = `CREATE TABLE scheduled_tasks (
key text PRIMARY KEY NOT NULL,
enabled integer DEFAULT true NOT NULL,
interval_ms integer NOT NULL,
last_status text DEFAULT 'idle' NOT NULL,
last_run_at integer,
last_duration_ms integer,
last_error text,
last_result_json text,
next_run_at integer,
run_requested integer DEFAULT false NOT NULL,
running_since integer,
updated_at integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL
);`;
const KEY = 'test-task';
async function seedRow(overrides: Partial<typeof tasks.$inferInsert>) {
await holder.db.insert(tasks).values({
key: KEY,
enabled: true,
intervalMs: 1000,
lastStatus: 'idle',
nextRunAt: Date.now() - 1,
runRequested: false,
...overrides,
});
}
async function readRow() {
const rows = await holder.db.select().from(tasks).where(eq(tasks.key, KEY));
return rows[0];
}
function registryWith(run: () => Promise<{ summary?: string } | void>): TaskRegistry {
return { [KEY]: { name: 'Test task', defaultIntervalMs: 1000, run } };
}
beforeEach(() => {
const sqlite = new Database(':memory:');
sqlite.exec(CREATE_TABLE);
holder.db = drizzle(sqlite, { schema: sqliteSchema });
});
describe('scheduled task engine', () => {
test('runs a due task and records success + next run', async () => {
const handler = vi.fn(async () => ({ summary: 'did 3 things' }));
await seedRow({ nextRunAt: Date.now() - 1 });
await runDueTasks({ registry: registryWith(handler) });
expect(handler).toHaveBeenCalledTimes(1);
const row = await readRow();
expect(row.lastStatus).toBe('ok');
expect(row.lastRunAt).not.toBeNull();
expect(row.lastResultJson).toBe('did 3 things');
expect(row.runningSince).toBeNull();
expect(Number(row.nextRunAt)).toBeGreaterThan(Date.now());
});
test('does not run a task that is not yet due', async () => {
const handler = vi.fn(async () => undefined);
await seedRow({ nextRunAt: Date.now() + 60_000, lastStatus: 'idle' });
await runDueTasks({ registry: registryWith(handler) });
expect(handler).not.toHaveBeenCalled();
expect((await readRow()).lastStatus).toBe('idle');
});
test('runs when a manual run is requested even if not due', async () => {
const handler = vi.fn(async () => ({ summary: 'manual' }));
await seedRow({ nextRunAt: Date.now() + 60_000, runRequested: true });
await runDueTasks({ registry: registryWith(handler) });
expect(handler).toHaveBeenCalledTimes(1);
const row = await readRow();
expect(row.lastStatus).toBe('ok');
expect(row.runRequested).toBe(false);
});
test('records errors without throwing', async () => {
const handler = vi.fn(async () => {
throw new Error('boom');
});
await seedRow({ nextRunAt: Date.now() - 1 });
await expect(runDueTasks({ registry: registryWith(handler) })).resolves.toBeUndefined();
const row = await readRow();
expect(row.lastStatus).toBe('error');
expect(row.lastError).toContain('boom');
});
test('single-flight: skips a task already marked running', async () => {
const handler = vi.fn(async () => undefined);
await seedRow({ nextRunAt: Date.now() - 1, lastStatus: 'running', runningSince: Date.now() });
await runDueTasks({ registry: registryWith(handler) });
expect(handler).not.toHaveBeenCalled();
expect((await readRow()).lastStatus).toBe('running');
});
test('runs a requested task even when disabled', async () => {
const handler = vi.fn(async () => undefined);
await seedRow({ nextRunAt: Date.now() + 60_000, enabled: false, runRequested: true });
await runDueTasks({ registry: registryWith(handler) });
expect(handler).toHaveBeenCalledTimes(1);
});
});

88
tests/unit/user-data-cleanup.vitest.spec.ts
View file

@ -1,11 +1,8 @@
import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
import { beforeEach, describe, expect, test, vi } from 'vitest';
const mocks = vi.hoisted(() => ({
selectResults: [] as unknown[][],
documentDeleteResults: [] as unknown[][],
deleteWhere: vi.fn(async () => undefined),
insertValues: vi.fn(() => ({ onConflictDoNothing: vi.fn(async () => undefined) })),
execute: vi.fn(async () => undefined),
deleteDocumentBlob: vi.fn(async () => undefined),
deleteDocumentPrefix: vi.fn(async () => 0),
deleteDocumentPreviewArtifacts: vi.fn(async () => 0),
@ -15,14 +12,7 @@ const mocks = vi.hoisted(() => ({
}));
function resultBuilder(result: unknown[]) {
const limitedResult = {
all: () => result,
then: (resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) =>
Promise.resolve(result).then(resolve, reject),
};
return {
all: () => result,
limit: () => limitedResult,
then: (resolve: (value: unknown[]) => unknown, reject: (error: unknown) => unknown) =>
Promise.resolve(result).then(resolve, reject),
};
@ -41,15 +31,9 @@ vi.mock('@/db', () => {
return {
then: promise.then.bind(promise),
catch: promise.catch.bind(promise),
returning: () => resultBuilder(mocks.documentDeleteResults.shift() ?? []),
};
}),
})),
insert: vi.fn(() => ({
values: mocks.insertValues,
})),
execute: mocks.execute,
transaction: vi.fn((callback: (tx: unknown) => unknown) => callback(database)),
};
return { db: database };
});
@ -94,21 +78,14 @@ vi.mock('@/lib/server/errors/logging', () => ({
import { deleteUserStorageData } from '../../src/lib/server/user/data-cleanup';
describe('user data cleanup', () => {
afterEach(() => {
vi.unstubAllEnvs();
});
beforeEach(() => {
mocks.selectResults = [];
mocks.documentDeleteResults = [];
for (const mock of Object.values(mocks)) {
if (typeof mock === 'function' && 'mockReset' in mock) {
mock.mockReset();
}
}
mocks.deleteWhere.mockResolvedValue(undefined);
mocks.insertValues.mockReturnValue({ onConflictDoNothing: vi.fn(async () => undefined) });
mocks.execute.mockResolvedValue(undefined);
mocks.deleteDocumentBlob.mockResolvedValue(undefined);
mocks.deleteDocumentPrefix.mockResolvedValue(0);
mocks.deleteDocumentPreviewArtifacts.mockResolvedValue(0);
@ -117,24 +94,21 @@ describe('user data cleanup', () => {
mocks.deleteTtsSegmentPrefix.mockResolvedValue(0);
});
test('keeps shared document blobs and previews', async () => {
mocks.selectResults = [
[{ id: 'shared-doc' }],
[{ id: 'shared-doc' }],
[],
];
mocks.documentDeleteResults = [[{ id: 'shared-doc' }]];
test('defers document blobs/previews to the reaper on the canonical pass', async () => {
mocks.selectResults = [[]]; // no audiobooks
await deleteUserStorageData('user-1', null);
// Shared document storage is reclaimed by the reap-orphaned-blobs task, not here.
expect(mocks.deleteDocumentBlob).not.toHaveBeenCalled();
expect(mocks.deleteDocumentPreviewArtifacts).not.toHaveBeenCalled();
expect(mocks.deleteDocumentPreviewRows).not.toHaveBeenCalled();
expect(mocks.deleteWhere).toHaveBeenCalledTimes(4);
// Only the three non-cascading DB row deletes (tts usage, job events, verification).
expect(mocks.deleteWhere).toHaveBeenCalledTimes(3);
});
test('blocks database cleanup when storage cleanup fails', async () => {
mocks.selectResults = [[], []];
mocks.selectResults = [[]]; // no audiobooks
mocks.deleteDocumentPrefix.mockRejectedValueOnce(new Error('storage unavailable'));
await expect(deleteUserStorageData('user-1', null)).rejects.toThrow(
@ -143,50 +117,18 @@ describe('user data cleanup', () => {
expect(mocks.deleteWhere).not.toHaveBeenCalled();
});
test('restores document ownership when document storage cleanup fails', async () => {
test('deletes namespaced document storage inline and skips global DB rows', async () => {
mocks.selectResults = [
[{ id: 'doc-1' }],
[],
[],
[{ id: 'doc-1' }], // userDocs (namespaced pass)
[], // audiobooks
];
mocks.documentDeleteResults = [[{ id: 'doc-1' }]];
mocks.deleteDocumentBlob.mockRejectedValueOnce(new Error('storage unavailable'));
await expect(deleteUserStorageData('user-1', null)).rejects.toThrow(
'User storage cleanup failed',
);
expect(mocks.insertValues).toHaveBeenCalledWith([{ id: 'doc-1' }]);
});
test('does not delete global preview rows during namespaced cleanup', async () => {
mocks.selectResults = [
[{ id: 'doc-1' }],
[],
[],
];
mocks.documentDeleteResults = [[{ id: 'doc-1' }]];
await deleteUserStorageData('user-1', 'test-ns');
expect(mocks.deleteDocumentPreviewRows).not.toHaveBeenCalled();
expect(mocks.insertValues).toHaveBeenCalledWith([{ id: 'doc-1' }]);
});
test('acquires a Postgres advisory lock before the ownership decision', async () => {
vi.stubEnv('POSTGRES_URL', 'postgres://test');
mocks.selectResults = [
[{ id: 'doc-1' }],
[{ id: 'doc-1' }],
[],
[],
];
mocks.documentDeleteResults = [[{ id: 'doc-1' }]];
await deleteUserStorageData('user-1', null);
// The mutation lock serializes the read-modify-write via a transaction-scoped
// advisory lock instead of SELECT ... FOR UPDATE.
expect(mocks.execute).toHaveBeenCalled();
expect(mocks.deleteDocumentBlob).toHaveBeenCalledWith('doc-1', 'test-ns');
expect(mocks.deleteDocumentPreviewArtifacts).toHaveBeenCalledWith('doc-1', 'test-ns');
expect(mocks.deleteDocumentPreviewRows).toHaveBeenCalledWith('doc-1', 'test-ns');
// Global DB rows are only removed on the canonical pass.
expect(mocks.deleteWhere).not.toHaveBeenCalled();
});
});

8
vercel.json Normal file
View file

@ -0,0 +1,8 @@
{
"crons": [
{
"path": "/api/admin/tasks/tick",
"schedule": "0 * * * *"
}
]
}