diff --git a/app/cron.mjs b/app/cron.mjs index 2a18eb2..0581c07 100644 --- a/app/cron.mjs +++ b/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); } diff --git a/app/data/ArchiveGenerator.mjs b/app/data/ArchiveGenerator.mjs new file mode 100644 index 0000000..cde67cc --- /dev/null +++ b/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; + } +} diff --git a/app/index.mjs b/app/index.mjs index 67bb15e..49dadbb 100644 --- a/app/index.mjs +++ b/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}`); } - diff --git a/docker/app/Dockerfile b/docker/app/Dockerfile index 412506b..5367c78 100644 --- a/docker/app/Dockerfile +++ b/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 diff --git a/package.json b/package.json index a98b70b..2b67305 100644 --- a/package.json +++ b/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"