feat(locator): unify locator with seperate clients

seperates locator logic from the underlying client, which is abstracted to behave like the indexer
client, with a content claims adaptation

BREAKING CHANGE: content claims is now just a client, not a locator

package.json

@@ -43,13 +43,13 @@
       "import": "./src/fetcher/batching.js",
       "types": "./dist/src/fetcher/batching.d.ts"
     },
-    "./locator/content-claims": {
-      "import": "./src/locator/content-claims.js",
-      "types": "./dist/src/locator/content-claims.d.ts"
+    "./locator/content-claims-client": {
+      "import": "./src/locator/content-claims-client.js",
+      "types": "./dist/src/locator/content-claims-client.d.ts"
     },
-    "./locator/indexing-service": {
-      "import": "./src/locator/indexing-service/index.js",
-      "types": "./dist/src/locator/indexing-service/index.d.ts"
+    "./locator": {
+      "import": "./src/locator/index.js",
+      "types": "./dist/src/locator/index.d.ts"
     },
     "./tracing/tracing": {
       "import": "./src/tracing/tracing.js",
@@ -60,9 +60,9 @@
     "@cloudflare/workers-types": "^4.20241022.0",
     "@ipld/dag-ucan": "^3.4.0",
     "@opentelemetry/api": "^1.9.0",
-    "@storacha/indexing-service-client": "2.0.1-rc.5",
+    "@storacha/indexing-service-client": "2.0.1-rc.6",
     "@web3-storage/blob-index": "^1.0.2",
-    "@web3-storage/content-claims": "5.1.4-rc.5",
+    "@web3-storage/content-claims": "5.2.0",
     "multiformats": "^13.1.0",
     "multipart-byte-range": "^3.0.1",
     "p-defer": "^4.0.1",

src/locator/content-claims-client.js

@@ -0,0 +1,193 @@
+import { DigestMap, ShardedDAGIndex } from '@web3-storage/blob-index'
+import { Assert } from '@web3-storage/content-claims/capability'
+import * as Claims from '@web3-storage/content-claims/client'
+import { withSimpleSpan } from '../tracing/tracing.js'
+import * as ed25519 from '@ucanto/principal/ed25519'
+import { base58btc } from 'multiformats/bases/base58'
+import { NotFoundError } from '../lib.js'
+import { from } from '@storacha/indexing-service-client/query-result'
+
+/**
+ * @import * as API from '../api.js'
+ * @import {Claim, IndexingServiceClient, Query, QueryError, QueryOk, Result, Kind} from '@storacha/indexing-service-client/api'
+ * @import {KnownClaimTypes, LocationClaim} from '@web3-storage/content-claims/client/api'
+ */
+
+/**
+ * @typedef {{ serviceURL?: URL, carpark?: import('@cloudflare/workers-types').R2Bucket, carparkPublicBucketURL?: URL}} LocatorOptions
+ */
+
+/**
+ * ContentClaimsClient mimics the indexing service client using the content claims service
+ * @implements {IndexingServiceClient}
+ */
+export class ContentClaimsClient {
+  /**
+   * @type {DigestMap<API.MultihashDigest, boolean}
+   */
+  #indexCids
+
+  /**
+   * @type {URL|undefined}
+   */
+  #serviceURL
+
+  /**
+   * @type {import('@cloudflare/workers-types').R2Bucket|undefined}
+   */
+  #carpark
+
+  /**
+   * @type {URL | Undefined}
+   */
+  #carparkPublicBucketURL
+
+  /**
+   * @type {ed25519.EdSigner | undefined}
+   */
+  #signer
+
+  /**
+   * @param {LocatorOptions} [options]
+   */
+  constructor (options) {
+    this.#indexCids = new DigestMap()
+    this.#serviceURL = options?.serviceURL
+    this.#carpark = options?.carpark
+    this.#carparkPublicBucketURL = options?.carparkPublicBucketURL
+  }
+
+  async #getSigner () {
+    if (!this.#signer) {
+      this.#signer = await ed25519.Signer.generate()
+    }
+    return this.#signer
+  }
+
+  /**
+   *
+   * @param {Query} q
+   * @returns {Promise<Result<QueryOk, QueryError>>}
+   */
+  async queryClaims (q) {
+    /** @type {Claim[]} */
+    const claims = []
+    /** @type {Map<string, import('@web3-storage/blob-index/types').ShardedDAGIndexView} */
+    const indexes = new Map()
+    const kind = q.kind || 'standard'
+    for (const digest of q.hashes) {
+      const digestClaims = (await Claims.read(digest, { serviceURL: this.#serviceURL })).filter((claim) => allowedClaimTypes[kind].includes(claim.type))
+      let indexBytes
+      if (digestClaims.length === 0) {
+        const backups = await this.#carparkBackup(digest)
+        if (backups) {
+          claims.push(backups.claim)
+          indexBytes = backups.indexBytes
+        }
+      } else {
+        claims.push(...digestClaims)
+        for (const claim of digestClaims) {
+          if (claim.type === 'assert/index') {
+            this.#indexCids.set(claim.index.multihash, true)
+          }
+          if (claim.type === 'assert/location' && this.#indexCids.has(Claims.contentMultihash(claim))) {
+            try {
+              const fetchRes = await fetchIndex(claim)
+              indexBytes = await fetchRes.bytes()
+            } catch (err) {
+              console.warn('unable to fetch index', err instanceof Error ? err.message : 'unknown error')
+            }
+          }
+        }
+      }
+      if (indexBytes) {
+        const decodeRes = ShardedDAGIndex.extract(indexBytes)
+        if (decodeRes.error) {
+          console.warn('failed to decode index', decodeRes.error)
+          continue
+        }
+        indexes.set(base58btc.encode(digest.bytes), decodeRes.ok)
+      }
+    }
+    return /** @type {Result<QueryOk, QueryError>} */(await from({ claims, indexes }))
+  }
+
+  /**
+   *
+   * @param {*} digest
+   * @returns
+   */
+  async #carparkBackup (digest) {
+    let indexBytes
+    if (this.#carpark === undefined || this.#carparkPublicBucketURL === undefined) {
+      return
+    }
+    if (this.#indexCids.has(digest)) {
+      const obj = await withSimpleSpan('carPark.get', this.#carpark.get, this.#carpark)(toBlobKey(digest))
+      if (!obj) {
+        return
+      }
+      indexBytes = new Uint8Array(await obj.arrayBuffer())
+    } else {
+      const obj = await this.#carpark.head(toBlobKey(digest))
+      if (!obj) {
+        return
+      }
+    }
+    return {
+      claim: await Claims.decodeDelegation(await Assert.location
+        .invoke({
+          issuer: await this.#getSigner(),
+          audience: await this.#getSigner(),
+          with: (await this.#getSigner()).did(),
+          nb: {
+            content: { digest: digest.bytes },
+            location: [
+            /** @type {API.URI<import('@ucanto/principal/ed25519').Protocol>} */((new URL(toBlobKey(digest), this.#carparkPublicBucketURL)).href)
+            ]
+          }
+        })
+        .delegate()),
+      indexBytes
+    }
+  }
+}
+
+/** @type {Record<Kind, Array<KnownClaimTypes | "unknown">>} */
+const allowedClaimTypes = {
+  standard: ['assert/location', 'assert/partition', 'assert/inclusion', 'assert/index', 'assert/equals', 'assert/relation'],
+  index_or_location: ['assert/location', 'assert/index'],
+  location: ['assert/location']
+}
+
+/** @param {import('multiformats').MultihashDigest} digest */
+const toBlobKey = digest => {
+  const mhStr = base58btc.encode(digest.bytes)
+  return `${mhStr}/${mhStr}.blob`
+}
+
+const fetchIndex = withSimpleSpan('fetchIndex',
+  /**
+ * Fetch a blob from the passed location.
+ * @param {LocationClaim} locationClaim
+ */
+  async (locationClaim) => {
+    for (const uri of locationClaim.location) {
+      const url = new URL(uri)
+      /** @type {HeadersInit} */
+      const headers = {}
+      if (locationClaim.range) {
+        headers.Range = `bytes=${locationClaim.range.offset}-${
+            locationClaim.range.length ? locationClaim.range.offset + locationClaim.range.length - 1 : ''}`
+      }
+      const res = await fetch(url, { headers })
+      if (!res.ok || !res.body) {
+        console.warn(`failed to fetch ${url}: ${res.status} ${await res.text()}`)
+        continue
+      }
+      return res
+    }
+
+    throw new NotFoundError(Claims.contentMultihash(locationClaim))
+  }
+)