From 04095eebf7d15a0045a75e4897aadcc165f96b26 Mon Sep 17 00:00:00 2001 From: sudacode Date: Sat, 11 Jul 2026 20:14:28 -0700 Subject: [PATCH] refactor(sync): extract stats-sync engine behind a DB-driver interface Move snapshot/merge/quiescence engine and ssh helpers from launcher/sync into src/core/services/stats-sync, parameterized on a minimal SyncDb driver. The launcher binds it to bun:sqlite via launcher/sync/bun-driver; the sync command becomes a thin adapter over the shared sync flow. --- launcher/commands/sync-command.ts | 349 ++--------------- launcher/history-db.ts | 47 +-- launcher/sync/bun-driver.ts | 23 ++ launcher/sync/ssh.ts | 116 +----- launcher/sync/sync-db.ts | 105 +---- launcher/sync/sync-shared.ts | 184 ++------- src/core/services/stats-sync/driver.ts | 42 ++ .../services/stats-sync}/merge-catalog.ts | 123 +++--- .../services/stats-sync}/merge-rollups.ts | 52 ++- .../services/stats-sync}/merge-sessions.ts | 165 ++++---- src/core/services/stats-sync/merge.ts | 108 +++++ src/core/services/stats-sync/shared.ts | 157 ++++++++ src/core/services/stats-sync/ssh.ts | 130 ++++++ src/core/services/stats-sync/sync-flow.ts | 370 ++++++++++++++++++ src/core/services/stats-sync/wal-retry.ts | 50 +++ 15 files changed, 1124 insertions(+), 897 deletions(-) create mode 100644 launcher/sync/bun-driver.ts create mode 100644 src/core/services/stats-sync/driver.ts rename {launcher/sync => src/core/services/stats-sync}/merge-catalog.ts (80%) rename {launcher/sync => src/core/services/stats-sync}/merge-rollups.ts (89%) rename {launcher/sync => src/core/services/stats-sync}/merge-sessions.ts (80%) create mode 100644 src/core/services/stats-sync/merge.ts create mode 100644 src/core/services/stats-sync/shared.ts create mode 100644 src/core/services/stats-sync/ssh.ts create mode 100644 src/core/services/stats-sync/sync-flow.ts create mode 100644 src/core/services/stats-sync/wal-retry.ts diff --git a/launcher/commands/sync-command.ts b/launcher/commands/sync-command.ts index 2e6acc46..7320d027 100644 --- a/launcher/commands/sync-command.ts +++ b/launcher/commands/sync-command.ts @@ -1,7 +1,6 @@ import fs from 'node:fs'; -import os from 'node:os'; -import path from 'node:path'; import { fail, log } from '../log.js'; +import type { LogLevel } from '../types.js'; import { resolveImmersionDbPath } from '../history-db.js'; import { createDbSnapshot, @@ -14,74 +13,26 @@ import { resolveRemoteSubminerCommand, runScp, runSsh, - shellQuote, } from '../sync/ssh.js'; import { resolvePathMaybe } from '../util.js'; import { canConnectUnixSocket } from '../mpv.js'; import { recordHostSyncResultToDisk } from '../sync/sync-hosts.js'; +import { + ensureTrackerQuiescentFlow, + runSyncFlow, + runCheckMode as runCheckModeFlow, + runHostSync as runHostSyncFlow, + runMergeMode as runMergeModeFlow, + runSnapshotMode as runSnapshotModeFlow, + type SyncFlowContext, + type SyncFlowDeps, +} from '../../src/core/services/stats-sync/sync-flow.js'; import type { LauncherCommandContext } from './context.js'; -import type { RemoteRunResult } from '../sync/ssh.js'; -import type { SyncMergeSummary, SyncProgressEvent } from '../../src/shared/sync/sync-events.js'; -import type { SyncResultStatus } from '../../src/shared/sync/sync-hosts-store.js'; -export interface SyncCommandDeps { - createDbSnapshot: typeof createDbSnapshot; - mergeSnapshotIntoDb: typeof mergeSnapshotIntoDb; - formatMergeSummary: typeof formatMergeSummary; - findLiveStatsDaemonPid: typeof findLiveStatsDaemonPid; - assertSafeSshHost: typeof assertSafeSshHost; - resolveRemoteSubminerCommand: typeof resolveRemoteSubminerCommand; - runScp: typeof runScp; - runSsh: typeof runSsh; - fail: typeof fail; - log: typeof log; - canConnectUnixSocket: typeof canConnectUnixSocket; - realpathSync: typeof fs.realpathSync; - mkdtempSync: typeof fs.mkdtempSync; - rmSync: typeof fs.rmSync; - consoleLog: typeof console.log; - writeStdout: typeof process.stdout.write; - ensureTrackerQuiescent: (context: LauncherCommandContext, dbPath: string) => Promise; - emitEvent: (event: SyncProgressEvent) => void; - recordHostSyncResult: (host: string, status: SyncResultStatus, detail: string | null) => void; -} - -function resolveDbPath(context: LauncherCommandContext): string { - const override = context.args.syncDbPath.trim(); - return override ? resolvePathMaybe(override) : resolveImmersionDbPath(); -} - -function isTrackerDb(dbPath: string, deps: SyncCommandDeps): boolean { - const trackerDbPath = resolveImmersionDbPath(); - try { - return deps.realpathSync(dbPath) === deps.realpathSync(trackerDbPath); - } catch { - return dbPath === trackerDbPath; - } -} - -export async function ensureTrackerQuiescent( - context: LauncherCommandContext, - dbPath: string, - inputDeps: Partial = {}, -): Promise { - const deps = resolveSyncCommandDeps(inputDeps); - if (context.args.syncForce) return; - // A running SubMiner only holds the tracker's own database; --db pointed - // elsewhere needs no guard. - if (!isTrackerDb(dbPath, deps)) return; - const daemonPid = deps.findLiveStatsDaemonPid(dbPath); - if (daemonPid !== null) { - deps.fail( - `The SubMiner stats server is running (pid ${daemonPid}). Stop it with "subminer stats -s" (or close SubMiner) before syncing, or pass --force.`, - ); - } - if (context.mpvSocketPath && (await deps.canConnectUnixSocket(context.mpvSocketPath))) { - deps.fail( - `An mpv/SubMiner session appears to be running (socket ${context.mpvSocketPath}). Close it before syncing, or pass --force.`, - ); - } -} +// The sync flow itself lives in src/core/services/stats-sync (shared with the +// app's --sync-cli mode); this module binds it to the launcher's bun:sqlite +// engine, logger, and config paths. +export type SyncCommandDeps = SyncFlowDeps; const defaultSyncCommandDeps: SyncCommandDeps = { createDbSnapshot, @@ -93,7 +44,7 @@ const defaultSyncCommandDeps: SyncCommandDeps = { runScp, runSsh, fail, - log, + log: (level, configured, message) => log(level as LogLevel, configured as LogLevel, message), canConnectUnixSocket, realpathSync: fs.realpathSync, mkdtempSync: fs.mkdtempSync, @@ -103,22 +54,20 @@ const defaultSyncCommandDeps: SyncCommandDeps = { ensureTrackerQuiescent: async (context, dbPath) => ensureTrackerQuiescent(context, dbPath), emitEvent: () => {}, recordHostSyncResult: recordHostSyncResultToDisk, + resolveDefaultDbPath: resolveImmersionDbPath, + resolvePath: resolvePathMaybe, }; function resolveSyncCommandDeps(inputDeps: Partial = {}): SyncCommandDeps { return { ...defaultSyncCommandDeps, ...inputDeps }; } -// In --json mode every line on stdout is an NDJSON event: human console output -// is silenced and events are written through the original console logger. -function withJsonEvents(deps: SyncCommandDeps): SyncCommandDeps { - const writeLine = deps.consoleLog; - return { - ...deps, - consoleLog: () => {}, - writeStdout: (() => true) as typeof process.stdout.write, - emitEvent: (event) => writeLine(JSON.stringify(event)), - }; +export async function ensureTrackerQuiescent( + context: SyncFlowContext, + dbPath: string, + inputDeps: Partial = {}, +): Promise { + await ensureTrackerQuiescentFlow(context, dbPath, resolveSyncCommandDeps(inputDeps)); } export function runSnapshotMode( @@ -126,16 +75,7 @@ export function runSnapshotMode( dbPath: string, inputDeps: Partial = {}, ): void { - const deps = resolveSyncCommandDeps(inputDeps); - const outPath = resolvePathMaybe(context.args.syncSnapshotPath); - deps.emitEvent({ - type: 'stage', - stage: 'snapshot-local', - message: `Snapshotting local database (${dbPath})`, - }); - deps.createDbSnapshot(dbPath, outPath); - deps.emitEvent({ type: 'snapshot-created', path: outPath }); - deps.consoleLog(outPath); + runSnapshotModeFlow(context, dbPath, resolveSyncCommandDeps(inputDeps)); } export async function runMergeMode( @@ -143,84 +83,14 @@ export async function runMergeMode( dbPath: string, inputDeps: Partial = {}, ): Promise { - const deps = resolveSyncCommandDeps(inputDeps); - await deps.ensureTrackerQuiescent(context, dbPath); - const snapshotPath = resolvePathMaybe(context.args.syncMergePath); - deps.emitEvent({ - type: 'stage', - stage: 'merge-local', - message: `Merging ${snapshotPath} into the local database`, - }); - const summary = deps.mergeSnapshotIntoDb(dbPath, snapshotPath); - deps.emitEvent({ type: 'merge-summary', target: 'local', summary }); - deps.consoleLog(deps.formatMergeSummary(summary)); -} - -function formatHostSyncDetail( - direction: 'both' | 'push' | 'pull', - pulledSummary: SyncMergeSummary | null, -): string { - if (!pulledSummary) return direction === 'push' ? 'Pushed local stats' : 'Sync complete'; - const merged = `${pulledSummary.sessionsMerged} session${pulledSummary.sessionsMerged === 1 ? '' : 's'} merged`; - return direction === 'pull' ? merged : `${merged}; pushed local stats`; + await runMergeModeFlow(context, dbPath, resolveSyncCommandDeps(inputDeps)); } export async function runCheckMode( context: LauncherCommandContext, inputDeps: Partial = {}, ): Promise { - const deps = resolveSyncCommandDeps(inputDeps); - const { args } = context; - const host = args.syncHost; - deps.assertSafeSshHost(host); - - deps.consoleLog(`Checking SSH connection to ${host}...`); - let remoteCommand: string | null = null; - let remoteVersion: string | null = null; - let error: string | null = null; - - const probe = deps.runSsh(host, 'echo subminer-check-ok'); - const sshOk = probe.status === 0 && probe.stdout.includes('subminer-check-ok'); - if (!sshOk) { - error = formatRemoteRunError(`Could not reach ${host} over SSH.`, probe); - } else { - deps.consoleLog('SSH connection: ok'); - try { - remoteCommand = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null); - const version = deps.runSsh(host, `${remoteCommand} --version`); - remoteVersion = version.status === 0 ? version.stdout.trim() || null : null; - deps.consoleLog( - `Remote subminer: ${remoteCommand}${remoteVersion ? ` (${remoteVersion})` : ''}`, - ); - } catch (resolveError) { - error = resolveError instanceof Error ? resolveError.message : String(resolveError); - } - } - - const ok = sshOk && remoteCommand !== null; - deps.emitEvent({ - type: 'check-result', - host, - sshOk, - remoteCommand, - remoteVersion, - ok, - error, - }); - if (!ok) { - throw new Error(error ?? `Connection check failed for ${host}.`); - } - deps.consoleLog('Check passed.'); -} - -function cleanupRemote(host: string, remoteTmpDir: string, deps: SyncCommandDeps): void { - if (!remoteTmpDir.startsWith('/tmp/')) return; - deps.runSsh(host, `rm -rf ${shellQuote(remoteTmpDir)}`); -} - -function formatRemoteRunError(message: string, run: RemoteRunResult): string { - const stderr = run.stderr.trim(); - return stderr ? `${message}\n${stderr}` : message; + await runCheckModeFlow(context, resolveSyncCommandDeps(inputDeps)); } export async function runHostSync( @@ -228,173 +98,12 @@ export async function runHostSync( dbPath: string, inputDeps: Partial = {}, ): Promise { - const deps = resolveSyncCommandDeps(inputDeps); - const { args } = context; - const host = args.syncHost; - const direction = args.syncDirection ?? 'both'; - const shouldPull = direction !== 'push'; - const shouldPush = direction !== 'pull'; - deps.assertSafeSshHost(host); - - await deps.ensureTrackerQuiescent(context, dbPath); - - const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null); - deps.log('debug', args.logLevel, `Remote subminer command: ${remoteCmd}`); - - const localTmpDir = deps.mkdtempSync(path.join(os.tmpdir(), 'subminer-sync-')); - let remoteTmpDir = ''; - let pulledSummary: SyncMergeSummary | null = null; - try { - // Signal failures by throwing (not fail(), which exits synchronously and - // would skip the finally cleanup, leaking temp dirs holding snapshot data). - // main().catch() reports the message the same way fail() would. - const mktemp = deps.runSsh(host, 'mktemp -d /tmp/subminer-sync.XXXXXX'); - remoteTmpDir = mktemp.stdout.trim(); - if (mktemp.status !== 0 || !remoteTmpDir.startsWith('/tmp/')) { - throw new Error(`Could not create a temporary directory on ${host}.`); - } - - const forceFlag = args.syncForce ? ' --force' : ''; - - const localSnapshot = path.join(localTmpDir, 'local.sqlite'); - if (shouldPush) { - deps.consoleLog(`Snapshotting local database (${dbPath})...`); - deps.emitEvent({ - type: 'stage', - stage: 'snapshot-local', - message: `Snapshotting local database (${dbPath})`, - }); - deps.createDbSnapshot(dbPath, localSnapshot); - } - - const remoteSnapshot = `${remoteTmpDir}/snapshot.sqlite`; - if (shouldPull) { - deps.consoleLog(`Snapshotting ${host}...`); - deps.emitEvent({ type: 'stage', stage: 'snapshot-remote', message: `Snapshotting ${host}` }); - const snapshotRun = deps.runSsh( - host, - `${remoteCmd} sync --snapshot ${shellQuote(remoteSnapshot)}${forceFlag}`, - ); - if (snapshotRun.status !== 0) { - throw new Error(formatRemoteRunError(`Remote snapshot failed on ${host}.`, snapshotRun)); - } - } - - const pulledSnapshot = path.join(localTmpDir, 'remote.sqlite'); - if (shouldPull) { - deps.emitEvent({ - type: 'stage', - stage: 'download', - message: `Copying snapshot from ${host}`, - }); - deps.runScp(`${host}:${remoteSnapshot}`, pulledSnapshot); - } - const incomingSnapshot = `${remoteTmpDir}/incoming.sqlite`; - if (shouldPush) { - deps.emitEvent({ type: 'stage', stage: 'upload', message: `Copying snapshot to ${host}` }); - deps.runScp(localSnapshot, `${host}:${incomingSnapshot}`); - } - - if (shouldPull) { - deps.consoleLog(`\nMerging ${host} -> local:`); - deps.emitEvent({ - type: 'stage', - stage: 'merge-local', - message: `Merging ${host} into the local database`, - }); - await deps.ensureTrackerQuiescent(context, dbPath); - const summary = deps.mergeSnapshotIntoDb(dbPath, pulledSnapshot); - pulledSummary = summary; - deps.emitEvent({ type: 'merge-summary', target: 'local', summary }); - deps.consoleLog(deps.formatMergeSummary(summary)); - } - - if (shouldPush) { - deps.consoleLog(`\nMerging local -> ${host}:`); - deps.emitEvent({ - type: 'stage', - stage: 'merge-remote', - message: `Merging the local database into ${host}`, - }); - await deps.ensureTrackerQuiescent(context, dbPath); - const mergeRun = deps.runSsh( - host, - `${remoteCmd} sync --merge ${shellQuote(incomingSnapshot)}${forceFlag}`, - ); - deps.writeStdout(mergeRun.stdout); - if (mergeRun.stdout.trim()) { - deps.emitEvent({ type: 'remote-output', text: mergeRun.stdout }); - } - if (mergeRun.status !== 0) { - const retryCommand = - direction === 'push' ? `subminer sync ${host} --push` : `subminer sync ${host}`; - const localUpdate = shouldPull ? ' The local database was updated;' : ''; - throw new Error( - formatRemoteRunError( - `Remote merge failed on ${host}.${localUpdate} re-run "${retryCommand}" once the remote issue is fixed.`, - mergeRun, - ), - ); - } - } - - deps.consoleLog('\nSync complete.'); - deps.recordHostSyncResult(host, 'success', formatHostSyncDetail(direction, pulledSummary)); - } catch (error) { - try { - deps.recordHostSyncResult( - host, - 'error', - error instanceof Error ? error.message : String(error), - ); - } catch { - // best effort - } - throw error; - } finally { - deps.rmSync(localTmpDir, { recursive: true, force: true }); - if (remoteTmpDir) { - try { - cleanupRemote(host, remoteTmpDir, deps); - } catch { - // best effort - } - } - } + await runHostSyncFlow(context, dbPath, resolveSyncCommandDeps(inputDeps)); } export async function runSyncCommand( context: LauncherCommandContext, inputDeps: Partial = {}, ): Promise { - let deps = resolveSyncCommandDeps(inputDeps); - const { args } = context; - if (!args.sync) return false; - if (args.syncJson) deps = withJsonEvents(deps); - - const dbPath = resolveDbPath(context); - try { - if (args.syncCheck) { - await runCheckMode(context, deps); - } else if (args.syncSnapshotPath) { - runSnapshotMode(context, dbPath, deps); - } else if (args.syncMergePath) { - await runMergeMode(context, dbPath, deps); - } else if (args.syncHost) { - await runHostSync(context, dbPath, deps); - } else { - deps.fail('sync requires a host, --snapshot , or --merge .'); - } - } catch (error) { - if (args.syncJson) { - deps.emitEvent({ - type: 'result', - ok: false, - error: error instanceof Error ? error.message : String(error), - }); - } - throw error; - } - if (args.syncJson) deps.emitEvent({ type: 'result', ok: true, error: null }); - return true; + return runSyncFlow(context, resolveSyncCommandDeps(inputDeps)); } diff --git a/launcher/history-db.ts b/launcher/history-db.ts index 68e5911b..66dc20b7 100644 --- a/launcher/history-db.ts +++ b/launcher/history-db.ts @@ -43,48 +43,11 @@ export function queryLocalWatchHistory(dbPath: string): HistoryVideoRow[] { return withReadonlyWalRetry(dbPath, (options) => readHistoryRows(dbPath, options)); } -export function withReadonlyWalRetry( - dbPath: string, - query: (options: { readonly?: boolean; readwrite?: boolean; create?: boolean }) => T, -): T { - try { - return query({ readonly: true }); - } catch (error) { - if (!isReadonlyWalRetryError(error, dbPath)) throw error; - return query({ readwrite: true, create: false }); - } -} - -export function isReadonlyWalRetryError(error: unknown, dbPath: string): boolean { - if (!isWalModeSqliteDatabase(dbPath)) return false; - const code = - typeof error === 'object' && error !== null && 'code' in error - ? String((error as { code?: unknown }).code ?? '') - : ''; - const message = error instanceof Error ? error.message : String(error); - const text = `${code} ${message}`.toLowerCase(); - return ( - text.includes('readonly') || - text.includes('read-only') || - text.includes('attempt to write a readonly database') || - text.includes('sqlite_cantopen') || - text.includes('unable to open database file') - ); -} - -function isWalModeSqliteDatabase(dbPath: string): boolean { - const header = Buffer.alloc(20); - let fd: number | null = null; - try { - fd = fs.openSync(dbPath, 'r'); - if (fs.readSync(fd, header, 0, header.length, 0) < header.length) return false; - } catch { - return false; - } finally { - if (fd !== null) fs.closeSync(fd); - } - return header.subarray(0, 16).toString('ascii') === 'SQLite format 3\0' && header[18] === 2; -} +export { + withReadonlyWalRetry, + isReadonlyWalRetryError, +} from '../src/core/services/stats-sync/wal-retry.js'; +import { withReadonlyWalRetry } from '../src/core/services/stats-sync/wal-retry.js'; function tableExists(db: Database, tableName: string): boolean { return Boolean( diff --git a/launcher/sync/bun-driver.ts b/launcher/sync/bun-driver.ts new file mode 100644 index 00000000..86ab0bb8 --- /dev/null +++ b/launcher/sync/bun-driver.ts @@ -0,0 +1,23 @@ +import { Database } from 'bun:sqlite'; +import type { + OpenSyncDb, + SyncDb, + SyncDbStatement, +} from '../../src/core/services/stats-sync/driver.js'; + +// bun:sqlite's Database.query() already caches prepared statements per SQL +// string, which is exactly what the SyncDb contract asks for. +export const openBunSyncDb: OpenSyncDb = (dbPath, options): SyncDb => { + const db = new Database(dbPath, options); + return { + query(sql: string): SyncDbStatement { + return db.query(sql); + }, + exec(sql: string): void { + db.run(sql); + }, + close(): void { + db.close(); + }, + }; +}; diff --git a/launcher/sync/ssh.ts b/launcher/sync/ssh.ts index 220a4a6c..4395fbda 100644 --- a/launcher/sync/ssh.ts +++ b/launcher/sync/ssh.ts @@ -1,106 +1,10 @@ -import { spawnSync } from 'node:child_process'; - -export interface RemoteRunResult { - status: number; - stdout: string; - stderr: string; -} - -/** - * ssh/scp have no `--` terminator for the destination, so a host that starts - * with `-` (e.g. `-oProxyCommand=...`) is parsed as an option. Reject those - * before spawning. - */ -export function assertSafeSshHost(host: string): void { - if (host.startsWith('-')) { - throw new Error(`Refusing to use SSH host that looks like an option: ${host}`); - } -} - -/** - * Run a command on the SSH host. stdin stays attached so interactive prompts - * can still read from the terminal; stdout/stderr are captured for callers - * that need actionable remote failure messages. - */ -export function runSsh(host: string, remoteCommand: string): RemoteRunResult { - assertSafeSshHost(host); - const result = spawnSync('ssh', [host, remoteCommand], { - encoding: 'utf8', - stdio: ['inherit', 'pipe', 'pipe'], - }); - if (result.error) { - throw new Error(`Failed to run ssh: ${(result.error as Error).message}`); - } - return { status: result.status ?? 1, stdout: result.stdout ?? '', stderr: result.stderr ?? '' }; -} - -function assertSafeScpEndpoint(endpoint: string): void { - const colon = endpoint.indexOf(':'); - const slash = endpoint.indexOf('/'); - if (colon <= 0 || (slash !== -1 && slash < colon)) { - if (endpoint.startsWith('-')) { - throw new Error(`Refusing to use scp endpoint that looks like an option: ${endpoint}`); - } - return; - } - - const host = endpoint.slice(0, colon); - const remotePath = endpoint.slice(colon + 1); - assertSafeSshHost(host); - if (remotePath.startsWith('-')) { - throw new Error(`Refusing to use scp remote path that looks like an option: ${remotePath}`); - } -} - -export function runScp(from: string, to: string): void { - assertSafeScpEndpoint(from); - assertSafeScpEndpoint(to); - const result = spawnSync('scp', ['-q', from, to], { - encoding: 'utf8', - stdio: ['inherit', 'inherit', 'inherit'], - }); - if (result.error) { - throw new Error(`Failed to run scp: ${(result.error as Error).message}`); - } - if ((result.status ?? 1) !== 0) { - throw new Error(`scp failed copying ${from} -> ${to}`); - } -} - -export function shellQuote(value: string): string { - return `'${value.replaceAll("'", `'\\''`)}'`; -} - -const REMOTE_RUNTIME_PATH = - 'PATH="$HOME/.local/bin:$HOME/.bun/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:$PATH"'; - -/** - * Non-interactive SSH shells often miss user-installed launchers and Bun. - * Probe the launcher under the same deterministic PATH used by sync itself. - */ -export function resolveRemoteSubminerCommand( - host: string, - preferred: string | null, - runRemote: typeof runSsh = runSsh, -): string { - // Trusted defaults stay unquoted so the remote shell expands `~`; a - // user-supplied override is shell-quoted to prevent command injection. - const candidates: Array<{ value: string; invocation: string }> = preferred - ? [{ value: preferred, invocation: shellQuote(preferred) }] - : [ - { value: 'subminer', invocation: 'subminer' }, - { value: '~/.local/bin/subminer', invocation: '~/.local/bin/subminer' }, - ]; - for (const candidate of candidates) { - const command = `${REMOTE_RUNTIME_PATH} ${candidate.invocation}`; - const probe = runRemote(host, `${command} --help >/dev/null 2>&1`); - if (probe.status === 0) { - return command; - } - } - throw new Error( - preferred - ? `Remote command not found on ${host}: ${preferred}` - : `subminer not found on ${host} (tried PATH and ~/.local/bin/subminer). Pass --remote-cmd .`, - ); -} +// Re-exported from the shared stats-sync engine so the launcher and the +// app's --sync-cli mode resolve remotes and shell out identically. +export { + assertSafeSshHost, + resolveRemoteSubminerCommand, + runScp, + runSsh, + shellQuote, + type RemoteRunResult, +} from '../../src/core/services/stats-sync/ssh.js'; diff --git a/launcher/sync/sync-db.ts b/launcher/sync/sync-db.ts index ef1fcb3c..0d45f611 100644 --- a/launcher/sync/sync-db.ts +++ b/launcher/sync/sync-db.ts @@ -1,105 +1,12 @@ -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'; +import { mergeSnapshotIntoDb as mergeSnapshotIntoDbWith } from '../../src/core/services/stats-sync/merge.js'; +import { openBunSyncDb } from './bun-driver.js'; +import type { SyncMergeSummary } from './sync-shared.js'; export type { SyncMergeSummary } from './sync-shared.js'; export { createDbSnapshot, findLiveStatsDaemonPid } from './sync-shared.js'; +export { formatMergeSummary } from '../../src/core/services/stats-sync/merge.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. - */ +/** bun:sqlite binding of the shared snapshot-merge engine. */ 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 }); - let local: Database; - try { - local = new Database(localDbPath, { readwrite: true, create: false }); - } catch (error) { - remote.close(); - throw error; - } - 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'); + return mergeSnapshotIntoDbWith(openBunSyncDb, localDbPath, snapshotPath); } diff --git a/launcher/sync/sync-shared.ts b/launcher/sync/sync-shared.ts index 98bc2ac4..eeb1c49e 100644 --- a/launcher/sync/sync-shared.ts +++ b/launcher/sync/sync-shared.ts @@ -1,161 +1,31 @@ -import fs from 'node:fs'; -import os from 'node:os'; -import path from 'node:path'; -import { Database } from 'bun:sqlite'; -import { SCHEMA_VERSION } from '../../src/core/services/immersion-tracker/types.js'; -import { withReadonlyWalRetry } from '../history-db.js'; -import { resolveConfigDir } from '../../src/config/path-resolution.js'; +// bun:sqlite bindings for the shared stats-sync engine. The engine itself +// lives in src/core/services/stats-sync so the Electron app (libsql) can run +// the exact same snapshot/merge code via its own driver. +import { + createDbSnapshot as createDbSnapshotWith, + createEmptyMergeSummary, + findLiveStatsDaemonPid, + nowDbTimestamp, + readSchemaVersion, + assertMergeableSchema, + insertRow, + tableExists, + SCHEMA_VERSION, +} from '../../src/core/services/stats-sync/shared.js'; +import { openBunSyncDb } from './bun-driver.js'; -export { SCHEMA_VERSION }; - -export type { SyncMergeSummary } from '../../src/shared/sync/sync-events.js'; -import type { SyncMergeSummary } from '../../src/shared/sync/sync-events.js'; - -export function createEmptyMergeSummary(): SyncMergeSummary { - return { - sessionsMerged: 0, - sessionsAlreadyPresent: 0, - activeSessionsSkipped: 0, - animeAdded: 0, - videosAdded: 0, - wordsAdded: 0, - kanjiAdded: 0, - subtitleLinesAdded: 0, - telemetryRowsAdded: 0, - eventsAdded: 0, - excludedWordsAdded: 0, - dailyRollupsCopied: 0, - monthlyRollupsCopied: 0, - rollupGroupsRecomputed: 0, - }; -} - -export function nowDbTimestamp(): string { - return String(Date.now()); -} - -export function tableExists(db: Database, tableName: string): boolean { - return Boolean( - db.query(`SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?`).get(tableName), - ); -} - -export function readSchemaVersion(db: Database): number | null { - if (!tableExists(db, 'imm_schema_version')) return null; - const row = db - .query<{ - schema_version: number; - }>('SELECT MAX(schema_version) AS schema_version FROM imm_schema_version') - .get(); - return typeof row?.schema_version === 'number' ? row.schema_version : null; -} - -export function assertMergeableSchema(db: Database, label: string): void { - const version = readSchemaVersion(db); - if (version === null) { - throw new Error( - `${label} database has no schema version. Run SubMiner once on that machine so the stats database is initialized.`, - ); - } - if (version !== SCHEMA_VERSION) { - throw new Error( - `${label} database is at schema version ${version} but this launcher expects ${SCHEMA_VERSION}. Update SubMiner on both machines to the same version and run each app once before syncing.`, - ); - } - for (const table of ['imm_sessions', 'imm_videos', 'imm_lifetime_global']) { - if (!tableExists(db, table)) { - throw new Error(`${label} database is missing table ${table}; cannot sync.`); - } - } -} - -export function insertRow( - db: Database, - table: string, - columns: readonly string[], - values: unknown[], -): number { - const sql = `INSERT INTO ${table} (${columns.join(', ')}) VALUES (${columns.map(() => '?').join(', ')})`; - // db.query() caches the prepared statement per SQL string; this runs once - // per copied row, so re-preparing via db.prepare() would dominate merge time. - const result = db.query(sql).run(...values); - return Number(result.lastInsertRowid); -} +export { + createEmptyMergeSummary, + findLiveStatsDaemonPid, + nowDbTimestamp, + readSchemaVersion, + assertMergeableSchema, + insertRow, + tableExists, + SCHEMA_VERSION, +}; +export type { SyncMergeSummary } from '../../src/core/services/stats-sync/shared.js'; export function createDbSnapshot(dbPath: string, outPath: string): void { - if (!fs.existsSync(dbPath)) { - throw new Error(`Stats database not found: ${dbPath}`); - } - fs.rmSync(outPath, { force: true }); - fs.mkdirSync(path.dirname(outPath), { recursive: true }); - withReadonlyWalRetry(dbPath, (options) => { - const db = new Database(dbPath, options); - try { - assertMergeableSchema(db, 'Local'); - db.prepare('VACUUM INTO ?').run(outPath); - } finally { - db.close(); - } - }); -} - -interface DaemonStateFile { - pid?: unknown; -} - -function isProcessAlive(pid: number): boolean { - try { - process.kill(pid, 0); - return true; - } catch (error) { - // EPERM means the process exists but we can't signal it → still alive. - // Only ESRCH (no such process) means it's actually gone. - return (error as NodeJS.ErrnoException)?.code === 'EPERM'; - } -} - -function statsDaemonStateCandidates(dbPath: string): string[] { - const homeDir = os.homedir(); - const candidates = new Set([path.join(path.dirname(dbPath), 'stats-daemon.json')]); - const configDir = resolveConfigDir({ - platform: process.platform, - appDataDir: process.env.APPDATA, - xdgConfigHome: process.env.XDG_CONFIG_HOME, - homeDir, - existsSync: fs.existsSync, - }); - candidates.add(path.join(configDir, 'stats-daemon.json')); - if (process.platform === 'darwin') { - candidates.add( - path.join(homeDir, 'Library', 'Application Support', 'SubMiner', 'stats-daemon.json'), - ); - } - return [...candidates]; -} - -/** - * Best-effort guard against merging while a SubMiner process holds the - * tracker's write queue in memory. Detects the background stats daemon via - * its pid state file; the interactive app is caught by the mpv-socket check - * in the sync command. - */ -export function findLiveStatsDaemonPid(dbPath: string): number | null { - for (const statePath of statsDaemonStateCandidates(dbPath)) { - let raw: string; - try { - raw = fs.readFileSync(statePath, 'utf8'); - } catch { - continue; - } - try { - const parsed = JSON.parse(raw) as DaemonStateFile; - const pid = typeof parsed.pid === 'number' && Number.isInteger(parsed.pid) ? parsed.pid : 0; - if (pid > 0 && isProcessAlive(pid)) { - return pid; - } - } catch { - continue; - } - } - return null; + createDbSnapshotWith(openBunSyncDb, dbPath, outPath); } diff --git a/src/core/services/stats-sync/driver.ts b/src/core/services/stats-sync/driver.ts new file mode 100644 index 00000000..e5fe3d1b --- /dev/null +++ b/src/core/services/stats-sync/driver.ts @@ -0,0 +1,42 @@ +// Minimal SQLite driver surface the stats-sync engine runs against. The +// launcher binds it to bun:sqlite; the Electron app binds it to libsql. Both +// implementations must cache prepared statements per SQL string in query(): +// the merge runs one insert per copied row, so re-preparing would dominate +// merge time. + +export interface SyncDbRunResult { + changes: number; + lastInsertRowid: number | bigint; +} + +export interface SyncDbStatement { + run(...params: unknown[]): SyncDbRunResult; + get(...params: unknown[]): unknown; + all(...params: unknown[]): unknown[]; +} + +export interface SyncDbOpenOptions { + readonly?: boolean; + readwrite?: boolean; + create?: boolean; +} + +export interface SyncDb { + /** Prepare (or reuse a cached prepared statement for) the given SQL. */ + query(sql: string): SyncDbStatement; + /** Execute SQL that returns no rows (pragmas, transaction control). */ + exec(sql: string): void; + close(): void; +} + +export type OpenSyncDb = (dbPath: string, options: SyncDbOpenOptions) => SyncDb; + +export type SqlRow = Record; + +export function selectAll(db: SyncDb, sql: string, params: unknown[] = []): SqlRow[] { + return db.query(sql).all(...params) as SqlRow[]; +} + +export function selectOne(db: SyncDb, sql: string, params: unknown[] = []): SqlRow | undefined { + return (db.query(sql).get(...params) ?? undefined) as SqlRow | undefined; +} diff --git a/launcher/sync/merge-catalog.ts b/src/core/services/stats-sync/merge-catalog.ts similarity index 80% rename from launcher/sync/merge-catalog.ts rename to src/core/services/stats-sync/merge-catalog.ts index b3c9d60b..265abbbd 100644 --- a/launcher/sync/merge-catalog.ts +++ b/src/core/services/stats-sync/merge-catalog.ts @@ -1,5 +1,5 @@ -import { Database } from 'bun:sqlite'; -import { insertRow, tableExists, type SyncMergeSummary } from './sync-shared.js'; +import { selectAll, selectOne, type SqlRow, type SyncDb } from './driver'; +import { insertRow, tableExists, type SyncMergeSummary } from './shared'; const ANIME_COPY_COLUMNS = [ 'normalized_title_key', @@ -90,23 +90,15 @@ const WORD_COPY_COLUMNS = [ 'frequency_rank', ] as const; -type SqlRow = Record; - -function selectAll(db: Database, sql: string, params: unknown[] = []): SqlRow[] { - return db.query(sql).all(...params); -} - export function mergeAnime( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, summary: SyncMergeSummary, ): Map { const map = new Map(); - const byAnilist = local.prepare('SELECT anime_id FROM imm_anime WHERE anilist_id = ?'); - const byTitleKey = local.prepare( - 'SELECT anime_id FROM imm_anime WHERE normalized_title_key = ?', - ); - const fillMissing = local.prepare( + const byAnilist = local.query('SELECT anime_id FROM imm_anime WHERE anilist_id = ?'); + const byTitleKey = local.query('SELECT anime_id FROM imm_anime WHERE normalized_title_key = ?'); + const fillMissing = local.query( `UPDATE imm_anime SET title_romaji = COALESCE(title_romaji, ?), @@ -122,9 +114,8 @@ export function mergeAnime( `SELECT anime_id, ${ANIME_COPY_COLUMNS.join(', ')} FROM imm_anime`, )) { const remoteId = Number(row.anime_id); - const existing = - (row.anilist_id !== null ? byAnilist.get(row.anilist_id) : undefined) ?? - byTitleKey.get(row.normalized_title_key); + const existing = ((row.anilist_id !== null ? byAnilist.get(row.anilist_id) : undefined) ?? + byTitleKey.get(row.normalized_title_key)) as SqlRow | undefined; if (existing) { const localId = Number(existing.anime_id); map.set(remoteId, localId); @@ -153,17 +144,15 @@ export interface VideoMergeResult { } export function mergeVideos( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, animeIdMap: Map, summary: SyncMergeSummary, ): VideoMergeResult { const videoIdMap = new Map(); const addedVideoIds = new Set(); - const byKey = local.prepare( - 'SELECT video_id, watched FROM imm_videos WHERE video_key = ?', - ); - const setWatched = local.prepare('UPDATE imm_videos SET watched = 1 WHERE video_id = ?'); + const byKey = local.query('SELECT video_id, watched FROM imm_videos WHERE video_key = ?'); + const setWatched = local.query('UPDATE imm_videos SET watched = 1 WHERE video_id = ?'); for (const row of selectAll( remote, @@ -172,7 +161,7 @@ export function mergeVideos( const remoteId = Number(row.video_id); const mappedAnimeId = row.anime_id === null ? null : (animeIdMap.get(Number(row.anime_id)) ?? null); - const existing = byKey.get(row.video_key); + const existing = byKey.get(row.video_key) as SqlRow | undefined; if (existing) { const localId = Number(existing.video_id); videoIdMap.set(remoteId, localId); @@ -192,8 +181,8 @@ export function mergeVideos( } export function mergeMediaMetadata( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, videoIdMap: Map, addedVideoIds: Set, ): void { @@ -203,31 +192,29 @@ export function mergeMediaMetadata( const hasBlobStore = tableExists(local, 'imm_cover_art_blobs') && tableExists(remote, 'imm_cover_art_blobs'); const copyBlob = hasBlobStore - ? local.prepare( + ? local.query( `INSERT INTO imm_cover_art_blobs (blob_hash, cover_blob, CREATED_DATE, LAST_UPDATE_DATE) VALUES (?, ?, ?, ?) ON CONFLICT(blob_hash) DO NOTHING`, ) : null; const readBlob = hasBlobStore - ? remote.prepare('SELECT * FROM imm_cover_art_blobs WHERE blob_hash = ?') + ? remote.query('SELECT * FROM imm_cover_art_blobs WHERE blob_hash = ?') : null; if (tableExists(remote, 'imm_media_art') && tableExists(local, 'imm_media_art')) { - const localArtExists = local.prepare( - 'SELECT 1 FROM imm_media_art WHERE video_id = ? LIMIT 1', - ); + const localArtExists = local.query('SELECT 1 FROM imm_media_art WHERE video_id = ? LIMIT 1'); for (const remoteVideoId of metadataVideoIds) { const localVideoId = videoIdMap.get(remoteVideoId)!; if (localArtExists.get(localVideoId)) continue; - const row = remote - .query( - `SELECT ${MEDIA_ART_COPY_COLUMNS.join(', ')} FROM imm_media_art WHERE video_id = ?`, - ) - .get(remoteVideoId); + const row = selectOne( + remote, + `SELECT ${MEDIA_ART_COPY_COLUMNS.join(', ')} FROM imm_media_art WHERE video_id = ?`, + [remoteVideoId], + ); if (!row) continue; if (row.cover_blob_hash && copyBlob && readBlob) { - const blob = readBlob.get(row.cover_blob_hash); + const blob = readBlob.get(row.cover_blob_hash) as SqlRow | undefined; if (blob) { copyBlob.run(blob.blob_hash, blob.cover_blob, blob.CREATED_DATE, blob.LAST_UPDATE_DATE); } @@ -242,17 +229,17 @@ export function mergeMediaMetadata( } if (tableExists(remote, 'imm_youtube_videos') && tableExists(local, 'imm_youtube_videos')) { - const localYoutubeExists = local.prepare( + const localYoutubeExists = local.query( 'SELECT 1 FROM imm_youtube_videos WHERE video_id = ? LIMIT 1', ); for (const remoteVideoId of metadataVideoIds) { const localVideoId = videoIdMap.get(remoteVideoId)!; if (localYoutubeExists.get(localVideoId)) continue; - const row = remote - .query( - `SELECT ${YOUTUBE_COPY_COLUMNS.join(', ')} FROM imm_youtube_videos WHERE video_id = ?`, - ) - .get(remoteVideoId); + const row = selectOne( + remote, + `SELECT ${YOUTUBE_COPY_COLUMNS.join(', ')} FROM imm_youtube_videos WHERE video_id = ?`, + [remoteVideoId], + ); if (!row) continue; insertRow( local, @@ -265,8 +252,8 @@ export function mergeMediaMetadata( } export function mergeExcludedWords( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, summary: SyncMergeSummary, ): void { if ( @@ -275,7 +262,7 @@ export function mergeExcludedWords( ) { return; } - const insert = local.prepare( + const insert = local.query( `INSERT INTO imm_stats_excluded_words (headword, word, reading, CREATED_DATE, LAST_UPDATE_DATE) VALUES (?, ?, ?, ?, ?) ON CONFLICT(headword, word, reading) DO NOTHING`, @@ -309,8 +296,8 @@ export class LexiconResolver { readonly kanjiFrequencyDeltas = new Map(); constructor( - private readonly local: Database, - private readonly remote: Database, + private readonly local: SyncDb, + private readonly remote: SyncDb, private readonly summary: SyncMergeSummary, ) {} @@ -318,19 +305,23 @@ export class LexiconResolver { const cached = this.wordMap.get(remoteWordId); if (cached) return cached.localId; - const row = this.remote - .query(`SELECT ${WORD_COPY_COLUMNS.join(', ')} FROM imm_words WHERE id = ?`) - .get(remoteWordId); + const row = selectOne( + this.remote, + `SELECT ${WORD_COPY_COLUMNS.join(', ')} FROM imm_words WHERE id = ?`, + [remoteWordId], + ); if (!row) throw new Error(`Snapshot references missing imm_words row ${remoteWordId}`); - const existing = this.local - .query('SELECT id FROM imm_words WHERE headword IS ? AND word IS ? AND reading IS ?') - .get(row.headword, row.word, row.reading); + const existing = selectOne( + this.local, + 'SELECT id FROM imm_words WHERE headword IS ? AND word IS ? AND reading IS ?', + [row.headword, row.word, row.reading], + ); let entry: { localId: number; isNew: boolean }; if (existing) { entry = { localId: Number(existing.id), isNew: false }; this.local - .prepare( + .query( `UPDATE imm_words SET first_seen = MIN(COALESCE(first_seen, ?), COALESCE(?, first_seen)), last_seen = MAX(COALESCE(last_seen, ?), COALESCE(?, last_seen)) @@ -355,19 +346,21 @@ export class LexiconResolver { const cached = this.kanjiMap.get(remoteKanjiId); if (cached) return cached.localId; - const row = this.remote - .query('SELECT kanji, first_seen, last_seen, frequency FROM imm_kanji WHERE id = ?') - .get(remoteKanjiId); + const row = selectOne( + this.remote, + 'SELECT kanji, first_seen, last_seen, frequency FROM imm_kanji WHERE id = ?', + [remoteKanjiId], + ); if (!row) throw new Error(`Snapshot references missing imm_kanji row ${remoteKanjiId}`); - const existing = this.local - .query('SELECT id FROM imm_kanji WHERE kanji IS ?') - .get(row.kanji); + const existing = selectOne(this.local, 'SELECT id FROM imm_kanji WHERE kanji IS ?', [ + row.kanji, + ]); let entry: { localId: number; isNew: boolean }; if (existing) { entry = { localId: Number(existing.id), isNew: false }; this.local - .prepare( + .query( `UPDATE imm_kanji SET first_seen = MIN(COALESCE(first_seen, ?), COALESCE(?, first_seen)), last_seen = MAX(COALESCE(last_seen, ?), COALESCE(?, last_seen)) @@ -407,13 +400,13 @@ export class LexiconResolver { } applyFrequencyDeltas(): void { - const updateWord = this.local.prepare( + const updateWord = this.local.query( 'UPDATE imm_words SET frequency = COALESCE(frequency, 0) + ? WHERE id = ?', ); for (const [localId, delta] of this.wordFrequencyDeltas) { updateWord.run(delta, localId); } - const updateKanji = this.local.prepare( + const updateKanji = this.local.query( 'UPDATE imm_kanji SET frequency = COALESCE(frequency, 0) + ? WHERE id = ?', ); for (const [localId, delta] of this.kanjiFrequencyDeltas) { diff --git a/launcher/sync/merge-rollups.ts b/src/core/services/stats-sync/merge-rollups.ts similarity index 89% rename from launcher/sync/merge-rollups.ts rename to src/core/services/stats-sync/merge-rollups.ts index c4f6892a..ee91e339 100644 --- a/launcher/sync/merge-rollups.ts +++ b/src/core/services/stats-sync/merge-rollups.ts @@ -1,7 +1,5 @@ -import { Database } from 'bun:sqlite'; -import { nowDbTimestamp, tableExists, type SyncMergeSummary } from './sync-shared.js'; - -type SqlRow = Record; +import { selectAll, type SqlRow, type SyncDb } from './driver'; +import { nowDbTimestamp, tableExists, type SyncMergeSummary } from './shared'; const LOCAL_DAY_EXPR = `CAST(julianday(CAST(started_at_ms AS REAL) / 1000, 'unixepoch', 'localtime') - 2440587.5 AS INTEGER)`; const LOCAL_MONTH_EXPR = `CAST(strftime('%Y%m', CAST(started_at_ms AS REAL) / 1000, 'unixepoch', 'localtime') AS INTEGER)`; @@ -125,7 +123,7 @@ const MONTHLY_ROLLUP_UPSERT = ` * the app later, which is idempotent. */ export function refreshRollupsForNewSessions( - local: Database, + local: SyncDb, newSessionIds: number[], summary: SyncMergeSummary, ): void { @@ -134,12 +132,12 @@ export function refreshRollupsForNewSessions( const groups = new Map(); for (let offset = 0; offset < newSessionIds.length; offset += 500) { const chunk = newSessionIds.slice(offset, offset + 500); - const rows = local - .query( - `SELECT DISTINCT ${LOCAL_DAY_EXPR} AS rollup_day, ${LOCAL_MONTH_EXPR} AS rollup_month, video_id - FROM imm_sessions WHERE session_id IN (${chunk.map(() => '?').join(',')})`, - ) - .all(...chunk); + const rows = selectAll( + local, + `SELECT DISTINCT ${LOCAL_DAY_EXPR} AS rollup_day, ${LOCAL_MONTH_EXPR} AS rollup_month, video_id + FROM imm_sessions WHERE session_id IN (${chunk.map(() => '?').join(',')})`, + chunk, + ); for (const row of rows) { const day = Number(row.rollup_day); const month = Number(row.rollup_month); @@ -149,14 +147,12 @@ export function refreshRollupsForNewSessions( } const stampMs = nowDbTimestamp(); - const deleteDaily = local.prepare( - 'DELETE FROM imm_daily_rollups WHERE rollup_day = ? AND video_id = ?', - ); - const deleteMonthly = local.prepare( + const deleteDaily = local.query('DELETE FROM imm_daily_rollups WHERE rollup_day = ? AND video_id = ?'); + const deleteMonthly = local.query( 'DELETE FROM imm_monthly_rollups WHERE rollup_month = ? AND video_id = ?', ); - const upsertDaily = local.prepare(DAILY_ROLLUP_UPSERT); - const upsertMonthly = local.prepare(MONTHLY_ROLLUP_UPSERT); + const upsertDaily = local.query(DAILY_ROLLUP_UPSERT); + const upsertMonthly = local.query(MONTHLY_ROLLUP_UPSERT); const monthlyGroups = new Set(); for (const { day, month, videoId } of groups.values()) { @@ -181,36 +177,36 @@ export function refreshRollupsForNewSessions( * double-counting sessions that earlier syncs already shared. */ export function copyRemoteOnlyRollups( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, videoIdMap: Map, summary: SyncMergeSummary, ): void { if (!tableExists(remote, 'imm_daily_rollups') || !tableExists(local, 'imm_daily_rollups')) return; - const localDailyExists = local.prepare( + const localDailyExists = local.query( 'SELECT 1 FROM imm_daily_rollups WHERE rollup_day = ? AND video_id = ? LIMIT 1', ); - const localDaySessions = local.prepare( + const localDaySessions = local.query( `SELECT 1 FROM imm_sessions WHERE video_id = ? AND ${LOCAL_DAY_EXPR} = ? LIMIT 1`, ); - const localMonthSessions = local.prepare( + const localMonthSessions = local.query( `SELECT 1 FROM imm_sessions WHERE video_id = ? AND ${LOCAL_MONTH_EXPR} = ? LIMIT 1`, ); - const localMonthSessionsForDay = local.prepare( + const localMonthSessionsForDay = local.query( `SELECT 1 FROM imm_sessions WHERE video_id = ? AND ${LOCAL_MONTH_EXPR} = CAST(strftime('%Y%m', CAST(? AS INTEGER) * 86400, 'unixepoch', 'localtime') AS INTEGER) LIMIT 1`, ); - const insertDaily = local.prepare( + const insertDaily = local.query( `INSERT INTO imm_daily_rollups ( rollup_day, video_id, total_sessions, total_active_min, total_lines_seen, total_tokens_seen, total_cards, cards_per_hour, tokens_per_min, lookup_hit_rate, CREATED_DATE, LAST_UPDATE_DATE ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); - for (const row of remote.query('SELECT * FROM imm_daily_rollups').all()) { + for (const row of selectAll(remote, 'SELECT * FROM imm_daily_rollups')) { if (row.video_id === null) continue; const localVideoId = videoIdMap.get(Number(row.video_id)); if (localVideoId === undefined) continue; @@ -234,16 +230,16 @@ export function copyRemoteOnlyRollups( summary.dailyRollupsCopied += 1; } - const localMonthlyExists = local.prepare( + const localMonthlyExists = local.query( 'SELECT 1 FROM imm_monthly_rollups WHERE rollup_month = ? AND video_id = ? LIMIT 1', ); - const insertMonthly = local.prepare( + const insertMonthly = local.query( `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 (?, ?, ?, ?, ?, ?, ?, ?, ?)`, ); - for (const row of remote.query('SELECT * FROM imm_monthly_rollups').all()) { + for (const row of selectAll(remote, 'SELECT * FROM imm_monthly_rollups')) { if (row.video_id === null) continue; const localVideoId = videoIdMap.get(Number(row.video_id)); if (localVideoId === undefined) continue; diff --git a/launcher/sync/merge-sessions.ts b/src/core/services/stats-sync/merge-sessions.ts similarity index 80% rename from launcher/sync/merge-sessions.ts rename to src/core/services/stats-sync/merge-sessions.ts index ecc6ca46..a5324d44 100644 --- a/launcher/sync/merge-sessions.ts +++ b/src/core/services/stats-sync/merge-sessions.ts @@ -1,6 +1,6 @@ -import { Database } from 'bun:sqlite'; -import type { LexiconResolver } from './merge-catalog.js'; -import { insertRow, nowDbTimestamp, type SyncMergeSummary } from './sync-shared.js'; +import type { LexiconResolver } from './merge-catalog'; +import { selectAll, selectOne, type SqlRow, type SyncDb } from './driver'; +import { insertRow, nowDbTimestamp, type SyncMergeSummary } from './shared'; const SESSION_COPY_COLUMNS = [ 'session_uuid', @@ -71,32 +71,27 @@ const LINE_COPY_COLUMNS = [ 'LAST_UPDATE_DATE', ] as const; -type SqlRow = Record; - export interface SessionMergeResult { newSessionIds: number[]; } export function mergeSessions( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, videoIdMap: Map, animeIdMap: Map, lexicon: LexiconResolver, summary: SyncMergeSummary, ): SessionMergeResult { const newSessionIds: number[] = []; - const uuidExists = local.prepare( - 'SELECT session_id FROM imm_sessions WHERE session_uuid = ?', - ); + const uuidExists = local.query('SELECT session_id FROM imm_sessions WHERE session_uuid = ?'); - const remoteSessions = remote - .query( - `SELECT session_id, video_id, ${SESSION_COPY_COLUMNS.join(', ')} - FROM imm_sessions - ORDER BY CAST(started_at_ms AS REAL) ASC, session_id ASC`, - ) - .all(); + const remoteSessions = selectAll( + remote, + `SELECT session_id, video_id, ${SESSION_COPY_COLUMNS.join(', ')} + FROM imm_sessions + ORDER BY CAST(started_at_ms AS REAL) ASC, session_id ASC`, + ); for (const session of remoteSessions) { if (session.ended_at_ms === null) { @@ -111,7 +106,9 @@ export function mergeSessions( } const localVideoId = videoIdMap.get(Number(session.video_id)); if (localVideoId === undefined) { - throw new Error(`Snapshot session ${String(session.session_uuid)} references missing video row`); + throw new Error( + `Snapshot session ${String(session.session_uuid)} references missing video row`, + ); } const localSessionId = insertRow( @@ -144,18 +141,18 @@ export function mergeSessions( } function copyTelemetry( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, remoteSessionId: number, localSessionId: number, summary: SyncMergeSummary, ): void { - const rows = remote - .query( - `SELECT ${TELEMETRY_COPY_COLUMNS.join(', ')} FROM imm_session_telemetry - WHERE session_id = ? ORDER BY telemetry_id ASC`, - ) - .all(remoteSessionId); + const rows = selectAll( + remote, + `SELECT ${TELEMETRY_COPY_COLUMNS.join(', ')} FROM imm_session_telemetry + WHERE session_id = ? ORDER BY telemetry_id ASC`, + [remoteSessionId], + ); for (const row of rows) { insertRow( local, @@ -168,19 +165,19 @@ function copyTelemetry( } function copyEvents( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, remoteSessionId: number, localSessionId: number, summary: SyncMergeSummary, ): Map { const eventIdMap = new Map(); - const rows = remote - .query( - `SELECT event_id, ${EVENT_COPY_COLUMNS.join(', ')} FROM imm_session_events - WHERE session_id = ? ORDER BY event_id ASC`, - ) - .all(remoteSessionId); + const rows = selectAll( + remote, + `SELECT event_id, ${EVENT_COPY_COLUMNS.join(', ')} FROM imm_session_events + WHERE session_id = ? ORDER BY event_id ASC`, + [remoteSessionId], + ); for (const row of rows) { const localEventId = insertRow( local, @@ -195,8 +192,8 @@ function copyEvents( } function copySubtitleLines( - local: Database, - remote: Database, + local: SyncDb, + remote: SyncDb, remoteSessionId: number, localSessionId: number, localVideoId: number, @@ -205,30 +202,32 @@ function copySubtitleLines( lexicon: LexiconResolver, summary: SyncMergeSummary, ): void { - const rows = remote - .query( - `SELECT line_id, event_id, anime_id, ${LINE_COPY_COLUMNS.join(', ')} FROM imm_subtitle_lines - WHERE session_id = ? ORDER BY line_id ASC`, - ) - .all(remoteSessionId); - const wordOccurrences = remote.prepare( + const rows = selectAll( + remote, + `SELECT line_id, event_id, anime_id, ${LINE_COPY_COLUMNS.join(', ')} FROM imm_subtitle_lines + WHERE session_id = ? ORDER BY line_id ASC`, + [remoteSessionId], + ); + const wordOccurrences = remote.query( 'SELECT word_id, occurrence_count FROM imm_word_line_occurrences WHERE line_id = ?', ); - const kanjiOccurrences = remote.prepare( + const kanjiOccurrences = remote.query( 'SELECT kanji_id, occurrence_count FROM imm_kanji_line_occurrences WHERE line_id = ?', ); - const insertWordOccurrence = local.prepare( + const insertWordOccurrence = local.query( `INSERT INTO imm_word_line_occurrences (line_id, word_id, occurrence_count) VALUES (?, ?, ?) ON CONFLICT(line_id, word_id) DO UPDATE SET occurrence_count = occurrence_count + excluded.occurrence_count`, ); - const insertKanjiOccurrence = local.prepare( + const insertKanjiOccurrence = local.query( `INSERT INTO imm_kanji_line_occurrences (line_id, kanji_id, occurrence_count) VALUES (?, ?, ?) ON CONFLICT(line_id, kanji_id) DO UPDATE SET occurrence_count = occurrence_count + excluded.occurrence_count`, ); for (const row of rows) { - const localEventId = row.event_id === null ? null : (eventIdMap.get(Number(row.event_id)) ?? null); - const localAnimeId = row.anime_id === null ? null : (animeIdMap.get(Number(row.anime_id)) ?? null); + const localEventId = + row.event_id === null ? null : (eventIdMap.get(Number(row.event_id)) ?? null); + const localAnimeId = + row.anime_id === null ? null : (animeIdMap.get(Number(row.anime_id)) ?? null); const localLineId = insertRow( local, 'imm_subtitle_lines', @@ -243,13 +242,13 @@ function copySubtitleLines( ); summary.subtitleLinesAdded += 1; - for (const occurrence of wordOccurrences.all(row.line_id)) { + for (const occurrence of wordOccurrences.all(row.line_id) as SqlRow[]) { const localWordId = lexicon.resolveWord(Number(occurrence.word_id)); const count = Number(occurrence.occurrence_count); insertWordOccurrence.run(localLineId, localWordId, count); lexicon.addWordOccurrences(Number(occurrence.word_id), count); } - for (const occurrence of kanjiOccurrences.all(row.line_id)) { + for (const occurrence of kanjiOccurrences.all(row.line_id) as SqlRow[]) { const localKanjiId = lexicon.resolveKanji(Number(occurrence.kanji_id)); const count = Number(occurrence.occurrence_count); insertKanjiOccurrence.run(localLineId, localKanjiId, count); @@ -267,14 +266,14 @@ function copySubtitleLines( * applied, even if the merged session started earlier that day. */ function applyMergedSessionLifetime( - local: Database, + local: SyncDb, sessionId: number, videoId: number, session: SqlRow, ): void { const updatedAtMs = nowDbTimestamp(); const applied = local - .prepare( + .query( `INSERT INTO imm_lifetime_applied_sessions (session_id, applied_at_ms, CREATED_DATE, LAST_UPDATE_DATE) VALUES (?, ?, ?, ?) ON CONFLICT(session_id) DO NOTHING`, @@ -282,15 +281,15 @@ function applyMergedSessionLifetime( .run(sessionId, session.ended_at_ms, updatedAtMs, updatedAtMs); if (applied.changes <= 0) return; - const telemetry = local - .query( - `SELECT active_watched_ms, cards_mined, lines_seen, tokens_seen - FROM imm_session_telemetry - WHERE session_id = ? - ORDER BY sample_ms DESC, telemetry_id DESC - LIMIT 1`, - ) - .get(sessionId); + const telemetry = selectOne( + local, + `SELECT active_watched_ms, cards_mined, lines_seen, tokens_seen + FROM imm_session_telemetry + WHERE session_id = ? + ORDER BY sample_ms DESC, telemetry_id DESC + LIMIT 1`, + [sessionId], + ); const metric = (telemetryValue: unknown, sessionValue: unknown): number => { const fromTelemetry = telemetry ? Number(telemetryValue) : Number.NaN; @@ -302,15 +301,18 @@ function applyMergedSessionLifetime( const linesSeen = metric(telemetry?.lines_seen, session.lines_seen); const tokensSeen = metric(telemetry?.tokens_seen, session.tokens_seen); - const video = local - .query('SELECT anime_id, watched FROM imm_videos WHERE video_id = ?') - .get(videoId); + const video = selectOne(local, 'SELECT anime_id, watched FROM imm_videos WHERE video_id = ?', [ + videoId, + ]); const watched = Number(video?.watched ?? 0); - const animeId = video?.anime_id === null || video?.anime_id === undefined ? null : Number(video.anime_id); + const animeId = + video?.anime_id === null || video?.anime_id === undefined ? null : Number(video.anime_id); - const mediaLifetime = local - .query('SELECT completed FROM imm_lifetime_media WHERE video_id = ?') - .get(videoId); + const mediaLifetime = selectOne( + local, + 'SELECT completed FROM imm_lifetime_media WHERE video_id = ?', + [videoId], + ); const hasOtherSessionForVideo = Boolean( local .query('SELECT 1 FROM imm_sessions WHERE video_id = ? AND session_id != ? LIMIT 1') @@ -333,16 +335,19 @@ function applyMergedSessionLifetime( let animeCompletedDelta = 0; if (animeId !== null && watched > 0 && isFirstCompletedSessionForVideoRun) { - const animeLifetime = local - .query('SELECT episodes_completed FROM imm_lifetime_anime WHERE anime_id = ?') - .get(animeId); - const anime = local - .query('SELECT episodes_total FROM imm_anime WHERE anime_id = ?') - .get(animeId); + const animeLifetime = selectOne( + local, + 'SELECT episodes_completed FROM imm_lifetime_anime WHERE anime_id = ?', + [animeId], + ); + const anime = selectOne(local, 'SELECT episodes_total FROM imm_anime WHERE anime_id = ?', [ + animeId, + ]); const episodesCompletedBefore = Number(animeLifetime?.episodes_completed ?? 0); - const episodesTotal = anime?.episodes_total === null || anime?.episodes_total === undefined - ? null - : Number(anime.episodes_total); + const episodesTotal = + anime?.episodes_total === null || anime?.episodes_total === undefined + ? null + : Number(anime.episodes_total); if ( episodesTotal !== null && episodesTotal > 0 && @@ -354,7 +359,7 @@ function applyMergedSessionLifetime( } local - .prepare( + .query( `UPDATE imm_lifetime_global SET total_sessions = total_sessions + 1, total_active_ms = total_active_ms + ?, @@ -377,7 +382,7 @@ function applyMergedSessionLifetime( ); local - .prepare( + .query( `INSERT INTO imm_lifetime_media( video_id, total_sessions, total_active_ms, total_cards, total_lines_seen, total_tokens_seen, completed, first_watched_ms, last_watched_ms, CREATED_DATE, LAST_UPDATE_DATE @@ -419,7 +424,7 @@ function applyMergedSessionLifetime( if (animeId !== null) { local - .prepare( + .query( `INSERT INTO imm_lifetime_anime( anime_id, total_sessions, total_active_ms, total_cards, total_lines_seen, total_tokens_seen, episodes_started, episodes_completed, first_watched_ms, diff --git a/src/core/services/stats-sync/merge.ts b/src/core/services/stats-sync/merge.ts new file mode 100644 index 00000000..b7b0c1c6 --- /dev/null +++ b/src/core/services/stats-sync/merge.ts @@ -0,0 +1,108 @@ +import fs from 'node:fs'; +import { + LexiconResolver, + mergeAnime, + mergeExcludedWords, + mergeMediaMetadata, + mergeVideos, +} from './merge-catalog'; +import { mergeSessions } from './merge-sessions'; +import { copyRemoteOnlyRollups, refreshRollupsForNewSessions } from './merge-rollups'; +import { + assertMergeableSchema, + createEmptyMergeSummary, + type SyncMergeSummary, +} from './shared'; +import type { OpenSyncDb, SyncDb } from './driver'; + +export type { SyncMergeSummary } from './shared'; + +/** + * 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( + open: OpenSyncDb, + 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 = open(snapshotPath, { readonly: true }); + let local: SyncDb; + try { + local = open(localDbPath, { readwrite: true, create: false }); + } catch (error) { + remote.close(); + throw error; + } + try { + assertMergeableSchema(remote, 'Snapshot'); + assertMergeableSchema(local, 'Local'); + + const summary = createEmptyMergeSummary(); + local.exec('PRAGMA foreign_keys = ON'); + local.exec('PRAGMA busy_timeout = 5000'); + local.exec('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.exec('COMMIT'); + return summary; + } catch (error) { + local.exec('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'); +} diff --git a/src/core/services/stats-sync/shared.ts b/src/core/services/stats-sync/shared.ts new file mode 100644 index 00000000..aefd5b3d --- /dev/null +++ b/src/core/services/stats-sync/shared.ts @@ -0,0 +1,157 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { SCHEMA_VERSION } from '../immersion-tracker/types'; +import { resolveConfigDir } from '../../../config/path-resolution'; +import { withReadonlyWalRetry } from './wal-retry'; +import { selectOne, type OpenSyncDb, type SyncDb } from './driver'; + +export { SCHEMA_VERSION }; + +export type { SyncMergeSummary } from '../../../shared/sync/sync-events'; +import type { SyncMergeSummary } from '../../../shared/sync/sync-events'; + +export function createEmptyMergeSummary(): SyncMergeSummary { + return { + sessionsMerged: 0, + sessionsAlreadyPresent: 0, + activeSessionsSkipped: 0, + animeAdded: 0, + videosAdded: 0, + wordsAdded: 0, + kanjiAdded: 0, + subtitleLinesAdded: 0, + telemetryRowsAdded: 0, + eventsAdded: 0, + excludedWordsAdded: 0, + dailyRollupsCopied: 0, + monthlyRollupsCopied: 0, + rollupGroupsRecomputed: 0, + }; +} + +export function nowDbTimestamp(): string { + return String(Date.now()); +} + +export function tableExists(db: SyncDb, tableName: string): boolean { + return Boolean( + db.query(`SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?`).get(tableName), + ); +} + +export function readSchemaVersion(db: SyncDb): number | null { + if (!tableExists(db, 'imm_schema_version')) return null; + const row = selectOne(db, 'SELECT MAX(schema_version) AS schema_version FROM imm_schema_version'); + return typeof row?.schema_version === 'number' ? row.schema_version : null; +} + +export function assertMergeableSchema(db: SyncDb, label: string): void { + const version = readSchemaVersion(db); + if (version === null) { + throw new Error( + `${label} database has no schema version. Run SubMiner once on that machine so the stats database is initialized.`, + ); + } + if (version !== SCHEMA_VERSION) { + throw new Error( + `${label} database is at schema version ${version} but this SubMiner install expects ${SCHEMA_VERSION}. Update SubMiner on both machines to the same version and run each app once before syncing.`, + ); + } + for (const table of ['imm_sessions', 'imm_videos', 'imm_lifetime_global']) { + if (!tableExists(db, table)) { + throw new Error(`${label} database is missing table ${table}; cannot sync.`); + } + } +} + +export function insertRow( + db: SyncDb, + table: string, + columns: readonly string[], + values: unknown[], +): number { + const sql = `INSERT INTO ${table} (${columns.join(', ')}) VALUES (${columns.map(() => '?').join(', ')})`; + // query() caches the prepared statement per SQL string; this runs once per + // copied row, so re-preparing each time would dominate merge time. + const result = db.query(sql).run(...values); + return Number(result.lastInsertRowid); +} + +export function createDbSnapshot(open: OpenSyncDb, dbPath: string, outPath: string): void { + if (!fs.existsSync(dbPath)) { + throw new Error(`Stats database not found: ${dbPath}`); + } + fs.rmSync(outPath, { force: true }); + fs.mkdirSync(path.dirname(outPath), { recursive: true }); + withReadonlyWalRetry(dbPath, (options) => { + const db = open(dbPath, options); + try { + assertMergeableSchema(db, 'Local'); + db.query('VACUUM INTO ?').run(outPath); + } finally { + db.close(); + } + }); +} + +interface DaemonStateFile { + pid?: unknown; +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + // EPERM means the process exists but we can't signal it → still alive. + // Only ESRCH (no such process) means it's actually gone. + return (error as NodeJS.ErrnoException)?.code === 'EPERM'; + } +} + +function statsDaemonStateCandidates(dbPath: string): string[] { + const homeDir = os.homedir(); + const candidates = new Set([path.join(path.dirname(dbPath), 'stats-daemon.json')]); + const configDir = resolveConfigDir({ + platform: process.platform, + appDataDir: process.env.APPDATA, + xdgConfigHome: process.env.XDG_CONFIG_HOME, + homeDir, + existsSync: fs.existsSync, + }); + candidates.add(path.join(configDir, 'stats-daemon.json')); + if (process.platform === 'darwin') { + candidates.add( + path.join(homeDir, 'Library', 'Application Support', 'SubMiner', 'stats-daemon.json'), + ); + } + return [...candidates]; +} + +/** + * Best-effort guard against merging while a SubMiner process holds the + * tracker's write queue in memory. Detects the background stats daemon via + * its pid state file; the interactive app is caught by the mpv-socket check + * in the sync flow. + */ +export function findLiveStatsDaemonPid(dbPath: string): number | null { + for (const statePath of statsDaemonStateCandidates(dbPath)) { + let raw: string; + try { + raw = fs.readFileSync(statePath, 'utf8'); + } catch { + continue; + } + try { + const parsed = JSON.parse(raw) as DaemonStateFile; + const pid = typeof parsed.pid === 'number' && Number.isInteger(parsed.pid) ? parsed.pid : 0; + if (pid > 0 && isProcessAlive(pid)) { + return pid; + } + } catch { + continue; + } + } + return null; +} diff --git a/src/core/services/stats-sync/ssh.ts b/src/core/services/stats-sync/ssh.ts new file mode 100644 index 00000000..4bb16198 --- /dev/null +++ b/src/core/services/stats-sync/ssh.ts @@ -0,0 +1,130 @@ +import { spawnSync } from 'node:child_process'; + +export interface RemoteRunResult { + status: number; + stdout: string; + stderr: string; +} + +/** + * ssh/scp have no `--` terminator for the destination, so a host that starts + * with `-` (e.g. `-oProxyCommand=...`) is parsed as an option. Reject those + * before spawning. + */ +export function assertSafeSshHost(host: string): void { + if (host.startsWith('-')) { + throw new Error(`Refusing to use SSH host that looks like an option: ${host}`); + } +} + +/** + * Run a command on the SSH host. stdin stays attached so interactive prompts + * can still read from the terminal; stdout/stderr are captured for callers + * that need actionable remote failure messages. + */ +export function runSsh(host: string, remoteCommand: string): RemoteRunResult { + assertSafeSshHost(host); + const result = spawnSync('ssh', [host, remoteCommand], { + encoding: 'utf8', + stdio: ['inherit', 'pipe', 'pipe'], + }); + if (result.error) { + throw new Error(`Failed to run ssh: ${(result.error as Error).message}`); + } + return { status: result.status ?? 1, stdout: result.stdout ?? '', stderr: result.stderr ?? '' }; +} + +function assertSafeScpEndpoint(endpoint: string): void { + const colon = endpoint.indexOf(':'); + const slash = endpoint.indexOf('/'); + if (colon <= 0 || (slash !== -1 && slash < colon)) { + if (endpoint.startsWith('-')) { + throw new Error(`Refusing to use scp endpoint that looks like an option: ${endpoint}`); + } + return; + } + + const host = endpoint.slice(0, colon); + const remotePath = endpoint.slice(colon + 1); + assertSafeSshHost(host); + if (remotePath.startsWith('-')) { + throw new Error(`Refusing to use scp remote path that looks like an option: ${remotePath}`); + } +} + +export function runScp(from: string, to: string): void { + assertSafeScpEndpoint(from); + assertSafeScpEndpoint(to); + const result = spawnSync('scp', ['-q', from, to], { + encoding: 'utf8', + stdio: ['inherit', 'inherit', 'inherit'], + }); + if (result.error) { + throw new Error(`Failed to run scp: ${(result.error as Error).message}`); + } + if ((result.status ?? 1) !== 0) { + throw new Error(`scp failed copying ${from} -> ${to}`); + } +} + +export function shellQuote(value: string): string { + return `'${value.replaceAll("'", `'\\''`)}'`; +} + +const REMOTE_RUNTIME_PATH = + 'PATH="$HOME/.local/bin:$HOME/.bun/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:$PATH"'; + +// The Electron app answers the same launcher-style `sync ...` argv when +// invoked with --sync-cli, so a remote machine only needs the app installed; +// the command-line launcher is one candidate, not a requirement. +const APP_SYNC_CLI_FLAG = '--sync-cli'; + +interface RemoteCandidate { + invocation: string; +} + +function defaultRemoteCandidates(): RemoteCandidate[] { + return [ + // Command-line launcher (bun script) on PATH or in its default install dir. + { invocation: 'subminer' }, + { invocation: '~/.local/bin/subminer' }, + // The app binary itself in sync-CLI mode. + { invocation: `SubMiner ${APP_SYNC_CLI_FLAG}` }, + { invocation: `/Applications/SubMiner.app/Contents/MacOS/SubMiner ${APP_SYNC_CLI_FLAG}` }, + { invocation: `~/Applications/SubMiner.app/Contents/MacOS/SubMiner ${APP_SYNC_CLI_FLAG}` }, + ]; +} + +/** + * Non-interactive SSH shells often miss user-installed launchers and Bun. + * Probe candidates under the same deterministic PATH used by sync itself. + * Trusted defaults stay unquoted so the remote shell expands `~`; a + * user-supplied override is shell-quoted to prevent command injection and + * probed both as an app binary (--sync-cli) and as a launcher. + */ +export function resolveRemoteSubminerCommand( + host: string, + preferred: string | null, + runRemote: typeof runSsh = runSsh, +): string { + const candidates: RemoteCandidate[] = preferred + ? [ + // App binaries also answer plain --help by opening the GUI-oriented + // help path, so probe the sync-CLI shape first. + { invocation: `${shellQuote(preferred)} ${APP_SYNC_CLI_FLAG}` }, + { invocation: shellQuote(preferred) }, + ] + : defaultRemoteCandidates(); + for (const candidate of candidates) { + const command = `${REMOTE_RUNTIME_PATH} ${candidate.invocation}`; + const probe = runRemote(host, `${command} --help >/dev/null 2>&1`); + if (probe.status === 0) { + return command; + } + } + throw new Error( + preferred + ? `Remote command not found on ${host}: ${preferred}` + : `SubMiner not found on ${host} (tried the subminer launcher on PATH and ~/.local/bin, and the SubMiner app binary). Pass --remote-cmd pointing at the SubMiner app or launcher.`, + ); +} diff --git a/src/core/services/stats-sync/sync-flow.ts b/src/core/services/stats-sync/sync-flow.ts new file mode 100644 index 00000000..40d64e62 --- /dev/null +++ b/src/core/services/stats-sync/sync-flow.ts @@ -0,0 +1,370 @@ +import os from 'node:os'; +import path from 'node:path'; +import { shellQuote } from './ssh'; +import type { RemoteRunResult } from './ssh'; +import type { SyncMergeSummary, SyncProgressEvent } from '../../../shared/sync/sync-events'; +import type { SyncResultStatus } from '../../../shared/sync/sync-hosts-store'; + +export interface SyncFlowArgs { + sync: boolean; + syncHost: string; + syncSnapshotPath: string; + syncMergePath: string; + syncDirection: 'both' | 'push' | 'pull' | null; + syncRemoteCmd: string; + syncDbPath: string; + syncForce: boolean; + syncJson: boolean; + syncCheck: boolean; + logLevel: string; +} + +export interface SyncFlowContext { + args: SyncFlowArgs; + mpvSocketPath: string; +} + +/** + * Everything the sync flow touches outside plain computation. The launcher + * binds these to bun:sqlite-backed engine functions and its own logger; the + * app's --sync-cli mode binds libsql and console output. Tests stub freely. + */ +export interface SyncFlowDeps { + createDbSnapshot: (dbPath: string, outPath: string) => void; + mergeSnapshotIntoDb: (localDbPath: string, snapshotPath: string) => SyncMergeSummary; + formatMergeSummary: (summary: SyncMergeSummary) => string; + findLiveStatsDaemonPid: (dbPath: string) => number | null; + assertSafeSshHost: (host: string) => void; + resolveRemoteSubminerCommand: (host: string, preferred: string | null) => string; + runScp: (from: string, to: string) => void; + runSsh: (host: string, remoteCommand: string) => RemoteRunResult; + fail: (message: string) => never; + log: (level: string, configuredLevel: string, message: string) => void; + canConnectUnixSocket: (socketPath: string) => Promise; + realpathSync: (candidate: string) => string; + mkdtempSync: (prefix: string) => string; + rmSync: (target: string, options: { recursive: boolean; force: boolean }) => void; + consoleLog: (message: string) => void; + writeStdout: (text: string) => boolean; + ensureTrackerQuiescent: (context: SyncFlowContext, dbPath: string) => Promise; + emitEvent: (event: SyncProgressEvent) => void; + recordHostSyncResult: (host: string, status: SyncResultStatus, detail: string | null) => void; + resolveDefaultDbPath: () => string; + resolvePath: (value: string) => string; +} + +export function resolveSyncDbPath(context: SyncFlowContext, deps: SyncFlowDeps): string { + const override = context.args.syncDbPath.trim(); + return override ? deps.resolvePath(override) : deps.resolveDefaultDbPath(); +} + +function isTrackerDb(dbPath: string, deps: SyncFlowDeps): boolean { + const trackerDbPath = deps.resolveDefaultDbPath(); + try { + return deps.realpathSync(dbPath) === deps.realpathSync(trackerDbPath); + } catch { + return dbPath === trackerDbPath; + } +} + +export async function ensureTrackerQuiescentFlow( + context: SyncFlowContext, + dbPath: string, + deps: SyncFlowDeps, +): Promise { + if (context.args.syncForce) return; + // A running SubMiner only holds the tracker's own database; --db pointed + // elsewhere needs no guard. + if (!isTrackerDb(dbPath, deps)) return; + const daemonPid = deps.findLiveStatsDaemonPid(dbPath); + if (daemonPid !== null) { + deps.fail( + `The SubMiner stats server is running (pid ${daemonPid}). Stop it with "subminer stats -s" (or close SubMiner) before syncing, or pass --force.`, + ); + } + if (context.mpvSocketPath && (await deps.canConnectUnixSocket(context.mpvSocketPath))) { + deps.fail( + `An mpv/SubMiner session appears to be running (socket ${context.mpvSocketPath}). Close it before syncing, or pass --force.`, + ); + } +} + +// In --json mode every line on stdout is an NDJSON event: human console output +// is silenced and events are written through the original console logger. +export function withJsonEvents(deps: SyncFlowDeps): SyncFlowDeps { + const writeLine = deps.consoleLog; + return { + ...deps, + consoleLog: () => {}, + writeStdout: () => true, + emitEvent: (event) => writeLine(JSON.stringify(event)), + }; +} + +export function runSnapshotMode( + context: SyncFlowContext, + dbPath: string, + deps: SyncFlowDeps, +): void { + const outPath = deps.resolvePath(context.args.syncSnapshotPath); + deps.emitEvent({ + type: 'stage', + stage: 'snapshot-local', + message: `Snapshotting local database (${dbPath})`, + }); + deps.createDbSnapshot(dbPath, outPath); + deps.emitEvent({ type: 'snapshot-created', path: outPath }); + deps.consoleLog(outPath); +} + +export async function runMergeMode( + context: SyncFlowContext, + dbPath: string, + deps: SyncFlowDeps, +): Promise { + await deps.ensureTrackerQuiescent(context, dbPath); + const snapshotPath = deps.resolvePath(context.args.syncMergePath); + deps.emitEvent({ + type: 'stage', + stage: 'merge-local', + message: `Merging ${snapshotPath} into the local database`, + }); + const summary = deps.mergeSnapshotIntoDb(dbPath, snapshotPath); + deps.emitEvent({ type: 'merge-summary', target: 'local', summary }); + deps.consoleLog(deps.formatMergeSummary(summary)); +} + +function formatHostSyncDetail( + direction: 'both' | 'push' | 'pull', + pulledSummary: SyncMergeSummary | null, +): string { + if (!pulledSummary) return direction === 'push' ? 'Pushed local stats' : 'Sync complete'; + const merged = `${pulledSummary.sessionsMerged} session${pulledSummary.sessionsMerged === 1 ? '' : 's'} merged`; + return direction === 'pull' ? merged : `${merged}; pushed local stats`; +} + +export async function runCheckMode(context: SyncFlowContext, deps: SyncFlowDeps): Promise { + const { args } = context; + const host = args.syncHost; + deps.assertSafeSshHost(host); + + deps.consoleLog(`Checking SSH connection to ${host}...`); + let remoteCommand: string | null = null; + let remoteVersion: string | null = null; + let error: string | null = null; + + const probe = deps.runSsh(host, 'echo subminer-check-ok'); + const sshOk = probe.status === 0 && probe.stdout.includes('subminer-check-ok'); + if (!sshOk) { + error = formatRemoteRunError(`Could not reach ${host} over SSH.`, probe); + } else { + deps.consoleLog('SSH connection: ok'); + try { + remoteCommand = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null); + const version = deps.runSsh(host, `${remoteCommand} --version`); + remoteVersion = version.status === 0 ? version.stdout.trim() || null : null; + deps.consoleLog( + `Remote subminer: ${remoteCommand}${remoteVersion ? ` (${remoteVersion})` : ''}`, + ); + } catch (resolveError) { + error = resolveError instanceof Error ? resolveError.message : String(resolveError); + } + } + + const ok = sshOk && remoteCommand !== null; + deps.emitEvent({ + type: 'check-result', + host, + sshOk, + remoteCommand, + remoteVersion, + ok, + error, + }); + if (!ok) { + throw new Error(error ?? `Connection check failed for ${host}.`); + } + deps.consoleLog('Check passed.'); +} + +function cleanupRemote(host: string, remoteTmpDir: string, deps: SyncFlowDeps): void { + if (!remoteTmpDir.startsWith('/tmp/')) return; + deps.runSsh(host, `rm -rf ${shellQuote(remoteTmpDir)}`); +} + +function formatRemoteRunError(message: string, run: RemoteRunResult): string { + const stderr = run.stderr.trim(); + return stderr ? `${message}\n${stderr}` : message; +} + +export async function runHostSync( + context: SyncFlowContext, + dbPath: string, + deps: SyncFlowDeps, +): Promise { + const { args } = context; + const host = args.syncHost; + const direction = args.syncDirection ?? 'both'; + const shouldPull = direction !== 'push'; + const shouldPush = direction !== 'pull'; + deps.assertSafeSshHost(host); + + await deps.ensureTrackerQuiescent(context, dbPath); + + const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null); + deps.log('debug', args.logLevel, `Remote subminer command: ${remoteCmd}`); + + const localTmpDir = deps.mkdtempSync(path.join(os.tmpdir(), 'subminer-sync-')); + let remoteTmpDir = ''; + let pulledSummary: SyncMergeSummary | null = null; + try { + // Signal failures by throwing (not fail(), which exits synchronously and + // would skip the finally cleanup, leaking temp dirs holding snapshot data). + // main().catch() reports the message the same way fail() would. + const mktemp = deps.runSsh(host, 'mktemp -d /tmp/subminer-sync.XXXXXX'); + remoteTmpDir = mktemp.stdout.trim(); + if (mktemp.status !== 0 || !remoteTmpDir.startsWith('/tmp/')) { + throw new Error(`Could not create a temporary directory on ${host}.`); + } + + const forceFlag = args.syncForce ? ' --force' : ''; + + const localSnapshot = path.join(localTmpDir, 'local.sqlite'); + if (shouldPush) { + deps.consoleLog(`Snapshotting local database (${dbPath})...`); + deps.emitEvent({ + type: 'stage', + stage: 'snapshot-local', + message: `Snapshotting local database (${dbPath})`, + }); + deps.createDbSnapshot(dbPath, localSnapshot); + } + + const remoteSnapshot = `${remoteTmpDir}/snapshot.sqlite`; + if (shouldPull) { + deps.consoleLog(`Snapshotting ${host}...`); + deps.emitEvent({ type: 'stage', stage: 'snapshot-remote', message: `Snapshotting ${host}` }); + const snapshotRun = deps.runSsh( + host, + `${remoteCmd} sync --snapshot ${shellQuote(remoteSnapshot)}${forceFlag}`, + ); + if (snapshotRun.status !== 0) { + throw new Error(formatRemoteRunError(`Remote snapshot failed on ${host}.`, snapshotRun)); + } + } + + const pulledSnapshot = path.join(localTmpDir, 'remote.sqlite'); + if (shouldPull) { + deps.emitEvent({ + type: 'stage', + stage: 'download', + message: `Copying snapshot from ${host}`, + }); + deps.runScp(`${host}:${remoteSnapshot}`, pulledSnapshot); + } + const incomingSnapshot = `${remoteTmpDir}/incoming.sqlite`; + if (shouldPush) { + deps.emitEvent({ type: 'stage', stage: 'upload', message: `Copying snapshot to ${host}` }); + deps.runScp(localSnapshot, `${host}:${incomingSnapshot}`); + } + + if (shouldPull) { + deps.consoleLog(`\nMerging ${host} -> local:`); + deps.emitEvent({ + type: 'stage', + stage: 'merge-local', + message: `Merging ${host} into the local database`, + }); + await deps.ensureTrackerQuiescent(context, dbPath); + const summary = deps.mergeSnapshotIntoDb(dbPath, pulledSnapshot); + pulledSummary = summary; + deps.emitEvent({ type: 'merge-summary', target: 'local', summary }); + deps.consoleLog(deps.formatMergeSummary(summary)); + } + + if (shouldPush) { + deps.consoleLog(`\nMerging local -> ${host}:`); + deps.emitEvent({ + type: 'stage', + stage: 'merge-remote', + message: `Merging the local database into ${host}`, + }); + await deps.ensureTrackerQuiescent(context, dbPath); + const mergeRun = deps.runSsh( + host, + `${remoteCmd} sync --merge ${shellQuote(incomingSnapshot)}${forceFlag}`, + ); + deps.writeStdout(mergeRun.stdout); + if (mergeRun.stdout.trim()) { + deps.emitEvent({ type: 'remote-output', text: mergeRun.stdout }); + } + if (mergeRun.status !== 0) { + const retryCommand = + direction === 'push' ? `subminer sync ${host} --push` : `subminer sync ${host}`; + const localUpdate = shouldPull ? ' The local database was updated;' : ''; + throw new Error( + formatRemoteRunError( + `Remote merge failed on ${host}.${localUpdate} re-run "${retryCommand}" once the remote issue is fixed.`, + mergeRun, + ), + ); + } + } + + deps.consoleLog('\nSync complete.'); + deps.recordHostSyncResult(host, 'success', formatHostSyncDetail(direction, pulledSummary)); + } catch (error) { + try { + deps.recordHostSyncResult( + host, + 'error', + error instanceof Error ? error.message : String(error), + ); + } catch { + // best effort + } + throw error; + } finally { + deps.rmSync(localTmpDir, { recursive: true, force: true }); + if (remoteTmpDir) { + try { + cleanupRemote(host, remoteTmpDir, deps); + } catch { + // best effort + } + } + } +} + +export async function runSyncFlow(context: SyncFlowContext, inputDeps: SyncFlowDeps): Promise { + let deps = inputDeps; + const { args } = context; + if (!args.sync) return false; + if (args.syncJson) deps = withJsonEvents(deps); + + const dbPath = resolveSyncDbPath(context, deps); + try { + if (args.syncCheck) { + await runCheckMode(context, deps); + } else if (args.syncSnapshotPath) { + runSnapshotMode(context, dbPath, deps); + } else if (args.syncMergePath) { + await runMergeMode(context, dbPath, deps); + } else if (args.syncHost) { + await runHostSync(context, dbPath, deps); + } else { + deps.fail('sync requires a host, --snapshot , or --merge .'); + } + } catch (error) { + if (args.syncJson) { + deps.emitEvent({ + type: 'result', + ok: false, + error: error instanceof Error ? error.message : String(error), + }); + } + throw error; + } + if (args.syncJson) deps.emitEvent({ type: 'result', ok: true, error: null }); + return true; +} diff --git a/src/core/services/stats-sync/wal-retry.ts b/src/core/services/stats-sync/wal-retry.ts new file mode 100644 index 00000000..b534a278 --- /dev/null +++ b/src/core/services/stats-sync/wal-retry.ts @@ -0,0 +1,50 @@ +import fs from 'node:fs'; +import type { SyncDbOpenOptions } from './driver'; + +/** + * Opening a WAL-mode SQLite database strictly read-only fails when the -shm + * file is missing or stale (the reader must be able to create it). Retry such + * failures with a read-write handle; the caller still only issues reads. + */ +export function withReadonlyWalRetry( + dbPath: string, + query: (options: SyncDbOpenOptions) => T, +): T { + try { + return query({ readonly: true }); + } catch (error) { + if (!isReadonlyWalRetryError(error, dbPath)) throw error; + return query({ readwrite: true, create: false }); + } +} + +export function isReadonlyWalRetryError(error: unknown, dbPath: string): boolean { + if (!isWalModeSqliteDatabase(dbPath)) return false; + const code = + typeof error === 'object' && error !== null && 'code' in error + ? String((error as { code?: unknown }).code ?? '') + : ''; + const message = error instanceof Error ? error.message : String(error); + const text = `${code} ${message}`.toLowerCase(); + return ( + text.includes('readonly') || + text.includes('read-only') || + text.includes('attempt to write a readonly database') || + text.includes('sqlite_cantopen') || + text.includes('unable to open database file') + ); +} + +function isWalModeSqliteDatabase(dbPath: string): boolean { + const header = Buffer.alloc(20); + let fd: number | null = null; + try { + fd = fs.openSync(dbPath, 'r'); + if (fs.readSync(fd, header, 0, header.length, 0) < header.length) return false; + } catch { + return false; + } finally { + if (fd !== null) fs.closeSync(fd); + } + return header.subarray(0, 16).toString('ascii') === 'SQLite format 3\0' && header[18] === 2; +}