diff --git a/scripts/sync-releases.ts b/scripts/sync-releases.ts index c1fa91f..6a2edb4 100644 --- a/scripts/sync-releases.ts +++ b/scripts/sync-releases.ts @@ -8,54 +8,28 @@ import { Readable } from "node:stream"; import { pipeline } from "node:stream/promises"; import { createInterface } from "node:readline/promises"; -import { - GetObjectCommand, - HeadObjectCommand, - ListObjectsV2Command, - S3Client, -} from "@aws-sdk/client-s3"; +import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3"; import { PrismaClient } from "@prisma/client"; import semver from "semver"; -import { objectKeyFromArtifactUrl, streamToString } from "../src/helpers"; -import { OTA_PREFIXES, legacyCompatibleSkus, otaFileForPrefix, skusForPrefix } from "../src/skus"; - -/** An R2 prefix, which is also the Release.type column value. */ -type ReleaseType = string; +import { objectKeyFromArtifactUrl } from "../src/helpers"; +import { baseUrl, bucketName, s3Client, s3ObjectExists } from "../src/s3"; +import { + DEFAULT_ROLLOUT_PERCENTAGE, + createAtDefaultRollout, + syncReleases, + type ReleaseArtifactInput, + type ReleaseDecider, + type ReleaseType, + type SyncClients, + type SyncConfig, +} from "../src/release-sync"; + +// Operator front end for src/release-sync.ts: signature verification output +// and a confirmation prompt before each production DB write. const OTA_ROOT_KEY_FPR = "AF5A36A993D828FEFE7C18C2D1B9856C26A79E95"; -interface SyncClients { - prisma: PrismaClient; - s3Client: S3Client; -} - -interface SyncConfig { - bucketName: string; - baseUrl: string; - skus?: string[]; -} - -interface ReleaseArtifactInput { - url: string; - hash: string; - compatibleSkus: string[]; -} - -const DEFAULT_ROLLOUT_PERCENTAGE = 10; - -type ReleaseOutcome = - | "created" - | "already-synced" - | "no-artifacts" - | "skipped" - | "aborted"; - -type ReleaseDecision = - | { kind: "create"; rolloutPercentage: number } - | { kind: "skip" } - | { kind: "abort" }; - interface LatestExistingRelease { version: string; rolloutPercentage: number; @@ -360,305 +334,41 @@ async function promptRolloutPercentage( } } -async function confirmProductionCreate( +function confirmProductionCreate( clients: SyncClients, config: SyncConfig, - type: ReleaseType, - version: string, - artifacts: ReleaseArtifactInput[], -): Promise { - if (process.env.NODE_ENV !== "production") { - return { kind: "create", rolloutPercentage: DEFAULT_ROLLOUT_PERCENTAGE }; - } - - if (!stdin.isTTY || !stdout.isTTY) { - throw new Error( - "Production release sync requires an interactive terminal for DB write confirmation.", - ); - } - - const [artifactInfos, latestExisting] = await Promise.all([ - loadArtifactDisplayInfo(clients, config, artifacts), - findLatestExistingRelease(clients.prisma, type), - ]); +): ReleaseDecider { + return async (type, version, artifacts) => { + const [artifactInfos, latestExisting] = await Promise.all([ + loadArtifactDisplayInfo(clients, config, artifacts), + findLatestExistingRelease(clients.prisma, type), + ]); - printArtifactSummary(type, version, artifactInfos, latestExisting); + printArtifactSummary(type, version, artifactInfos, latestExisting); - const readline = createInterface({ input: stdin, output: stdout }); - try { - const rolloutPercentage = await promptRolloutPercentage(readline); + const readline = createInterface({ input: stdin, output: stdout }); + try { + const rolloutPercentage = await promptRolloutPercentage(readline); - const confirmation = ( - await readline.question( - ` Create production ${type} release ${version} at ${rolloutPercentage}% rollout? [y/N/a (abort run)] `, + const confirmation = ( + await readline.question( + ` Create production ${type} release ${version} at ${rolloutPercentage}% rollout? [y/N/a (abort run)] `, + ) ) - ) - .trim() - .toLowerCase(); - - if (["a", "abort"].includes(confirmation)) { - return { kind: "abort" }; - } - if (!["y", "yes"].includes(confirmation)) { - return { kind: "skip" }; - } - return { kind: "create", rolloutPercentage }; - } finally { - readline.close(); - } -} - -function isS3NotFound(error: any): boolean { - return ( - error.name === "NotFound" || - error.name === "NoSuchKey" || - error.$metadata?.httpStatusCode === 404 - ); -} - -async function s3ObjectExists( - s3Client: S3Client, - bucketName: string, - key: string, -): Promise { - try { - await s3Client.send(new HeadObjectCommand({ Bucket: bucketName, Key: key })); - return true; - } catch (error: any) { - if (isS3NotFound(error)) { - return false; - } - throw error; - } -} - -async function versionHasSkuSupport( - s3Client: S3Client, - bucketName: string, - type: ReleaseType, - version: string, -): Promise { - const response = await s3Client.send( - new ListObjectsV2Command({ - Bucket: bucketName, - Prefix: `${type}/${version}/skus/`, - MaxKeys: 1, - }), - ); - return (response.Contents?.length ?? 0) > 0; -} - -async function readHash( - s3Client: S3Client, - bucketName: string, - artifactPath: string, -): Promise { - try { - const response = await s3Client.send( - new GetObjectCommand({ - Bucket: bucketName, - Key: `${artifactPath}.sha256`, - }), - ); - return streamToString(response.Body); - } catch (error: any) { - if (isS3NotFound(error)) { - return undefined; - } - throw error; - } -} - -function addArtifact( - artifactsByUrl: Map, - url: string, - hash: string, - sku: string, -): void { - const artifact = artifactsByUrl.get(url); - if (artifact) { - if (!artifact.compatibleSkus.includes(sku)) { - artifact.compatibleSkus.push(sku); - } - return; - } - - artifactsByUrl.set(url, { url, hash, compatibleSkus: [sku] }); -} - -export async function collectReleaseArtifacts( - clients: Pick, - config: SyncConfig, - type: ReleaseType, - version: string, -): Promise { - const skus = config.skus ?? skusForPrefix(type); - const artifactFileName = otaFileForPrefix(type); - - if (!(await versionHasSkuSupport(clients.s3Client, config.bucketName, type, version))) { - // Pre-SKU artifacts (no skus/ folder) are only safe on the SKUs that - // predate the layout. A type with no legacy form treats a version - // without skus/ as an upload mistake, not a release. - const compatibleSkus = legacyCompatibleSkus(type); - if (compatibleSkus.length === 0) { - return []; - } - - const artifactPath = `${type}/${version}/${artifactFileName}`; - const hash = await readHash(clients.s3Client, config.bucketName, artifactPath); - if (!hash) { - return []; - } - - return [ - { - url: `${config.baseUrl}/${artifactPath}`, - hash, - compatibleSkus, - }, - ]; - } - - const artifactsByUrl = new Map(); - for (const sku of skus) { - const artifactPath = `${type}/${version}/skus/${sku}/${artifactFileName}`; - if (!(await s3ObjectExists(clients.s3Client, config.bucketName, artifactPath))) { - continue; - } - - const hash = await readHash(clients.s3Client, config.bucketName, artifactPath); - if (!hash) { - continue; - } - addArtifact(artifactsByUrl, `${config.baseUrl}/${artifactPath}`, hash, sku); - } - - return Array.from(artifactsByUrl.values()); -} - -async function listStableVersions( - s3Client: S3Client, - bucketName: string, - type: ReleaseType, -): Promise { - const response = await s3Client.send( - new ListObjectsV2Command({ - Bucket: bucketName, - Prefix: `${type}/`, - Delimiter: "/", - }), - ); - - return (response.CommonPrefixes ?? []) - .map(cp => cp.Prefix?.split("/")[1]) - .filter((version): version is string => Boolean(version)) - .filter( - version => Boolean(semver.valid(version)) && semver.prerelease(version) === null, - ) - .sort(semver.compare); -} - -async function syncRelease( - clients: SyncClients, - config: SyncConfig, - type: ReleaseType, - version: string, - artifacts: ReleaseArtifactInput[], -): Promise { - if (artifacts.length === 0) { - console.log(`[sync-releases] ${type} ${version}: skipped, no compatible artifacts`); - return "no-artifacts"; - } - - // Sync only registers brand-new releases. Existing rows (rollout state, URLs, - // artifact compatibility) are left untouched — backfills/repairs are handled - // by one-off scripts so a routine sync run can never rewrite production data. - const existing = await clients.prisma.release.findUnique({ - where: { version_type: { version, type } }, - select: { id: true }, - }); - - if (existing) { - console.log(`[sync-releases] ${type} ${version}: already synced, skipping`); - return "already-synced"; - } - - const decision = await confirmProductionCreate( - clients, - config, - type, - version, - artifacts, - ); - if (decision.kind === "abort") { - console.log(`[sync-releases] ${type} ${version}: aborted by user`); - return "aborted"; - } - if (decision.kind === "skip") { - console.log(`[sync-releases] ${type} ${version}: skipped by user`); - return "skipped"; - } - - const primaryArtifact = artifacts[0]; - await clients.prisma.release.create({ - data: { - version, - type, - rolloutPercentage: decision.rolloutPercentage, - url: primaryArtifact.url, - hash: primaryArtifact.hash, - artifacts: { - create: artifacts.map(artifact => ({ - url: artifact.url, - hash: artifact.hash, - compatibleSkus: artifact.compatibleSkus, - })), - }, - }, - }); - - console.log( - `[sync-releases] ${type} ${version}: created with ${artifacts.length} artifact(s) at ${decision.rolloutPercentage}% rollout`, - ); - return "created"; -} - -export async function syncReleases( - clients: SyncClients, - config: SyncConfig, -): Promise { - const stats: Record = { - created: 0, - "already-synced": 0, - "no-artifacts": 0, - skipped: 0, - aborted: 0, - }; - let abortedAt: { type: ReleaseType; version: string } | null = null; - - outer: for (const type of OTA_PREFIXES) { - const versions = await listStableVersions(clients.s3Client, config.bucketName, type); - - for (const version of versions) { - const artifacts = await collectReleaseArtifacts(clients, config, type, version); - const outcome = await syncRelease(clients, config, type, version, artifacts); - stats[outcome]++; + .trim() + .toLowerCase(); - if (outcome === "aborted") { - abortedAt = { type, version }; - break outer; + if (["a", "abort"].includes(confirmation)) { + return { kind: "abort" }; } + if (!["y", "yes"].includes(confirmation)) { + return { kind: "skip" }; + } + return { kind: "create", rolloutPercentage }; + } finally { + readline.close(); } - } - - if (abortedAt) { - console.log( - `[sync-releases] aborted at ${abortedAt.type} ${abortedAt.version}; remaining versions in this run were not processed`, - ); - } - console.log( - `[sync-releases] done: created=${stats.created} skipped-by-user=${stats.skipped} already-synced=${stats["already-synced"]} no-artifacts=${stats["no-artifacts"]}`, - ); + }; } function describeDbTarget(): string { @@ -678,35 +388,32 @@ function describeDbTarget(): string { async function main(): Promise { console.log( - `[sync-releases] env=${process.env.NODE_ENV ?? "(unset)"} db=${describeDbTarget()} bucket=${process.env.R2_BUCKET ?? "(unset)"}`, + `[sync-releases] env=${process.env.NODE_ENV ?? "(unset)"} db=${describeDbTarget()} bucket=${bucketName ?? "(unset)"}`, ); + const isProduction = process.env.NODE_ENV === "production"; + if (isProduction && (!stdin.isTTY || !stdout.isTTY)) { + throw new Error( + "Production release sync requires an interactive terminal for DB write confirmation.", + ); + } + const prisma = new PrismaClient(); - const s3Client = new S3Client({ - endpoint: process.env.R2_ENDPOINT!, - credentials: { - accessKeyId: process.env.R2_ACCESS_KEY_ID!, - secretAccessKey: process.env.R2_SECRET_ACCESS_KEY!, - }, - region: "auto", - }); + const clients: SyncClients = { prisma, s3Client }; + const config: SyncConfig = { bucketName, baseUrl }; try { await syncReleases( - { prisma, s3Client }, - { - bucketName: process.env.R2_BUCKET!, - baseUrl: process.env.R2_CDN_URL!, - }, + clients, + config, + isProduction ? confirmProductionCreate(clients, config) : createAtDefaultRollout, ); } finally { await prisma.$disconnect(); } } -if (require.main === module) { - main().catch(error => { - console.error("[sync-releases] failed", error); - process.exit(1); - }); -} +main().catch(error => { + console.error("[sync-releases] failed", error); + process.exit(1); +}); diff --git a/src/index.ts b/src/index.ts index b6fdecb..2a235c1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,6 +13,8 @@ import * as Releases from "./releases"; import { HttpError } from "./errors"; import { authenticated } from "./auth"; import { prisma } from "./db"; +import { baseUrl, bucketName, s3Client } from "./s3"; +import { scheduleReleaseSync } from "./release-sync"; import { initializeWebRTCSignaling } from "./webrtc-signaling"; declare global { @@ -218,3 +220,6 @@ const server = app.listen(PORT, () => { }); initializeWebRTCSignaling(server); + +// Register new R2 releases at the default rollout, now and every 30 minutes. +scheduleReleaseSync({ prisma, s3Client }, { bucketName, baseUrl }); diff --git a/src/release-sync.ts b/src/release-sync.ts new file mode 100644 index 0000000..2911427 --- /dev/null +++ b/src/release-sync.ts @@ -0,0 +1,390 @@ +import { GetObjectCommand, S3Client, paginateListObjectsV2 } from "@aws-sdk/client-s3"; +import { Prisma, PrismaClient } from "@prisma/client"; +import semver from "semver"; + +import { streamToString } from "./helpers"; +import { isS3NotFound, s3ObjectExists, versionHasSkuSupport } from "./s3"; +import { OTA_PREFIXES, legacyCompatibleSkus, otaFileForPrefix, skusForPrefix } from "./skus"; + +/** An R2 prefix, which is also the Release.type column value. */ +export type ReleaseType = string; + +export interface SyncClients { + prisma: PrismaClient; + s3Client: S3Client; +} + +export interface SyncConfig { + bucketName: string; + baseUrl: string; + skus?: string[]; + /** + * Defer a version whose newest object changed within this many ms, checked + * after the artifact scan: the upload script writes several files per SKU, + * and sync never rewrites a row, so a scan that overlapped an upload would + * freeze a partial SKU set or a stale hash. Unset or 0 disables the check + * (an operator at a terminal can judge the artifact list for themselves). + */ + uploadSettleMs?: number; +} + +export interface ReleaseArtifactInput { + url: string; + hash: string; + compatibleSkus: string[]; +} + +export const DEFAULT_ROLLOUT_PERCENTAGE = 10; + +export type ReleaseOutcome = + | "created" + | "already-synced" + | "uploading" + | "no-artifacts" + | "skipped" + | "aborted"; + +/** Settle window for unattended runs; shorter than the interval, so it costs at most one tick. */ +export const UPLOAD_SETTLE_MS = 10 * 60 * 1000; + +export type ReleaseDecision = + | { kind: "create"; rolloutPercentage: number } + | { kind: "skip" } + | { kind: "abort" }; + +/** + * Decides whether a release that exists in R2 but not in the DB gets created. + * The interactive script asks the operator; the in-process scheduler always + * creates at the default rollout. + */ +export type ReleaseDecider = ( + type: ReleaseType, + version: string, + artifacts: ReleaseArtifactInput[], +) => Promise; + +export const createAtDefaultRollout: ReleaseDecider = async () => ({ + kind: "create", + rolloutPercentage: DEFAULT_ROLLOUT_PERCENTAGE, +}); + +async function readHash( + s3Client: S3Client, + bucketName: string, + artifactPath: string, +): Promise { + try { + const response = await s3Client.send( + new GetObjectCommand({ + Bucket: bucketName, + Key: `${artifactPath}.sha256`, + }), + ); + return streamToString(response.Body); + } catch (error: any) { + if (isS3NotFound(error)) { + return undefined; + } + throw error; + } +} + +function addArtifact( + artifactsByUrl: Map, + url: string, + hash: string, + sku: string, +): void { + const artifact = artifactsByUrl.get(url); + if (artifact) { + if (!artifact.compatibleSkus.includes(sku)) { + artifact.compatibleSkus.push(sku); + } + return; + } + + artifactsByUrl.set(url, { url, hash, compatibleSkus: [sku] }); +} + +export async function collectReleaseArtifacts( + clients: Pick, + config: SyncConfig, + type: ReleaseType, + version: string, +): Promise { + const skus = config.skus ?? skusForPrefix(type); + const artifactFileName = otaFileForPrefix(type); + + if (!(await versionHasSkuSupport(clients.s3Client, config.bucketName, type, version))) { + // Pre-SKU artifacts (no skus/ folder) are only safe on the SKUs that + // predate the layout. A type with no legacy form treats a version + // without skus/ as an upload mistake, not a release. + const compatibleSkus = legacyCompatibleSkus(type); + if (compatibleSkus.length === 0) { + return []; + } + + const artifactPath = `${type}/${version}/${artifactFileName}`; + const hash = await readHash(clients.s3Client, config.bucketName, artifactPath); + if (!hash) { + return []; + } + + return [ + { + url: `${config.baseUrl}/${artifactPath}`, + hash, + compatibleSkus, + }, + ]; + } + + // SKUs are probed concurrently; folding afterwards in `skus` order keeps + // the primary artifact (artifacts[0]) and compatibleSkus order stable. + const found = await Promise.all( + skus.map(async sku => { + const artifactPath = `${type}/${version}/skus/${sku}/${artifactFileName}`; + if (!(await s3ObjectExists(clients.s3Client, config.bucketName, artifactPath))) { + return undefined; + } + const hash = await readHash(clients.s3Client, config.bucketName, artifactPath); + return hash ? { sku, url: `${config.baseUrl}/${artifactPath}`, hash } : undefined; + }), + ); + + const artifactsByUrl = new Map(); + for (const artifact of found) { + if (artifact) { + addArtifact(artifactsByUrl, artifact.url, artifact.hash, artifact.sku); + } + } + return Array.from(artifactsByUrl.values()); +} + +async function listStableVersions( + s3Client: S3Client, + bucketName: string, + type: ReleaseType, +): Promise { + const prefixes: string[] = []; + for await (const page of paginateListObjectsV2( + { client: s3Client }, + { Bucket: bucketName, Prefix: `${type}/`, Delimiter: "/" }, + )) { + for (const cp of page.CommonPrefixes ?? []) { + if (cp.Prefix) { + prefixes.push(cp.Prefix); + } + } + } + + return prefixes + .map(prefix => prefix.split("/")[1]) + .filter((version): version is string => Boolean(version)) + .filter( + version => Boolean(semver.valid(version)) && semver.prerelease(version) === null, + ) + .sort(semver.compare); +} + +/** Newest LastModified among every object under `${type}/${version}/`, or undefined when empty. */ +async function newestUploadTime( + s3Client: S3Client, + bucketName: string, + type: ReleaseType, + version: string, +): Promise { + let newest: Date | undefined; + for await (const page of paginateListObjectsV2( + { client: s3Client }, + { Bucket: bucketName, Prefix: `${type}/${version}/` }, + )) { + for (const object of page.Contents ?? []) { + if (object.LastModified && (!newest || object.LastModified > newest)) { + newest = object.LastModified; + } + } + } + return newest; +} + +async function listSyncedVersions(prisma: PrismaClient, type: ReleaseType): Promise> { + const releases = await prisma.release.findMany({ + where: { type }, + select: { version: true }, + }); + return new Set(releases.map(release => release.version)); +} + +function isUniqueViolation(error: unknown): boolean { + return ( + error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002" + ); +} + +async function createRelease( + clients: SyncClients, + config: SyncConfig, + decide: ReleaseDecider, + type: ReleaseType, + version: string, +): Promise { + const artifacts = await collectReleaseArtifacts(clients, config, type, version); + + // Listed after the artifact scan on purpose: an upload that was active at + // any point during the scan leaves an object newer than the window, so the + // snapshot above is discarded rather than registered. + if (config.uploadSettleMs) { + const newest = await newestUploadTime(clients.s3Client, config.bucketName, type, version); + if (newest && Date.now() - newest.getTime() < config.uploadSettleMs) { + console.log( + `[sync-releases] ${type} ${version}: upload still settling (last object ${newest.toISOString()}), retrying next run`, + ); + return "uploading"; + } + } + + if (artifacts.length === 0) { + console.log(`[sync-releases] ${type} ${version}: skipped, no compatible artifacts`); + return "no-artifacts"; + } + + const decision = await decide(type, version, artifacts); + if (decision.kind === "abort") { + console.log(`[sync-releases] ${type} ${version}: aborted by user`); + return "aborted"; + } + if (decision.kind === "skip") { + console.log(`[sync-releases] ${type} ${version}: skipped by user`); + return "skipped"; + } + + const primaryArtifact = artifacts[0]; + try { + await clients.prisma.release.create({ + data: { + version, + type, + rolloutPercentage: decision.rolloutPercentage, + url: primaryArtifact.url, + hash: primaryArtifact.hash, + artifacts: { + create: artifacts.map(artifact => ({ + url: artifact.url, + hash: artifact.hash, + compatibleSkus: artifact.compatibleSkus, + })), + }, + }, + }); + } catch (error) { + // Another API instance can win the race between the version listing and + // this insert. The row it wrote is the one we wanted, so treat it as synced. + if (isUniqueViolation(error)) { + console.log(`[sync-releases] ${type} ${version}: created concurrently elsewhere, skipping`); + return "already-synced"; + } + throw error; + } + + console.log( + `[sync-releases] ${type} ${version}: created with ${artifacts.length} artifact(s) at ${decision.rolloutPercentage}% rollout`, + ); + return "created"; +} + +/** + * Registers every stable version in R2 that has no Release row yet. + * + * Sync only registers brand-new releases. Existing rows (rollout state, URLs, + * artifact compatibility) are left untouched — backfills/repairs are handled + * by one-off scripts so a routine sync run can never rewrite production data. + * Known versions are loaded once per prefix, so a run over an already-synced + * bucket costs one R2 list and one DB query per prefix. + * + * Returns the per-outcome counts so callers can log or assert on them. + */ +export async function syncReleases( + clients: SyncClients, + config: SyncConfig, + decide: ReleaseDecider, +): Promise> { + const stats: Record = { + created: 0, + "already-synced": 0, + uploading: 0, + "no-artifacts": 0, + skipped: 0, + aborted: 0, + }; + let abortedAt: { type: ReleaseType; version: string } | null = null; + + outer: for (const type of OTA_PREFIXES) { + const [versions, synced] = await Promise.all([ + listStableVersions(clients.s3Client, config.bucketName, type), + listSyncedVersions(clients.prisma, type), + ]); + + for (const version of versions) { + const outcome = synced.has(version) + ? "already-synced" + : await createRelease(clients, config, decide, type, version); + stats[outcome]++; + + if (outcome === "aborted") { + abortedAt = { type, version }; + break outer; + } + } + } + + if (abortedAt) { + console.log( + `[sync-releases] aborted at ${abortedAt.type} ${abortedAt.version}; remaining versions in this run were not processed`, + ); + } + console.log( + `[sync-releases] done: created=${stats.created} skipped-by-user=${stats.skipped} already-synced=${stats["already-synced"]} uploading=${stats.uploading} no-artifacts=${stats["no-artifacts"]}`, + ); + return stats; +} + +const RELEASE_SYNC_INTERVAL_MS = 30 * 60 * 1000; + +/** + * Runs the sync now and then every `intervalMs`, inside the API process. + * A tick that fires while the previous run is still going is skipped, not + * queued. A failed run is logged and the schedule continues, so one bad R2 + * or DB response never stops future syncs. + * Returns a function that stops the schedule. + */ +export function scheduleReleaseSync( + clients: SyncClients, + config: SyncConfig, + intervalMs: number = RELEASE_SYNC_INTERVAL_MS, +): () => void { + let running = false; + + const run = async () => { + if (running) { + console.warn("[sync-releases] previous run still in progress, skipping this tick"); + return; + } + running = true; + try { + await syncReleases( + clients, + { uploadSettleMs: UPLOAD_SETTLE_MS, ...config }, + createAtDefaultRollout, + ); + } catch (error) { + console.error("[sync-releases] scheduled run failed", error); + } finally { + running = false; + } + }; + + void run(); + const timer = setInterval(run, intervalMs); + return () => clearInterval(timer); +} diff --git a/src/releases.ts b/src/releases.ts index fb0293f..33d5323 100644 --- a/src/releases.ts +++ b/src/releases.ts @@ -3,13 +3,9 @@ import { prisma } from "./db"; import { BadRequestError, InternalServerError, NotFoundError } from "./errors"; import semver from "semver"; -import { - GetObjectCommand, - HeadObjectCommand, - ListObjectsV2Command, - S3Client, -} from "@aws-sdk/client-s3"; +import { GetObjectCommand, ListObjectsV2Command } from "@aws-sdk/client-s3"; import { LRUCache } from "lru-cache"; +import { baseUrl, bucketName, s3Client, s3ObjectExists, versionHasSkuSupport } from "./s3"; import { getDeviceRolloutBucket, @@ -101,15 +97,6 @@ interface DbRelease { }[]; } -const s3Client = new S3Client({ - endpoint: process.env.R2_ENDPOINT!, - credentials: { - accessKeyId: process.env.R2_ACCESS_KEY_ID!, - secretAccessKey: process.env.R2_SECRET_ACCESS_KEY!, - }, - region: "auto", -}); - const releaseCache = new LRUCache({ max: 1000, ttl: 5 * 60 * 1000, // 5 minutes @@ -134,9 +121,6 @@ export function clearCaches() { sigUrlCache.clear(); } -const bucketName = process.env.R2_BUCKET; -const baseUrl = process.env.R2_CDN_URL; - /** * The one error for "this version ships no artifact for this SKU", whichever * path detects it: a skus/ folder without this SKU, a pre-SKU version asked @@ -146,41 +130,7 @@ function noArtifactForSku(version: string, sku: string): NotFoundError { return new NotFoundError(`Version ${version} has no artifact for SKU "${sku}"`); } -/** - * Checks if an object exists in S3/R2 by attempting a HeadObjectCommand. - * Returns true if the object exists, false otherwise. - */ -async function s3ObjectExists(key: string): Promise { - try { - await s3Client.send(new HeadObjectCommand({ Bucket: bucketName, Key: key })); - return true; - } catch (error: any) { - // HeadObjectCommand throws NotFound, but some S3-compatible stores (like R2) may throw NoSuchKey - if ( - error.name === "NotFound" || - error.name === "NoSuchKey" || - error.$metadata?.httpStatusCode === 404 - ) { - return false; - } - throw error; - } -} - -/** - * Checks if a version was uploaded with SKU folder structure. - * Returns true if any skus/ subfolder exists for this version. - */ -async function versionHasSkuSupport(prefix: string, version: string): Promise { - const response = await s3Client.send( - new ListObjectsV2Command({ - Bucket: bucketName, - Prefix: `${prefix}/${version}/skus/`, - MaxKeys: 1, - }), - ); - return (response.Contents?.length ?? 0) > 0; -} +const objectExists = (key: string) => s3ObjectExists(s3Client, bucketName, key); /** * Resolves the artifact path for a given version and SKU. @@ -205,10 +155,10 @@ async function resolveArtifactPath( sku: string, file: string, ): Promise { - if (await versionHasSkuSupport(prefix, version)) { + if (await versionHasSkuSupport(s3Client, bucketName, prefix, version)) { const skuPath = `${prefix}/${version}/skus/${sku}/${file}`; - if (await s3ObjectExists(skuPath)) { + if (await objectExists(skuPath)) { return skuPath; } @@ -243,7 +193,7 @@ async function resolveSigUrl( try { const path = await resolveArtifactPath(artifact.prefix, version, sku, artifact.file); const sigKey = `${path}.sig`; - if (await s3ObjectExists(sigKey)) { + if (await objectExists(sigKey)) { const url = `${baseUrl}/${sigKey}`; sigUrlCache.set(cacheKey, url); return url; @@ -389,7 +339,7 @@ async function resolveSigUrlFromArtifactUrl( const sigUrl = `${artifactUrl}.sig`; try { const sigKey = `${objectKeyFromArtifactUrl(artifactUrl)}.sig`; - if (await s3ObjectExists(sigKey)) { + if (await objectExists(sigKey)) { sigUrlCache.set(cacheKey, sigUrl); return sigUrl; } @@ -693,7 +643,7 @@ export const RetrieveLatestSystemRecovery = cachedRedirect( recovery.file, ); - if (!(await s3ObjectExists(artifactPath))) { + if (!(await objectExists(artifactPath))) { throw new NotFoundError(`Recovery image not found for version ${latestVersion}`); } @@ -750,7 +700,7 @@ function latestArtifactRedirect(kind: OtaKind) { artifact.file, ); - if (!(await s3ObjectExists(artifactPath))) { + if (!(await objectExists(artifactPath))) { throw new NotFoundError(`${prefix} artifact not found for version ${latestVersion}`); } diff --git a/src/s3.ts b/src/s3.ts new file mode 100644 index 0000000..1c702bb --- /dev/null +++ b/src/s3.ts @@ -0,0 +1,57 @@ +import { HeadObjectCommand, ListObjectsV2Command, S3Client } from "@aws-sdk/client-s3"; + +/** The one R2 client the API process shares between request handlers and background jobs. */ +export const s3Client = new S3Client({ + endpoint: process.env.R2_ENDPOINT!, + credentials: { + accessKeyId: process.env.R2_ACCESS_KEY_ID!, + secretAccessKey: process.env.R2_SECRET_ACCESS_KEY!, + }, + region: "auto", +}); + +/** Bucket that holds release artifacts, and the public CDN origin they are served from. */ +export const bucketName = process.env.R2_BUCKET!; +export const baseUrl = process.env.R2_CDN_URL!; + +/** HeadObject throws NotFound, but some S3-compatible stores (like R2) may throw NoSuchKey. */ +export function isS3NotFound(error: any): boolean { + return ( + error.name === "NotFound" || + error.name === "NoSuchKey" || + error.$metadata?.httpStatusCode === 404 + ); +} + +export async function s3ObjectExists( + client: S3Client, + bucketName: string, + key: string, +): Promise { + try { + await client.send(new HeadObjectCommand({ Bucket: bucketName, Key: key })); + return true; + } catch (error: any) { + if (isS3NotFound(error)) { + return false; + } + throw error; + } +} + +/** True when the version was uploaded with the skus// folder layout. */ +export async function versionHasSkuSupport( + client: S3Client, + bucketName: string, + prefix: string, + version: string, +): Promise { + const response = await client.send( + new ListObjectsV2Command({ + Bucket: bucketName, + Prefix: `${prefix}/${version}/skus/`, + MaxKeys: 1, + }), + ); + return (response.Contents?.length ?? 0) > 0; +} diff --git a/test/sync-releases.test.ts b/test/sync-releases.test.ts index c0f1a92..51b3fbd 100644 --- a/test/sync-releases.test.ts +++ b/test/sync-releases.test.ts @@ -4,12 +4,20 @@ import { ListObjectsV2Command, S3Client, } from "@aws-sdk/client-s3"; -import { describe, expect, beforeEach, it } from "vitest"; +import { PrismaClient } from "@prisma/client"; +import { afterEach, describe, expect, beforeEach, it, vi } from "vitest"; -import { collectReleaseArtifacts, syncReleases } from "../scripts/sync-releases"; +import { + collectReleaseArtifacts, + createAtDefaultRollout, + scheduleReleaseSync, + syncReleases, + UPLOAD_SETTLE_MS, + type ReleaseDecider, + type ReleaseType, +} from "../src/release-sync"; import { otaFileForPrefix } from "../src/skus"; -type ReleaseType = string; import { createAsyncIterable, s3Mock, testPrisma } from "./setup"; const DEFAULT_SKU = "jetkvm-v2"; @@ -20,9 +28,27 @@ const SYNC_BUCKET = "test-bucket"; const SYNC_BASE_URL = "https://cdn.test.com"; const syncS3Client = new S3Client({}); -function mockS3ListVersions(prefix: ReleaseType, versions: string[]) { - s3Mock.on(ListObjectsV2Command, { Prefix: `${prefix}/` }).resolves({ - CommonPrefixes: versions.map(v => ({ Prefix: `${prefix}/${v}/` })), +/** Makes the settle check see one object under the version uploaded at `at`. */ +function mockS3UploadedAt(prefix: ReleaseType, version: string, at: Date) { + s3Mock.on(ListObjectsV2Command, { Prefix: `${prefix}/${version}/` }).resolves({ + Contents: [{ Key: `${prefix}/${version}/${otaFileForPrefix(prefix)}`, LastModified: at }], + }); +} + +/** Lists `versions` under `prefix`, one page per inner array, as R2 does past 1,000 keys. */ +function mockS3ListVersions(prefix: ReleaseType, ...pages: string[][]) { + pages.forEach((versions, index) => { + const last = index === pages.length - 1; + s3Mock + .on(ListObjectsV2Command, { + Prefix: `${prefix}/`, + ContinuationToken: index === 0 ? undefined : `page-${index}`, + }) + .resolves({ + CommonPrefixes: versions.map(v => ({ Prefix: `${prefix}/${v}/` })), + IsTruncated: !last, + NextContinuationToken: last ? undefined : `page-${index + 1}`, + }); }); } @@ -56,14 +82,17 @@ function mockS3SkuVersion( }); } -describe("sync-releases script", () => { - beforeEach(() => { - s3Mock.reset(); - s3Mock - .on(HeadObjectCommand) - .rejects({ name: "NotFound", $metadata: { httpStatusCode: 404 } }); - }); +beforeEach(() => { + s3Mock.reset(); + s3Mock + .on(HeadObjectCommand) + .rejects({ name: "NotFound", $metadata: { httpStatusCode: 404 } }); + // Listings a test does not stub (other prefixes, the upload settle check) see + // an empty folder. More specific .on(..., { Prefix }) stubs registered later win. + s3Mock.on(ListObjectsV2Command).resolves({ Contents: [] }); +}); +describe("syncReleases", () => { it("marks legacy app artifacts compatible with the default SKU only", async () => { mockS3HashFile("app", "9.9.1", "legacy-app-hash"); @@ -178,16 +207,25 @@ describe("sync-releases script", () => { mockS3ListVersions("app", [version, "10.0.0-beta.1"]); mockS3ListVersions("system", [version]); - mockS3ListVersions("mini", []); mockS3HashFile("app", version, "app-hash"); mockS3SkuVersion("system", version, DEFAULT_SKU, "system-hash-v2"); mockS3SkuVersion("system", version, SDMMC_SKU, "system-hash-sdmmc"); - await syncReleases( + const stats = await syncReleases( { prisma: testPrisma, s3Client: syncS3Client }, { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + createAtDefaultRollout, ); + expect(stats).toEqual({ + created: 1, + "already-synced": 1, + uploading: 0, + "no-artifacts": 0, + skipped: 0, + aborted: 0, + }); + const appRelease = await testPrisma.release.findUniqueOrThrow({ where: { version_type: { version, type: "app" } }, include: { artifacts: true }, @@ -220,4 +258,225 @@ describe("sync-releases script", () => { // Prereleases are filtered out by listStableVersions. expect(prerelease).toBeNull(); }); + + it("registers versions from every page of a truncated listing", async () => { + const firstPage = "9.9.20"; + const secondPage = "9.9.21"; + mockS3ListVersions("app", [firstPage], [secondPage]); + mockS3HashFile("app", firstPage, "first-page-hash"); + mockS3HashFile("app", secondPage, "second-page-hash"); + + const stats = await syncReleases( + { prisma: testPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + createAtDefaultRollout, + ); + + expect(stats).toMatchObject({ created: 2 }); + const created = await testPrisma.release.findMany({ + where: { type: "app", version: { in: [firstPage, secondPage] } }, + orderBy: { version: "asc" }, + }); + expect(created.map(release => release.version)).toEqual([firstPage, secondPage]); + }); + + it("defers a version whose objects changed within the settle window", async () => { + const fresh = "9.9.10"; + const settled = "9.9.11"; + mockS3ListVersions("app", [fresh, settled]); + mockS3HashFile("app", fresh, "fresh-hash"); + mockS3HashFile("app", settled, "settled-hash"); + mockS3UploadedAt("app", fresh, new Date(Date.now() - 60 * 1000)); + mockS3UploadedAt("app", settled, new Date(Date.now() - UPLOAD_SETTLE_MS - 60 * 1000)); + + const stats = await syncReleases( + { prisma: testPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL, uploadSettleMs: UPLOAD_SETTLE_MS }, + createAtDefaultRollout, + ); + + expect(stats).toMatchObject({ created: 1, uploading: 1 }); + expect( + await testPrisma.release.findUnique({ + where: { version_type: { version: fresh, type: "app" } }, + }), + ).toBeNull(); + // The settle listing must come after the artifact scan, so an upload that + // overlaps the scan is seen. The deferred version was therefore scanned. + const calls = s3Mock.calls().map(call => call.args[0].input as { Key?: string; Prefix?: string }); + const scanIndex = calls.findIndex(input => input.Key === `app/${fresh}/jetkvm_app.sha256`); + const settleIndex = calls.findIndex(input => input.Prefix === `app/${fresh}/`); + expect(scanIndex).toBeGreaterThanOrEqual(0); + expect(settleIndex).toBeGreaterThan(scanIndex); + expect( + await testPrisma.release.findUnique({ + where: { version_type: { version: settled, type: "app" } }, + }), + ).toMatchObject({ hash: "settled-hash" }); + }); + + it("honours the decider's rollout, skip and abort answers", async () => { + mockS3ListVersions("app", ["9.9.5", "9.9.6", "9.9.7"]); + mockS3ListVersions("system", ["9.9.5"]); + for (const version of ["9.9.5", "9.9.6", "9.9.7"]) { + mockS3HashFile("app", version, `app-hash-${version}`); + } + mockS3HashFile("system", "9.9.5", "system-hash"); + + const answers: Record>> = { + "app 9.9.5": { kind: "create", rolloutPercentage: 42 }, + "app 9.9.6": { kind: "skip" }, + "app 9.9.7": { kind: "abort" }, + }; + const decide: ReleaseDecider = async (type, version) => answers[`${type} ${version}`]; + + const stats = await syncReleases( + { prisma: testPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + decide, + ); + + expect(stats).toMatchObject({ created: 1, skipped: 1, aborted: 1 }); + + const created = await testPrisma.release.findUniqueOrThrow({ + where: { version_type: { version: "9.9.5", type: "app" } }, + }); + expect(created.rolloutPercentage).toBe(42); + + const notCreated = await testPrisma.release.findMany({ + where: { + OR: [ + { version: "9.9.6", type: "app" }, + { version: "9.9.7", type: "app" }, + // Abort stops the whole run, so system is never reached. + { version: "9.9.5", type: "system" }, + ], + }, + }); + expect(notCreated).toEqual([]); + }); + + it("treats a release created by another instance mid-run as already synced", async () => { + const version = "9.9.8"; + mockS3ListVersions("app", [version]); + mockS3HashFile("app", version, "app-hash"); + + // Simulate the race: the known-versions query sees nothing, but by the + // time this instance inserts, the row is there (written here up front so + // the real unique constraint fires on create). + await testPrisma.release.create({ + data: { + version, + type: "app", + rolloutPercentage: 10, + url: "https://cdn.test.com/other-instance", + hash: "other-instance-hash", + }, + }); + const racingPrisma = { + release: { + findMany: async () => [], + create: (args: unknown) => testPrisma.release.create(args as any), + }, + } as unknown as PrismaClient; + + const stats = await syncReleases( + { prisma: racingPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + createAtDefaultRollout, + ); + + expect(stats).toMatchObject({ created: 0, "already-synced": 1 }); + const release = await testPrisma.release.findUniqueOrThrow({ + where: { version_type: { version, type: "app" } }, + }); + expect(release.url).toBe("https://cdn.test.com/other-instance"); + }); +}); + +describe("scheduleReleaseSync", () => { + const INTERVAL_MS = 1000; + let stop: (() => void) | undefined; + + beforeEach(() => { + // Only the scheduler's own timer is faked; DB and S3 mock I/O stay real. + vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + }); + + afterEach(() => { + stop?.(); + stop = undefined; + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("runs at start, keeps the schedule after a failed run and creates on the next tick", async () => { + const version = "9.9.9"; + s3Mock + .on(ListObjectsV2Command, { Prefix: "app/" }) + .rejectsOnce(new Error("R2 unavailable")) + .resolves({ CommonPrefixes: [{ Prefix: `app/${version}/` }] }); + mockS3HashFile("app", version, "app-hash"); + const errorLog = vi.spyOn(console, "error").mockImplementation(() => {}); + + stop = scheduleReleaseSync( + { prisma: testPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + INTERVAL_MS, + ); + + await vi.waitFor(() => expect(errorLog).toHaveBeenCalledOnce()); + expect( + await testPrisma.release.findUnique({ + where: { version_type: { version, type: "app" } }, + }), + ).toBeNull(); + + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + + await vi.waitFor(async () => { + const release = await testPrisma.release.findUnique({ + where: { version_type: { version, type: "app" } }, + }); + expect(release?.rolloutPercentage).toBe(10); + }); + }); + + it("skips a tick while the previous run is still in progress", async () => { + let finishFirstRun!: () => void; + const firstListing = new Promise<{ CommonPrefixes: never[] }>(resolve => { + finishFirstRun = () => resolve({ CommonPrefixes: [] }); + }); + s3Mock + .on(ListObjectsV2Command, { Prefix: "app/" }) + .callsFakeOnce(() => firstListing) + .resolves({ CommonPrefixes: [] }); + const warnLog = vi.spyOn(console, "warn").mockImplementation(() => {}); + + stop = scheduleReleaseSync( + { prisma: testPrisma, s3Client: syncS3Client }, + { bucketName: SYNC_BUCKET, baseUrl: SYNC_BASE_URL }, + INTERVAL_MS, + ); + + // Two ticks fire while the first run is still waiting on R2. + await vi.advanceTimersByTimeAsync(INTERVAL_MS * 2); + + expect(warnLog).toHaveBeenCalledTimes(2); + // The stalled run's list call is the only R2 traffic so far: the skipped + // ticks did not start a second walk of the bucket. + expect(s3Mock.commandCalls(ListObjectsV2Command)).toHaveLength(1); + + finishFirstRun(); + await vi.waitFor(() => + expect(s3Mock.commandCalls(ListObjectsV2Command)).toHaveLength(3), + ); + + // The next tick after the run completed starts a fresh run. + await vi.advanceTimersByTimeAsync(INTERVAL_MS); + await vi.waitFor(() => + expect(s3Mock.commandCalls(ListObjectsV2Command)).toHaveLength(6), + ); + expect(warnLog).toHaveBeenCalledTimes(2); + }); });