Sync Stats & History window, headless --sync-cli, and Windows remote support (#160)

This commit is contained in:
2026-07-13 18:56:51 -07:00
committed by GitHub
parent 66f8ca4f80
commit 49b926e08c
111 changed files with 6983 additions and 1130 deletions
+81 -2
View File
@@ -16,6 +16,7 @@ function makeArgs(overrides: Partial<CliArgs> = {}): CliArgs {
togglePrimarySubtitleBar: false,
yomitan: false,
settings: false,
syncWindow: false,
setup: false,
show: false,
hide: false,
@@ -140,6 +141,65 @@ test('startAppLifecycle still acquires lock for startup commands', () => {
assert.equal(getLockCalls(), 1);
});
test('startAppLifecycle defers quit until async cleanup settles', async () => {
let willQuit: ((event: { preventDefault(): void }) => void) | null = null;
let releaseCleanup: (() => void) | null = null;
const cleanupDone = new Promise<void>((resolve) => {
releaseCleanup = resolve;
});
let prevented = false;
const { deps, calls } = createDeps({
shouldStartApp: () => true,
onWillQuit: (handler) => {
willQuit = handler;
},
onWillQuitCleanup: () => cleanupDone,
});
startAppLifecycle(makeArgs({ start: true }), deps);
assert.ok(willQuit);
(willQuit as (event: { preventDefault(): void }) => void)({
preventDefault: () => {
prevented = true;
},
});
assert.equal(prevented, true);
assert.deepEqual(calls, []);
assert.ok(releaseCleanup);
(releaseCleanup as () => void)();
await cleanupDone;
// The re-quit must not fire in the microtask turn of the will-quit
// dispatch: Electron drops a quit issued while the prevented quit is
// still unwinding, leaving a windowless process alive.
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
assert.deepEqual(calls, []);
await new Promise((resolve) => setImmediate(resolve));
assert.deepEqual(calls, ['quitApp']);
});
test('startAppLifecycle contains synchronous quit cleanup failures', () => {
let willQuit: ((event: { preventDefault(): void }) => void) | null = null;
const { deps, calls } = createDeps({
shouldStartApp: () => true,
onWillQuit: (handler) => {
willQuit = handler;
},
onWillQuitCleanup: () => {
throw new Error('cleanup exploded');
},
});
startAppLifecycle(makeArgs({ start: true }), deps);
assert.ok(willQuit);
assert.doesNotThrow(() =>
(willQuit as (event: { preventDefault(): void }) => void)({ preventDefault: () => {} }),
);
assert.deepEqual(calls, []);
});
test('startAppLifecycle app ping exits non-zero immediately when no running instance owns the lock', () => {
const { deps, calls, getLockCalls } = createDeps({
shouldStartApp: () => false,
@@ -252,7 +312,7 @@ test('startAppLifecycle routes control socket commands through the second-instan
},
});
let willQuitHandler: (() => void) | null = null;
let willQuitHandler: ((event: { preventDefault(): void }) => void) | null = null;
deps.onWillQuit = (handler) => {
willQuitHandler = handler;
};
@@ -274,7 +334,7 @@ test('startAppLifecycle routes control socket commands through the second-instan
assert.deepEqual(handled, ['ready', 'second-instance:start']);
assert.ok(willQuitHandler);
(willQuitHandler as () => void)();
(willQuitHandler as (event: { preventDefault(): void }) => void)({ preventDefault: () => {} });
assert.deepEqual(handled, ['ready', 'second-instance:start', 'control-close']);
});
@@ -353,3 +413,22 @@ test('startAppLifecycle quits macOS setup-only launch when all windows close', (
handler();
assert.deepEqual(calls, ['quitApp']);
});
test('startAppLifecycle quits macOS sync-window launch when its window closes', () => {
let windowAllClosedHandler: (() => void) | null = null;
const { deps, calls } = createDeps({
shouldStartApp: () => true,
isDarwinPlatform: () => true,
shouldQuitOnWindowAllClosed: () => true,
onWindowAllClosed: (handler) => {
windowAllClosedHandler = handler;
},
});
startAppLifecycle(makeArgs({ syncWindow: true }), deps);
const handler = windowAllClosedHandler as (() => void) | null;
assert.ok(handler);
handler();
assert.deepEqual(calls, ['quitApp']);
});
+38 -6
View File
@@ -16,11 +16,11 @@ export interface AppLifecycleServiceDeps {
startControlServer?: (handleArgv: (argv: string[]) => void) => (() => void) | void;
whenReady: (handler: () => Promise<void>) => void;
onWindowAllClosed: (handler: () => void) => void;
onWillQuit: (handler: () => void) => void;
onWillQuit: (handler: (event: { preventDefault(): void }) => void) => void;
onActivate: (handler: () => void) => void;
isDarwinPlatform: () => boolean;
onReady: () => Promise<void>;
onWillQuitCleanup: () => void;
onWillQuitCleanup: () => void | Promise<void>;
shouldRestoreWindowsOnActivate: () => boolean;
restoreWindowsOnActivate: () => void;
shouldQuitOnWindowAllClosed: () => boolean;
@@ -44,7 +44,7 @@ export interface AppLifecycleDepsRuntimeOptions {
logNoRunningInstance: () => void;
startControlServer?: (handleArgv: (argv: string[]) => void) => (() => void) | void;
onReady: () => Promise<void>;
onWillQuitCleanup: () => void;
onWillQuitCleanup: () => void | Promise<void>;
shouldRestoreWindowsOnActivate: () => boolean;
restoreWindowsOnActivate: () => void;
shouldQuitOnWindowAllClosed: () => boolean;
@@ -183,16 +183,48 @@ export function startAppLifecycle(initialArgs: CliArgs, deps: AppLifecycleServic
deps.onWindowAllClosed(() => {
if (
deps.shouldQuitOnWindowAllClosed() &&
(!deps.isDarwinPlatform() || initialArgs.settings || initialArgs.setup)
(!deps.isDarwinPlatform() ||
initialArgs.settings ||
initialArgs.setup ||
initialArgs.syncWindow)
) {
deps.quitApp();
}
});
deps.onWillQuit(() => {
let quitCleanupPending = false;
let quitCleanupComplete = false;
deps.onWillQuit((event) => {
if (quitCleanupComplete) return;
stopControlServer?.();
stopControlServer = null;
deps.onWillQuitCleanup();
if (quitCleanupPending) {
event.preventDefault();
return;
}
let cleanup: void | Promise<void>;
try {
cleanup = deps.onWillQuitCleanup();
} catch (error) {
logger.error('App quit cleanup failed:', error);
return;
}
if (!(cleanup instanceof Promise)) return;
quitCleanupPending = true;
event.preventDefault();
void cleanup
.catch((error) => {
logger.error('App quit cleanup failed:', error);
})
.finally(() => {
quitCleanupPending = false;
quitCleanupComplete = true;
// A cleanup promise that settles in a microtask would re-quit while
// Electron is still unwinding the prevented quit, and that quit call
// is silently dropped, leaving a windowless process alive. Re-issue
// the quit from a fresh macrotask instead.
setImmediate(() => deps.quitApp());
});
});
deps.onActivate(() => {
+5
View File
@@ -21,6 +21,7 @@ function makeArgs(overrides: Partial<CliArgs> = {}): CliArgs {
toggleVisibleOverlay: false,
yomitan: false,
settings: false,
syncWindow: false,
setup: false,
show: false,
hide: false,
@@ -138,6 +139,9 @@ function createDeps(overrides: Partial<CliCommandServiceDeps> = {}) {
openConfigSettingsWindow: () => {
calls.push('openConfigSettingsWindow');
},
openSyncUiWindow: () => {
calls.push('openSyncUiWindow');
},
openFirstRunSetup: (force?: boolean) => {
calls.push(`openFirstRunSetup:${force === true ? 'force' : 'default'}`);
},
@@ -660,6 +664,7 @@ test('createCliCommandDepsRuntime reconnects MPV client when reconnect hook exis
openFirstRunSetup: () => {},
openYomitanSettings: () => {},
openConfigSettingsWindow: () => {},
openSyncUiWindow: () => {},
cycleSecondarySubMode: () => {},
openRuntimeOptionsPalette: () => {},
printHelp: () => {},
+5
View File
@@ -44,6 +44,7 @@ export interface CliCommandServiceDeps {
openFirstRunSetup: (force?: boolean) => void;
openYomitanSettingsDelayed: (delayMs: number) => void;
openConfigSettingsWindow: () => void;
openSyncUiWindow: () => void;
setVisibleOverlayVisible: (visible: boolean) => void;
copyCurrentSubtitle: () => void;
startPendingMultiCopy: (timeoutMs: number) => void;
@@ -170,6 +171,7 @@ interface UiCliRuntime {
openFirstRunSetup: (force?: boolean) => void;
openYomitanSettings: () => void;
openConfigSettingsWindow: () => void;
openSyncUiWindow: () => void;
cycleSecondarySubMode: () => void;
openRuntimeOptionsPalette: () => void;
printHelp: () => void;
@@ -274,6 +276,7 @@ export function createCliCommandDepsRuntime(
}, delayMs);
},
openConfigSettingsWindow: options.ui.openConfigSettingsWindow,
openSyncUiWindow: options.ui.openSyncUiWindow,
setVisibleOverlayVisible: options.overlay.setVisible,
copyCurrentSubtitle: options.mining.copyCurrentSubtitle,
startPendingMultiCopy: options.mining.startPendingMultiCopy,
@@ -417,6 +420,8 @@ export function handleCliCommand(
deps.openYomitanSettingsDelayed(1000);
} else if (args.settings) {
deps.openConfigSettingsWindow();
} else if (args.syncWindow) {
deps.openSyncUiWindow();
} else if (args.show || args.showVisibleOverlay) {
deps.setVisibleOverlayVisible(true);
} else if (args.hide || args.hideVisibleOverlay) {
@@ -16,6 +16,7 @@ function makeArgs(overrides: Partial<CliArgs> = {}): CliArgs {
togglePrimarySubtitleBar: false,
yomitan: false,
settings: false,
syncWindow: false,
setup: false,
show: false,
hide: false,
@@ -0,0 +1,82 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import { extractSyncCliTokens, parseSyncCliTokens } from './cli-args';
test('extractSyncCliTokens returns tokens after --sync-cli', () => {
assert.equal(extractSyncCliTokens(['/bin/electron', '/app']), null);
assert.deepEqual(extractSyncCliTokens(['/bin/electron', '/app', '--sync-cli', 'sync', 'host']), [
'sync',
'host',
]);
// A repeated flag (e.g. resolved remote command + forwarded argv) is ignored.
assert.deepEqual(extractSyncCliTokens(['/app', '--sync-cli', '--sync-cli', 'sync', 'h']), [
'sync',
'h',
]);
});
test('parseSyncCliTokens handles help, version, and run modes', () => {
assert.deepEqual(parseSyncCliTokens(['--help']), { kind: 'help' });
assert.deepEqual(parseSyncCliTokens(['--version']), { kind: 'version' });
const run = parseSyncCliTokens(['sync', 'media-box', '--pull', '--force', '--json']);
assert.equal(run.kind, 'run');
if (run.kind === 'run') {
assert.equal(run.args.syncHost, 'media-box');
assert.equal(run.args.syncDirection, 'pull');
assert.equal(run.args.syncForce, true);
assert.equal(run.args.syncJson, true);
}
const snapshot = parseSyncCliTokens(['sync', '--snapshot', '/tmp/x.sqlite', '--db', '/tmp/db']);
assert.equal(snapshot.kind, 'run');
if (snapshot.kind === 'run') {
assert.equal(snapshot.args.syncSnapshotPath, '/tmp/x.sqlite');
assert.equal(snapshot.args.syncDbPath, '/tmp/db');
}
});
test('parseSyncCliTokens handles the temp-dir protocol modes', () => {
const makeTemp = parseSyncCliTokens(['sync', '--make-temp']);
assert.equal(makeTemp.kind, 'run');
if (makeTemp.kind === 'run') assert.equal(makeTemp.args.syncMakeTemp, true);
const removeTemp = parseSyncCliTokens(['sync', '--remove-temp', '/tmp/subminer-sync-x']);
assert.equal(removeTemp.kind, 'run');
if (removeTemp.kind === 'run') {
assert.equal(removeTemp.args.syncRemoveTempPath, '/tmp/subminer-sync-x');
}
assert.equal(parseSyncCliTokens(['sync', '--make-temp', 'host']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', '--make-temp', '--remove-temp', '/tmp/x']).kind, 'error');
});
test('parseSyncCliTokens owns the sync CLI validation rules', () => {
assert.equal(parseSyncCliTokens([]).kind, 'error');
assert.equal(parseSyncCliTokens(['sync']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', 'h', '--push', '--pull']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', '--snapshot', '/tmp/x', '--push']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', '--check']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', '--check', '--snapshot', '/tmp/x', 'h']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', 'h', '--snapshot', '/tmp/x']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', 'h', '--bogus']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', 'h', 'extra']).kind, 'error');
assert.equal(parseSyncCliTokens(['sync', '--snapshot']).kind, 'error');
});
test('parseSyncCliTokens rejects an option-like token where a value is required', () => {
// Without this guard `--snapshot --force` writes a snapshot to a file literally
// named "--force" and silently drops the flag.
assert.deepEqual(parseSyncCliTokens(['sync', '--snapshot', '--force']), {
kind: 'error',
message: 'Missing value for --snapshot.',
});
assert.deepEqual(parseSyncCliTokens(['sync', '--remove-temp', '--force']), {
kind: 'error',
message: 'Missing value for --remove-temp.',
});
// The `--flag=<value>` form still accepts values that begin with "-".
const parsed = parseSyncCliTokens(['sync', '--snapshot=-weird-name.sqlite']);
assert.equal(parsed.kind, 'run');
assert.equal(parsed.kind === 'run' && parsed.args.syncSnapshotPath, '-weird-name.sqlite');
});
+165
View File
@@ -0,0 +1,165 @@
import type { SyncFlowArgs } from './sync-flow';
export const SYNC_CLI_FLAG = '--sync-cli';
export type ParsedSyncCli =
| { kind: 'help' }
| { kind: 'version' }
| { kind: 'run'; args: SyncFlowArgs }
| { kind: 'error'; message: string };
export function extractSyncCliTokens(argv: readonly string[]): string[] | null {
const index = argv.indexOf(SYNC_CLI_FLAG);
if (index === -1) return null;
return argv.slice(index + 1).filter((token) => token !== SYNC_CLI_FLAG);
}
/**
* Parse launcher-style sync argv (`sync [host] [--snapshot f] ...`) for the
* app's --sync-cli mode. This is the single owner of sync CLI validation:
* the launcher forwards `subminer sync` tokens verbatim, so both entry
* points accept the same command lines and fail the same way.
*/
export function parseSyncCliTokens(tokens: readonly string[]): ParsedSyncCli {
if (tokens.includes('--help') || tokens.includes('-h')) return { kind: 'help' };
if (tokens.includes('--version') || tokens.includes('-V')) return { kind: 'version' };
const rest = [...tokens];
if (rest[0] !== 'sync') {
return {
kind: 'error',
message: `Expected a "sync" command after ${SYNC_CLI_FLAG} (e.g. ${SYNC_CLI_FLAG} sync --snapshot <file>).`,
};
}
rest.shift();
let host = '';
let snapshot = '';
let merge = '';
let push = false;
let pull = false;
let check = false;
let force = false;
let json = false;
let makeTemp = false;
let removeTemp = '';
let remoteCmd = '';
let dbPath = '';
let logLevel = 'warn';
const valueFlags = new Map<string, (value: string) => void>([
['--snapshot', (value) => (snapshot = value.trim())],
['--merge', (value) => (merge = value.trim())],
['--remove-temp', (value) => (removeTemp = value.trim())],
['--remote-cmd', (value) => (remoteCmd = value.trim())],
['--db', (value) => (dbPath = value.trim())],
['--log-level', (value) => (logLevel = value.trim() || 'warn')],
]);
for (let i = 0; i < rest.length; i += 1) {
const token = rest[i]!;
const assignValue = valueFlags.get(token.includes('=') ? token.slice(0, token.indexOf('=')) : token);
if (assignValue) {
if (token.includes('=')) {
assignValue(token.slice(token.indexOf('=') + 1));
continue;
}
const value = rest[i + 1];
// An option-like token is never a value: `--snapshot --force` must fail
// loudly instead of writing a snapshot to a file named "--force". Paths
// that really do start with "-" can still be passed as `--snapshot=-x`.
if (value === undefined || value.startsWith('-')) {
return { kind: 'error', message: `Missing value for ${token}.` };
}
assignValue(value);
i += 1;
continue;
}
if (token === '--push') push = true;
else if (token === '--pull') pull = true;
else if (token === '--check') check = true;
else if (token === '--force' || token === '-f') force = true;
else if (token === '--json') json = true;
else if (token === '--make-temp') makeTemp = true;
else if (token.startsWith('-')) {
return { kind: 'error', message: `Unknown sync option: ${token}` };
} else if (host) {
return { kind: 'error', message: `Unexpected extra argument: ${token}` };
} else {
host = token.trim();
}
}
if (push && pull) return { kind: 'error', message: 'Sync --push and --pull cannot be combined.' };
if ((push || pull) && !host) {
return { kind: 'error', message: 'Sync --push and --pull require a host.' };
}
if (check && !host) return { kind: 'error', message: 'Sync --check requires a host.' };
if (check && (push || pull || snapshot || merge)) {
return {
kind: 'error',
message: 'Sync --check cannot be combined with --push, --pull, --snapshot, or --merge.',
};
}
const modes = [
Boolean(host),
Boolean(snapshot),
Boolean(merge),
makeTemp,
Boolean(removeTemp),
].filter(Boolean).length;
if (modes === 0) {
return { kind: 'error', message: 'Sync requires a host, --snapshot <file>, or --merge <file>.' };
}
if (modes > 1) {
return {
kind: 'error',
message: 'Sync host, --snapshot, --merge, --make-temp, and --remove-temp cannot be combined.',
};
}
return {
kind: 'run',
args: {
syncHost: host,
syncSnapshotPath: snapshot,
syncMergePath: merge,
syncDirection: push ? 'push' : pull ? 'pull' : 'both',
syncRemoteCmd: remoteCmd,
syncDbPath: dbPath,
syncForce: force,
syncJson: json,
syncCheck: check,
syncMakeTemp: makeTemp,
syncRemoveTempPath: removeTemp,
logLevel,
},
};
}
export function syncCliUsage(): string {
return [
'SubMiner sync CLI',
'',
`Usage: SubMiner ${SYNC_CLI_FLAG} sync [host] [options]`,
'',
'Modes (exactly one):',
' <host> Sync stats with an SSH destination (user@host or ssh alias)',
' --snapshot <file> Write a consistent snapshot of the local stats database',
' --merge <file> Merge a snapshot database file into the local stats database',
' --make-temp Create a sync temp directory and print its path (used over SSH)',
' --remove-temp <dir> Remove a sync temp directory created by --make-temp',
'',
'Options:',
' --push Only merge local stats into the SSH host',
' --pull Only merge stats from the SSH host into the local database',
' --check Test the SSH connection and remote SubMiner availability',
' --db <file> Override the local stats database path',
' --remote-cmd <cmd> SubMiner app or launcher command to run on the remote host',
' -f, --force Skip the running-app safety check',
' --json Emit machine-readable NDJSON progress output',
' --log-level <level> Log level',
' --help Show this help',
' --version Show the SubMiner version',
].join('\n');
}
+42
View File
@@ -0,0 +1,42 @@
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { resolveConfigFilePath } from '../../../config/path-resolution';
import { parseConfigContent } from '../../../config/parse';
import { getDefaultConfigDir } from '../../../shared/setup-state';
/**
* Default immersion stats database location, shared by the launcher's history
* command and the app's --sync-cli mode: honor a configured
* immersionTracking.dbPath (raw main-config read, tolerant of comments), else
* <configDir>/immersion.sqlite. Electron- and libsql-free on purpose so the
* bun launcher can import it.
*/
export function resolveImmersionDbPath(): string {
const configPath = resolveConfigFilePath({
appDataDir: process.env.APPDATA,
xdgConfigHome: process.env.XDG_CONFIG_HOME,
homeDir: os.homedir(),
existsSync: fs.existsSync,
});
let configured = '';
try {
const parsed = parseConfigContent(configPath, fs.readFileSync(configPath, 'utf8'));
const tracking =
parsed && typeof parsed === 'object' && !Array.isArray(parsed)
? (parsed as Record<string, unknown>).immersionTracking
: null;
if (tracking && typeof tracking === 'object' && !Array.isArray(tracking)) {
const dbPath = (tracking as Record<string, unknown>).dbPath;
if (typeof dbPath === 'string') configured = dbPath.trim();
}
} catch {
// no config or unreadable config → default location
}
if (configured) {
return configured.startsWith('~')
? path.join(os.homedir(), configured.slice(1))
: configured;
}
return path.join(getDefaultConfigDir(), 'immersion.sqlite');
}
@@ -0,0 +1,68 @@
import { Database } from '../immersion-tracker/sqlite';
import type { SyncDbOpenOptions } from './wal-retry';
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 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 SqlRow = Record<string, unknown>;
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;
}
interface LibsqlDatabase {
prepare(sql: string): SyncDbStatement;
exec(sql: string): unknown;
close(): unknown;
}
/**
* libsql (better-sqlite3 API) SQLite connection for the stats-sync engine.
* prepare() is not cached by libsql, so query() keeps a per-connection
* statement cache because the merge prepares a handful of statements and runs them
* once per copied row, so re-preparing would dominate merge time.
*/
export function openLibsqlSyncDb(dbPath: string, options: SyncDbOpenOptions): SyncDb {
const db = new Database(dbPath, {
readonly: options.readonly === true,
fileMustExist: options.create !== true,
}) as unknown as LibsqlDatabase;
const statements = new Map<string, SyncDbStatement>();
return {
query(sql: string): SyncDbStatement {
let statement = statements.get(sql);
if (!statement) {
statement = db.prepare(sql);
statements.set(sql, statement);
}
return statement;
},
exec(sql: string): void {
db.exec(sql);
},
close(): void {
statements.clear();
db.close();
},
};
}
@@ -0,0 +1,459 @@
import { selectAll, selectOne, type SqlRow, type SyncDb } from './libsql-driver';
import { insertRow, tableExists, type SyncMergeSummary } from './shared';
const ANIME_COPY_COLUMNS = [
'normalized_title_key',
'canonical_title',
'anilist_id',
'title_romaji',
'title_english',
'title_native',
'episodes_total',
'description',
'metadata_json',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const VIDEO_COPY_COLUMNS = [
'video_key',
'canonical_title',
'source_type',
'source_path',
'source_url',
'parsed_basename',
'parsed_title',
'parsed_season',
'parsed_episode',
'parser_source',
'parser_confidence',
'parse_metadata_json',
'watched',
'duration_ms',
'file_size_bytes',
'codec_id',
'container_id',
'width_px',
'height_px',
'fps_x100',
'bitrate_kbps',
'audio_codec_id',
'hash_sha256',
'screenshot_path',
'metadata_json',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const MEDIA_ART_COPY_COLUMNS = [
'anilist_id',
'cover_url',
'cover_blob',
'cover_blob_hash',
'title_romaji',
'title_english',
'episodes_total',
'fetched_at_ms',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const YOUTUBE_COPY_COLUMNS = [
'youtube_video_id',
'video_url',
'video_title',
'video_thumbnail_url',
'channel_id',
'channel_name',
'channel_url',
'channel_thumbnail_url',
'uploader_id',
'uploader_url',
'description',
'metadata_json',
'fetched_at_ms',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const WORD_COPY_COLUMNS = [
'headword',
'word',
'reading',
'part_of_speech',
'pos1',
'pos2',
'pos3',
'first_seen',
'last_seen',
'frequency',
'frequency_rank',
] as const;
export function mergeAnime(
local: SyncDb,
remote: SyncDb,
summary: SyncMergeSummary,
): Map<number, number> {
const map = new Map<number, number>();
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, ?),
title_english = COALESCE(title_english, ?),
title_native = COALESCE(title_native, ?),
episodes_total = COALESCE(episodes_total, ?),
description = COALESCE(description, ?)
WHERE anime_id = ?`,
);
for (const row of selectAll(
remote,
`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)) as SqlRow | undefined;
if (existing) {
const localId = Number(existing.anime_id);
map.set(remoteId, localId);
fillMissing.run(
row.title_romaji,
row.title_english,
row.title_native,
row.episodes_total,
row.description,
localId,
);
continue;
}
// No local row matched by anilist_id (checked first in `existing` above)
// or title key, so the remote anilist_id — if any — is free to insert as-is.
const values = ANIME_COPY_COLUMNS.map((column) => row[column]);
map.set(remoteId, insertRow(local, 'imm_anime', ANIME_COPY_COLUMNS, values));
summary.animeAdded += 1;
}
return map;
}
export interface VideoMergeResult {
videoIdMap: Map<number, number>;
addedVideoIds: Set<number>;
}
export function mergeVideos(
local: SyncDb,
remote: SyncDb,
animeIdMap: Map<number, number>,
summary: SyncMergeSummary,
): VideoMergeResult {
const videoIdMap = new Map<number, number>();
const addedVideoIds = new Set<number>();
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,
`SELECT video_id, anime_id, ${VIDEO_COPY_COLUMNS.join(', ')} FROM imm_videos`,
)) {
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) as SqlRow | undefined;
if (existing) {
const localId = Number(existing.video_id);
videoIdMap.set(remoteId, localId);
if (Number(row.watched) > 0 && Number(existing.watched) <= 0) {
setWatched.run(localId);
}
continue;
}
const columns = ['anime_id', ...VIDEO_COPY_COLUMNS];
const values = [mappedAnimeId, ...VIDEO_COPY_COLUMNS.map((column) => row[column])];
const localId = insertRow(local, 'imm_videos', columns, values);
videoIdMap.set(remoteId, localId);
addedVideoIds.add(remoteId);
summary.videosAdded += 1;
}
return { videoIdMap, addedVideoIds };
}
export function mergeMediaMetadata(
local: SyncDb,
remote: SyncDb,
videoIdMap: Map<number, number>,
addedVideoIds: Set<number>,
): void {
if (videoIdMap.size === 0) return;
const metadataVideoIds = new Set<number>([...addedVideoIds, ...videoIdMap.keys()]);
const hasBlobStore =
tableExists(local, 'imm_cover_art_blobs') && tableExists(remote, 'imm_cover_art_blobs');
const copyBlob = hasBlobStore
? 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.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.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 = 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) as SqlRow | undefined;
if (blob) {
copyBlob.run(blob.blob_hash, blob.cover_blob, blob.CREATED_DATE, blob.LAST_UPDATE_DATE);
}
}
insertRow(
local,
'imm_media_art',
['video_id', ...MEDIA_ART_COPY_COLUMNS],
[localVideoId, ...MEDIA_ART_COPY_COLUMNS.map((column) => row[column])],
);
}
}
if (tableExists(remote, 'imm_youtube_videos') && tableExists(local, 'imm_youtube_videos')) {
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 = selectOne(
remote,
`SELECT ${YOUTUBE_COPY_COLUMNS.join(', ')} FROM imm_youtube_videos WHERE video_id = ?`,
[remoteVideoId],
);
if (!row) continue;
insertRow(
local,
'imm_youtube_videos',
['video_id', ...YOUTUBE_COPY_COLUMNS],
[localVideoId, ...YOUTUBE_COPY_COLUMNS.map((column) => row[column])],
);
}
}
}
export function mergeExcludedWords(
local: SyncDb,
remote: SyncDb,
summary: SyncMergeSummary,
): void {
if (
!tableExists(remote, 'imm_stats_excluded_words') ||
!tableExists(local, 'imm_stats_excluded_words')
) {
return;
}
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`,
);
for (const row of selectAll(
remote,
'SELECT headword, word, reading, CREATED_DATE, LAST_UPDATE_DATE FROM imm_stats_excluded_words',
)) {
const result = insert.run(
row.headword,
row.word,
row.reading,
row.CREATED_DATE,
row.LAST_UPDATE_DATE,
);
summary.excludedWordsAdded += result.changes;
}
}
/**
* Lazily maps remote imm_words / imm_kanji ids onto local rows by natural key
* ((headword, word, reading) / kanji). New rows are copied with the remote's
* accumulated frequency minus any counts owed to skipped ACTIVE sessions
* (those merge later and re-add their counts); rows that already exist locally
* get their frequency incremented later with only the occurrence counts this
* merge adds (the remote total would double-count lines merged in earlier
* syncs).
*/
export class LexiconResolver {
private readonly wordMap = new Map<number, { localId: number; isNew: boolean }>();
private readonly kanjiMap = new Map<number, { localId: number; isNew: boolean }>();
readonly wordFrequencyDeltas = new Map<number, number>();
readonly kanjiFrequencyDeltas = new Map<number, number>();
constructor(
private readonly local: SyncDb,
private readonly remote: SyncDb,
private readonly summary: SyncMergeSummary,
) {}
/**
* Occurrence counts the remote's live tracker already baked into `frequency`
* but that belong to skipped ACTIVE sessions. Those lines are not copied this
* merge; they are re-added via addWordOccurrences/addKanjiOccurrences when
* the session finalizes and syncs, so a newly adopted row must not carry them
* or that slice would be counted twice.
*/
private pendingActiveSessionOccurrences(
occurrenceTable: 'imm_word_line_occurrences' | 'imm_kanji_line_occurrences',
idColumn: 'word_id' | 'kanji_id',
remoteId: number,
): number {
const row = selectOne(
this.remote,
`SELECT COALESCE(SUM(o.occurrence_count), 0) AS pending
FROM ${occurrenceTable} o
JOIN imm_subtitle_lines l ON l.line_id = o.line_id
JOIN imm_sessions s ON s.session_id = l.session_id
WHERE o.${idColumn} = ? AND s.ended_at_ms IS NULL`,
[remoteId],
);
return Number(row?.pending ?? 0);
}
private adoptedFrequency(frequency: unknown, pending: number): unknown {
if (pending <= 0 || typeof frequency !== 'number') return frequency;
return Math.max(0, frequency - pending);
}
resolveWord(remoteWordId: number): number {
const cached = this.wordMap.get(remoteWordId);
if (cached) return cached.localId;
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 = 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
.query(
`UPDATE imm_words
SET first_seen = MIN(COALESCE(first_seen, ?), COALESCE(?, first_seen)),
last_seen = MAX(COALESCE(last_seen, ?), COALESCE(?, last_seen))
WHERE id = ?`,
)
.run(row.first_seen, row.first_seen, row.last_seen, row.last_seen, entry.localId);
} else {
const pending = this.pendingActiveSessionOccurrences(
'imm_word_line_occurrences',
'word_id',
remoteWordId,
);
const localId = insertRow(
this.local,
'imm_words',
WORD_COPY_COLUMNS,
WORD_COPY_COLUMNS.map((column) =>
column === 'frequency' ? this.adoptedFrequency(row[column], pending) : row[column],
),
);
entry = { localId, isNew: true };
this.summary.wordsAdded += 1;
}
this.wordMap.set(remoteWordId, entry);
return entry.localId;
}
resolveKanji(remoteKanjiId: number): number {
const cached = this.kanjiMap.get(remoteKanjiId);
if (cached) return cached.localId;
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 = 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
.query(
`UPDATE imm_kanji
SET first_seen = MIN(COALESCE(first_seen, ?), COALESCE(?, first_seen)),
last_seen = MAX(COALESCE(last_seen, ?), COALESCE(?, last_seen))
WHERE id = ?`,
)
.run(row.first_seen, row.first_seen, row.last_seen, row.last_seen, entry.localId);
} else {
const pending = this.pendingActiveSessionOccurrences(
'imm_kanji_line_occurrences',
'kanji_id',
remoteKanjiId,
);
const localId = insertRow(
this.local,
'imm_kanji',
['kanji', 'first_seen', 'last_seen', 'frequency'],
[row.kanji, row.first_seen, row.last_seen, this.adoptedFrequency(row.frequency, pending)],
);
entry = { localId, isNew: true };
this.summary.kanjiAdded += 1;
}
this.kanjiMap.set(remoteKanjiId, entry);
return entry.localId;
}
addWordOccurrences(remoteWordId: number, count: number): void {
const entry = this.wordMap.get(remoteWordId);
if (!entry || entry.isNew) return;
this.wordFrequencyDeltas.set(
entry.localId,
(this.wordFrequencyDeltas.get(entry.localId) ?? 0) + count,
);
}
addKanjiOccurrences(remoteKanjiId: number, count: number): void {
const entry = this.kanjiMap.get(remoteKanjiId);
if (!entry || entry.isNew) return;
this.kanjiFrequencyDeltas.set(
entry.localId,
(this.kanjiFrequencyDeltas.get(entry.localId) ?? 0) + count,
);
}
applyFrequencyDeltas(): void {
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.query(
'UPDATE imm_kanji SET frequency = COALESCE(frequency, 0) + ? WHERE id = ?',
);
for (const [localId, delta] of this.kanjiFrequencyDeltas) {
updateKanji.run(delta, localId);
}
}
}
@@ -0,0 +1,264 @@
import { selectAll, type SqlRow, type SyncDb } from './libsql-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)`;
// Ported from upsertDailyRollupsForGroups / upsertMonthlyRollupsForGroups in
// src/core/services/immersion-tracker/maintenance.ts — must stay in sync.
const DAILY_ROLLUP_UPSERT = `
WITH matching_sessions AS (
SELECT * FROM imm_sessions
WHERE ${LOCAL_DAY_EXPR} = ? AND video_id = ?
),
session_metrics AS (
SELECT
t.session_id,
MAX(t.active_watched_ms) AS max_active_ms,
MAX(t.lines_seen) AS max_lines,
MAX(t.tokens_seen) AS max_tokens,
MAX(t.cards_mined) AS max_cards,
MAX(t.lookup_count) AS max_lookups,
MAX(t.lookup_hits) AS max_hits
FROM imm_session_telemetry t
JOIN matching_sessions s ON s.session_id = t.session_id
GROUP BY t.session_id
)
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
)
SELECT
${LOCAL_DAY_EXPR.replace('started_at_ms', 's.started_at_ms')} AS rollup_day,
s.video_id AS video_id,
COUNT(DISTINCT s.session_id) AS total_sessions,
COALESCE(SUM(COALESCE(sm.max_active_ms, s.active_watched_ms)), 0) / 60000.0 AS total_active_min,
COALESCE(SUM(COALESCE(sm.max_lines, s.lines_seen)), 0) AS total_lines_seen,
COALESCE(SUM(COALESCE(sm.max_tokens, s.tokens_seen)), 0) AS total_tokens_seen,
COALESCE(SUM(COALESCE(sm.max_cards, s.cards_mined)), 0) AS total_cards,
CASE
WHEN COALESCE(SUM(COALESCE(sm.max_active_ms, s.active_watched_ms)), 0) > 0
THEN (COALESCE(SUM(COALESCE(sm.max_cards, s.cards_mined)), 0) * 60.0)
/ (COALESCE(SUM(COALESCE(sm.max_active_ms, s.active_watched_ms)), 0) / 60000.0)
ELSE NULL
END AS cards_per_hour,
CASE
WHEN COALESCE(SUM(COALESCE(sm.max_active_ms, s.active_watched_ms)), 0) > 0
THEN COALESCE(SUM(COALESCE(sm.max_tokens, s.tokens_seen)), 0)
/ (COALESCE(SUM(COALESCE(sm.max_active_ms, s.active_watched_ms)), 0) / 60000.0)
ELSE NULL
END AS tokens_per_min,
CASE
WHEN COALESCE(SUM(COALESCE(sm.max_lookups, s.lookup_count)), 0) > 0
THEN CAST(COALESCE(SUM(COALESCE(sm.max_hits, s.lookup_hits)), 0) AS REAL)
/ CAST(COALESCE(SUM(COALESCE(sm.max_lookups, s.lookup_count)), 0) AS REAL)
ELSE NULL
END AS lookup_hit_rate,
? AS CREATED_DATE,
? AS LAST_UPDATE_DATE
FROM matching_sessions s
LEFT JOIN session_metrics sm ON s.session_id = sm.session_id
GROUP BY rollup_day, s.video_id
ON CONFLICT (rollup_day, video_id) DO UPDATE SET
total_sessions = excluded.total_sessions,
total_active_min = excluded.total_active_min,
total_lines_seen = excluded.total_lines_seen,
total_tokens_seen = excluded.total_tokens_seen,
total_cards = excluded.total_cards,
cards_per_hour = excluded.cards_per_hour,
tokens_per_min = excluded.tokens_per_min,
lookup_hit_rate = excluded.lookup_hit_rate,
CREATED_DATE = COALESCE(imm_daily_rollups.CREATED_DATE, excluded.CREATED_DATE),
LAST_UPDATE_DATE = excluded.LAST_UPDATE_DATE
`;
const MONTHLY_ROLLUP_UPSERT = `
WITH matching_sessions AS (
SELECT * FROM imm_sessions
WHERE ${LOCAL_MONTH_EXPR} = ? AND video_id = ?
),
session_metrics AS (
SELECT
t.session_id,
MAX(t.active_watched_ms) AS max_active_ms,
MAX(t.lines_seen) AS max_lines,
MAX(t.tokens_seen) AS max_tokens,
MAX(t.cards_mined) AS max_cards
FROM imm_session_telemetry t
JOIN matching_sessions s ON s.session_id = t.session_id
GROUP BY t.session_id
)
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
)
SELECT
${LOCAL_MONTH_EXPR.replace('started_at_ms', 's.started_at_ms')} AS rollup_month,
s.video_id AS video_id,
COUNT(DISTINCT s.session_id) AS total_sessions,
COALESCE(SUM(COALESCE(sm.max_active_ms, s.active_watched_ms)), 0) / 60000.0 AS total_active_min,
COALESCE(SUM(COALESCE(sm.max_lines, s.lines_seen)), 0) AS total_lines_seen,
COALESCE(SUM(COALESCE(sm.max_tokens, s.tokens_seen)), 0) AS total_tokens_seen,
COALESCE(SUM(COALESCE(sm.max_cards, s.cards_mined)), 0) AS total_cards,
? AS CREATED_DATE,
? AS LAST_UPDATE_DATE
FROM matching_sessions s
LEFT JOIN session_metrics sm ON s.session_id = sm.session_id
GROUP BY rollup_month, s.video_id
ON CONFLICT (rollup_month, video_id) DO UPDATE SET
total_sessions = excluded.total_sessions,
total_active_min = excluded.total_active_min,
total_lines_seen = excluded.total_lines_seen,
total_tokens_seen = excluded.total_tokens_seen,
total_cards = excluded.total_cards,
CREATED_DATE = COALESCE(imm_monthly_rollups.CREATED_DATE, excluded.CREATED_DATE),
LAST_UPDATE_DATE = excluded.LAST_UPDATE_DATE
`;
/**
* Recompute daily/monthly rollup groups touched by the newly merged sessions
* from the (now merged) local session + telemetry data. The maintenance
* watermark is left alone: telemetry newer than it gets recomputed again by
* the app later, which is idempotent.
*/
export function refreshRollupsForNewSessions(
local: SyncDb,
newSessionIds: number[],
summary: SyncMergeSummary,
): void {
if (newSessionIds.length === 0) return;
const groups = new Map<string, { day: number; month: number; videoId: number }>();
for (let offset = 0; offset < newSessionIds.length; offset += 500) {
const chunk = newSessionIds.slice(offset, offset + 500);
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);
const videoId = Number(row.video_id);
groups.set(`${day}-${videoId}`, { day, month, videoId });
}
}
const stampMs = nowDbTimestamp();
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.query(DAILY_ROLLUP_UPSERT);
const upsertMonthly = local.query(MONTHLY_ROLLUP_UPSERT);
const monthlyGroups = new Set<string>();
for (const { day, month, videoId } of groups.values()) {
deleteDaily.run(day, videoId);
upsertDaily.run(day, videoId, stampMs, stampMs);
summary.rollupGroupsRecomputed += 1;
const monthKey = `${month}-${videoId}`;
if (!monthlyGroups.has(monthKey)) {
monthlyGroups.add(monthKey);
deleteMonthly.run(month, videoId);
upsertMonthly.run(month, videoId, stampMs, stampMs);
}
}
}
/**
* Sessions are pruned after a retention window, but rollups are kept much
* longer — the remote's older rollup history can't be reconstructed from
* merged sessions. Copy remote rollup rows for groups where the local DB has
* neither a rollup row nor any sessions (i.e. history only the remote knows).
* Groups both machines have data for are never summed, to avoid
* double-counting sessions that earlier syncs already shared.
*/
export function copyRemoteOnlyRollups(
local: SyncDb,
remote: SyncDb,
videoIdMap: Map<number, number>,
summary: SyncMergeSummary,
): void {
if (!tableExists(remote, 'imm_daily_rollups') || !tableExists(local, 'imm_daily_rollups')) return;
const localDailyExists = local.query(
'SELECT 1 FROM imm_daily_rollups WHERE rollup_day = ? AND video_id = ? LIMIT 1',
);
const localDaySessions = local.query(
`SELECT 1 FROM imm_sessions WHERE video_id = ? AND ${LOCAL_DAY_EXPR} = ? LIMIT 1`,
);
const localMonthSessions = local.query(
`SELECT 1 FROM imm_sessions WHERE video_id = ? AND ${LOCAL_MONTH_EXPR} = ? LIMIT 1`,
);
// rollup_day is a *local* epoch day, so anchor it at local noon (+43200)
// before reading its month back: plain UTC midnight lands in the previous
// civil month for the 1st of a month at any negative UTC offset.
const localMonthSessionsForDay = local.query(
`SELECT 1 FROM imm_sessions
WHERE video_id = ?
AND ${LOCAL_MONTH_EXPR} = CAST(strftime('%Y%m', CAST(? AS INTEGER) * 86400 + 43200, 'unixepoch', 'localtime') AS INTEGER)
LIMIT 1`,
);
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 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;
if (localDailyExists.get(row.rollup_day, localVideoId)) continue;
if (localDaySessions.get(localVideoId, row.rollup_day)) continue;
if (localMonthSessionsForDay.get(localVideoId, row.rollup_day)) continue;
insertDaily.run(
row.rollup_day,
localVideoId,
row.total_sessions,
row.total_active_min,
row.total_lines_seen,
row.total_tokens_seen,
row.total_cards,
row.cards_per_hour,
row.tokens_per_min,
row.lookup_hit_rate,
row.CREATED_DATE,
row.LAST_UPDATE_DATE,
);
summary.dailyRollupsCopied += 1;
}
const localMonthlyExists = local.query(
'SELECT 1 FROM imm_monthly_rollups WHERE rollup_month = ? AND video_id = ? LIMIT 1',
);
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 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;
if (localMonthlyExists.get(row.rollup_month, localVideoId)) continue;
if (localMonthSessions.get(localVideoId, row.rollup_month)) continue;
insertMonthly.run(
row.rollup_month,
localVideoId,
row.total_sessions,
row.total_active_min,
row.total_lines_seen,
row.total_tokens_seen,
row.total_cards,
row.CREATED_DATE,
row.LAST_UPDATE_DATE,
);
summary.monthlyRollupsCopied += 1;
}
}
@@ -0,0 +1,470 @@
import type { LexiconResolver } from './merge-catalog';
import { selectAll, selectOne, type SqlRow, type SyncDb } from './libsql-driver';
import { insertRow, nowDbTimestamp, type SyncMergeSummary } from './shared';
const SESSION_COPY_COLUMNS = [
'session_uuid',
'started_at_ms',
'ended_at_ms',
'status',
'locale_id',
'target_lang_id',
'difficulty_tier',
'subtitle_mode',
'ended_media_ms',
'total_watched_ms',
'active_watched_ms',
'lines_seen',
'tokens_seen',
'cards_mined',
'lookup_count',
'lookup_hits',
'yomitan_lookup_count',
'pause_count',
'pause_ms',
'seek_forward_count',
'seek_backward_count',
'media_buffer_events',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const TELEMETRY_COPY_COLUMNS = [
'sample_ms',
'total_watched_ms',
'active_watched_ms',
'lines_seen',
'tokens_seen',
'cards_mined',
'lookup_count',
'lookup_hits',
'yomitan_lookup_count',
'pause_count',
'pause_ms',
'seek_forward_count',
'seek_backward_count',
'media_buffer_events',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const EVENT_COPY_COLUMNS = [
'ts_ms',
'event_type',
'line_index',
'segment_start_ms',
'segment_end_ms',
'tokens_delta',
'cards_delta',
'payload_json',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
const LINE_COPY_COLUMNS = [
'line_index',
'segment_start_ms',
'segment_end_ms',
'text',
'secondary_text',
'CREATED_DATE',
'LAST_UPDATE_DATE',
] as const;
export interface SessionMergeResult {
newSessionIds: number[];
}
export function mergeSessions(
local: SyncDb,
remote: SyncDb,
videoIdMap: Map<number, number>,
animeIdMap: Map<number, number>,
lexicon: LexiconResolver,
summary: SyncMergeSummary,
): SessionMergeResult {
const newSessionIds: number[] = [];
const uuidExists = local.query('SELECT session_id FROM imm_sessions WHERE session_uuid = ?');
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) {
// Stale ACTIVE sessions are finalized by the app on its next startup;
// they will sync once they carry final numbers.
summary.activeSessionsSkipped += 1;
continue;
}
if (uuidExists.get(session.session_uuid)) {
summary.sessionsAlreadyPresent += 1;
continue;
}
const localVideoId = videoIdMap.get(Number(session.video_id));
if (localVideoId === undefined) {
throw new Error(
`Snapshot session ${String(session.session_uuid)} references missing video row`,
);
}
const localSessionId = insertRow(
local,
'imm_sessions',
['video_id', ...SESSION_COPY_COLUMNS],
[localVideoId, ...SESSION_COPY_COLUMNS.map((column) => session[column])],
);
newSessionIds.push(localSessionId);
summary.sessionsMerged += 1;
const remoteSessionId = Number(session.session_id);
copyTelemetry(local, remote, remoteSessionId, localSessionId, summary);
const eventIdMap = copyEvents(local, remote, remoteSessionId, localSessionId, summary);
copySubtitleLines(
local,
remote,
remoteSessionId,
localSessionId,
localVideoId,
animeIdMap,
eventIdMap,
lexicon,
summary,
);
applyMergedSessionLifetime(local, localSessionId, localVideoId, session);
}
return { newSessionIds };
}
function copyTelemetry(
local: SyncDb,
remote: SyncDb,
remoteSessionId: number,
localSessionId: number,
summary: SyncMergeSummary,
): void {
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,
'imm_session_telemetry',
['session_id', ...TELEMETRY_COPY_COLUMNS],
[localSessionId, ...TELEMETRY_COPY_COLUMNS.map((column) => row[column])],
);
summary.telemetryRowsAdded += 1;
}
}
function copyEvents(
local: SyncDb,
remote: SyncDb,
remoteSessionId: number,
localSessionId: number,
summary: SyncMergeSummary,
): Map<number, number> {
const eventIdMap = new Map<number, number>();
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,
'imm_session_events',
['session_id', ...EVENT_COPY_COLUMNS],
[localSessionId, ...EVENT_COPY_COLUMNS.map((column) => row[column])],
);
eventIdMap.set(Number(row.event_id), localEventId);
summary.eventsAdded += 1;
}
return eventIdMap;
}
function copySubtitleLines(
local: SyncDb,
remote: SyncDb,
remoteSessionId: number,
localSessionId: number,
localVideoId: number,
animeIdMap: Map<number, number>,
eventIdMap: Map<number, number>,
lexicon: LexiconResolver,
summary: SyncMergeSummary,
): void {
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.query(
'SELECT kanji_id, occurrence_count FROM imm_kanji_line_occurrences WHERE line_id = ?',
);
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.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 localLineId = insertRow(
local,
'imm_subtitle_lines',
['session_id', 'event_id', 'video_id', 'anime_id', ...LINE_COPY_COLUMNS],
[
localSessionId,
localEventId,
localVideoId,
localAnimeId,
...LINE_COPY_COLUMNS.map((column) => row[column]),
],
);
summary.subtitleLinesAdded += 1;
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) as SqlRow[]) {
const localKanjiId = lexicon.resolveKanji(Number(occurrence.kanji_id));
const count = Number(occurrence.occurrence_count);
insertKanjiOccurrence.run(localLineId, localKanjiId, count);
lexicon.addKanjiOccurrences(Number(occurrence.kanji_id), count);
}
}
}
/**
* Port of applySessionLifetimeSummary (src/core/services/immersion-tracker/
* lifetime.ts) for sessions arriving out of chronological order. The
* "first session of the day / for this video" checks are order-independent
* here (any other session counts, not just earlier ones): the local machine
* already credited active_days/episodes_started when its own session was
* applied, even if the merged session started earlier that day.
*/
function applyMergedSessionLifetime(
local: SyncDb,
sessionId: number,
videoId: number,
session: SqlRow,
): void {
const updatedAtMs = nowDbTimestamp();
const applied = local
.query(
`INSERT INTO imm_lifetime_applied_sessions (session_id, applied_at_ms, CREATED_DATE, LAST_UPDATE_DATE)
VALUES (?, ?, ?, ?)
ON CONFLICT(session_id) DO NOTHING`,
)
.run(sessionId, session.ended_at_ms, updatedAtMs, updatedAtMs);
if (applied.changes <= 0) return;
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;
const value = Number.isFinite(fromTelemetry) ? fromTelemetry : Number(sessionValue);
return Math.max(0, Math.floor(Number.isFinite(value) ? value : 0));
};
const activeMs = metric(telemetry?.active_watched_ms, session.active_watched_ms);
const cardsMined = metric(telemetry?.cards_mined, session.cards_mined);
const linesSeen = metric(telemetry?.lines_seen, session.lines_seen);
const tokensSeen = metric(telemetry?.tokens_seen, session.tokens_seen);
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 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')
.get(videoId, sessionId),
);
const isFirstSessionForVideoRun = !mediaLifetime && !hasOtherSessionForVideo;
const isFirstCompletedSessionForVideoRun = watched > 0 && Number(mediaLifetime?.completed ?? 0) <= 0;
const hasOtherSessionOnDay = Boolean(
local
.query(
`SELECT 1 FROM imm_sessions
WHERE session_id != ?
AND CAST(julianday(CAST(started_at_ms AS REAL) / 1000, 'unixepoch', 'localtime') - 2440587.5 AS INTEGER)
= CAST(julianday(CAST(? AS REAL) / 1000, 'unixepoch', 'localtime') - 2440587.5 AS INTEGER)
LIMIT 1`,
)
.get(sessionId, session.started_at_ms),
);
let animeCompletedDelta = 0;
if (animeId !== null && watched > 0 && isFirstCompletedSessionForVideoRun) {
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);
if (
episodesTotal !== null &&
episodesTotal > 0 &&
episodesCompletedBefore < episodesTotal &&
episodesCompletedBefore + 1 >= episodesTotal
) {
animeCompletedDelta = 1;
}
}
local
.query(
`UPDATE imm_lifetime_global
SET total_sessions = total_sessions + 1,
total_active_ms = total_active_ms + ?,
total_cards = total_cards + ?,
active_days = active_days + ?,
episodes_started = episodes_started + ?,
episodes_completed = episodes_completed + ?,
anime_completed = anime_completed + ?,
LAST_UPDATE_DATE = ?
WHERE global_id = 1`,
)
.run(
activeMs,
cardsMined,
hasOtherSessionOnDay ? 0 : 1,
isFirstSessionForVideoRun ? 1 : 0,
isFirstCompletedSessionForVideoRun ? 1 : 0,
animeCompletedDelta,
updatedAtMs,
);
local
.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
)
VALUES (?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(video_id) DO UPDATE SET
total_sessions = total_sessions + 1,
total_active_ms = total_active_ms + excluded.total_active_ms,
total_cards = total_cards + excluded.total_cards,
total_lines_seen = total_lines_seen + excluded.total_lines_seen,
total_tokens_seen = total_tokens_seen + excluded.total_tokens_seen,
completed = MAX(completed, excluded.completed),
first_watched_ms = CASE
WHEN excluded.first_watched_ms IS NULL THEN first_watched_ms
WHEN first_watched_ms IS NULL THEN excluded.first_watched_ms
WHEN excluded.first_watched_ms < first_watched_ms THEN excluded.first_watched_ms
ELSE first_watched_ms
END,
last_watched_ms = CASE
WHEN excluded.last_watched_ms IS NULL THEN last_watched_ms
WHEN last_watched_ms IS NULL THEN excluded.last_watched_ms
WHEN excluded.last_watched_ms > last_watched_ms THEN excluded.last_watched_ms
ELSE last_watched_ms
END,
LAST_UPDATE_DATE = excluded.LAST_UPDATE_DATE`,
)
.run(
videoId,
activeMs,
cardsMined,
linesSeen,
tokensSeen,
watched > 0 ? 1 : 0,
session.started_at_ms,
session.ended_at_ms,
updatedAtMs,
updatedAtMs,
);
if (animeId !== null) {
local
.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,
last_watched_ms, CREATED_DATE, LAST_UPDATE_DATE
)
VALUES (?, 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(anime_id) DO UPDATE SET
total_sessions = total_sessions + 1,
total_active_ms = total_active_ms + excluded.total_active_ms,
total_cards = total_cards + excluded.total_cards,
total_lines_seen = total_lines_seen + excluded.total_lines_seen,
total_tokens_seen = total_tokens_seen + excluded.total_tokens_seen,
episodes_started = episodes_started + excluded.episodes_started,
episodes_completed = episodes_completed + excluded.episodes_completed,
first_watched_ms = CASE
WHEN excluded.first_watched_ms IS NULL THEN first_watched_ms
WHEN first_watched_ms IS NULL THEN excluded.first_watched_ms
WHEN excluded.first_watched_ms < first_watched_ms THEN excluded.first_watched_ms
ELSE first_watched_ms
END,
last_watched_ms = CASE
WHEN excluded.last_watched_ms IS NULL THEN last_watched_ms
WHEN last_watched_ms IS NULL THEN excluded.last_watched_ms
WHEN excluded.last_watched_ms > last_watched_ms THEN excluded.last_watched_ms
ELSE last_watched_ms
END,
LAST_UPDATE_DATE = excluded.LAST_UPDATE_DATE`,
)
.run(
animeId,
activeMs,
cardsMined,
linesSeen,
tokensSeen,
isFirstSessionForVideoRun ? 1 : 0,
isFirstCompletedSessionForVideoRun ? 1 : 0,
session.started_at_ms,
session.ended_at_ms,
updatedAtMs,
updatedAtMs,
);
}
}
+104
View File
@@ -0,0 +1,104 @@
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 { openLibsqlSyncDb, type SyncDb } from './libsql-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(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 = openLibsqlSyncDb(snapshotPath, { readonly: true });
let local: SyncDb;
try {
local = openLibsqlSyncDb(localDbPath, { 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');
}
+150
View File
@@ -0,0 +1,150 @@
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { SCHEMA_VERSION } from '../immersion-tracker/types';
import { getDefaultConfigDir } from '../../../shared/setup-state';
import { withReadonlyWalRetry } from './wal-retry';
import { openLibsqlSyncDb, selectOne, type SyncDb } from './libsql-driver';
export { SCHEMA_VERSION };
import type { SyncMergeSummary } from '../../../shared/sync/sync-events';
export type { SyncMergeSummary };
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),
);
}
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(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 = openLibsqlSyncDb(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<string>([path.join(path.dirname(dbPath), 'stats-daemon.json')]);
candidates.add(path.join(getDefaultConfigDir(), '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;
}
+191
View File
@@ -0,0 +1,191 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import {
assertSafeSshHost,
detectRemoteShellFlavor,
quoteForRemoteShell,
resolveRemoteSubminerCommand,
runScp,
shellQuote,
type RemoteRunResult,
} from './ssh';
function remoteResult(status: number, stdout = ''): RemoteRunResult {
return { status, stdout, stderr: '' };
}
test('assertSafeSshHost rejects option-like hosts', () => {
assert.throws(() => assertSafeSshHost('-oProxyCommand=touch pwned'), /looks like an option/);
assert.throws(() => assertSafeSshHost('-lroot'), /looks like an option/);
});
test('assertSafeSshHost accepts normal destinations', () => {
assert.doesNotThrow(() => assertSafeSshHost('macbook'));
assert.doesNotThrow(() => assertSafeSshHost('user@192.168.1.20'));
assert.doesNotThrow(() => assertSafeSshHost('ssh-alias'));
});
test('shellQuote escapes single quotes and wraps in quotes', () => {
assert.equal(shellQuote('subminer'), `'subminer'`);
assert.equal(shellQuote(`a'; rm -rf ~; '`), `'a'\\''; rm -rf ~; '\\'''`);
});
test('runScp rejects option-like local endpoints before spawning scp', () => {
assert.throws(() => runScp('-oProxyCommand=sh', '/tmp/out.sqlite'), /looks like an option/);
assert.throws(() => runScp('/tmp/in.sqlite', '-bad-destination'), /looks like an option/);
});
test('runScp rejects option-like remote host components', () => {
assert.throws(
() => runScp('-oProxyCommand=sh:/tmp/in.sqlite', '/tmp/out.sqlite'),
/SSH host that looks like an option/,
);
});
test('resolveRemoteSubminerCommand verifies the launcher under the remote runtime PATH', () => {
const calls: Array<{ host: string; remoteCommand: string }> = [];
const command = resolveRemoteSubminerCommand('macbook', null, 'posix', (host, remoteCommand) => {
calls.push({ host, remoteCommand });
return remoteResult(0);
});
assert.equal(
command,
'PATH="$HOME/.local/bin:$HOME/.bun/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:$PATH" subminer',
);
assert.deepEqual(calls, [
{
host: 'macbook',
remoteCommand:
'PATH="$HOME/.local/bin:$HOME/.bun/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:$PATH" subminer --help >/dev/null 2>&1',
},
]);
});
test('resolveRemoteSubminerCommand falls back to the app binary in --sync-cli mode', () => {
const probed: string[] = [];
const command = resolveRemoteSubminerCommand(
'media-box',
null,
'posix',
(_host, remoteCommand) => {
probed.push(remoteCommand);
return remoteResult(remoteCommand.includes('SubMiner --sync-cli') ? 0 : 1);
},
);
assert.match(command, / SubMiner --sync-cli$/);
// Launcher candidates (PATH + ~/.local/bin) are tried before app binaries.
assert.equal(probed.length, 3);
assert.match(probed[0]!, / subminer --help /);
assert.match(probed[1]!, / ~\/\.local\/bin\/subminer --help /);
});
test('resolveRemoteSubminerCommand probes a user override as app first, then launcher', () => {
const asApp = resolveRemoteSubminerCommand(
'media-box',
'/opt/SubMiner.AppImage',
'posix',
(_host, cmd) => remoteResult(cmd.includes('--sync-cli') ? 0 : 1),
);
assert.match(asApp, /'\/opt\/SubMiner\.AppImage' --sync-cli$/);
const asLauncher = resolveRemoteSubminerCommand(
'media-box',
'/opt/subminer',
'posix',
(_host, cmd) => remoteResult(cmd.includes('--sync-cli') ? 1 : 0),
);
assert.match(asLauncher, /'\/opt\/subminer'$/);
assert.throws(
() => resolveRemoteSubminerCommand('media-box', '/missing', 'posix', () => remoteResult(127)),
/Remote command not found on media-box: \/missing/,
);
});
test('resolveRemoteSubminerCommand probes Windows install locations without a PATH prefix', () => {
const probed: string[] = [];
const command = resolveRemoteSubminerCommand('win-box', null, 'windows-cmd', (_host, cmd) => {
probed.push(cmd);
return remoteResult(cmd.includes('Programs\\SubMiner\\SubMiner.exe') ? 0 : 1);
});
assert.equal(command, '"%LOCALAPPDATA%\\Programs\\SubMiner\\SubMiner.exe" --sync-cli');
assert.equal(probed[0], 'subminer --help');
assert.equal(probed[1], '"%LOCALAPPDATA%\\SubMiner\\bin\\subminer.cmd" --help');
const powershell = resolveRemoteSubminerCommand(
'win-box',
null,
'windows-powershell',
(_host, cmd) => remoteResult(cmd.includes('Programs\\SubMiner\\SubMiner.exe') ? 0 : 1),
);
assert.equal(powershell, '& "$env:LOCALAPPDATA\\Programs\\SubMiner\\SubMiner.exe" --sync-cli');
});
test('resolveRemoteSubminerCommand quotes Windows overrides with double quotes', () => {
const command = resolveRemoteSubminerCommand(
'win-box',
'C:/Apps/SubMiner/SubMiner.exe',
'windows-cmd',
(_host, cmd) => remoteResult(cmd.includes('--sync-cli') ? 0 : 1),
);
assert.equal(command, '"C:/Apps/SubMiner/SubMiner.exe" --sync-cli');
});
test('detectRemoteShellFlavor identifies posix, cmd, and powershell remotes', () => {
assert.equal(
detectRemoteShellFlavor('linux-box', (_host, cmd) =>
cmd === 'uname -s' ? remoteResult(0, 'Linux\n') : remoteResult(1),
),
'posix',
);
assert.equal(
detectRemoteShellFlavor('win-box', (_host, cmd) => {
if (cmd === 'uname -s') return remoteResult(1);
if (cmd === 'echo %OS%') return remoteResult(0, 'Windows_NT\r\n');
return remoteResult(1);
}),
'windows-cmd',
);
assert.equal(
detectRemoteShellFlavor('ps-box', (_host, cmd) => {
if (cmd === 'uname -s') return remoteResult(1);
if (cmd === 'echo %OS%') return remoteResult(0, '%OS%\r\n');
if (cmd === 'echo $env:OS') return remoteResult(0, 'Windows_NT\r\n');
return remoteResult(1);
}),
'windows-powershell',
);
// Unidentifiable remotes keep the pre-detection POSIX behavior.
assert.equal(
detectRemoteShellFlavor('odd-box', () => remoteResult(1)),
'posix',
);
});
test('quoteForRemoteShell quotes per flavor and rejects unsafe Windows values', () => {
assert.equal(quoteForRemoteShell('posix', "/tmp/it's"), `'/tmp/it'\\''s'`);
assert.equal(
quoteForRemoteShell('windows-cmd', 'C:/Users/First Last/AppData/Local/Temp/subminer-sync-ab'),
'"C:/Users/First Last/AppData/Local/Temp/subminer-sync-ab"',
);
assert.throws(() => quoteForRemoteShell('windows-cmd', 'a"b'), /Refusing to quote/);
assert.throws(() => quoteForRemoteShell('windows-cmd', 'C:/tmp/%TEMP%/db.sqlite'), /percent/);
assert.throws(() => quoteForRemoteShell('windows-powershell', 'a\nb'), /Refusing to quote/);
});
test('quoteForRemoteShell does not let PowerShell expand a quoted value', () => {
// PowerShell expands $(...) and $var inside double quotes, so a single-quoted
// literal is the only safe form; '' is the escape for an embedded quote.
assert.equal(
quoteForRemoteShell('windows-powershell', 'C:/tmp/$(calc.exe)'),
`'C:/tmp/$(calc.exe)'`,
);
assert.equal(quoteForRemoteShell('windows-powershell', "C:/tmp/it's"), `'C:/tmp/it''s'`);
assert.equal(
quoteForRemoteShell('windows-powershell', 'C:/Users/First Last/Temp/subminer-sync-ab'),
`'C:/Users/First Last/Temp/subminer-sync-ab'`,
);
});
+223
View File
@@ -0,0 +1,223 @@
import { spawnSync } from 'node:child_process';
import { SYNC_CLI_FLAG } from './cli-args';
export interface RemoteRunResult {
status: number;
stdout: string;
stderr: string;
}
export interface RunSshOptions {
batchMode?: boolean;
connectTimeoutSeconds?: number;
timeoutMs?: number;
}
/**
* 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,
options: RunSshOptions = {},
): RemoteRunResult {
assertSafeSshHost(host);
const args: string[] = [];
if (options.batchMode) args.push('-o', 'BatchMode=yes');
if (options.connectTimeoutSeconds !== undefined) {
args.push('-o', `ConnectTimeout=${options.connectTimeoutSeconds}`);
}
args.push(host, remoteCommand);
const result = spawnSync('ssh', args, {
encoding: 'utf8',
stdio: ['inherit', 'pipe', 'pipe'],
timeout: options.timeoutMs,
});
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 {
if (/^[A-Za-z]:[\\/]/.test(endpoint)) return;
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("'", `'\\''`)}'`;
}
/**
* The shell that Windows OpenSSH hands remote commands to (cmd.exe by
* default, PowerShell when DefaultShell is changed); it decides quoting,
* environment-variable expansion, and which SubMiner install paths to probe.
*/
export type RemoteShellFlavor = 'posix' | 'windows-cmd' | 'windows-powershell';
/**
* Identify the remote shell with probes that are harmless everywhere:
* `uname -s` only succeeds on a POSIX shell, `echo %OS%` only expands under
* cmd.exe, and `echo $env:OS` only expands under PowerShell. Defaults to
* posix so unreachable/odd hosts fail with the familiar POSIX errors.
*/
export function detectRemoteShellFlavor(
host: string,
runRemote: (host: string, remoteCommand: string) => RemoteRunResult,
): RemoteShellFlavor {
const posixProbe = runRemote(host, 'uname -s');
if (posixProbe.status === 0 && posixProbe.stdout.trim().length > 0) return 'posix';
const cmdProbe = runRemote(host, 'echo %OS%');
if (cmdProbe.status === 0 && cmdProbe.stdout.includes('Windows_NT')) return 'windows-cmd';
const powershellProbe = runRemote(host, 'echo $env:OS');
if (powershellProbe.status === 0 && powershellProbe.stdout.includes('Windows_NT')) {
return 'windows-powershell';
}
return 'posix';
}
/**
* Quote one argument for the detected remote shell. PowerShell expands $(...)
* and $var inside double quotes, so it gets a single-quoted literal ('' escapes
* a quote). cmd.exe has no single-quote form and treats ' literally, so it keeps
* double quotes and rejects values carrying a double quote of their own.
*/
export function quoteForRemoteShell(flavor: RemoteShellFlavor, value: string): string {
if (flavor === 'posix') return shellQuote(value);
if (/[\r\n]/.test(value)) {
throw new Error(`Refusing to quote a value with newlines for a Windows shell: ${value}`);
}
if (flavor === 'windows-powershell') {
return `'${value.replaceAll("'", "''")}'`;
}
if (value.includes('"')) {
throw new Error(`Refusing to quote a value with quotes for a Windows shell: ${value}`);
}
if (value.includes('%')) {
throw new Error(`Refusing to quote a value with percent signs for cmd.exe: ${value}`);
}
return `"${value}"`;
}
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 (SYNC_CLI_FLAG), so a remote machine only needs the
// app installed; the command-line launcher is one candidate, not a
// requirement. Each candidate is the remote invocation to probe, in order.
function defaultRemoteCandidates(flavor: RemoteShellFlavor): string[] {
if (flavor === 'windows-cmd' || flavor === 'windows-powershell') {
// cmd.exe expands %VAR% inside double quotes; PowerShell needs $env: and
// the & call operator to run a quoted path.
const launcherShim =
flavor === 'windows-cmd'
? `"%LOCALAPPDATA%\\SubMiner\\bin\\subminer.cmd"`
: `& "$env:LOCALAPPDATA\\SubMiner\\bin\\subminer.cmd"`;
const appInstall =
flavor === 'windows-cmd'
? `"%LOCALAPPDATA%\\Programs\\SubMiner\\SubMiner.exe"`
: `& "$env:LOCALAPPDATA\\Programs\\SubMiner\\SubMiner.exe"`;
return [
// Command-line launcher shim on PATH or in its default install dir.
'subminer',
launcherShim,
// The app binary itself in sync-CLI mode (default NSIS install dir).
`${appInstall} ${SYNC_CLI_FLAG}`,
`SubMiner ${SYNC_CLI_FLAG}`,
];
}
return [
// Command-line launcher (bun script) on PATH or in its default install dir.
'subminer',
'~/.local/bin/subminer',
// The app binary itself in sync-CLI mode.
`SubMiner ${SYNC_CLI_FLAG}`,
`/Applications/SubMiner.app/Contents/MacOS/SubMiner ${SYNC_CLI_FLAG}`,
`~/Applications/SubMiner.app/Contents/MacOS/SubMiner ${SYNC_CLI_FLAG}`,
];
}
function preferredCandidates(flavor: RemoteShellFlavor, preferred: string): string[] {
const quoted = quoteForRemoteShell(flavor, preferred);
const invocation = flavor === 'windows-powershell' ? `& ${quoted}` : quoted;
// App binaries also answer plain --help by opening the GUI-oriented help
// path, so probe the sync-CLI shape first.
return [`${invocation} ${SYNC_CLI_FLAG}`, invocation];
}
/**
* Non-interactive POSIX SSH shells often miss user-installed launchers and
* Bun, so those candidates are probed under the same deterministic PATH sync
* itself uses; Windows shells get their default install locations instead.
* Trusted defaults stay unquoted so the remote shell expands `~`/%VAR%; a
* user-supplied override is 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,
flavor: RemoteShellFlavor = 'posix',
runRemote: typeof runSsh = runSsh,
): string {
const candidates = preferred
? preferredCandidates(flavor, preferred)
: defaultRemoteCandidates(flavor);
for (const candidate of candidates) {
const command = flavor === 'posix' ? `${REMOTE_RUNTIME_PATH} ${candidate}` : candidate;
const probe = runRemote(
host,
flavor === 'posix' ? `${command} --help >/dev/null 2>&1` : `${command} --help`,
);
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 and the SubMiner app binary in their default install locations). Pass --remote-cmd <path> pointing at the SubMiner app or launcher.`,
);
}
@@ -0,0 +1,492 @@
import test from 'node:test';
import assert from 'node:assert/strict';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { createEmptyMergeSummary } from './shared';
import {
ensureTrackerQuiescentFlow,
runSyncFlow,
type SyncFlowContext,
type SyncFlowDeps,
} from './sync-flow';
function makeContext(overrides: Partial<SyncFlowContext['args']> = {}): SyncFlowContext {
return {
args: {
syncHost: '',
syncSnapshotPath: '',
syncMergePath: '',
syncDirection: 'both',
syncRemoteCmd: '',
syncDbPath: '',
syncForce: false,
syncJson: false,
syncCheck: false,
syncMakeTemp: false,
syncRemoveTempPath: '',
logLevel: 'warn',
...overrides,
},
mpvSocketPath: '',
};
}
function ok(stdout = ''): { status: number; stdout: string; stderr: string } {
return { status: 0, stdout, stderr: '' };
}
// recordHostSyncResult defaults to a no-op here: the real disk-writing
// implementations are bound by the entry points, never by the flow itself.
function makeDeps(overrides: Partial<SyncFlowDeps> = {}): SyncFlowDeps {
return {
createDbSnapshot: () => {},
mergeSnapshotIntoDb: () => createEmptyMergeSummary(),
findLiveStatsDaemonPid: () => null,
assertSafeSshHost: () => {},
detectRemoteShellFlavor: () => 'posix',
resolveRemoteSubminerCommand: () => 'subminer',
runScp: () => {},
runSsh: () => ok(),
canConnectUnixSocket: async () => false,
realpathSync: (candidate) => candidate,
mkdtempSync: (prefix) => fs.mkdtempSync(prefix),
rmSync: (target, options) => fs.rmSync(target, options),
consoleLog: () => {},
writeStdout: () => true,
ensureTrackerQuiescent: async () => {},
emitEvent: () => {},
recordHostSyncResult: () => {},
resolveDefaultDbPath: () => '/tracker.sqlite',
...overrides,
};
}
test('ensureTrackerQuiescentFlow ignores stale sockets but rejects live sockets', async () => {
const context = makeContext({ syncDbPath: '/tmp/local.sqlite' });
context.mpvSocketPath = '/tmp/subminer-socket';
let socketConnectable = false;
const deps = makeDeps({
realpathSync: () => '/tracker.sqlite',
canConnectUnixSocket: async () => socketConnectable,
});
await ensureTrackerQuiescentFlow(context, '/tmp/local.sqlite', deps);
socketConnectable = true;
await assert.rejects(
async () => ensureTrackerQuiescentFlow(context, '/tmp/local.sqlite', deps),
/mpv\/SubMiner session appears to be running/,
);
});
test('ensureTrackerQuiescentFlow rejects a live stats daemon and honors --force', async () => {
const context = makeContext({ syncDbPath: '/tmp/local.sqlite' });
const deps = makeDeps({
realpathSync: () => '/tracker.sqlite',
findLiveStatsDaemonPid: () => 4242,
});
await assert.rejects(
async () => ensureTrackerQuiescentFlow(context, '/tmp/local.sqlite', deps),
/stats server is running \(pid 4242\)/,
);
const forced = makeContext({ syncDbPath: '/tmp/local.sqlite', syncForce: true });
await ensureTrackerQuiescentFlow(forced, '/tmp/local.sqlite', deps);
});
test('runSyncFlow dispatches snapshot, merge, host, and missing-target modes', async () => {
const calls: string[] = [];
const deps = makeDeps({
createDbSnapshot: (dbPath, outPath) => {
calls.push(`snapshot:${dbPath}->${outPath}`);
},
mergeSnapshotIntoDb: (dbPath, snapshotPath) => {
calls.push(`merge:${dbPath}<-${snapshotPath}`);
return createEmptyMergeSummary();
},
ensureTrackerQuiescent: async () => {
calls.push('quiescent');
},
assertSafeSshHost: (host) => {
calls.push(`host:${host}`);
},
runSsh: (_host, command) => {
calls.push(`ssh:${command}`);
return command.includes(' sync --make-temp') ? ok('/tmp/subminer-sync-remote\n') : ok();
},
runScp: (from, to) => {
calls.push(`scp:${from}->${to}`);
},
});
await runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncSnapshotPath: '/tmp/out.sqlite' }),
deps,
);
assert.ok(calls.includes('snapshot:/tmp/local.sqlite->/tmp/out.sqlite'));
assert.ok(
calls.indexOf('quiescent') < calls.indexOf('snapshot:/tmp/local.sqlite->/tmp/out.sqlite'),
);
await runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncMergePath: '/tmp/in.sqlite' }),
deps,
);
assert.ok(calls.includes('quiescent'));
assert.ok(calls.includes('merge:/tmp/local.sqlite<-/tmp/in.sqlite'));
await runSyncFlow(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), deps);
assert.ok(calls.includes('host:media-box'));
await assert.rejects(
() => runSyncFlow(makeContext({ syncDbPath: '/tmp/local.sqlite' }), deps),
/sync requires a host, --snapshot <file>, or --merge <file>/,
);
});
function makeHostDeps(calls: string[], overrides: Partial<SyncFlowDeps> = {}): SyncFlowDeps {
return makeDeps({
createDbSnapshot: (_dbPath, outPath) => {
calls.push(`snapshot:${outPath}`);
fs.writeFileSync(outPath, 'snapshot');
},
mergeSnapshotIntoDb: () => {
calls.push('local-merge');
return createEmptyMergeSummary();
},
ensureTrackerQuiescent: async () => {
calls.push('quiescent');
},
runSsh: (_host, command) => {
calls.push(`ssh:${command}`);
if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
return ok();
},
runScp: (from, to) => {
calls.push(`scp:${from}->${to}`);
if (!to.includes(':')) fs.writeFileSync(to, 'pulled');
},
...overrides,
});
}
test('runHostSync keeps tracker quiescent through both merges and cleans up after failure', async () => {
const calls: string[] = [];
let localTmpDir = '';
const deps = makeHostDeps(calls, {
mkdtempSync: (prefix) => {
localTmpDir = fs.mkdtempSync(prefix);
return localTmpDir;
},
runSsh: (_host, command) => {
calls.push(`ssh:${command}`);
if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
if (command.includes(' sync --merge ')) {
return { status: 9, stdout: 'remote output', stderr: 'remote merge exploded' };
}
return ok();
},
});
await assert.rejects(
() =>
runSyncFlow(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), deps),
/Remote merge failed on media-box[\s\S]*remote merge exploded/,
);
assert.equal(calls.filter((call) => call === 'quiescent').length, 3);
assert.ok(calls.indexOf('quiescent') < calls.findIndex((call) => call.startsWith('snapshot:')));
assert.ok(calls.includes('local-merge'));
assert.ok(calls.some((call) => call.includes(' sync --remove-temp ')));
assert.equal(fs.existsSync(localTmpDir), false);
});
test('runHostSync includes remote snapshot stderr in failures', async () => {
const deps = makeHostDeps([], {
runSsh: (_host, command) => {
if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
if (command.includes(' sync --snapshot ')) {
return { status: 5, stdout: '', stderr: 'snapshot permission denied' };
}
return ok();
},
});
await assert.rejects(
() =>
runSyncFlow(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }), deps),
/Remote snapshot failed on media-box[\s\S]*snapshot permission denied/,
);
});
test('runHostSync push only snapshots locally and merges remotely', async () => {
const calls: string[] = [];
await runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box', syncDirection: 'push' }),
makeHostDeps(calls),
);
assert.ok(calls.some((call) => call.startsWith('snapshot:')));
assert.ok(calls.some((call) => call.includes(' sync --merge ')));
assert.ok(calls.some((call) => call.startsWith('scp:') && call.includes('->media-box:')));
assert.ok(!calls.some((call) => call.includes(' sync --snapshot ')));
assert.ok(!calls.includes('local-merge'));
});
test('runHostSync pull only snapshots remotely and merges locally', async () => {
const calls: string[] = [];
await runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box', syncDirection: 'pull' }),
makeHostDeps(calls),
);
assert.ok(calls.some((call) => call.includes(' sync --snapshot ')));
assert.ok(calls.some((call) => call.startsWith('scp:media-box:')));
assert.ok(calls.includes('local-merge'));
assert.ok(!calls.some((call) => call.startsWith('snapshot:')));
assert.ok(!calls.some((call) => call.includes(' sync --merge ')));
});
test('runSyncFlow --json emits NDJSON progress events and a final result', async () => {
const lines: string[] = [];
const remoteSummary = {
...createEmptyMergeSummary(),
sessionsMerged: 2,
videosAdded: 1,
};
const deps = makeHostDeps([], {
consoleLog: (line) => {
lines.push(line);
},
runSsh: (_host, command) => {
if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
if (command.includes(' sync --merge ')) {
assert.match(command, / --json(?: |$)/);
return ok(
`${JSON.stringify({ type: 'merge-summary', target: 'local', summary: remoteSummary })}\n` +
`${JSON.stringify({ type: 'result', ok: true, error: null })}\n`,
);
}
return ok();
},
});
await runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box', syncJson: true }),
deps,
);
const events = lines.map((line) => JSON.parse(line));
assert.ok(events.some((event) => event.type === 'stage' && event.stage === 'snapshot-local'));
assert.ok(events.some((event) => event.type === 'merge-summary' && event.target === 'local'));
assert.deepEqual(
events.find((event) => event.type === 'merge-summary' && event.target === 'remote'),
{ type: 'merge-summary', target: 'remote', summary: remoteSummary },
);
assert.deepEqual(events[events.length - 1], { type: 'result', ok: true, error: null });
});
test('runSyncFlow --json emits an error result when the sync fails', async () => {
const lines: string[] = [];
const deps = makeHostDeps([], {
consoleLog: (line) => {
lines.push(line);
},
runSsh: (_host, command) => {
if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
if (command.includes(' sync --merge ')) return { status: 9, stdout: '', stderr: 'boom' };
return ok();
},
});
await assert.rejects(() =>
runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box', syncJson: true }),
deps,
),
);
const events = lines.map((line) => JSON.parse(line));
const last = events[events.length - 1];
assert.equal(last.type, 'result');
assert.equal(last.ok, false);
assert.match(last.error, /Remote merge failed/);
});
test('runHostSync records host sync results for saved-host bookkeeping', async () => {
const recorded: Array<{ host: string; status: string; detail: string | null }> = [];
const record: SyncFlowDeps['recordHostSyncResult'] = (host, status, detail) => {
recorded.push({ host, status, detail });
};
await runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }),
makeHostDeps([], { recordHostSyncResult: record }),
);
assert.deepEqual(recorded[0], {
host: 'media-box',
status: 'success',
detail: '0 sessions merged; pushed local stats',
});
await assert.rejects(() =>
runSyncFlow(
makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'media-box' }),
makeHostDeps([], {
recordHostSyncResult: record,
runSsh: (_host, command) => {
if (command.includes(' sync --make-temp')) return ok('/tmp/subminer-sync-remote\n');
if (command.includes(' sync --merge ')) return { status: 9, stdout: '', stderr: 'boom' };
return ok();
},
}),
),
);
assert.equal(recorded.length, 2);
assert.equal(recorded[1]!.status, 'error');
});
test('runCheckMode --json reports ssh and remote SubMiner status', async () => {
const lines: string[] = [];
const sshOptions: unknown[] = [];
const deps = makeDeps({
consoleLog: (line) => {
lines.push(line);
},
resolveRemoteSubminerCommand: (host, _preferred, _flavor, runRemote) => {
runRemote!(host, 'subminer --help');
return 'subminer';
},
runSsh: (_host, command, options) => {
sshOptions.push(options);
if (command.includes('--version')) return ok('SubMiner 0.18.0\n');
return ok('subminer-check-ok\n');
},
});
await runSyncFlow(
makeContext({
syncDbPath: '/tmp/local.sqlite',
syncHost: 'media-box',
syncCheck: true,
syncJson: true,
}),
deps,
);
const events = lines.map((line) => JSON.parse(line));
const check = events.find((event) => event.type === 'check-result');
assert.ok(check);
assert.equal(check.host, 'media-box');
assert.equal(check.sshOk, true);
assert.equal(check.remoteCommand, 'subminer');
assert.equal(check.remoteVersion, 'SubMiner 0.18.0');
assert.equal(check.ok, true);
assert.equal(check.error, null);
assert.equal(sshOptions.length, 3);
assert.ok(
sshOptions.every(
(options) =>
JSON.stringify(options) ===
JSON.stringify({ batchMode: true, connectTimeoutSeconds: 10, timeoutMs: 15_000 }),
),
);
const failLines: string[] = [];
await assert.rejects(() =>
runSyncFlow(
makeContext({
syncDbPath: '/tmp/local.sqlite',
syncHost: 'media-box',
syncCheck: true,
syncJson: true,
}),
makeDeps({
consoleLog: (line) => {
failLines.push(line);
},
resolveRemoteSubminerCommand: () => {
throw new Error('Could not find a runnable "subminer" on media-box.');
},
runSsh: () => ok('subminer-check-ok\n'),
}),
),
);
const failEvents = failLines.map((line) => JSON.parse(line));
const failedCheck = failEvents.find((event) => event.type === 'check-result');
assert.ok(failedCheck);
assert.equal(failedCheck.ok, false);
assert.match(failedCheck.error, /Could not find a runnable/);
});
test('runHostSync speaks Windows shells: app command, double quotes, temp protocol', async () => {
const sshCommands: string[] = [];
const scpCalls: string[] = [];
const winTemp = 'C:\\Users\\First Last\\AppData\\Local\\Temp\\subminer-sync-remote';
const appCmd = '"%LOCALAPPDATA%\\Programs\\SubMiner\\SubMiner.exe" --sync-cli';
const deps = makeHostDeps([], {
detectRemoteShellFlavor: () => 'windows-cmd',
resolveRemoteSubminerCommand: () => appCmd,
createDbSnapshot: (_dbPath, outPath) => {
fs.writeFileSync(outPath, 'snapshot');
},
runSsh: (_host, command) => {
sshCommands.push(command);
if (command.includes(' sync --make-temp')) return ok(`${winTemp}\r\n`);
return ok();
},
runScp: (from, to) => {
scpCalls.push(`${from}->${to}`);
if (!to.includes(':')) fs.writeFileSync(to, 'pulled');
},
});
await runSyncFlow(makeContext({ syncDbPath: '/tmp/local.sqlite', syncHost: 'win-box' }), deps);
const expectedDir = 'C:/Users/First Last/AppData/Local/Temp/subminer-sync-remote';
assert.ok(sshCommands.includes(`${appCmd} sync --snapshot "${expectedDir}/snapshot.sqlite"`));
assert.ok(sshCommands.includes(`${appCmd} sync --merge "${expectedDir}/incoming.sqlite"`));
assert.ok(sshCommands.includes(`${appCmd} sync --remove-temp "${expectedDir}"`));
assert.ok(scpCalls.some((call) => call.startsWith(`win-box:${expectedDir}/snapshot.sqlite->`)));
assert.ok(scpCalls.some((call) => call.endsWith(`->win-box:${expectedDir}/incoming.sqlite`)));
// No POSIX shell-isms reach a Windows remote.
assert.ok(
!sshCommands.some((command) => command.startsWith('mktemp') || command.startsWith('rm ')),
);
assert.ok(!sshCommands.some((command) => command.includes('PATH="$HOME')));
});
test('runSyncFlow --make-temp prints a temp dir and --remove-temp only removes sync temp dirs', async () => {
const printed: string[] = [];
const removed: string[] = [];
const madeDir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-sync-'));
try {
await runSyncFlow(
makeContext({ syncMakeTemp: true }),
makeDeps({
mkdtempSync: () => madeDir,
consoleLog: (line) => {
printed.push(line);
},
}),
);
assert.deepEqual(printed, [madeDir]);
const removeDeps = makeDeps({
rmSync: (target) => {
removed.push(target);
},
});
await runSyncFlow(makeContext({ syncRemoveTempPath: madeDir }), removeDeps);
assert.deepEqual(removed, [madeDir]);
await assert.rejects(
() => runSyncFlow(makeContext({ syncRemoveTempPath: '/etc' }), removeDeps),
/Refusing to remove a directory outside the sync temp area/,
);
assert.equal(removed.length, 1);
} finally {
fs.rmSync(madeDir, { recursive: true, force: true });
}
});
+482
View File
@@ -0,0 +1,482 @@
import os from 'node:os';
import path from 'node:path';
import { formatMergeSummary } from './merge';
import { quoteForRemoteShell } from './ssh';
import type { RemoteRunResult, RemoteShellFlavor, RunSshOptions } from './ssh';
import {
parseSyncProgressLine,
type SyncMergeSummary,
type SyncProgressEvent,
} from '../../../shared/sync/sync-events';
import type { SyncResultStatus } from '../../../shared/sync/sync-hosts-store';
export interface SyncFlowArgs {
syncHost: string;
syncSnapshotPath: string;
syncMergePath: string;
syncDirection: 'both' | 'push' | 'pull' | null;
syncRemoteCmd: string;
syncDbPath: string;
syncForce: boolean;
syncJson: boolean;
syncCheck: boolean;
syncMakeTemp: boolean;
syncRemoveTempPath: string;
logLevel: string;
}
export interface SyncFlowContext {
args: SyncFlowArgs;
mpvSocketPath: string;
}
/**
* Process/IO seams the sync flow needs stubbed in tests: SSH/scp, the DB
* snapshot/merge engine, filesystem, and progress/bookkeeping output. The
* app's --sync-cli mode (src/main/sync-cli.ts) provides the only production
* binding; pure helpers are imported directly.
*/
export interface SyncFlowDeps {
createDbSnapshot: (dbPath: string, outPath: string) => void;
mergeSnapshotIntoDb: (localDbPath: string, snapshotPath: string) => SyncMergeSummary;
findLiveStatsDaemonPid: (dbPath: string) => number | null;
assertSafeSshHost: (host: string) => void;
detectRemoteShellFlavor: (
host: string,
runRemote: (host: string, remoteCommand: string) => RemoteRunResult,
) => RemoteShellFlavor;
resolveRemoteSubminerCommand: (
host: string,
preferred: string | null,
flavor: RemoteShellFlavor,
runRemote?: (host: string, remoteCommand: string) => RemoteRunResult,
) => string;
runScp: (from: string, to: string) => void;
runSsh: (host: string, remoteCommand: string, options?: RunSshOptions) => RemoteRunResult;
canConnectUnixSocket: (socketPath: string) => Promise<boolean>;
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<void>;
emitEvent: (event: SyncProgressEvent) => void;
recordHostSyncResult: (host: string, status: SyncResultStatus, detail: string | null) => void;
resolveDefaultDbPath: () => string;
}
/** Expand a leading `~` (as ssh users write paths) and make the path absolute. */
function resolveCliPath(input: string): string {
return input.startsWith('~') ? path.join(os.homedir(), input.slice(1)) : path.resolve(input);
}
/**
* Prefix shared by every sync temp dir, local or remote. Remote temp dirs are
* created and removed by the remote SubMiner itself (sync --make-temp /
* --remove-temp) so the flow never depends on mktemp/rm existing in the
* remote shell. This is what makes Windows remotes work.
*/
const SYNC_TEMP_PREFIX = 'subminer-sync-';
function makeSyncTempDir(mkdtempSync: SyncFlowDeps['mkdtempSync']): string {
return mkdtempSync(path.join(os.tmpdir(), SYNC_TEMP_PREFIX));
}
/** Only dirs directly under os.tmpdir() with the sync prefix may be removed. */
function assertRemovableSyncTempDir(target: string): string {
const resolved = path.resolve(target.trim());
const normalizeCase = (value: string) =>
process.platform === 'win32' ? value.toLowerCase() : value;
const insideTmp =
normalizeCase(path.dirname(resolved)) === normalizeCase(path.resolve(os.tmpdir()));
if (!insideTmp || !path.basename(resolved).startsWith(SYNC_TEMP_PREFIX)) {
throw new Error(`Refusing to remove a directory outside the sync temp area: ${target}`);
}
return resolved;
}
function runMakeTempMode(deps: SyncFlowDeps): void {
deps.consoleLog(makeSyncTempDir(deps.mkdtempSync));
}
function runRemoveTempMode(context: SyncFlowContext, deps: SyncFlowDeps): void {
const target = assertRemovableSyncTempDir(context.args.syncRemoveTempPath);
deps.rmSync(target, { recursive: true, force: true });
}
function resolveSyncDbPath(context: SyncFlowContext, deps: SyncFlowDeps): string {
const override = context.args.syncDbPath.trim();
return override ? resolveCliPath(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<void> {
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) {
throw new Error(
`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))) {
throw new Error(
`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.
function withJsonEvents(deps: SyncFlowDeps): SyncFlowDeps {
const writeLine = deps.consoleLog;
return {
...deps,
consoleLog: () => {},
writeStdout: () => true,
emitEvent: (event) => writeLine(JSON.stringify(event)),
};
}
async function runSnapshotMode(
context: SyncFlowContext,
dbPath: string,
deps: SyncFlowDeps,
): Promise<void> {
await deps.ensureTrackerQuiescent(context, dbPath);
const outPath = resolveCliPath(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);
}
async function runMergeMode(
context: SyncFlowContext,
dbPath: string,
deps: SyncFlowDeps,
): Promise<void> {
await deps.ensureTrackerQuiescent(context, dbPath);
const snapshotPath = resolveCliPath(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(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<void> {
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 runCheck = (checkHost: string, command: string) =>
deps.runSsh(checkHost, command, {
batchMode: true,
connectTimeoutSeconds: 10,
timeoutMs: 15_000,
});
const probe = runCheck(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 {
const flavor = deps.detectRemoteShellFlavor(host, runCheck);
if (flavor !== 'posix') deps.consoleLog(`Remote platform: Windows (${flavor})`);
remoteCommand = deps.resolveRemoteSubminerCommand(
host,
args.syncRemoteCmd || null,
flavor,
runCheck,
);
const version = runCheck(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.');
}
// The remote validates --remove-temp against its own tmpdir; this guard only
// keeps garbage output from an earlier failure out of the remote command.
function cleanupRemote(
host: string,
remoteCmd: string,
remoteTmpDir: string,
quote: (value: string) => string,
deps: SyncFlowDeps,
): void {
if (!path.posix.basename(remoteTmpDir).startsWith(SYNC_TEMP_PREFIX)) return;
deps.runSsh(host, `${remoteCmd} sync --remove-temp ${quote(remoteTmpDir)}`);
}
/**
* `sync --make-temp` prints the created dir as its last stdout line (a
* launcher wrapper may log above it). Backslashes are normalized to forward
* slashes: scp, the remote SubMiner, and Windows itself all accept them, and
* it keeps the later `${dir}/file` compositions valid on every platform.
*/
function parseRemoteTempDir(stdout: string): string {
const lines = stdout
.split('\n')
.map((line) => line.trim())
.filter((line) => line.length > 0);
const candidate = (lines[lines.length - 1] ?? '').replaceAll('\\', '/');
return path.posix.basename(candidate).startsWith(SYNC_TEMP_PREFIX) ? candidate : '';
}
function parseRemoteMergeSummary(stdout: string): SyncMergeSummary | null {
for (const line of stdout.split('\n')) {
const event = parseSyncProgressLine(line);
if (event?.type === 'merge-summary' && event.target === 'local') return event.summary;
}
return null;
}
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<void> {
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 flavor = deps.detectRemoteShellFlavor(host, deps.runSsh);
const remoteCmd = deps.resolveRemoteSubminerCommand(host, args.syncRemoteCmd || null, flavor);
const quote = (value: string) => quoteForRemoteShell(flavor, value);
if (args.logLevel === 'debug') {
console.error(`Remote subminer command (${flavor}): ${remoteCmd}`);
}
const localTmpDir = makeSyncTempDir(deps.mkdtempSync);
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, `${remoteCmd} sync --make-temp`);
remoteTmpDir = mktemp.status === 0 ? parseRemoteTempDir(mktemp.stdout) : '';
if (!remoteTmpDir) {
throw new Error(
formatRemoteRunError(`Could not create a temporary directory on ${host}.`, mktemp),
);
}
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 ${quote(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(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 ${quote(incomingSnapshot)}${forceFlag}${args.syncJson ? ' --json' : ''}`,
);
deps.writeStdout(mergeRun.stdout);
const remoteSummary = args.syncJson ? parseRemoteMergeSummary(mergeRun.stdout) : null;
if (remoteSummary) {
deps.emitEvent({ type: 'merge-summary', target: 'remote', summary: remoteSummary });
} else 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, remoteCmd, remoteTmpDir, quote, deps);
} catch {
// best effort
}
}
}
}
export async function runSyncFlow(
context: SyncFlowContext,
inputDeps: SyncFlowDeps,
): Promise<void> {
let deps = inputDeps;
const { args } = context;
if (args.syncJson) deps = withJsonEvents(deps);
try {
if (args.syncMakeTemp) {
runMakeTempMode(deps);
} else if (args.syncRemoveTempPath) {
runRemoveTempMode(context, deps);
} else {
const dbPath = resolveSyncDbPath(context, deps);
if (args.syncCheck) {
await runCheckMode(context, deps);
} else if (args.syncSnapshotPath) {
await runSnapshotMode(context, dbPath, deps);
} else if (args.syncMergePath) {
await runMergeMode(context, dbPath, deps);
} else if (args.syncHost) {
await runHostSync(context, dbPath, deps);
} else {
throw new Error('sync requires a host, --snapshot <file>, or --merge <file>.');
}
}
} 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 });
}
+61
View File
@@ -0,0 +1,61 @@
import fs from 'node:fs';
export interface SyncDbOpenOptions {
readonly?: boolean;
/**
* Read-write open. The libsql opener treats "not readonly" as read-write,
* but the launcher's bun:sqlite history reader passes these options straight
* to `new Database(...)`, where an explicit readwrite flag is what prevents
* bun from defaulting to readwrite+create. Keep it.
*/
readwrite?: boolean;
create?: boolean;
}
/**
* 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<T>(
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;
}