Add daily archive generator

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 { generateArchives } from './data/ArchiveGenerator.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 * * *', () => generateArchives(10), null, true, 'UTC');
   new CronJob('0 55 4 * * *', restartToRefreshConfig, null, true);
 }

app/data/ArchiveGenerator.mjs

@@ -0,0 +1,396 @@
+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 { spawn } from 'node:child_process';
+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';
+
+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 generateArchives(maxDays = Infinity) {
+  let generator = new ArchiveGenerator;
+  generator.maxDays = maxDays;
+
+  return generator.process();
+}
+
+export function generateArchivesFromCli(args) {
+  if (args.length === 0) {
+    return generateArchives();
+  }
+
+  if (args.length !== 2 || args[0] !== '--max-days' || !/^\d+$/.test(args[1])) {
+    throw new Error('Usage: npm run data:archive:generate -- [--max-days DAYS]');
+  }
+
+  let maxDays = Number(args[1]);
+  if (maxDays < 1) {
+    throw new Error('--max-days must be at least 1');
+  }
+
+  return generateArchives(maxDays);
+}
+
+export default class ArchiveGenerator
+{
+  maxDays = Infinity;
+
+  async process() {
+    if (!this.canRun) {
+      this.console.log('Skipping archive generator');
+
+      return;
+    }
+
+    let currentDate;
+    let generated = 0;
+
+    try {
+      let candidates = await this.getCandidates();
+      this.console.log(`Found ${candidates.length} dates to archive`);
+
+      for (let candidate of candidates) {
+        if (generated >= this.maxDays) {
+          break;
+        }
+
+        currentDate = candidate.date;
+        if (await this.archiveDate(candidate)) {
+          generated++;
+        }
+      }
+    } 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(`Generated ${generated} archives`);
+  }
+
+  // Properties
+
+  get console() {
+    this._console ??= prefixedConsole('Archive Generator');
+
+    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 generation
+
+  async archiveDate(candidate) {
+    let objects = await this.getSourceObjects(candidate.prefix);
+    if (objects.length === 0) {
+      return false;
+    }
+
+    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 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.compress(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,
+      };
+
+      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 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 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)) {
+      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/index.mjs

@@ -9,6 +9,7 @@ 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 { generateArchivesFromCli } from './data/ArchiveGenerator.mjs';
 import { sentryInit } from './common/sentry.mjs';
 import { sync, syncUpload, syncDownload } from './sync/index.mjs';
 import { updateAvatars } from './social/updateAvatars.mjs';
@@ -28,6 +29,7 @@ const actions = {
   splatnet: update,
   warmCaches,
   dataArchive: archiveData,
+  archiveGenerate: (...args) => generateArchivesFromCli(args),
   sync,
   syncUpload,
   syncDownload,
@@ -38,8 +40,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,7 @@
     "splatnet:all": "node app/index.mjs splatnet all",
     "warmCaches": "node app/index.mjs warmCaches",
     "data:archive": "node app/index.mjs dataArchive",
+    "data:archive:generate": "node app/index.mjs archiveGenerate",
     "sync": "node app/index.mjs sync",
     "sync:upload": "node app/index.mjs syncUpload",
     "sync:download": "node app/index.mjs syncDownload"