fix: download zip file to tmp BM-1417

package.json

@@ -17,12 +17,12 @@
     "@types/aws-lambda": "^8.10.83",
     "@types/geojson": "^7946.0.8",
     "@types/node": "^20.10.4",
-    "@types/unzipper": "^0.10.11",
     "@types/yazl": "^3.3.0",
+    "yazl": "^3.3.1",
+    "@types/yauzl": "^2.10.3",
     "aws-cdk": "^2.1004.0",
     "constructs": "^10.4.2",
-    "esbuild": "^0.19.9",
-    "yazl": "^3.3.1"
+    "esbuild": "^0.19.9"
   },
   "scripts": {
     "build": "tsc",
@@ -44,6 +44,6 @@
     "@linzjs/lambda": "^4.1.0",
     "aws-cdk-lib": "^2.184.1",
     "stac-ts": "^1.0.3",
-    "unzipper": "^0.12.3"
+    "yauzl": "^3.2.0"
   }
 }

src/infra/lds.export.cache.ts

@@ -1,5 +1,5 @@
 import type { StackProps } from 'aws-cdk-lib';
-import { CfnOutput, Duration, Stack } from 'aws-cdk-lib';
+import { CfnOutput, Duration, Size, Stack } from 'aws-cdk-lib';
 import { EventBus, Rule, Schedule } from 'aws-cdk-lib/aws-events';
 import { LambdaFunction } from 'aws-cdk-lib/aws-events-targets';
 import { Runtime } from 'aws-cdk-lib/aws-lambda';
@@ -31,6 +31,7 @@ export class LdsExportCache extends Stack {
       runtime: Runtime.NODEJS_22_X,
       timeout: Duration.minutes(10),
       memorySize: 4096,
+      ephemeralStorageSize: Size.gibibytes(4),
       environment: {
         CACHE_PREFIX: `s3://${cacheBucket.bucketName}`,
         KX_API_KEY: kxApiKey.stringValue,

src/storage.ts

@@ -5,9 +5,11 @@ import { fsa } from '@chunkd/fs';
 import { HashTransform } from '@chunkd/fs/build/src/hash.stream.js';
 import { FsAwsS3 } from '@chunkd/fs-aws';
 import type { LambdaRequest, LogType } from '@linzjs/lambda';
+import { createWriteStream, rmSync } from 'fs';
 import { Readable } from 'stream';
-import type { Entry, ParseStream } from 'unzipper';
-import { Parse } from 'unzipper';
+import { pipeline } from 'stream/promises';
+import type { Entry } from 'yauzl';
+import yauzl from 'yauzl';
 
 import { CachePrefix, ExportLayerId, kx } from './config.ts';
 import type { KxDatasetExport, KxDatasetVersionDetail } from './kx.ts';
@@ -56,33 +58,56 @@ export async function extractAndWritePackage(
   datasetId: number,
   log: LogType,
 ): Promise<void> {
-  const unzipperParser: ParseStream = Parse();
+  const tmpZipFile = `/tmp/${datasetId}.zip`;
+  try {
+    await pipeline(stream, createWriteStream(tmpZipFile));
+    const fileNames: string[] = [];
+
+    const writeProms: Promise<void>[] = [];
+
+    return new Promise((resolve, reject) => {
+      yauzl.open(tmpZipFile, { lazyEntries: true }, (err, zipFile) => {
+        if (err) reject(`Failed to open zip file ${tmpZipFile}`);
+
+        zipFile.once('end', () => {
+          Promise.all(writeProms)
+            .then(() => {
+              resolve();
+            })
+            .catch(reject);
+        });
 
-  let writeProm: Promise<void> | undefined;
-  let fileName: string | null = null;
+        zipFile.once('error', reject);
 
-  await stream
-    .pipe(unzipperParser)
-    .on('entry', (entry: Entry) => {
-      log.debug({ datasetId, path: entry.path }, 'Export:Zip:File');
-      if (entry.path.endsWith(PackageExtension)) {
-        log.info({ datasetId, path: entry.path, target: targetFileUri.href }, 'Ingest:Read:Start');
+        zipFile.on('entry', (entry: Entry) => {
+          log.debug({ datasetId, path: entry.fileName }, 'Export:Zip:File');
+          if (entry.fileName.endsWith(PackageExtension)) {
+            log.info({ datasetId, path: entry.fileName, target: targetFileUri.href }, 'Ingest:Read:Start');
 
-        if (fileName != null) throw Error(`Duplicate export package: ${fileName} vs ${entry.path}`);
-        fileName = entry.path;
+            if (fileNames.includes(entry.fileName)) reject(`Duplicate export package: ${entry.fileName}`);
+            fileNames.push(entry.fileName);
 
-        const gzipOut = entry.pipe(ht).pipe(createGzip({ level: 9 }));
-        writeProm = fsa.write(targetFileUri, gzipOut, {
-          contentType: 'application/geopackage+vnd.sqlite3',
-          contentEncoding: 'gzip',
-        });
-      } else {
-        entry.autodrain();
-      }
-    })
-    .promise();
+            zipFile.openReadStream(entry, (err, readStream) => {
+              if (err) reject(`Failed to read zip entry ${entry.fileName}`);
+
+              const gzipOut = readStream.pipe(ht).pipe(createGzip({ level: 9 }));
+              writeProms.push(
+                fsa.write(targetFileUri, gzipOut, {
+                  contentType: 'application/geopackage+vnd.sqlite3',
+                  contentEncoding: 'gzip',
+                }),
+              );
 
-  if (writeProm != null) await writeProm;
+              zipFile.readEntry();
+            });
+          }
+        });
+        zipFile.readEntry();
+      });
+    });
+  } finally {
+    rmSync(tmpZipFile);
+  }
 }
 
 /** Ingest the export into our cache */