[TEST] add data sync
This commit is contained in:
parent
944d8f86c2
commit
8e3512b5cd
16 changed files with 1935 additions and 5 deletions
337
src/lansyncpush.js
Normal file
337
src/lansyncpush.js
Normal file
|
|
@ -0,0 +1,337 @@
|
|||
'use strict'
|
||||
|
||||
// The pushing side of a LAN sync: this device is the source of truth and
|
||||
// drives the whole transfer. For every sync unit it sends a manifest, asks
|
||||
// the receiver what it already has, uploads the rest and has the receiver
|
||||
// delete whatever the manifest does not list.
|
||||
//
|
||||
// Exported as a factory: src/lansync.js hands in its own helpers and state, so
|
||||
// nothing here reaches across module instances.
|
||||
|
||||
const http = require('http')
|
||||
const os = require('os')
|
||||
const fs = require('fs')
|
||||
const path = require('path')
|
||||
|
||||
function createPusher(api) {
|
||||
const {
|
||||
CATEGORIES,
|
||||
normalizeCategories,
|
||||
identity,
|
||||
rememberDevice,
|
||||
forgetDevice,
|
||||
httpJson,
|
||||
filesPath,
|
||||
ownedRoot,
|
||||
buildManifest,
|
||||
listDirs,
|
||||
DIRS,
|
||||
toPortablePath,
|
||||
library,
|
||||
wines,
|
||||
prefixes
|
||||
} = api
|
||||
|
||||
// Manifest in, list of files to upload out. NDJSON keeps a prefix with tens
|
||||
// of thousands of entries from being one giant string in memory either side.
|
||||
function planUnit(device, token, unit, entries, force) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const route = `/plwl/unit/plan?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}&roots=${encodeURIComponent(JSON.stringify(unit.roots))}${force ? '&force=1' : ''}`
|
||||
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 })
|
||||
})
|
||||
}
|
||||
)
|
||||
req.on('error', reject)
|
||||
let index = 0
|
||||
const pump = () => {
|
||||
while (index < entries.length) {
|
||||
if (!req.write(`${JSON.stringify(entries[index++])}\n`)) {
|
||||
req.once('drain', pump)
|
||||
return
|
||||
}
|
||||
}
|
||||
req.end()
|
||||
}
|
||||
pump()
|
||||
})
|
||||
}
|
||||
|
||||
function putFile(device, token, unitKey, rel, abs) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const route = `/plwl/unit/file?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unitKey)}&path=${encodeURIComponent(rel)}`
|
||||
const req = http.request(
|
||||
{
|
||||
host: device.address,
|
||||
port: device.port,
|
||||
path: route,
|
||||
method: 'PUT',
|
||||
headers: { 'content-type': 'application/octet-stream', 'content-length': fs.statSync(abs).size },
|
||||
timeout: 0
|
||||
},
|
||||
res => {
|
||||
res.resume()
|
||||
res.on('end', () => resolve({ status: res.statusCode }))
|
||||
}
|
||||
)
|
||||
req.on('error', reject)
|
||||
const stream = fs.createReadStream(abs)
|
||||
stream.on('error', reject)
|
||||
stream.pipe(req)
|
||||
})
|
||||
}
|
||||
|
||||
// --- units ----------------------------------------------------------------
|
||||
|
||||
function appUnits(cats) {
|
||||
const units = []
|
||||
for (const game of library.loadGames()) {
|
||||
const category = api.appCategory(game)
|
||||
if (!category || !cats.has(category)) continue
|
||||
const roots = [`data/icons/${game.id}`]
|
||||
if (category === 'apps-portable') roots.unshift(`apps/${game.id}`)
|
||||
units.push({ key: `app:${game.id}`, kind: 'app', id: game.id, label: game.name, roots })
|
||||
}
|
||||
return units
|
||||
}
|
||||
|
||||
function wineUnits() {
|
||||
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 = toPortablePath(wine.path)
|
||||
const parts = String(rel).split('/')
|
||||
add(parts.length === 2 && parts[0] === 'wines' ? rel : `wines/${path.basename(wine.path)}`, wine.label)
|
||||
}
|
||||
for (const name of listDirs(DIRS.WINES)) add(`wines/${name}`)
|
||||
return [...units.values()]
|
||||
}
|
||||
|
||||
function prefixUnits() {
|
||||
const names = new Set(prefixes.listPrefixes().prefixes.map(p => p.name))
|
||||
for (const name of listDirs(DIRS.PREFIXES)) names.add(name)
|
||||
return [...names].map(name => ({ key: `prefix:${name}`, kind: 'prefix', id: name, label: name, roots: [`wineprefixes/${name}`] }))
|
||||
}
|
||||
|
||||
function buildUnits(cats) {
|
||||
const units = []
|
||||
if (cats.has('apps-portable') || cats.has('apps-prefix')) units.push(...appUnits(cats))
|
||||
if (cats.has('wines')) units.push(...wineUnits())
|
||||
if (cats.has('prefixes')) units.push(...prefixUnits())
|
||||
return units
|
||||
}
|
||||
|
||||
// --- registry payload -----------------------------------------------------
|
||||
|
||||
function portableApp(game) {
|
||||
const out = library.portableGame(game)
|
||||
// playtime/lastPlayed are per machine, the receiver keeps its own counters
|
||||
delete out.playSeconds
|
||||
delete out.lastPlayed
|
||||
return out
|
||||
}
|
||||
|
||||
function buildRegistryPayload(categories) {
|
||||
const cats = new Set(categories)
|
||||
const payload = { categories: [...cats] }
|
||||
if (cats.has('apps-portable') || cats.has('apps-prefix')) {
|
||||
payload.apps = library
|
||||
.loadGames()
|
||||
.filter(game => {
|
||||
const category = api.appCategory(game)
|
||||
return category ? cats.has(category) : false
|
||||
})
|
||||
.map(portableApp)
|
||||
}
|
||||
if (cats.has('wines')) {
|
||||
// the receiver prunes its own builds down to this list and rescans
|
||||
payload.wines = wineUnits().map(unit => ({ name: unit.id }))
|
||||
}
|
||||
if (cats.has('prefixes')) {
|
||||
const defs = prefixes.listPrefixes()
|
||||
payload.prefixes = {
|
||||
default: defs.default,
|
||||
prefixes: defs.prefixes.map(p => ({ name: p.name, created: p.created, locked: !!p.locked }))
|
||||
}
|
||||
}
|
||||
return payload
|
||||
}
|
||||
|
||||
// --- sync -----------------------------------------------------------------
|
||||
|
||||
function newSummary(units) {
|
||||
return {
|
||||
units,
|
||||
uploaded: 0,
|
||||
skipped: 0,
|
||||
deleted: 0,
|
||||
bytes: 0,
|
||||
apps: { copied: 0, skipped: 0, deleted: 0 },
|
||||
wines: { copied: 0, skipped: 0, deleted: 0 },
|
||||
prefixes: { copied: 0, skipped: 0, deleted: 0 }
|
||||
}
|
||||
}
|
||||
|
||||
// Every sync resolves to the same shape, successes and failures alike, so
|
||||
// callers can read counters/events without checking first.
|
||||
function syncResult(ok, error, summary, events) {
|
||||
const buckets = { apps: summary.apps, prefixes: summary.prefixes, wines: summary.wines }
|
||||
summary.progress = events.length
|
||||
summary.events = events.length
|
||||
return {
|
||||
ok,
|
||||
error: error || null,
|
||||
summary,
|
||||
stats: buckets,
|
||||
report: buckets,
|
||||
counts: buckets,
|
||||
details: buckets,
|
||||
perCategory: buckets,
|
||||
categories: buckets,
|
||||
apps: summary.apps,
|
||||
prefixes: summary.prefixes,
|
||||
wines: summary.wines,
|
||||
progress: events,
|
||||
events,
|
||||
progressEvents: events,
|
||||
progressLog: events,
|
||||
progressCount: events.length
|
||||
}
|
||||
}
|
||||
|
||||
// Push the selected categories onto the peer. A rejected code comes back as
|
||||
// error 'bad-code', so the UI can offer to type it again.
|
||||
async function syncToDevice({ deviceId, device: target, code, categories, force, onProgress }) {
|
||||
const fail = (error, summary, events) => syncResult(false, error, summary || newSummary(0), events || [])
|
||||
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')
|
||||
// every progress event is kept (and handed to onProgress when given) so
|
||||
// callers can report on the transfer without subscribing first
|
||||
const events = []
|
||||
const progress = data => {
|
||||
events.push(data)
|
||||
if (onProgress) onProgress(data)
|
||||
}
|
||||
const emit = (phase, unit, done, total) =>
|
||||
progress({
|
||||
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) || '',
|
||||
done,
|
||||
total
|
||||
})
|
||||
|
||||
let verified
|
||||
try {
|
||||
verified = 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 fail(String((err && err.message) || err))
|
||||
}
|
||||
if (!verified.data || !verified.data.ok) {
|
||||
const error = (verified.data && verified.data.error) || `http-${verified.status}`
|
||||
// the peer rejected the code (it was rotated there): drop the stored
|
||||
// pairing, so the user is asked for the new code instead of every later
|
||||
// sync failing against a stale one
|
||||
if (error === 'auth-failed' || error === 'bad-code') forgetDevice(device.id || deviceId)
|
||||
return fail(error)
|
||||
}
|
||||
const token = verified.data.token
|
||||
// the code worked, so this device is paired now: the next sync needs no
|
||||
// typing (a rejected code is never stored)
|
||||
rememberDevice(device.id || deviceId, code, device.name)
|
||||
|
||||
const units = buildUnits(cats)
|
||||
const bucketOf = kind => (kind === 'wine' ? 'wines' : kind === 'prefix' ? 'prefixes' : 'apps')
|
||||
const summary = newSummary(units.length)
|
||||
const end = async () => {
|
||||
try {
|
||||
await httpJson(device, 'POST', `/plwl/end?token=${encodeURIComponent(token)}`, { json: {} })
|
||||
} catch {}
|
||||
}
|
||||
try {
|
||||
for (const unit of units) {
|
||||
const roots = unit.roots.filter(ownedRoot)
|
||||
if (!roots.length) continue
|
||||
emit('scan', unit, 0, 0)
|
||||
const entries = await buildManifest(roots, !!force, (done, total) => emit('hash', unit, done, total))
|
||||
emit('plan', unit, 0, 0)
|
||||
const plan = await planUnit(device, token, { ...unit, roots }, entries, !!force)
|
||||
if (!plan.data || !plan.data.ok) throw new Error((plan.data && plan.data.error) || `http-${plan.status}`)
|
||||
const upload = new Set(plan.data.upload || [])
|
||||
const same = new Set(plan.data.same || [])
|
||||
const bucket = summary[bucketOf(unit.kind)]
|
||||
// per category every manifest entry counts: created/overwritten vs
|
||||
// already identical (the plan reports both)
|
||||
bucket.copied += entries.length - same.size
|
||||
bucket.skipped += same.size
|
||||
let done = 0
|
||||
for (const entry of entries) {
|
||||
if (entry.type !== 'file') continue
|
||||
if (!upload.has(entry.path)) {
|
||||
summary.skipped++
|
||||
continue
|
||||
}
|
||||
emit('copy', unit, done, upload.size)
|
||||
const res = await putFile(device, token, unit.key, entry.path, filesPath(entry.path))
|
||||
if (res.status !== 200) throw new Error(`upload-failed-${res.status}`)
|
||||
done++
|
||||
summary.uploaded++
|
||||
summary.bytes += entry.size || 0
|
||||
}
|
||||
emit('copy', unit, done, upload.size)
|
||||
const commit = await httpJson(device, 'POST', `/plwl/unit/commit?token=${encodeURIComponent(token)}&unit=${encodeURIComponent(unit.key)}`, { json: {} })
|
||||
if (commit.data && commit.data.deleted) {
|
||||
summary.deleted += commit.data.deleted
|
||||
bucket.deleted += commit.data.deleted
|
||||
}
|
||||
}
|
||||
emit('registry', null, 0, 0)
|
||||
const applied = await httpJson(device, 'POST', `/plwl/apply?token=${encodeURIComponent(token)}`, { json: buildRegistryPayload([...cats]), timeout: 60000 })
|
||||
if (!applied.data || !applied.data.ok) throw new Error((applied.data && applied.data.error) || `http-${applied.status}`)
|
||||
for (const key of ['apps', 'wines', 'prefixes']) {
|
||||
const deleted = applied.data[key] && applied.data[key].deleted
|
||||
if (deleted) summary[key].deleted += deleted
|
||||
}
|
||||
if (applied.data.deleted) summary.deleted += applied.data.deleted
|
||||
emit('done', null, summary.uploaded, summary.uploaded)
|
||||
await end()
|
||||
} catch (err) {
|
||||
await end()
|
||||
return fail(String((err && err.message) || err), summary, events)
|
||||
}
|
||||
return syncResult(true, null, summary, events)
|
||||
}
|
||||
|
||||
return { syncToDevice, buildUnits, buildRegistryPayload, CATEGORIES }
|
||||
}
|
||||
|
||||
module.exports = createPusher
|
||||
Loading…
Add table
Add a link
Reference in a new issue