674 lines
28 KiB
JavaScript
674 lines
28 KiB
JavaScript
'use strict'
|
|
|
|
// The pushing side of a LAN sync: per unit, stream a manifest while it is
|
|
// still being hashed, learn what the receiver has, upload the rest, delete
|
|
// the surplus. Exported as a factory; src/lansync.js hands in its own
|
|
// helpers and state.
|
|
|
|
const http = require('http')
|
|
const os = require('os')
|
|
const fs = require('fs')
|
|
const path = require('path')
|
|
|
|
// Files sent at once: one at a time wastes LAN round trips on small files.
|
|
const UPLOAD_CONCURRENCY = 4
|
|
|
|
// Two lost heartbeats in a row: the receiver is gone (40 s of silence).
|
|
const MAX_MISSED_PINGS = 2
|
|
|
|
// One file, one PUT. A file that has not finished on its own after this long
|
|
// (stalled receiver, lost socket while the OS still reports it open, a file
|
|
// that changed size after the manifest) fails the whole sync instead of
|
|
// hanging it forever. Env override for tests.
|
|
const UPLOAD_TIMEOUT_MS = Number(process.env.PLWL_LANSYNC_UPLOAD_TIMEOUT) || 10 * 60 * 1000
|
|
|
|
// Keeps the receiver alive while this side hashes (minutes of silence); env
|
|
// override for tests.
|
|
const HEARTBEAT_MS = Number(process.env.PLWL_LANSYNC_HEARTBEAT) || 20000
|
|
|
|
// One ping round trip; env override for tests.
|
|
const PING_TIMEOUT_MS = Number(process.env.PLWL_LANSYNC_PING_TIMEOUT) || 10000
|
|
|
|
// How often the sender asks the receiver's popup whether the deletion list is
|
|
// settled, and how long it waits for a human before giving up (nothing is
|
|
// deleted on a timeout, so the safe answer is also the fallback).
|
|
const REVIEW_POLL_MS = Number(process.env.PLWL_LANSYNC_REVIEW_POLL) || 1500
|
|
const REVIEW_WAIT_MS = Number(process.env.PLWL_LANSYNC_REVIEW_WAIT) || 15 * 60 * 1000
|
|
|
|
function createPusher(api) {
|
|
const {
|
|
CATEGORIES,
|
|
normalizeCategories,
|
|
identity,
|
|
rememberDevice,
|
|
forgetDevice,
|
|
httpJson,
|
|
filesPath,
|
|
ownedRoot,
|
|
buildManifestStream,
|
|
listDirs,
|
|
DIRS,
|
|
toPortablePath,
|
|
library,
|
|
wines,
|
|
prefixes
|
|
} = api
|
|
|
|
// Manifest in, upload list out; NDJSON avoids one giant string. The body
|
|
// is written by `produce` WHILE the sender hashes, so the receiver can
|
|
// compare entries as they arrive (both sides hash at the same time).
|
|
function planUnit(device, token, unit, force, meta, produce) {
|
|
// announced up front, so the receiver's popup can say "Apps: 2/5"
|
|
const totals = meta && meta.totals ? `&totals=${encodeURIComponent(JSON.stringify(meta.totals))}` : ''
|
|
// folders a selection kept off the wire must not be pruned on the other side
|
|
const exclude = unit.exclude && unit.exclude.length ? `&exclude=${encodeURIComponent(JSON.stringify(unit.exclude))}` : ''
|
|
return new Promise((resolve, reject) => {
|
|
const route = `/plwl/unit/plan?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}&label=${encodeURIComponent(unit.label || '')}&roots=${encodeURIComponent(JSON.stringify(unit.roots))}${force ? '&force=1' : ''}${totals}${exclude}`
|
|
const req = http.request(
|
|
{
|
|
host: device.address,
|
|
port: device.port,
|
|
path: route,
|
|
method: 'POST',
|
|
headers: { 'content-type': 'application/x-ndjson' },
|
|
timeout: 0
|
|
},
|
|
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 })
|
|
})
|
|
}
|
|
)
|
|
let dead = null
|
|
req.on('error', err => {
|
|
dead = err
|
|
reject(err)
|
|
})
|
|
// ordered writes with backpressure: the receiver reads continuously,
|
|
// but a slow socket must not buffer the whole manifest in memory
|
|
const write = obj => new Promise((resolveWrite, rejectWrite) => {
|
|
if (dead) return rejectWrite(dead)
|
|
const onDrain = () => {
|
|
cleanup()
|
|
resolveWrite()
|
|
}
|
|
const onErr = () => {
|
|
cleanup()
|
|
rejectWrite(dead || new Error('plan-write-failed'))
|
|
}
|
|
const cleanup = () => {
|
|
req.off('drain', onDrain)
|
|
req.off('error', onErr)
|
|
}
|
|
if (req.write(`${JSON.stringify(obj)}\n`)) {
|
|
cleanup()
|
|
return resolveWrite()
|
|
}
|
|
req.once('drain', onDrain)
|
|
req.once('error', onErr)
|
|
})
|
|
produce(write).then(() => req.end(), err => {
|
|
try { req.destroy() } catch {}
|
|
reject(err)
|
|
})
|
|
})
|
|
}
|
|
|
|
function putFile(device, token, unitKey, rel, abs, onBytes) {
|
|
return new Promise((resolve, reject) => {
|
|
const settle = (fn, value) => {
|
|
clearTimeout(timer)
|
|
fn(value)
|
|
}
|
|
// Unit and path travel in the query, percent-encoded: Node refuses any
|
|
// header value holding a code above U+00FF (ERR_INVALID_CHAR), which a
|
|
// real library has (Minecraft pack names, Polish file names), and that
|
|
// refusal used to kill the sync mid-unit.
|
|
const query = `token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unitKey)}&path=${encodeURIComponent(rel)}`
|
|
const req = http.request(
|
|
{
|
|
host: device.address,
|
|
port: device.port,
|
|
path: `/plwl/unit/file?${query}`,
|
|
method: 'PUT',
|
|
headers: {
|
|
'content-type': 'application/octet-stream',
|
|
'content-length': fs.statSync(abs).size
|
|
},
|
|
timeout: 0
|
|
},
|
|
res => {
|
|
res.resume()
|
|
res.on('end', () => settle(resolve, { status: res.statusCode }))
|
|
res.on('aborted', () => settle(reject, new Error('response-aborted')))
|
|
}
|
|
)
|
|
// The timer spans the WHOLE transfer, response included: a stalled
|
|
// receiver or a silently lost socket must fail the sync instead of
|
|
// hanging the popup on the last file. Generous - a file would have to
|
|
// take over ten minutes at 20 MB/s to hit it.
|
|
const timer = setTimeout(() => {
|
|
req.destroy(new Error('upload-timeout'))
|
|
settle(reject, new Error('upload-timeout'))
|
|
}, UPLOAD_TIMEOUT_MS)
|
|
if (timer.unref) timer.unref()
|
|
req.on('error', err => settle(reject, err))
|
|
const stream = fs.createReadStream(abs)
|
|
stream.on('error', err => settle(reject, err))
|
|
// live bytes: a multi-GB file has to move the counter while it flies,
|
|
// not only when its response lands (read = what this side hands over)
|
|
if (onBytes) stream.on('data', chunk => onBytes(chunk.length))
|
|
// pipe ends the request when the file ends; a body shorter than the
|
|
// declared content-length (file changed mid-sync) wedges the receiver
|
|
// - its own idle watchdog breaks that, this side's timer breaks every
|
|
// other silent case
|
|
stream.pipe(req)
|
|
})
|
|
}
|
|
|
|
// --- units ----------------------------------------------------------------
|
|
|
|
// A selection lists unit keys ("app:1", "wine:GE-Proton9-20", "prefix:Main");
|
|
// absent means everything travels. `apps` is the older app-id-only shape.
|
|
function selectionOf(selection) {
|
|
const units = selection && Array.isArray(selection.units) ? new Set(selection.units.map(String)) : null
|
|
const apps = selection && Array.isArray(selection.apps) ? new Set(selection.apps.map(String)) : null
|
|
return {
|
|
active: !!units || !!apps,
|
|
has(key, id) {
|
|
if (units && !units.has(key)) return false
|
|
if (apps && String(key).startsWith('app:') && !apps.has(String(id))) return false
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
|
|
function appUnits(cats, sel) {
|
|
const units = []
|
|
for (const game of library.loadGames()) {
|
|
const category = api.appCategory(game)
|
|
if (!category || !cats.has(category)) continue
|
|
if (!sel.has(`app:${game.id}`, game.id)) continue
|
|
const roots = [`data/icons/${game.id}`]
|
|
if (category === 'apps-portable') {
|
|
// the app's own folder, never apps/<id> guessed from the entry
|
|
const rel = String(toPortablePath(library.appFolder(game)) || '')
|
|
roots.unshift(rel.startsWith('apps/') ? rel : `apps/${game.id}`)
|
|
}
|
|
units.push({ key: `app:${game.id}`, kind: 'app', id: game.id, label: game.name, roots })
|
|
}
|
|
return units
|
|
}
|
|
|
|
// A prefix app a selection leaves out stays on this side: its folder must not
|
|
// travel with the prefix, and the receiver must not prune it as a leftover.
|
|
// Every unselected prefix app inside the unit is found by where its folder
|
|
// really is, not by the prefix name it was registered with. The path is named
|
|
// the way the prefix walk names it: a folder imported through the unified
|
|
// `pfx` link is "drive_c/..." in the manifest, and the stored spelling alone
|
|
// would never match - the folder would travel.
|
|
function prefixExcludes(name, sel) {
|
|
const rootRel = `wineprefixes/${name}`
|
|
const out = []
|
|
for (const game of library.loadGames()) {
|
|
if (api.appCategory(game) !== 'apps-prefix') continue
|
|
if (sel.has(`app:${game.id}`, game.id)) continue
|
|
const folder = library.appFolder(game)
|
|
const inPrefix = prefixes.prefixRelFor(name, folder)
|
|
const candidates = [inPrefix ? `${rootRel}/${inPrefix}` : '', String(toPortablePath(folder) || '')]
|
|
for (const rel of candidates) {
|
|
if (rel === rootRel || !rel.startsWith(rootRel + '/')) continue
|
|
if (!out.includes(rel)) out.push(rel)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
function wineUnits(sel) {
|
|
const units = new Map()
|
|
const add = (root, label) => {
|
|
if (units.has(root)) return
|
|
units.set(root, { key: `wine:${root.slice('wines/'.length)}`, kind: 'wine', id: root.slice('wines/'.length), label: label || root, roots: [root] })
|
|
}
|
|
for (const wine of wines.getWines()) {
|
|
const rel = String(toPortablePath(wine.path) || '')
|
|
const parts = rel.split('/')
|
|
// the unit is the TOP folder under wines/, not the detected build dir
|
|
add(parts[0] === 'wines' && parts[1] ? `wines/${parts[1]}` : `wines/${path.basename(wine.path)}`, wine.label)
|
|
}
|
|
for (const name of listDirs(DIRS.WINES)) add(`wines/${name}`)
|
|
return [...units.values()].filter(unit => sel.has(unit.key, unit.id))
|
|
}
|
|
|
|
function prefixUnits(sel) {
|
|
const names = new Set(prefixes.listPrefixes().prefixes.map(p => p.name))
|
|
for (const name of listDirs(DIRS.PREFIXES)) names.add(name)
|
|
const units = []
|
|
for (const name of names) {
|
|
if (!sel.has(`prefix:${name}`)) continue
|
|
const unit = { key: `prefix:${name}`, kind: 'prefix', id: name, label: name, roots: [`wineprefixes/${name}`] }
|
|
const exclude = prefixExcludes(name, sel)
|
|
if (exclude.length) unit.exclude = exclude
|
|
units.push(unit)
|
|
}
|
|
return units
|
|
}
|
|
|
|
function buildUnits(cats, selection) {
|
|
const sel = selectionOf(selection)
|
|
const units = []
|
|
if (cats.has('apps-portable') || cats.has('apps-prefix')) units.push(...appUnits(cats, sel))
|
|
if (cats.has('wines')) units.push(...wineUnits(sel))
|
|
if (cats.has('prefixes')) units.push(...prefixUnits(sel))
|
|
return units
|
|
}
|
|
|
|
// --- registry payload -----------------------------------------------------
|
|
|
|
// The whole app config travels (paths rewritten to relative form).
|
|
function portableApp(game) {
|
|
return library.portableGame(game)
|
|
}
|
|
|
|
function buildRegistryPayload(categories, selection) {
|
|
const cats = new Set(categories)
|
|
const sel = selectionOf(selection)
|
|
const payload = { categories: [...cats] }
|
|
if (cats.has('apps-portable') || cats.has('apps-prefix')) {
|
|
payload.apps = library
|
|
.loadGames()
|
|
.filter(game => {
|
|
const category = api.appCategory(game)
|
|
if (!category || !cats.has(category)) return false
|
|
return sel.has(`app:${game.id}`, game.id)
|
|
})
|
|
.map(portableApp)
|
|
// A partial list means "only these apps travel"; without the marker the
|
|
// receiver mirrors the category and drops the apps that were left out.
|
|
if (sel.active) payload.appScope = 'selected'
|
|
}
|
|
if (cats.has('wines')) {
|
|
// the receiver mirrors its own builds down to this list and rescans
|
|
payload.wines = wineUnits(sel).map(unit => ({ name: unit.id }))
|
|
if (sel.active) payload.wineScope = 'selected'
|
|
}
|
|
if (cats.has('settings')) {
|
|
// "Sync settings": registry-level values only; opt-in - apps must not
|
|
// silently rewrite the other machine's settings.
|
|
const defs = prefixes.listPrefixes()
|
|
const games = library.loadGames()
|
|
payload.defaults = {
|
|
defaultWine: (library.loadSettings().defaultWine) || 'system',
|
|
defaultPrefix: defs.default || ''
|
|
}
|
|
payload.wineSettings = wines
|
|
.getWines()
|
|
.filter(w => w.type !== 'system')
|
|
.map(w => ({ name: w.id, displayName: w.displayName || '', disabled: !!w.disabled }))
|
|
payload.prefixSettings = {
|
|
default: defs.default || '',
|
|
prefixes: defs.prefixes.map(p => ({ name: p.name, created: p.created, locked: !!p.locked }))
|
|
}
|
|
// an app a selection leaves out is untouched there, config included
|
|
payload.appSettings = games
|
|
.filter(g => sel.has(`app:${g.id}`, g.id))
|
|
.map(g => ({
|
|
id: g.id,
|
|
name: g.name,
|
|
wine: g.wine || 'default',
|
|
prefix: g.prefix || '',
|
|
args: g.args || '',
|
|
env: g.env || {},
|
|
dllOverrides: g.dllOverrides || [],
|
|
disableUmu: !!g.disableUmu,
|
|
// Linux-only app config, part of the app's configuration as well
|
|
isolation: g.isolation,
|
|
forceX11: !!g.forceX11
|
|
}))
|
|
payload.appOrder = games.map(g => g.id)
|
|
}
|
|
if (cats.has('prefixes')) {
|
|
const defs = prefixes.listPrefixes()
|
|
payload.prefixes = {
|
|
default: defs.default,
|
|
prefixes: defs.prefixes
|
|
.filter(p => sel.has(`prefix:${p.name}`))
|
|
.map(p => ({ name: p.name, created: p.created, locked: !!p.locked }))
|
|
}
|
|
if (sel.active) payload.prefixScope = 'selected'
|
|
}
|
|
return payload
|
|
}
|
|
|
|
// --- sync -----------------------------------------------------------------
|
|
|
|
function newSummary(units) {
|
|
return {
|
|
units,
|
|
uploaded: 0,
|
|
skipped: 0,
|
|
deleted: 0,
|
|
// files the receiver asked to keep when the deletion list was reviewed
|
|
kept: 0,
|
|
bytes: 0,
|
|
// size of the synced trees as they now sit on the other device
|
|
sizeBytes: 0,
|
|
// the popup reports "3 games, 1 prefix" instead of file counts
|
|
items: { apps: 0, wines: 0, prefixes: 0 }
|
|
}
|
|
}
|
|
|
|
// Every sync resolves to the same shape, success or failure.
|
|
function syncResult(ok, error, summary) {
|
|
return { ok, error: error || null, summary }
|
|
}
|
|
|
|
// Push to the peer; a rejected code comes back as 'bad-code'.
|
|
async function syncToDevice({ deviceId, device: target, code, categories, force, selection, onProgress }) {
|
|
const fail = (error, summary) => syncResult(false, error, summary || newSummary(0))
|
|
const device = target || api.listDevices().find(d => d.id === deviceId)
|
|
if (!device || !device.address || !device.port) return fail('offline')
|
|
const cats = new Set(normalizeCategories(categories))
|
|
if (!cats.size) return fail('nothing-selected')
|
|
// units are known before anything is read: announce them with the verify
|
|
const units = buildUnits(cats, selection)
|
|
const bucketOf = kind => (kind === 'wine' ? 'wines' : kind === 'prefix' ? 'prefixes' : 'apps')
|
|
const totals = { apps: 0, wines: 0, prefixes: 0, settings: 0 }
|
|
for (const unit of units) totals[bucketOf(unit.kind)] += 1
|
|
if (cats.has('settings')) totals.settings = 1
|
|
const progress = data => {
|
|
if (onProgress) onProgress(data)
|
|
}
|
|
|
|
const summary = newSummary(units.length)
|
|
const finished = { apps: 0, wines: 0, prefixes: 0 }
|
|
// planned bytes so far, so the popup can show "1.2 GB / 3.4 GB"
|
|
let plannedBytes = 0
|
|
let itemsSeen = 0
|
|
// files from previous units: the popup counter is cumulative
|
|
let allDone = 0
|
|
let allTotal = 0
|
|
// the bar tracks the current phase only: hashing starts empty
|
|
let barBytes = 0
|
|
let barTotal = 0
|
|
const snapshot = (phase, unit, done, total) => {
|
|
const data = {
|
|
stage: phase,
|
|
phase,
|
|
kind: (unit && unit.kind) || '',
|
|
id: (unit && unit.id) || '',
|
|
key: (unit && unit.key) || '',
|
|
unit: (unit && unit.label) || '',
|
|
label: (unit && unit.label) || '',
|
|
name: (unit && unit.label) || '',
|
|
bytes: summary.bytes,
|
|
totalBytes: plannedBytes,
|
|
units: summary.units,
|
|
unitsDone: itemsSeen,
|
|
items: { ...summary.items },
|
|
itemsDone: { ...finished },
|
|
totals: { ...totals },
|
|
barBytes,
|
|
barTotalBytes: barTotal
|
|
}
|
|
// only the copy phase carries a file count, so the popup never jumps back
|
|
if (Number.isFinite(done) && Number.isFinite(total)) {
|
|
data.done = done
|
|
data.total = total
|
|
}
|
|
return data
|
|
}
|
|
// Snapshot per file, forward a few times per second (phase changes
|
|
// always). The snapshot is built when the gate flushes, not when it is
|
|
// pushed: a throttled byte count must not be one gate window old.
|
|
const emitNow = data => progress(typeof data === 'function' ? data() : data)
|
|
const gate = typeof api.progressGate === 'function' ? api.progressGate(emitNow) : emitNow
|
|
const emit = (phase, unit, done, total, force) => gate(() => snapshot(phase, unit, done, total), !!force)
|
|
// consent may wait on a human: say so before anything else happens
|
|
emit('wait', null, null, null, true)
|
|
|
|
let verified
|
|
try {
|
|
// longer than the receiver's incoming-request timeout (popup wait)
|
|
verified = await httpJson(device, 'POST', '/plwl/verify', {
|
|
json: { deviceId: identity().deviceId, name: os.hostname(), code: String(code == null ? '' : code).trim().toUpperCase(), totals },
|
|
timeout: 90000
|
|
})
|
|
} catch (err) {
|
|
return fail(String((err && err.message) || err))
|
|
}
|
|
if (!verified.data || !verified.data.ok) {
|
|
const error = (verified.data && verified.data.error) || `http-${verified.status}`
|
|
// a rotated code drops the pairing; a plain 'rejected' keeps it
|
|
if (error === 'auth-failed' || error === 'bad-code') forgetDevice(device.id || deviceId)
|
|
return fail(error)
|
|
}
|
|
// An older peer prunes inside its commit, with no list to approve: units
|
|
// must not travel there at all. Settings alone are additive and fine.
|
|
if (!verified.data.review && units.length) return fail('old-peer')
|
|
const token = verified.data.token
|
|
// paired now: the next sync needs no typing
|
|
rememberDevice(device.id || deviceId, code, device.name)
|
|
|
|
// The heartbeat keeps the receiver's popup alive while this side hashes
|
|
// (minutes of silence). Consecutive ping failures mean the peer dropped
|
|
// (crash, sleep, lost network): without this a dead receiver hangs the
|
|
// sync until the next request notices, which may be never.
|
|
let missedPings = 0
|
|
const abort = { dead: false, reject: null }
|
|
const abortPromise = new Promise((resolve, reject) => { abort.reject = reject })
|
|
abortPromise.catch(() => {}) // only awaited via Promise.race below
|
|
const heartbeat = setInterval(() => {
|
|
httpJson(device, 'POST', `/plwl/ping?token=${encodeURIComponent(token)}`, { json: {}, timeout: PING_TIMEOUT_MS })
|
|
.then(() => { missedPings = 0 })
|
|
.catch(() => {
|
|
missedPings++
|
|
if (missedPings >= MAX_MISSED_PINGS && !abort.dead) {
|
|
abort.dead = true
|
|
abort.reject(new Error('receiver-lost'))
|
|
}
|
|
})
|
|
}, HEARTBEAT_MS)
|
|
if (heartbeat.unref) heartbeat.unref()
|
|
|
|
const end = async () => {
|
|
try {
|
|
await httpJson(device, 'POST', `/plwl/end?token=${encodeURIComponent(token)}`, { json: {} })
|
|
} catch {}
|
|
}
|
|
// The receiver's popup is a human, so the sender waits - visibly, and with
|
|
// a deadline. A timeout deletes nothing on purpose.
|
|
const waitForReview = async (device, token, review, emit, abort) => {
|
|
emit('review', null, null, null, true)
|
|
const deadline = Date.now() + REVIEW_WAIT_MS
|
|
while (Date.now() < deadline) {
|
|
if (abort.dead) throw new Error('receiver-lost')
|
|
try {
|
|
const res = await httpJson(device, 'GET', `/plwl/review?token=${encodeURIComponent(token)}&id=${encodeURIComponent(review.id)}`, { timeout: 20000 })
|
|
const data = res && res.data ? res.data : {}
|
|
if (data.state === 'done' || data.state === 'timeout' || data.state === 'skipped') {
|
|
return { deleted: Number(data.deleted) || 0, kept: Number(data.kept) || 0 }
|
|
}
|
|
} catch {}
|
|
await new Promise(resolve => setTimeout(resolve, REVIEW_POLL_MS))
|
|
}
|
|
return { deleted: 0, kept: 0 }
|
|
}
|
|
const runUnits = async () => {
|
|
for (const unit of units) {
|
|
if (abort.dead) throw new Error('receiver-lost')
|
|
const roots = unit.roots.filter(ownedRoot)
|
|
if (!roots.length) continue
|
|
const kind = bucketOf(unit.kind)
|
|
summary.items[kind] += 1
|
|
itemsSeen += 1
|
|
barBytes = 0
|
|
barTotal = 0
|
|
emit('scan', unit, null, null, true)
|
|
// The manifest streams WHILE this side hashes its tail: the receiver
|
|
// compares entries as they arrive, so both devices hash at the same
|
|
// time. Once this side is done hashing, the popup keeps a full bar
|
|
// and the waiting text until the plan answers (a cached sender waits
|
|
// for a receiver that still has to hash).
|
|
const entries = []
|
|
let writes = Promise.resolve()
|
|
// A read failure means this manifest is not the sender's whole tree:
|
|
// tell the receiver, so it never prunes what only looked missing.
|
|
const incomplete = () => {
|
|
writes = writes.then(() => write({ unitIncomplete: true }))
|
|
}
|
|
const plan = await planUnit(device, token, { ...unit, roots }, !!force, { totals }, async write => {
|
|
await buildManifestStream(roots, !!force, {
|
|
// an app the selection left out does not travel with its prefix
|
|
skip: unit.exclude,
|
|
// the receiver sizes its compare bar from this, before any hashing
|
|
onTotal: totalBytes => {
|
|
writes = writes.then(() => write({ unitBytes: totalBytes }))
|
|
},
|
|
onEntry: entry => {
|
|
entries.push(entry)
|
|
writes = writes.then(() => write(entry))
|
|
},
|
|
onIncomplete: incomplete,
|
|
onProgress: (hashedDone, hashedTotal, hashedBytes, totalBytes) => {
|
|
barBytes = hashedBytes
|
|
barTotal = totalBytes
|
|
emit('hash', unit, null, null)
|
|
}
|
|
})
|
|
await writes
|
|
// this side is done hashing; the receiver may still be comparing
|
|
barTotal = barTotal || 1
|
|
barBytes = barTotal
|
|
emit('wait', unit, null, null, true)
|
|
})
|
|
if (abort.dead) throw new Error('receiver-lost')
|
|
if (!plan.data || !plan.data.ok) {
|
|
// a path makes a layout refusal actionable (which folder to reconcile)
|
|
const detail = plan.data && Array.isArray(plan.data.paths) && plan.data.paths.length ? `:${plan.data.paths[0]}` : ''
|
|
throw new Error(((plan.data && plan.data.error) || `http-${plan.status}`) + detail)
|
|
}
|
|
// every file in the unit, including the ones already identical there
|
|
for (const entry of entries) if (entry.type === 'file') summary.sizeBytes += entry.size || 0
|
|
const upload = new Set(plan.data.upload || [])
|
|
const same = new Set(plan.data.same || [])
|
|
const filePaths = new Set(entries.filter(e => e.type === 'file').map(e => e.path))
|
|
let sameFiles = 0
|
|
for (const p of same) if (filePaths.has(p)) sameFiles++
|
|
summary.skipped += sameFiles
|
|
const pending = []
|
|
for (const entry of entries) {
|
|
if (entry.type !== 'file') continue
|
|
if (!upload.has(entry.path)) continue // identical -> the receiver keeps its copy
|
|
pending.push(entry)
|
|
plannedBytes += entry.size || 0
|
|
}
|
|
// a few files in flight at once: small files cost round trips; the
|
|
// bar counts only bytes that move and starts empty for the copy
|
|
if (pending.length) {
|
|
barBytes = 0
|
|
barTotal = pending.reduce((sum, entry) => sum + (entry.size || 0), 0)
|
|
emit('copy', unit, allDone, allTotal + pending.length, true)
|
|
let failed = null
|
|
let done = 0
|
|
const send = async entry => {
|
|
if (failed || abort.dead) return
|
|
const res = await putFile(device, token, unit.key, entry.path, filesPath(entry.path), delta => {
|
|
summary.bytes += delta
|
|
barBytes += delta
|
|
emit('copy', unit, allDone + done, allTotal + pending.length)
|
|
})
|
|
if (res.status !== 200) {
|
|
failed = new Error(`upload-failed-${res.status}`)
|
|
return
|
|
}
|
|
done++
|
|
summary.uploaded++
|
|
// the byte counters already moved while the file was read; the
|
|
// completion still has to push the file count next
|
|
emit('copy', unit, allDone + done, allTotal + pending.length)
|
|
}
|
|
if (typeof api.mapLimit === 'function') await api.mapLimit(pending, UPLOAD_CONCURRENCY, send)
|
|
else for (const entry of pending) await send(entry)
|
|
if (failed) throw failed
|
|
// A dead receiver made `send` skip files; committing now would prune
|
|
// exactly the files that never arrived (the commit itself could even
|
|
// succeed while only the heartbeats were failing).
|
|
if (abort.dead) throw new Error('receiver-lost')
|
|
}
|
|
emit('copy', unit, allDone + pending.length, allTotal + pending.length, true)
|
|
// the totals count a unit's files only once they are on the wire
|
|
allDone += pending.length
|
|
allTotal += pending.length
|
|
const commit = await httpJson(device, 'POST', `/plwl/unit/commit?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}`, { json: {} })
|
|
// a unit that did not commit (unknown unit, failed prune) must not be
|
|
// counted as done while the sync moves on to the registry
|
|
if (!commit.data || !commit.data.ok) throw new Error((commit.data && commit.data.error) || `commit-${commit.status}`)
|
|
if (commit.data.deleted) summary.deleted += commit.data.deleted
|
|
finished[kind] += 1
|
|
// a finished unit leaves a finished bar behind, even one that sent none
|
|
barTotal = barTotal || 1
|
|
barBytes = barTotal
|
|
emit('commit', unit, null, null, true)
|
|
}
|
|
emit('registry', null, null, null, true)
|
|
if (abort.dead) throw new Error('receiver-lost')
|
|
const applied = await httpJson(device, 'POST', `/plwl/apply?token=${encodeURIComponent(token)}`, { json: buildRegistryPayload([...cats], selection), timeout: 60000 })
|
|
if (!applied.data || !applied.data.ok) throw new Error((applied.data && applied.data.error) || `http-${applied.status}`)
|
|
if (applied.data.deleted) summary.deleted += applied.data.deleted
|
|
// Deletions wait for the receiver's approval; nothing is removed on a
|
|
// timeout, so a lost popup can only ever keep too much, never too little.
|
|
if (applied.data.review && applied.data.review.id) {
|
|
const outcome = await waitForReview(device, token, applied.data.review, emit, abort)
|
|
summary.deleted += outcome.deleted || 0
|
|
summary.kept += outcome.kept || 0
|
|
}
|
|
// a sync with nothing to send still ends with a full bar
|
|
barTotal = barTotal || 1
|
|
barBytes = barTotal
|
|
emit('done', null, allDone, allTotal, true)
|
|
await end()
|
|
}
|
|
try {
|
|
await Promise.race([runUnits(), abortPromise])
|
|
} catch (err) {
|
|
abort.dead = true
|
|
await end()
|
|
return fail(String((err && err.message) || err), summary)
|
|
} finally {
|
|
clearInterval(heartbeat)
|
|
}
|
|
return syncResult(true, null, summary)
|
|
}
|
|
|
|
// Everything a selection can name, in the same ids buildUnits uses - the UI
|
|
// must never derive a unit key on its own.
|
|
function items() {
|
|
const all = { active: false, has: () => true }
|
|
return {
|
|
apps: library
|
|
.loadGames()
|
|
.filter(game => api.appCategory(game))
|
|
.map(game => ({
|
|
key: `app:${game.id}`,
|
|
name: game.name || game.id,
|
|
category: api.appCategory(game),
|
|
prefix: game.prefix || ''
|
|
})),
|
|
wines: wineUnits(all).map(unit => ({ key: unit.key, name: unit.label })),
|
|
prefixes: prefixUnits(all).map(unit => ({ key: unit.key, name: unit.label }))
|
|
}
|
|
}
|
|
|
|
// putFile is exposed for the stall tests (it is the piece that must never
|
|
// leave a request half-open)
|
|
return { syncToDevice, buildUnits, buildRegistryPayload, items, putFile, CATEGORIES }
|
|
}
|
|
|
|
module.exports = createPusher
|