feat: trigger content claims equals on piece cid computed

filecoin/functions/piece-cid-report.js

@@ -2,6 +2,7 @@ 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 { reportPieceCid } from '../index.js'
 import { getServiceConnection, getServiceSigner } from '../service.js'
@@ -28,27 +29,36 @@ 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 claimsServiceConnection = getServiceConnection({
+    did: aggregatorDid,
+    url: aggregatorUrl
+  })
   const issuer = getServiceSigner({
     privateKey
   })
-  const audience = aggregateServiceConnection.id
-  /** @type {import('@web3-storage/filecoin-client/types').InvocationConfig} */
-  const invocationConfig = {
-    issuer,
-    audience,
-    with: issuer.did(),
-  }
 
   const { ok, error } = await reportPieceCid({
     piece,
+    content,
     group: issuer.did(),
     aggregateServiceConnection,
-    invocationConfig
+    aggregateInvocationConfig: /** @type {import('@web3-storage/filecoin-client/types').InvocationConfig} */ ({
+      issuer,
+      audience: aggregateServiceConnection.id,
+      with: issuer.did(),
+    }),
+    claimsServiceConnection,
+    claimsInvocationConfig: /** @type {import('../types').ClaimsInvocationConfig} */ ({
+      issuer,
+      audience: claimsServiceConnection.id,
+      with: issuer.did(),
+    })
   })
 
   if (error) {
@@ -73,6 +83,8 @@ function getEnv() {
   return {
     aggregatorDid: mustGetEnv('AGGREGATOR_DID'),
     aggregatorUrl: mustGetEnv('AGGREGATOR_URL'),
+    contentClaimsDid: mustGetEnv('CONTENT_CLAIMS_DID'),
+    contentClaimsUrl: mustGetEnv('CONTENT_CLAIMS_URL'),
   }
 }
 

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,44 @@ 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 +135,7 @@ export async function reportPieceCid ({
       error: aggregateQueue.out.error
     }
   }
+
   return {
     ok: {},
   }

filecoin/package.json

@@ -11,8 +11,10 @@
     "@aws-sdk/client-sqs": "^3.226.0",
     "@sentry/serverless": "^7.22.0",
     "@ucanto/client": "^8.0.1",
+    "@ucanto/interface": "^8.1.0",
     "@ucanto/principal": "^8.1.0",
     "@ucanto/transport": "^8.0.0",
+    "@web3-storage/content-claims": "^3.0.1",
     "@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",

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,13 @@ export async function getAggregatorServiceCtx () {
       did: aggregator.did(),
       privateKey: Signer.format(aggregator),
       raw: aggregator
+    },
+    claims: {
+      did: claims.did(),
+      privateKey: Signer.format(claims),
+      raw: claims
     }
   }
 }
+
+