'use strict' // LAN sync: instances beacon and push to each other over HTTP. The sender's // tree wins; extra receiver files are deleted. const http = require('http') const crypto = require('crypto') const os = require('os') const path = require('path') const fs = require('fs') const fsp = fs.promises const readline = require('readline') const { DIRS, readJson, writeJson, toPortablePath } = require('./paths') const { rewireSymlinks, restoreExecutableBits } = require('./symlinks') const { progressGate, mapLimit } = require('./fsutil') const { createHashCache } = require('./hashcache') const { createIdentity, sameCode } = require('./lansync-identity') const { createManifestBuilder } = require('./lansync-manifest') const { createDiscovery } = require('./lansync-discovery') const library = require('./library') const wines = require('./wines') const prefixes = require('./prefixes') const { isPrefixRel } = prefixes const CONFIG = path.join(DIRS.DATA, 'lansync.json') // The beacon doubles as liveness: a device drops off a few seconds after close. const DISCOVERY_PORT = Number(process.env.PLWL_LANSYNC_PORT) || 47821 const BEACON_TAG = 'plwl-lansync' const PROTOCOL = 1 // 2.5 s beacon against 6 s TTL: one lost datagram still fits. const BEACON_MS = 2500 const DEVICE_TTL_MS = 6000 // Cost + one-in-flight limit make guessing the 8 character code pointless. const VERIFY_DELAY_MS = 2000 // Session token TTL; env override for tests. const TOKEN_TTL_MS = Number(process.env.PLWL_LANSYNC_TOKEN_TTL) || 30 * 60 * 1000 const SESSION_IDLE_MS = 5 * 60 * 1000 // Receiver consent window; the sender's verify timeout must exceed it. const INCOMING_TIMEOUT_MS = Number(process.env.PLWL_LANSYNC_INCOMING_TIMEOUT) || 60 * 1000 // The only trees a sync may touch; the rest of files/ is local state. const SYNC_ROOTS = ['apps/', 'wines/', 'wineprefixes/', 'data/icons/'] // Sync categories plus lenient aliases for older senders. const CATEGORIES = ['apps-portable', 'apps-prefix', 'wines', 'prefixes', 'settings'] const CATEGORY_ALIASES = { // the whole app library: portable + prefix-import launcher entries apps: ['apps-portable', 'apps-prefix'], app: ['apps-portable', 'apps-prefix'], appsportable: ['apps-portable'], portableapps: ['apps-portable'], portable: ['apps-portable'], appsprefix: ['apps-prefix'], prefixapps: ['apps-prefix'], appsimportedfromprefix: ['apps-prefix'], wine: ['wines'], wines: ['wines'], wineversions: ['wines'], wineprefix: ['prefixes'], wineprefixes: ['prefixes'], prefixes: ['prefixes'], prefix: ['prefixes'], // launcher settings: defaults, wine labels, prefix registry, app configs settings: ['settings'], syncsettings: ['settings'], defaults: ['settings'], default: ['settings'], defaultsettings: ['settings'] } // An app entry's category: portable imports own their apps/ folder. function appCategory(game) { if (!game || typeof game !== 'object') return null if (game.importSource === 'prefix' || game.mode === 'prefix') return 'apps-prefix' if (game.imported === true || game.mode === 'copy' || game.mode === 'move') return 'apps-portable' return null } // An app entry's files, read off its stored path; outside the data = not synced. function classifyApp(app) { if (!app || (typeof app !== 'string' && typeof app !== 'object')) return null const raw = typeof app === 'string' ? app : app.path || app.exePath || app.workdir // external entry: nothing to sync if (!raw || typeof raw !== 'string') return 'external' const value = raw.replace(/\\/g, '/') const outside = value.startsWith('..') || path.isAbsolute(raw) || /^[a-z]:\//i.test(value) || /^[a-z][a-z0-9+.-]*:\/\//i.test(value) if (isPrefixRel(value)) return 'prefix' if (/^apps\//.test(value)) return 'portable' return outside ? 'external' : null } function normalizeCategories(list) { const out = new Set() for (const raw of Array.isArray(list) ? list : []) { const aliases = CATEGORY_ALIASES[String(raw || '').toLowerCase().replace(/[^a-z]+/g, '')] for (const alias of aliases || []) out.add(alias) } return [...out] } // --- identity --------------------------------------------------------------- const identityApi = createIdentity(CONFIG) const { identity, rotateDeviceId, rotateSyncCode, knownDevice, knownPeers, rememberDevice, forgetDevice, makeDeviceId, makeSyncCode } = identityApi // --- files: hashing, paths, manifests --------------------------------------- // A few digests in flight keep the disk busy on many small files. const HASH_CONCURRENCY = Math.max(4, Math.min(16, (os.cpus() || []).length || 4)) // Digests cached per file (size + mtime); one store per launcher state. const hashStore = createHashCache(DIRS.DATA) const hashFile = hashStore.hashFile // Protocol paths are FILES-relative; escapes are rejected (untrusted). function filesPath(rel) { if (typeof rel !== 'string' || !rel || rel.includes('\0') || path.isAbsolute(rel)) return null const root = path.resolve(DIRS.FILES) const abs = path.resolve(root, rel) if (abs !== root && !abs.startsWith(root + path.sep)) return null return abs } // A sync unit may only own one of the sync trees, nothing else. function ownedRoot(rel) { const clean = String(rel || '').replace(/\\/g, '/').replace(/\/+$/, '') if (!clean || clean.split('/').includes('..')) return null return SYNC_ROOTS.some(prefix => clean.startsWith(prefix)) ? clean : null } // --- manifests (the sender's hashing) --------------------------------------- const manifest = createManifestBuilder({ filesPath, filesRoot: DIRS.FILES, hash: hashStore, mapLimit, concurrency: HASH_CONCURRENCY }) const buildManifest = manifest.buildManifest // --- HTTP helpers (shared with the pusher) ---------------------------------- // One-shot JSON request; timeout 0 = none (remote hashing is slow). function httpJson(device, method, route, { json, timeout = 15000 } = {}) { return new Promise((resolve, reject) => { const req = http.request( { host: device.address, port: device.port, path: route, method, headers: json ? { 'content-type': 'application/json' } : {}, timeout }, res => { const chunks = [] res.on('data', chunk => chunks.push(chunk)) res.on('end', () => { let data = null try { data = JSON.parse(Buffer.concat(chunks).toString('utf8')) } catch {} resolve({ status: res.statusCode, data }) }) } ) req.on('timeout', () => req.destroy(new Error('timeout'))) req.on('error', reject) if (json) req.write(JSON.stringify(json)) req.end() }) } // Accept a discovered device record, a bare port (tests) or 'host:port'. function targetOf(target) { if (typeof target === 'number') return { address: '127.0.0.1', port: target } if (typeof target === 'string') { const [host, port] = target.split(':') return { address: host || '127.0.0.1', port: Number(port) || 0 } } if (target && typeof target === 'object') { return { address: target.address || target.host || target.hostname || '127.0.0.1', port: Number(target.port) || 0 } } return null } // Trade the sync code for a token; a rejected code is never stored. async function verifyPeer(target, code) { // { host, port, code } in one object, or a target plus a separate code if (target && typeof target === 'object' && code === undefined && target.code !== undefined) code = target.code const device = targetOf(target) if (!device || !device.port) return { ok: false, error: 'bad-target' } let res try { res = await httpJson(device, 'POST', '/plwl/verify', { json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase() }, timeout: 20000 }) } catch (err) { return { ok: false, error: String((err && err.message) || err) } } if (!res.data || !res.data.ok) return { ok: false, error: (res.data && res.data.error) || `http-${res.status}` } const peer = res.data.device || { id: res.data.deviceId, name: res.data.name } const peerId = (target && target.deviceId) || (target && target.id) || (peer && peer.id) if (peerId) rememberDevice(peerId, code, (peer && peer.name) || (target && target.name)) return { ok: true, token: res.data.token, deviceId: peer && peer.id, device: peer, peer: peer } } // --- receiving side --------------------------------------------------------- function sendJson(res, status, data) { if (res.writableEnded) return const body = JSON.stringify(data) res.writeHead(status, { 'content-type': 'application/json', 'content-length': Buffer.byteLength(body) }) res.end(body) } function readJsonBody(req, limit = 16 * 1024 * 1024) { return new Promise(resolve => { let size = 0 const chunks = [] req.on('data', chunk => { size += chunk.length if (size > limit) { req.destroy() return } chunks.push(chunk) }) req.on('end', () => { try { resolve(JSON.parse(Buffer.concat(chunks).toString('utf8'))) } catch { resolve(null) } }) req.on('error', () => resolve(null)) }) } function readNdjson(req, onEntry) { return new Promise(resolve => { const rl = readline.createInterface({ input: req, crlfDelay: Infinity }) let chain = Promise.resolve() rl.on('line', line => { if (!line.trim()) return // entries are handled strictly in order: a handler may be async (the // streamed plan compares/hashes while the body keeps arriving), but // the response must never reorder the manifest let parsed = null try { parsed = JSON.parse(line) } catch { return } chain = chain.then(() => onEntry(parsed)).catch(() => {}) }) const done = () => chain.then(resolve, () => resolve()) rl.on('close', done) rl.on('error', done) }) } function inlineSame(entry) { return entry && typeof entry.path === 'string' && (entry.type === 'dir' || entry.type === 'link' || entry.type === 'file') } function withinRoots(rel, roots) { return roots.some(root => rel === root || rel.startsWith(root + '/')) } async function entrySame(entry) { const abs = filesPath(entry.path) if (!abs) return false if (entry.type === 'dir') { try { const st = await fsp.lstat(abs) return st.isDirectory() && !st.isSymbolicLink() } catch { return false } } if (entry.type === 'link') { try { return (await fsp.readlink(abs)) === entry.target } catch { return false } } if (!entry.hash) return false // forced sync (or unreadable file) -> replace it // A size mismatch alone settles it - no need to hash. let st try { st = fs.lstatSync(abs) } catch { return false } if (!st.isFile() || st.size !== entry.size) return false let same = false try { same = (await hashStore.cachedHash(entry.path, abs, st.size, st.mtimeMs || 0)) === entry.hash } catch { return false } if (!same) return false // Same bytes, wrong mode: repair locally instead of re-downloading. if (typeof entry.mode === 'number' && (st.mode & 0o777) !== entry.mode) applyMode(abs, entry.mode) return true } function ensureDir(abs) { let st = null try { st = fs.lstatSync(abs) } catch {} if (st && st.isDirectory() && !st.isSymbolicLink()) return if (st) removeAbs(abs) try { fs.mkdirSync(abs, { recursive: true }) } catch {} } // After a unit lands, rewire links and restore modes, or it cannot start. async function healUnitRoots(roots) { const healed = { links: 0, executables: 0 } for (const rel of Array.isArray(roots) ? roots : []) { const abs = filesPath(ownedRoot(rel)) if (!abs) continue try { // a symlinked unit root is not ours to walk; its own unit heals it if (!fs.lstatSync(abs).isDirectory()) continue healed.links += await rewireSymlinks(abs, [DIRS.FILES]) if (String(rel).startsWith('wines/')) healed.executables += await restoreExecutableBits(abs) } catch {} } return healed } function ensureLink(abs, target) { if (typeof target !== 'string' || !target) return let current = null try { current = fs.readlinkSync(abs) } catch {} if (current === target) return removeAbs(abs) try { fs.mkdirSync(path.dirname(abs), { recursive: true }) fs.symlinkSync(target, abs) } catch {} } // rmSync(force) "succeeds" on missing paths; only real removals count. function removeAbs(abs) { try { if (!fs.lstatSync(abs)) return false } catch { return false } try { fs.rmSync(abs, { recursive: true, force: true }) return true } catch { return false } } function removeTree(rel) { const abs = filesPath(rel) if (!abs) return 0 try { if (!fs.lstatSync(abs)) return 0 // nothing to delete -> do not count it } catch { return 0 } return removeAbs(abs) ? 1 : 0 } function removeOrphans(dir, keep) { let names = [] try { names = fs.readdirSync(dir) } catch { return 0 } let removed = 0 for (const name of names) { if (keep.has(name)) continue if (removeAbs(path.join(dir, name))) removed++ } return removed } // Registry folder names; symlinked folders count as the link they are. function listDirs(dir) { const out = [] let entries try { entries = fs.readdirSync(dir, { withFileTypes: true }) } catch { return out } for (const entry of entries) { if (entry.isDirectory()) { out.push(entry.name) continue } if (!entry.isSymbolicLink()) continue try { if (fs.statSync(path.join(dir, entry.name)).isDirectory()) out.push(entry.name) } catch {} } return out } // Delete everything under a unit root that is not in the sender's manifest. async function pruneTree(abs, rel, keep) { // A symlinked unit root: readdir would delete the target's real files. let rootStat = null try { rootStat = await fsp.lstat(abs) } catch { return 0 } if (!rootStat.isDirectory()) return 0 let entries try { entries = await fsp.readdir(abs, { withFileTypes: true }) } catch { return 0 } let removed = 0 for (const entry of entries) { const childAbs = path.join(abs, entry.name) const childRel = `${rel}/${entry.name}` if (!keep.has(childRel)) { if (removeAbs(childAbs)) removed++ continue } if (entry.isDirectory()) removed += await pruneTree(childAbs, childRel, keep) } return removed } // Modes go on after the bytes; a build without the exec bit is unusable. function applyMode(abs, mode) { if (typeof mode !== 'number' || !mode) return try { fs.chmodSync(abs, mode & 0o777) } catch {} } function writeStream(abs, stream, onBytes) { return new Promise((resolve, reject) => { try { fs.mkdirSync(path.dirname(abs), { recursive: true }) } catch {} const tmp = `${abs}.plwl-sync-tmp` const out = fs.createWriteStream(tmp) stream.on('error', err => { removeAbs(tmp) reject(err) }) out.on('error', err => { removeAbs(tmp) reject(err) }) out.on('finish', () => { try { if (fs.lstatSync(abs).isDirectory()) removeAbs(abs) } catch {} // nothing there (the common case) or a link - both fine to rename over try { fs.renameSync(tmp, abs) resolve(out.bytesWritten || 0) } catch (err) { // a rename race must not undo the transfer removeAbs(tmp) resolve(0) } }) if (onBytes) { // flowing mode delivers every chunk to each listener stream.on('data', chunk => onBytes(chunk.length)) } stream.pipe(out) }) } // --- registries on the receiving side --------------------------------------- function applyApps(payload, cats) { if (!cats.has('apps-portable') && !cats.has('apps-prefix')) return 0 const incoming = (Array.isArray(payload.apps) ? payload.apps : []).filter(g => g && g.id) const incomingIds = new Set(incoming.map(g => g.id)) const existing = library.loadGames() const kept = [] const dropped = new Set() for (const game of existing) { const category = appCategory(game) if (category && cats.has(category)) { if (!incomingIds.has(game.id)) dropped.add(game.id) continue // replaced by the sender's copy } kept.push(game) } // App settings travel as stored; playtime only grows (higher wins). const merged = [...kept] for (const g of incoming) { const prev = existing.find(x => x.id === g.id) const hasPlay = Number.isFinite(Number(g.playSeconds)) const playSeconds = hasPlay ? Math.max(Number(prev && prev.playSeconds) || 0, Number(g.playSeconds)) : prev && prev.playSeconds const lastPlayed = Math.max(Number(prev && prev.lastPlayed) || 0, Number(g.lastPlayed) || 0) merged.push({ ...g, playSeconds, lastPlayed }) // the per-app playtime file wins over the registry; write it too if (hasPlay) library.writePlaySeconds(g.id, playSeconds) } library.saveGames(merged) let removed = 0 for (const id of dropped) removed += removeTree(`data/icons/${id}`) const iconsRemoved = removed if (cats.has('apps-portable')) { // files/apps holds only portable imports; the rest is mirror leftover removed += removeOrphans(DIRS.APPS, incomingIds) } return removed } async function applyWines(payload) { const names = new Set((Array.isArray(payload.wines) ? payload.wines : []).map(w => w && w.name).filter(Boolean)) const removed = removeOrphans(DIRS.WINES, names) // the registry derives from disk; a rescan picks up the synced builds await wines.refreshWines() return removed } function applyPrefixes(payload, useSenderDefault) { const sender = payload.prefixes && typeof payload.prefixes === 'object' ? payload.prefixes : {} const incoming = (Array.isArray(sender.prefixes) ? sender.prefixes : []).filter(p => p && p.name) const existing = prefixes.listPrefixes() const byName = new Map(existing.prefixes.map(p => [p.name, p])) const keep = new Map(incoming.map(p => [p.name, p])) // never delete the receiver's own Main prefix if (!keep.has(prefixes.MAIN)) { keep.set(prefixes.MAIN, byName.get(prefixes.MAIN) || { name: prefixes.MAIN, created: Date.now() }) } const removed = removeOrphans(DIRS.PREFIXES, new Set(keep.keys())) const list = [...keep.values()].map(p => ({ ...p, locked: p.name === prefixes.MAIN ? true : !!p.locked })) // the default prefix moves only with synced default settings const wanted = useSenderDefault ? sender.default : existing.default const def = keep.has(wanted) ? wanted : prefixes.MAIN prefixes.savePrefixes({ default: def, prefixes: list }) return removed } // Wine display names/disabled flags by build id; empty name clears the label. function applyWineSettings(payload) { const incoming = Array.isArray(payload.wineSettings) ? payload.wineSettings : [] let applied = 0 for (const entry of incoming) { if (!entry || !entry.name) continue const id = String(entry.name) const wine = wines.getWine(id) if (!wine || wine.type === 'system') continue let changed = false const wanted = typeof entry.displayName === 'string' ? entry.displayName.trim() : '' if (wanted !== (wine.displayName || '')) { // a sender that never renamed the build clears the custom label here if (wanted) wines.setWineDisplayName(id, wanted) else wines.clearWineDisplayName(id) changed = true } if (entry.disabled !== undefined && !!entry.disabled !== !!wine.disabled) { wines.setWineDisabled(id, !!entry.disabled) changed = true } if (changed) applied++ } return applied } // Prefix registry names only; registers sender entries as known here. function applyPrefixSettings(payload) { const sender = payload.prefixSettings && typeof payload.prefixSettings === 'object' ? payload.prefixSettings : null if (!sender) return 0 const incoming = (Array.isArray(sender.prefixes) ? sender.prefixes : []).filter(p => p && p.name) if (!incoming.length) return 0 const defs = prefixes.listPrefixes() const known = new Set(defs.prefixes.map(p => p.name)) let added = 0 for (const entry of incoming) { const name = String(entry.name) if (known.has(name)) continue known.add(name) defs.prefixes.push({ name, created: Number(entry.created) || Date.now(), locked: name === prefixes.MAIN ? true : !!entry.locked }) added++ } if (added) prefixes.savePrefixes(defs) return added } // Per-app config: patch existing apps only; unknown ids are skipped. function applyAppSettings(payload) { const incoming = Array.isArray(payload.appSettings) ? payload.appSettings : [] if (!incoming.length) return 0 const byId = new Map(incoming.filter(a => a && a.id).map(a => [String(a.id), a])) const knownPrefixes = new Set(prefixes.listPrefixes().prefixes.map(p => p.name)) let patched = 0 for (const game of library.loadGames()) { const from = byId.get(game.id) if (!from) continue const patch = {} if (typeof from.name === 'string' && from.name && from.name !== game.name) patch.name = from.name if (from.wine !== undefined) { const wanted = String(from.wine || 'default') if ((wanted === 'default' || wanted === 'system' || wines.getWine(wanted)) && wanted !== (game.wine || 'default')) patch.wine = wanted } if (from.prefix !== undefined && game.importSource !== 'prefix') { const wanted = String(from.prefix || '') if ((!wanted || knownPrefixes.has(wanted)) && wanted !== (game.prefix || '')) patch.prefix = wanted } if (typeof from.args === 'string' && from.args !== (game.args || '')) patch.args = from.args if (from.env && typeof from.env === 'object' && JSON.stringify(from.env) !== JSON.stringify(game.env || {})) patch.env = from.env if (Array.isArray(from.dllOverrides) && JSON.stringify(from.dllOverrides) !== JSON.stringify(game.dllOverrides || [])) patch.dllOverrides = from.dllOverrides if (from.disableUmu !== undefined && !!from.disableUmu !== !!game.disableUmu) patch.disableUmu = !!from.disableUmu if (Object.keys(patch).length) { library.updateGame(game.id, patch) patched++ } } return patched } // Library order: entries unknown to the sender stay at the end. function applyAppOrder(payload) { const order = Array.isArray(payload.appOrder) ? payload.appOrder.filter(id => typeof id === 'string' && id) : [] if (!order.length) return 0 const current = library.loadGames().map(g => g.id) if (current.length === order.length && current.every((id, i) => id === order[i])) return 0 library.reorderGames(order) return order.length } // "Sync settings" payload; order matters (names -> defaults -> configs -> // order). function applySettings(payload, cats) { if (!cats.has('settings')) return null return { prefixes: applyPrefixSettings(payload), wines: applyWineSettings(payload), apps: applyAppSettings(payload), order: applyAppOrder(payload), defaults: applyDefaults(payload) } } // Launcher defaults; only values that exist here are taken. function applyDefaults(payload) { const wanted = payload.defaults && typeof payload.defaults === 'object' ? payload.defaults : {} const applied = {} if (wanted.defaultWine) { const res = wines.setDefaultWine(String(wanted.defaultWine)) applied.defaultWine = res && res.ok ? String(wanted.defaultWine) : null } if (wanted.defaultPrefix) { const defs = prefixes.listPrefixes() if (defs.prefixes.some(p => p.name === wanted.defaultPrefix)) { prefixes.setDefaultPrefix(String(wanted.defaultPrefix)) applied.defaultPrefix = String(wanted.defaultPrefix) } else { applied.defaultPrefix = null } } return applied } // --- HTTP server (what other devices talk to) ------------------------------- function createServerState() { return { tokens: new Map(), verifying: new Set(), pending: new Map(), session: null } } // The user's popup answer; unknown/expired ids are a no-op. let serverState = null function respondIncoming(requestId, accept) { const key = String(requestId || '') const resolve = serverState && serverState.pending ? serverState.pending.get(key) : null if (!resolve) return false serverState.pending.delete(key) resolve(accept ? 'accept' : 'reject') return true } // The receiving-side counter a unit feeds on ("app:" etc.). const UNIT_KINDS = { app: 'apps', wine: 'wines', prefix: 'prefixes' } // Progress of a sync this device receives, for the receiver's popup. function emitReceive(hooks, session, stage, extra = {}) { if (!hooks || !hooks.onReceive) return try { hooks.onReceive({ stage, phase: stage, from: session.name || '', deviceId: session.deviceId || '', unit: extra.unit || '', label: extra.unit || '', done: extra.done == null ? session.received || 0 : extra.done, total: extra.total == null ? session.planned || 0 : extra.total, bytes: session.bytes || 0, totalBytes: session.plannedBytes || 0, // items: announced units, itemsDone: written units, totals: sender's plan items: { ...(session.items || {}) }, itemsDone: { ...(session.itemsDone || {}) }, totals: { ...(session.totals || {}) }, // the bar shows the current phase, so hashing starts empty barBytes: session.barBytes || 0, barTotalBytes: session.barTotal || 0, deleted: extra.deleted || 0 }) } catch {} } function activeSession(state) { if (!state.session) return null if (Date.now() - state.session.lastAt > SESSION_IDLE_MS) { state.session = null return null } return state.session } function authorize(state, token) { const session = token && state.tokens.get(token) if (!session) return null if (Date.now() > session.expires) { state.tokens.delete(token) return null } session.lastAt = Date.now() // The expiry slides with every request: hashing can outlive any fixed TTL. session.expires = Date.now() + TOKEN_TTL_MS if (state.session && state.session.token === token) state.session.lastAt = session.lastAt return session } function createHttpServer(state, hooks) { const server = http.createServer(async (req, res) => { let url = null try { url = new URL(req.url, 'http://plwl.local') } catch {} if (!url) return sendJson(res, 400, { ok: false, error: 'bad-request' }) const route = url.pathname try { if (route === '/plwl/info' && req.method === 'GET') { const me = identity() return sendJson(res, 200, { ok: true, protocol: PROTOCOL, id: me.deviceId, name: os.hostname() }) } if (route === '/plwl/verify' && req.method === 'POST') return handleVerify(state, req, res, hooks) const session = authorize(state, url.searchParams.get('token')) if (!session) return sendJson(res, 401, { ok: false, error: 'unauthorized' }) if (route === '/plwl/ping' && req.method === 'POST') return handlePing(session, res, hooks) if (route === '/plwl/unit/plan' && req.method === 'POST') return handlePlan(state, session, url, req, res, hooks) if (route === '/plwl/unit/file' && req.method === 'PUT') return handlePut(session, url, req, res, hooks) if (route === '/plwl/unit/commit' && req.method === 'POST') return handleCommit(session, url, res, hooks) if (route === '/plwl/apply' && req.method === 'POST') return handleApply(session, req, res, hooks) if (route === '/plwl/end' && req.method === 'POST') return handleEnd(state, session, res, hooks) sendJson(res, 404, { ok: false, error: 'not-found' }) } catch (err) { sendJson(res, 500, { ok: false, error: String((err && err.message) || err) }) } }) // a prefix manifest can take minutes to hash on the receiving side server.requestTimeout = 0 server.headersTimeout = 60000 server.timeout = 0 return server } async function handleVerify(state, req, res, hooks) { const payload = await readJsonBody(req) if (!payload || !payload.deviceId) return sendJson(res, 400, { ok: false, error: 'bad-request' }) const me = identity() const from = String(payload.deviceId) // one verification per device at a time if (state.verifying.has(from)) return sendJson(res, 429, { ok: false, error: 'busy' }) const running = activeSession(state) if (running && running.deviceId !== from) return sendJson(res, 429, { ok: false, error: 'busy' }) // one unanswered consent popup at a time if (state.pending.size) return sendJson(res, 429, { ok: false, error: 'busy' }) if (hooks.isBusy && hooks.isBusy(from)) return sendJson(res, 429, { ok: false, error: 'busy' }) state.verifying.add(from) try { await new Promise(resolve => setTimeout(resolve, VERIFY_DELAY_MS)) if (!sameCode(payload.code, me.syncCode)) return sendJson(res, 403, { ok: false, error: 'auth-failed', reason: 'wrong-code' }) // the sender may announce totals up front ("0/5" in the popup) const announced = payload.totals && typeof payload.totals === 'object' ? payload.totals : {} // a correct code is not consent; without a UI to ask, the gate stays open if (typeof hooks.onIncoming === 'function') { const requestId = crypto.randomBytes(12).toString('hex') const decision = await new Promise(resolve => { const timer = setTimeout(() => { state.pending.delete(requestId) resolve('timeout') }, INCOMING_TIMEOUT_MS) if (timer.unref) timer.unref() state.pending.set(requestId, resolve) try { hooks.onIncoming({ requestId, deviceId: from, name: String(payload.name || ''), totals: { apps: Number(announced.apps) || 0, wines: Number(announced.wines) || 0, prefixes: Number(announced.prefixes) || 0, settings: Number(announced.settings) || 0 } }) } catch { resolve('reject') // no window to answer with -> never auto-accept silently } }) if (decision !== 'accept') return sendJson(res, 403, { ok: false, error: decision === 'reject' ? 'rejected' : 'incoming-timeout' }) } const token = crypto.randomBytes(24).toString('hex') const session = { token, deviceId: from, name: String(payload.name || ''), expires: Date.now() + TOKEN_TTL_MS, lastAt: Date.now(), units: new Map(), // popup counters: files, bytes, sender's totals learned per unit counted: new Set(), received: 0, planned: 0, bytes: 0, plannedBytes: 0, items: { apps: 0, wines: 0, prefixes: 0 }, itemsDone: { apps: 0, wines: 0, prefixes: 0 }, barBytes: 0, barTotal: 0, totals: { apps: Number(announced.apps) || 0, wines: Number(announced.wines) || 0, prefixes: Number(announced.prefixes) || 0 } } state.tokens.set(token, session) state.session = session emitReceive(hooks, session, 'start') sendJson(res, 200, { ok: true, token, deviceId: me.deviceId, name: os.hostname(), device: { id: me.deviceId, name: os.hostname() } }) } finally { state.verifying.delete(from) } } async function handlePlan(state, session, url, req, res, hooks) { const key = url.searchParams.get('unit') || '' const label = url.searchParams.get('label') || '' let roots = [] try { roots = JSON.parse(url.searchParams.get('roots') || '[]') } catch {} const owned = (Array.isArray(roots) ? roots : []).map(ownedRoot).filter(Boolean) if (!key || !owned.length) return sendJson(res, 400, { ok: false, error: 'bad-unit' }) // comparing is its own phase: the bar restarts for the copy phase const announce = progressGate((data, force) => emitReceive(hooks, session, 'compare', data)) // the unit counts from the start of its comparison, not from its commit const kind = UNIT_KINDS[String(key).split(':')[0]] if (kind && !session.counted.has(key)) { session.counted.add(key) session.items[kind] = (session.items[kind] || 0) + 1 } // a force sync compares nothing: every entry counts as different const forced = url.searchParams.get('force') === '1' // the sender's totals, so the popup can say "2/5" from the start try { const totals = JSON.parse(url.searchParams.get('totals') || 'null') if (totals && typeof totals === 'object') { session.totals = { apps: Number(totals.apps) || 0, wines: Number(totals.wines) || 0, prefixes: Number(totals.prefixes) || 0 } } } catch {} // The manifest is compared as it streams in (both sides hash at the same // time). The sender sends its unit total first, so the bar starts with its // full width; files missing from the sender's total compare as the usual // entries - the bar just grows when that happens. const entries = [] const results = new Map() let unitTotalBytes = 0 let compared = 0 let comparedBytes = 0 const compareOne = async entry => { const same = forced ? false : await entrySame(entry) results.set(entry.path, same === true) compared++ comparedBytes += entry.size || 0 session.barBytes = comparedBytes // the sender's announced total (or a provisional full bar until it // arrives) - clamped at the top by the renderer, never over 100% session.barTotal = unitTotalBytes || 1 // no file counting yet: dirs/links are compared too announce({ unit: label }) } // files compared concurrently: the sender streams its manifest, this side // keeps at most HASH_CONCURRENCY digests in flight (the sender's hashing // queue is the wider throttle) const jobs = [] let inFlight = 0 const settleOldest = () => Promise.race(jobs.splice(0, 1)) await readNdjson(req, async entry => { // the sender announces its unit total before hashing starts (a control // record, no path): it carries the compare bar's width, so it must be // read before the entry filter drops it if (entry.unitBytes != null) { unitTotalBytes = Number(entry.unitBytes) || 0 session.barTotal = unitTotalBytes || 1 return } if (!inlineSame(entry)) return entries.push(entry) if (!withinRoots(entry.path, owned)) return if (forced) { await compareOne(entry) return } // the sender's hashing queue is the throttle on its side; this side's // own concurrency limit keeps a saturated disk from stacking up here if (inFlight >= HASH_CONCURRENCY) await settleOldest() inFlight++ const job = compareOne(entry).finally(() => { inFlight-- }) jobs.push(job) }) await Promise.all(jobs) hashStore.save() const allowed = entries.filter(entry => withinRoots(entry.path, owned)) const identical = allowed.map(entry => results.get(entry.path) === true) // the mode each file has to end up with on this side const modes = new Map() for (const entry of allowed) if (entry.type === 'file' && typeof entry.mode === 'number') modes.set(entry.path, entry.mode) const paths = new Set() const upload = [] const same = [] let uploadBytes = 0 allowed.forEach((entry, index) => { const abs = filesPath(entry.path) if (!abs) return paths.add(entry.path) if (identical[index]) { same.push(entry.path) return } if (entry.type === 'dir') ensureDir(abs) else if (entry.type === 'link') ensureLink(abs, entry.target) else { upload.push(entry.path) uploadBytes += entry.size || 0 } }) session.units.set(key, { roots: owned, paths, label, modes }) // this unit's files/bytes, reported as they arrive session.planned += upload.length session.plannedBytes += uploadBytes // the compare is finished: report it full, then reset for the copy phase // (the first byte that arrives emits the new bar) session.barTotal = unitTotalBytes || session.barTotal || 1 session.barBytes = session.barTotal announce({ unit: label }, true) session.unitWritten = 0 session.unitLabel = label // named up front, so file events can label it session.barBytes = 0 session.barTotal = uploadBytes sendJson(res, 200, { ok: true, upload, same, entries: paths.size }) } // One PUT, one file. The sender declares a content-length up front; if the // body stops short of it (sender crashed, lost link, file changed size after // the manifest), the request would otherwise stay open forever and wedge // this file, the session and the sync. Any arriving byte re-arms the timer, // so a slow honest transfer never trips it. Env override for tests. const PUT_IDLE_MS = Number(process.env.PLWL_LANSYNC_PUT_IDLE) || 60000 async function handlePut(session, url, req, res, hooks) { // unit and path travel as headers: constant per request const key = req.headers['x-plwl-unit'] || url.searchParams.get('unit') || '' const rel = req.headers['x-plwl-path'] || url.searchParams.get('path') || '' const unit = session.units.get(key) const abs = filesPath(rel) if (!unit || !abs || !unit.paths.has(rel)) { req.resume() return sendJson(res, 400, { ok: false, error: 'bad-path' }) } // report bytes while the file is still arriving const announce = progressGate((data, force) => emitReceive(hooks, session, 'copy', data)) // a ping must not flash "waiting" while transfers are running session.writing = (session.writing || 0) + 1 let idleTimer = null const armIdle = () => { if (idleTimer) clearTimeout(idleTimer) idleTimer = setTimeout(() => { // the write's stream 'error' handler removes the tmp file; the sender's // request errors out and its own timer/heartbeat handles the rest req.destroy(new Error('body-stalled')) }, PUT_IDLE_MS) if (idleTimer.unref) idleTimer.unref() } armIdle() try { await writeStream(abs, req, delta => { armIdle() session.bytes = (session.bytes || 0) + delta session.unitWritten = (session.unitWritten || 0) + delta session.barBytes = session.unitWritten announce({ unit: session.unitLabel || '' }) }) // modes do not travel with the bytes; restore from the manifest applyMode(abs, unit.modes.get(rel)) session.received = (session.received || 0) + 1 sendJson(res, 200, { ok: true }) announce({ unit: session.unitLabel || '' }, true) } catch (err) { sendJson(res, 500, { ok: false, error: String((err && err.message) || err) }) } finally { if (idleTimer) clearTimeout(idleTimer) session.writing-- } } async function handleCommit(session, url, res, hooks) { const key = url.searchParams.get('unit') || '' const unit = session.units.get(key) if (!unit) return sendJson(res, 400, { ok: false, error: 'no-unit' }) let removed = 0 for (const root of unit.roots) { const abs = filesPath(root) if (abs) removed += await pruneTree(abs, root, unit.paths) } // rewire links only once the whole tree is there try { await healUnitRoots(unit.roots) } catch {} session.units.delete(key) session.deleted = (session.deleted || 0) + removed // a finished unit completes its "2/5" slice const kind = UNIT_KINDS[String(key).split(':')[0]] if (kind) session.itemsDone[kind] = (session.itemsDone[kind] || 0) + 1 // a finished unit leaves a finished bar behind session.barTotal = session.barTotal || 1 session.barBytes = session.barTotal sendJson(res, 200, { ok: true, deleted: removed }) emitReceive(hooks, session, 'commit', { unit: session.unitLabel || '', deleted: removed }) } async function handleApply(session, req, res, hooks) { const payload = await readJsonBody(req) if (!payload) return sendJson(res, 400, { ok: false, error: 'bad-request' }) // sender categories go through the aliases (older PLWL says "defaults") const cats = new Set(normalizeCategories(payload.categories)) const appsDeleted = applyApps(payload, cats) const winesDeleted = cats.has('wines') ? await applyWines(payload) : 0 const prefixesDeleted = cats.has('prefixes') ? applyPrefixes(payload, cats.has('settings')) : 0 // after the registries, so synced builds/prefixes are usable as defaults const settings = applySettings(payload, cats) const removed = appsDeleted + winesDeleted + prefixesDeleted emitReceive(hooks, session, 'registry', { deleted: removed }) if (hooks.onApplied) { hooks.onApplied({ from: session.name, deviceId: session.deviceId, categories: [...cats], files: session.received || 0, bytes: session.bytes || 0, items: { ...(session.items || {}) }, deleted: removed, settings, defaults: settings ? settings.defaults : null }) } sendJson(res, 200, { ok: true, deleted: removed, apps: { deleted: appsDeleted }, wines: { deleted: winesDeleted }, prefixes: { deleted: prefixesDeleted }, settings, defaults: settings ? settings.defaults : null }) } function handleEnd(state, session, res, hooks) { state.tokens.delete(session.token) if (state.session && state.session.token === session.token) state.session = null // a finished sync leaves a finished bar, even one that moved no files session.barTotal = session.barTotal || 1 session.barBytes = session.barTotal sendJson(res, 200, { ok: true }) emitReceive(hooks, session, 'done', { deleted: session.deleted || 0 }) } // Liveness ping: hashing can take minutes; re-arms the popup watchdog. The // event must repeat on every ping (it is what keeps the popup's watchdog // armed), but never while a file is being written: a copy in flight reports // itself and must not flash the waiting text over the receiving phase. function handlePing(session, res, hooks) { session.lastAt = Date.now() if (!session.writing) emitReceive(hooks, session, 'wait') sendJson(res, 200, { ok: true }) } // --- discovery -------------------------------------------------------------- const discovery = createDiscovery({ identity, tag: BEACON_TAG, protocol: PROTOCOL, port: DISCOVERY_PORT, beaconMs: BEACON_MS, deviceTtlMs: DEVICE_TTL_MS, getPort: () => httpPort }) const startDiscovery = discovery.start const stopDiscovery = discovery.stop const listDevices = discovery.listDevices // --- service ---------------------------------------------------------------- let httpPort = 0 let httpServer = null let serverHooks = {} let listening = null // promise resolved once the receiving server is up // Bind the receiving server; resolves with { port, close() }. function startServer(options = {}) { if (options && typeof options === 'object') serverHooks = { ...serverHooks, ...options } if (listening) return listening const state = createServerState() serverState = state // respondIncoming() answers the popups of THIS server httpServer = createHttpServer(state, serverHooks) httpServer.on('error', () => {}) const port = Number(options && options.port) || 0 listening = new Promise(resolve => { httpServer.listen(port, '0.0.0.0', () => { const address = httpServer && httpServer.address() httpPort = address && typeof address === 'object' ? address.port : 0 resolve({ ok: true, port: httpPort, deviceId: identity().deviceId, server: httpServer, close: () => stopServer() }) }) }) return listening } function stopServer() { // throttled mid-sync saves may have skipped the last digests hashStore.save(true) if (httpServer) { try { httpServer.close() } catch {} httpServer = null } // a waiting popup would never be answered: resolve it as declined if (serverState && serverState.pending) { for (const resolve of serverState.pending.values()) resolve('reject') serverState.pending.clear() } serverState = null listening = null httpPort = 0 } function start(hooks = {}) { serverHooks = { ...serverHooks, ...hooks } return startServer().then(info => { startDiscovery(serverHooks.onDevices) return info }) } function stop() { stopDiscovery() stopServer() } function serverPort() { return httpPort } function getState() { const me = identity() return { deviceId: me.deviceId, syncCode: me.syncCode, hash: hashStore.HASH_ALGO, port: httpPort, devices: listDevices() } } // The pusher is built with this module's helpers, so launcher states never mix. let pusher = null function getPusher() { if (!pusher) { pusher = require('./lansyncpush')({ CATEGORIES, normalizeCategories, appCategory, identity, rememberDevice, forgetDevice, httpJson, filesPath, ownedRoot, buildManifest, buildManifestStream: manifest.buildManifestStream, listDirs, listDevices, mapLimit, progressGate, hashFile, DIRS, toPortablePath, library, wines, prefixes }) } return pusher } function syncTo(payload) { return getPusher().syncToDevice(payload) } // Push to one peer; options: peer record, code, categories, force. function runSync(options = {}, onProgress) { const target = options.port || options.host || options.address ? targetOf(options) : null const deviceId = options.deviceId || options.id // paired before -> reuse the stored code, no typing needed const stored = (deviceId && knownDevice(deviceId)) || null const result = getPusher().syncToDevice({ device: target, deviceId, code: options.code || (target && target.code) || (stored && stored.code), categories: options.categories || options.kinds || CATEGORIES, force: !!options.force, onProgress: [onProgress, options.onProgress, options.onEvent, options.progress].find(fn => typeof fn === 'function') }) return result } const exported = { start, startServer, stop, stopServer, closeServer: stopServer, close: stopServer, httpJson, verifyPeer, runSync, appCategory, classifyApp, CATEGORIES, normalizeCategories, serverPort, getServerPort: serverPort, getState, listDevices, identity, loadIdentity: identity, makeDeviceId, makeSyncCode, rotateDeviceId, rotateSyncCode, rerollDeviceId: rotateDeviceId, rerollSyncCode: rotateSyncCode, knownDevice, knownPeers, rememberDevice, forgetDevice, respondIncoming, sameCode, HASH_ALGO: hashStore.HASH_ALGO, hashFile, hashCacheStats: hashStore.stats, resetHashCacheStats: hashStore.resetStats, forgetHashCache: hashStore.forget, clearHashCache: hashStore.clear, HASH_CONCURRENCY, progressGate, filesPath, ownedRoot, buildManifest, buildManifestStream: manifest.buildManifestStream, mapLimit, listDirs, syncTo } module.exports = exported