diff --git a/src/core/services/immersion-tracker-service.test.ts b/src/core/services/immersion-tracker-service.test.ts index f7dbf77d..a40936e6 100644 --- a/src/core/services/immersion-tracker-service.test.ts +++ b/src/core/services/immersion-tracker-service.test.ts @@ -2498,6 +2498,63 @@ test('Jellyfin link repair removes merged leaked anime rows and sanitizes orphan } }); +test('mergeAnime drains a queue larger than one batch before rebuilding summaries', async () => { + const dbPath = makeDbPath(); + let tracker: ImmersionTrackerService | null = null; + + try { + const Ctor = await loadTrackerCtor(); + // A batch size well below the queued write count: one flushNow() pass would + // leave the forced telemetry sample, appended last, unwritten. + tracker = new Ctor({ dbPath, policy: { batchSize: 2 } }); + const privateApi = tracker as unknown as { + db: DatabaseSync; + queue: unknown[]; + mergeAnime: (targetAnimeId: number, sourceAnimeIds: number[]) => Promise; + }; + + privateApi.db.exec(` + INSERT INTO imm_anime (anime_id, normalized_title_key, canonical_title, CREATED_DATE, LAST_UPDATE_DATE) + VALUES (1, 'show', 'Show', 1000, 1000), (2, 'show season 1', 'Show Season 1', 1000, 1000); + INSERT INTO imm_videos (video_id, video_key, canonical_title, anime_id, source_type, watched, duration_ms, CREATED_DATE, LAST_UPDATE_DATE) + VALUES (1, 'local:/tmp/a.mkv', 'A', 1, 1, 0, 1440000, 1000, 1000), + (2, 'local:/tmp/b.mkv', 'B', 2, 1, 0, 1440000, 1000, 1000); + INSERT INTO imm_sessions (session_id, session_uuid, video_id, started_at_ms, ended_at_ms, status, active_watched_ms, CREATED_DATE, LAST_UPDATE_DATE) + VALUES (1, 'drain-session', 2, '1000', '2000', 2, 1000, 1000, 2000); + `); + + for (let index = 0; index < 8; index += 1) { + (tracker as unknown as { recordWrite: (write: Record) => void }).recordWrite( + { + kind: 'subtitleLine', + sessionId: 1, + videoId: 2, + lineIndex: index, + segmentStartMs: index * 1000, + segmentEndMs: index * 1000 + 900, + text: `line ${index}`, + wordOccurrences: [], + kanjiOccurrences: [], + firstSeen: 1000, + lastSeen: 2000, + }, + ); + } + assert.ok(privateApi.queue.length > 2, 'expected more queued writes than one batch'); + + await privateApi.mergeAnime(1, [2]); + + assert.equal(privateApi.queue.length, 0); + const lines = privateApi.db + .prepare('SELECT COUNT(*) AS total FROM imm_subtitle_lines WHERE anime_id = 1') + .get() as { total: number }; + assert.equal(Number(lines.total), 8); + } finally { + tracker?.destroy(); + cleanupDbPath(dbPath); + } +}); + test('applies configurable queue, flush, and retention policy', 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 ce2464ec..95dfd42b 100644 --- a/src/core/services/immersion-tracker-service.ts +++ b/src/core/services/immersion-tracker-service.ts @@ -602,8 +602,7 @@ export class ImmersionTrackerService { } async rebuildLifetimeSummaries(): Promise { - this.flushTelemetry(true); - this.flushNow(); + this.drainWriteQueue('rebuilding lifetime summaries'); return rebuildLifetimeSummaryTables(this.db); } @@ -772,21 +771,47 @@ export class ImmersionTrackerService { if (pendingVideoId !== undefined) { await this.pendingAnimeMetadataUpdates.get(pendingVideoId); } - // Both of these rebuild the lifetime summaries, which recompute from the - // database: queued telemetry has to land first or the active session's - // watch time is dropped from the merged totals. - this.flushTelemetry(true); - this.flushNow(); + // This rebuilds the lifetime summaries, which recompute from the database: + // queued writes have to land first or the active session is dropped from + // the merged totals. + this.drainWriteQueue('merging library entries'); return mergeAnimeRecords(this.db, targetAnimeId, sourceAnimeIds); } async moveVideoToAnime(videoId: number, targetAnimeId: number): Promise { await this.pendingAnimeMetadataUpdates.get(videoId); - this.flushTelemetry(true); - this.flushNow(); + this.drainWriteQueue('moving an episode'); return moveVideoToAnimeQuery(this.db, videoId, targetAnimeId); } + /** + * Persist every queued write before a caller recomputes summaries from the + * database. + * + * A single `flushNow()` is not enough: forced telemetry is appended to the + * back of the queue while `flushNow()` writes at most `batchSize` entries off + * the front, so a busy session leaves the newest sample unwritten. Stops as + * soon as a pass makes no progress โ€” a rolled-back batch is pushed back onto + * the queue, and looping on that would spin forever. + * + * Returns false when the queue could not be emptied, in which case the + * rebuild runs against a database still missing those writes. + */ + private drainWriteQueue(context: string): boolean { + this.flushTelemetry(true); + while (this.queue.length > 0) { + const pending = this.queue.length; + this.flushNow(); + if (this.queue.length >= pending) { + this.logger.warn( + `Immersion tracker queue did not drain before ${context}; summaries may lag by ${this.queue.length} writes`, + ); + return false; + } + } + return true; + } + async reassignAnimeAnilist( animeId: number, info: { diff --git a/stats/src/components/anime/AnimeMergeDialog.tsx b/stats/src/components/anime/AnimeMergeDialog.tsx index a9319c56..b5d26eff 100644 --- a/stats/src/components/anime/AnimeMergeDialog.tsx +++ b/stats/src/components/anime/AnimeMergeDialog.tsx @@ -1,4 +1,4 @@ -import { useState } from 'react'; +import { useId, useState } from 'react'; import { apiClient } from '../../lib/api-client'; import { formatDuration, formatNumber } from '../../lib/formatters'; import { AnimeCoverImage } from './AnimeCoverImage'; @@ -19,6 +19,7 @@ function pickDefaultKeeper(entries: AnimeLibraryItem[]): number { } export function AnimeMergeDialog({ entries, onClose, onMerged }: AnimeMergeDialogProps) { + const headingId = useId(); const [keeperId, setKeeperId] = useState(() => pickDefaultKeeper(entries)); const [merging, setMerging] = useState(false); const [error, setError] = useState(null); @@ -57,18 +58,22 @@ export function AnimeMergeDialog({ entries, onClose, onMerged }: AnimeMergeDialo >
e.stopPropagation()} >
-

+

Merge {entries.length} Library Entries

- {error ?
{error}
: null} + {error ? ( +
+ {error} +
+ ) : null}
Result: {totalEpisodes} episode{totalEpisodes !== 1 ? 's' : ''} ยท{' '} diff --git a/stats/src/components/anime/LibraryEntryPicker.tsx b/stats/src/components/anime/LibraryEntryPicker.tsx index dd914009..3de42ef4 100644 --- a/stats/src/components/anime/LibraryEntryPicker.tsx +++ b/stats/src/components/anime/LibraryEntryPicker.tsx @@ -1,4 +1,4 @@ -import { useEffect, useMemo, useRef, useState } from 'react'; +import { useEffect, useId, useMemo, useRef, useState } from 'react'; import { apiClient } from '../../lib/api-client'; import { formatDuration } from '../../lib/formatters'; import { AnimeCoverImage } from './AnimeCoverImage'; @@ -28,6 +28,8 @@ export function LibraryEntryPicker({ const [loadFailed, setLoadFailed] = useState(false); const [query, setQuery] = useState(initialQuery); const inputRef = useRef(null); + const headingId = useId(); + const searchId = useId(); useEffect(() => { inputRef.current?.focus(); @@ -62,40 +64,56 @@ export function LibraryEntryPicker({
e.stopPropagation()} >
-

{heading}

+

+ {heading} +

+ setQuery(e.target.value)} placeholder="Search library..." className="w-full bg-ctp-surface0 border border-ctp-surface1 rounded-lg px-3 py-2 text-sm text-ctp-text placeholder:text-ctp-overlay2 focus:outline-none focus:border-ctp-blue" /> - {error ?
{error}
: null} + {error ? ( +
+ {error} +
+ ) : null}
{entries === null &&
Loading...
} {loadFailed && ( -
+
Could not load the library. Close this dialog and try again.
)} {!loadFailed && entries !== null && visible.length === 0 && ( -
No other titles
+
+ {query.trim() ? 'No matches' : 'No other titles'} +
)} {visible.map((entry) => (