Merge pull request #110 from misenhower/codex/daily-archive-generator

Add daily archive compression

app/cron.mjs

@@ -4,6 +4,7 @@ import { update } from './data/index.mjs';
 import { warmCaches } from './splatnet/index.mjs';
 import { sendStatuses } from './social/index.mjs';
 import { archiveData } from './data/DataArchiver.mjs';
+import { compressArchives } from './data/ArchiveCompressor.mjs';
 import { updateAvatars } from './social/updateAvatars.mjs';
 
 let updating = false;
@@ -62,5 +63,6 @@ export default function() {
   }, null, true);
 
   new CronJob('30 * * * *', updateAvatars, null, true);
+  new CronJob('30 0 * * *', () => compressArchives(10), null, true, 'UTC');
   new CronJob('0 55 4 * * *', restartToRefreshConfig, null, true);
 }

app/data/ArchiveCompressor.mjs

@@ -0,0 +1,386 @@
+import crypto from 'node:crypto';
+import { createReadStream, createWriteStream } from 'node:fs';
+import fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { pipeline } from 'node:stream/promises';
+import * as Sentry from '@sentry/node';
+import {
+  GetObjectCommand,
+  ListObjectsV2Command,
+  PutObjectCommand,
+  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;
+
+export function compressArchives(maxDays = Infinity, dryRun = false) {
+  let compressor = new ArchiveCompressor;
+  compressor.maxDays = maxDays;
+  compressor.dryRun = dryRun;
+
+  return compressor.process();
+}
+
+export function compressArchivesFromCli(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:compress -- [--dry-run] [--max-days DAYS]',
+      );
+    }
+  }
+
+  if (maxDays < 1) {
+    throw new Error('--max-days must be at least 1');
+  }
+
+  return compressArchives(maxDays, dryRun);
+}
+
+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');
+
+      return;
+    }
+
+    let currentDate;
+    let compressed = 0;
+
+    try {
+      let candidates = await this.getCandidates();
+      this.console.log(`Found ${candidates.length} dates to archive`);
+
+      for (let candidate of candidates) {
+        if (compressed >= this.maxDays) {
+          break;
+        }
+
+        currentDate = candidate.date;
+        let completed = this.dryRun
+          ? await this.previewDate(candidate)
+          : await this.archiveDate(candidate);
+        if (completed) {
+          compressed++;
+        }
+      }
+    } 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;
+    }
+
+    this.console.log(
+      this.dryRun
+        ? `Would compress ${compressed} daily archives`
+        : `Compressed ${compressed} daily archives`,
+    );
+  }
+
+  // Properties
+
+  get console() {
+    this._console ??= prefixedConsole('Archive Compressor');
+
+    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,
+      },
+    });
+  }
+
+  // Archive compression
+
+  async previewDate(candidate) {
+    let objects = await this.getReadyObjects(candidate);
+    if (!objects) {
+      return false;
+    }
+
+    this.console.log(
+      `Would create ${candidate.archiveKey} and ${candidate.manifestKey} `
+      + `from ${objects.length} files`,
+    );
+
+    return true;
+  }
+
+  async archiveDate(candidate) {
+    let objects = await this.getReadyObjects(candidate);
+    if (!objects) {
+      return false;
+    }
+
+    let temporaryDirectory = await fs.mkdtemp(
+      path.join(os.tmpdir(), `splatoon3ink-${candidate.date}-`),
+    );
+    let sourceDirectory = path.join(temporaryDirectory, 'source');
+    let archivePath = path.join(temporaryDirectory, `${candidate.date}.tar.zst`);
+
+    try {
+      await fs.mkdir(sourceDirectory);
+      this.console.log(`Downloading ${objects.length} files for ${candidate.date}`);
+      let files = await this.downloadObjects(candidate.prefix, objects, sourceDirectory);
+      files.sort((a, b) => a.path.localeCompare(b.path));
+
+      this.console.log(`Compressing ${candidate.date}`);
+      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 = createArchiveManifest(candidate.date, archive, files);
+
+      this.console.log(`Uploading ${candidate.archiveKey}`);
+      await this.upload(candidate.archiveKey, await fs.readFile(archivePath), 'application/zstd');
+      await this.upload(
+        candidate.manifestKey,
+        Buffer.from(`${JSON.stringify(manifest, null, 2)}\n`),
+        'application/json',
+      );
+
+      return true;
+    } finally {
+      await fs.rm(temporaryDirectory, { force: true, recursive: true });
+    }
+  }
+
+  async getReadyObjects(candidate) {
+    let objects = await this.getSourceObjects(candidate.prefix);
+    if (objects.length === 0) {
+      return null;
+    }
+
+    let newestObject = Math.max(...objects.map(object => object.lastModified.getTime()));
+    if (newestObject > Date.now() - quietPeriod) {
+      this.console.log(`Skipping ${candidate.date}; its source files are still changing`);
+
+      return null;
+    }
+
+    return objects;
+  }
+
+  async downloadObjects(prefix, objects, sourceDirectory) {
+    let files = [];
+    let nextObject = 0;
+    let error;
+    let worker = async () => {
+      while (!error && nextObject < objects.length) {
+        let index = nextObject++;
+        try {
+          files[index] = await this.downloadObject(prefix, objects[index], sourceDirectory);
+        } catch (e) {
+          error ??= e;
+        }
+      }
+    };
+
+    let workers = Array.from({ length: Math.min(downloadLimit, objects.length) }, worker);
+    await Promise.all(workers);
+    if (error) {
+      throw error;
+    }
+
+    return files;
+  }
+
+  async downloadObject(prefix, object, sourceDirectory) {
+    let relativePath = this.relativePath(prefix, object.key);
+    let destination = path.join(sourceDirectory, relativePath);
+    await fs.mkdir(path.dirname(destination), { recursive: true });
+
+    let response = await this.s3Client.send(new GetObjectCommand({
+      Bucket: process.env.AWS_S3_ARCHIVE_BUCKET,
+      Key: object.key,
+    }));
+    if (!response.Body) {
+      throw new Error(`S3 returned no body for ${object.key}`);
+    }
+
+    await pipeline(response.Body, createWriteStream(destination, { flags: 'wx' }));
+    let bytes = (await fs.stat(destination)).size;
+    if (bytes !== object.bytes) {
+      throw new Error(`Downloaded size does not match S3 listing for ${object.key}`);
+    }
+
+    return {
+      path: relativePath,
+      bytes,
+      hash: `sha256:${await this.hashFile(destination)}`,
+    };
+  }
+
+  async hashFile(file) {
+    let hash = crypto.createHash('sha256');
+    for await (let chunk of createReadStream(file)) {
+      hash.update(chunk);
+    }
+
+    return hash.digest('hex');
+  }
+
+  async upload(key, body, contentType) {
+    await this.s3Client.send(new PutObjectCommand({
+      ACL: 'public-read',
+      Body: body,
+      Bucket: process.env.AWS_S3_ARCHIVE_BUCKET,
+      ContentLength: body.length,
+      ContentType: contentType,
+      Key: key,
+    }));
+  }
+
+  // S3 discovery
+
+  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 || date >= new Date().toISOString().slice(0, 10)) {
+            continue;
+          }
+
+          let archiveKey = `${month}${date}.tar.zst`;
+          let manifestKey = `${archiveKey}.manifest.json`;
+          if (existing.has(archiveKey) && existing.has(manifestKey)) {
+            continue;
+          }
+
+          candidates.push({ archiveKey, date, manifestKey, prefix });
+        }
+      }
+    }
+
+    return candidates.sort((a, b) => a.date.localeCompare(b.date));
+  }
+
+  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 (!object.LastModified || !Number.isFinite(object.Size)) {
+          throw new Error(`S3 returned incomplete metadata for ${object.Key}`);
+        }
+
+        return {
+          key: object.Key,
+          bytes: object.Size,
+          lastModified: object.LastModified,
+        };
+      });
+  }
+
+  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 };
+  }
+
+  // Helpers
+
+  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`);
+    if (Number.isNaN(parsedDate.getTime()) || parsedDate.toISOString().slice(0, 10) !== date) {
+      return null;
+    }
+
+    return date;
+  }
+
+  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/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.toSorted()).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

@@ -0,0 +1,190 @@
+import {
+  GetObjectCommand,
+  ListObjectsV2Command,
+  S3Client,
+} 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);
+
+export function reportArchiveStats(verbose = false) {
+  let stats = new ArchiveStats;
+  stats.verbose = verbose;
+
+  return stats.process();
+}
+
+export function reportArchiveStatsFromCli(args) {
+  if (args.some(argument => argument !== '--verbose')) {
+    throw new Error('Usage: npm run data:archive:stats -- [--verbose]');
+  }
+
+  return reportArchiveStats(args.includes('--verbose'));
+}
+
+export default class ArchiveStats
+{
+  verbose = false;
+
+  async process() {
+    if (!this.canRun) {
+      this.console.log('Skipping archive stats');
+
+      return;
+    }
+
+    this.console.log('Reading archive manifests...');
+    let keys = await this.getManifestKeys();
+    let archives = await Promise.all(keys.map(key => requestLimit(() => this.readManifest(key))));
+    archives.sort((a, b) => a.path.localeCompare(b.path));
+
+    if (this.verbose) {
+      for (let archive of archives) {
+        this.console.log(this.describeArchive(archive));
+      }
+    }
+
+    let originalBytes = archives.reduce((total, archive) => total + archive.originalBytes, 0);
+    let compressedBytes = archives.reduce((total, archive) => total + archive.compressedBytes, 0);
+    let fileCount = archives.reduce((total, archive) => total + archive.fileCount, 0);
+    let savedBytes = originalBytes - compressedBytes;
+
+    this.console.log(
+      `${archives.length} archives containing ${fileCount} files: `
+      + `${this.formatBytes(originalBytes)} source data -> ${this.formatBytes(compressedBytes)} compressed; `
+      + `projected source-file savings after pruning ${this.formatBytes(savedBytes)} `
+      + `(${this.formatPercent(savedBytes, originalBytes)}; manifests excluded), `
+      + `${this.formatRatio(originalBytes, compressedBytes)}:1 compression`,
+    );
+  }
+
+  // Properties
+
+  get console() {
+    this._console ??= prefixedConsole('Archive Stats');
+
+    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,
+      },
+    });
+  }
+
+  // Manifests
+
+  async getManifestKeys() {
+    let years = (await this.list('', '/')).prefixes.filter(prefix => /^\d{4}\/$/.test(prefix));
+    let yearListings = await Promise.all(years.map(year => {
+      return requestLimit(() => this.list(year, '/'));
+    }));
+    let months = yearListings.flatMap(listing => listing.prefixes)
+      .filter(prefix => /^\d{4}\/\d{2}\/$/.test(prefix));
+    let monthListings = await Promise.all(months.map(month => {
+      return requestLimit(() => this.list(month, '/'));
+    }));
+
+    return monthListings.flatMap(listing => listing.objects)
+      .map(object => object.Key)
+      .filter(key => key && key.endsWith(manifestSuffix));
+  }
+
+  async readManifest(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}`);
+    }
+
+    let archivePath = key.slice(0, -'.manifest.json'.length);
+    let manifest = parseArchiveManifest(await response.Body.transformToString(), archivePath);
+
+    return getArchiveManifestStats(manifest);
+  }
+
+  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 };
+  }
+
+  // Formatting
+
+  describeArchive(archive) {
+    let savedBytes = archive.originalBytes - archive.compressedBytes;
+
+    return `${archive.path}: ${archive.fileCount} ${archive.fileCount === 1 ? 'file' : 'files'}, `
+      + `${this.formatBytes(archive.originalBytes)} -> ${this.formatBytes(archive.compressedBytes)}, `
+      + `${this.formatPercent(savedBytes, archive.originalBytes)} saved, `
+      + `${this.formatRatio(archive.originalBytes, archive.compressedBytes)}:1`;
+  }
+
+  formatBytes(bytes) {
+    let units = ['B', 'KiB', 'MiB', 'GiB', 'TiB'];
+    let unit = 0;
+    let value = bytes;
+
+    while (Math.abs(value) >= 1024 && unit < units.length - 1) {
+      value /= 1024;
+      unit++;
+    }
+
+    return `${new Intl.NumberFormat('en-US', { maximumFractionDigits: 2 }).format(value)} ${units[unit]}`;
+  }
+
+  formatPercent(part, total) {
+    if (total === 0) {
+      return '0%';
+    }
+
+    return `${new Intl.NumberFormat('en-US', { maximumFractionDigits: 2 }).format(part / total * 100)}%`;
+  }
+
+  formatRatio(originalBytes, compressedBytes) {
+    if (compressedBytes === 0) {
+      return '0';
+    }
+
+    return new Intl.NumberFormat('en-US', { maximumFractionDigits: 1 })
+      .format(originalBytes / compressedBytes);
+  }
+}

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)}`);
+    }
+  }
+}

app/index.mjs

@@ -9,6 +9,8 @@ import ImageWriter from './social/clients/ImageWriter.mjs';
 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 { reportArchiveStatsFromCli } from './data/ArchiveStats.mjs';
 import { sentryInit } from './common/sentry.mjs';
 import { sync, syncUpload, syncDownload } from './sync/index.mjs';
 import { updateAvatars } from './social/updateAvatars.mjs';
@@ -28,6 +30,8 @@ const actions = {
   splatnet: update,
   warmCaches,
   dataArchive: archiveData,
+  archiveCompress: (...args) => compressArchivesFromCli(args),
+  archiveStats: (...args) => reportArchiveStatsFromCli(args),
   sync,
   syncUpload,
   syncDownload,
@@ -38,8 +42,7 @@ const command = process.argv[2];
 const params = process.argv.slice(3);
 const action = actions[command];
 if (action) {
-  action(...params);
+  await action(...params);
 } else {
   console.error(`Unrecognized command: ${command}`);
 }
-

docker/app/Dockerfile

@@ -4,6 +4,11 @@ FROM node:22
 WORKDIR /app
 ENV PUPPETEER_SKIP_DOWNLOAD=true
 
+# Archive compression
+RUN apt-get update \
+  && apt-get install -y --no-install-recommends zstd \
+  && rm -rf /var/lib/apt/lists/*
+
 # Install NPM dependencies
 COPY package*.json ./
 RUN npm ci

package.json

@@ -23,6 +23,9 @@
     "splatnet:all": "node app/index.mjs splatnet all",
     "warmCaches": "node app/index.mjs warmCaches",
     "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:stats": "node app/index.mjs archiveStats",
     "sync": "node app/index.mjs sync",
     "sync:upload": "node app/index.mjs syncUpload",
     "sync:download": "node app/index.mjs syncDownload"