feat(ota): mirror release payloads into R2

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"

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,
+    )
+  })
+})

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<OtaRelease["platforms"][OtaPlatform]>
+type PlatformAsset = PlatformPayload["launchAsset"] | PlatformPayload["assets"][number]
+
+export interface MirroredFile {
+  key: string
+  body: Uint8Array
+  contentType: string
+}
+
+export function buildMirroredAssetKey(
+  release: Pick<OtaRelease, "product" | "channel" | "runtimeVersion" | "releaseVersion">,
+  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<MirroredFile[]> {
+  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<string, PlatformAsset>()
+
+  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<string, Uint8Array>()
+  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)
+}

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,
+      },
+    })
+  }
+}

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<string>,
+) {
+  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
+}

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)