feat: add dynamo block index migrator

packages/cli/bin.js

@@ -64,7 +64,7 @@ cli
   .option('-r, --region', 'Bucket region.')
   .option('-b, --bucket', 'Bucket name.')
   .action(async (/** @type {string|undefined} */ key, /** @type {Record<string, string|undefined>} */ options) => {
-    const endpoint = new URL(options.endpoint ?? notNully(process.env, 'SERVICE_ENDPOINT', 'missing required option'))
+    const endpoint = new URL(options.endpoint ?? notNully(process.env, 'HASH_SERVICE_ENDPOINT', 'missing required option'))
     if (key) {
       const region = notNully(options, 'region', 'missing required option')
       const bucket = notNully(options, 'bucket', 'missing required option')
@@ -281,4 +281,65 @@ cli
       .pipeTo(Writable.toWeb(process.stdout))
   })
 
+cli.command('index [key] [cid]')
+  .option('-e, --endpoint', 'Service endpoint.')
+  .option('-r, --region', 'Bucket region.')
+  .option('-b, --bucket', 'Bucket name.')
+  .action(async (/** @type {string|undefined} */ key, /** @type {string|undefined} */ cidstr, /** @type {Record<string, string|undefined>} */ options) => {
+    const endpoint = new URL(options.endpoint ?? notNully(process.env, 'INDEX_SERVICE_ENDPOINT', 'missing required option'))
+    if (key && cidstr) {
+      const region = notNully(options, 'region', 'missing required option')
+      const bucket = notNully(options, 'bucket', 'missing required option')
+      /** @type {import('multiformats').Link} */
+      const cid = Link.parse(cidstr)
+      try {
+        await index(endpoint, region, bucket, key, cid)
+        return console.log(dagJSON.stringify({ region, bucket, key, cid }))
+      } catch (err) {
+        console.warn(`failed index of ${region}/${bucket}/${key}`, err)
+        return console.log(dagJSON.stringify({ region, bucket, key, cid, error: err.message }))
+      }
+    }
+
+    const source = /** @type {ReadableStream<Uint8Array>} */ (Readable.toWeb(process.stdin))
+    await source
+      .pipeThrough(/** @type {Parse<{ region?: string, bucket?: string, key: string, cid: { '/': string }, root?: { '/': string } }|{ error: string }>} */ (new Parse()))
+      .pipeThrough(new Parallel(concurrency, async item => {
+        if ('error' in item) return { ...item, error: 'missing shard CID' }
+        const region = item.region ?? notNully(options, 'region', 'missing required option')
+        const bucket = item.bucket ?? notNully(options, 'bucket', 'missing required option')
+        const { key } = item
+        /** @type {import('multiformats').Link} */
+        const cid = Link.parse(item.cid['/'])
+        try {
+          await retry(() => index(endpoint, region, bucket, key, cid))
+          return { region, bucket, key, cid }
+        } catch (err) {
+          console.warn(`failed index of ${region}/${bucket}/${key}`, err)
+          return { region, bucket, key, cid, error: err.message }
+        }
+      }))
+      .pipeThrough(new Stringify(dagJSON.stringify))
+      .pipeTo(Writable.toWeb(process.stdout))
+  })
+
+/**
+ * @param {URL} endpoint
+ * @param {string} region
+ * @param {string} bucket
+ * @param {string} key
+ * @param {import('multiformats').Link} shard
+ */
+const index = async (endpoint, region, bucket, key, shard) => {
+  const url = new URL(endpoint)
+  url.searchParams.set('region', region)
+  url.searchParams.set('bucket', bucket)
+  url.searchParams.set('key', key)
+  url.searchParams.set('shard', shard.toString())
+  const res = await fetch(url, { dispatcher })
+  const text = await res.text()
+  if (!res.ok) throw new Error(`index failed: ${text}`)
+  return dagJSON.parse(text)
+}
+
 cli.parse(process.argv)

packages/functions/package.json

@@ -9,16 +9,22 @@
   "devDependencies": {
     "@types/aws-lambda": "^8.10.119",
     "@types/node": "^20.4.7",
+    "@types/varint": "^6.0.1",
     "nanoid": "^4.0.2",
     "sst": "^2.23.1",
     "testcontainers": "^10.2.1",
     "vitest": "^0.34.1"
   },
   "dependencies": {
+    "@aws-sdk/client-dynamodb": "^3.410.0",
     "@aws-sdk/client-s3": "^3.383.0",
+    "@aws-sdk/util-dynamodb": "^3.410.0",
     "cardex": "^2.3.1",
     "carstream": "^1.1.0",
     "multiformats": "^12.0.1",
-    "uint8arraylist": "^2.4.3"
+    "p-retry": "^6.0.0",
+    "parallel-transform-web": "^1.0.0",
+    "uint8arraylist": "^2.4.3",
+    "varint": "^6.0.0"
   }
 }

packages/functions/src/copy.ts

@@ -12,35 +12,12 @@ import { Uint8ArrayList } from 'uint8arraylist'
 import { CARReaderStream } from 'carstream'
 import { MultihashIndexSortedWriter } from 'cardex/multihash-index-sorted'
 import { mustGetEnv, errorResponse } from './lib/util'
+import { ShardLink, ObjectID, ContentAddressedObjectID, ShardObjectID } from './lib/api.js'
+import { CAR_CODEC } from './lib/constants.js'
 
-const CAR_CODEC = 0x0202
 const MAX_PUT_SIZE = 1024 * 1024 * 1024 * 5
 const TARGET_PART_SIZE = 1024 * 1024 * 100
 
-type ShardLink = Link.Link<Uint8Array, typeof CAR_CODEC>
-
-interface ObjectID {
-  region: string
-  bucket: string
-  key: string
-  endpoint?: string
-  credentials?: {
-    accessKeyId: string,
-    secretAccessKey: string
-  }
-}
-
-interface ContentAddressedObjectID<
-  Data extends unknown = unknown,
-  Format extends number = number,
-  Alg extends number = number,
-  V extends Link.Version = 1
-> extends ObjectID {
-  cid: Link.Link<Data, Format, Alg, V>
-}
-
-interface ShardObjectID extends ContentAddressedObjectID<Uint8Array, typeof CAR_CODEC> {}
-
 interface ShardSource extends ContentAddressedObjectID<Uint8Array, typeof CAR_CODEC> {
   size: number
   body: ReadableStream<Uint8Array>

packages/functions/src/index.ts

@@ -0,0 +1,189 @@
+import { ApiHandler } from 'sst/node/api'
+import { GetObjectCommand, S3Client } from '@aws-sdk/client-s3'
+import { MultihashDigest } from 'multiformats'
+import * as Link from 'multiformats/link'
+import { base58btc } from 'multiformats/bases/base58'
+import { CARReaderStream } from 'carstream'
+import { mustGetEnv, errorResponse } from './lib/util'
+import { BatchGetItemCommand, BatchWriteItemCommand, DynamoDBClient, WriteRequest } from '@aws-sdk/client-dynamodb'
+import { Block, Position } from 'carstream/api'
+import { marshall, unmarshall } from '@aws-sdk/util-dynamodb'
+import retry from 'p-retry'
+import { Parallel } from 'parallel-transform-web'
+import { MultihashIndexSortedReader } from 'cardex/multihash-index-sorted'
+import { Credentials, Endpoint, ShardLink, ShardObjectID } from './lib/api.js'
+import { CAR_CODEC } from './lib/constants.js'
+
+interface TableID extends Endpoint, Credentials {
+  tableName: string
+}
+
+export const handler = ApiHandler(event => _handler.call(null, new Request(`http://localhost/?${event.rawQueryString}`), process.env))
+
+export const _handler = async (request: Request, env: Record<string, string|undefined>) => {
+  try {
+    const { searchParams } = new URL(request.url)
+
+    const srcRegion = searchParams.get('region')
+    if (!srcRegion) return errorResponse('Missing "region" search parameter', 400)
+    if (!['us-east-2', 'us-west-2'].includes(srcRegion)) return errorResponse('Invalid region', 400)
+
+    const srcBucketName = searchParams.get('bucket')
+    if (!srcBucketName) return errorResponse('Missing "bucket" search parameter', 400)
+    if (!srcBucketName.startsWith('dotstorage')) return errorResponse('Invalid bucket', 400)
+
+    const srcKey = searchParams.get('key')
+    if (!srcKey) return errorResponse('Missing "key" search parameter', 400)
+    if (!srcKey.endsWith('.car')) return errorResponse('Only keys for CARs supported', 400)
+
+    const shardstr = searchParams.get('shard')
+    if (!shardstr) return errorResponse('Missing "shard" search parameter', 400)
+    const shard: ShardLink = Link.parse(shardstr)
+    if (shard.code !== CAR_CODEC) return errorResponse('Not a CAR file hash', 400)
+
+    const src = {
+      cid: shard,
+      region: srcRegion,
+      bucket: srcBucketName,
+      key: srcKey
+    }
+
+    const dest = {
+      region: mustGetEnv(env, 'BLOCK_INDEX_REGION'),
+      tableName: mustGetEnv(env, 'BLOCK_INDEX_TABLE')
+    }
+
+    return await index(src, dest)
+  } catch (err: any) {
+    console.error(err)
+    return errorResponse(err.message, 500)
+  }
+}
+
+class Batcher<I> extends TransformStream<I, I[]> {
+  constructor (size: number) {
+    let batch: I[] = []
+    super({
+      transform (chunk, controller) {
+        batch.push(chunk)
+        if (batch.length < size) return
+        controller.enqueue(batch)
+        batch = []
+      },
+      flush (controller) {
+        if (batch.length) controller.enqueue(batch)
+        batch = []
+      }
+    })
+  }
+}
+
+interface BlockIndexItem {
+  blockmultihash: string
+  carpath: string
+  offset: number
+  length: number
+}
+
+export const index = async (src: ShardObjectID, dest: TableID) => {
+  const dynamo = new DynamoDBClient(dest)
+  const multihashes = await shardMultihashes(src)
+  let total = 0
+  await multihashes
+    .pipeThrough(new Batcher(100))
+    .pipeThrough(new TransformStream<MultihashDigest[], BlockIndexItem>({
+      async transform (batch, controller) {
+        const cmd = new BatchGetItemCommand({
+          RequestItems: {
+            [dest.tableName]: {
+              Keys: batch.map(multihash => marshall({
+                blockmultihash: base58btc.encode(multihash.bytes),
+                carpath: `${src.region}/${src.bucket}/${src.key}`
+              }))
+            }
+          }
+        })
+        const res = await dynamo.send(cmd)
+        for (const item of res.Responses?.[dest.tableName] ?? []) {
+          controller.enqueue(unmarshall(item) as BlockIndexItem)
+        }
+      }
+    }))
+    .pipeThrough(new Batcher(25))
+    .pipeThrough(new Parallel(5, async batch => {
+      const writeItems = (items: WriteRequest[]) => retry(async () => {
+        const cmd = new BatchWriteItemCommand({ RequestItems: { [dest.tableName]: items } })
+        const res = await dynamo.send(cmd)
+        if (res.UnprocessedItems && res.UnprocessedItems[dest.tableName]?.length) {
+          items = res.UnprocessedItems[dest.tableName]
+          throw new Error(`${res.UnprocessedItems[dest.tableName].length} unprocessed items`)
+        }
+      }, { retries: 2 })
+
+      let items: WriteRequest[] = batch.map(b => {
+        const item = { ...b, carpath: `auto/carpark-prod-0/${src.cid}/${src.cid}.car` }
+        console.warn('write', JSON.stringify(item))
+        return { PutRequest: { Item: marshall(item) } }
+      })
+
+      // write new items
+      await writeItems(items)
+
+      items = batch.map(b => {
+        const key = { blockmultihash: b.blockmultihash, carpath: b.carpath }
+        console.warn('delete', JSON.stringify(key))
+        return { DeleteRequest: { Key: marshall(key) } }
+      })
+
+      // delete old items
+      await writeItems(items)
+
+      return batch.length
+    }))
+    .pipeTo(new WritableStream({
+      async write (updated) {
+        total += updated
+      }
+    }))
+
+  return { statusCode: 200, body: JSON.stringify({ ok: true, updated: total }) }
+}
+
+/** Retrieve multihashes of blocks in the CAR. */
+const shardMultihashes = async (src: ShardObjectID): Promise<ReadableStream<MultihashDigest>> => {
+  const s3 = new S3Client(src)
+  try {
+    const cmd = new GetObjectCommand({ Bucket: src.bucket, Key: `${src.key}.idx` })
+    const res = await s3.send(cmd)
+    if (!res.Body) throw new Error('missing body')
+    const reader = MultihashIndexSortedReader.createReader({ reader: res.Body.transformToWebStream().getReader() })
+    return new ReadableStream({
+      async pull (controller) {
+        const { done, value } = await reader.read()
+        if (done) return controller.close()
+        controller.enqueue(value.multihash)
+      }
+    })
+  } catch (err: any) {
+    if (err.$metadata?.httpStatusCode !== 404) {
+      console.error(`failed to read index: ${src.key}.idx`, err)
+      throw new Error(`failed to read index: ${src.key}.idx`, { cause: err })
+    }
+  }
+
+  const blocks = await shardBlocks(src)
+  return blocks.pipeThrough(new TransformStream({
+    transform (block, controller) {
+      controller.enqueue(block.cid.multihash)
+    }
+  }))
+}
+
+/** Retrieve blocks from the CAR. */
+const shardBlocks = async (src: ShardObjectID): Promise<ReadableStream<Block & Position>> => {
+  const s3 = new S3Client(src)
+  const cmd = new GetObjectCommand({ Bucket: src.bucket, Key: src.key })
+  const res = await s3.send(cmd)
+  if (!res.Body) throw new Error('missing body')
+  return res.Body.transformToWebStream().pipeThrough(new CARReaderStream())
+}

packages/functions/src/lib/api.ts

@@ -0,0 +1,32 @@
+import { Link, Version } from 'multiformats'
+import { CAR_CODEC } from './constants.js'
+
+export type ShardLink = Link<Uint8Array, typeof CAR_CODEC>
+
+export interface Endpoint {
+  endpoint?: string
+}
+
+export interface Credentials {
+  credentials?: {
+    accessKeyId: string,
+    secretAccessKey: string
+  }
+}
+
+export interface ObjectID extends Endpoint, Credentials {
+  region: string
+  bucket: string
+  key: string
+}
+
+export interface ContentAddressedObjectID<
+  Data extends unknown = unknown,
+  Format extends number = number,
+  Alg extends number = number,
+  V extends Version = 1
+> extends ObjectID {
+  cid: Link<Data, Format, Alg, V>
+}
+
+export interface ShardObjectID extends ContentAddressedObjectID<Uint8Array, typeof CAR_CODEC> {}

packages/functions/src/lib/constants.ts

@@ -0,0 +1 @@
+export const CAR_CODEC = 0x0202