feat: trigger content claims from piece cid

README.md

@@ -116,6 +116,14 @@ DID of the filecoin aggregator service.
 
 URL of the filecoin aggregator service.
 
+#### `CONTENT_CLAIMS_DID`
+
+DID of the [content claims service](https://github.com/web3-storage/content-claims).
+
+#### `CONTENT_CLAIMS_URL`
+
+URL of the [content claims service](https://github.com/web3-storage/content-claims).
+
 #### `UPLOAD_API_DID`
 
 [DID](https://www.w3.org/TR/did-core/) of the upload-api ucanto server. e.g. `did:web:up.web3.storage`. Optional: if omitted, a `did:key` will be derrived from `PRIVATE_KEY`
@@ -208,6 +216,14 @@ Generated by [@ucanto/principal `EdSigner`](https://github.com/web3-storage/ucan
 
 _Example:_ `MgCZG7EvaA...1pX9as=`
 
+#### `CONTENT_CLAIMS_PRIVATE_KEY`
+
+The `base64pad` [`multibase`](https://github.com/multiformats/multibase) encoded ED25519 keypair used as the signing key for [content-claims](https://github.com/web3-storage/content-claims).
+
+Generated by [@ucanto/principal `EdSigner`](https://github.com/web3-storage/ucanto) via [`ucan-key`](https://www.npmjs.com/package/ucan-key)
+
+_Example:_ `MgCZG7EvaA...1pX9as=`
+
 #### `UCAN_INVOCATION_POST_BASIC_AUTH`
 
 The HTTP Basic auth token for the UCAN Invocation entrypoint, where UCAN invocations can be stored and proxied to the UCAN Stream.

filecoin/functions/piece-cid-report.js

@@ -2,6 +2,10 @@ import * as Sentry from '@sentry/serverless'
 import { Config } from '@serverless-stack/node/config/index.js'
 import { unmarshall } from '@aws-sdk/util-dynamodb'
 import { Piece } from '@web3-storage/data-segment'
+import { CID } from 'multiformats/cid'
+import * as Delegation from '@ucanto/core/delegation'
+import { fromString } from 'uint8arrays/from-string'
+import * as DID from '@ipld/dag-ucan/did'
 
 import { reportPieceCid } from '../index.js'
 import { getServiceConnection, getServiceSigner } from '../service.js'
@@ -17,8 +21,8 @@ Sentry.AWSLambda.init({
  * @param {import('aws-lambda').DynamoDBStreamEvent} event
  */
 async function pieceCidReport (event) {
-  const { aggregatorDid, aggregatorUrl } = getEnv()
-  const { PRIVATE_KEY: privateKey } = Config
+  const { aggregatorDid, aggregatorUrl, contentClaimsDid, contentClaimsUrl, contentClaimsProof } = getEnv()
+  const { PRIVATE_KEY: privateKey, CONTENT_CLAIMS_PRIVATE_KEY: contentClaimsPrivateKey } = Config
 
   const records = parseDynamoDbEvent(event)
   if (records.length > 1) {
@@ -28,30 +32,53 @@ async function pieceCidReport (event) {
   // @ts-expect-error can't figure out type of new
   const pieceRecord = unmarshall(records[0].new)
   const piece = Piece.fromString(pieceRecord.piece).link
+  const content = CID.parse(pieceRecord.link)
 
   const aggregateServiceConnection = getServiceConnection({
     did: aggregatorDid,
     url: aggregatorUrl
   })
-  const issuer = getServiceSigner({
+  const claimsServiceConnection = getServiceConnection({
+    did: contentClaimsDid,
+    url: contentClaimsUrl
+  })
+  const storefrontIssuer = getServiceSigner({
     privateKey
   })
-  const audience = aggregateServiceConnection.id
-  /** @type {import('@web3-storage/filecoin-client/types').InvocationConfig} */
-  const invocationConfig = {
-    issuer,
-    audience,
-    with: issuer.did(),
+  let claimsIssuer = getServiceSigner({
+    privateKey: contentClaimsPrivateKey
+  })
+  const claimsProofs = []
+  if (contentClaimsProof) {
+    const proof = await Delegation.extract(fromString(contentClaimsProof, 'base64pad'))
+      if (!proof.ok) throw new Error('failed to extract proof', { cause: proof.error })
+      claimsProofs.push(proof.ok)
+  } else {
+    // if no proofs, we must be using the service private key to sign
+    claimsIssuer = claimsIssuer.withDID(DID.parse(contentClaimsDid).did())
   }
 
   const { ok, error } = await reportPieceCid({
     piece,
-    group: issuer.did(),
+    content,
+    group: storefrontIssuer.did(),
     aggregateServiceConnection,
-    invocationConfig
+    aggregateInvocationConfig: /** @type {import('@web3-storage/filecoin-client/types').InvocationConfig} */ ({
+      issuer: storefrontIssuer,
+      audience: aggregateServiceConnection.id,
+      with: storefrontIssuer.did(),
+    }),
+    claimsServiceConnection,
+    claimsInvocationConfig: /** @type {import('../types').ClaimsInvocationConfig} */ ({
+      issuer: claimsIssuer,
+      audience: claimsServiceConnection.id,
+      with: claimsIssuer.did(),
+    })
   })
 
   if (error) {
+    console.error(error)
+
     return {
       statusCode: 500,
       body: error.message || 'failed to add aggregate'
@@ -73,6 +100,9 @@ function getEnv() {
   return {
     aggregatorDid: mustGetEnv('AGGREGATOR_DID'),
     aggregatorUrl: mustGetEnv('AGGREGATOR_URL'),
+    contentClaimsDid: mustGetEnv('CONTENT_CLAIMS_DID'),
+    contentClaimsUrl: mustGetEnv('CONTENT_CLAIMS_URL'),
+    contentClaimsProof: process.env.CONTENT_CLAIMS_PROOF,
   }
 }
 

filecoin/index.js

@@ -4,6 +4,7 @@ import * as Hasher from 'fr32-sha2-256-trunc254-padded-binary-tree-multihash'
 import * as Digest from 'multiformats/hashes/digest'
 import { Piece } from '@web3-storage/data-segment'
 import { CID } from 'multiformats/cid'
+import { Assert } from '@web3-storage/content-claims/capability'
 import { Aggregator } from '@web3-storage/filecoin-client'
 
 import { GetCarFailed, ComputePieceFailed } from './errors.js'
@@ -86,19 +87,43 @@ export async function computePieceCid({
 /**
  * @param {object} props
  * @param {import('@web3-storage/data-segment').PieceLink} props.piece
+ * @param {import('multiformats').CID} props.content
  * @param {string} props.group
- * @param {import('@web3-storage/filecoin-client/types').InvocationConfig} props.invocationConfig
  * @param {import('@ucanto/principal/ed25519').ConnectionView<any>} props.aggregateServiceConnection
+ * @param {import('@web3-storage/filecoin-client/types').InvocationConfig} props.aggregateInvocationConfig
+ * @param {import('@ucanto/principal/ed25519').ConnectionView<any>} props.claimsServiceConnection
+ * @param {import('./types.js').ClaimsInvocationConfig} props.claimsInvocationConfig
  */
 export async function reportPieceCid ({
   piece,
+  content,
   group,
-  invocationConfig,
-  aggregateServiceConnection
+  aggregateServiceConnection,
+  aggregateInvocationConfig,
+  claimsServiceConnection,
+  claimsInvocationConfig
 }) {
+  // Add claim for reading
+  const claimResult = await Assert.equals
+    .invoke({
+      issuer: claimsInvocationConfig.issuer,
+      audience: claimsInvocationConfig.audience,
+      with: claimsInvocationConfig.with,
+      nb: {
+        content,
+        equals: piece
+      }
+    })
+    .execute(claimsServiceConnection)
+  if (claimResult.out.error) {
+    return {
+      error: claimResult.out.error
+    }
+  }
+
   // Add piece for aggregation
   const aggregateQueue = await Aggregator.aggregateQueue(
-    invocationConfig,
+    aggregateInvocationConfig,
     piece,
     group,
     { connection: aggregateServiceConnection }
@@ -109,6 +134,7 @@ export async function reportPieceCid ({
       error: aggregateQueue.out.error
     }
   }
+
   return {
     ok: {},
   }

filecoin/package.json

@@ -11,12 +11,16 @@
     "@aws-sdk/client-sqs": "^3.226.0",
     "@sentry/serverless": "^7.22.0",
     "@ucanto/client": "^8.0.1",
+    "@ucanto/core": "^8.2.0",
+    "@ucanto/interface": "^8.1.0",
     "@ucanto/principal": "^8.1.0",
     "@ucanto/transport": "^8.0.0",
+    "@web3-storage/content-claims": "^3.1.0",
     "@web3-storage/data-segment": "^3.0.1",
     "@web3-storage/filecoin-client": "^1.3.0",
     "fr32-sha2-256-trunc254-padded-binary-tree-multihash": "^1.0.0",
-    "multiformats": "^12.1.1"
+    "multiformats": "^12.1.1",
+    "uint8arrays": "4.0.6"
   },
   "devDependencies": {
     "@serverless-stack/resources": "*",

filecoin/test/helpers/mocks.js

@@ -6,7 +6,8 @@ const notImplemented = () => {
 
 /**
  * @param {Partial<
- * import('@web3-storage/filecoin-client/types').AggregatorService
+ * import('@web3-storage/filecoin-client/types').AggregatorService &
+ * { assert: Partial<import('@web3-storage/content-claims/server/service/api').AssertService> }
  * >} impl
  */
 export function mockService(impl) {
@@ -15,6 +16,9 @@ export function mockService(impl) {
       add: withCallCount(impl.aggregate?.add ?? notImplemented),
       queue: withCallCount(impl.aggregate?.queue ?? notImplemented),
     },
+    assert: {
+      equals: withCallCount(impl.assert?.equals ?? notImplemented)
+    }
  }
 }
 

filecoin/test/helpers/ucanto.js

@@ -3,12 +3,68 @@ import * as Server from '@ucanto/server'
 import * as Client from '@ucanto/client'
 import * as CAR from '@ucanto/transport/car'
 import * as FilecoinCapabilities from '@web3-storage/capabilities/filecoin'
+import { Assert } from '@web3-storage/content-claims/capability'
 
 import { OperationFailed } from './errors.js'
 import { mockService } from './mocks.js'
 
 const nop = (/** @type {any} */ invCap) => {}
 
+/**
+ * @param {any} serviceProvider
+ * @param {object} [options]
+ * @param {(inCap: any) => void} [options.onCall]
+ * @param {boolean} [options.mustFail]
+ */
+export async function getClaimsServiceServer (serviceProvider, options = {}) {
+  const onCall = options.onCall || nop
+  const equalsStore = new Map()
+
+  const service = mockService({
+    assert: {
+      equals: Server.provide(Assert.equals, async ({ capability, invocation }) => {
+        const invCap = invocation.capabilities[0]
+        const { content, equals } = capability.nb
+
+        if (options.mustFail) {
+          return {
+            error: new OperationFailed(
+              'failed to add to aggregate',
+              // @ts-ignore wrong dep
+              invCap.nb?.content
+            )
+          }
+        }
+
+        equalsStore.set(content.toString(), equals.toString())
+        equalsStore.set(equals.toString(), content.toString())
+
+        onCall(invCap)
+
+        return {
+          ok: {}
+        }
+      })
+    }
+  })
+
+  const server = Server.create({
+    id: serviceProvider,
+    service,
+    codec: CAR.inbound,
+  })
+  const connection = Client.connect({
+    id: serviceProvider,
+    codec: CAR.outbound,
+    channel: server,
+  })
+
+  return {
+    service,
+    connection
+  }
+}
+
 /**
  * @param {any} serviceProvider
  * @param {object} [options]
@@ -112,9 +168,10 @@ export async function getAggregatorServiceServer (serviceProvider, options = {})
   }
 }
 
-export async function getAggregatorServiceCtx () {
+export async function getServiceCtx () {
   const storefront = await Signer.generate()
   const aggregator = await Signer.generate()
+  const claims = await Signer.generate()
 
   return {
     storefront: {
@@ -126,6 +183,11 @@ export async function getAggregatorServiceCtx () {
       did: aggregator.did(),
       privateKey: Signer.format(aggregator),
       raw: aggregator
+    },
+    claims: {
+      did: claims.did(),
+      privateKey: Signer.format(claims),
+      raw: claims
     }
   }
 }