diff --git a/drizzle/postgres/0011_scheduled-tasks.sql b/drizzle/postgres/0011_scheduled-tasks.sql index 8eef94b..cb05f2f 100644 --- a/drizzle/postgres/0011_scheduled-tasks.sql +++ b/drizzle/postgres/0011_scheduled-tasks.sql @@ -10,5 +10,6 @@ CREATE TABLE "scheduled_tasks" ( "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 + "updated_at" bigint DEFAULT (extract(epoch from now()) * 1000)::bigint NOT NULL, + CONSTRAINT "scheduled_tasks_interval_ms_positive" CHECK ("scheduled_tasks"."interval_ms" > 0) ); diff --git a/drizzle/postgres/meta/0011_snapshot.json b/drizzle/postgres/meta/0011_snapshot.json index 33c6003..443520f 100644 --- a/drizzle/postgres/meta/0011_snapshot.json +++ b/drizzle/postgres/meta/0011_snapshot.json @@ -1,5 +1,5 @@ { - "id": "195ce074-f013-420d-9238-30c32f38365e", + "id": "afe1d934-9b20-4d50-82fc-aeebd49fe43f", "prevId": "1099cbe6-0c2e-4ea8-ae3c-6883c931169d", "version": "7", "dialect": "postgresql", @@ -780,7 +780,12 @@ "compositePrimaryKeys": {}, "uniqueConstraints": {}, "policies": {}, - "checkConstraints": {}, + "checkConstraints": { + "scheduled_tasks_interval_ms_positive": { + "name": "scheduled_tasks_interval_ms_positive", + "value": "\"scheduled_tasks\".\"interval_ms\" > 0" + } + }, "isRLSEnabled": false }, "public.tts_segment_entries": { diff --git a/drizzle/postgres/meta/_journal.json b/drizzle/postgres/meta/_journal.json index 488ccd2..5731fec 100644 --- a/drizzle/postgres/meta/_journal.json +++ b/drizzle/postgres/meta/_journal.json @@ -82,7 +82,7 @@ { "idx": 11, "version": "7", - "when": 1780825886154, + "when": 1780847571383, "tag": "0011_scheduled-tasks", "breakpoints": true } diff --git a/drizzle/sqlite/0011_scheduled-tasks.sql b/drizzle/sqlite/0011_scheduled-tasks.sql index af6466c..cca23bd 100644 --- a/drizzle/sqlite/0011_scheduled-tasks.sql +++ b/drizzle/sqlite/0011_scheduled-tasks.sql @@ -10,5 +10,6 @@ CREATE TABLE `scheduled_tasks` ( `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 + `updated_at` integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL, + CONSTRAINT "scheduled_tasks_interval_ms_positive" CHECK("scheduled_tasks"."interval_ms" > 0) ); diff --git a/drizzle/sqlite/meta/0011_snapshot.json b/drizzle/sqlite/meta/0011_snapshot.json index 66b1549..45d8f1f 100644 --- a/drizzle/sqlite/meta/0011_snapshot.json +++ b/drizzle/sqlite/meta/0011_snapshot.json @@ -1,7 +1,7 @@ { "version": "6", "dialect": "sqlite", - "id": "504118c7-02f7-4fc5-806d-4a369e415173", + "id": "4c4a203e-9b2b-4eb2-b31c-2613a7c1d377", "prevId": "a32ae7d4-c074-44f0-9e90-2473949376ef", "tables": { "admin_providers": { @@ -791,7 +791,12 @@ "foreignKeys": {}, "compositePrimaryKeys": {}, "uniqueConstraints": {}, - "checkConstraints": {} + "checkConstraints": { + "scheduled_tasks_interval_ms_positive": { + "name": "scheduled_tasks_interval_ms_positive", + "value": "\"scheduled_tasks\".\"interval_ms\" > 0" + } + } }, "tts_segment_entries": { "name": "tts_segment_entries", diff --git a/drizzle/sqlite/meta/_journal.json b/drizzle/sqlite/meta/_journal.json index 5546859..513ea7f 100644 --- a/drizzle/sqlite/meta/_journal.json +++ b/drizzle/sqlite/meta/_journal.json @@ -82,7 +82,7 @@ { "idx": 11, "version": "6", - "when": 1780825885684, + "when": 1780847571062, "tag": "0011_scheduled-tasks", "breakpoints": true } diff --git a/src/app/api/admin/tasks/[key]/route.ts b/src/app/api/admin/tasks/[key]/route.ts index cb302f9..4ac2460 100644 --- a/src/app/api/admin/tasks/[key]/route.ts +++ b/src/app/api/admin/tasks/[key]/route.ts @@ -13,7 +13,7 @@ export async function PATCH( if (ctx instanceof Response) return ctx; const { key } = await params; - if (!(key in TASK_REGISTRY)) { + if (!Object.hasOwn(TASK_REGISTRY, key)) { return NextResponse.json({ error: 'Unknown task' }, { status: 404 }); } @@ -27,11 +27,26 @@ export async function PATCH( return NextResponse.json({ error: 'Expected JSON object' }, { status: 400 }); } - const { enabled, intervalMs } = body as { enabled?: unknown; intervalMs?: unknown }; + const payload = 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); + if (Object.hasOwn(payload, 'enabled')) { + if (typeof payload.enabled !== 'boolean') { + return NextResponse.json({ error: 'enabled must be a boolean' }, { status: 400 }); + } + patch.enabled = payload.enabled; + } + if (Object.hasOwn(payload, 'intervalMs')) { + if ( + typeof payload.intervalMs !== 'number' + || !Number.isFinite(payload.intervalMs) + || payload.intervalMs <= 0 + ) { + return NextResponse.json({ error: 'intervalMs must be a finite positive number' }, { status: 400 }); + } + patch.intervalMs = Math.max(1, Math.floor(payload.intervalMs)); + } + if (Object.keys(patch).length === 0) { + return NextResponse.json({ error: 'Expected enabled or intervalMs' }, { status: 400 }); } await updateTask(key, patch); diff --git a/src/app/api/admin/tasks/[key]/run/route.ts b/src/app/api/admin/tasks/[key]/run/route.ts index 5c37fc2..8b67a8a 100644 --- a/src/app/api/admin/tasks/[key]/run/route.ts +++ b/src/app/api/admin/tasks/[key]/run/route.ts @@ -13,7 +13,7 @@ export async function POST( if (ctx instanceof Response) return ctx; const { key } = await params; - if (!(key in TASK_REGISTRY)) { + if (!Object.hasOwn(TASK_REGISTRY, key)) { return NextResponse.json({ error: 'Unknown task' }, { status: 404 }); } diff --git a/src/components/admin/AdminTasksPanel.tsx b/src/components/admin/AdminTasksPanel.tsx index 3affd25..e5a232c 100644 --- a/src/components/admin/AdminTasksPanel.tsx +++ b/src/components/admin/AdminTasksPanel.tsx @@ -141,15 +141,16 @@ function TaskRow({ onSaveInterval: (intervalMs: number) => void; onRun: () => void; }) { - const [minutes, setMinutes] = useState(String(Math.round(task.intervalMs / 60000))); + const [minutes, setMinutes] = useState(String(task.intervalMs / 60000)); useEffect(() => { - setMinutes(String(Math.round(task.intervalMs / 60000))); + setMinutes(String(task.intervalMs / 60000)); }, [task.intervalMs]); - const parsedMinutes = Number(minutes); + const parsedMinutes = parseFloat(minutes); + const newIntervalMs = parsedMinutes * 60000; const intervalDirty = - Number.isFinite(parsedMinutes) && parsedMinutes > 0 && parsedMinutes * 60000 !== task.intervalMs; + Number.isFinite(parsedMinutes) && parsedMinutes > 0 && newIntervalMs !== task.intervalMs; return (
@@ -177,7 +178,8 @@ function TaskRow({ Every onSaveInterval(Math.floor(parsedMinutes) * 60000)} + onClick={() => onSaveInterval(newIntervalMs)} disabled={busy} > Save diff --git a/src/db/schema_postgres.ts b/src/db/schema_postgres.ts index 5171903..d771d87 100644 --- a/src/db/schema_postgres.ts +++ b/src/db/schema_postgres.ts @@ -1,5 +1,5 @@ import { sql } from 'drizzle-orm'; -import { pgTable, text, integer, real, date, bigint, boolean, primaryKey, index, jsonb, foreignKey } from 'drizzle-orm/pg-core'; +import { pgTable, text, integer, real, date, bigint, boolean, primaryKey, index, jsonb, foreignKey, check } from 'drizzle-orm/pg-core'; import { user } from './schema_auth_postgres'; const PG_NOW_MS = sql`(extract(epoch from now()) * 1000)::bigint`; @@ -223,7 +223,9 @@ export const scheduledTasks = pgTable('scheduled_tasks', { runRequested: boolean('run_requested').notNull().default(false), runningSince: bigint('running_since', { mode: 'number' }), updatedAt: bigint('updated_at', { mode: 'number' }).notNull().default(PG_NOW_MS), -}); +}, (table) => [ + check('scheduled_tasks_interval_ms_positive', sql`${table.intervalMs} > 0`), +]); export const ttsSegmentVariants = pgTable('tts_segment_variants', { segmentId: text('segment_id').notNull(), diff --git a/src/db/schema_sqlite.ts b/src/db/schema_sqlite.ts index 78f0588..2ae1c54 100644 --- a/src/db/schema_sqlite.ts +++ b/src/db/schema_sqlite.ts @@ -1,4 +1,4 @@ -import { sqliteTable, text, integer, real, primaryKey, index, foreignKey } from 'drizzle-orm/sqlite-core'; +import { sqliteTable, text, integer, real, primaryKey, index, foreignKey, check } from 'drizzle-orm/sqlite-core'; import { sql } from 'drizzle-orm'; import { user } from './schema_auth_sqlite'; @@ -223,7 +223,9 @@ export const scheduledTasks = sqliteTable('scheduled_tasks', { runRequested: integer('run_requested', { mode: 'boolean' }).notNull().default(false), runningSince: integer('running_since'), updatedAt: integer('updated_at').notNull().default(SQLITE_NOW_MS), -}); +}, (table) => [ + check('scheduled_tasks_interval_ms_positive', sql`${table.intervalMs} > 0`), +]); export const ttsSegmentVariants = sqliteTable('tts_segment_variants', { segmentId: text('segment_id').notNull(), diff --git a/src/lib/server/documents/blobstore.ts b/src/lib/server/documents/blobstore.ts index 438f3b8..61ceb06 100644 --- a/src/lib/server/documents/blobstore.ts +++ b/src/lib/server/documents/blobstore.ts @@ -654,7 +654,7 @@ export async function deleteAllExpiredTempDocumentUploads( const nsSegment = ns ? `ns/${ns}/` : ''; const prefix = `${cfg.prefix}/document_uploads_temp_v1/${nsSegment}`; let continuationToken: string | undefined; - const keys: string[] = []; + let deleted = 0; do { const listRes = await client.send( @@ -665,33 +665,35 @@ export async function deleteAllExpiredTempDocumentUploads( }), ); + const batch: string[] = []; for (const item of listRes.Contents ?? []) { const key = item.Key; const lastModified = item.LastModified?.getTime() ?? 0; if (!key || lastModified <= 0 || lastModified >= olderThanMs) continue; - keys.push(key); + batch.push(key); + } + + if (batch.length > 0) { + const deleteRes = await client.send( + new DeleteObjectsCommand({ + Bucket: cfg.bucket, + Delete: { + Objects: batch.map((Key) => ({ Key })), + Quiet: true, + }, + }), + ); + if (deleteRes.Errors?.length) { + throw new Error( + `Failed to delete temporary uploads from bucket "${cfg.bucket}" ` + + `(keys: ${JSON.stringify(batch)}, errors: ${JSON.stringify(deleteRes.Errors)})`, + ); + } + deleted += batch.length; } continuationToken = listRes.IsTruncated ? listRes.NextContinuationToken : undefined; } while (continuationToken); - if (keys.length === 0) return 0; - - let deleted = 0; - for (let i = 0; i < keys.length; i += 1000) { - const batch = keys.slice(i, i + 1000); - const deleteRes = await client.send( - new DeleteObjectsCommand({ - Bucket: cfg.bucket, - Delete: { - Objects: batch.map((Key) => ({ Key })), - Quiet: true, - }, - }), - ); - deleted += deleteRes.Deleted?.length ?? batch.length - (deleteRes.Errors?.length ?? 0); - } - return deleted; } - diff --git a/src/lib/server/tasks/engine.ts b/src/lib/server/tasks/engine.ts index ac198ee..aecc113 100644 --- a/src/lib/server/tasks/engine.ts +++ b/src/lib/server/tasks/engine.ts @@ -69,7 +69,6 @@ async function finishTask( // 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)); @@ -136,12 +135,29 @@ export async function updateTask( key: string, patch: { enabled?: boolean; intervalMs?: number }, ): Promise { - const set: Record = { updatedAt: Date.now() }; + const def = Object.hasOwn(TASK_REGISTRY, key) ? TASK_REGISTRY[key] : undefined; + if (!def) return; + + const now = Date.now(); + const set: Record = { updatedAt: 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); + set.intervalMs = Math.max(1, Math.floor(patch.intervalMs)); } - await db.update(scheduledTasks).set(set).where(eq(scheduledTasks.key, key)); + await db + .insert(scheduledTasks) + .values({ + key, + enabled: typeof set.enabled === 'boolean' ? set.enabled : true, + intervalMs: typeof set.intervalMs === 'number' ? set.intervalMs : def.defaultIntervalMs, + lastStatus: 'idle', + nextRunAt: now, + updatedAt: now, + }) + .onConflictDoUpdate({ + target: scheduledTasks.key, + set, + }); } export type TaskView = { diff --git a/src/lib/server/user/claim-data.ts b/src/lib/server/user/claim-data.ts index 28167fd..47a52c9 100644 --- a/src/lib/server/user/claim-data.ts +++ b/src/lib/server/user/claim-data.ts @@ -499,15 +499,28 @@ export async function transferUserDocumentSettings( .select() .from(documentSettings) .where(eq(documentSettings.userId, fromUserId)); - for (const row of rows) { - const [existing] = await database - .select({ clientUpdatedAtMs: documentSettings.clientUpdatedAtMs }) + const documentIds = rows.map((row: { documentId: string }) => row.documentId); + const existingRows = documentIds.length > 0 + ? await database + .select({ + documentId: documentSettings.documentId, + clientUpdatedAtMs: documentSettings.clientUpdatedAtMs, + }) .from(documentSettings) .where(and( eq(documentSettings.userId, toUserId), - eq(documentSettings.documentId, row.documentId), + inArray(documentSettings.documentId, documentIds), )) - .limit(1); + : []; + const existingByDocumentId = new Map( + existingRows.map((row: { documentId: string; clientUpdatedAtMs: number | null }) => [ + row.documentId, + row, + ] as const), + ); + + for (const row of rows) { + const existing = existingByDocumentId.get(row.documentId); if (existing && Number(existing.clientUpdatedAtMs ?? 0) >= Number(row.clientUpdatedAtMs ?? 0)) { continue; } diff --git a/tests/unit/scheduled-tasks-engine.vitest.spec.ts b/tests/unit/scheduled-tasks-engine.vitest.spec.ts index c7587de..737fdf8 100644 --- a/tests/unit/scheduled-tasks-engine.vitest.spec.ts +++ b/tests/unit/scheduled-tasks-engine.vitest.spec.ts @@ -16,7 +16,7 @@ vi.mock('@/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 { runDueTasks, updateTask } from '../../src/lib/server/tasks/engine'; import type { TaskRegistry } from '../../src/lib/server/tasks/types'; const tasks = sqliteSchema.scheduledTasks; @@ -33,7 +33,8 @@ const CREATE_TABLE = `CREATE TABLE scheduled_tasks ( 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 + updated_at integer DEFAULT (cast(unixepoch('subsecond') * 1000 as integer)) NOT NULL, + CONSTRAINT scheduled_tasks_interval_ms_positive CHECK(interval_ms > 0) );`; const KEY = 'test-task'; @@ -134,4 +135,37 @@ describe('scheduled task engine', () => { expect(handler).toHaveBeenCalledTimes(1); }); + + test('preserves a manual rerun requested while a task is running', async () => { + const handler = vi.fn(async () => { + await holder.db + .update(tasks) + .set({ runRequested: true }) + .where(eq(tasks.key, KEY)); + }); + await seedRow({ nextRunAt: Date.now() - 1 }); + + await runDueTasks({ registry: registryWith(handler) }); + + expect(handler).toHaveBeenCalledTimes(1); + expect((await readRow()).runRequested).toBe(true); + }); + + test('rejects non-positive intervals at the database boundary', async () => { + await expect(seedRow({ intervalMs: 0 })).rejects.toThrow(); + }); + + test('updates a registered task even when its row has not been seeded', async () => { + await updateTask('prune-job-events', { enabled: false, intervalMs: 12_345 }); + + const [row] = await holder.db + .select() + .from(tasks) + .where(eq(tasks.key, 'prune-job-events')); + expect(row).toEqual(expect.objectContaining({ + enabled: false, + intervalMs: 12_345, + lastStatus: 'idle', + })); + }); }); diff --git a/tests/unit/storage-prefix-cleanup.vitest.spec.ts b/tests/unit/storage-prefix-cleanup.vitest.spec.ts index 6267527..7afb7de 100644 --- a/tests/unit/storage-prefix-cleanup.vitest.spec.ts +++ b/tests/unit/storage-prefix-cleanup.vitest.spec.ts @@ -11,7 +11,11 @@ vi.mock('@/lib/server/storage/s3', () => ({ })); import { deleteAudiobookPrefix } from '../../src/lib/server/audiobooks/blobstore'; -import { deleteDocumentBlob, deleteDocumentPrefix } from '../../src/lib/server/documents/blobstore'; +import { + deleteAllExpiredTempDocumentUploads, + deleteDocumentBlob, + deleteDocumentPrefix, +} from '../../src/lib/server/documents/blobstore'; describe('storage prefix cleanup', () => { beforeEach(() => { @@ -85,4 +89,44 @@ describe('storage prefix cleanup', () => { && command.input.Key === `openreader-test/documents_v1/${documentId}`); expect(sourceDelete).toBeUndefined(); }); + + test('deletes expired temporary uploads page by page', async () => { + const old = new Date(Date.now() - 60_000); + mocks.send + .mockResolvedValueOnce({ + Contents: [{ Key: 'temp/a', LastModified: old }], + IsTruncated: true, + NextContinuationToken: 'next', + }) + .mockResolvedValueOnce({}) + .mockResolvedValueOnce({ + Contents: [{ Key: 'temp/b', LastModified: old }], + IsTruncated: false, + }) + .mockResolvedValueOnce({}); + + await expect(deleteAllExpiredTempDocumentUploads(null, Date.now())).resolves.toBe(2); + + expect(mocks.send.mock.calls.map(([command]) => command.constructor.name)).toEqual([ + 'ListObjectsV2Command', + 'DeleteObjectsCommand', + 'ListObjectsV2Command', + 'DeleteObjectsCommand', + ]); + }); + + test('temporary upload cleanup fails on per-object storage errors', async () => { + const old = new Date(Date.now() - 60_000); + mocks.send + .mockResolvedValueOnce({ + Contents: [{ Key: 'temp/a', LastModified: old }], + IsTruncated: false, + }) + .mockResolvedValueOnce({ + Errors: [{ Key: 'temp/a', Code: 'AccessDenied' }], + }); + + await expect(deleteAllExpiredTempDocumentUploads(null, Date.now())) + .rejects.toThrow('test-bucket'); + }); });