Test archive compression workflow

app/data/ArchiveCompressor.mjs

@@ -3,7 +3,6 @@ import { createReadStream, createWriteStream } from 'node:fs';
 import fs from 'node:fs/promises';
 import os from 'node:os';
 import path from 'node:path';
-import { spawn } from 'node:child_process';
 import { pipeline } from 'node:stream/promises';
 import * as Sentry from '@sentry/node';
 import {
@@ -13,31 +12,12 @@ import {
   S3Client,
 } from '@aws-sdk/client-s3';
 import prefixedConsole from '../common/prefixedConsole.mjs';
+import { createArchiveManifest } from './ArchiveManifest.mjs';
+import TarZstdWriter from './TarZstdWriter.mjs';
 
 const downloadLimit = 5;
 const quietPeriod = 30 * 60 * 1000;
 
-function processCompletion(child, name) {
-  let stderr = '';
-  child.stderr.setEncoding('utf8');
-  child.stderr.on('data', chunk => {
-    stderr += chunk;
-  });
-
-  return new Promise((resolve, reject) => {
-    child.once('error', reject);
-    child.once('close', (code, signal) => {
-      if (code === 0) {
-        resolve();
-      } else {
-        reject(new Error(
-          `${name} failed (${signal ? `signal ${signal}` : `exit ${code}`}): ${stderr.trim()}`,
-        ));
-      }
-    });
-  });
-}
-
 export function compressArchives(maxDays = Infinity, dryRun = false) {
   let compressor = new ArchiveCompressor;
   compressor.maxDays = maxDays;
@@ -74,6 +54,11 @@ export default class ArchiveCompressor
   dryRun = false;
   maxDays = Infinity;
 
+  constructor(s3Client, archiveWriter = new TarZstdWriter) {
+    this._client = s3Client;
+    this.archiveWriter = archiveWriter;
+  }
+
   async process() {
     if (!this.canRun) {
       this.console.log('Skipping archive compressor');
@@ -184,20 +169,14 @@ export default class ArchiveCompressor
       files.sort((a, b) => a.path.localeCompare(b.path));
 
       this.console.log(`Compressing ${candidate.date}`);
-      await this.compress(sourceDirectory, archivePath, files.map(file => file.path));
+      await this.archiveWriter.write(sourceDirectory, archivePath, files.map(file => file.path));
 
       let archive = {
         path: candidate.archiveKey,
         bytes: (await fs.stat(archivePath)).size,
         hash: `sha256:${await this.hashFile(archivePath)}`,
       };
-      let manifest = {
-        version: 1,
-        date: candidate.date,
-        createdAt: new Date().toISOString(),
-        archive,
-        files,
-      };
+      let manifest = createArchiveManifest(candidate.date, archive, files);
 
       this.console.log(`Uploading ${candidate.archiveKey}`);
       await this.upload(candidate.archiveKey, await fs.readFile(archivePath), 'application/zstd');
@@ -279,33 +258,6 @@ export default class ArchiveCompressor
     };
   }
 
-  async compress(sourceDirectory, archivePath, files) {
-    let tar = spawn('tar', ['-cf', '-', '-C', sourceDirectory, '--', ...files], {
-      stdio: ['ignore', 'pipe', 'pipe'],
-    });
-    let zstd = spawn('zstd', [
-      '-19',
-      '--long=27',
-      '--single-thread',
-      '-f',
-      '-o', archivePath,
-    ], {
-      stdio: ['pipe', 'ignore', 'pipe'],
-    });
-
-    let results = await Promise.allSettled([
-      processCompletion(tar, 'tar'),
-      pipeline(tar.stdout, zstd.stdin),
-      processCompletion(zstd, 'zstd'),
-    ]);
-    let errors = results
-      .filter(result => result.status === 'rejected')
-      .map(result => result.reason);
-    if (errors.length > 0) {
-      throw new AggregateError(errors, `Could not create ${path.basename(archivePath)}`);
-    }
-  }
-
   async hashFile(file) {
     let hash = crypto.createHash('sha256');
     for await (let chunk of createReadStream(file)) {

app/data/ArchiveCompressor.test.mjs

@@ -0,0 +1,249 @@
+import { Readable } from 'node:stream';
+import fs from 'node:fs/promises';
+import {
+  GetObjectCommand,
+  ListObjectsV2Command,
+  PutObjectCommand,
+} from '@aws-sdk/client-s3';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import ArchiveCompressor from './ArchiveCompressor.mjs';
+
+class FakeS3Client
+{
+  completed = false;
+  downloads = [];
+  files = new Map([
+    ['2025/03/05/alpha.json', Buffer.from('alpha\n')],
+    ['2025/03/05/nested/beta.json', Buffer.from('beta\n')],
+  ]);
+  secondDay = false;
+  uploads = [];
+
+  async send(command) {
+    if (command instanceof ListObjectsV2Command) {
+      return this.list(command.input.Prefix);
+    }
+
+    if (command instanceof GetObjectCommand) {
+      this.downloads.push(command.input.Key);
+
+      return { Body: Readable.from(this.files.get(command.input.Key)) };
+    }
+
+    if (command instanceof PutObjectCommand) {
+      this.uploads.push(command.input);
+
+      return {};
+    }
+
+    throw new Error(`Unexpected S3 command: ${command.constructor.name}`);
+  }
+
+  list(prefix) {
+    if (prefix === '') {
+      return { CommonPrefixes: [{ Prefix: '2025/' }] };
+    }
+    if (prefix === '2025/') {
+      return { CommonPrefixes: [{ Prefix: '2025/03/' }] };
+    }
+    if (prefix === '2025/03/') {
+      return {
+        CommonPrefixes: [
+          { Prefix: '2025/03/05/' },
+          ...(this.secondDay ? [{ Prefix: '2025/03/06/' }] : []),
+        ],
+        Contents: this.completed ? [
+          { Key: '2025/03/2025-03-05.tar.zst' },
+          { Key: '2025/03/2025-03-05.tar.zst.manifest.json' },
+        ] : [],
+      };
+    }
+    if (/^2025\/03\/\d{2}\/$/.test(prefix)) {
+      return { Contents: [...this.files]
+        .filter(([Key]) => Key.startsWith(prefix))
+        .map(([Key, body]) => ({
+          Key,
+          LastModified: new Date('2025-03-05T23:45:00.000Z'),
+          Size: body.length,
+        })) };
+    }
+
+    throw new Error(`Unexpected S3 prefix: ${prefix}`);
+  }
+}
+
+class PaginatedS3Client extends FakeS3Client
+{
+  async send(command) {
+    if (command instanceof ListObjectsV2Command && command.input.Prefix === '') {
+      if (!command.input.ContinuationToken) {
+        return {
+          CommonPrefixes: [],
+          IsTruncated: true,
+          NextContinuationToken: 'next-page',
+        };
+      }
+
+      expect(command.input.ContinuationToken).toBe('next-page');
+
+      return { CommonPrefixes: [{ Prefix: '2025/' }] };
+    }
+
+    return super.send(command);
+  }
+}
+
+describe('ArchiveCompressor', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2026-08-18T12:00:00.000Z'));
+    process.env.AWS_S3_ENDPOINT = 'https://example.invalid';
+    process.env.AWS_REGION = 'test';
+    process.env.AWS_S3_ARCHIVE_BUCKET = 'archive';
+    process.env.AWS_ACCESS_KEY_ID = 'key';
+    process.env.AWS_SECRET_ACCESS_KEY = 'secret';
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it('downloads a day and publishes its archive and manifest', async () => {
+    let s3Client = new FakeS3Client;
+    let temporaryDirectory;
+    let archiveWriter = {
+      async write(sourceDirectory, archivePath, files) {
+        temporaryDirectory = sourceDirectory.slice(0, -'/source'.length);
+        expect(files).toEqual(['alpha.json', 'nested/beta.json']);
+        expect(await fs.readFile(`${sourceDirectory}/alpha.json`, 'utf8')).toBe('alpha\n');
+        expect(await fs.readFile(`${sourceDirectory}/nested/beta.json`, 'utf8')).toBe('beta\n');
+        await fs.writeFile(archivePath, 'fake compressed archive');
+      },
+    };
+    let compressor = new ArchiveCompressor(s3Client, archiveWriter);
+    compressor._console = { error: vi.fn(), log: vi.fn() };
+
+    await compressor.process();
+
+    expect(s3Client.uploads.map(upload => upload.Key)).toEqual([
+      '2025/03/2025-03-05.tar.zst',
+      '2025/03/2025-03-05.tar.zst.manifest.json',
+    ]);
+    expect(s3Client.uploads[0].Body.toString()).toBe('fake compressed archive');
+    expect(JSON.parse(s3Client.uploads[1].Body)).toEqual({
+      version: 1,
+      date: '2025-03-05',
+      createdAt: '2026-08-18T12:00:00.000Z',
+      archive: {
+        path: '2025/03/2025-03-05.tar.zst',
+        bytes: 23,
+        hash: 'sha256:01bba9b6e21a70b88e2fb19741b8aa5e24f7d0ee368224528e7a451fcce5cbc6',
+      },
+      files: [
+        {
+          path: 'alpha.json',
+          bytes: 6,
+          hash: 'sha256:b6a98d9ce9a2d9149288fa3df42d377c3e42737afdcdaf714e33c0a100b51060',
+        },
+        {
+          path: 'nested/beta.json',
+          bytes: 5,
+          hash: 'sha256:f2c82decdd7181cf98945929a62598db7e6b477e11f6e0eb0ae97020eff151ad',
+        },
+      ],
+    });
+    await expect(fs.stat(temporaryDirectory)).rejects.toMatchObject({ code: 'ENOENT' });
+  });
+
+  it('does not download or upload files during a dry run', async () => {
+    let s3Client = new FakeS3Client;
+    let archiveWriter = { write: vi.fn() };
+    let compressor = new ArchiveCompressor(s3Client, archiveWriter);
+    compressor.dryRun = true;
+    compressor._console = { error: vi.fn(), log: vi.fn() };
+
+    await compressor.process();
+
+    expect(s3Client.downloads).toEqual([]);
+    expect(s3Client.uploads).toEqual([]);
+    expect(archiveWriter.write).not.toHaveBeenCalled();
+  });
+
+  it('leaves a completed archive alone', async () => {
+    let s3Client = new FakeS3Client;
+    s3Client.completed = true;
+    let archiveWriter = { write: vi.fn() };
+    let compressor = new ArchiveCompressor(s3Client, archiveWriter);
+    compressor._console = { error: vi.fn(), log: vi.fn() };
+
+    await compressor.process();
+
+    expect(s3Client.downloads).toEqual([]);
+    expect(s3Client.uploads).toEqual([]);
+    expect(archiveWriter.write).not.toHaveBeenCalled();
+  });
+
+  it('stops after a failed day and removes its temporary files', async () => {
+    vi.useRealTimers();
+    let s3Client = new FakeS3Client;
+    s3Client.secondDay = true;
+    s3Client.files.set('2025/03/06/gamma.json', Buffer.from('gamma\n'));
+    let temporaryDirectory;
+    let archiveWriter = {
+      async write(sourceDirectory) {
+        temporaryDirectory = sourceDirectory.slice(0, -'/source'.length);
+        throw new Error('compression failed');
+      },
+    };
+    let compressor = new ArchiveCompressor(s3Client, archiveWriter);
+    compressor._console = { error: vi.fn(), log: vi.fn() };
+
+    await expect(compressor.process()).rejects.toThrow('compression failed');
+
+    expect(s3Client.downloads).toEqual([
+      '2025/03/05/alpha.json',
+      '2025/03/05/nested/beta.json',
+    ]);
+    expect(s3Client.uploads).toEqual([]);
+    await expect(fs.stat(temporaryDirectory)).rejects.toMatchObject({ code: 'ENOENT' });
+  });
+
+  it('continues S3 discovery across paginated listings', async () => {
+    let s3Client = new PaginatedS3Client;
+    let archiveWriter = {
+      async write(sourceDirectory, archivePath) {
+        await fs.writeFile(archivePath, 'archive');
+      },
+    };
+    let compressor = new ArchiveCompressor(s3Client, archiveWriter);
+    compressor._console = { error: vi.fn(), log: vi.fn() };
+
+    await compressor.process();
+
+    expect(s3Client.uploads.map(upload => upload.Key)).toEqual([
+      '2025/03/2025-03-05.tar.zst',
+      '2025/03/2025-03-05.tar.zst.manifest.json',
+    ]);
+  });
+
+  it('does not process more than maxDays', async () => {
+    let s3Client = new FakeS3Client;
+    s3Client.secondDay = true;
+    s3Client.files.set('2025/03/06/gamma.json', Buffer.from('gamma\n'));
+    let archiveWriter = {
+      async write(sourceDirectory, archivePath) {
+        await fs.writeFile(archivePath, 'archive');
+      },
+    };
+    let compressor = new ArchiveCompressor(s3Client, archiveWriter);
+    compressor.maxDays = 1;
+    compressor._console = { error: vi.fn(), log: vi.fn() };
+
+    await compressor.process();
+
+    expect(s3Client.uploads.map(upload => upload.Key)).toEqual([
+      '2025/03/2025-03-05.tar.zst',
+      '2025/03/2025-03-05.tar.zst.manifest.json',
+    ]);
+  });
+});

app/data/ArchiveManifest.mjs

@@ -0,0 +1,71 @@
+import path from 'node:path';
+
+const sha256Pattern = /^sha256:[a-f0-9]{64}$/;
+
+export function createArchiveManifest(date, archive, files) {
+  return {
+    version: 1,
+    date,
+    createdAt: new Date().toISOString(),
+    archive,
+    files,
+  };
+}
+
+export function getArchiveManifestStats(manifest) {
+  return {
+    path: manifest.archive.path,
+    compressedBytes: manifest.archive.bytes,
+    fileCount: manifest.files.length,
+    originalBytes: manifest.files.reduce((total, file) => total + file.bytes, 0),
+  };
+}
+
+export function parseArchiveManifest(contents, archivePath) {
+  let manifest = JSON.parse(contents);
+  let filePaths = new Set;
+  if (manifest.version !== 1
+    || !isDate(manifest.date)
+    || typeof manifest.createdAt !== 'string'
+    || !Number.isFinite(Date.parse(manifest.createdAt))
+    || manifest.archive?.path !== archivePath
+    || !Number.isSafeInteger(manifest.archive?.bytes)
+    || manifest.archive.bytes < 1
+    || !sha256Pattern.test(manifest.archive?.hash)
+    || !Array.isArray(manifest.files)
+    || manifest.files.some(file => {
+      if (!isRelativePath(file.path)
+        || filePaths.has(file.path)
+        || !Number.isSafeInteger(file.bytes)
+        || file.bytes < 0
+        || !sha256Pattern.test(file.hash)) {
+        return true;
+      }
+
+      filePaths.add(file.path);
+
+      return false;
+    })) {
+    throw new Error(`Invalid archive manifest: ${archivePath}`);
+  }
+
+  return manifest;
+}
+
+function isDate(date) {
+  if (typeof date !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(date)) {
+    return false;
+  }
+
+  let parsedDate = new Date(`${date}T00:00:00.000Z`);
+
+  return !Number.isNaN(parsedDate.getTime()) && parsedDate.toISOString().slice(0, 10) === date;
+}
+
+function isRelativePath(filePath) {
+  return typeof filePath === 'string'
+    && filePath.length > 0
+    && !path.posix.isAbsolute(filePath)
+    && path.posix.normalize(filePath) === filePath
+    && !filePath.split('/').includes('..');
+}

app/data/ArchiveManifest.test.mjs

@@ -0,0 +1,95 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import {
+  createArchiveManifest,
+  getArchiveManifestStats,
+  parseArchiveManifest,
+} from './ArchiveManifest.mjs';
+
+describe('ArchiveManifest', () => {
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it('creates the public manifest format', () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2026-08-18T12:00:00.000Z'));
+    let archive = {
+      path: '2025/03/2025-03-05.tar.zst',
+      bytes: 123,
+      hash: `sha256:${'a'.repeat(64)}`,
+    };
+    let files = [{
+      path: 'sample.json',
+      bytes: 456,
+      hash: `sha256:${'b'.repeat(64)}`,
+    }];
+
+    expect(createArchiveManifest('2025-03-05', archive, files)).toEqual({
+      version: 1,
+      date: '2025-03-05',
+      createdAt: '2026-08-18T12:00:00.000Z',
+      archive,
+      files,
+    });
+  });
+
+  it('parses a valid manifest', () => {
+    let manifest = {
+      version: 1,
+      date: '2025-03-05',
+      createdAt: '2026-08-18T12:00:00.000Z',
+      archive: {
+        path: '2025/03/2025-03-05.tar.zst',
+        bytes: 123,
+        hash: `sha256:${'a'.repeat(64)}`,
+      },
+      files: [{
+        path: 'sample.json',
+        bytes: 456,
+        hash: `sha256:${'b'.repeat(64)}`,
+      }],
+    };
+
+    expect(parseArchiveManifest(
+      JSON.stringify(manifest),
+      '2025/03/2025-03-05.tar.zst',
+    )).toEqual(manifest);
+  });
+
+  it('rejects a manifest without valid SHA-256 hashes', () => {
+    let manifest = {
+      version: 1,
+      date: '2025-03-05',
+      createdAt: '2026-08-18T12:00:00.000Z',
+      archive: {
+        path: '2025/03/2025-03-05.tar.zst',
+        bytes: 123,
+        hash: 'not-a-hash',
+      },
+      files: [{
+        path: 'sample.json',
+        bytes: 456,
+        hash: `sha256:${'b'.repeat(64)}`,
+      }],
+    };
+
+    expect(() => parseArchiveManifest(
+      JSON.stringify(manifest),
+      '2025/03/2025-03-05.tar.zst',
+    )).toThrow('Invalid archive manifest');
+  });
+
+  it('summarizes the sizes recorded in a manifest', () => {
+    let manifest = {
+      archive: { path: '2025/03/2025-03-05.tar.zst', bytes: 123 },
+      files: [{ bytes: 400 }, { bytes: 600 }],
+    };
+
+    expect(getArchiveManifestStats(manifest)).toEqual({
+      path: '2025/03/2025-03-05.tar.zst',
+      compressedBytes: 123,
+      fileCount: 2,
+      originalBytes: 1000,
+    });
+  });
+});

app/data/ArchiveStats.mjs

@@ -5,6 +5,7 @@ import {
 } from '@aws-sdk/client-s3';
 import pLimit from 'p-limit';
 import prefixedConsole from '../common/prefixedConsole.mjs';
+import { getArchiveManifestStats, parseArchiveManifest } from './ArchiveManifest.mjs';
 
 const manifestSuffix = '.tar.zst.manifest.json';
 const requestLimit = pLimit(10);
@@ -115,22 +116,10 @@ export default class ArchiveStats
       throw new Error(`S3 returned no body for ${key}`);
     }
 
-    let manifest = JSON.parse(await response.Body.transformToString());
     let archivePath = key.slice(0, -'.manifest.json'.length);
-    if (manifest.archive?.path !== archivePath
-      || !Number.isSafeInteger(manifest.archive?.bytes)
-      || manifest.archive.bytes < 1
-      || !Array.isArray(manifest.files)
-      || manifest.files.some(file => !Number.isSafeInteger(file.bytes) || file.bytes < 0)) {
-      throw new Error(`Invalid archive manifest: ${key}`);
-    }
+    let manifest = parseArchiveManifest(await response.Body.transformToString(), archivePath);
 
-    return {
-      path: manifest.archive.path,
-      compressedBytes: manifest.archive.bytes,
-      fileCount: manifest.files.length,
-      originalBytes: manifest.files.reduce((total, file) => total + file.bytes, 0),
-    };
+    return getArchiveManifestStats(manifest);
   }
 
   async list(prefix, delimiter) {

app/data/TarZstdWriter.mjs

@@ -0,0 +1,54 @@
+import path from 'node:path';
+import { spawn } from 'node:child_process';
+import { pipeline } from 'node:stream/promises';
+
+function processCompletion(child, name) {
+  let stderr = '';
+  child.stderr.setEncoding('utf8');
+  child.stderr.on('data', chunk => {
+    stderr += chunk;
+  });
+
+  return new Promise((resolve, reject) => {
+    child.once('error', reject);
+    child.once('close', (code, signal) => {
+      if (code === 0) {
+        resolve();
+      } else {
+        reject(new Error(
+          `${name} failed (${signal ? `signal ${signal}` : `exit ${code}`}): ${stderr.trim()}`,
+        ));
+      }
+    });
+  });
+}
+
+export default class TarZstdWriter
+{
+  async write(sourceDirectory, archivePath, files) {
+    let tar = spawn('tar', ['-cf', '-', '-C', sourceDirectory, '--', ...files], {
+      stdio: ['ignore', 'pipe', 'pipe'],
+    });
+    let zstd = spawn('zstd', [
+      '-19',
+      '--long=27',
+      '--single-thread',
+      '-f',
+      '-o', archivePath,
+    ], {
+      stdio: ['pipe', 'ignore', 'pipe'],
+    });
+
+    let results = await Promise.allSettled([
+      processCompletion(tar, 'tar'),
+      pipeline(tar.stdout, zstd.stdin),
+      processCompletion(zstd, 'zstd'),
+    ]);
+    let errors = results
+      .filter(result => result.status === 'rejected')
+      .map(result => result.reason);
+    if (errors.length > 0) {
+      throw new AggregateError(errors, `Could not create ${path.basename(archivePath)}`);
+    }
+  }
+}