'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

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))}` : ''
    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}`
      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 ----------------------------------------------------------------

  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 = 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()]
  }

  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 -----------------------------------------------------

  // The whole app config travels (paths rewritten to relative form).
  function portableApp(game) {
    return library.portableGame(game)
  }

  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('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 }))
      }
      payload.appSettings = games.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.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,
      // 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, 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)
    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)
    }
    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 {}
    }
    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()
        const plan = await planUnit(device, token, { ...unit, roots }, !!force, { totals }, async write => {
          await buildManifestStream(roots, !!force, {
            // 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))
            },
            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) 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 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
        }
        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: {} })
        if (commit.data && 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)
      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}`)
      if (applied.data.deleted) summary.deleted += applied.data.deleted
      // 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)
  }

  // putFile is exposed for the stall tests (it is the piece that must never
  // leave a request half-open)
  return { syncToDevice, buildUnits, buildRegistryPayload, putFile, CATEGORIES }
}

module.exports = createPusher