feat(launcher): add sync command to merge stats and history between machines over SSH

subminer sync <host> exchanges VACUUM INTO snapshots over ssh/scp and each side merges the other's data as an insert-only union keyed on session UUIDs, video keys, series title keys, and word/kanji identity. Lifetime totals and daily/monthly rollups are applied incrementally so pre-retention history survives, remote-only historical rollups are copied, and re-syncing is idempotent. sync --snapshot/--merge expose the underlying steps for manual transfers; a pid-file/mpv-socket guard refuses to run while SubMiner may be writing the database and schema-version mismatches abort the merge.
2026-07-27 16:49:51 -07:00 · 2026-07-08 23:28:10 -07:00
parent cdb1475a54
commit be31f96f02
21 changed files with 2762 additions and 0 deletions
@@ -0,0 +1,99 @@
+import fs from 'node:fs';
+import { Database } from 'bun:sqlite';
+import {
+  LexiconResolver,
+  mergeAnime,
+  mergeExcludedWords,
+  mergeMediaMetadata,
+  mergeVideos,
+} from './merge-catalog.js';
+import { mergeSessions } from './merge-sessions.js';
+import { copyRemoteOnlyRollups, refreshRollupsForNewSessions } from './merge-rollups.js';
+import {
+  assertMergeableSchema,
+  createEmptyMergeSummary,
+  type SyncMergeSummary,
+} from './sync-shared.js';
+
+export type { SyncMergeSummary } from './sync-shared.js';
+export { createDbSnapshot, findLiveStatsDaemonPid } from './sync-shared.js';
+
+/**
+ * Merge a snapshot of another machine's immersion database into the local
+ * one. Insert-only union keyed on natural keys (session_uuid, video_key,
+ * normalized_title_key, word/kanji identity); lifetime and rollup aggregates
+ * are updated incrementally so history older than the session retention
+ * window is preserved on both sides. Idempotent: re-merging the same
+ * snapshot is a no-op.
+ */
+export function mergeSnapshotIntoDb(localDbPath: string, snapshotPath: string): SyncMergeSummary {
+  if (!fs.existsSync(localDbPath)) {
+    throw new Error(`Local stats database not found: ${localDbPath}`);
+  }
+  if (!fs.existsSync(snapshotPath)) {
+    throw new Error(`Snapshot database not found: ${snapshotPath}`);
+  }
+
+  const remote = new Database(snapshotPath, { readonly: true });
+  const local = new Database(localDbPath, { readwrite: true, create: false });
+  try {
+    assertMergeableSchema(remote, 'Snapshot');
+    assertMergeableSchema(local, 'Local');
+
+    const summary = createEmptyMergeSummary();
+    local.run('PRAGMA foreign_keys = ON');
+    local.run('PRAGMA busy_timeout = 5000');
+    local.run('BEGIN IMMEDIATE');
+    try {
+      const animeIdMap = mergeAnime(local, remote, summary);
+      const { videoIdMap, addedVideoIds } = mergeVideos(local, remote, animeIdMap, summary);
+      mergeMediaMetadata(local, remote, videoIdMap, addedVideoIds);
+      mergeExcludedWords(local, remote, summary);
+
+      const lexicon = new LexiconResolver(local, remote, summary);
+      const { newSessionIds } = mergeSessions(
+        local,
+        remote,
+        videoIdMap,
+        animeIdMap,
+        lexicon,
+        summary,
+      );
+      lexicon.applyFrequencyDeltas();
+      refreshRollupsForNewSessions(local, newSessionIds, summary);
+      copyRemoteOnlyRollups(local, remote, videoIdMap, summary);
+
+      local.run('COMMIT');
+      return summary;
+    } catch (error) {
+      local.run('ROLLBACK');
+      throw error;
+    }
+  } finally {
+    local.close();
+    remote.close();
+  }
+}
+
+export function formatMergeSummary(summary: SyncMergeSummary): string {
+  const lines = [
+    `Sessions merged: ${summary.sessionsMerged} (${summary.sessionsAlreadyPresent} already present, ${summary.activeSessionsSkipped} unfinished skipped)`,
+  ];
+  const detail: string[] = [];
+  if (summary.animeAdded) detail.push(`${summary.animeAdded} series`);
+  if (summary.videosAdded) detail.push(`${summary.videosAdded} videos`);
+  if (summary.wordsAdded) detail.push(`${summary.wordsAdded} words`);
+  if (summary.kanjiAdded) detail.push(`${summary.kanjiAdded} kanji`);
+  if (summary.subtitleLinesAdded) detail.push(`${summary.subtitleLinesAdded} subtitle lines`);
+  if (summary.excludedWordsAdded) detail.push(`${summary.excludedWordsAdded} excluded words`);
+  if (detail.length > 0) lines.push(`Added: ${detail.join(', ')}`);
+  if (summary.dailyRollupsCopied || summary.monthlyRollupsCopied) {
+    lines.push(
+      `Historical rollups copied: ${summary.dailyRollupsCopied} daily, ${summary.monthlyRollupsCopied} monthly`,
+    );
+  }
+  if (summary.rollupGroupsRecomputed) {
+    lines.push(`Rollup groups recomputed: ${summary.rollupGroupsRecomputed}`);
+  }
+  return lines.join('\n');
+}