feat(anime): filter episodes and mark them watched by hand

- Add a filter box above the episode list for a number, range, or name substring, with a "N of M" counter
- Read watch marks from the immersion tracker stats and show them per episode, with a watched count in the header, refreshed on window focus
- Add a right-click menu to mark one episode or a whole catch-up span (this and everything below) watched/unwatched
- Add setWatched/getWatchState IPC plumbing so a manual mark creates the stats row for an episode that was never played

src/core/services/immersion-tracker-service.ts

@@ -91,6 +91,10 @@ import {
   markVideoWatched,
   upsertCoverArt,
 } from './immersion-tracker/query-maintenance';
+import {
+  getVideoIdByVideoKey,
+  getWatchStateByVideoKeys,
+} from './immersion-tracker/query-watch-state';
 import { repairJellyfinStreamVideoLinks } from './immersion-tracker/jellyfin-link-repair';
 import {
   repairLegacySeasonlessAnimeRows,
@@ -338,6 +342,15 @@ export interface StreamPlaybackMetadataInput {
   episodeNumber: number | null;
 }
 
+/** What a caller needs to show "already watched" against a streamed episode. */
+export interface StreamWatchState {
+  /** Set once a session passed the completion threshold, or marked by hand. */
+  watched: boolean;
+  /** Start of the most recent session, or null when it was never played. */
+  lastWatchedMs: number | null;
+  sessionCount: number;
+}
+
 /**
  * Parser sources that are recorded before playback starts. A video carrying one
  * already has better metadata than filename guessing could produce, so the
@@ -722,6 +735,77 @@ export class ImmersionTrackerService {
     markVideoWatched(this.db, videoId, watched);
   }
 
+  /**
+   * Set the watch mark on streamed episodes by hand.
+   *
+   * Marking watched creates the video row when the episode was never played, so
+   * a series watched elsewhere can be caught up on; the row carries the same
+   * series/season/episode metadata playback would have recorded. Both library
+   * views join the lifetime tables, so a row created this way stays out of the
+   * stats lists until it is actually watched.
+   *
+   * Clearing a mark never creates anything: with no row there is nothing to
+   * clear.
+   */
+  async setStreamWatchState(
+    episodes: StreamPlaybackMetadataInput[],
+    watched: boolean,
+  ): Promise<number> {
+    let changed = 0;
+    // Every row this creates would otherwise rebuild the lifetime summaries on
+    // its own, and a season's worth of episodes arrives in one call.
+    let needsLifetimeRebuild = false;
+    for (const episode of episodes) {
+      const statsPath = normalizeMediaPath(episode.statsPath);
+      if (!statsPath) continue;
+      if (watched) {
+        needsLifetimeRebuild =
+          this.recordStreamPlaybackMetadata(episode, { deferLifetimeRebuild: true }) ||
+          needsLifetimeRebuild;
+      }
+
+      const videoId = getVideoIdByVideoKey(this.db, buildVideoKey(statsPath, SOURCE_TYPE_REMOTE));
+      if (videoId === null) continue;
+      markVideoWatched(this.db, videoId, watched);
+      changed += 1;
+
+      // Clearing the mark on what is playing right now would otherwise be undone
+      // the moment the session passes the completion threshold again.
+      if (!watched && this.sessionState?.videoId === videoId) {
+        this.sessionState.markedWatched = true;
+      }
+    }
+
+    if (needsLifetimeRebuild) rebuildLifetimeSummaryTables(this.db);
+    return changed;
+  }
+
+  /**
+   * Watch state for streamed episodes, keyed by the stats path the anime
+   * browser derives for each one. Paths never played are absent from the map,
+   * so a caller can treat "missing" as unwatched without a probe per episode.
+   */
+  async getStreamWatchState(statsPaths: string[]): Promise<Map<string, StreamWatchState>> {
+    const byKey = new Map<string, string>();
+    for (const path of statsPaths) {
+      const normalized = normalizeMediaPath(path);
+      if (!normalized) continue;
+      byKey.set(buildVideoKey(normalized, SOURCE_TYPE_REMOTE), normalized);
+    }
+
+    const state = new Map<string, StreamWatchState>();
+    for (const row of getWatchStateByVideoKeys(this.db, [...byKey.keys()])) {
+      const statsPath = byKey.get(row.videoKey);
+      if (!statsPath) continue;
+      state.set(statsPath, {
+        watched: row.watched,
+        lastWatchedMs: row.lastWatchedMs,
+        sessionCount: row.sessionCount,
+      });
+    }
+    return state;
+  }
+
   async markActiveVideoWatched(): Promise<boolean> {
     if (!this.sessionState) return false;
     markVideoWatched(this.db, this.sessionState.videoId, true);
@@ -1310,23 +1394,29 @@ export class ImmersionTrackerService {
     });
   }
 
-  recordStreamPlaybackMetadata(metadata: StreamPlaybackMetadataInput): void {
+  /** Returns whether the lifetime summaries still need rebuilding; see
+   * `recordPrePlaybackMetadata` for why a batch defers that. */
+  recordStreamPlaybackMetadata(
+    metadata: StreamPlaybackMetadataInput,
+    options: { deferLifetimeRebuild?: boolean } = {},
+  ): boolean {
     const rawPath = normalizeMediaPath(metadata.mediaPath);
     const statsPath = normalizeMediaPath(metadata.statsPath) || rawPath;
     if (!statsPath) {
-      return;
+      return false;
     }
     const seriesTitle = normalizeText(metadata.seriesTitle);
     const displayTitle =
       normalizeText(metadata.displayTitle) || seriesTitle || deriveCanonicalTitle(statsPath);
     const libraryTitle = seriesTitle || displayTitle;
     if (!libraryTitle) {
-      return;
+      return false;
     }
     const seasonNumber = normalizeMetadataInt(metadata.seasonNumber);
     const episodeNumber = normalizeMetadataInt(metadata.episodeNumber);
 
-    this.recordPrePlaybackMetadata({
+    return this.recordPrePlaybackMetadata({
+      deferLifetimeRebuild: options.deferLifetimeRebuild,
       statsPath,
       aliases: rawPath ? buildMediaPathAliasCandidates(rawPath) : [],
       displayTitle,
@@ -1347,6 +1437,10 @@ export class ImmersionTrackerService {
   /**
    * Creates the video row and its series link ahead of playback, so the session
    * mpv's path change starts already belongs to the right anime.
+   *
+   * Returns whether the lifetime summaries need rebuilding. A new row always
+   * does, so a caller recording a batch passes `deferLifetimeRebuild` and
+   * rebuilds once at the end rather than once per episode.
    */
   private recordPrePlaybackMetadata(params: {
     statsPath: string;
@@ -1357,7 +1451,8 @@ export class ImmersionTrackerService {
     episodeNumber: number | null;
     parserSource: string;
     metadataJson: string;
-  }): void {
+    deferLifetimeRebuild?: boolean;
+  }): boolean {
     for (const alias of params.aliases) {
       this.mediaPathAliases.set(alias, params.statsPath);
     }
@@ -1399,9 +1494,13 @@ export class ImmersionTrackerService {
     const hasLifetimeMedia = Boolean(
       this.db.prepare('SELECT 1 FROM imm_lifetime_media WHERE video_id = ?').get(videoId),
     );
-    if (hasLifetimeMedia || (previousLink && previousLink.animeId !== animeId)) {
+    const needsRebuild = Boolean(
+      hasLifetimeMedia || (previousLink && previousLink.animeId !== animeId),
+    );
+    if (needsRebuild && !params.deferLifetimeRebuild) {
       rebuildLifetimeSummaryTables(this.db);
     }
+    return needsRebuild;
   }
 
   private hasPrePlaybackMetadata(videoId: number): boolean {

src/core/services/immersion-tracker-watch-marks.test.ts

@@ -0,0 +1,127 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+
+type ImmersionTrackerService = import('./immersion-tracker-service').ImmersionTrackerService;
+
+const POLICY = {
+  batchSize: 10,
+  flushIntervalMs: 5_000,
+  queueCap: 100,
+  payloadCapBytes: 512,
+  maintenanceIntervalMs: 60 * 60 * 1000,
+  retention: {
+    eventsDays: 14,
+    telemetryDays: 45,
+    sessionsDays: 60,
+    dailyRollupsDays: 730,
+    monthlyRollupsDays: 3650,
+    vacuumIntervalDays: 14,
+  },
+};
+
+async function createTracker(): Promise<{ tracker: ImmersionTrackerService; dir: string }> {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-watch-marks-'));
+  const { ImmersionTrackerService: Ctor } = await import('./immersion-tracker-service');
+  return { tracker: new Ctor({ dbPath: path.join(dir, 'immersion.sqlite'), policy: POLICY }), dir };
+}
+
+function episode(statsPath: string, episodeNumber: number) {
+  return {
+    mediaPath: '',
+    statsPath,
+    displayTitle: `Test Series S03E0${episodeNumber}`,
+    seriesTitle: 'Test Series',
+    seasonNumber: 3,
+    episodeNumber,
+  };
+}
+
+const EP1 = 'animebrowser://src/anime/ep1';
+const EP2 = 'animebrowser://src/anime/ep2';
+
+test('marking an episode nobody played records it, and clearing it takes the mark away', async () => {
+  const { tracker, dir } = await createTracker();
+  try {
+    assert.equal(await tracker.setStreamWatchState([episode(EP1, 1), episode(EP2, 2)], true), 2);
+
+    const marked = await tracker.getStreamWatchState([EP1, EP2]);
+    assert.equal(marked.get(EP1)?.watched, true);
+    assert.equal(marked.get(EP2)?.watched, true);
+    // Nothing was played, so there is no session behind the mark.
+    assert.equal(marked.get(EP1)?.sessionCount, 0);
+    assert.equal(marked.get(EP1)?.lastWatchedMs, null);
+
+    assert.equal(await tracker.setStreamWatchState([episode(EP1, 1)], false), 1);
+    const cleared = await tracker.getStreamWatchState([EP1, EP2]);
+    assert.equal(cleared.get(EP1)?.watched, false);
+    assert.equal(cleared.get(EP2)?.watched, true);
+  } finally {
+    tracker.destroy();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('clearing a mark on an episode with no history creates nothing', async () => {
+  const { tracker, dir } = await createTracker();
+  try {
+    assert.equal(await tracker.setStreamWatchState([episode(EP1, 1)], false), 0);
+    assert.equal((await tracker.getStreamWatchState([EP1])).size, 0);
+  } finally {
+    tracker.destroy();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('a manual mark stays out of the stats library until the episode is watched', async () => {
+  const { tracker, dir } = await createTracker();
+  try {
+    await tracker.setStreamWatchState([episode(EP1, 1)], true);
+    // Both library views join the lifetime tables, which only playback fills.
+    assert.deepEqual(await tracker.getMediaLibrary(), []);
+    assert.deepEqual(await tracker.getAnimeLibrary(), []);
+  } finally {
+    tracker.destroy();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('a batch rebuilds the lifetime summaries once, not once per episode', async () => {
+  const { tracker, dir } = await createTracker();
+  const privateApi = tracker as unknown as {
+    recordPrePlaybackMetadata: (params: { deferLifetimeRebuild?: boolean }) => boolean;
+  };
+  const original = privateApi.recordPrePlaybackMetadata.bind(tracker);
+  let deferred = 0;
+  let immediate = 0;
+  privateApi.recordPrePlaybackMetadata = (params) => {
+    if (params.deferLifetimeRebuild) deferred += 1;
+    else immediate += 1;
+    return original(params);
+  };
+
+  try {
+    const season = Array.from({ length: 12 }, (_, index) =>
+      episode(`animebrowser://src/anime/batch-${index}`, index + 1),
+    );
+    await tracker.setStreamWatchState(season, true);
+
+    assert.equal(deferred, 12);
+    assert.equal(immediate, 0, 'every row in the batch defers its rebuild to the end of the call');
+  } finally {
+    tracker.destroy();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('an episode with no stats path is skipped rather than recorded as unknown', async () => {
+  const { tracker, dir } = await createTracker();
+  try {
+    assert.equal(await tracker.setStreamWatchState([episode('  ', 1)], true), 0);
+  } finally {
+    tracker.destroy();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});

src/core/services/immersion-tracker/__tests__/query-watch-state.test.ts

@@ -0,0 +1,91 @@
+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 { Database } from '../sqlite.js';
+import { ensureSchema, getOrCreateVideoRecord } from '../storage.js';
+import { startSessionRecord } from '../session.js';
+import { markVideoWatched } from '../query-maintenance.js';
+import { getWatchStateByVideoKeys } from '../query-watch-state.js';
+import { SOURCE_TYPE_REMOTE } from '../types.js';
+
+function createDb() {
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-imm-watch-state-test-'));
+  const dbPath = path.join(dir, 'immersion.sqlite');
+  const db = new Database(dbPath);
+  ensureSchema(db);
+  return { db, dir };
+}
+
+function addStreamVideo(db: ReturnType<typeof createDb>['db'], statsPath: string): number {
+  return getOrCreateVideoRecord(db, `remote:${statsPath}`, {
+    canonicalTitle: statsPath,
+    sourcePath: null,
+    sourceUrl: statsPath,
+    sourceType: SOURCE_TYPE_REMOTE,
+  });
+}
+
+test('getWatchStateByVideoKeys reports watched marks and the newest session', () => {
+  const { db, dir } = createDb();
+  try {
+    const watchedPath = 'animebrowser://src/anime/ep1';
+    const startedPath = 'animebrowser://src/anime/ep2';
+    const watchedId = addStreamVideo(db, watchedPath);
+    const startedId = addStreamVideo(db, startedPath);
+
+    startSessionRecord(db, watchedId, 1_000_000);
+    startSessionRecord(db, watchedId, 3_000_000);
+    startSessionRecord(db, startedId, 2_000_000);
+    markVideoWatched(db, watchedId, true);
+
+    const rows = getWatchStateByVideoKeys(db, [
+      `remote:${watchedPath}`,
+      `remote:${startedPath}`,
+      'remote:animebrowser://src/anime/never-played',
+    ]);
+    const byKey = new Map(rows.map((row) => [row.videoKey, row]));
+
+    assert.equal(rows.length, 2, 'a key with no video row comes back absent, not unwatched');
+    assert.deepEqual(byKey.get(`remote:${watchedPath}`), {
+      videoKey: `remote:${watchedPath}`,
+      watched: true,
+      lastWatchedMs: 3_000_000,
+      sessionCount: 2,
+    });
+    // Started but never finished: a row exists, the watch mark does not.
+    assert.equal(byKey.get(`remote:${startedPath}`)?.watched, false);
+    assert.equal(byKey.get(`remote:${startedPath}`)?.lastWatchedMs, 2_000_000);
+  } finally {
+    db.close();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('getWatchStateByVideoKeys handles a video that was never played', () => {
+  const { db, dir } = createDb();
+  try {
+    const statsPath = 'animebrowser://src/anime/ep3';
+    addStreamVideo(db, statsPath);
+    const [row] = getWatchStateByVideoKeys(db, [`remote:${statsPath}`]);
+    assert.equal(row?.lastWatchedMs, null);
+    assert.equal(row?.sessionCount, 0);
+  } finally {
+    db.close();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});
+
+test('getWatchStateByVideoKeys ignores empty keys and dedupes the rest', () => {
+  const { db, dir } = createDb();
+  try {
+    const statsPath = 'animebrowser://src/anime/ep4';
+    addStreamVideo(db, statsPath);
+    const rows = getWatchStateByVideoKeys(db, ['', `remote:${statsPath}`, `remote:${statsPath}`]);
+    assert.equal(rows.length, 1);
+  } finally {
+    db.close();
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});

src/core/services/immersion-tracker/query-watch-state.ts

@@ -0,0 +1,79 @@
+import type { DatabaseSync } from './sqlite';
+import { fromDbTimestamp } from './query-shared';
+
+/** Watch state of one video row, addressed by the key playback recorded it under. */
+export interface VideoWatchStateRow {
+  videoKey: string;
+  watched: boolean;
+  /** Start of the most recent session on this video, or null when never played. */
+  lastWatchedMs: number | null;
+  sessionCount: number;
+}
+
+/** The video row a key belongs to, or null when nothing has recorded it yet. */
+export function getVideoIdByVideoKey(db: DatabaseSync, videoKey: string): number | null {
+  const row = db.prepare('SELECT video_id FROM imm_videos WHERE video_key = ?').get(videoKey) as {
+    video_id: number;
+  } | null;
+  return row?.video_id ?? null;
+}
+
+/**
+ * How many keys one statement binds. SQLite caps parameters per statement, and
+ * an episode list can be long, so the lookup runs in chunks.
+ */
+const CHUNK_SIZE = 400;
+
+/**
+ * Look up watch state for a set of video keys.
+ *
+ * Keys that have never been played simply do not come back — the caller treats
+ * a missing key as unwatched rather than needing a row for it.
+ *
+ * `last_watched_ms` on the lifetime tables is rebuilt in batches and can lag, so
+ * the timestamp comes from the sessions themselves. Timestamps are epoch
+ * milliseconds stored as text, hence the cast before `MAX`.
+ */
+export function getWatchStateByVideoKeys(
+  db: DatabaseSync,
+  videoKeys: string[],
+): VideoWatchStateRow[] {
+  const unique = [...new Set(videoKeys.filter((key) => key.length > 0))];
+  const rows: VideoWatchStateRow[] = [];
+
+  for (let start = 0; start < unique.length; start += CHUNK_SIZE) {
+    const chunk = unique.slice(start, start + CHUNK_SIZE);
+    const placeholders = chunk.map(() => '?').join(', ');
+    const chunkRows = db
+      .prepare(
+        `
+    SELECT
+      v.video_key AS videoKey,
+      v.watched AS watched,
+      MAX(CAST(s.started_at_ms AS INTEGER)) AS lastWatchedMs,
+      COUNT(s.session_id) AS sessionCount
+    FROM imm_videos v
+    LEFT JOIN imm_sessions s ON s.video_id = v.video_id
+    WHERE v.video_key IN (${placeholders})
+    GROUP BY v.video_id
+  `,
+      )
+      .all(...chunk) as Array<{
+      videoKey: string;
+      watched: number;
+      lastWatchedMs: number | string | null;
+      sessionCount: number;
+    }>;
+
+    for (const row of chunkRows) {
+      rows.push({
+        videoKey: row.videoKey,
+        watched: row.watched === 1,
+        lastWatchedMs: fromDbTimestamp(row.lastWatchedMs),
+        sessionCount: Number(row.sessionCount) || 0,
+      });
+    }
+  }
+
+  return rows;
+}