splatoon3.ink/app/data/ArchiveBuilder.mjs

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 {
  GetObjectCommand,
  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;

export default class ArchiveBuilder
{
  constructor(s3Client, archiveWriter = new TarZstdWriter) {
    this._client = s3Client;
    this.archiveWriter = archiveWriter;
  }

  async build(candidate, objects) {
    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(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);

      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',
      );
    } finally {
      await fs.rm(temporaryDirectory, { force: true, recursive: true });
    }
  }

  get console() {
    this._console ??= prefixedConsole('Archive Builder');

    return this._console;
  }

  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 downloadObjects(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(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(object, sourceDirectory) {
    let destination = path.join(sourceDirectory, object.path);
    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: object.path,
      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,
    }));
  }

}