u

src/api/index.js

@@ -13,6 +13,7 @@ import { ensureConnection } from '../db/store.js';
 import { closeAll as closeAllSse } from './sse.js';
 import { closeAllManagers } from '../services/conversationManager.js';
 import { shutdownImageCropWorkers } from '../services/imageCrop.js';
+import { shutdownImageHashWorkers } from '../services/companyImport/imageHasher.js';
 import { shutdownJavascriptSandboxWorkers } from '../services/javascriptSandbox.js';
 import { getOrCreateCert } from '../util/cert.js';
 import { createLogger } from '../util/logger.js';
@@ -279,6 +280,7 @@ export async function stopApi() {
     // Pooled image-crop workers are live threads and must be terminated too, or the
     // process stays alive after the last request has been served.
     await shutdownImageCropWorkers();
+    await shutdownImageHashWorkers();
     // Same for the pooled sandbox threads that run `javascript_execute` / applets.
     await shutdownJavascriptSandboxWorkers();
     await stopBackupService();

src/services/companyImport/articleImageImporter.js

@@ -1,12 +1,13 @@
 // MARK: - Imports
 // #region Imports
 import fs from 'fs';
-import { computeSkeleHash } from './imageHasher.js';
 import { createLogger } from '../../util/logger.js';
 // #endregion
 
 const log = createLogger('import');
 
+const IMAGE_CONCURRENCY = 4;
+
 // MARK: - Article Image Importer
 // #region Article Image Importer
 
@@ -28,20 +29,25 @@ export async function importArticleImages({
 }) {
     let totalImagesImported = 0;
 
-    for (const task of articleImageTasks) {
-        try {
-            const imgBuf = fs.readFileSync(task.fullPath);
-            await store.article.addImage(googleAuthId, task.sku, imgBuf);
-            totalImagesImported++;
-
-            const skele = await computeSkeleHash(imgBuf);
-            progressTracker.emitSkeleHash(skele?.base62 || null);
-        } catch (err) {
-            log.warn(`Article image add failed for ${task.sku} (${task.relPath}):`, err.message);
+    // addImage decodes/encodes in the store's worker pool before its (serialized) DB transaction, so
+    // several in flight overlap the CPU work. The store also returns the skelehash — no second decode.
+    let next = 0;
+    const runNext = async () => {
+        while (next < articleImageTasks.length) {
+            const task = articleImageTasks[next++];
+            try {
+                const imgBuf = await fs.promises.readFile(task.fullPath);
+                const added = await store.article.addImage(googleAuthId, task.sku, imgBuf);
+                totalImagesImported++;
+                progressTracker.emitSkeleHash(added?.skelehash || null);
+            } catch (err) {
+                log.warn(`Article image add failed for ${task.sku} (${task.relPath}):`, err.message);
+            }
+            progressTracker.counts.imagesdone++;
+            progressTracker.emitProgress();
         }
-        progressTracker.counts.imagesdone++;
-        progressTracker.emitProgress();
-    }
+    };
+    await Promise.all(Array.from({ length: Math.min(IMAGE_CONCURRENCY, articleImageTasks.length) }, runNext));
 
     progressTracker.flushSkeleHashes();
     progressTracker.emitProgress(true);

src/services/companyImport/articleImporter.js

@@ -4,7 +4,6 @@ import fs from 'fs';
 import { cleanValue, parseGermanFloat, parseSuppliers } from './csvParser.js';
 import { setupVariantAttributes } from './variantImporter.js';
 import { setupPfandArticles, linkArticlePfand, extractPfandValue } from './pfandImporter.js';
-import { computeSkeleHash } from './imageHasher.js';
 import { createLogger } from '../../util/logger.js';
 // #endregion
 
@@ -154,6 +153,9 @@ export async function importArticles({
     let pfandOptionsLinked = 0;
     let suppliersLinked = 0;
     let articlePicturesImported = 0;
+    const IMAGE_CONCURRENCY = 4;
+    /** @type {Set<Promise<void>>} */
+    const inFlightImageJobs = new Set();
 
     // 6. Unified single loop over all articles
     for (const a of sortedArticles) {
@@ -452,26 +454,33 @@ export async function importArticles({
         // --- Pictures handling for this article ---
         const imagesForThisArt = articleImagesBySku.get(sku);
         if (imagesForThisArt && imagesForThisArt.length > 0) {
-            for (const imgTask of imagesForThisArt) {
-                try {
-                    const imgBuf = fs.readFileSync(imgTask.fullPath);
-                    await store.article.addImage(googleAuthId, sku, imgBuf);
-                    articlePicturesImported++;
-
-                    const skele = await computeSkeleHash(imgBuf);
-                    progressTracker.emitSkeleHash(skele?.base62 || null);
-                } catch (imgErr) {
-                    log.warn(`Article image add failed for ${sku} (${imgTask.relPath}):`, imgErr.message);
+            // Don't await: addImage's decode/encode runs in the store's worker pool, so keep creating the
+            // next articles while up to IMAGE_CONCURRENCY images are processed. Images of one article stay
+            // sequential (sort order). Awaiting happens only when the in-flight limit is reached.
+            const job = (async () => {
+                for (const imgTask of imagesForThisArt) {
+                    try {
+                        const imgBuf = await fs.promises.readFile(imgTask.fullPath);
+                        const added = await store.article.addImage(googleAuthId, sku, imgBuf);
+                        articlePicturesImported++;
+                        progressTracker.emitSkeleHash(added?.skelehash || null);
+                    } catch (imgErr) {
+                        log.warn(`Article image add failed for ${sku} (${imgTask.relPath}):`, imgErr.message);
+                    }
+                    progressTracker.counts.imagesdone++;
+                    progressTracker.emitProgress();
                 }
-                progressTracker.counts.imagesdone++;
-                progressTracker.emitProgress();
-            }
+            })();
+            inFlightImageJobs.add(job);
+            job.finally(() => inFlightImageJobs.delete(job));
+            if (inFlightImageJobs.size >= IMAGE_CONCURRENCY) await Promise.race(inFlightImageJobs);
         }
 
         progressTracker.counts.articlesdone++;
         progressTracker.emitProgress();
     }
 
+    await Promise.all(inFlightImageJobs);
     progressTracker.flushSkeleHashes();
     progressTracker.emitProgress(true);
     log.info(`Articles complete: New: ${articlesImported}, Updated: ${articlesUpdated}, Variants: ${variantsLinked}, Pfand: ${pfandOptionsLinked}, Suppliers: ${suppliersLinked}, Pictures: ${articlePicturesImported}`);

src/services/companyImport/imageHashWorker.js

@@ -0,0 +1,36 @@
+/**
+ * Worker-thread entry for SkeleHash computation (decode + resize + encodeRGBA).
+ * Protocol: parent sends `{ id, buffer }` (ArrayBuffer, transferred); worker replies
+ * `{ id, ok: true, result }` or `{ id, ok: false, error }`.
+ */
+import { parentPort } from 'worker_threads';
+import { optimizeImage } from 'wasm-image-optimization';
+import { encodeRGBA } from 'skelehash-wasm';
+
+parentPort.postMessage({ ready: true });
+
+parentPort.on('message', async ({ id, buffer }) => {
+    try {
+        const raw = await optimizeImage({
+            image: new Uint8Array(buffer),
+            width: 200,
+            height: 200,
+            fit: 'contain',
+            format: 'raw',
+        });
+        const result = await encodeRGBA(raw.width, raw.height, raw.data);
+        if (!result?.base62) throw new Error('encodeRGBA failed to produce a base62 skelehash');
+        parentPort.postMessage({
+            id,
+            ok: true,
+            result: {
+                base62: result.base62,
+                width: raw.width,
+                height: raw.height,
+                hashLength: result.hash?.length ?? 0,
+            },
+        });
+    } catch (err) {
+        parentPort.postMessage({ id, ok: false, error: err?.message || String(err) });
+    }
+});

src/services/companyImport/imageHasher.js

@@ -1,17 +1,69 @@
 // MARK: - Imports
 // #region Imports
-import { optimizeImage } from 'wasm-image-optimization';
-import { encodeRGBA } from 'skelehash-wasm';
+import { Worker } from 'worker_threads';
+import { fileURLToPath } from 'url';
+import os from 'os';
 import { createLogger } from '../../util/logger.js';
 // #endregion
 
 const log = createLogger('import');
 
+// `wasm-image-optimization/workers` (worker-lib) cannot be loaded under native Node ESM
+// (extensionless `./pool` import -> ERR_MODULE_NOT_FOUND), so use a plain worker_threads pool.
+const WORKER_PATH = fileURLToPath(new URL('./imageHashWorker.js', import.meta.url));
+const POOL_SIZE = Math.max(1, Math.min(4, (os.cpus?.()?.length || 2) - 1));
+
+/** @type {Array<{ worker: Worker, busy: boolean, pending: Map<number, {resolve: Function, reject: Function}> }>} */
+const pool = [];
+const queue = [];
+let jobSeq = 0;
+
+function spawn() {
+    const entry = { worker: new Worker(WORKER_PATH), busy: false, pending: new Map(), ready: false };
+    entry.worker.on('message', (msg) => {
+        if (msg.ready) { entry.ready = true; drain(); return; }
+        const p = entry.pending.get(msg.id);
+        if (!p) return;
+        entry.pending.delete(msg.id);
+        entry.busy = false;
+        entry.worker.unref();
+        if (msg.ok) p.resolve(msg.result); else p.reject(new Error(msg.error));
+        drain();
+    });
+    entry.worker.on('error', (err) => {
+        for (const p of entry.pending.values()) p.reject(err);
+        entry.pending.clear();
+        pool.splice(pool.indexOf(entry), 1);
+    });
+    entry.worker.unref();
+    pool.push(entry);
+}
+
+function drain() {
+    while (queue.length) {
+        const entry = pool.find((e) => e.ready && !e.busy);
+        if (!entry) {
+            if (pool.length < POOL_SIZE) spawn();
+            return;
+        }
+        const job = queue.shift();
+        entry.busy = true;
+        entry.worker.ref(); // keep the process alive while a job is in flight
+        entry.pending.set(job.id, job);
+        entry.worker.postMessage({ id: job.id, buffer: job.buffer }, [job.buffer]);
+    }
+}
+
+/** Terminates the hashing workers (call on shutdown). */
+export async function shutdownImageHashWorkers() {
+    await Promise.all(pool.splice(0).map((e) => e.worker.terminate()));
+}
+
 // MARK: - SkeleHash Computation
 // #region SkeleHash Computation
 
 /**
- * Computes SkeleHash base62 hash for an image buffer or base64 string.
+ * Computes SkeleHash base62 hash for an image buffer or base64 string, off the main thread.
  *
  * @param {Buffer|string} buffer
  * @returns {Promise<{base62: string, width: number, height: number, hashLength: number}|null>}
@@ -23,24 +75,11 @@ export async function computeSkeleHash(buffer) {
         const base64Data = input.includes(',') ? input.split(',')[1] : input;
         input = Buffer.from(base64Data, 'base64');
     }
-    const raw = await optimizeImage({
-        image: input,
-        width: 200,
-        height: 200,
-        fit: 'contain',
-        format: 'raw',
+    const copy = Uint8Array.from(input).buffer; // standalone ArrayBuffer, safe to transfer
+    return new Promise((resolve, reject) => {
+        queue.push({ id: ++jobSeq, buffer: copy, resolve, reject });
+        drain();
     });
-
-    const result = await encodeRGBA(raw.width, raw.height, raw.data);
-    if (!result?.base62) {
-        throw new Error('encodeRGBA failed to produce a base62 skelehash');
-    }
-    return {
-        base62: result.base62,
-        width: raw.width,
-        height: raw.height,
-        hashLength: result.hash?.length ?? 0,
-    };
 }
 
 // #endregion