diff --git a/package.json b/package.json index 8b5780a..0d5e85e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@ztimson/zim-utils", - "version": "0.4.6", + "version": "0.4.7", "description": "Native, dependency-light ZIM archive reader/searcher and Kiwix catalog downloader for Node.js", "author": "Zak Timson", "license": "MIT", diff --git a/src/manager.js b/src/manager.js index c0a4d11..391ce83 100644 --- a/src/manager.js +++ b/src/manager.js @@ -2,6 +2,7 @@ import fs from 'node:fs'; import path from 'node:path'; import {pipeline} from 'node:stream/promises'; import {Readable} from 'node:stream'; +import {PromiseProgress} from '@ztimson/utils'; import {KiwixServer} from './server.js'; import {zimCatalog, zimCatalogInfo, CATALOG_URL} from './catalog.js'; @@ -17,6 +18,7 @@ export class ZimManager { #dir; #server; #ownsServer; + #active = {}; // name -> in-flight PromiseProgress, for on-demand status lookups /** @param {{catalog?: string, server?: KiwixServer, port?: number, host?: string, binDir?: string, url?: string}} [opts] * server: reuse an existing KiwixServer instance directly. @@ -41,11 +43,16 @@ export class ZimManager { return this.#server; } - async #download(url, destPath) { + /** Streams `url` to `destPath`, reporting raw bytes/total via `onBytes(size, total)` as chunks arrive. */ + async #download(url, destPath, onBytes) { const {res} = await this.#resolveUrl(url); if (!res.ok) throw new Error(`${res.status} ${res.statusText}`); + const total = Number(res.headers.get('content-length')) || 0; const tmpPath = `${destPath}.part`; - await pipeline(Readable.fromWeb(res.body), fs.createWriteStream(tmpPath)); + let size = 0; + const readable = Readable.fromWeb(res.body); + if (onBytes) readable.on('data', chunk => onBytes(size += chunk.length, total)); + await pipeline(readable, fs.createWriteStream(tmpPath)); await fs.promises.rename(tmpPath, destPath); } @@ -75,7 +82,8 @@ export class ZimManager { return false; } - async #update(name, catalogEntry, localMatch, force) { + /** Performs the actual download + post-processing for a resolved catalog entry, forwarding byte progress via `onBytes`. */ + async #update(name, catalogEntry, localMatch, force, onBytes) { const remoteDate = catalogEntry.updated ? new Date(catalogEntry.updated) : null; const localDate = localMatch?.updated ?? null; if (!force && localMatch && remoteDate && localDate && remoteDate <= localDate) @@ -84,13 +92,22 @@ export class ZimManager { const filename = path.basename(new URL(catalogEntry.href).pathname).replace(/\.meta4$/i, ''); const destPath = path.join(this.#dir, filename); - await this.#download(catalogEntry.href, destPath); + await this.#download(catalogEntry.href, destPath, onBytes); if (localMatch && localMatch.file !== filename) await fs.promises.rm(path.join(this.#dir, localMatch.file), {force: true}); await this.#ensureServer(); await this.#waitUntilVisible(filename, {expect: true}); return {name, status: 'updated', file: filename}; } + /** Wraps `#update` in a tracked `PromiseProgress`, registered in `#active` under `name` until it settles. */ + #trackedUpdate(name, catalogEntry, localMatch, force) { + const dl = new PromiseProgress((resolve, reject, prog) => + this.#update(name, catalogEntry, localMatch, force, prog).then(resolve, reject)); + this.#active[name] = dl; + dl.catch(() => {}).finally(() => { if (this.#active[name] === dl) delete this.#active[name]; }); + return dl; + } + catalog(search, opts = {}) { return zimCatalog(search, {url: this.#catalogUrl, ...opts}); } @@ -111,18 +128,28 @@ export class ZimManager { return {file: match.file, status: 'deleted'}; } - /** Downloads/updates a single ZIM by direct href, matching against any existing local copy by name. */ - async download(href, {force = false} = {}) { - const {url: finalUrl} = await this.#resolveUrl(href); - const filename = path.basename(new URL(finalUrl).pathname).replace(/\.meta4$/i, ''); - await fs.promises.mkdir(this.#dir, {recursive: true}); - const local = await this.list(); - const exact = local.find(l => l.file === filename); - if (exact && !force) return {name: exact.name ?? this.#stripDate(filename), status: 'skipped', reason: 'already downloaded', file: filename}; - const name = this.#stripDate(filename); - const localMatch = local.find(l => l.name === name) ?? null; - const catalogEntry = await zimCatalogInfo(name, this.#catalogUrl) || {name, updated: null, href}; - return this.#update(name, catalogEntry, localMatch, force); + /** Downloads/updates a single ZIM by direct href, matching against any existing local copy by name. + * Returns a `PromiseProgress` with live `.progress`/`.speed`/`.eta`/etc getters & `onProgress` callback, + * while also tracking it internally (see `progress()`/`downloads()`) in case the reference is dropped. */ + download(href, {force = false} = {}) { + let dl; + dl = new PromiseProgress(async (resolve, reject, prog) => { + try { + const {url: finalUrl} = await this.#resolveUrl(href); + const filename = path.basename(new URL(finalUrl).pathname).replace(/\.meta4$/i, ''); + await fs.promises.mkdir(this.#dir, {recursive: true}); + const local = await this.list(); + const exact = local.find(l => l.file === filename); + if (exact && !force) return resolve({name: exact.name ?? this.#stripDate(filename), status: 'skipped', reason: 'already downloaded', file: filename}); + const name = this.#stripDate(filename); + const localMatch = local.find(l => l.name === name) ?? null; + const catalogEntry = await zimCatalogInfo(name, this.#catalogUrl) || {name, updated: null, href}; + this.#active[name] = dl; + dl.catch(() => {}).finally(() => { if (this.#active[name] === dl) delete this.#active[name]; }); + resolve(await this.#update(name, catalogEntry, localMatch, force, prog)); + } catch (e) { reject(e); } + }); + return dl; } async link(href) { @@ -159,7 +186,8 @@ export class ZimManager { return server.search(terms, {limit, sources}); } - /** Checks all local ZIMs against the catalog and updates any that are outdated. */ + /** Checks all local ZIMs against the catalog and updates any that are outdated. + * Each in-flight update is individually tracked in `#active` - use `progress(name)`/`downloads()` to poll them mid-run. */ async updateAll({force = false} = {}) { const local = await this.list(); const results = []; @@ -168,11 +196,21 @@ export class ZimManager { const catalogEntry = await zimCatalogInfo(entry.name, this.#catalogUrl); if (!catalogEntry) { results.push({name: entry.name, status: 'skipped', reason: 'missing from catalog'}); continue; } try { - results.push(await this.#update(entry.name, catalogEntry, entry, force)); + results.push(await this.#trackedUpdate(entry.name, catalogEntry, entry, force)); } catch (e) { results.push({name: entry.name, status: 'error', reason: e.message}); } } return results; } + + /** Live progress snapshot (speed/eta/size/total/...) for a single named download, or `null` if nothing's in flight. */ + progress(name) { + return this.#active[name] ? {name, ...this.#active[name].meta} : null; + } + + /** Live progress snapshots for every currently active download. */ + downloads() { + return Object.entries(this.#active).map(([name, dl]) => ({name, ...dl.meta})); + } }