From 8a06a0f8216bd92d82c8d58917358fd3545cc52d Mon Sep 17 00:00:00 2001 From: DIYgod Date: Fri, 10 Apr 2026 19:30:29 +0800 Subject: [PATCH] feat(ota): mirror release payloads into R2 --- apps/ota/package.json | 1 + apps/ota/src/__tests__/sync.test.ts | 83 +++++++++++++++ apps/ota/src/lib/archive.ts | 151 ++++++++++++++++++++++++++++ apps/ota/src/lib/r2.ts | 11 ++ apps/ota/src/lib/sync.ts | 70 +++++++++++++ pnpm-lock.yaml | 8 ++ 6 files changed, 324 insertions(+) create mode 100644 apps/ota/src/lib/archive.ts create mode 100644 apps/ota/src/lib/r2.ts create mode 100644 apps/ota/src/lib/sync.ts diff --git a/apps/ota/package.json b/apps/ota/package.json index f28d0ebcb..f417230cc 100644 --- a/apps/ota/package.json +++ b/apps/ota/package.json @@ -10,6 +10,7 @@ "typecheck": "tsc --noEmit" }, "dependencies": { + "fzstd": "0.1.1", "hono": "4.12.1", "ofetch": "1.5.1", "zod": "3.25.76" diff --git a/apps/ota/src/__tests__/sync.test.ts b/apps/ota/src/__tests__/sync.test.ts index c10bba258..a623f5eb0 100644 --- a/apps/ota/src/__tests__/sync.test.ts +++ b/apps/ota/src/__tests__/sync.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" import type { GitHubRequestError } from "../lib/github" import { listPublishedOtaReleases } from "../lib/github" +import { mirrorReleaseToStorage } from "../lib/sync" describe("listPublishedOtaReleases", () => { beforeEach(() => { @@ -153,3 +154,85 @@ describe("listPublishedOtaReleases", () => { ) }) }) + +describe("mirrorReleaseToStorage", () => { + it("stores mirrored release metadata and latest pointers only for platforms with mirrored payloads", async () => { + const kvWrites: string[] = [] + const r2Writes: string[] = [] + + const kv = { + put: vi.fn(async (key: string) => { + kvWrites.push(key) + }), + get: vi.fn(async () => null), + } as unknown as KVNamespace + + const bucket = { + put: vi.fn(async (key: string) => { + r2Writes.push(key) + }), + } as unknown as R2Bucket + + await mirrorReleaseToStorage( + { + release: { + schemaVersion: 1, + product: "mobile", + channel: "production", + releaseVersion: "0.4.2", + releaseKind: "ota", + runtimeVersion: "0.4.1", + publishedAt: "2026-04-10T12:00:00Z", + git: { tag: "mobile/v0.4.2", commit: "abcdef1234567890" }, + policy: { + storeRequired: false, + minSupportedBinaryVersion: "0.4.1", + message: null, + }, + platforms: { + ios: { + launchAsset: { + path: "bundles/ios-main.js", + sha256: "a".repeat(64), + contentType: "application/javascript", + }, + assets: [], + }, + android: { + launchAsset: { + path: "bundles/android-main.js", + sha256: "b".repeat(64), + contentType: "application/javascript", + }, + assets: [], + }, + }, + }, + files: [ + { + key: "mobile/production/0.4.1/0.4.2/ios/bundles/ios-main.js", + body: new Uint8Array([1, 2, 3]), + contentType: "application/javascript", + }, + ], + }, + { kv, bucket }, + ) + + expect(r2Writes).toContain("mobile/production/0.4.1/0.4.2/ios/bundles/ios-main.js") + + const releaseWriteIndex = kvWrites.findIndex((key) => + key.includes("release:mobile:0.4.2"), + ) + const latestWriteIndex = kvWrites.findIndex((key) => + key.includes("latest:mobile:production:0.4.1:ios"), + ) + + expect(releaseWriteIndex).toBeGreaterThanOrEqual(0) + expect(latestWriteIndex).toBeGreaterThan(releaseWriteIndex) + expect(kvWrites.some((key) => key.includes("latest:mobile:production:0.4.1:ios"))).toBe(true) + expect(kvWrites.some((key) => key.includes("latest:mobile:production:0.4.1:android"))).toBe( + false, + ) + }) +}) diff --git a/apps/ota/src/lib/archive.ts b/apps/ota/src/lib/archive.ts new file mode 100644 index 000000000..903558de4 --- /dev/null +++ b/apps/ota/src/lib/archive.ts @@ -0,0 +1,151 @@ +import { decompress } from "fzstd" + +import type { OtaPlatform, OtaRelease } from "./schema" + +const TAR_BLOCK_SIZE = 512 +const textDecoder = new TextDecoder() +const OTA_PLATFORMS: OtaPlatform[] = ["ios", "android", "macos", "windows", "linux"] + +type PlatformPayload = NonNullable +type PlatformAsset = PlatformPayload["launchAsset"] | PlatformPayload["assets"][number] + +export interface MirroredFile { + key: string + body: Uint8Array + contentType: string +} + +export function buildMirroredAssetKey( + release: Pick, + platform: OtaPlatform, + assetPath: string, +) { + return [ + release.product, + release.channel, + release.runtimeVersion, + release.releaseVersion, + platform, + normalizeArchivePath(assetPath), + ].join("/") +} + +export async function extractMirroredFiles(input: { + release: OtaRelease + archiveBuffer: ArrayBuffer | Uint8Array +}): Promise { + const compressedArchive = + input.archiveBuffer instanceof Uint8Array ? input.archiveBuffer : new Uint8Array(input.archiveBuffer) + const archiveEntries = parseTarArchive(decompress(compressedArchive)) + const files: MirroredFile[] = [] + + for (const platform of OTA_PLATFORMS) { + const platformPayload = input.release.platforms[platform] + + if (!platformPayload) { + continue + } + + for (const asset of listReferencedAssets(platformPayload)) { + const assetPath = normalizeArchivePath(asset.path) + const archiveBody = archiveEntries.get(assetPath) + + if (!archiveBody) { + throw new Error( + `Archive is missing referenced file "${assetPath}" for ${platform} in ${input.release.releaseVersion}`, + ) + } + + files.push({ + key: buildMirroredAssetKey(input.release, platform, assetPath), + body: archiveBody, + contentType: asset.contentType, + }) + } + } + + return files +} + +function listReferencedAssets(platformPayload: PlatformPayload): PlatformAsset[] { + const dedupedAssets = new Map() + + for (const asset of [platformPayload.launchAsset, ...platformPayload.assets]) { + dedupedAssets.set(normalizeArchivePath(asset.path), asset) + } + + return [...dedupedAssets.values()] +} + +function parseTarArchive(archive: Uint8Array) { + const entries = new Map() + let offset = 0 + + while (offset + TAR_BLOCK_SIZE <= archive.length) { + const header = archive.subarray(offset, offset + TAR_BLOCK_SIZE) + + if (isZeroBlock(header)) { + break + } + + const fileName = readTarPath(header) + const fileSize = parseTarOctal(header.subarray(124, 136)) + const fileType = header[156] ?? 0 + const fileStart = offset + TAR_BLOCK_SIZE + const fileEnd = fileStart + fileSize + + if (fileEnd > archive.length) { + throw new Error(`Archive entry "${fileName}" exceeds archive bounds`) + } + + if (isRegularFile(fileType)) { + entries.set(normalizeArchivePath(fileName), archive.slice(fileStart, fileEnd)) + } + + offset = fileStart + Math.ceil(fileSize / TAR_BLOCK_SIZE) * TAR_BLOCK_SIZE + } + + return entries +} + +function readTarPath(header: Uint8Array) { + const name = readTarString(header.subarray(0, 100)) + const prefix = readTarString(header.subarray(345, 500)) + + return prefix ? `${prefix}/${name}` : name +} + +function readTarString(value: Uint8Array) { + const terminatorIndex = value.indexOf(0) + const sliceEnd = terminatorIndex === -1 ? value.length : terminatorIndex + + return textDecoder.decode(value.subarray(0, sliceEnd)).trim() +} + +function parseTarOctal(value: Uint8Array) { + const rawValue = readTarString(value).replaceAll("\0", "").trim() + + if (!rawValue) { + return 0 + } + + const parsedValue = Number.parseInt(rawValue, 8) + + if (Number.isNaN(parsedValue)) { + throw new TypeError(`Invalid tar entry size "${rawValue}"`) + } + + return parsedValue +} + +function normalizeArchivePath(path: string) { + return path.replace(/^\/+/, "").replace(/^(\.\/)+/, "").replaceAll(/\/{2,}/g, "/") +} + +function isRegularFile(fileType: number) { + return fileType === 0 || fileType === 48 +} + +function isZeroBlock(block: Uint8Array) { + return block.every((byte) => byte === 0) +} diff --git a/apps/ota/src/lib/r2.ts b/apps/ota/src/lib/r2.ts new file mode 100644 index 000000000..15f536a29 --- /dev/null +++ b/apps/ota/src/lib/r2.ts @@ -0,0 +1,11 @@ +import type { MirroredFile } from "./archive" + +export async function putMirroredFiles(bucket: R2Bucket, files: readonly MirroredFile[]) { + for (const file of files) { + await bucket.put(file.key, file.body, { + httpMetadata: { + contentType: file.contentType, + }, + }) + } +} diff --git a/apps/ota/src/lib/sync.ts b/apps/ota/src/lib/sync.ts new file mode 100644 index 000000000..1a0365eb7 --- /dev/null +++ b/apps/ota/src/lib/sync.ts @@ -0,0 +1,70 @@ +import type { MirroredFile } from "./archive" +import { buildMirroredAssetKey } from "./archive" +import { KV_KEYS } from "./constants" +import type { LatestReleasePointerRecord } from "./kv" +import { putReleaseRecord } from "./kv" +import { putMirroredFiles } from "./r2" +import type { OtaPlatform, OtaRelease } from "./schema" + +const OTA_PLATFORMS: OtaPlatform[] = ["ios", "android", "macos", "windows", "linux"] + +export async function mirrorReleaseToStorage( + input: { + release: OtaRelease + files: readonly MirroredFile[] + }, + env: { + kv: KVNamespace + bucket: R2Bucket + }, +) { + await putMirroredFiles(env.bucket, input.files) + await putReleaseRecord( + env.kv, + input.release.product, + input.release.releaseVersion, + input.release, + ) + + const mirroredFileKeys = new Set(input.files.map((file) => file.key)) + + for (const platform of OTA_PLATFORMS) { + if (!hasCompleteMirroredPayload(input.release, platform, mirroredFileKeys)) { + continue + } + + const latestReleasePointer: LatestReleasePointerRecord = { + releaseVersion: input.release.releaseVersion, + } + + await env.kv.put( + KV_KEYS.latest( + input.release.product, + input.release.channel, + input.release.runtimeVersion, + platform, + ), + JSON.stringify(latestReleasePointer), + ) + } +} + +function hasCompleteMirroredPayload( + release: OtaRelease, + platform: OtaPlatform, + mirroredFileKeys: ReadonlySet, +) { + const platformPayload = release.platforms[platform] + + if (!platformPayload) { + return false + } + + for (const asset of [platformPayload.launchAsset, ...platformPayload.assets]) { + if (!mirroredFileKeys.has(buildMirroredAssetKey(release, platform, asset.path))) { + return false + } + } + + return true +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b40351dc9..8555a8f08 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1444,6 +1444,9 @@ importers: apps/ota: dependencies: + fzstd: + specifier: 0.1.1 + version: 0.1.1 hono: specifier: 4.12.1 version: 4.12.1 @@ -11533,6 +11536,9 @@ packages: resolution: {integrity: sha512-trLf4SzuuUxfusZADLINj+dE8clK1frKdmqiJNb1Es75fmI5oY6X2mxLVUciLLjxqw/xr72Dhy+lER6dGd02FQ==} engines: {node: '>=10'} + fzstd@0.1.1: + resolution: {integrity: sha512-dkuVSOKKwh3eas5VkJy1AW1vFpet8TA/fGmVA5krThl8YcOVE/8ZIoEA1+U1vEn5ckxxhLirSdY837azmbaNHA==} + galactus@1.0.0: resolution: {integrity: sha512-R1fam6D4CyKQGNlvJne4dkNF+PvUUl7TAJInvTGa9fti9qAv95quQz29GXapA4d8Ec266mJJxFVh82M4GIIGDQ==} engines: {node: '>= 12'} @@ -29673,6 +29679,8 @@ snapshots: fuse.js@7.1.0: {} + fzstd@0.1.1: {} + galactus@1.0.0: dependencies: debug: 4.4.3(supports-color@8.1.1)