Add verified archive pruning command

app/data/ArchivePruner.mjs

@@ -0,0 +1,328 @@
+import path from 'node:path';
+import * as Sentry from '@sentry/node';
+import {
+  DeleteObjectsCommand,
+  GetObjectCommand,
+  ListObjectsV2Command,
+  S3Client,
+} from '@aws-sdk/client-s3';
+import prefixedConsole from '../common/prefixedConsole.mjs';
+import { parseArchiveManifest } from './ArchiveManifest.mjs';
+import ArchiveVerifier from './ArchiveVerifier.mjs';
+
+const retentionPeriod = 7 * 24 * 60 * 60 * 1000;
+
+export function pruneArchives(maxDays = Infinity, dryRun = false) {
+  let pruner = new ArchivePruner;
+  pruner.maxDays = maxDays;
+  pruner.dryRun = dryRun;
+
+  return pruner.process();
+}
+
+export function pruneArchivesFromCli(args) {
+  let dryRun = false;
+  let maxDays = Infinity;
+
+  for (let i = 0; i < args.length; i++) {
+    if (args[i] === '--dry-run') {
+      dryRun = true;
+    } else if (args[i] === '--max-days' && /^\d+$/.test(args[i + 1])) {
+      maxDays = Number(args[++i]);
+    } else {
+      throw new Error(
+        'Usage: npm run data:archive:prune -- [--dry-run] [--max-days DAYS]',
+      );
+    }
+  }
+
+  if (maxDays < 1) {
+    throw new Error('--max-days must be at least 1');
+  }
+
+  return pruneArchives(maxDays, dryRun);
+}
+
+export default class ArchivePruner
+{
+  dryRun = false;
+  maxDays = Infinity;
+
+  constructor(s3Client, archiveVerifier = new ArchiveVerifier) {
+    this._client = s3Client;
+    this.archiveVerifier = archiveVerifier;
+  }
+
+  async process() {
+    if (!this.canRun) {
+      this.console.log('Skipping archive pruner');
+
+      return;
+    }
+
+    let currentDate;
+
+    try {
+      let candidates = await this.getCandidates();
+      this.console.log(`Found ${candidates.length} dates to verify`);
+      let processedDays = 0;
+
+      for (let candidate of candidates) {
+        if (processedDays >= this.maxDays) {
+          break;
+        }
+
+        currentDate = candidate.date;
+        let manifest = await this.getManifest(candidate);
+        if (!this.dryRun && !this.isEligible(manifest)) {
+          this.console.log(
+            `Skipping ${candidate.date}; not eligible for deletion until ${this.eligibleDate(manifest)}`,
+          );
+          continue;
+        }
+
+        let sourceObjects = await this.getSourceObjects(candidate.prefix);
+        let archiveStream = await this.getArchive(candidate.archiveKey);
+        await this.archiveVerifier.verify(archiveStream, manifest, sourceObjects);
+        processedDays++;
+
+        if (this.dryRun) {
+          if (this.isEligible(manifest)) {
+            this.console.log(
+              `Verified ${candidate.date}; would delete ${this.fileCount(sourceObjects)}`,
+            );
+          } else {
+            this.console.log(
+              `Verified ${candidate.date}; not eligible for deletion until ${this.eligibleDate(manifest)}`,
+            );
+          }
+          continue;
+        }
+
+        let currentObjects = await this.getSourceObjects(candidate.prefix);
+        this.assertObjectsUnchanged(sourceObjects, currentObjects, candidate.date);
+        await this.deleteObjects(currentObjects);
+        this.console.log(`Pruned ${candidate.date}; deleted ${this.fileCount(currentObjects)}`);
+      }
+
+      this.console.log(
+        this.dryRun
+          ? `Verified ${processedDays} daily archives`
+          : `Pruned ${processedDays} daily archives`,
+      );
+    } catch (e) {
+      this.console.error(e);
+      Sentry.withScope(scope => {
+        if (currentDate) {
+          scope.setTag('archive.date', currentDate);
+        }
+        Sentry.captureException(e);
+      });
+      await Sentry.flush(2000).catch(() => {});
+      throw e;
+    }
+  }
+
+  get console() {
+    this._console ??= prefixedConsole('Archive Pruner');
+
+    return this._console;
+  }
+
+  get canRun() {
+    return process.env.AWS_S3_ENDPOINT
+      && process.env.AWS_REGION
+      && process.env.AWS_S3_ARCHIVE_BUCKET
+      && process.env.AWS_ACCESS_KEY_ID
+      && process.env.AWS_SECRET_ACCESS_KEY;
+  }
+
+  get s3Client() {
+    return this._client ??= new S3Client({
+      endpoint: process.env.AWS_S3_ENDPOINT,
+      region: process.env.AWS_REGION,
+      requestChecksumCalculation: 'WHEN_REQUIRED',
+      responseChecksumValidation: 'WHEN_REQUIRED',
+      credentials: {
+        accessKeyId: process.env.AWS_ACCESS_KEY_ID,
+        secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
+      },
+    });
+  }
+
+  async getCandidates() {
+    let candidates = [];
+    let years = (await this.list('', '/')).prefixes.filter(prefix => /^\d{4}\/$/.test(prefix));
+
+    for (let year of years.sort()) {
+      let months = (await this.list(year, '/')).prefixes
+        .filter(prefix => /^\d{4}\/\d{2}\/$/.test(prefix));
+
+      for (let month of months.sort()) {
+        let listing = await this.list(month, '/');
+        let existing = new Set(listing.objects.map(object => object.Key));
+
+        for (let prefix of listing.prefixes.sort()) {
+          let date = this.dateFromPrefix(prefix);
+          if (!date) {
+            continue;
+          }
+
+          let archiveKey = `${month}${date}.tar.zst`;
+          let manifestKey = `${archiveKey}.manifest.json`;
+          let hasArchive = existing.has(archiveKey);
+          let hasManifest = existing.has(manifestKey);
+          if (hasArchive !== hasManifest) {
+            throw new Error(`Archive and manifest are incomplete for ${date}`);
+          }
+          if (hasArchive) {
+            candidates.push({ archiveKey, date, manifestKey, prefix });
+          }
+        }
+      }
+    }
+
+    return candidates.sort((a, b) => a.date.localeCompare(b.date));
+  }
+
+  async getManifest(candidate) {
+    let response = await this.s3Client.send(new GetObjectCommand({
+      Bucket: process.env.AWS_S3_ARCHIVE_BUCKET,
+      Key: candidate.manifestKey,
+    }));
+    if (!response.Body) {
+      throw new Error(`S3 returned no body for ${candidate.manifestKey}`);
+    }
+
+    return parseArchiveManifest(
+      await response.Body.transformToString(),
+      candidate.archiveKey,
+    );
+  }
+
+  async getArchive(key) {
+    let response = await this.s3Client.send(new GetObjectCommand({
+      Bucket: process.env.AWS_S3_ARCHIVE_BUCKET,
+      Key: key,
+    }));
+    if (!response.Body) {
+      throw new Error(`S3 returned no body for ${key}`);
+    }
+
+    return response.Body;
+  }
+
+  async getSourceObjects(prefix) {
+    let listing = await this.list(prefix);
+
+    return listing.objects
+      .filter(object => object.Key && object.Key !== prefix && !object.Key.endsWith('/'))
+      .map(object => {
+        if (!Number.isFinite(object.Size) || typeof object.ETag !== 'string') {
+          throw new Error(`S3 returned incomplete metadata for ${object.Key}`);
+        }
+
+        return {
+          path: this.relativePath(prefix, object.Key),
+          bytes: object.Size,
+          etag: object.ETag,
+          key: object.Key,
+        };
+      })
+      .sort((a, b) => a.path.localeCompare(b.path));
+  }
+
+  async deleteObjects(objects) {
+    for (let i = 0; i < objects.length; i += 1000) {
+      let batch = objects.slice(i, i + 1000);
+      let response = await this.s3Client.send(new DeleteObjectsCommand({
+        Bucket: process.env.AWS_S3_ARCHIVE_BUCKET,
+        Delete: {
+          Objects: batch.map(object => ({ Key: object.key })),
+          Quiet: true,
+        },
+      }));
+      if (response.Errors?.length) {
+        let failures = response.Errors.map(error => `${error.Key}: ${error.Code}`).join(', ');
+        throw new Error(`Could not delete archive source files: ${failures}`);
+      }
+    }
+  }
+
+  async list(prefix, delimiter) {
+    let prefixes = [];
+    let objects = [];
+    let continuationToken;
+
+    do {
+      let response = await this.s3Client.send(new ListObjectsV2Command({
+        Bucket: process.env.AWS_S3_ARCHIVE_BUCKET,
+        ContinuationToken: continuationToken,
+        Delimiter: delimiter,
+        Prefix: prefix,
+      }));
+      prefixes.push(...(response.CommonPrefixes ?? []).map(item => item.Prefix).filter(Boolean));
+      objects.push(...(response.Contents ?? []));
+
+      if (response.IsTruncated && !response.NextContinuationToken) {
+        throw new Error(`S3 listing for ${prefix} was truncated without a continuation token`);
+      }
+      continuationToken = response.IsTruncated ? response.NextContinuationToken : undefined;
+    } while (continuationToken);
+
+    return { objects, prefixes };
+  }
+
+  isEligible(manifest) {
+    return Date.parse(manifest.createdAt) <= Date.now() - retentionPeriod;
+  }
+
+  eligibleDate(manifest) {
+    return new Date(Date.parse(manifest.createdAt) + retentionPeriod).toISOString().slice(0, 10);
+  }
+
+  assertObjectsUnchanged(expected, actual, date) {
+    let unchanged = expected.length === actual.length
+      && expected.every((object, index) => {
+        let current = actual[index];
+
+        return object.key === current.key
+          && object.bytes === current.bytes
+          && object.etag === current.etag;
+      });
+    if (!unchanged) {
+      throw new Error(`Source files changed while verifying ${date}`);
+    }
+  }
+
+  fileCount(objects) {
+    return `${objects.length} ${objects.length === 1 ? 'file' : 'files'}`;
+  }
+
+  dateFromPrefix(prefix) {
+    let match = prefix.match(/^(\d{4})\/(\d{2})\/(\d{2})\/$/);
+    if (!match) {
+      return null;
+    }
+
+    let date = `${match[1]}-${match[2]}-${match[3]}`;
+    let parsedDate = new Date(`${date}T00:00:00.000Z`);
+
+    return !Number.isNaN(parsedDate.getTime()) && parsedDate.toISOString().slice(0, 10) === date
+      ? date
+      : null;
+  }
+
+  relativePath(prefix, key) {
+    let relativePath = key.slice(prefix.length);
+    if (!key.startsWith(prefix)
+      || !relativePath
+      || path.posix.normalize(relativePath) !== relativePath
+      || path.posix.isAbsolute(relativePath)
+      || relativePath.split('/').includes('..')) {
+      throw new Error(`Invalid archive object path: ${key}`);
+    }
+
+    return relativePath;
+  }
+}

app/data/ArchivePruner.test.mjs

@@ -0,0 +1,171 @@
+import { Readable } from 'node:stream';
+import {
+  DeleteObjectsCommand,
+  GetObjectCommand,
+  ListObjectsV2Command,
+} from '@aws-sdk/client-s3';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import ArchivePruner from './ArchivePruner.mjs';
+
+const archiveKey = '2025/03/2025-03-05.tar.zst';
+const manifestKey = `${archiveKey}.manifest.json`;
+
+function manifest(createdAt = '2026-08-21T00:00:00.000Z') {
+  return {
+    version: 1,
+    date: '2025-03-05',
+    createdAt,
+    archive: {
+      path: archiveKey,
+      bytes: 7,
+      hash: `sha256:${'a'.repeat(64)}`,
+    },
+    files: [{
+      path: 'alpha.json',
+      bytes: 6,
+      hash: `sha256:${'b'.repeat(64)}`,
+    }],
+  };
+}
+
+class FakeS3Client
+{
+  deletes = [];
+  events = [];
+  manifest = manifest();
+  manifestMissing = false;
+
+  async send(command) {
+    if (command instanceof ListObjectsV2Command) {
+      return this.list(command.input.Prefix);
+    }
+    if (command instanceof GetObjectCommand && command.input.Key === manifestKey) {
+      return {
+        Body: {
+          transformToString: async () => JSON.stringify(this.manifest),
+        },
+      };
+    }
+    if (command instanceof GetObjectCommand && command.input.Key === archiveKey) {
+      return { Body: Readable.from('archive') };
+    }
+    if (command instanceof DeleteObjectsCommand) {
+      this.events.push('delete');
+      this.deletes.push(...command.input.Delete.Objects.map(object => object.Key));
+
+      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/' }],
+        Contents: [
+          { Key: archiveKey },
+          ...(this.manifestMissing ? [] : [{ Key: manifestKey }]),
+        ],
+      };
+    }
+    if (prefix === '2025/03/05/') {
+      return {
+        Contents: [{
+          ETag: '"9f9f90dbe3e5ee1218c86b8839db1995"',
+          Key: '2025/03/05/alpha.json',
+          Size: 6,
+        }],
+      };
+    }
+
+    throw new Error(`Unexpected S3 prefix: ${prefix}`);
+  }
+}
+
+describe('ArchivePruner', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2026-08-22T12: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('fully verifies recent archives during a dry run without deleting files', async () => {
+    let s3Client = new FakeS3Client;
+    let verifier = { verify: vi.fn() };
+    let pruner = new ArchivePruner(s3Client, verifier);
+    pruner.dryRun = true;
+    pruner._console = { error: vi.fn(), log: vi.fn() };
+
+    await pruner.process();
+
+    expect(verifier.verify).toHaveBeenCalledOnce();
+    expect(s3Client.deletes).toEqual([]);
+    expect(pruner._console.log).toHaveBeenCalledWith(
+      'Verified 2025-03-05; not eligible for deletion until 2026-08-28',
+    );
+  });
+
+  it('deletes an eligible day only after its archive is verified', async () => {
+    let s3Client = new FakeS3Client;
+    s3Client.manifest = manifest('2026-08-01T00:00:00.000Z');
+    let verifier = {
+      verify: vi.fn(async () => {
+        s3Client.events.push('verify');
+      }),
+    };
+    let pruner = new ArchivePruner(s3Client, verifier);
+    pruner._console = { error: vi.fn(), log: vi.fn() };
+
+    await pruner.process();
+
+    expect(s3Client.events).toEqual(['verify', 'delete']);
+    expect(s3Client.deletes).toEqual(['2025/03/05/alpha.json']);
+    expect(pruner._console.log).toHaveBeenCalledWith('Pruned 2025-03-05; deleted 1 file');
+  });
+
+  it('stops when an archive exists without its manifest', async () => {
+    let s3Client = new FakeS3Client;
+    s3Client.manifestMissing = true;
+    let verifier = { verify: vi.fn() };
+    let pruner = new ArchivePruner(s3Client, verifier);
+    pruner._console = { error: vi.fn(), log: vi.fn() };
+
+    await expect(pruner.process()).rejects.toThrow(
+      'Archive and manifest are incomplete for 2025-03-05',
+    );
+
+    expect(verifier.verify).not.toHaveBeenCalled();
+    expect(s3Client.deletes).toEqual([]);
+  });
+
+  it('stops without deleting when archive verification fails', async () => {
+    let s3Client = new FakeS3Client;
+    s3Client.manifest = manifest('2026-08-01T00:00:00.000Z');
+    let verifier = {
+      verify: vi.fn(async () => {
+        throw new Error('archive does not match');
+      }),
+    };
+    let pruner = new ArchivePruner(s3Client, verifier);
+    pruner._console = { error: vi.fn(), log: vi.fn() };
+
+    await expect(pruner.process()).rejects.toThrow('archive does not match');
+
+    expect(s3Client.deletes).toEqual([]);
+  });
+});

app/data/ArchiveVerifier.mjs

@@ -0,0 +1,164 @@
+import crypto from 'node:crypto';
+import { spawn } from 'node:child_process';
+import { Transform } from 'node:stream';
+import { pipeline } from 'node:stream/promises';
+import * as tar from 'tar-stream';
+
+function processCompletion(child) {
+  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(
+          `zstd failed (${signal ? `signal ${signal}` : `exit ${code}`}): ${stderr.trim()}`,
+        ));
+      }
+    });
+  });
+}
+
+function decompressZstd(stream) {
+  let zstd = spawn('zstd', ['-dc'], {
+    stdio: ['pipe', 'pipe', 'pipe'],
+  });
+
+  return {
+    completed: Promise.all([
+      pipeline(stream, zstd.stdin),
+      processCompletion(zstd),
+    ]),
+    stream: zstd.stdout,
+  };
+}
+
+export default class ArchiveVerifier
+{
+  constructor(decompress = decompressZstd) {
+    this.decompress = decompress;
+  }
+
+  async verify(archiveStream, manifest, sourceObjects) {
+    let inventory;
+    try {
+      inventory = this.matchFiles(manifest.files, sourceObjects);
+    } catch (error) {
+      archiveStream.destroy?.();
+      throw error;
+    }
+
+    let archiveBytes = 0;
+    let archiveHash = crypto.createHash('sha256');
+    let inspector = new Transform({
+      transform(chunk, encoding, callback) {
+        archiveBytes += chunk.length;
+        archiveHash.update(chunk);
+        callback(null, chunk);
+      },
+    });
+    let decompressor = this.decompress(inspector);
+    let results = await Promise.allSettled([
+      pipeline(archiveStream, inspector),
+      decompressor.completed,
+      this.verifyTar(decompressor.stream, inventory),
+    ]);
+    let errors = results
+      .filter(result => result.status === 'rejected')
+      .map(result => result.reason);
+    if (errors.length === 1) {
+      throw errors[0];
+    }
+    if (errors.length > 0) {
+      throw new AggregateError(errors, 'Could not verify archive');
+    }
+
+    if (archiveBytes !== manifest.archive.bytes) {
+      throw new Error('Archive size does not match its manifest');
+    }
+    if (`sha256:${archiveHash.digest('hex')}` !== manifest.archive.hash) {
+      throw new Error('Archive SHA-256 does not match its manifest');
+    }
+  }
+
+  matchFiles(manifestFiles, sourceObjects) {
+    let manifestByPath = new Map(manifestFiles.map(file => [file.path, file]));
+    let sourceByPath = new Map(sourceObjects.map(object => [object.path, object]));
+    if (manifestByPath.size !== manifestFiles.length
+      || sourceByPath.size !== sourceObjects.length
+      || manifestByPath.size !== sourceByPath.size
+      || [...manifestByPath].some(([filePath, file]) => {
+        let source = sourceByPath.get(filePath);
+
+        return !source || source.bytes !== file.bytes;
+      })) {
+      throw new Error('Live source files do not match the archive manifest');
+    }
+
+    return { manifestByPath, sourceByPath };
+  }
+
+  async verifyTar(stream, { manifestByPath, sourceByPath }) {
+    let extract = tar.extract();
+    let verified = new Set;
+    extract.on('entry', (header, entry, next) => {
+      this.verifyEntry(header, entry, manifestByPath, sourceByPath, verified)
+        .then(next)
+        .catch(error => extract.destroy(error));
+    });
+    await pipeline(stream, extract);
+
+    if (verified.size !== manifestByPath.size) {
+      throw new Error('Tar archive is missing files from its manifest');
+    }
+  }
+
+  async verifyEntry(header, entry, manifestByPath, sourceByPath, verified) {
+    if (header.type !== 'file') {
+      throw new Error(`Tar archive contains a non-file entry: ${header.name}`);
+    }
+    if (verified.has(header.name)) {
+      throw new Error(`Tar archive contains a duplicate file: ${header.name}`);
+    }
+
+    let manifestFile = manifestByPath.get(header.name);
+    let sourceObject = sourceByPath.get(header.name);
+    if (!manifestFile || !sourceObject) {
+      throw new Error(`Tar archive contains an unexpected file: ${header.name}`);
+    }
+    verified.add(header.name);
+
+    let bytes = 0;
+    let md5 = crypto.createHash('md5');
+    let sha256 = crypto.createHash('sha256');
+    for await (let chunk of entry) {
+      bytes += chunk.length;
+      md5.update(chunk);
+      sha256.update(chunk);
+    }
+
+    if (bytes !== manifestFile.bytes) {
+      throw new Error(`Size does not match the manifest for ${header.name}`);
+    }
+    if (`sha256:${sha256.digest('hex')}` !== manifestFile.hash) {
+      throw new Error(`SHA-256 does not match the manifest for ${header.name}`);
+    }
+
+    let etag = sourceObject.etag;
+    if (etag.startsWith('"') && etag.endsWith('"')) {
+      etag = etag.slice(1, -1);
+    }
+    if (!/^[a-fA-F0-9]{32}$/.test(etag)) {
+      throw new Error(`S3 ETag is not an MD5 hash for ${header.name}`);
+    }
+    if (md5.digest('hex') !== etag.toLowerCase()) {
+      throw new Error(`MD5 does not match the S3 ETag for ${header.name}`);
+    }
+  }
+}

app/data/ArchiveVerifier.test.mjs

@@ -0,0 +1,124 @@
+import { Readable } from 'node:stream';
+import { describe, expect, it } from 'vitest';
+import * as tar from 'tar-stream';
+import ArchiveVerifier from './ArchiveVerifier.mjs';
+
+async function createTar(files) {
+  let pack = tar.pack();
+  for (let [name, contents] of files) {
+    pack.entry({
+      mode: 0o644,
+      mtime: new Date(0),
+      name,
+    }, contents);
+  }
+  pack.finalize();
+
+  let chunks = [];
+  for await (let chunk of pack) {
+    chunks.push(chunk);
+  }
+
+  return Buffer.concat(chunks);
+}
+
+function manifestFor(archive, fileHash = 'b6a98d9ce9a2d9149288fa3df42d377c3e42737afdcdaf714e33c0a100b51060') {
+  return {
+    archive: {
+      bytes: archive.length,
+      hash: 'sha256:5a500b4490cc8d86f7c6970ae673b15933dc1d29f45e6098b730572ec5d2396a',
+    },
+    files: [{
+      path: 'alpha.json',
+      bytes: 6,
+      hash: `sha256:${fileHash}`,
+    }],
+  };
+}
+
+function sourceObject(etag = '9f9f90dbe3e5ee1218c86b8839db1995') {
+  return {
+    path: 'alpha.json',
+    bytes: 6,
+    etag: `"${etag}"`,
+  };
+}
+
+const identityDecompressor = stream => ({ completed: Promise.resolve(), stream });
+
+describe('ArchiveVerifier', () => {
+  it('verifies a tar stream against its manifest and live S3 objects', async () => {
+    let archive = await createTar([['alpha.json', 'alpha\n']]);
+    let verifier = new ArchiveVerifier(identityDecompressor);
+
+    await expect(verifier.verify(
+      Readable.from(archive),
+      manifestFor(archive),
+      [sourceObject()],
+    )).resolves.toBeUndefined();
+  });
+
+  it('rejects duplicate live S3 paths', async () => {
+    let archive = await createTar([['alpha.json', 'alpha\n']]);
+    let source = sourceObject();
+    let verifier = new ArchiveVerifier(identityDecompressor);
+    let archiveStream = new Readable({
+      read() {
+        this.destroy(new Error('Archive should not be read'));
+      },
+    });
+
+    await expect(verifier.verify(
+      archiveStream,
+      manifestFor(archive),
+      [source, source],
+    )).rejects.toThrow('Live source files do not match the archive manifest');
+  });
+
+  it('rejects a file whose SHA-256 does not match the manifest', async () => {
+    let archive = await createTar([['alpha.json', 'alpha\n']]);
+    let verifier = new ArchiveVerifier(identityDecompressor);
+
+    await expect(verifier.verify(
+      Readable.from(archive),
+      manifestFor(archive, '0'.repeat(64)),
+      [sourceObject()],
+    )).rejects.toThrow('SHA-256 does not match the manifest for alpha.json');
+  });
+
+  it('rejects an S3 ETag that is not a plain MD5 hash', async () => {
+    let archive = await createTar([['alpha.json', 'alpha\n']]);
+    let verifier = new ArchiveVerifier(identityDecompressor);
+
+    await expect(verifier.verify(
+      Readable.from(archive),
+      manifestFor(archive),
+      [sourceObject('multipart-etag-2')],
+    )).rejects.toThrow('S3 ETag is not an MD5 hash for alpha.json');
+  });
+
+  it('rejects a file that exists only in the tar archive', async () => {
+    let archive = await createTar([
+      ['alpha.json', 'alpha\n'],
+      ['extra.json', 'extra\n'],
+    ]);
+    let verifier = new ArchiveVerifier(identityDecompressor);
+
+    await expect(verifier.verify(
+      Readable.from(archive),
+      manifestFor(archive),
+      [sourceObject()],
+    )).rejects.toThrow('Tar archive contains an unexpected file: extra.json');
+  });
+
+  it('rejects a file whose MD5 does not match its S3 ETag', async () => {
+    let archive = await createTar([['alpha.json', 'alpha\n']]);
+    let verifier = new ArchiveVerifier(identityDecompressor);
+
+    await expect(verifier.verify(
+      Readable.from(archive),
+      manifestFor(archive),
+      [sourceObject('0'.repeat(32))],
+    )).rejects.toThrow('MD5 does not match the S3 ETag for alpha.json');
+  });
+});

app/index.mjs

@@ -10,6 +10,7 @@ import BlueskyClient from './social/clients/BlueskyClient.mjs';
 import ThreadsClient from './social/clients/ThreadsClient.mjs';
 import { archiveData } from './data/DataArchiver.mjs';
 import { compressArchivesFromCli } from './data/ArchiveCompressor.mjs';
+import { pruneArchivesFromCli } from './data/ArchivePruner.mjs';
 import { reportArchiveStatsFromCli } from './data/ArchiveStats.mjs';
 import { sentryInit } from './common/sentry.mjs';
 import { sync, syncUpload, syncDownload } from './sync/index.mjs';
@@ -31,6 +32,7 @@ const actions = {
   warmCaches,
   dataArchive: archiveData,
   archiveCompress: (...args) => compressArchivesFromCli(args),
+  archivePrune: (...args) => pruneArchivesFromCli(args),
   archiveStats: (...args) => reportArchiveStatsFromCli(args),
   sync,
   syncUpload,

package-lock.json

@@ -30,6 +30,7 @@
         "s3-sync-client": "^4.3.1",
         "sharp": "^0.34.5",
         "sirv": "^3.0.2",
+        "tar-stream": "^3.2.0",
         "twitter-api-v2": "^1.29.0",
         "vue": "^3.5.28",
         "vue-i18n": "^11.2.8",
@@ -5872,11 +5873,10 @@
       }
     },
     "node_modules/bare-fs": {
-      "version": "4.5.4",
-      "resolved": "https://registry.npmjs.org/bare-fs/-/bare-fs-4.5.4.tgz",
-      "integrity": "sha512-POK4oplfA7P7gqvetNmCs4CNtm9fNsx+IAh7jH7GgU0OJdge2rso0R20TNWVq6VoWcCvsTdlNDaleLHGaKx8CA==",
+      "version": "4.8.0",
+      "resolved": "https://registry.npmjs.org/bare-fs/-/bare-fs-4.8.0.tgz",
+      "integrity": "sha512-fM+MhCvdQhZ7NV6S95a07gPSqjIYKn6mFaXfx266wN3ajZGl/+1AzH+ubkXQ0fFZvOe2nk9VHkzdYkQE5zMV3Q==",
       "license": "Apache-2.0",
-      "optional": true,
       "dependencies": {
         "bare-events": "^2.5.4",
         "bare-path": "^3.0.0",
@@ -5885,7 +5885,7 @@
         "fast-fifo": "^1.3.2"
       },
       "engines": {
-        "bare": ">=1.16.0"
+        "bare": ">=1.28.0"
       },
       "peerDependencies": {
         "bare-buffer": "*"
@@ -5901,7 +5901,6 @@
       "resolved": "https://registry.npmjs.org/bare-os/-/bare-os-3.6.2.tgz",
       "integrity": "sha512-T+V1+1srU2qYNBmJCXZkUY5vQ0B4FSlL3QDROnKQYOqeiQR8UbjNHlPa+TIbM4cuidiN9GaTaOZgSEgsvPbh5A==",
       "license": "Apache-2.0",
-      "optional": true,
       "engines": {
         "bare": ">=1.14.0"
       }
@@ -5911,7 +5910,6 @@
       "resolved": "https://registry.npmjs.org/bare-path/-/bare-path-3.0.0.tgz",
       "integrity": "sha512-tyfW2cQcB5NN8Saijrhqn0Zh7AnFNsnczRcuWODH0eYAXBsJ5gVxAUuNr7tsHSC6IZ77cA0SitzT+s47kot8Mw==",
       "license": "Apache-2.0",
-      "optional": true,
       "dependencies": {
         "bare-os": "^3.0.1"
       }
@@ -5921,7 +5919,6 @@
       "resolved": "https://registry.npmjs.org/bare-stream/-/bare-stream-2.7.0.tgz",
       "integrity": "sha512-oyXQNicV1y8nc2aKffH+BUHFRXmx6VrPzlnaEvMhram0nPBrKcEdcyBg5r08D0i8VxngHFAiVyn1QKXpSG0B8A==",
       "license": "Apache-2.0",
-      "optional": true,
       "dependencies": {
         "streamx": "^2.21.0"
       },
@@ -5943,7 +5940,6 @@
       "resolved": "https://registry.npmjs.org/bare-url/-/bare-url-2.3.2.tgz",
       "integrity": "sha512-ZMq4gd9ngV5aTMa5p9+UfY0b3skwhHELaDkhEHetMdX0LRkW9kzaym4oo/Eh+Ghm0CCDuMTsRIGM/ytUc1ZYmw==",
       "license": "Apache-2.0",
-      "optional": true,
       "dependencies": {
         "bare-path": "^3.0.0"
       }
@@ -10914,12 +10910,13 @@
       }
     },
     "node_modules/tar-stream": {
-      "version": "3.1.7",
-      "resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-3.1.7.tgz",
-      "integrity": "sha512-qJj60CXt7IU1Ffyc3NJMjh6EkuCFej46zUqJ4J7pqYlThyd9bO0XBTmcOIhSzZJVWfsLks0+nle/j538YAW9RQ==",
+      "version": "3.2.0",
+      "resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-3.2.0.tgz",
+      "integrity": "sha512-ojzvCvVaNp6aOTFmG7jaRD0meowIAuPc3cMMhSgKiVWws1GyHbGd/xvnyuRKcKlMpt3qvxx6r0hreCNITP9hIg==",
       "license": "MIT",
       "dependencies": {
         "b4a": "^1.6.4",
+        "bare-fs": "^4.5.5",
         "fast-fifo": "^1.2.0",
         "streamx": "^2.15.0"
       }

package.json

@@ -25,6 +25,8 @@
     "data:archive": "node app/index.mjs dataArchive",
     "data:archive:compress": "node app/index.mjs archiveCompress",
     "data:archive:compress:dry-run": "node app/index.mjs archiveCompress --dry-run",
+    "data:archive:prune": "node app/index.mjs archivePrune",
+    "data:archive:prune:dry-run": "node app/index.mjs archivePrune --dry-run",
     "data:archive:stats": "node app/index.mjs archiveStats",
     "sync": "node app/index.mjs sync",
     "sync:upload": "node app/index.mjs syncUpload",
@@ -53,6 +55,7 @@
     "s3-sync-client": "^4.3.1",
     "sharp": "^0.34.5",
     "sirv": "^3.0.2",
+    "tar-stream": "^3.2.0",
     "twitter-api-v2": "^1.29.0",
     "vue": "^3.5.28",
     "vue-i18n": "^11.2.8",