diff --git a/changes/stats-delete-responsiveness.md b/changes/stats-delete-responsiveness.md new file mode 100644 index 00000000..5bd00ef8 --- /dev/null +++ b/changes/stats-delete-responsiveness.md @@ -0,0 +1,4 @@ +type: fixed +area: stats + +- Kept the stats page and active video player responsive during deletes, and batched concurrent session, episode, and library deletes into one transaction and summary rebuild. diff --git a/docs/architecture/domains.md b/docs/architecture/domains.md index 789a4142..e34a0ab3 100644 --- a/docs/architecture/domains.md +++ b/docs/architecture/domains.md @@ -25,6 +25,7 @@ Read when: you need to find the owner module for a behavior or test surface - Anki workflow: `src/anki-integration/`, `src/core/services/anki-jimaku*.ts` - Immersion tracking: `src/core/services/immersion-tracker/` Includes stats storage/query schema such as `imm_videos`, `imm_media_art`, and `imm_youtube_videos` for per-video and YouTube-specific library metadata. + `delete-maintenance-scheduler.ts` coalesces and serializes stats deletes; expensive deletion and summary rebuilds run in `delete-maintenance-worker-thread.ts` while the tracker queues playback writes. Each batch uses one transaction, lexical update, rollup refresh, and lifetime rebuild. - AniList tracking + character dictionary: `src/core/services/anilist/`, `src/main/runtime/composers/anilist-*`, `src/main/character-dictionary-runtime.ts`, `src/main/character-dictionary-runtime/` - Jellyfin integration: `src/core/services/jellyfin*.ts`, `src/main/runtime/composers/jellyfin-*` - Window trackers: `src/window-trackers/` diff --git a/src/core/services/immersion-tracker-service.test.ts b/src/core/services/immersion-tracker-service.test.ts index f7dbf77d..90e1b909 100644 --- a/src/core/services/immersion-tracker-service.test.ts +++ b/src/core/services/immersion-tracker-service.test.ts @@ -1414,6 +1414,353 @@ test('deleteSession ignores the currently active session and keeps new writes fl } }); +test('deleteSession yields the main event loop while delete maintenance is pending', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + const deleteGate: { release?: () => void } = {}; + let deleteRunnerCalled = false; + let bufferedWritesAtDeleteStart = -1; + + try { + const Ctor = await loadTrackerCtor(); + const createdTracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async () => { + deleteRunnerCalled = true; + bufferedWritesAtDeleteStart = (tracker as unknown as { queue: unknown[] }).queue.length; + await new Promise((resolve) => { + deleteGate.release = resolve; + }); + }, + }, + ); + tracker = createdTracker; + createdTracker.handleMediaChange('/tmp/delete-yield-first.mkv', 'Delete Yield First'); + createdTracker.handleMediaChange('/tmp/delete-yield-active.mkv', 'Delete Yield Active'); + + const privateApi = createdTracker as unknown as { + db: DatabaseSync; + queue: unknown[]; + flushNow: () => void; + }; + const sessionId = ( + privateApi.db + .prepare( + `SELECT session_id AS sessionId + FROM imm_sessions + WHERE ended_at_ms IS NOT NULL + ORDER BY session_id + LIMIT 1`, + ) + .get() as { sessionId: number } | null + )?.sessionId; + assert.ok(sessionId); + + const deletePromise = createdTracker.deleteSession(sessionId); + let timerAdvanced = false; + setTimeout(() => { + timerAdvanced = true; + }, 0); + + await waitForCondition(() => deleteRunnerCalled); + assert.equal(deleteRunnerCalled, true, 'delete should be dispatched to the maintenance runner'); + assert.equal( + bufferedWritesAtDeleteStart, + 0, + 'writes buffered before delete should flush first', + ); + await waitForCondition(() => timerAdvanced); + + createdTracker.recordSubtitleLine('queued during delete', 0, 1); + privateApi.flushNow(); + assert.ok(privateApi.queue.length > 0, 'tracking writes should wait for delete maintenance'); + + assert.ok(deleteGate.release); + deleteGate.release(); + await deletePromise; + } finally { + deleteGate.release?.(); + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + +test('delete maintenance flushes the entire write queue before locking writes', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + const deleteGate: { release?: () => void } = {}; + let queuedWritesAtDeleteStart = -1; + let writeLockedAtDeleteStart = false; + + try { + const Ctor = await loadTrackerCtor(); + tracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async () => { + const privateApi = tracker as unknown as { + queue: unknown[]; + writeLock: { locked: boolean }; + }; + queuedWritesAtDeleteStart = privateApi.queue.length; + writeLockedAtDeleteStart = privateApi.writeLock.locked; + await new Promise((resolve) => { + deleteGate.release = resolve; + }); + }, + }, + ); + + const privateApi = tracker as unknown as { + batchSize: number; + flushNow: () => void; + queue: unknown[]; + }; + privateApi.batchSize = 1; + privateApi.queue.push({}, {}, {}); + privateApi.flushNow = () => { + privateApi.queue.shift(); + }; + + const deletePromise = tracker.deleteSession(101); + await waitForCondition(() => deleteGate.release !== undefined); + + assert.equal(queuedWritesAtDeleteStart, 0); + assert.equal(writeLockedAtDeleteStart, true); + + deleteGate.release?.(); + await deletePromise; + } finally { + deleteGate.release?.(); + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + +test('delete maintenance tasks stay serialized under concurrent requests', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + const releases: Array<() => void> = []; + let activeTasks = 0; + let maxActiveTasks = 0; + + try { + const Ctor = await loadTrackerCtor(); + tracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async () => { + activeTasks += 1; + maxActiveTasks = Math.max(maxActiveTasks, activeTasks); + await new Promise((resolve) => { + releases.push(resolve); + }); + activeTasks -= 1; + }, + }, + ); + + const firstDelete = tracker.deleteSession(101); + await waitForCondition(() => releases.length === 1); + assert.equal(maxActiveTasks, 1); + + const secondDelete = tracker.deleteSession(102); + + releases[0]?.(); + await waitForCondition(() => releases.length === 2); + assert.equal(maxActiveTasks, 1); + + releases[1]?.(); + await Promise.all([firstDelete, secondDelete]); + } finally { + for (const release of releases) release(); + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + +test('concurrent delete requests share one maintenance worker batch', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + const tasks: unknown[] = []; + + try { + const Ctor = await loadTrackerCtor(); + tracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async (_path, task) => { + tasks.push(task); + }, + }, + ); + + const firstDelete = tracker.deleteSession(201); + const secondDelete = tracker.deleteSessions([202, 203]); + const thirdDelete = tracker.deleteVideo(204); + await Promise.all([firstDelete, secondDelete, thirdDelete]); + + assert.equal(tasks.length, 1, 'concurrent deletes should use one maintenance pass'); + assert.deepEqual(tasks[0], { + kind: 'batch', + tasks: [ + { kind: 'session', sessionId: 201 }, + { kind: 'sessions', sessionIds: [202, 203] }, + { kind: 'video', videoId: 204 }, + ], + }); + } finally { + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + +test('destroy rejects delete requests waiting behind active maintenance', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + let releaseFirstTask: () => void = () => {}; + + try { + const Ctor = await loadTrackerCtor(); + let markFirstTaskStarted: () => void = () => {}; + const firstTaskStarted = new Promise((resolve) => { + markFirstTaskStarted = resolve; + }); + tracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async () => { + markFirstTaskStarted(); + await new Promise((resolve) => { + releaseFirstTask = resolve; + }); + }, + }, + ); + + const firstDelete = tracker.deleteSession(301); + await firstTaskStarted; + const queuedDelete = tracker.deleteSession(302); + tracker.destroy(); + + const queuedOutcome = await Promise.race([ + queuedDelete.then( + () => 'resolved', + (error: unknown) => + error instanceof Error && /shutting down/.test(error.message) + ? 'rejected' + : 'wrong-error', + ), + new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), 25)), + ]); + assert.equal(queuedOutcome, 'rejected'); + releaseFirstTask(); + await firstDelete; + } finally { + releaseFirstTask(); + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + +test('delete requested after destroy rejects without running maintenance', async () => { + const dbPath = makeDbPath(); + let maintenanceCalls = 0; + const Ctor = await loadTrackerCtor(); + const tracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async () => { + maintenanceCalls += 1; + }, + }, + ); + + tracker.destroy(); + + await assert.rejects(tracker.deleteSession(303), /shutting down/); + assert.equal(maintenanceCalls, 0); + cleanupDbPath(dbPath); +}); + +test('deleteSessions skips maintenance when no sessions are deletable', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + const tasks: unknown[] = []; + + try { + const Ctor = await loadTrackerCtor(); + tracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async (_path, task) => { + tasks.push(task); + }, + }, + ); + + await tracker.deleteSessions([]); + + assert.deepEqual(tasks, []); + } finally { + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + +test('queued video delete is skipped when that video becomes active before dispatch', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + const tasks: Array<{ kind: string }> = []; + let releaseFirstTask: () => void = () => {}; + + try { + const Ctor = await loadTrackerCtor(); + const createdTracker = new Ctor( + { dbPath }, + { + runDeleteMaintenanceTask: async (_path, task) => { + tasks.push(task); + if (tasks.length === 1) { + await new Promise((resolve) => { + releaseFirstTask = resolve; + }); + } + }, + }, + ); + tracker = createdTracker; + createdTracker.handleMediaChange('/tmp/delete-race-target.mkv', 'Delete Race Target'); + createdTracker.handleMediaChange('/tmp/delete-race-other.mkv', 'Delete Race Other'); + + const privateApi = createdTracker as unknown as { db: DatabaseSync }; + const targetVideoId = ( + privateApi.db + .prepare(`SELECT video_id AS videoId FROM imm_videos WHERE video_key LIKE '%target.mkv'`) + .get() as { videoId: number } | null + )?.videoId; + assert.ok(targetVideoId); + + const firstDelete = createdTracker.deleteSession(999_001); + await waitForCondition(() => tasks.length === 1); + + const queuedVideoDelete = createdTracker.deleteVideo(targetVideoId); + createdTracker.handleMediaChange('/tmp/delete-race-target.mkv', 'Delete Race Target'); + releaseFirstTask(); + await Promise.all([firstDelete, queuedVideoDelete]); + + assert.deepEqual( + tasks.map((task) => task.kind), + ['session'], + ); + } finally { + releaseFirstTask(); + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + test('deleteVideo ignores the currently active video and keeps new writes flushable', async () => { const dbPath = makeDbPath(); let tracker: ImmersionTrackerService | null = null; diff --git a/src/core/services/immersion-tracker-service.ts b/src/core/services/immersion-tracker-service.ts index cbbe7c73..865006f5 100644 --- a/src/core/services/immersion-tracker-service.ts +++ b/src/core/services/immersion-tracker-service.ts @@ -83,14 +83,15 @@ import { } from './immersion-tracker/query-library'; import { cleanupVocabularyStats, - deleteAnime as deleteAnimeQuery, - deleteSession as deleteSessionQuery, - deleteSessions as deleteSessionsQuery, - deleteVideo as deleteVideoQuery, getVideoDurationMs, markVideoWatched, upsertCoverArt, } from './immersion-tracker/query-maintenance'; +import { + DeleteMaintenanceWorkerRuntime, + type RunDeleteMaintenanceTask, +} from './immersion-tracker/delete-maintenance-worker-runtime'; +import { DeleteMaintenanceScheduler } from './immersion-tracker/delete-maintenance-scheduler'; import { repairJellyfinStreamVideoLinks } from './immersion-tracker/jellyfin-link-repair'; import { repairLegacySeasonlessAnimeRows, @@ -182,6 +183,7 @@ const YOUTUBE_SCREENSHOT_MAX_SECONDS = 120; const YOUTUBE_OEMBED_ENDPOINT = 'https://www.youtube.com/oembed'; const YOUTUBE_ID_PATTERN = /^[A-Za-z0-9_-]{6,}$/; const YOUTUBE_METADATA_REFRESH_MS = 24 * 60 * 60 * 1000; +const DELETE_MAINTENANCE_BATCH_WINDOW_MS = 10; function isValidYouTubeVideoId(value: string | null): boolean { return Boolean(value && YOUTUBE_ID_PATTERN.test(value)); @@ -385,6 +387,8 @@ export class ImmersionTrackerService { private readonly vacuumIntervalMs: number; private readonly dbPath: string; private readonly writeLock = { locked: false }; + private readonly destroyDeleteMaintenanceRunner: () => void; + private readonly deleteMaintenanceScheduler: DeleteMaintenanceScheduler; private flushTimer: ReturnType | null = null; private maintenanceTimer: ReturnType | null = null; private flushScheduled = false; @@ -406,9 +410,38 @@ export class ImmersionTrackerService { | ((row: LegacyVocabularyPosRow) => Promise) | undefined; - constructor(options: ImmersionTrackerOptions) { + constructor( + options: ImmersionTrackerOptions, + dependencies: { + runDeleteMaintenanceTask?: RunDeleteMaintenanceTask; + destroyDeleteMaintenanceRunner?: () => void; + } = {}, + ) { this.dbPath = options.dbPath; this.resolveLegacyVocabularyPos = options.resolveLegacyVocabularyPos; + let runDeleteMaintenanceTask: RunDeleteMaintenanceTask; + if (dependencies.runDeleteMaintenanceTask) { + runDeleteMaintenanceTask = dependencies.runDeleteMaintenanceTask; + this.destroyDeleteMaintenanceRunner = + dependencies.destroyDeleteMaintenanceRunner ?? (() => {}); + } else { + const deleteMaintenanceRuntime = new DeleteMaintenanceWorkerRuntime(); + runDeleteMaintenanceTask = (dbPath, task) => deleteMaintenanceRuntime.run(dbPath, task); + this.destroyDeleteMaintenanceRunner = () => deleteMaintenanceRuntime.destroy(); + } + this.deleteMaintenanceScheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: DELETE_MAINTENANCE_BATCH_WINDOW_MS, + runTask: (task) => runDeleteMaintenanceTask(this.dbPath, task), + onBusy: () => { + this.flushTelemetry(true); + while (this.queue.length > 0) this.flushNow(); + this.writeLock.locked = true; + }, + onIdle: () => { + this.writeLock.locked = false; + if (!this.isDestroyed && this.queue.length > 0) this.scheduleFlush(0); + }, + }); const parentDir = path.dirname(this.dbPath); if (!fs.existsSync(parentDir)) { fs.mkdirSync(parentDir, { recursive: true }); @@ -512,6 +545,8 @@ export class ImmersionTrackerService { } this.finalizeActiveSession(); this.isDestroyed = true; + this.deleteMaintenanceScheduler.destroy(); + this.destroyDeleteMaintenanceRunner(); this.db.close(); } @@ -709,51 +744,66 @@ export class ImmersionTrackerService { this.logger.warn(`Ignoring delete request for active immersion session ${sessionId}`); return; } - deleteSessionQuery(this.db, sessionId); + await this.enqueueDeleteMaintenanceTask(() => ({ kind: 'session', sessionId })); } async deleteSessions(sessionIds: number[]): Promise { - const activeSessionId = this.sessionState?.sessionId; - const deletableSessionIds = - activeSessionId === undefined - ? sessionIds - : sessionIds.filter((sessionId) => sessionId !== activeSessionId); - if (deletableSessionIds.length !== sessionIds.length) { - this.logger.warn( - `Ignoring bulk delete request for active immersion session ${activeSessionId}`, - ); - } - deleteSessionsQuery(this.db, deletableSessionIds); + await this.enqueueDeleteMaintenanceTask(() => { + const activeSessionId = this.sessionState?.sessionId; + const deletableSessionIds = + activeSessionId === undefined + ? sessionIds + : sessionIds.filter((sessionId) => sessionId !== activeSessionId); + if (deletableSessionIds.length !== sessionIds.length) { + this.logger.warn( + `Ignoring bulk delete request for active immersion session ${activeSessionId}`, + ); + } + if (deletableSessionIds.length === 0) return null; + return { kind: 'sessions', sessionIds: deletableSessionIds }; + }); } async deleteVideo(videoId: number): Promise { - if (this.sessionState?.videoId === videoId) { - this.logger.warn(`Ignoring delete request for active immersion video ${videoId}`); - return; - } - deleteVideoQuery(this.db, videoId); + await this.enqueueDeleteMaintenanceTask(() => { + if (this.sessionState?.videoId === videoId) { + this.logger.warn(`Ignoring delete request for active immersion video ${videoId}`); + return null; + } + return { kind: 'video', videoId }; + }); } async deleteAnime(animeId: number): Promise { - // The active video's anime link is assigned asynchronously after the title - // is parsed, so a guard reading imm_videos too early sees a null and lets - // the delete through — then the late update recreates the anime row. - const pendingVideoId = this.sessionState?.videoId; - if (pendingVideoId !== undefined) { - await this.pendingAnimeMetadataUpdates.get(pendingVideoId); - } - - const activeVideoId = this.sessionState?.videoId; - if (activeVideoId !== undefined) { - const activeAnime = this.db - .prepare('SELECT anime_id FROM imm_videos WHERE video_id = ?') - .get(activeVideoId) as { anime_id: number | null } | null; - if (activeAnime?.anime_id === animeId) { - this.logger.warn(`Ignoring delete request for active immersion anime ${animeId}`); - return; + await this.enqueueDeleteMaintenanceTask(async () => { + // Resolve this at dispatch time because another queued delete can leave + // enough time for playback to switch to an episode of this anime. + const pendingVideoId = this.sessionState?.videoId; + if (pendingVideoId !== undefined) { + await this.pendingAnimeMetadataUpdates.get(pendingVideoId); } + + const activeVideoId = this.sessionState?.videoId; + if (activeVideoId !== undefined) { + const activeAnime = this.db + .prepare('SELECT anime_id FROM imm_videos WHERE video_id = ?') + .get(activeVideoId) as { anime_id: number | null } | null; + if (activeAnime?.anime_id === animeId) { + this.logger.warn(`Ignoring delete request for active immersion anime ${animeId}`); + return null; + } + } + return { kind: 'anime', animeId }; + }); + } + + private enqueueDeleteMaintenanceTask( + resolveTask: Parameters[0], + ): Promise { + if (this.isDestroyed) { + return Promise.reject(new Error('Immersion tracker is shutting down')); } - deleteAnimeQuery(this.db, animeId); + return this.deleteMaintenanceScheduler.enqueue(resolveTask); } async reassignAnimeAnilist( @@ -1811,7 +1861,7 @@ export class ImmersionTrackerService { } private runMaintenance(): void { - if (this.isDestroyed) return; + if (this.isDestroyed || this.writeLock.locked) return; try { this.flushTelemetry(true); this.flushNow(); diff --git a/src/core/services/immersion-tracker/__tests__/query-split-modules.test.ts b/src/core/services/immersion-tracker/__tests__/query-split-modules.test.ts index 6c530f73..efd6c574 100644 --- a/src/core/services/immersion-tracker/__tests__/query-split-modules.test.ts +++ b/src/core/services/immersion-tracker/__tests__/query-split-modules.test.ts @@ -50,6 +50,7 @@ import { updateAnimeAnilistInfo, upsertCoverArt, } from '../query-maintenance.js'; +import { deleteMaintenanceBatch } from '../query-delete-maintenance.js'; import { getLocalEpochDay } from '../query-shared.js'; import { EVENT_CARD_MINED, EVENT_SUBTITLE_LINE, SOURCE_TYPE_LOCAL } from '../types.js'; @@ -985,3 +986,197 @@ test('split maintenance helpers delete multiple sessions and whole videos with d cleanupDbPath(dbPath); } }); + +test('delete maintenance batch preserves retained data across overlapping session, video, and anime targets', () => { + const { db, dbPath, stmts } = createDb(); + + try { + const retainedAnimeId = getOrCreateAnimeRecord(db, { + parsedTitle: 'Retained Anime', + canonicalTitle: 'Retained Anime', + anilistId: null, + titleRomaji: null, + titleEnglish: null, + titleNative: null, + metadataJson: null, + }); + const deletedAnimeId = getOrCreateAnimeRecord(db, { + parsedTitle: 'Deleted Anime', + canonicalTitle: 'Deleted Anime', + anilistId: null, + titleRomaji: null, + titleEnglish: null, + titleNative: null, + metadataJson: null, + }); + const retainedVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-retain.mkv', { + canonicalTitle: 'Batch Retain', + sourcePath: '/tmp/batch-retain.mkv', + sourceUrl: null, + sourceType: SOURCE_TYPE_LOCAL, + }); + const deletedVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-video.mkv', { + canonicalTitle: 'Batch Video', + sourcePath: '/tmp/batch-video.mkv', + sourceUrl: null, + sourceType: SOURCE_TYPE_LOCAL, + }); + const animeVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-anime.mkv', { + canonicalTitle: 'Batch Anime', + sourcePath: '/tmp/batch-anime.mkv', + sourceUrl: null, + sourceType: SOURCE_TYPE_LOCAL, + }); + for (const [videoId, animeId, episode] of [ + [retainedVideoId, retainedAnimeId, 1], + [deletedVideoId, retainedAnimeId, 2], + [animeVideoId, deletedAnimeId, 1], + ] as const) { + linkVideoToAnimeRecord(db, videoId, { + animeId, + parsedBasename: `batch-${episode}.mkv`, + parsedTitle: animeId === retainedAnimeId ? 'Retained Anime' : 'Deleted Anime', + parsedSeason: 1, + parsedEpisode: episode, + parserSource: 'test', + parserConfidence: 1, + parseMetadataJson: null, + }); + } + + const startedAtMs = 1_700_000_000_000; + const deletedSessionId = startSessionRecord(db, retainedVideoId, startedAtMs).sessionId; + const retainedSessionId = startSessionRecord( + db, + retainedVideoId, + startedAtMs + 1_000, + ).sessionId; + const videoSessionId = startSessionRecord(db, deletedVideoId, startedAtMs + 2_000).sessionId; + const animeSessionId = startSessionRecord(db, animeVideoId, startedAtMs + 3_000).sessionId; + for (const [sessionId, sessionStartedAtMs] of [ + [deletedSessionId, startedAtMs], + [retainedSessionId, startedAtMs + 1_000], + [videoSessionId, startedAtMs + 2_000], + [animeSessionId, startedAtMs + 3_000], + ] as const) { + finalizeSessionMetrics(db, sessionId, sessionStartedAtMs); + } + + for (const [index, sessionId, videoId, animeId] of [ + [1, deletedSessionId, retainedVideoId, retainedAnimeId], + [2, retainedSessionId, retainedVideoId, retainedAnimeId], + [3, videoSessionId, deletedVideoId, retainedAnimeId], + [4, animeSessionId, animeVideoId, deletedAnimeId], + ] as const) { + insertWordOccurrence(db, stmts, { + sessionId, + videoId, + animeId, + lineIndex: index, + text: '猫日', + word: { headword: '猫', word: '猫', reading: 'ねこ' }, + }); + insertKanjiOccurrence(db, stmts, { + sessionId, + videoId, + animeId, + lineIndex: index + 10, + text: '猫日', + kanji: '日', + }); + } + + const rollupDay = getLocalEpochDay(db, startedAtMs); + const rollupMonth = ( + db + .prepare( + `SELECT CAST(strftime('%Y%m', CAST(? AS REAL) / 1000, 'unixepoch', 'localtime') AS INTEGER) AS rollupMonth`, + ) + .get(startedAtMs) as { rollupMonth: number } + ).rollupMonth; + for (const videoId of [retainedVideoId, deletedVideoId, animeVideoId]) { + db.prepare( + `INSERT INTO imm_daily_rollups ( + rollup_day, video_id, total_sessions, total_active_min, total_lines_seen, + total_tokens_seen, total_cards, CREATED_DATE, LAST_UPDATE_DATE + ) VALUES (?, ?, 99, 99, 99, 99, 99, ?, ?)`, + ).run(rollupDay, videoId, startedAtMs, startedAtMs); + db.prepare( + `INSERT INTO imm_monthly_rollups ( + rollup_month, video_id, total_sessions, total_active_min, total_lines_seen, + total_tokens_seen, total_cards, CREATED_DATE, LAST_UPDATE_DATE + ) VALUES (?, ?, 99, 99, 99, 99, 99, ?, ?)`, + ).run(rollupMonth, videoId, startedAtMs, startedAtMs); + } + + deleteMaintenanceBatch(db, [ + { kind: 'session', sessionId: deletedSessionId }, + { kind: 'session', sessionId: videoSessionId }, + { kind: 'video', videoId: deletedVideoId }, + { kind: 'video', videoId: animeVideoId }, + { kind: 'anime', animeId: deletedAnimeId }, + ]); + + assert.deepEqual(db.prepare('SELECT session_id FROM imm_sessions').all(), [ + { session_id: retainedSessionId }, + ]); + assert.deepEqual(db.prepare('SELECT video_id FROM imm_videos').all(), [ + { video_id: retainedVideoId }, + ]); + assert.deepEqual(db.prepare('SELECT anime_id FROM imm_anime').all(), [ + { anime_id: retainedAnimeId }, + ]); + assert.equal( + ( + db.prepare(`SELECT frequency FROM imm_words WHERE headword = '猫'`).get() as { + frequency: number; + } + ).frequency, + 1, + ); + assert.equal( + ( + db.prepare(`SELECT frequency FROM imm_kanji WHERE kanji = '日'`).get() as { + frequency: number; + } + ).frequency, + 1, + ); + assert.deepEqual( + db.prepare('SELECT video_id, total_sessions FROM imm_daily_rollups').all() as Array<{ + video_id: number; + total_sessions: number; + }>, + [{ video_id: retainedVideoId, total_sessions: 1 }], + ); + assert.deepEqual( + db.prepare('SELECT video_id, total_sessions FROM imm_monthly_rollups').all() as Array<{ + video_id: number; + total_sessions: number; + }>, + [{ video_id: retainedVideoId, total_sessions: 1 }], + ); + } finally { + db.close(); + cleanupDbPath(dbPath); + } +}); + +test('delete maintenance batch chunks id lists below the SQLite variable limit', () => { + const { db, dbPath } = createDb(); + + try { + const ids = Array.from({ length: 32_767 }, (_, index) => index + 1); + + assert.doesNotThrow(() => { + deleteMaintenanceBatch(db, [ + { kind: 'sessions', sessionIds: ids }, + ...ids.map((videoId) => ({ kind: 'video' as const, videoId })), + ...ids.map((animeId) => ({ kind: 'anime' as const, animeId })), + ]); + }); + } finally { + db.close(); + cleanupDbPath(dbPath); + } +}); diff --git a/src/core/services/immersion-tracker/delete-maintenance-scheduler.test.ts b/src/core/services/immersion-tracker/delete-maintenance-scheduler.test.ts new file mode 100644 index 00000000..45f597cd --- /dev/null +++ b/src/core/services/immersion-tracker/delete-maintenance-scheduler.test.ts @@ -0,0 +1,160 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { DeleteMaintenanceScheduler } from './delete-maintenance-scheduler'; +import type { DeleteMaintenanceTask } from './delete-maintenance'; + +test('scheduler batches same-turn requests and balances busy state', async () => { + const tasks: DeleteMaintenanceTask[] = []; + const states: string[] = []; + const scheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: 0, + runTask: async (task) => { + tasks.push(task); + }, + onBusy: () => states.push('busy'), + onIdle: () => states.push('idle'), + }); + + const first = scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 })); + const second = scheduler.enqueue(() => ({ kind: 'sessions', sessionIds: [2, 3] })); + const third = scheduler.enqueue(() => null); + await Promise.all([first, second, third]); + + assert.deepEqual(tasks, [ + { + kind: 'batch', + tasks: [ + { kind: 'session', sessionId: 1 }, + { kind: 'sessions', sessionIds: [2, 3] }, + ], + }, + ]); + assert.deepEqual(states, ['busy', 'idle']); +}); + +test('scheduler rejects enqueue after destruction without entering busy state', async () => { + let busyCalls = 0; + let runCalls = 0; + const scheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: 0, + runTask: async () => { + runCalls += 1; + }, + onBusy: () => { + busyCalls += 1; + }, + onIdle: () => {}, + }); + scheduler.destroy(); + + await assert.rejects( + scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 })), + /shutting down/, + ); + assert.equal(busyCalls, 0); + assert.equal(runCalls, 0); +}); + +test('scheduler rejects every request in a batch when the maintenance task fails', async () => { + const failure = new Error('maintenance failed'); + const scheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: 0, + runTask: async () => { + throw failure; + }, + onBusy: () => {}, + onIdle: () => {}, + }); + + const first = scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 })); + const second = scheduler.enqueue(() => ({ kind: 'session', sessionId: 2 })); + + const results = await Promise.allSettled([first, second]); + assert.deepEqual( + results.map((result) => (result.status === 'rejected' ? result.reason : null)), + [failure, failure], + ); +}); + +test('scheduler rejects only the request whose task resolution fails', async () => { + const failure = new Error('resolution failed'); + const tasks: DeleteMaintenanceTask[] = []; + const scheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: 0, + runTask: async (task) => { + tasks.push(task); + }, + onBusy: () => {}, + onIdle: () => {}, + }); + + const failed = scheduler.enqueue(() => { + throw failure; + }); + const succeeded = scheduler.enqueue(() => ({ kind: 'session', sessionId: 2 })); + + const results = await Promise.allSettled([failed, succeeded]); + assert.equal(results[0]?.status, 'rejected'); + assert.equal(results[0]?.status === 'rejected' ? results[0].reason : null, failure); + assert.equal(results[1]?.status, 'fulfilled'); + assert.deepEqual(tasks, [{ kind: 'session', sessionId: 2 }]); +}); + +test('scheduler does not schedule another drain when the queue is empty', async () => { + const originalSetTimeout = globalThis.setTimeout; + let timerCalls = 0; + globalThis.setTimeout = ((handler: TimerHandler, timeout?: number, ...args: unknown[]) => { + timerCalls += 1; + return originalSetTimeout(handler, timeout, ...args); + }) as typeof setTimeout; + + try { + const scheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: 0, + runTask: async () => {}, + onBusy: () => {}, + onIdle: () => {}, + }); + + await scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 })); + assert.equal(timerCalls, 1); + } finally { + globalThis.setTimeout = originalSetTimeout; + } +}); + +test('scheduler serializes batches and rejects requests queued at destruction', async () => { + const releases: Array<() => void> = []; + let activeTasks = 0; + let maxActiveTasks = 0; + const scheduler = new DeleteMaintenanceScheduler({ + batchWindowMs: 0, + runTask: async () => { + activeTasks += 1; + maxActiveTasks = Math.max(maxActiveTasks, activeTasks); + await new Promise((resolve) => releases.push(resolve)); + activeTasks -= 1; + }, + onBusy: () => {}, + onIdle: () => {}, + }); + + const first = scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 })); + const maxPollAttempts = 100; + let pollAttempts = 0; + while (releases.length === 0 && pollAttempts < maxPollAttempts) { + pollAttempts += 1; + await new Promise((resolve) => setTimeout(resolve, 0)); + } + assert.ok( + releases.length > 0, + `runTask did not produce a release after ${maxPollAttempts} polling attempts`, + ); + const queued = scheduler.enqueue(() => ({ kind: 'session', sessionId: 2 })); + scheduler.destroy(); + + await assert.rejects(queued, /shutting down/); + releases[0]?.(); + await first; + assert.equal(maxActiveTasks, 1); +}); diff --git a/src/core/services/immersion-tracker/delete-maintenance-scheduler.ts b/src/core/services/immersion-tracker/delete-maintenance-scheduler.ts new file mode 100644 index 00000000..45fd8483 --- /dev/null +++ b/src/core/services/immersion-tracker/delete-maintenance-scheduler.ts @@ -0,0 +1,105 @@ +import type { DeleteMaintenanceOperation, DeleteMaintenanceTask } from './delete-maintenance'; + +type ResolveDeleteMaintenanceOperation = () => + | DeleteMaintenanceOperation + | null + | Promise; + +interface PendingDeleteMaintenanceRequest { + resolveTask: ResolveDeleteMaintenanceOperation; + resolve: () => void; + reject: (error: unknown) => void; +} + +interface DeleteMaintenanceSchedulerOptions { + batchWindowMs: number; + runTask: (task: DeleteMaintenanceTask) => Promise; + onBusy: () => void; + onIdle: () => void; +} + +export class DeleteMaintenanceScheduler { + private readonly pendingRequests: PendingDeleteMaintenanceRequest[] = []; + private running = false; + private drainTimer: ReturnType | null = null; + private pendingTaskCount = 0; + private destroyed = false; + + constructor(private readonly options: DeleteMaintenanceSchedulerOptions) {} + + enqueue(resolveTask: ResolveDeleteMaintenanceOperation): Promise { + if (this.destroyed) { + return Promise.reject(new Error('Immersion tracker is shutting down')); + } + + if (this.pendingTaskCount === 0) this.options.onBusy(); + this.pendingTaskCount += 1; + + const result = new Promise((resolve, reject) => { + this.pendingRequests.push({ resolveTask, resolve, reject }); + this.scheduleDrain(); + }); + + return result.finally(() => { + this.pendingTaskCount -= 1; + if (this.pendingTaskCount === 0) this.options.onIdle(); + }); + } + + destroy(): void { + if (this.destroyed) return; + this.destroyed = true; + if (this.drainTimer) { + clearTimeout(this.drainTimer); + this.drainTimer = null; + } + const error = new Error('Immersion tracker is shutting down'); + for (const request of this.pendingRequests.splice(0)) request.reject(error); + } + + private scheduleDrain(): void { + if (this.destroyed || this.running || this.drainTimer || this.pendingRequests.length === 0) { + return; + } + this.drainTimer = setTimeout(() => { + this.drainTimer = null; + void this.drain(); + }, this.options.batchWindowMs); + } + + private async drain(): Promise { + if (this.running || this.pendingRequests.length === 0) return; + this.running = true; + const requests = this.pendingRequests.splice(0); + const runnable: Array<{ + request: PendingDeleteMaintenanceRequest; + task: DeleteMaintenanceOperation; + }> = []; + + for (const request of requests) { + try { + const task = await request.resolveTask(); + if (task) runnable.push({ request, task }); + else request.resolve(); + } catch (error) { + request.reject(error); + } + } + + if (runnable.length > 0) { + const task: DeleteMaintenanceTask = + runnable.length === 1 + ? runnable[0]!.task + : { kind: 'batch', tasks: runnable.map((entry) => entry.task) }; + try { + await this.options.runTask(task); + for (const { request } of runnable) request.resolve(); + } catch (error) { + for (const { request } of runnable) request.reject(error); + } + } + + this.running = false; + this.scheduleDrain(); + } +} diff --git a/src/core/services/immersion-tracker/delete-maintenance-worker-runtime.test.ts b/src/core/services/immersion-tracker/delete-maintenance-worker-runtime.test.ts new file mode 100644 index 00000000..33f80432 --- /dev/null +++ b/src/core/services/immersion-tracker/delete-maintenance-worker-runtime.test.ts @@ -0,0 +1,239 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import test from 'node:test'; +import { + DeleteMaintenanceWorkerRuntime, + resolveDeleteMaintenanceWorkerPath, +} from './delete-maintenance-worker-runtime'; +import { executeDeleteMaintenanceTask } from './delete-maintenance'; +import { startSessionRecord } from './session'; +import { Database } from './sqlite'; +import { applyPragmas, ensureSchema, getOrCreateVideoRecord } from './storage'; + +type FakeWorkerListener = (value: never) => void; + +function createFakeWorker() { + const listeners = new Map(); + const terminationState = { calls: 0 }; + const worker = { + once(event: string, listener: FakeWorkerListener) { + listeners.set(event, listener); + return this; + }, + terminate: async () => { + terminationState.calls += 1; + return 0; + }, + }; + return { worker, listeners, terminationState }; +} + +type FakeWorker = ReturnType['worker']; + +test('a delete batch rebuilds lifetime summaries once', () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-delete-batch-test-')); + const dbPath = path.join(tempDir, 'immersion.sqlite'); + let db = new Database(dbPath); + + try { + applyPragmas(db); + ensureSchema(db); + const videoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-delete.mkv', { + canonicalTitle: 'Batch Delete', + sourcePath: '/tmp/batch-delete.mkv', + sourceUrl: null, + sourceType: 1, + }); + const firstSessionId = startSessionRecord(db, videoId, 1_000).sessionId; + const secondSessionId = startSessionRecord(db, videoId, 2_000).sessionId; + const deletedVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-delete-video.mkv', { + canonicalTitle: 'Batch Delete Video', + sourcePath: '/tmp/batch-delete-video.mkv', + sourceUrl: null, + sourceType: 1, + }); + startSessionRecord(db, deletedVideoId, 3_000); + db.exec(` + CREATE TABLE delete_rebuild_audit (id INTEGER PRIMARY KEY); + CREATE TRIGGER count_delete_lifetime_rebuild + AFTER UPDATE OF last_rebuilt_ms ON imm_lifetime_global + BEGIN + INSERT INTO delete_rebuild_audit (id) VALUES (NULL); + END; + `); + db.close(); + + executeDeleteMaintenanceTask(dbPath, { + kind: 'batch', + tasks: [ + { kind: 'session', sessionId: firstSessionId }, + { kind: 'video', videoId: deletedVideoId }, + ], + }); + + db = new Database(dbPath); + const audit = db.prepare('SELECT COUNT(*) AS total FROM delete_rebuild_audit').get() as { + total: number; + }; + const retainedSession = db + .prepare('SELECT session_id AS sessionId FROM imm_sessions WHERE video_id = ?') + .get(videoId) as { sessionId: number } | null; + const deletedVideo = db + .prepare('SELECT video_id AS videoId FROM imm_videos WHERE video_id = ?') + .get(deletedVideoId) as { videoId: number } | null; + assert.equal(retainedSession?.sessionId, secondSessionId); + assert.equal(deletedVideo, undefined); + assert.equal( + audit.total, + 2, + 'one rebuild performs exactly its reset and final global summary writes', + ); + } finally { + try { + db.close(); + } catch { + // The setup connection closes before maintenance runs. + } + fs.rmSync(tempDir, { recursive: true, force: true }); + } +}); + +test( + 'compiled delete worker removes data through its separate database connection', + { skip: resolveDeleteMaintenanceWorkerPath() === null }, + async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-delete-worker-test-')); + const dbPath = path.join(tempDir, 'immersion.sqlite'); + const runtime = new DeleteMaintenanceWorkerRuntime(); + let db = new Database(dbPath); + + try { + applyPragmas(db); + ensureSchema(db); + const videoId = getOrCreateVideoRecord(db, 'local:/tmp/worker-delete.mkv', { + canonicalTitle: 'Worker Delete', + sourcePath: '/tmp/worker-delete.mkv', + sourceUrl: null, + sourceType: 1, + }); + const firstSessionId = startSessionRecord(db, videoId, 1_000).sessionId; + const secondSessionId = startSessionRecord(db, videoId, 2_000).sessionId; + db.close(); + + await runtime.run(dbPath, { + kind: 'batch', + tasks: [ + { kind: 'session', sessionId: firstSessionId }, + { kind: 'session', sessionId: secondSessionId }, + ], + }); + + db = new Database(dbPath); + const row = db + .prepare('SELECT COUNT(*) AS total FROM imm_sessions WHERE video_id = ?') + .get(videoId) as { total: number }; + assert.equal(row.total, 0); + } finally { + runtime.destroy(); + try { + db.close(); + } catch { + // The setup connection is already closed before the worker starts. + } + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }, +); + +test('worker runtime warns before falling back when no emitted worker is available', async () => { + const warnings: unknown[][] = []; + const fallbackTasks: unknown[] = []; + const runtime = new DeleteMaintenanceWorkerRuntime({ + resolveWorkerPath: () => null, + warn: (...args) => warnings.push(args), + executeFallback: (_dbPath, task) => fallbackTasks.push(task), + }); + + await runtime.run('/tmp/fallback.sqlite', { kind: 'session', sessionId: 1 }); + + assert.equal(warnings.length, 1); + assert.match(String(warnings[0]?.[0]), /worker unavailable/i); + assert.deepEqual(fallbackTasks, [{ kind: 'session', sessionId: 1 }]); +}); + +test('worker runtime terminates a worker after successful settlement', async () => { + const { worker, listeners, terminationState } = createFakeWorker(); + const runtime = new DeleteMaintenanceWorkerRuntime({ + resolveWorkerPath: () => '/tmp/delete-worker.js', + createWorker: async () => worker, + }); + + const result = runtime.run('/tmp/test.sqlite', { kind: 'session', sessionId: 1 }); + await new Promise((resolve) => setTimeout(resolve, 0)); + listeners.get('message')?.({ ok: true } as never); + await result; + + assert.equal(terminationState.calls, 1); +}); + +test('worker runtime terminates a worker after failed settlement', async () => { + const { worker, listeners, terminationState } = createFakeWorker(); + const runtime = new DeleteMaintenanceWorkerRuntime({ + resolveWorkerPath: () => '/tmp/delete-worker.js', + createWorker: async () => worker, + }); + + const result = runtime.run('/tmp/test.sqlite', { kind: 'session', sessionId: 1 }); + await new Promise((resolve) => setTimeout(resolve, 0)); + listeners.get('error')?.(new Error('worker failed') as never); + + await assert.rejects(result, /worker failed/); + assert.equal(terminationState.calls, 1); +}); + +test('worker runtime terminates a worker created after shutdown begins', async () => { + const { worker, listeners, terminationState } = createFakeWorker(); + const createGate: { resolve?: (worker: FakeWorker) => void } = {}; + const fallbackTasks: unknown[] = []; + const runtime = new DeleteMaintenanceWorkerRuntime({ + resolveWorkerPath: () => '/tmp/delete-worker.js', + createWorker: () => + new Promise((resolve) => { + createGate.resolve = resolve; + }), + executeFallback: (_dbPath, task) => fallbackTasks.push(task), + }); + + const result = runtime.run('/tmp/test.sqlite', { kind: 'session', sessionId: 1 }); + await new Promise((resolve) => setTimeout(resolve, 0)); + runtime.destroy(); + createGate.resolve?.(worker); + + await assert.rejects(result, /shut down/); + assert.equal(terminationState.calls, 1); + assert.equal(listeners.size, 0); + assert.deepEqual(fallbackTasks, []); +}); + +test('worker runtime does not fall back when worker creation fails during shutdown', async () => { + const createGate: { reject?: (error: Error) => void } = {}; + const fallbackTasks: unknown[] = []; + const runtime = new DeleteMaintenanceWorkerRuntime({ + resolveWorkerPath: () => '/tmp/delete-worker.js', + createWorker: () => + new Promise((_resolve, reject) => { + createGate.reject = reject; + }), + executeFallback: (_dbPath, task) => fallbackTasks.push(task), + }); + + const result = runtime.run('/tmp/test.sqlite', { kind: 'session', sessionId: 1 }); + await new Promise((resolve) => setTimeout(resolve, 0)); + runtime.destroy(); + createGate.reject?.(new Error('creation failed')); + + await assert.rejects(result, /shut down/); + assert.deepEqual(fallbackTasks, []); +}); diff --git a/src/core/services/immersion-tracker/delete-maintenance-worker-runtime.ts b/src/core/services/immersion-tracker/delete-maintenance-worker-runtime.ts new file mode 100644 index 00000000..0a92c417 --- /dev/null +++ b/src/core/services/immersion-tracker/delete-maintenance-worker-runtime.ts @@ -0,0 +1,121 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { createLogger } from '../../../logger'; +import { executeDeleteMaintenanceTask, type DeleteMaintenanceTask } from './delete-maintenance'; + +interface DeleteMaintenanceWorkerResponse { + ok?: unknown; + error?: unknown; +} + +export type RunDeleteMaintenanceTask = ( + dbPath: string, + task: DeleteMaintenanceTask, +) => Promise; + +interface DeleteMaintenanceWorkerHandle { + once(event: 'message', listener: (message: DeleteMaintenanceWorkerResponse) => void): this; + once(event: 'error', listener: (error: Error) => void): this; + once(event: 'exit', listener: (code: number) => void): this; + terminate(): Promise; +} + +interface DeleteMaintenanceWorkerRuntimeOptions { + resolveWorkerPath?: () => string | null; + createWorker?: ( + workerPath: string, + workerData: { dbPath: string; task: DeleteMaintenanceTask }, + ) => Promise; + executeFallback?: typeof executeDeleteMaintenanceTask; + warn?: (message: string, ...meta: unknown[]) => void; +} + +export function resolveDeleteMaintenanceWorkerPath(): string | null { + const workerPath = path.join(__dirname, 'delete-maintenance-worker-thread.js'); + return fs.existsSync(workerPath) ? workerPath : null; +} + +const logger = createLogger('main:immersion-tracker:delete-worker'); + +export class DeleteMaintenanceWorkerRuntime { + private readonly activeWorkers = new Set(); + private destroyed = false; + + constructor(private readonly options: DeleteMaintenanceWorkerRuntimeOptions = {}) {} + + async run(dbPath: string, task: DeleteMaintenanceTask): Promise { + if (this.destroyed) { + throw new Error('Delete maintenance worker is shut down'); + } + + let worker: DeleteMaintenanceWorkerHandle; + try { + const workerPath = (this.options.resolveWorkerPath ?? resolveDeleteMaintenanceWorkerPath)(); + if (!workerPath) throw new Error('Emitted delete-maintenance worker module was not found'); + const createWorker = + this.options.createWorker ?? + (async (resolvedPath, workerData) => { + const { Worker } = await import('node:worker_threads'); + return new Worker(resolvedPath, { workerData }); + }); + worker = await createWorker(workerPath, { dbPath, task }); + } catch (error) { + if (this.destroyed) { + throw new Error('Delete maintenance worker is shut down'); + } + (this.options.warn ?? logger.warn)( + 'Delete maintenance worker unavailable; running maintenance on the current thread', + error, + ); + (this.options.executeFallback ?? executeDeleteMaintenanceTask)(dbPath, task); + return; + } + + if (this.destroyed) { + await worker.terminate().catch(() => undefined); + throw new Error('Delete maintenance worker is shut down'); + } + + await new Promise((resolve, reject) => { + let settled = false; + this.activeWorkers.add(worker); + + const settle = (error?: Error) => { + if (settled) return; + settled = true; + this.activeWorkers.delete(worker); + if (error) reject(error); + else resolve(); + void worker.terminate(); + }; + + worker.once('message', (message: DeleteMaintenanceWorkerResponse) => { + if (message.ok === true) { + settle(); + return; + } + const detail = typeof message.error === 'string' ? message.error : 'unknown worker error'; + settle(new Error(`Delete maintenance failed: ${detail}`)); + }); + worker.once('error', (error) => settle(error)); + worker.once('exit', (code) => { + settle( + new Error( + code === 0 + ? 'Delete maintenance worker exited without a response' + : `Delete maintenance worker exited with code ${code}`, + ), + ); + }); + }); + } + + destroy(): void { + if (this.destroyed) return; + this.destroyed = true; + for (const worker of this.activeWorkers) { + void worker.terminate(); + } + this.activeWorkers.clear(); + } +} diff --git a/src/core/services/immersion-tracker/delete-maintenance-worker-thread.ts b/src/core/services/immersion-tracker/delete-maintenance-worker-thread.ts new file mode 100644 index 00000000..496c3685 --- /dev/null +++ b/src/core/services/immersion-tracker/delete-maintenance-worker-thread.ts @@ -0,0 +1,22 @@ +import { parentPort, workerData } from 'node:worker_threads'; +import { executeDeleteMaintenanceTask, type DeleteMaintenanceTask } from './delete-maintenance'; + +interface DeleteMaintenanceWorkerData { + dbPath: string; + task: DeleteMaintenanceTask; +} + +if (!parentPort) { + throw new Error('delete maintenance worker missing parent port'); +} + +const port = parentPort; +const request = workerData as DeleteMaintenanceWorkerData; + +try { + executeDeleteMaintenanceTask(request.dbPath, request.task); + port.postMessage({ ok: true }); +} catch (error) { + const message = error instanceof Error ? error.message : String(error); + port.postMessage({ ok: false, error: message }); +} diff --git a/src/core/services/immersion-tracker/delete-maintenance.ts b/src/core/services/immersion-tracker/delete-maintenance.ts new file mode 100644 index 00000000..065d3601 --- /dev/null +++ b/src/core/services/immersion-tracker/delete-maintenance.ts @@ -0,0 +1,47 @@ +import { Database } from './sqlite'; +import { applyPragmas } from './storage'; +import { deleteAnime, deleteSession, deleteSessions, deleteVideo } from './query-maintenance'; +import { + deleteMaintenanceBatch, + type DeleteMaintenanceOperation, +} from './query-delete-maintenance'; + +export type { DeleteMaintenanceOperation } from './query-delete-maintenance'; + +export type DeleteMaintenanceTask = + | DeleteMaintenanceOperation + | { kind: 'batch'; tasks: DeleteMaintenanceOperation[] }; + +function executeDeleteMaintenanceOperation( + db: InstanceType, + task: DeleteMaintenanceOperation, +): void { + switch (task.kind) { + case 'session': + deleteSession(db, task.sessionId); + return; + case 'sessions': + deleteSessions(db, task.sessionIds); + return; + case 'video': + deleteVideo(db, task.videoId); + return; + case 'anime': + deleteAnime(db, task.animeId); + return; + } +} + +export function executeDeleteMaintenanceTask(dbPath: string, task: DeleteMaintenanceTask): void { + const db = new Database(dbPath); + try { + applyPragmas(db); + if (task.kind === 'batch') { + deleteMaintenanceBatch(db, task.tasks); + return; + } + executeDeleteMaintenanceOperation(db, task); + } finally { + db.close(); + } +} diff --git a/src/core/services/immersion-tracker/query-delete-maintenance.ts b/src/core/services/immersion-tracker/query-delete-maintenance.ts new file mode 100644 index 00000000..d1df1b60 --- /dev/null +++ b/src/core/services/immersion-tracker/query-delete-maintenance.ts @@ -0,0 +1,196 @@ +import type { DatabaseSync } from './sqlite'; +import { rebuildLifetimeSummariesInTransaction } from './lifetime'; +import { getRollupGroupsForSessions, refreshRollupsForGroupsInTransaction } from './maintenance'; +import { + applyLexicalRemovals, + cleanupUnusedCoverArtBlobHash, + deleteSessionsByIds, + forEachIdChunk, + makePlaceholders, + planLexicalRemovalsForSessions, + SQLITE_ID_CHUNK_SIZE, + type LexicalRemovalPlan, +} from './query-shared'; + +export type DeleteMaintenanceOperation = + | { kind: 'session'; sessionId: number } + | { kind: 'sessions'; sessionIds: number[] } + | { kind: 'video'; videoId: number } + | { kind: 'anime'; animeId: number }; + +function addOperationTargets( + operations: DeleteMaintenanceOperation[], + sessionIds: Set, + videoIds: Set, + animeIds: Set, +): void { + for (const operation of operations) { + switch (operation.kind) { + case 'session': + sessionIds.add(operation.sessionId); + break; + case 'sessions': + for (const sessionId of operation.sessionIds) sessionIds.add(sessionId); + break; + case 'video': + videoIds.add(operation.videoId); + break; + case 'anime': + animeIds.add(operation.animeId); + break; + } + } +} + +function selectIds( + db: DatabaseSync, + buildSql: (placeholders: string) => string, + params: number[], + column: string, +): number[] { + if (params.length === 0) return []; + const ids: number[] = []; + forEachIdChunk(params, (chunk) => { + const rows = db.prepare(buildSql(makePlaceholders(chunk))).all(...chunk) as Array< + Record + >; + for (const row of rows) ids.push(row[column]!); + }); + return ids; +} + +function planLexicalRemovalsInChunks(db: DatabaseSync, sessionIds: number[]): LexicalRemovalPlan { + const combined: LexicalRemovalPlan = { words: [], kanji: [] }; + const merge = (target: LexicalRemovalPlan['words'], source: LexicalRemovalPlan['words']) => { + const byId = new Map(target.map((entry) => [entry.id, entry])); + for (const entry of source) { + const existing = byId.get(entry.id); + if (!existing) { + const added = { ...entry }; + target.push(added); + byId.set(entry.id, added); + continue; + } + existing.removedFrequency += entry.removedFrequency; + if ( + entry.removedFirstSeenMs !== null && + (existing.removedFirstSeenMs === null || + entry.removedFirstSeenMs < existing.removedFirstSeenMs) + ) { + existing.removedFirstSeenMs = entry.removedFirstSeenMs; + } + if ( + entry.removedLastSeenMs !== null && + (existing.removedLastSeenMs === null || + entry.removedLastSeenMs > existing.removedLastSeenMs) + ) { + existing.removedLastSeenMs = entry.removedLastSeenMs; + } + } + }; + + forEachIdChunk(sessionIds, (chunk) => { + const plan = planLexicalRemovalsForSessions(db, chunk); + merge(combined.words, plan.words); + merge(combined.kanji, plan.kanji); + }); + return combined; +} + +export function deleteMaintenanceBatch( + db: DatabaseSync, + operations: DeleteMaintenanceOperation[], +): void { + if (operations.length === 0) return; + + db.exec('BEGIN IMMEDIATE'); + try { + const sessionIds = new Set(); + const videoIds = new Set(); + const animeIds = new Set(); + addOperationTargets(operations, sessionIds, videoIds, animeIds); + + const animeIdList = [...animeIds]; + for (const videoId of selectIds( + db, + (placeholders) => `SELECT video_id FROM imm_videos WHERE anime_id IN (${placeholders})`, + animeIdList, + 'video_id', + )) { + videoIds.add(videoId); + } + + const videoIdList = [...videoIds]; + for (const sessionId of selectIds( + db, + (placeholders) => `SELECT session_id FROM imm_sessions WHERE video_id IN (${placeholders})`, + videoIdList, + 'session_id', + )) { + sessionIds.add(sessionId); + } + + const sessionIdList = [...sessionIds]; + const lexicalRemovals = planLexicalRemovalsInChunks(db, sessionIdList); + const affectedRollupGroups = sessionIdList + .flatMap((_, index) => + index % SQLITE_ID_CHUNK_SIZE === 0 + ? getRollupGroupsForSessions(db, sessionIdList.slice(index, index + SQLITE_ID_CHUNK_SIZE)) + : [], + ) + .filter((group) => !videoIds.has(group.videoId)); + const coverBlobHashes = new Set(); + if (videoIdList.length > 0) { + forEachIdChunk(videoIdList, (chunk) => { + const placeholders = makePlaceholders(chunk); + const artRows = db + .prepare( + `SELECT cover_blob_hash AS coverBlobHash + FROM imm_media_art + WHERE video_id IN (${placeholders}) AND cover_blob_hash IS NOT NULL`, + ) + .all(...chunk) as Array<{ coverBlobHash: string }>; + for (const row of artRows) coverBlobHashes.add(row.coverBlobHash); + }); + + deleteSessionsByIds(db, sessionIdList); + forEachIdChunk(videoIdList, (chunk) => { + const placeholders = makePlaceholders(chunk); + db.prepare(`DELETE FROM imm_subtitle_lines WHERE video_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_daily_rollups WHERE video_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_monthly_rollups WHERE video_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_media_art WHERE video_id IN (${placeholders})`).run(...chunk); + db.prepare(`DELETE FROM imm_videos WHERE video_id IN (${placeholders})`).run(...chunk); + }); + } else { + deleteSessionsByIds(db, sessionIdList); + } + + for (const coverBlobHash of coverBlobHashes) { + cleanupUnusedCoverArtBlobHash(db, coverBlobHash); + } + if (animeIdList.length > 0) { + forEachIdChunk(animeIdList, (chunk) => { + const placeholders = makePlaceholders(chunk); + db.prepare(`DELETE FROM imm_lifetime_anime WHERE anime_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_anime WHERE anime_id IN (${placeholders})`).run(...chunk); + }); + } + + applyLexicalRemovals(db, lexicalRemovals); + rebuildLifetimeSummariesInTransaction(db); + refreshRollupsForGroupsInTransaction(db, affectedRollupGroups); + db.exec('COMMIT'); + } catch (error) { + db.exec('ROLLBACK'); + throw error; + } +} diff --git a/src/core/services/immersion-tracker/query-shared.ts b/src/core/services/immersion-tracker/query-shared.ts index ede2ee04..0472a500 100644 --- a/src/core/services/immersion-tracker/query-shared.ts +++ b/src/core/services/immersion-tracker/query-shared.ts @@ -80,6 +80,14 @@ export function makePlaceholders(values: number[]): string { return values.map(() => '?').join(','); } +export const SQLITE_ID_CHUNK_SIZE = 1_000; + +export function forEachIdChunk(ids: number[], callback: (chunk: number[]) => void): void { + for (let start = 0; start < ids.length; start += SQLITE_ID_CHUNK_SIZE) { + callback(ids.slice(start, start + SQLITE_ID_CHUNK_SIZE)); + } +} + export function resolvedCoverBlobExpr(mediaAlias: string, blobStoreAlias: string): string { return `COALESCE(${blobStoreAlias}.cover_blob, CASE WHEN ${mediaAlias}.cover_blob_hash IS NULL THEN ${mediaAlias}.cover_blob ELSE NULL END)`; } @@ -490,17 +498,19 @@ export function deleteSessionsByIds(db: DatabaseSync, sessionIds: number[]): voi return; } - const placeholders = makePlaceholders(sessionIds); - db.prepare(`DELETE FROM imm_subtitle_lines WHERE session_id IN (${placeholders})`).run( - ...sessionIds, - ); - db.prepare(`DELETE FROM imm_session_telemetry WHERE session_id IN (${placeholders})`).run( - ...sessionIds, - ); - db.prepare(`DELETE FROM imm_session_events WHERE session_id IN (${placeholders})`).run( - ...sessionIds, - ); - db.prepare(`DELETE FROM imm_sessions WHERE session_id IN (${placeholders})`).run(...sessionIds); + forEachIdChunk(sessionIds, (chunk) => { + const placeholders = makePlaceholders(chunk); + db.prepare(`DELETE FROM imm_subtitle_lines WHERE session_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_session_telemetry WHERE session_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_session_events WHERE session_id IN (${placeholders})`).run( + ...chunk, + ); + db.prepare(`DELETE FROM imm_sessions WHERE session_id IN (${placeholders})`).run(...chunk); + }); } export function toDbMs(ms: number | bigint): bigint {