feat: submit content claims

packages/upload-client/README.md

@@ -466,6 +466,10 @@ export interface CARMetadata {
    * Size of the CAR file in bytes.
    */
   size: number
+  /**
+   * The CAR file data that was stored.
+   */
+  blob(): Promise<Blob>
 }
 ```
 

packages/upload-client/src/sharding.js

@@ -97,7 +97,7 @@ export class ShardStoringStream extends TransformStream {
                 const opts = { ...options, signal: abortController.signal }
                 const cid = await add(conf, car, opts)
                 const { version, roots, size } = car
-                controller.enqueue({ version, roots, cid, size })
+                controller.enqueue(new ShardMetadata(version, roots, cid, size, car))
               } catch (err) {
                 controller.error(err)
                 abortController.abort(err)
@@ -117,3 +117,26 @@ export class ShardStoringStream extends TransformStream {
     })
   }
 }
+
+class ShardMetadata {
+  #blob
+
+  /**
+   * @param {number} version
+   * @param {import('multiformats').UnknownLink[]} roots 
+   * @param {import('./types').CARLink} cid
+   * @param {number} size
+   * @param {Blob} blob
+   */
+  constructor (version, roots, cid, size, blob) {
+    this.version = version
+    this.roots = roots
+    this.cid = cid
+    this.size = size
+    this.#blob = blob
+  }
+
+  async blob () {
+    return this.#blob
+  }
+}

packages/upload-client/src/types.ts

@@ -175,6 +175,10 @@ export interface CARMetadata extends CARHeaderInfo {
    * Size of the CAR file in bytes.
    */
   size: number
+  /**
+   * The CAR file data that was stored.
+   */
+  blob(): Promise<Blob>
 }
 
 export interface Retryable {
@@ -254,6 +258,10 @@ export interface UploadOptions
     ShardingOptions,
     ShardStoringOptions,
     UploadProgressTrackable {
+  /**
+   * A function called after a DAG shard has been successfully stored by the
+   * service.
+   */
   onShardStored?: (meta: CARMetadata) => void
 }
 

packages/w3up-client/package.json

@@ -81,7 +81,11 @@
     "@ucanto/transport": "^8.0.0",
     "@web3-storage/access": "workspace:^",
     "@web3-storage/capabilities": "workspace:^",
-    "@web3-storage/upload-client": "workspace:^"
+    "@web3-storage/content-claims": "^3.0.1",
+    "@web3-storage/upload-client": "workspace:^",
+    "cardex": "^2.3.1",
+    "carstream": "^1.1.0",
+    "p-queue": "^7.3.0"
   },
   "devDependencies": {
     "@docusaurus/core": "^2.2.0",

packages/w3up-client/src/base.js

@@ -17,10 +17,10 @@ export class Base {
   /**
    * @param {import('@web3-storage/access').AgentData} agentData
    * @param {object} [options]
-   * @param {import('./types').ServiceConf} [options.serviceConf]
+   * @param {Partial<import('./types').ServiceConf>} [options.serviceConf]
    */
   constructor(agentData, options = {}) {
-    this._serviceConf = options.serviceConf ?? serviceConf
+    this._serviceConf = { ...serviceConf, ...options.serviceConf }
     this._agent = new Agent(agentData, {
       servicePrincipal: this._serviceConf.access.id,
       // @ts-expect-error I know but it will be HTTP for the forseeable.

packages/w3up-client/src/capability/assert.js

@@ -0,0 +1,86 @@
+import { Assert } from '@web3-storage/content-claims/capability'
+import { Base } from '../base.js'
+
+/**
+ * Client for interacting with the content claim `assert/*` capabilities.
+ */
+export class AssertClient extends Base {
+  /**
+   * Claims that a CID is available at a URL.
+   *
+   * @param {import('multiformats').UnknownLink} content - Claim subject.
+   * @param {URL[]} location - Location(s) the content may be found.
+   */
+  async location(content, location) {
+    const conf = await this._invocationConfig([Assert.location.can])
+    const locs = location.map(l => /** @type {import('@ucanto/interface').URI} */ (l.toString()))
+    const result = await Assert.location
+      .invoke({ ...conf, nb: { content, location: locs } })
+      .execute(this._serviceConf.claim)
+    if (result.out.error) {
+      const cause = result.out.error
+      throw new Error(`failed ${Assert.location.can} invocation`, { cause })
+    }
+    return result.out.ok
+  }
+
+  /**
+   * Claims that a CID's graph can be read from the blocks found in parts.
+   *
+   * @param {import('multiformats').UnknownLink} content - Claim subject.
+   * @param {import('multiformats').Link|undefined} blocks - CIDs CID.
+   * @param {import('multiformats').Link[]} parts - CIDs of CAR files the content can be found within.
+   */
+  async partition(content, blocks, parts) {
+    const conf = await this._invocationConfig([Assert.partition.can])
+    const result = await Assert.partition
+      .invoke({ ...conf, nb: { content, blocks, parts } })
+      .execute(this._serviceConf.claim)
+    if (result.out.error) {
+      const cause = result.out.error
+      throw new Error(`failed ${Assert.partition.can} invocation`, { cause })
+    }
+    return result.out.ok
+  }
+
+  /**
+   * Claims that a CID includes the contents claimed in another CID.
+   * 
+   * @param {import('multiformats').UnknownLink} content - Claim subject.
+   * @param {import('multiformats').Link} includes - Contents the claim content includes.
+   * @param {import('multiformats').Link} [proof] - Inclusion proof.
+   */
+  async inclusion(content, includes, proof) {
+    const conf = await this._invocationConfig([Assert.inclusion.can])
+    const result = await Assert.inclusion
+      .invoke({ ...conf, nb: { content, includes, proof } })
+      .execute(this._serviceConf.claim)
+
+    if (result.out.error) {
+      const cause = result.out.error
+      throw new Error(`failed ${Assert.inclusion.can} invocation`, { cause })
+    }
+
+    return result.out.ok
+  }
+
+  /**
+   * Claim that a CID is linked to directly or indirectly by another CID.
+   *
+   * @param {import('multiformats').UnknownLink} content - Claim subject.
+   * @param {import('multiformats').UnknownLink} ancestor - Ancestor content CID.
+   */
+  async descendant(content, ancestor) {
+    const conf = await this._invocationConfig([Assert.descendant.can])
+    const result = await Assert.descendant
+      .invoke({ ...conf, nb: { content, ancestor } })
+      .execute(this._serviceConf.claim)
+
+    if (result.out.error) {
+      const cause = result.out.error
+      throw new Error(`failed ${Assert.descendant.can} invocation`, { cause })
+    }
+
+    return result.out.ok
+  }
+}

packages/w3up-client/src/client.js

@@ -14,6 +14,7 @@ import { StoreClient } from './capability/store.js'
 import { UploadClient } from './capability/upload.js'
 import { SpaceClient } from './capability/space.js'
 import { AccessClient } from './capability/access.js'
+import { ContentClaimsBuilder } from './content-claims.js'
 
 export class Client extends Base {
   /**
@@ -59,7 +60,11 @@ export class Client extends Base {
       UploadCapabilities.add.can,
     ])
     options.connection = this._serviceConf.upload
-    return uploadFile(conf, file, options)
+    const claimsBuilder = new ContentClaimsBuilder(this.capability.store)
+    const root = await uploadFile(conf, file, options)
+    const claims = await claimsBuilder.setRoot(root).close()
+    // TODO: submit claims
+    return root
   }
 
   /**

packages/w3up-client/src/content-claims.js

@@ -0,0 +1,71 @@
+import { MultihashIndexSortedWriter } from 'cardex/multihash-index-sorted'
+import { CARReaderStream } from 'carstream'
+import Queue from 'p-queue'
+import { CAR } from '@web3-storage/upload-client'
+import * as Link from 'multiformats/link'
+import { sha256 } from 'multiformats/hashes/sha2'
+
+export class ContentClaimsBuilder {
+  #store
+  #queue
+  /** @type {import('multiformats').UnknownLink|undefined} */
+  #root
+  /** @type {import('./types').CARLink[]} */
+  #shards
+
+  /**
+   * @param {import('./capability/store').StoreClient} store 
+   */
+  constructor (store) {
+    this.#store = store
+    this.#queue = new Queue()
+    this.#shards = []
+  }
+
+  /**
+   * @param {import('./types').CARLink} shard 
+   * @param {Blob} data
+   */
+  async addShard (shard, data) {
+    this.#shards.push(shard)
+    this.#queue.add(async () => {
+      const { readable, writable } = new TransformStream()
+      const writer = MultihashIndexSortedWriter.createWriter({ writer: writable.getWriter() })
+
+      const [, indexBlock] = await Promise.all([
+        data.stream()
+          .pipeThrough(new CARReaderStream())
+          .pipeTo(new WritableStream({
+            async write (block) { await writer.add(block.cid, block.offset) },
+            async close () { await writer.close() }
+          })),
+        (async () => {
+          const bytes = new Uint8Array(await new Response(readable).arrayBuffer())
+          const cid = Link.create(MultihashIndexSortedWriter.codec, await sha256.digest(bytes))
+          return { cid, bytes }
+        })()
+      ])
+
+      const car = await CAR.encode([indexBlock], indexBlock.cid)
+      const indexCARCID = await this.#store.add(car)
+
+      // TODO: create inclusion claim for shard => indexBlock.cid
+      // TODO: create partition claim for indexBlock.cid => indexCARCID
+      // TODO: create relation claims for block => shard => indexBlock.cid => indexCARCID
+    })
+    return this
+  }
+
+  /**
+   * @param {import('multiformats').UnknownLink} root
+   */
+  setRoot (root) {
+    this.#root = root
+    return this
+  }
+
+  async close () {
+    await this.#queue.onIdle()
+    // TODO: generate partition claim for root
+  }
+}

packages/w3up-client/src/service.js

@@ -1,6 +1,7 @@
 import { connect } from '@ucanto/client'
 import { CAR, HTTP } from '@ucanto/transport'
 import * as DID from '@ipld/dag-ucan/did'
+import { connection as claimServiceConnection } from '@web3-storage/content-claims/client'
 
 export const accessServiceURL = new URL('https://up.web3.storage')
 export const accessServicePrincipal = DID.parse('did:web:web3.storage')
@@ -30,4 +31,5 @@ export const uploadServiceConnection = connect({
 export const serviceConf = {
   access: accessServiceConnection,
   upload: uploadServiceConnection,
+  claim: claimServiceConnection,
 }

packages/w3up-client/src/types.ts

@@ -4,12 +4,14 @@ import {
   type AgentDataExport,
 } from '@web3-storage/access/types'
 import { type Service as UploadService } from '@web3-storage/upload-client/types'
+import type { Service as ClaimService } from '@web3-storage/content-claims/server/service/api'
 import type { ConnectionView, Signer, DID } from '@ucanto/interface'
 import { type Client } from './client'
 
 export interface ServiceConf {
   access: ConnectionView<AccessService>
   upload: ConnectionView<UploadService>
+  claim: ConnectionView<ClaimService>
 }
 
 export interface ClientFactoryOptions {