mirror of
https://github.com/ksyasuda/SubMiner.git
synced 2026-08-13 13:55:57 -07:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
772635ab96
|
|||
|
675cb33519
|
|||
|
320db591ae
|
@@ -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.
|
||||||
@@ -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`
|
- Anki workflow: `src/anki-integration/`, `src/core/services/anki-jimaku*.ts`
|
||||||
- Immersion tracking: `src/core/services/immersion-tracker/`
|
- 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.
|
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/`
|
- 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-*`
|
- Jellyfin integration: `src/core/services/jellyfin*.ts`, `src/main/runtime/composers/jellyfin-*`
|
||||||
- Window trackers: `src/window-trackers/`
|
- Window trackers: `src/window-trackers/`
|
||||||
|
|||||||
@@ -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<void>((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<void>((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<void>((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<void>((resolve) => {
|
||||||
|
markFirstTaskStarted = resolve;
|
||||||
|
});
|
||||||
|
tracker = new Ctor(
|
||||||
|
{ dbPath },
|
||||||
|
{
|
||||||
|
runDeleteMaintenanceTask: async () => {
|
||||||
|
markFirstTaskStarted();
|
||||||
|
await new Promise<void>((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<void>((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 () => {
|
test('deleteVideo ignores the currently active video and keeps new writes flushable', async () => {
|
||||||
const dbPath = makeDbPath();
|
const dbPath = makeDbPath();
|
||||||
let tracker: ImmersionTrackerService | null = null;
|
let tracker: ImmersionTrackerService | null = null;
|
||||||
|
|||||||
@@ -83,14 +83,15 @@ import {
|
|||||||
} from './immersion-tracker/query-library';
|
} from './immersion-tracker/query-library';
|
||||||
import {
|
import {
|
||||||
cleanupVocabularyStats,
|
cleanupVocabularyStats,
|
||||||
deleteAnime as deleteAnimeQuery,
|
|
||||||
deleteSession as deleteSessionQuery,
|
|
||||||
deleteSessions as deleteSessionsQuery,
|
|
||||||
deleteVideo as deleteVideoQuery,
|
|
||||||
getVideoDurationMs,
|
getVideoDurationMs,
|
||||||
markVideoWatched,
|
markVideoWatched,
|
||||||
upsertCoverArt,
|
upsertCoverArt,
|
||||||
} from './immersion-tracker/query-maintenance';
|
} 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 { repairJellyfinStreamVideoLinks } from './immersion-tracker/jellyfin-link-repair';
|
||||||
import {
|
import {
|
||||||
repairLegacySeasonlessAnimeRows,
|
repairLegacySeasonlessAnimeRows,
|
||||||
@@ -182,6 +183,7 @@ const YOUTUBE_SCREENSHOT_MAX_SECONDS = 120;
|
|||||||
const YOUTUBE_OEMBED_ENDPOINT = 'https://www.youtube.com/oembed';
|
const YOUTUBE_OEMBED_ENDPOINT = 'https://www.youtube.com/oembed';
|
||||||
const YOUTUBE_ID_PATTERN = /^[A-Za-z0-9_-]{6,}$/;
|
const YOUTUBE_ID_PATTERN = /^[A-Za-z0-9_-]{6,}$/;
|
||||||
const YOUTUBE_METADATA_REFRESH_MS = 24 * 60 * 60 * 1000;
|
const YOUTUBE_METADATA_REFRESH_MS = 24 * 60 * 60 * 1000;
|
||||||
|
const DELETE_MAINTENANCE_BATCH_WINDOW_MS = 10;
|
||||||
|
|
||||||
function isValidYouTubeVideoId(value: string | null): boolean {
|
function isValidYouTubeVideoId(value: string | null): boolean {
|
||||||
return Boolean(value && YOUTUBE_ID_PATTERN.test(value));
|
return Boolean(value && YOUTUBE_ID_PATTERN.test(value));
|
||||||
@@ -385,6 +387,8 @@ export class ImmersionTrackerService {
|
|||||||
private readonly vacuumIntervalMs: number;
|
private readonly vacuumIntervalMs: number;
|
||||||
private readonly dbPath: string;
|
private readonly dbPath: string;
|
||||||
private readonly writeLock = { locked: false };
|
private readonly writeLock = { locked: false };
|
||||||
|
private readonly destroyDeleteMaintenanceRunner: () => void;
|
||||||
|
private readonly deleteMaintenanceScheduler: DeleteMaintenanceScheduler;
|
||||||
private flushTimer: ReturnType<typeof setTimeout> | null = null;
|
private flushTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
private maintenanceTimer: ReturnType<typeof setInterval> | null = null;
|
private maintenanceTimer: ReturnType<typeof setInterval> | null = null;
|
||||||
private flushScheduled = false;
|
private flushScheduled = false;
|
||||||
@@ -406,9 +410,38 @@ export class ImmersionTrackerService {
|
|||||||
| ((row: LegacyVocabularyPosRow) => Promise<LegacyVocabularyPosResolution | null>)
|
| ((row: LegacyVocabularyPosRow) => Promise<LegacyVocabularyPosResolution | null>)
|
||||||
| undefined;
|
| undefined;
|
||||||
|
|
||||||
constructor(options: ImmersionTrackerOptions) {
|
constructor(
|
||||||
|
options: ImmersionTrackerOptions,
|
||||||
|
dependencies: {
|
||||||
|
runDeleteMaintenanceTask?: RunDeleteMaintenanceTask;
|
||||||
|
destroyDeleteMaintenanceRunner?: () => void;
|
||||||
|
} = {},
|
||||||
|
) {
|
||||||
this.dbPath = options.dbPath;
|
this.dbPath = options.dbPath;
|
||||||
this.resolveLegacyVocabularyPos = options.resolveLegacyVocabularyPos;
|
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);
|
const parentDir = path.dirname(this.dbPath);
|
||||||
if (!fs.existsSync(parentDir)) {
|
if (!fs.existsSync(parentDir)) {
|
||||||
fs.mkdirSync(parentDir, { recursive: true });
|
fs.mkdirSync(parentDir, { recursive: true });
|
||||||
@@ -512,6 +545,8 @@ export class ImmersionTrackerService {
|
|||||||
}
|
}
|
||||||
this.finalizeActiveSession();
|
this.finalizeActiveSession();
|
||||||
this.isDestroyed = true;
|
this.isDestroyed = true;
|
||||||
|
this.deleteMaintenanceScheduler.destroy();
|
||||||
|
this.destroyDeleteMaintenanceRunner();
|
||||||
this.db.close();
|
this.db.close();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -709,10 +744,11 @@ export class ImmersionTrackerService {
|
|||||||
this.logger.warn(`Ignoring delete request for active immersion session ${sessionId}`);
|
this.logger.warn(`Ignoring delete request for active immersion session ${sessionId}`);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
deleteSessionQuery(this.db, sessionId);
|
await this.enqueueDeleteMaintenanceTask(() => ({ kind: 'session', sessionId }));
|
||||||
}
|
}
|
||||||
|
|
||||||
async deleteSessions(sessionIds: number[]): Promise<void> {
|
async deleteSessions(sessionIds: number[]): Promise<void> {
|
||||||
|
await this.enqueueDeleteMaintenanceTask(() => {
|
||||||
const activeSessionId = this.sessionState?.sessionId;
|
const activeSessionId = this.sessionState?.sessionId;
|
||||||
const deletableSessionIds =
|
const deletableSessionIds =
|
||||||
activeSessionId === undefined
|
activeSessionId === undefined
|
||||||
@@ -723,21 +759,25 @@ export class ImmersionTrackerService {
|
|||||||
`Ignoring bulk delete request for active immersion session ${activeSessionId}`,
|
`Ignoring bulk delete request for active immersion session ${activeSessionId}`,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
deleteSessionsQuery(this.db, deletableSessionIds);
|
if (deletableSessionIds.length === 0) return null;
|
||||||
|
return { kind: 'sessions', sessionIds: deletableSessionIds };
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
async deleteVideo(videoId: number): Promise<void> {
|
async deleteVideo(videoId: number): Promise<void> {
|
||||||
|
await this.enqueueDeleteMaintenanceTask(() => {
|
||||||
if (this.sessionState?.videoId === videoId) {
|
if (this.sessionState?.videoId === videoId) {
|
||||||
this.logger.warn(`Ignoring delete request for active immersion video ${videoId}`);
|
this.logger.warn(`Ignoring delete request for active immersion video ${videoId}`);
|
||||||
return;
|
return null;
|
||||||
}
|
}
|
||||||
deleteVideoQuery(this.db, videoId);
|
return { kind: 'video', videoId };
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
async deleteAnime(animeId: number): Promise<void> {
|
async deleteAnime(animeId: number): Promise<void> {
|
||||||
// The active video's anime link is assigned asynchronously after the title
|
await this.enqueueDeleteMaintenanceTask(async () => {
|
||||||
// is parsed, so a guard reading imm_videos too early sees a null and lets
|
// Resolve this at dispatch time because another queued delete can leave
|
||||||
// the delete through — then the late update recreates the anime row.
|
// enough time for playback to switch to an episode of this anime.
|
||||||
const pendingVideoId = this.sessionState?.videoId;
|
const pendingVideoId = this.sessionState?.videoId;
|
||||||
if (pendingVideoId !== undefined) {
|
if (pendingVideoId !== undefined) {
|
||||||
await this.pendingAnimeMetadataUpdates.get(pendingVideoId);
|
await this.pendingAnimeMetadataUpdates.get(pendingVideoId);
|
||||||
@@ -750,10 +790,20 @@ export class ImmersionTrackerService {
|
|||||||
.get(activeVideoId) as { anime_id: number | null } | null;
|
.get(activeVideoId) as { anime_id: number | null } | null;
|
||||||
if (activeAnime?.anime_id === animeId) {
|
if (activeAnime?.anime_id === animeId) {
|
||||||
this.logger.warn(`Ignoring delete request for active immersion anime ${animeId}`);
|
this.logger.warn(`Ignoring delete request for active immersion anime ${animeId}`);
|
||||||
return;
|
return null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
deleteAnimeQuery(this.db, animeId);
|
return { kind: 'anime', animeId };
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
private enqueueDeleteMaintenanceTask(
|
||||||
|
resolveTask: Parameters<DeleteMaintenanceScheduler['enqueue']>[0],
|
||||||
|
): Promise<void> {
|
||||||
|
if (this.isDestroyed) {
|
||||||
|
return Promise.reject(new Error('Immersion tracker is shutting down'));
|
||||||
|
}
|
||||||
|
return this.deleteMaintenanceScheduler.enqueue(resolveTask);
|
||||||
}
|
}
|
||||||
|
|
||||||
async reassignAnimeAnilist(
|
async reassignAnimeAnilist(
|
||||||
@@ -1811,7 +1861,7 @@ export class ImmersionTrackerService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private runMaintenance(): void {
|
private runMaintenance(): void {
|
||||||
if (this.isDestroyed) return;
|
if (this.isDestroyed || this.writeLock.locked) return;
|
||||||
try {
|
try {
|
||||||
this.flushTelemetry(true);
|
this.flushTelemetry(true);
|
||||||
this.flushNow();
|
this.flushNow();
|
||||||
|
|||||||
@@ -50,6 +50,7 @@ import {
|
|||||||
updateAnimeAnilistInfo,
|
updateAnimeAnilistInfo,
|
||||||
upsertCoverArt,
|
upsertCoverArt,
|
||||||
} from '../query-maintenance.js';
|
} from '../query-maintenance.js';
|
||||||
|
import { deleteMaintenanceBatch } from '../query-delete-maintenance.js';
|
||||||
import { getLocalEpochDay } from '../query-shared.js';
|
import { getLocalEpochDay } from '../query-shared.js';
|
||||||
import { EVENT_CARD_MINED, EVENT_SUBTITLE_LINE, SOURCE_TYPE_LOCAL } from '../types.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);
|
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);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|||||||
@@ -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<void>((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<void>((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);
|
||||||
|
});
|
||||||
@@ -0,0 +1,105 @@
|
|||||||
|
import type { DeleteMaintenanceOperation, DeleteMaintenanceTask } from './delete-maintenance';
|
||||||
|
|
||||||
|
type ResolveDeleteMaintenanceOperation = () =>
|
||||||
|
| DeleteMaintenanceOperation
|
||||||
|
| null
|
||||||
|
| Promise<DeleteMaintenanceOperation | null>;
|
||||||
|
|
||||||
|
interface PendingDeleteMaintenanceRequest {
|
||||||
|
resolveTask: ResolveDeleteMaintenanceOperation;
|
||||||
|
resolve: () => void;
|
||||||
|
reject: (error: unknown) => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface DeleteMaintenanceSchedulerOptions {
|
||||||
|
batchWindowMs: number;
|
||||||
|
runTask: (task: DeleteMaintenanceTask) => Promise<void>;
|
||||||
|
onBusy: () => void;
|
||||||
|
onIdle: () => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
export class DeleteMaintenanceScheduler {
|
||||||
|
private readonly pendingRequests: PendingDeleteMaintenanceRequest[] = [];
|
||||||
|
private running = false;
|
||||||
|
private drainTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
|
private pendingTaskCount = 0;
|
||||||
|
private destroyed = false;
|
||||||
|
|
||||||
|
constructor(private readonly options: DeleteMaintenanceSchedulerOptions) {}
|
||||||
|
|
||||||
|
enqueue(resolveTask: ResolveDeleteMaintenanceOperation): Promise<void> {
|
||||||
|
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<void>((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<void> {
|
||||||
|
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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<string, FakeWorkerListener>();
|
||||||
|
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<typeof createFakeWorker>['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<void>((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<void>((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<void>((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<void>((resolve) => setTimeout(resolve, 0));
|
||||||
|
runtime.destroy();
|
||||||
|
createGate.reject?.(new Error('creation failed'));
|
||||||
|
|
||||||
|
await assert.rejects(result, /shut down/);
|
||||||
|
assert.deepEqual(fallbackTasks, []);
|
||||||
|
});
|
||||||
@@ -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<void>;
|
||||||
|
|
||||||
|
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<number>;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface DeleteMaintenanceWorkerRuntimeOptions {
|
||||||
|
resolveWorkerPath?: () => string | null;
|
||||||
|
createWorker?: (
|
||||||
|
workerPath: string,
|
||||||
|
workerData: { dbPath: string; task: DeleteMaintenanceTask },
|
||||||
|
) => Promise<DeleteMaintenanceWorkerHandle>;
|
||||||
|
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<DeleteMaintenanceWorkerHandle>();
|
||||||
|
private destroyed = false;
|
||||||
|
|
||||||
|
constructor(private readonly options: DeleteMaintenanceWorkerRuntimeOptions = {}) {}
|
||||||
|
|
||||||
|
async run(dbPath: string, task: DeleteMaintenanceTask): Promise<void> {
|
||||||
|
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<void>((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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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 });
|
||||||
|
}
|
||||||
@@ -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<typeof Database>,
|
||||||
|
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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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<number>,
|
||||||
|
videoIds: Set<number>,
|
||||||
|
animeIds: Set<number>,
|
||||||
|
): 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<string, number>
|
||||||
|
>;
|
||||||
|
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<number>();
|
||||||
|
const videoIds = new Set<number>();
|
||||||
|
const animeIds = new Set<number>();
|
||||||
|
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<string>();
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -80,6 +80,14 @@ export function makePlaceholders(values: number[]): string {
|
|||||||
return values.map(() => '?').join(',');
|
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 {
|
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)`;
|
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;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
const placeholders = makePlaceholders(sessionIds);
|
forEachIdChunk(sessionIds, (chunk) => {
|
||||||
|
const placeholders = makePlaceholders(chunk);
|
||||||
db.prepare(`DELETE FROM imm_subtitle_lines WHERE session_id IN (${placeholders})`).run(
|
db.prepare(`DELETE FROM imm_subtitle_lines WHERE session_id IN (${placeholders})`).run(
|
||||||
...sessionIds,
|
...chunk,
|
||||||
);
|
);
|
||||||
db.prepare(`DELETE FROM imm_session_telemetry WHERE session_id IN (${placeholders})`).run(
|
db.prepare(`DELETE FROM imm_session_telemetry WHERE session_id IN (${placeholders})`).run(
|
||||||
...sessionIds,
|
...chunk,
|
||||||
);
|
);
|
||||||
db.prepare(`DELETE FROM imm_session_events WHERE session_id IN (${placeholders})`).run(
|
db.prepare(`DELETE FROM imm_session_events WHERE session_id IN (${placeholders})`).run(
|
||||||
...sessionIds,
|
...chunk,
|
||||||
);
|
);
|
||||||
db.prepare(`DELETE FROM imm_sessions WHERE session_id IN (${placeholders})`).run(...sessionIds);
|
db.prepare(`DELETE FROM imm_sessions WHERE session_id IN (${placeholders})`).run(...chunk);
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
export function toDbMs(ms: number | bigint): bigint {
|
export function toDbMs(ms: number | bigint): bigint {
|
||||||
|
|||||||
Reference in New Issue
Block a user