--help\` for command options.
+`.trim();
+
+const COMMAND_USAGE = {
+ login: `
+Usage: sinker login [options]
+
+Options:
+ --handle HANDLE atproto handle (default: ATPROTO_HANDLE)
+ --port PORT loopback callback port (default: 8765)
+ --login-host HOST interface to bind (default: 127.0.0.1; use 0.0.0.0 in Docker)
+ --no-open print the URL instead of opening a browser
+`.trim(),
+
+ logout: `
+Usage: sinker logout [options]
+`.trim(),
+
+ status: `
+Usage: sinker status [options]
+
+Options:
+ --json machine-readable output
+ --check verify the saved session against BookHive
+`.trim(),
+
+ sync: `
+Usage: sinker sync [file…] [options]
+
+Compresses and uploads new books once. With no files, scans the source folder.
+
+Options:
+ --source DIR source folder (default: SINKER_SOURCE_DIR)
+ --out DIR output folder (default: SINKER_OUTPUT_DIR)
+ --device X4|X3 target screen (default: X4)
+ --quality N JPEG quality 1-100 (default: 85)
+ --auto-crop trim uniform margins before resizing
+ --no-rename keep source filenames instead of "Title - Author.epub"
+ --dry-run do everything except write files, upload, or update state
+ --force upload even if the book is already in BookHive
+ --retry-failed retry files that previously failed permanently
+`.trim(),
+
+ watch: `
+Usage: sinker watch [options]
+
+Watches the source folder until Ctrl+C.
+
+Options: (same as sync, plus)
+ --no-backfill only process files added after startup
+ --poll poll the filesystem (network volumes)
+ --rescan-ms N periodic rescan interval in ms (default: 300000)
+ --retry-max N upload retry attempts (default: 5)
+ --timeout-ms N per-request timeout in ms (default: 120000)
+ --verbose log per-image detail and retry decisions
+`.trim(),
+};
+
+class UsageError extends Error {}
+
+function buildOverrides(values) {
+ return {
+ handle: values.handle,
+ sourceDir: values.source,
+ outputDir: values.out,
+ device: values.device,
+ quality: values.quality,
+ autoCrop: values['auto-crop'] ? true : undefined,
+ rename: values['no-rename'] ? false : undefined,
+ loginPort: values.port,
+ loginHost: values['login-host'],
+ openBrowser: values['no-open'] ? false : undefined,
+ poll: values.poll ? true : undefined,
+ backfill: values['no-backfill'] ? false : undefined,
+ rescanMs: values['rescan-ms'],
+ retryMax: values['retry-max'],
+ timeoutMs: values['timeout-ms'],
+ dryRun: values['dry-run'] ? true : undefined,
+ force: values.force ? true : undefined,
+ retryFailed: values['retry-failed'] ? true : undefined,
+ verbose: values.verbose ? true : undefined,
+ quiet: values.quiet ? true : undefined,
+ };
+}
+
+function parseCli(argv) {
+ let parsed;
+ try {
+ parsed = parseArgs({
+ args: argv,
+ allowPositionals: true,
+ strict: true,
+ options: {
+ 'env-file': { type: 'string' },
+ help: { type: 'boolean', short: 'h' },
+
+ verbose: { type: 'boolean' },
+ quiet: { type: 'boolean' },
+
+ handle: { type: 'string' },
+ port: { type: 'string' },
+ 'login-host': { type: 'string' },
+ 'no-open': { type: 'boolean' },
+
+ source: { type: 'string' },
+ out: { type: 'string' },
+ device: { type: 'string' },
+ quality: { type: 'string' },
+ 'auto-crop': { type: 'boolean' },
+ 'no-rename': { type: 'boolean' },
+ 'dry-run': { type: 'boolean' },
+ force: { type: 'boolean' },
+ 'retry-failed': { type: 'boolean' },
+
+ poll: { type: 'boolean' },
+ 'no-backfill': { type: 'boolean' },
+ 'rescan-ms': { type: 'string' },
+ 'retry-max': { type: 'string' },
+ 'timeout-ms': { type: 'string' },
+
+ json: { type: 'boolean' },
+ check: { type: 'boolean' },
+ },
+ });
+ } catch (error) {
+ throw new UsageError(error.message);
+ }
+
+ return parsed;
+}
+
+function helpFor(command) {
+ if (command && COMMAND_USAGE[command]) return `${COMMAND_USAGE[command]}\n\n${USAGE}`;
+ return USAGE;
+}
+
+async function promptHandle() {
+ if (!process.stdin.isTTY) return null;
+ const rl = createInterface({ input: process.stdin, output: process.stdout });
+ try {
+ const answer = await rl.question('Bluesky handle (e.g. you.bsky.social): ');
+ return answer.trim() || null;
+ } finally {
+ rl.close();
+ }
+}
+
+function ledgerFor(config, did) {
+ return new Ledger(stateFileFor(config.stateDir, did ?? 'anonymous')).load();
+}
+
+async function runLogin(config, log) {
+ let handle = config.handle;
+ if (!handle) handle = await promptHandle();
+ if (!handle) throw new UsageError('a handle is required: pass --handle or set ATPROTO_HANDLE');
+
+ const { did } = await login(config, { handle, log });
+ log.ok(`Signed in as ${did}.`);
+ return 0;
+}
+
+async function runLogout(config, log) {
+ const did = await logout(config);
+ if (did) log.ok(`Signed out ${did}.`);
+ else log.info('No saved session.');
+ return 0;
+}
+
+function sourceSummary(config) {
+ if (!existsSync(config.sourceDir)) {
+ return { exists: false, epubs: 0, other: 0 };
+ }
+ let epubs = 0;
+ let other = 0;
+ for (const entry of readdirSync(config.sourceDir, { withFileTypes: true })) {
+ if (entry.name.startsWith('.') || entry.isDirectory()) continue;
+ let isFile = entry.isFile();
+ if (!isFile) {
+ try { isFile = statSync(resolve(config.sourceDir, entry.name)).isFile(); } catch { continue; }
+ }
+ if (!isFile) continue;
+ if (/\.epub$/i.test(entry.name)) epubs++;
+ else other++;
+ }
+ return { exists: true, epubs, other };
+}
+
+async function runStatus(config, log, values) {
+ const pointer = readSessionPointer(config);
+ const did = pointer?.did ?? null;
+ const ledger = ledgerFor(config, did);
+ const source = sourceSummary(config);
+
+ const data = {
+ signedIn: Boolean(did),
+ did,
+ handle: pointer?.handle ?? config.handle ?? null,
+ signedInAt: pointer?.signedInAt ?? null,
+ bookhive: { url: config.bookhiveUrl, did: config.bookhiveDid },
+ source: { ...source, dir: config.sourceDir },
+ output: { dir: config.outputDir, exists: existsSync(config.outputDir) },
+ device: config.device,
+ quality: config.quality,
+ autoCrop: config.autoCrop,
+ rename: config.rename,
+ ledger: { file: stateFileFor(config.stateDir, did ?? 'anonymous'), counts: ledger.counts() },
+ };
+
+ if (values.check) {
+ if (!did) throw new AuthError('not signed in — run `sinker login` first');
+ const { session } = await restoreSession(config);
+ const client = createBookHiveClient({ config, session, logger: log });
+ const library = await client.refreshLibrary();
+ data.library = {
+ books: library.size,
+ storage: client.storage,
+ };
+ }
+
+ if (values.json) {
+ console.log(JSON.stringify(data, null, 2));
+ return 0;
+ }
+
+ console.log(describeConfig(config));
+ console.log('');
+ if (did) {
+ console.log(`session ${pointer.handle ? `@${pointer.handle} ` : ''}${did}`);
+ console.log(`signed in ${pointer.signedInAt ?? 'unknown'}`);
+ } else {
+ console.log('session not signed in (run `sinker login`)');
+ }
+
+ const counts = data.ledger.counts;
+ const parts = Object.entries(counts).map(([status, count]) => `${count} ${status}`);
+ console.log(`ledger ${parts.length ? parts.join(', ') : 'empty'}`);
+ console.log(`files ${source.epubs} epub(s)${source.other ? `, ${source.other} other file(s)` : ''}`);
+ if (data.library) {
+ const storage = data.library.storage
+ ? ` (${formatBytes(data.library.storage.usedBytes)} / ${formatBytes(data.library.storage.quotaBytes)})`
+ : '';
+ console.log(`library ${data.library.books} book(s)${storage}`);
+ }
+ return 0;
+}
+
+async function makeClient(config, log) {
+ const { did, session } = await restoreSession(config);
+ const client = createBookHiveClient({ config, session, logger: log });
+ return { did, client };
+}
+
+async function runSync(config, log, paths) {
+ const { did, client } = await makeClient(config, log);
+ const ctx = createContext({ config, bookhive: client, logger: log, did });
+
+ const explicit = paths.length
+ ? paths.map(path => {
+ const resolved = resolve(config.cwd, path);
+ if (!existsSync(resolved) || !statSync(resolved).isFile()) {
+ throw new UsageError(`not a file: ${resolved}`);
+ }
+ return resolved;
+ })
+ : undefined;
+
+ log.info(`Signed in as ${did}`);
+ log.info(`Target ${DEVICE_PROFILES[config.device].label} | quality ${config.quality}` +
+ `${config.autoCrop ? ' | auto-crop' : ''}${config.dryRun ? ' | DRY RUN' : ''}`);
+
+ const stats = await syncOnce(ctx, explicit ? { paths: explicit } : {});
+ log.info(`Done: ${summarize(stats)}`);
+ return stats.failed > 0 ? 1 : 0;
+}
+
+async function runWatch(config, log) {
+ const { did, client } = await makeClient(config, log);
+ const ctx = createContext({ config, bookhive: client, logger: log, did });
+
+ log.info(`Signed in as ${did}`);
+ log.info(`Target ${DEVICE_PROFILES[config.device].label} | quality ${config.quality}` +
+ `${config.autoCrop ? ' | auto-crop' : ''}${config.dryRun ? ' | DRY RUN' : ''}`);
+
+ return watch(ctx);
+}
+
+async function main(argv) {
+ const { values, positionals } = parseCli(argv);
+ const command = positionals[0];
+
+ if (values.help || command === 'help') {
+ console.log(helpFor(command === 'help' ? null : command));
+ return 0;
+ }
+
+ if (!command) {
+ console.log(USAGE);
+ return 2;
+ }
+
+ if (!['login', 'logout', 'status', 'sync', 'watch'].includes(command)) {
+ throw new UsageError(`unknown command "${command}"`);
+ }
+
+ const log = createLogger({ verbose: values.verbose, quiet: values.quiet });
+
+ const config = loadConfig({
+ envFile: values['env-file'] ? resolve(values['env-file']) : undefined,
+ cwd: process.cwd(),
+ overrides: buildOverrides(values),
+ });
+
+ const extra = positionals.slice(1);
+ if (extra.length && command !== 'sync') {
+ throw new UsageError(`unexpected argument "${extra[0]}"`);
+ }
+
+ switch (command) {
+ case 'login': return runLogin(config, log);
+ case 'logout': return runLogout(config, log);
+ case 'status': return runStatus(config, log, values);
+ case 'sync': return runSync(config, log, extra);
+ case 'watch': return runWatch(config, log);
+ default: throw new UsageError(`unknown command "${command}"`);
+ }
+}
+
+main(process.argv.slice(2))
+ .then(code => { process.exitCode = code; })
+ .catch(error => {
+ if (error instanceof UsageError) {
+ console.error(`error: ${error.message}`);
+ console.error('');
+ console.error(USAGE);
+ } else if (error instanceof ConfigError || error instanceof AuthError) {
+ console.error(`error: ${error.message}`);
+ } else {
+ console.error(`error: ${error?.message ?? error}`);
+ if (process.env.DEBUG) console.error(error.stack);
+ }
+ process.exitCode = 1;
+ });
diff --git a/src/sinker/auth.js b/src/sinker/auth.js
new file mode 100644
index 0000000..b78cc66
--- /dev/null
+++ b/src/sinker/auth.js
@@ -0,0 +1,323 @@
+// atproto OAuth for a loopback CLI client.
+//
+// We authenticate the user once with OAuth (public loopback client, no hosted
+// metadata or keyset), persist the session, and later use that session only to
+// mint short-lived inter-service auth tokens for BookHive's XRPC methods.
+
+import { execFile } from 'node:child_process';
+import { createServer } from 'node:http';
+import {
+ chmodSync, existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmSync,
+ unlinkSync, writeFileSync,
+} from 'node:fs';
+import { dirname, join } from 'node:path';
+
+import {
+ OAuthClient, TokenInvalidError, TokenRefreshError, TokenRevokedError,
+} from '@atcute/oauth-node-client';
+import {
+ CompositeDidDocumentResolver, CompositeHandleResolver, LocalActorResolver,
+ PlcDidDocumentResolver, WebDidDocumentResolver, WellKnownHandleResolver,
+} from '@atcute/identity-resolver';
+import { NodeDnsHandleResolver } from '@atcute/identity-resolver-node';
+
+/** BookHive RPC methods this tool needs to call. */
+export const BOOKHIVE_RPC_METHODS = [
+ 'buzz.bookhive.getPersonalLibrary',
+ 'buzz.bookhive.getPersonalBook',
+ 'buzz.bookhive.uploadPersonalBook',
+];
+
+/** Granular scopes. BookHive's published permission set predates these RPCs. */
+export function oauthScope() {
+ return [
+ 'atproto',
+ ...BOOKHIVE_RPC_METHODS.map(method => `rpc:${method}?aud=*`),
+ ];
+}
+
+export class AuthError extends Error {}
+
+function atomicWriteJson(file, value) {
+ mkdirSync(dirname(file), { recursive: true, mode: 0o700 });
+ const tmp = `${file}.${process.pid}.tmp`;
+ writeFileSync(tmp, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 });
+ renameSync(tmp, file);
+ try { chmodSync(file, 0o600); } catch { /* best effort */ }
+}
+
+function readJson(file) {
+ try {
+ return JSON.parse(readFileSync(file, 'utf8'));
+ } catch {
+ return undefined;
+ }
+}
+
+/** File-backed atcute Store implementation. */
+function fileStore(dir, prefix, { ttlMs } = {}) {
+ mkdirSync(dir, { recursive: true, mode: 0o700 });
+ const fileFor = key => join(dir, `${prefix}-${encodeURIComponent(String(key))}.json`);
+
+ return {
+ async get(key) {
+ const value = readJson(fileFor(key));
+ if (!value) return undefined;
+ if (ttlMs && value.expiresAt && Date.now() > value.expiresAt) {
+ try { unlinkSync(fileFor(key)); } catch { /* ignore */ }
+ return undefined;
+ }
+ return value;
+ },
+ async set(key, value) {
+ atomicWriteJson(fileFor(key), value);
+ },
+ async delete(key) {
+ try { unlinkSync(fileFor(key)); } catch { /* ignore */ }
+ },
+ async clear() {
+ if (!existsSync(dir)) return;
+ for (const name of readdirSync(dir)) {
+ if (name.startsWith(`${prefix}-`)) {
+ try { unlinkSync(join(dir, name)); } catch { /* ignore */ }
+ }
+ }
+ },
+ };
+}
+
+export function sessionPointerPath(config) {
+ return join(config.oauthDir, 'session.json');
+}
+
+/** The loopback port is part of the client id, so reuse the one from login. */
+function oauthConfigFor(config, pointer) {
+ const port = Number(pointer?.loginPort);
+ if (!Number.isInteger(port) || port === config.loginPort) return config;
+ return { ...config, loginPort: port };
+}
+
+export function readSessionPointer(config) {
+ return readJson(sessionPointerPath(config));
+}
+
+function writeSessionPointer(config, pointer) {
+ atomicWriteJson(sessionPointerPath(config), pointer);
+}
+
+function clearSessionPointer(config) {
+ rmSync(sessionPointerPath(config), { force: true });
+}
+
+/**
+ * Build the OAuth client. `loginPort` is baked into the redirect URI, so it
+ * must match the port the callback server actually binds.
+ */
+export function createOAuth(config) {
+ const redirectUri = `http://127.0.0.1:${config.loginPort}/callback`;
+ const oauthDir = join(config.oauthDir, 'oauth');
+
+ const sessions = fileStore(join(oauthDir, 'sessions'), 'session');
+ const states = fileStore(join(oauthDir, 'states'), 'state', { ttlMs: 15 * 60_000 });
+
+ const actorResolver = new LocalActorResolver({
+ handleResolver: new CompositeHandleResolver({
+ methods: {
+ dns: new NodeDnsHandleResolver(),
+ http: new WellKnownHandleResolver(),
+ },
+ }),
+ didDocumentResolver: new CompositeDidDocumentResolver({
+ methods: {
+ plc: new PlcDidDocumentResolver(),
+ web: new WebDidDocumentResolver(),
+ },
+ }),
+ });
+
+ const oauth = new OAuthClient({
+ metadata: {
+ redirect_uris: [redirectUri],
+ scope: oauthScope(),
+ },
+ stores: { sessions, states },
+ actorResolver,
+ });
+
+ return { oauth, sessions, states, redirectUri };
+}
+
+function openBrowser(url) {
+ const [command, args] = process.platform === 'darwin'
+ ? ['open', [url]]
+ : process.platform === 'win32'
+ ? ['cmd', ['/c', 'start', '', url]]
+ : ['xdg-open', [url]];
+
+ return new Promise(resolve => {
+ execFile(command, args, error => resolve(!error));
+ });
+}
+
+const CALLBACK_TIMEOUT_MS = 10 * 60_000;
+
+/**
+ * Run the interactive login flow.
+ * @param {object} config
+ * @param {{handle: string, log: object}} options
+ * @returns {Promise<{did: string, redirectUri: string}>}
+ */
+export async function login(config, { handle, log }) {
+ if (!handle) throw new AuthError('a handle is required (pass --handle or set ATPROTO_HANDLE)');
+
+ const { oauth, redirectUri } = createOAuth(config);
+ const host = config.loginHost;
+
+ return new Promise((resolve, reject) => {
+ let settled = false;
+ let timer;
+
+ const finish = (error, did) => {
+ if (settled) return;
+ settled = true;
+ clearTimeout(timer);
+ if (server.listening) server.close();
+ if (error) reject(error);
+ else resolve({ did, redirectUri });
+ };
+
+ const server = createServer(async (req, res) => {
+ const url = new URL(req.url ?? '/', redirectUri);
+ if (url.pathname !== '/callback') {
+ res.writeHead(404, { 'content-type': 'text/plain' });
+ res.end('Not found');
+ return;
+ }
+
+ const oauthError = url.searchParams.get('error');
+ if (oauthError) {
+ const description = url.searchParams.get('error_description') ?? '';
+ res.writeHead(400, { 'content-type': 'text/plain' });
+ res.end(`Authorization failed: ${oauthError} ${description}`);
+ finish(new AuthError(`authorization failed: ${oauthError} ${description}`.trim()));
+ return;
+ }
+
+ try {
+ const { session } = await oauth.callback(url.searchParams);
+ const did = session.did;
+ writeSessionPointer(config, {
+ did,
+ handle: handle.replace(/^@/, ''),
+ loginPort: config.loginPort,
+ signedInAt: new Date().toISOString(),
+ });
+
+ res.writeHead(200, { 'content-type': 'text/html; charset=utf-8' });
+ res.end(
+ 'Signed in' +
+ '' +
+ 'Signed in to BookHive
You can close this tab and return to the terminal.
',
+ );
+ finish(null, did);
+ } catch (error) {
+ res.writeHead(500, { 'content-type': 'text/plain' });
+ res.end(`Login failed: ${error.message}`);
+ finish(new AuthError(`login failed: ${error.message}`));
+ }
+ });
+
+ server.on('error', error => {
+ finish(new AuthError(
+ error.code === 'EADDRINUSE'
+ ? `port ${config.loginPort} is already in use; pass --port with a free port`
+ : `could not start the login server: ${error.message}`,
+ ));
+ });
+
+ timer = setTimeout(() => {
+ finish(new AuthError('login timed out after 10 minutes'));
+ }, CALLBACK_TIMEOUT_MS);
+
+ server.listen(config.loginPort, host, async () => {
+ try {
+ const { url } = await oauth.authorize({
+ target: { type: 'account', identifier: handle.replace(/^@/, '') },
+ state: { source: 'cross-reader-sinker' },
+ });
+
+ log.always('');
+ log.always('Open this URL to authorize Cross Reader Sinker:');
+ log.always('');
+ log.always(` ${url}`);
+ log.always('');
+ log.always(`Waiting for the callback on ${redirectUri} ...`);
+
+ if (config.openBrowser) {
+ const opened = await openBrowser(url.toString());
+ if (!opened) log.warn('could not open a browser automatically; open the URL above.');
+ }
+ } catch (error) {
+ finish(new AuthError(`could not start authorization: ${error.message}`));
+ }
+ });
+ });
+}
+
+/**
+ * Restore the saved session (refreshing tokens if needed).
+ * @returns {Promise<{did: string, session: import('@atcute/oauth-node-client').OAuthSession, oauth: OAuthClient}>}
+ */
+export async function restoreSession(config) {
+ const pointer = readSessionPointer(config);
+ if (!pointer?.did) {
+ throw new AuthError('not signed in — run `sinker login` first');
+ }
+
+ const { oauth, sessions } = createOAuth(oauthConfigFor(config, pointer));
+ const stored = await sessions.get(pointer.did);
+ if (!stored) {
+ throw new AuthError(
+ `no stored session for ${pointer.did} — run \`sinker login\` again`,
+ );
+ }
+ try {
+ const session = await oauth.restore(pointer.did);
+ return { did: session.did, session, oauth };
+ } catch (error) {
+ const message = error?.message ?? '';
+ if (/not\s+found|no\s+(stored\s+)?session|missing/i.test(message)) {
+ throw new AuthError(
+ `no stored session for ${pointer.did} — run \`sinker login\` again`,
+ );
+ }
+ if (error instanceof TokenRefreshError
+ || error instanceof TokenInvalidError
+ || error instanceof TokenRevokedError
+ || /refresh|session|token/i.test(message)) {
+ throw new AuthError(
+ `the saved session for ${pointer.did} can no longer be refreshed ` +
+ `(sessions expire after about two weeks of inactivity). Run \`sinker login\` again.`,
+ );
+ }
+ throw new AuthError(`could not restore the saved session: ${error?.message ?? error}`);
+ }
+}
+
+/** Revoke the saved session, if any. Returns the DID that was signed out. */
+export async function logout(config) {
+ const pointer = readSessionPointer(config);
+ if (!pointer?.did) {
+ clearSessionPointer(config);
+ return null;
+ }
+
+ const { oauth } = createOAuth(oauthConfigFor(config, pointer));
+ try {
+ await oauth.revoke(pointer.did);
+ } catch {
+ // Tokens may already be invalid; removing the local pointer is what matters.
+ }
+ clearSessionPointer(config);
+ return pointer.did;
+}
diff --git a/src/sinker/bookhive.js b/src/sinker/bookhive.js
new file mode 100644
index 0000000..d705c6c
--- /dev/null
+++ b/src/sinker/bookhive.js
@@ -0,0 +1,242 @@
+// BookHive XRPC client.
+//
+// BookHive does not accept OAuth access tokens directly. It accepts an atproto
+// inter-service auth JWT, minted by the user's PDS for one method (`lxm`) and
+// one audience. So every call is:
+//
+// OAuth session -> com.atproto.server.getServiceAuth -> BookHive /xrpc/
+//
+// Tokens are 60-second tokens; we cache them for 50 seconds per method.
+
+import { LibraryIndex } from './dedupe.js';
+
+const RETRYABLE_STATUS = new Set([429, 500, 502, 503, 504]);
+const TOKEN_TTL_MS = 50_000;
+
+export class BookHiveError extends Error {
+ constructor(message, { code = 'RequestFailed', status = 0, retryable = false } = {}) {
+ super(message);
+ this.name = 'BookHiveError';
+ this.code = code;
+ this.status = status;
+ this.retryable = retryable;
+ }
+}
+
+const defaultSleep = ms => new Promise(resolve => setTimeout(resolve, ms));
+
+export function createBookHiveClient({
+ config,
+ session,
+ logger,
+ fetchImpl = fetch,
+ sleep = defaultSleep,
+}) {
+ const tokenCache = new Map();
+ let library = new LibraryIndex();
+ let libraryMeta = { total: 0, storage: null };
+
+ async function mintToken(lxm) {
+ const exp = Math.floor(Date.now() / 1000) + 60;
+ const path = '/xrpc/com.atproto.server.getServiceAuth'
+ + `?aud=${encodeURIComponent(config.bookhiveDid)}`
+ + `&lxm=${encodeURIComponent(lxm)}`
+ + `&exp=${exp}`;
+
+ let response;
+ try {
+ response = await session.handle(path, {
+ method: 'GET',
+ signal: AbortSignal.timeout(config.timeoutMs),
+ });
+ } catch (error) {
+ throw new BookHiveError(`could not mint a service token: ${error.message}`, {
+ code: 'TokenError',
+ retryable: true,
+ });
+ }
+
+ const text = await response.text();
+ let data = null;
+ try { data = text ? JSON.parse(text) : null; } catch { /* non-JSON error body */ }
+
+ if (!response.ok || !data?.token) {
+ throw new BookHiveError(
+ `could not mint a service token: ${data?.message ?? data?.error ?? `HTTP ${response.status}`}`,
+ {
+ code: data?.error ?? 'TokenError',
+ status: response.status,
+ retryable: response.status >= 500,
+ },
+ );
+ }
+
+ tokenCache.set(lxm, { token: data.token, expiresAt: Date.now() + TOKEN_TTL_MS });
+ return data.token;
+ }
+
+ async function tokenFor(lxm) {
+ const cached = tokenCache.get(lxm);
+ if (cached && cached.expiresAt > Date.now()) return cached.token;
+ return mintToken(lxm);
+ }
+
+ function backoff(attempt) {
+ return Math.min(30_000, 500 * 2 ** (attempt - 1)) + Math.floor(Math.random() * 250);
+ }
+
+ async function call(method, { params, body, contentType } = {}) {
+ const url = new URL(`${config.bookhiveUrl}/xrpc/${method}`);
+ for (const [key, value] of Object.entries(params ?? {})) {
+ if (value !== undefined && value !== null) url.searchParams.set(key, String(value));
+ }
+
+ let attempt = 0;
+ let authRetried = false;
+
+ for (;;) {
+ attempt++;
+ let token;
+ try {
+ token = await tokenFor(method);
+ } catch (error) {
+ if (error instanceof BookHiveError && error.retryable && attempt < config.retryMax) {
+ await sleep(backoff(attempt));
+ continue;
+ }
+ throw error;
+ }
+
+ let response;
+ try {
+ response = await fetchImpl(url, {
+ method: body === undefined ? 'GET' : 'POST',
+ headers: {
+ authorization: `Bearer ${token}`,
+ ...(contentType ? { 'content-type': contentType } : {}),
+ },
+ body,
+ signal: AbortSignal.timeout(config.timeoutMs),
+ });
+ } catch (error) {
+ if (attempt < config.retryMax) {
+ logger.debug(`${method}: ${error.message}; retrying (${attempt}/${config.retryMax})`);
+ await sleep(backoff(attempt));
+ continue;
+ }
+ throw new BookHiveError(`${method} failed: ${error.message}`, {
+ code: 'NetworkError',
+ retryable: true,
+ });
+ }
+
+ // A cached token can be revoked or clock-skewed. Mint a fresh one once,
+ // without consuming a retry attempt.
+ if (response.status === 401 && !authRetried) {
+ authRetried = true;
+ attempt--;
+ tokenCache.delete(method);
+ continue;
+ }
+
+ const text = await response.text();
+ let data = null;
+ try { data = text ? JSON.parse(text) : null; } catch { /* non-JSON body */ }
+
+ if (response.ok) {
+ return data ?? {};
+ }
+
+ const code = data?.error ?? `HTTP${response.status}`;
+ const message = data?.message ?? (text ? text.slice(0, 200) : response.statusText);
+ const retryable = RETRYABLE_STATUS.has(response.status) || code === 'Busy';
+
+ if (retryable && attempt < config.retryMax) {
+ logger.debug(`${method}: ${code} (${message}); retrying (${attempt}/${config.retryMax})`);
+ await sleep(backoff(attempt));
+ continue;
+ }
+
+ throw new BookHiveError(`${method}: ${code}${message ? ` — ${message}` : ''}`, {
+ code,
+ status: response.status,
+ retryable,
+ });
+ }
+ }
+
+ return {
+ get library() {
+ return library;
+ },
+
+ get storage() {
+ return libraryMeta.storage;
+ },
+
+ /** Fetch (or 404 -> null) one personal-library book by content hash. */
+ async getPersonalBook(contentHash) {
+ try {
+ const data = await call('buzz.bookhive.getPersonalBook', { params: { contentHash } });
+ return data?.book ?? null;
+ } catch (error) {
+ if (error instanceof BookHiveError && (error.status === 404 || error.code === 'NotFound')) {
+ return null;
+ }
+ throw error;
+ }
+ },
+
+ /** Load every page of the personal library into the dedupe index. */
+ async refreshLibrary() {
+ const books = [];
+ let cursor;
+ let total = 0;
+ let storage = null;
+ let pages = 0;
+
+ do {
+ const data = await call('buzz.bookhive.getPersonalLibrary', {
+ params: { limit: 100, sort: 'recent', ...(cursor ? { cursor } : {}) },
+ });
+ if (Array.isArray(data?.books)) books.push(...data.books);
+ if (typeof data?.total === 'number') total = data.total;
+ if (data?.storage) storage = data.storage;
+ cursor = data?.cursor;
+ pages++;
+ } while (cursor && pages < 500);
+
+ library = new LibraryIndex(books);
+ libraryMeta = { total: total || books.length, storage };
+ logger.debug(`library index: ${library.size} book(s)`);
+ return library;
+ },
+
+ /**
+ * Upload raw EPUB bytes. Treats AlreadyExists as success — the caller's
+ * content hash is the same value BookHive used for its duplicate check.
+ */
+ async upload(buffer, filename) {
+ try {
+ const data = await call('buzz.bookhive.uploadPersonalBook', {
+ params: { filename },
+ body: buffer,
+ contentType: 'application/epub+zip',
+ });
+ if (data?.book) library.add(data.book);
+ if (data?.storageUsedBytes !== undefined) {
+ libraryMeta.storage = {
+ usedBytes: data.storageUsedBytes,
+ quotaBytes: data.storageQuotaBytes,
+ };
+ }
+ return { alreadyExists: false, ...data };
+ } catch (error) {
+ if (error instanceof BookHiveError && (error.status === 409 || error.code === 'AlreadyExists')) {
+ return { alreadyExists: true };
+ }
+ throw error;
+ }
+ },
+ };
+}
diff --git a/src/sinker/config.js b/src/sinker/config.js
new file mode 100644
index 0000000..d01102c
--- /dev/null
+++ b/src/sinker/config.js
@@ -0,0 +1,225 @@
+// Configuration: .env -> environment -> CLI flags.
+//
+// Flags win over the environment, which wins over defaults. Paths are
+// resolved against the working directory (except `~`, which is expanded).
+
+import { homedir } from 'node:os';
+import { isAbsolute, join, resolve } from 'node:path';
+import dotenv from 'dotenv';
+
+import { DEVICE_PROFILES, DEFAULT_DEVICE, DEFAULT_JPEG_QUALITY } from '../constants.js';
+
+export const BOOKHIVE_URL = 'https://bookhive.buzz';
+export const BOOKHIVE_DID = 'did:plc:enu2j5xjlqsjaylv3du4myh4';
+
+export class ConfigError extends Error {}
+
+const DEFAULTS = {
+ sourceDir: './unformatted-books',
+ outputDir: './formatted-books',
+ device: DEFAULT_DEVICE,
+ quality: DEFAULT_JPEG_QUALITY,
+ autoCrop: false,
+ rename: true,
+ loginPort: 8765,
+ loginHost: '127.0.0.1',
+ poll: false,
+ backfill: true,
+ rescanMs: 5 * 60_000,
+ retryMax: 5,
+ timeoutMs: 120_000,
+};
+
+function expandHome(value) {
+ if (value === '~') return homedir();
+ if (value.startsWith('~/')) return join(homedir(), value.slice(2));
+ return value;
+}
+
+function resolvePath(value, cwd) {
+ const expanded = expandHome(String(value));
+ return isAbsolute(expanded) ? expanded : resolve(cwd, expanded);
+}
+
+function defaultOauthDir(env) {
+ const base = env.XDG_CONFIG_HOME || join(homedir(), '.config');
+ return join(base, 'cross-reader-sinker');
+}
+
+function defaultStateDir(env) {
+ const base = env.XDG_STATE_HOME || join(homedir(), '.local', 'state');
+ return join(base, 'cross-reader-sinker');
+}
+
+function optionalString(value) {
+ if (value === undefined || value === null) return undefined;
+ const trimmed = String(value).trim();
+ return trimmed === '' ? undefined : trimmed;
+}
+
+function parseBool(name, value) {
+ if (value === undefined || value === null || value === '') return undefined;
+ if (typeof value === 'boolean') return value;
+ const normalized = String(value).trim().toLowerCase();
+ if (['1', 'true', 'yes', 'on'].includes(normalized)) return true;
+ if (['0', 'false', 'no', 'off'].includes(normalized)) return false;
+ throw new ConfigError(`${name} must be a boolean (got "${value}")`);
+}
+
+function parseInteger(name, value, { min, max } = {}) {
+ if (value === undefined || value === null || value === '') return undefined;
+ const number = Number(value);
+ if (!Number.isInteger(number)) throw new ConfigError(`${name} must be an integer (got "${value}")`);
+ if (min !== undefined && number < min) throw new ConfigError(`${name} must be >= ${min} (got ${number})`);
+ if (max !== undefined && number > max) throw new ConfigError(`${name} must be <= ${max} (got ${number})`);
+ return number;
+}
+
+/** Pick the first defined value. */
+function firstDefined(...values) {
+ return values.find(value => value !== undefined);
+}
+
+/**
+ * @param {object} [options]
+ * @param {string} [options.envFile] explicit env file; missing file is an error
+ * @param {Record} [options.env] environment source
+ * @param {string} [options.cwd]
+ * @param {object} [options.overrides] CLI values (undefined = not passed)
+ */
+export function loadConfig({ envFile, env = process.env, cwd = process.cwd(), overrides = {} } = {}) {
+ // Copy so callers (and tests) don't observe dotenv's mutations.
+ const mergedEnv = { ...env };
+ const envPath = envFile ?? join(cwd, '.env');
+ const result = dotenv.config({ path: envPath, processEnv: mergedEnv });
+ if (result.error && result.error.code !== 'ENOENT') {
+ throw new ConfigError(`could not read ${envPath}: ${result.error.message}`);
+ }
+ if (envFile && result.error) throw new ConfigError(`env file not found: ${envFile}`);
+
+ const device = String(firstDefined(overrides.device, mergedEnv.SINKER_DEVICE, DEFAULTS.device)).toUpperCase();
+ if (!DEVICE_PROFILES[device]) {
+ throw new ConfigError(`device must be one of ${Object.keys(DEVICE_PROFILES).join(', ')} (got "${device}")`);
+ }
+
+ const quality = firstDefined(
+ parseInteger('--quality', overrides.quality, { min: 1, max: 100 }),
+ parseInteger('SINKER_QUALITY', mergedEnv.SINKER_QUALITY, { min: 1, max: 100 }),
+ DEFAULTS.quality,
+ );
+
+ const autoCrop = firstDefined(
+ overrides.autoCrop,
+ parseBool('SINKER_AUTO_CROP', mergedEnv.SINKER_AUTO_CROP),
+ DEFAULTS.autoCrop,
+ );
+
+ const rename = firstDefined(
+ overrides.rename,
+ parseBool('SINKER_RENAME', mergedEnv.SINKER_RENAME),
+ DEFAULTS.rename,
+ );
+
+ const poll = firstDefined(
+ overrides.poll,
+ parseBool('SINKER_WATCH_POLL', mergedEnv.SINKER_WATCH_POLL),
+ DEFAULTS.poll,
+ );
+
+ const backfill = firstDefined(
+ overrides.backfill,
+ parseBool('SINKER_BACKFILL', mergedEnv.SINKER_BACKFILL),
+ DEFAULTS.backfill,
+ );
+
+ const rescanMs = firstDefined(
+ parseInteger('--rescan-ms', overrides.rescanMs, { min: 10_000 }),
+ parseInteger('SINKER_RESCAN_MS', mergedEnv.SINKER_RESCAN_MS, { min: 10_000 }),
+ DEFAULTS.rescanMs,
+ );
+
+ const retryMax = firstDefined(
+ parseInteger('--retry-max', overrides.retryMax, { min: 1, max: 20 }),
+ parseInteger('SINKER_RETRY_MAX', mergedEnv.SINKER_RETRY_MAX, { min: 1, max: 20 }),
+ DEFAULTS.retryMax,
+ );
+
+ const timeoutMs = firstDefined(
+ parseInteger('--timeout-ms', overrides.timeoutMs, { min: 1_000 }),
+ parseInteger('SINKER_TIMEOUT_MS', mergedEnv.SINKER_TIMEOUT_MS, { min: 1_000 }),
+ DEFAULTS.timeoutMs,
+ );
+
+ const loginPort = firstDefined(
+ parseInteger('--port', overrides.loginPort, { min: 1, max: 65535 }),
+ parseInteger('SINKER_LOGIN_PORT', mergedEnv.SINKER_LOGIN_PORT, { min: 1, max: 65535 }),
+ DEFAULTS.loginPort,
+ );
+
+ const bookhiveUrl = String(firstDefined(mergedEnv.BOOKHIVE_URL, BOOKHIVE_URL)).replace(/\/+$/, '');
+
+ const sourceDir = resolvePath(firstDefined(overrides.sourceDir, mergedEnv.SINKER_SOURCE_DIR, DEFAULTS.sourceDir), cwd);
+ const outputDir = resolvePath(firstDefined(overrides.outputDir, mergedEnv.SINKER_OUTPUT_DIR, DEFAULTS.outputDir), cwd);
+ if (sourceDir === outputDir) {
+ throw new ConfigError('the source and output folders must be different');
+ }
+
+ return Object.freeze({
+ cwd,
+ envFile: envFile ?? null,
+
+ handle: optionalString(firstDefined(overrides.handle, mergedEnv.ATPROTO_HANDLE)) ?? null,
+
+ bookhiveUrl,
+ bookhiveDid: String(firstDefined(mergedEnv.BOOKHIVE_DID, BOOKHIVE_DID)).trim(),
+
+ sourceDir,
+ outputDir,
+ oauthDir: resolvePath(firstDefined(mergedEnv.SINKER_OAUTH_DIR, defaultOauthDir(mergedEnv)), cwd),
+ stateDir: resolvePath(firstDefined(mergedEnv.SINKER_STATE_DIR, defaultStateDir(mergedEnv)), cwd),
+
+ device,
+ quality,
+ autoCrop,
+ rename,
+
+ loginPort,
+ loginHost: String(firstDefined(overrides.loginHost, mergedEnv.SINKER_LOGIN_HOST, DEFAULTS.loginHost)),
+ poll,
+ backfill,
+ rescanMs,
+ retryMax,
+ timeoutMs,
+
+ dryRun: Boolean(overrides.dryRun),
+ force: Boolean(overrides.force),
+ retryFailed: Boolean(overrides.retryFailed),
+ verbose: Boolean(overrides.verbose),
+ quiet: Boolean(overrides.quiet),
+ openBrowser: overrides.openBrowser !== false,
+
+ // Effective image conversion options, threaded into convertEpub().
+ convertOptions: Object.freeze({
+ maxWidth: DEVICE_PROFILES[device].width,
+ maxHeight: DEVICE_PROFILES[device].height,
+ quality,
+ autoCrop,
+ }),
+ });
+}
+
+export function describeConfig(config) {
+ return [
+ `source ${config.sourceDir}`,
+ `output ${config.outputDir}`,
+ `device ${DEVICE_PROFILES[config.device].label} (${DEVICE_PROFILES[config.device].width}x${DEVICE_PROFILES[config.device].height})`,
+ `quality ${config.quality}`,
+ `auto-crop ${config.autoCrop ? 'on' : 'off'}`,
+ `rename ${config.rename ? 'on' : 'off'}`,
+ `bookhive ${config.bookhiveUrl} (${config.bookhiveDid})`,
+ `login http://127.0.0.1:${config.loginPort}/callback` +
+ (config.loginHost === '127.0.0.1' ? '' : ` (bind ${config.loginHost})`),
+ `oauth dir ${config.oauthDir}`,
+ `state dir ${config.stateDir}`,
+ ].join('\n');
+}
diff --git a/src/sinker/dedupe.js b/src/sinker/dedupe.js
new file mode 100644
index 0000000..c36936d
--- /dev/null
+++ b/src/sinker/dedupe.js
@@ -0,0 +1,104 @@
+// Matching helpers for "is this book already in the library?".
+//
+// BookHive identifies a personal-library file by its KOReader partial MD5, so
+// a hash match is exact. Compression is not byte-deterministic across runs
+// (JSZip stamps fresh timestamps), so a previously uploaded *compressed* copy
+// can only be recognised by metadata. BookHive parses EPUB metadata as:
+//
+// title = first dc:title, falling back to the file name
+// authors = every dc:creator, joined with ", "
+//
+// which is what `normalizeText` is applied to on both sides.
+
+/** Lowercase, strip diacritics and punctuation, collapse whitespace. */
+export function normalizeText(value) {
+ return String(value ?? '')
+ .normalize('NFKD')
+ .replace(/\p{M}/gu, '')
+ .toLowerCase()
+ .replace(/['\u2018\u2019\u0060\u00b4]/g, '')
+ .replace(/[^\p{L}\p{N}]+/gu, ' ')
+ .trim()
+ .replace(/\s+/g, ' ');
+}
+
+/** Accepts an array of creators or BookHive's ", "-joined string. */
+export function authorsString(authors) {
+ if (Array.isArray(authors)) return authors.filter(Boolean).join(', ');
+ return String(authors ?? '');
+}
+
+/** Stable comparison key for a book: normalized title + normalized authors. */
+export function bookKey({ title, authors } = {}) {
+ return `${normalizeText(title)}::${normalizeText(authorsString(authors))}`;
+}
+
+/**
+ * In-memory index of the remote personal library, kept in sync as uploads
+ * succeed so two files in one run can't both upload the same book.
+ */
+export class LibraryIndex {
+ constructor(books = []) {
+ this.books = [];
+ this.byHash = new Map();
+ this.byKey = new Map();
+ this.byTitle = new Map();
+ for (const book of books) this.add(book);
+ }
+
+ get size() {
+ return this.books.length;
+ }
+
+ add(book) {
+ if (!book) return;
+ this.books.push(book);
+ if (book.contentHash) this.byHash.set(book.contentHash, book);
+
+ const key = bookKey(book);
+ if (!this.byKey.has(key)) this.byKey.set(key, book);
+
+ const title = normalizeText(book.title);
+ if (title) {
+ if (!this.byTitle.has(title)) this.byTitle.set(title, []);
+ this.byTitle.get(title).push(book);
+ }
+ }
+
+ hasHash(contentHash) {
+ return Boolean(contentHash) && this.byHash.has(contentHash);
+ }
+
+ getByHash(contentHash) {
+ return this.byHash.get(contentHash);
+ }
+
+ /**
+ * Find a metadata match. Returns `{ reason, book }` or null.
+ *
+ * Exact title+authors wins. When either side has no authors at all, fall
+ * back to a title-only match — the alternative is silently uploading a
+ * duplicate of every book whose OPF lacks a creator.
+ */
+ find({ title, authors } = {}) {
+ const wantedTitle = normalizeText(title);
+ if (!wantedTitle) return null;
+
+ const exact = this.byKey.get(bookKey({ title, authors }));
+ if (exact) return { reason: 'title-author', book: exact };
+
+ const candidates = this.byTitle.get(wantedTitle) ?? [];
+ if (candidates.length === 0) return null;
+
+ // No creator on the local side: the title is all we have.
+ if (!normalizeText(authorsString(authors))) {
+ return { reason: 'title', book: candidates[0] };
+ }
+
+ // Remote entry has no creator either (common for filename-only uploads).
+ const withoutAuthors = candidates.find(book => !normalizeText(authorsString(book.authors)));
+ if (withoutAuthors) return { reason: 'title', book: withoutAuthors };
+
+ return null;
+ }
+}
diff --git a/src/sinker/hash.js b/src/sinker/hash.js
new file mode 100644
index 0000000..48ed04f
--- /dev/null
+++ b/src/sinker/hash.js
@@ -0,0 +1,89 @@
+// KOReader document identifier (partial MD5).
+//
+// This is the exact value BookHive stores as `contentHash` for a personal
+// library upload (despite the lexicon describing it as SHA-256). Ported from
+// KOReader's util.partialMD5 and BookHive's core/bookMetadata/hash.ts.
+//
+// 12 x 1 KB windows at exponentially increasing offsets, so a byte-identical
+// file produces the same hash no matter where it was uploaded from.
+
+import { createHash } from 'node:crypto';
+import { closeSync, openSync, readSync, statSync } from 'node:fs';
+import { open } from 'node:fs/promises';
+
+const WINDOW = 1024;
+
+/**
+ * Offsets of the twelve sampled windows.
+ *
+ * The original Lua shifts a 32-bit integer: for i = -1 the shift count wraps
+ * to 30, so `1024 << 30` overflows to 0. Expressed directly as `1024 * 4^i`
+ * (i >= 0) with an explicit leading 0, which is the same sequence.
+ */
+export function partialMd5Offsets() {
+ const offsets = [0];
+ for (let i = 0; i < 11; i++) offsets.push(1024 * 4 ** i);
+ return offsets;
+}
+
+/**
+ * KOReader partial MD5 of an in-memory buffer.
+ * @param {Uint8Array} buffer
+ * @returns {string} lowercase hex digest
+ */
+export function partialMd5(buffer) {
+ const hasher = createHash('md5');
+ for (const offset of partialMd5Offsets()) {
+ if (offset >= buffer.length) break;
+ hasher.update(buffer.subarray(offset, Math.min(offset + WINDOW, buffer.length)));
+ }
+ return hasher.digest('hex');
+}
+
+/**
+ * Synchronous variant of {@link partialMd5File}, for the few places where an
+ * existing output file has to be compared before deciding on a name.
+ * @param {string} filePath
+ * @returns {string}
+ */
+export function partialMd5FileSync(filePath) {
+ const hasher = createHash('md5');
+ const size = statSync(filePath).size;
+ const handle = openSync(filePath, 'r');
+ try {
+ const scratch = Buffer.alloc(WINDOW);
+ for (const offset of partialMd5Offsets()) {
+ if (offset >= size) break;
+ const length = Math.min(WINDOW, size - offset);
+ const bytesRead = readSync(handle, scratch, 0, length, offset);
+ hasher.update(scratch.subarray(0, bytesRead));
+ }
+ } finally {
+ closeSync(handle);
+ }
+ return hasher.digest('hex');
+}
+
+/**
+ * The same hash, read straight off disk. At most 12 KB of reads regardless of
+ * file size, so this is cheap enough to run before deciding to compress.
+ * @param {string} filePath
+ * @returns {Promise}
+ */
+export async function partialMd5File(filePath) {
+ const hasher = createHash('md5');
+ const handle = await open(filePath, 'r');
+ try {
+ const stat = await handle.stat();
+ const scratch = Buffer.alloc(WINDOW);
+ for (const offset of partialMd5Offsets()) {
+ if (offset >= stat.size) break;
+ const length = Math.min(WINDOW, stat.size - offset);
+ const { bytesRead } = await handle.read(scratch, 0, length, offset);
+ hasher.update(scratch.subarray(0, bytesRead));
+ }
+ } finally {
+ await handle.close();
+ }
+ return hasher.digest('hex');
+}
diff --git a/src/sinker/log.js b/src/sinker/log.js
new file mode 100644
index 0000000..6708781
--- /dev/null
+++ b/src/sinker/log.js
@@ -0,0 +1,41 @@
+// Tiny timestamped logger. One line per event; `verbose` turns on debug lines
+// (per-image conversion detail) and `quiet` silences info.
+
+function stamp() {
+ const now = new Date();
+ const pad = n => String(n).padStart(2, '0');
+ return `${pad(now.getHours())}:${pad(now.getMinutes())}:${pad(now.getSeconds())}`;
+}
+
+export function createLogger({ verbose = false, quiet = false, out = console.log, err = console.error } = {}) {
+ const write = (stream, level, message) => {
+ stream(`[${stamp()}] ${level ? `${level} ` : ''}${message}`);
+ };
+
+ return {
+ verbose,
+ quiet,
+ info(message) {
+ if (!quiet) write(out, '', message);
+ },
+ /** For output that must never be suppressed (e.g. a login URL). */
+ always(message) {
+ write(out, '', message);
+ },
+ step(message) {
+ if (!quiet) write(out, '→', message);
+ },
+ ok(message) {
+ if (!quiet) write(out, '✓', message);
+ },
+ warn(message) {
+ write(err, 'warning:', message);
+ },
+ error(message) {
+ write(err, 'error:', message);
+ },
+ debug(message) {
+ if (verbose) write(out, 'debug:', message);
+ },
+ };
+}
diff --git a/src/sinker/metadata.js b/src/sinker/metadata.js
new file mode 100644
index 0000000..992ec27
--- /dev/null
+++ b/src/sinker/metadata.js
@@ -0,0 +1,109 @@
+// EPUB metadata in the shape BookHive stores it.
+//
+// Reads dc:title / dc:creator from the package document. Kept separate from
+// rename.js (which only needs the derived filename) because dedupe compares
+// the full creator list, not just the primary author used in the filename.
+
+import JSZip from 'jszip';
+import { basename } from 'node:path';
+
+import { findOPFPath } from '../rename.js';
+import { safeReadText } from '../util.js';
+import { parseXml, byLocalName, getAttrNS } from '../xml.js';
+import { OPF_NS } from '../constants.js';
+
+const ENTITIES = {
+ amp: '&', lt: '<', gt: '>', quot: '"', apos: "'", nbsp: '\u00a0',
+};
+
+function decodeEntities(value) {
+ return String(value ?? '').replace(/&(#x?[0-9a-f]+|[a-z]+);/gi, (match, body) => {
+ if (body[0] === '#') {
+ const code = body[1] === 'x' || body[1] === 'X'
+ ? Number.parseInt(body.slice(2), 16)
+ : Number.parseInt(body.slice(1), 10);
+ return Number.isFinite(code) ? String.fromCodePoint(code) : match;
+ }
+ return ENTITIES[body.toLowerCase()] ?? match;
+ });
+}
+
+function textOf(el) {
+ return decodeEntities(el?.textContent ?? '').replace(/\s+/g, ' ').trim();
+}
+
+/** Regex fallbacks for OPFs that linkedom can't parse. */
+function fallbackMetadata(opfText) {
+ const title = opfText.match(/]*>([\s\S]*?)<\/dc:title>/i)?.[1]
+ ?? opfText.match(/]*>([\s\S]*?)<\/title>/i)?.[1];
+ const authors = [...opfText.matchAll(/]*>([\s\S]*?)<\/dc:creator>/gi)]
+ .map(m => m[1]);
+ return {
+ title: title ? decodeEntities(title).replace(/\s+/g, ' ').trim() : '',
+ authors: authors.map(a => decodeEntities(a).replace(/\s+/g, ' ').trim()).filter(Boolean),
+ };
+}
+
+/**
+ * Extract title / authors from an already-loaded EPUB.
+ *
+ * @param {JSZip} zip
+ * @param {string} [fallbackTitle] used when the OPF has no title, mirroring
+ * BookHive's filename fallback.
+ * @returns {Promise<{title:string, authors:string[], primaryAuthor:string}>}
+ */
+export async function readMetadataFromZip(zip, fallbackTitle = '') {
+ const empty = { title: fallbackTitle, authors: [], primaryAuthor: '' };
+ const opfPath = await findOPFPath(zip);
+ if (!opfPath || !zip.files[opfPath]) return empty;
+
+ const opfText = await safeReadText(zip.files[opfPath]);
+ const { doc, ok } = parseXml(opfText);
+
+ let title = '';
+ let authors = [];
+ let primaryAuthor = '';
+
+ if (ok) {
+ title = textOf(byLocalName(doc, 'title')[0]);
+ const creators = byLocalName(doc, 'creator');
+ authors = creators.map(textOf).filter(Boolean);
+
+ const role = el => (getAttrNS(el, 'role', OPF_NS, ['opf']) || '').toLowerCase();
+ const primary = creators.find(el => role(el) === 'aut')
+ ?? creators.find(el => !role(el));
+ primaryAuthor = textOf(primary)
+ || (primary ? decodeEntities(getAttrNS(primary, 'file-as', OPF_NS, ['opf']) ?? '').trim() : '');
+ }
+
+ if (!title || authors.length === 0) {
+ const fallback = fallbackMetadata(opfText);
+ if (!title) title = fallback.title;
+ if (authors.length === 0) authors = fallback.authors;
+ if (!primaryAuthor && authors.length > 0) primaryAuthor = authors[0];
+ }
+
+ return {
+ title: title || fallbackTitle,
+ authors,
+ primaryAuthor: primaryAuthor || authors[0] || '',
+ };
+}
+
+/**
+ * Extract metadata from raw EPUB bytes.
+ * @param {Buffer} buffer
+ * @param {string} [filePath] used for the title fallback
+ */
+export async function readEpubMetadata(buffer, filePath = '') {
+ const fallbackTitle = filePath
+ ? basename(filePath).replace(/\.epub$/i, '')
+ : '';
+ let zip;
+ try {
+ zip = await JSZip.loadAsync(buffer);
+ } catch {
+ return { title: fallbackTitle, authors: [], primaryAuthor: '' };
+ }
+ return readMetadataFromZip(zip, fallbackTitle);
+}
diff --git a/src/sinker/pipeline.js b/src/sinker/pipeline.js
new file mode 100644
index 0000000..bf4f547
--- /dev/null
+++ b/src/sinker/pipeline.js
@@ -0,0 +1,375 @@
+// Compression + upload pipeline.
+//
+// One file at a time, deliberately: sharp is CPU-bound and BookHive asks
+// clients not to hammer uploads. The order of checks is cheapest-first:
+// ledger -> source hash -> metadata -> compress -> output hash -> upload.
+
+import { existsSync, mkdirSync, readFileSync, readdirSync, statSync, writeFileSync } from 'node:fs';
+import { basename, join } from 'node:path';
+
+import { convertEpub } from '../convert.js';
+import { buildMetadataFilename, reserveAvailableFilename } from '../rename.js';
+import { formatBytes } from '../util.js';
+import { AuthError } from './auth.js';
+import { BookHiveError } from './bookhive.js';
+import { authorsString } from './dedupe.js';
+import { partialMd5, partialMd5File, partialMd5FileSync } from './hash.js';
+import { readEpubMetadata } from './metadata.js';
+import { Ledger, stateFileFor } from './state.js';
+
+const TERMINAL_SKIP = new Set(['uploaded', 'skipped-existing', 'skipped-duplicate']);
+const PERMANENT_FAILURE = new Set(['failed-permanent']);
+
+/** Top-level .epub files, sorted, plus the count of ignored files. */
+export function scanSource(sourceDir) {
+ const epubs = [];
+ let ignored = 0;
+ for (const entry of readdirSync(sourceDir, { withFileTypes: true })) {
+ if (entry.name.startsWith('.')) continue;
+ if (entry.isDirectory()) continue;
+
+ // Dirent types can be UNKNOWN on network filesystems, and symlinked books
+ // should be followed. Fall back to stat() when the dirent doesn't say.
+ let isFile = entry.isFile();
+ if (!isFile) {
+ try { isFile = statSync(join(sourceDir, entry.name)).isFile(); } catch { continue; }
+ }
+ if (!isFile) continue;
+
+ if (/\.epub$/i.test(entry.name)) epubs.push(join(sourceDir, entry.name));
+ else ignored++;
+ }
+ epubs.sort();
+ return { epubs, ignored };
+}
+
+function shouldSkip(entry, stat, config) {
+ if (!entry || config.force) return false;
+ if (entry.size !== stat.size || entry.mtimeMs !== stat.mtimeMs) return false;
+ if (TERMINAL_SKIP.has(entry.status)) return true;
+ if (PERMANENT_FAILURE.has(entry.status) && !config.retryFailed) return true;
+ return false;
+}
+
+/** Small synchronous hash used only for "is this existing output identical?". */
+function partialMd5Sync(filePath) {
+ return partialMd5FileSync(filePath);
+}
+
+function percentChange(before, after) {
+ if (!before) return '0%';
+ const delta = Math.round(((after - before) / before) * 100);
+ return `${delta > 0 ? '+' : ''}${delta}%`;
+}
+
+function chooseOutputName(ctx, desiredName, outputHash) {
+ const candidate = join(ctx.config.outputDir, desiredName);
+
+ // A previous run (or `epub-optimize`) may already have written this exact
+ // output. Reuse the name instead of creating "Title (2).epub" forever.
+ if (existsSync(candidate)) {
+ try {
+ if (partialMd5Sync(candidate) === outputHash) return desiredName;
+ } catch { /* fall through to a fresh name */ }
+ }
+
+ return reserveAvailableFilename(desiredName, ctx.usedOutputNames);
+}
+
+function record(ctx, sourcePath, stat, entry) {
+ if (ctx.config.dryRun) return;
+ ctx.ledger.record(sourcePath, {
+ size: stat.size,
+ mtimeMs: stat.mtimeMs,
+ ...entry,
+ });
+}
+
+/**
+ * Build the per-run context: ledger, output-name bookkeeping, counters.
+ */
+export function createContext({ config, bookhive, logger, did }) {
+ const ledger = new Ledger(stateFileFor(config.stateDir, did), did).load();
+ const usedOutputNames = new Set();
+ if (existsSync(config.outputDir)) {
+ for (const name of readdirSync(config.outputDir)) usedOutputNames.add(name.toLowerCase());
+ }
+
+ return {
+ config,
+ bookhive,
+ ledger,
+ logger,
+ did,
+ usedOutputNames,
+ lastLibraryRefresh: 0,
+ stats: { uploaded: 0, already: 0, failed: 0, ignored: 0, wouldUpload: 0 },
+ };
+}
+
+/**
+ * Process one EPUB: dedupe, compress, save, upload.
+ * @returns {Promise<{status: string, outputName?: string}>}
+ */
+export async function processFile(ctx, sourcePath) {
+ const { config, bookhive, ledger, logger } = ctx;
+ const fileName = basename(sourcePath);
+
+ let stat;
+ try {
+ stat = statSync(sourcePath);
+ } catch (error) {
+ if (error.code === 'ENOENT') return { status: 'gone' };
+ throw error;
+ }
+
+ const previous = ledger.get(sourcePath);
+ if (shouldSkip(previous, stat, config)) {
+ logger.debug(`${fileName}: unchanged (${previous.status})`);
+ return { status: 'unchanged' };
+ }
+
+ logger.step(fileName);
+
+ // 1. Exact-file check, before paying for a compress.
+ const sourceHash = await partialMd5File(sourcePath);
+ if (!config.force) {
+ const existing = await bookhive.getPersonalBook(sourceHash);
+ if (existing) {
+ record(ctx, sourcePath, stat, {
+ status: 'skipped-existing',
+ sourceHash,
+ title: existing.title,
+ authors: existing.authors ?? '',
+ contentHash: existing.contentHash,
+ });
+ ctx.stats.already++;
+ logger.info(` already in BookHive as "${existing.title}" (identical file) — skipped`);
+ return { status: 'skipped-existing' };
+ }
+ }
+
+ // 2. Metadata. BookHive's own parse is dc:title / all dc:creator joined ", ".
+ const sourceBuffer = readFileSync(sourcePath);
+ const metadata = await readEpubMetadata(sourceBuffer, sourcePath);
+ const title = metadata.title || fileName.replace(/\.epub$/i, '');
+ const authors = authorsString(metadata.authors);
+
+ // 3. Metadata check, still before compressing.
+ if (!config.force) {
+ const match = bookhive.library.find({ title, authors });
+ if (match) {
+ record(ctx, sourcePath, stat, {
+ status: 'skipped-duplicate',
+ sourceHash,
+ title,
+ authors,
+ contentHash: match.book.contentHash,
+ matchedBy: match.reason,
+ });
+ ctx.stats.already++;
+ const how = match.reason === 'title-author' ? 'title + author' : 'title';
+ logger.info(` already in BookHive as "${match.book.title}" (${how}) — skipped`);
+ return { status: 'skipped-duplicate' };
+ }
+ }
+
+ // 4. Compress.
+ const hooks = {
+ image: () => {},
+ fix: () => {},
+ warn: message => logger.warn(message),
+ };
+ const { buffer, stats } = await convertEpub(sourceBuffer, config.convertOptions, hooks);
+ const outputHash = partialMd5(buffer);
+
+ // 5. Compressed-file check (catches copies uploaded from another machine,
+ // where the source hash differs but the bytes do not).
+ if (!config.force) {
+ if (bookhive.library.hasHash(outputHash)) {
+ const book = bookhive.library.getByHash(outputHash);
+ record(ctx, sourcePath, stat, {
+ status: 'skipped-existing', sourceHash, outputHash,
+ title, authors, contentHash: outputHash,
+ });
+ ctx.stats.already++;
+ logger.info(` identical compressed file is already in BookHive ("${book?.title ?? title}") — skipped`);
+ return { status: 'skipped-existing' };
+ }
+
+ const existing = await bookhive.getPersonalBook(outputHash);
+ if (existing) {
+ record(ctx, sourcePath, stat, {
+ status: 'skipped-existing', sourceHash, outputHash,
+ title, authors, contentHash: outputHash,
+ });
+ ctx.stats.already++;
+ logger.info(` identical compressed file is already in BookHive — skipped`);
+ return { status: 'skipped-existing' };
+ }
+ }
+
+ // 6. Name + write the archive copy.
+ const desiredName = config.rename
+ ? (buildMetadataFilename(title, metadata.primaryAuthor) || fileName)
+ : fileName;
+ const outputName = chooseOutputName(ctx, desiredName, outputHash);
+ const outputPath = join(config.outputDir, outputName);
+
+ if (!config.dryRun) {
+ mkdirSync(config.outputDir, { recursive: true });
+ let alreadyWritten = false;
+ if (existsSync(outputPath)) {
+ try { alreadyWritten = partialMd5Sync(outputPath) === outputHash; } catch { /* rewrite */ }
+ }
+ if (!alreadyWritten) writeFileSync(outputPath, buffer);
+ }
+
+ logger.info(
+ ` ${formatBytes(stats.originalSize)} → ${formatBytes(stats.newSize)} ` +
+ `(${percentChange(stats.originalSize, stats.newSize)}) → ${outputName}`,
+ );
+
+ if (config.dryRun) {
+ ctx.stats.wouldUpload++;
+ logger.info(' would upload (dry run)');
+ return { status: 'dry-run', outputName };
+ }
+
+ // 7. Upload.
+ const result = await bookhive.upload(buffer, outputName);
+ const storage = result.storageUsedBytes !== undefined && result.storageQuotaBytes
+ ? ` | storage ${formatBytes(result.storageUsedBytes)} / ${formatBytes(result.storageQuotaBytes)}`
+ : '';
+
+ if (result.alreadyExists) {
+ logger.ok(` uploaded as "${outputName}" (BookHive already had it)${storage}`);
+ } else {
+ logger.ok(` uploaded as "${outputName}"${storage}`);
+ }
+
+ // Keep the in-run index accurate even when BookHive returned 409 without a
+ // book view, so a second copy in the same batch is skipped rather than
+ // uploaded and rejected again.
+ if (!result.book) {
+ bookhive.library.add({ contentHash: outputHash, title, authors });
+ }
+
+ record(ctx, sourcePath, stat, {
+ status: 'uploaded',
+ sourceHash,
+ outputHash,
+ outputName,
+ title: result.book?.title ?? title,
+ authors: result.book?.authors ?? authors,
+ contentHash: result.book?.contentHash ?? outputHash,
+ });
+ ctx.stats.uploaded++;
+ return { status: 'uploaded', outputName };
+}
+
+/**
+ * One sync pass.
+ *
+ * @param {object} ctx from createContext()
+ * @param {{paths?: string[]}} [options] explicit file list (watcher events);
+ * omit for a full top-level scan.
+ * @returns {Promise