feat: integrate agent store for idempotence & invocation/receipt persistence

packages/access-client/src/agent-use-cases.js

@@ -38,17 +38,19 @@ export async function requestAccess(access, account, capabilities) {
  * @param {AccessAgent} access
  * @param {API.DID} [audienceOfClaimedDelegations] - audience of claimed delegations. defaults to access.connection.id.did()
  * @param {object} opts
+ * @param {string} [opts.nonce] - nonce to use for the claim
  * @param {boolean} [opts.addProofs] - whether to addProof to access agent
  * @returns
  */
 export async function claimAccess(
   access,
   audienceOfClaimedDelegations = access.connection.id.did(),
-  { addProofs = false } = {}
+  { addProofs = false, nonce } = {}
 ) {
   const res = await access.invokeAndExecute(Access.claim, {
     audience: access.connection.id,
     with: audienceOfClaimedDelegations,
+    nonce,
   })
   if (res.out.error) {
     throw res.out.error

packages/access-client/src/agent.js

@@ -581,6 +581,7 @@ export class Agent {
       }),
       issuer: this.issuer,
       proofs: [...proofs],
+      nonce: options.nonce,
     })
 
     return /** @type {API.IssuedInvocationView<API.InferInvokedCapability<CAP>>} */ (
@@ -592,13 +593,16 @@ export class Agent {
    * Get Space information from Access service
    *
    * @param {API.URI<"did:">} [space]
+   * @param {object} [options]
+   * @param {string} [options.nonce]
    */
-  async getSpaceInfo(space) {
+  async getSpaceInfo(space, options) {
     const _space = space || this.currentSpace()
     if (!_space) {
       throw new Error('No space selected, you need pass a resource.')
     }
     const inv = await this.invokeAndExecute(Capabilities.info, {
+      ...options,
       with: _space,
     })
 

packages/filecoin-api/src/storefront/api.ts

@@ -21,7 +21,7 @@ import {
   UpdatableAndQueryableStore,
   Queue,
   ServiceConfig,
-  StoreGetError
+  StoreGetError,
 } from '../types.js'
 
 export type PieceStore = UpdatableAndQueryableStore<

packages/upload-api/package.json

@@ -220,7 +220,6 @@
     "is-subset": "^0.1.1",
     "mocha": "^10.2.0",
     "one-webcrypto": "git://github.com/web3-storage/one-webcrypto",
-    "p-defer": "^4.0.1",
     "typescript": "5.2.2"
   },
   "eslintConfig": {

packages/upload-api/src/blob/accept.js

@@ -3,10 +3,15 @@ import * as DID from '@ipld/dag-ucan/did'
 import * as W3sBlob from '@web3-storage/capabilities/web3.storage/blob'
 import { Assert } from '@web3-storage/content-claims/capability'
 import { create as createLink } from 'multiformats/link'
+import { Invocation } from '@ucanto/core'
 import * as Digest from 'multiformats/hashes/digest'
 import { code as rawCode } from 'multiformats/codecs/raw'
 import * as API from '../types.js'
-import { AllocatedMemoryHadNotBeenWrittenTo } from './lib.js'
+import {
+  AllocatedMemoryHadNotBeenWrittenTo,
+  UnsupportedCapability,
+} from './lib.js'
+import * as HTTP from '@web3-storage/capabilities/http'
 
 /**
  * @param {API.W3ServiceContext} context
@@ -16,6 +21,13 @@ export function blobAcceptProvider(context) {
   return Server.provideAdvanced({
     capability: W3sBlob.accept,
     handler: async ({ capability }) => {
+      // Only service principal can perform an allocation
+      if (capability.with !== context.id.did()) {
+        return {
+          error: new UnsupportedCapability({ capability }),
+        }
+      }
+
       const { blob, space } = capability.nb
       // If blob is not stored, we must fail
       const hasBlob = await context.blobsStorage.has(blob.digest)
@@ -29,7 +41,9 @@ export function blobAcceptProvider(context) {
 
       const digest = Digest.decode(blob.digest)
       const content = createLink(rawCode, digest)
-      const createUrl = await context.blobsStorage.createDownloadUrl(digest.bytes)
+      const createUrl = await context.blobsStorage.createDownloadUrl(
+        digest.bytes
+      )
       if (createUrl.error) {
         return createUrl
       }
@@ -55,3 +69,76 @@ export function blobAcceptProvider(context) {
     },
   })
 }
+
+/**
+ * Polls `blob/accept` task whenever we receive a receipt. It may error if passed
+ * receipt is for `http/put` task that refers to the `blob/allocate` that we
+ * are unable to find.
+ *
+ * @param {API.ConcludeServiceContext} context
+ * @param {API.Receipt} receipt
+ * @returns {Promise<API.Result<{}, API.StorageGetError>>}
+ */
+export const poll = async (context, receipt) => {
+  const ran = Invocation.isInvocation(receipt.ran)
+    ? { ok: receipt.ran }
+    : await context.agentStore.invocations.get(receipt.ran)
+
+  // If can not find an invocation for this receipt there is nothing to do here,
+  // if it was receipt for `http/put` we would have invocation record.
+  if (!ran.ok) {
+    return { ok: {} }
+  }
+
+  // Detect if this receipt is for an `http/put` invocation
+  const put = /** @type {?API.HTTPPut} */ (
+    ran.ok.capabilities.find(({ can }) => can === HTTP.put.can)
+  )
+
+  // If it's not an http/put invocation nothing to do here.
+  if (put == null) {
+    return { ok: {} }
+  }
+
+  // Otherwise we are going to lookup allocation corresponding to this http/put
+  // in order to issue blob/accept.
+  const [, allocation] = /** @type {API.UCANAwait} */ (put.nb.url)['ucan/await']
+  const result = await context.agentStore.invocations.get(allocation)
+  // If could not find blob/allocate invocation there is something wrong in
+  // the system and we return error so it could be propagated to the user. It is
+  // not a proper solution as user can not really do anything, but still seems
+  // better than silently ignoring, this way user has a chance to report a
+  // problem. Client test could potentially also catch errors.
+  if (result.error) {
+    return result
+  }
+
+  const [allocate] = /** @type {[API.BlobAllocate]} */ (result.ok.capabilities)
+
+  // If this is a receipt for the http/put we will perform blob/accept.
+  const blobAccept = await W3sBlob.accept.invoke({
+    issuer: context.id,
+    audience: context.id,
+    with: context.id.did(),
+    nb: {
+      blob: put.nb.body,
+      space: allocate.nb.space,
+      _put: {
+        'ucan/await': ['.out.ok', receipt.ran.link()],
+      },
+    },
+    // ⚠️ We need invocation to be deterministic which is why we use exact
+    // same as it is on allocation which will guarantee that expiry is the
+    // same regardless when we received `http/put` receipt.
+    //
+    // ℹ️ This works around the fact that we index receipts by invocation link
+    // as opposed to task link which would not care about the expiration.
+    expiration: result.ok.expiration,
+  })
+
+  // We do not care about the result we just want receipt to be issued and
+  // stored.
+  await blobAccept.execute(context.getServiceConnection())
+
+  return { ok: {} }
+}