feat: agent store integration

filecoin/store/invocation.js

@@ -1,7 +1,6 @@
-import {
-  S3Client,
-  ListObjectsV2Command,
-} from '@aws-sdk/client-s3'
+import { S3Client } from '@aws-sdk/client-s3'
+import { parseLink } from '@ucanto/core'
+import * as Store from '../../upload-api/stores/agent/store.js'
 
 /**
  * Abstraction layer with Factory to perform operations on bucket storing
@@ -25,28 +24,26 @@ export function createInvocationStore(region, bucketName, options = {}) {
  * @returns {import('../types').InvocationBucket}
  */
 export const useInvocationStore = (s3client, bucketName) => {
+  const store = Store.open({
+    connection: { channel: s3client },
+    region: typeof s3client.config.region === 'string' ? s3client.config.region : 'us-west-2',
+    buckets: {
+      index: { name: bucketName },
+      message: { name: bucketName },
+    }
+  })
+
   return {
     /**
      * Get the agent message file CID for an invocation.
      *
      * @param {string} invocationCid 
      */
     getInLink: async (invocationCid) => {
-      const prefix = `${invocationCid}/`
-      // Multiple entries may match the key prefix. Picking an arbitrary one is fine given
-      //  can receive same invocations in multiple CAR files.
-      const listObjectCmd = new ListObjectsV2Command({
-        Bucket: bucketName,
-        Prefix: prefix,
-      })
-      const listObject = await s3client.send(listObjectCmd)
-      const carEntry = listObject.Contents?.find(
-        content => content.Key?.endsWith('.in')
-      )
-      if (!carEntry) {
-        return
+      const result = await Store.resolve(store, { invocation: parseLink(invocationCid) })
+      if (result.ok) {
+        return result.ok.message.toString()
       }
-      return carEntry.Key?.replace(prefix, '').replace('.in', '')
     },
   }
 }

filecoin/store/receipt.js

@@ -1,10 +1,6 @@
-import {
-  S3Client,
-  GetObjectCommand,
-  ListObjectsV2Command
-} from '@aws-sdk/client-s3'
-import * as CAR from '@ucanto/transport/car'
-import { StoreOperationFailed, RecordNotFound } from '@web3-storage/filecoin-api/errors'
+import { S3Client } from '@aws-sdk/client-s3'
+import { StoreOperationFailed } from '@web3-storage/filecoin-api/errors'
+import * as Store from '../../upload-api/stores/agent/store.js'
 
 /**
  * Abstraction layer with Factory to perform operations on bucket storing
@@ -30,97 +26,27 @@ export function createReceiptStore(region, invocationBucketName, workflowBucketN
  * @returns {import('@web3-storage/filecoin-api/storefront/api').ReceiptStore}
  */
 export const useReceiptStore = (s3client, invocationBucketName, workflowBucketName) => {
+  const store = Store.open({
+    connection: { channel: s3client },
+    region: typeof s3client.config.region === 'string' ? s3client.config.region : 'us-west-2',
+    buckets: {
+      index: { name: invocationBucketName },
+      message: { name: workflowBucketName },
+    }
+  })
+
   return {
     put: async (record) => {
       return {
         error: new StoreOperationFailed('no new receipt should be put by storefront')
       }
     },
-    get: async (taskCid) => {
-      // TODO: When we distinct between TaskCid and InvocationCid, we also need to see this mapping.
-      const invocationCid = taskCid
-
-      // Find agent message archive CID where this receipt was stored
-      const encodedInvocationKeyPrefix = `${invocationCid.toString()}/`
-      const listCmd = new ListObjectsV2Command({
-        Bucket: invocationBucketName,
-        Prefix: encodedInvocationKeyPrefix
-      })
-      let listRes
-      try {
-        listRes = await s3client.send(listCmd)
-      } catch (/** @type {any} */ error) {
-        if (error?.$metadata?.httpStatusCode === 404) {
-          return {
-            error: new RecordNotFound(`any pseudo symlink from invocation ${invocationCid.toString()} was found`)
-          }
-        }
-        return {
-          error: new StoreOperationFailed(error.message)
-        }
-      }
-      if (!listRes.Contents?.length) {
-        return {
-          error: new RecordNotFound(`any pseudo symlink from invocation ${invocationCid.toString()} was found`)
-        }
-      }
-      // Key in format `${invocation.cid}/${agentMessageArchive.cid}.out`
-      const agentMessageArchiveWithReceipt = listRes.Contents.find(c => c.Key?.endsWith('.out'))
-      if (!agentMessageArchiveWithReceipt || !agentMessageArchiveWithReceipt.Key) {
-        return {
-          error: new RecordNotFound(`any pseudo symlink from invocation ${invocationCid.toString()} was found with a receipt`)
-        }
-      }
-
-      // Get Message Archive with receipt
-      const agentMessageArchiveWithReceiptCid = agentMessageArchiveWithReceipt.Key
-        .replace(encodedInvocationKeyPrefix, '')
-        .replace('.out', '')
-
-      const encodedAgentMessageArchiveKey = `${agentMessageArchiveWithReceiptCid}/${agentMessageArchiveWithReceiptCid}`
-      const getCmd = new GetObjectCommand({
-        Bucket: workflowBucketName,
-        Key: encodedAgentMessageArchiveKey,
-      })
-
-      let res
-      try {
-        res = await s3client.send(getCmd)
-      } catch (/** @type {any} */ error) {
-        if (error?.$metadata?.httpStatusCode === 404) {
-          return {
-            error: new RecordNotFound(`agent message archive ${encodedAgentMessageArchiveKey} not found in store`)
-          }
-        }
-        return {
-          error: new StoreOperationFailed(error.message)
-        }
-      }
-      if (!res || !res.Body) {
-        return {
-          error: new RecordNotFound(`agent message archive ${encodedAgentMessageArchiveKey} not found in store`)
-        }
-      }
-
-
-      // Get receipt from Message Archive
-      const agentMessageBytes = await res.Body.transformToByteArray()
-      const agentMessage = await CAR.request.decode({
-        body: agentMessageBytes,
-        headers: {},
-      })
-
-      // @ts-expect-error unknown link does not mach expectations
-      const receipt = agentMessage.receipts.get(invocationCid.toString())
-      if (!receipt) {
-        return {
-          error: new RecordNotFound(`agent message archive ${encodedAgentMessageArchiveKey} does not include receipt for invocation ${invocationCid.toString()}`)
-        }
-      }
-      return {
-        ok: receipt
-      }
-    },
+    /**
+     * @param {import('@ucanto/interface').UnknownLink} taskCid
+     */
+    get: (taskCid) =>  
+      // @ts-expect-error - need to align RecordNotFoundError
+      Store.getReceipt(store, taskCid),
     has: async (record) => {
       return {
         error: new StoreOperationFailed('no receipt should checked by storefront')

filecoin/store/task.js

@@ -1,10 +1,6 @@
-import {
-  S3Client,
-  GetObjectCommand,
-  ListObjectsV2Command
-} from '@aws-sdk/client-s3'
-import * as CAR from '@ucanto/transport/car'
-import { StoreOperationFailed, RecordNotFound } from '@web3-storage/filecoin-api/errors'
+import { S3Client } from '@aws-sdk/client-s3'
+import { StoreOperationFailed } from '@web3-storage/filecoin-api/errors'
+import * as Store from '../../upload-api/stores/agent/store.js'
 
 /**
  * Abstraction layer with Factory to perform operations on bucket storing
@@ -30,98 +26,24 @@ export function createTaskStore(region, invocationBucketName, workflowBucketName
  * @returns {import('@web3-storage/filecoin-api/storefront/api').TaskStore}
  */
 export const useTaskStore = (s3client, invocationBucketName, workflowBucketName) => {
+  const store = Store.open({
+    connection: { channel: s3client },
+    region: typeof s3client.config.region === 'string' ? s3client.config.region : 'us-west-2',
+    buckets: {
+      index: { name: invocationBucketName },
+      message: { name: workflowBucketName },
+    }
+  })
+
   return {
     put: async (record) => {
       return {
         error: new StoreOperationFailed('no new task should be put by storefront')
       }
     },
-    get: async (taskCid) => {
-      // TODO: When we distinct between TaskCid and InvocationCid, we also need to see this mapping.
-      const invocationCid = taskCid
-
-      // Find agent message archive CID where this receipt was stored
-      const encodedInvocationKeyPrefix = `${invocationCid.toString()}/`
-      const listCmd = new ListObjectsV2Command({
-        Bucket: invocationBucketName,
-        Prefix: encodedInvocationKeyPrefix
-      })
-      let listRes
-      try {
-        listRes = await s3client.send(listCmd)
-      } catch (/** @type {any} */ error) {
-        if (error?.$metadata?.httpStatusCode === 404) {
-          return {
-            error: new RecordNotFound(`any pseudo symlink from invocation ${invocationCid.toString()} was found`)
-          }
-        }
-        return {
-          error: new StoreOperationFailed(error.message)
-        }
-      }
-      if (!listRes.Contents?.length) {
-        return {
-          error: new RecordNotFound(`any pseudo symlink from invocation ${invocationCid.toString()} was found`)
-        }
-      }
-
-      // Key in format `${invocation.cid}/${agentMessageArchive.cid}.in`
-      const agentMessageArchiveWithInvocation = listRes.Contents.find(c => c.Key?.endsWith('.in'))
-      if (!agentMessageArchiveWithInvocation || !agentMessageArchiveWithInvocation.Key) {
-        return {
-          error: new RecordNotFound(`any pseudo symlink from invocation ${invocationCid.toString()} was found with a receipt`)
-        }
-      }
-
-      // Get Message Archive with invocation
-      const agentMessageArchiveWithInvocationCid = agentMessageArchiveWithInvocation.Key
-        .replace(encodedInvocationKeyPrefix, '')
-        .replace('.in', '')
-
-      const encodedAgentMessageArchiveKey = `${agentMessageArchiveWithInvocationCid}/${agentMessageArchiveWithInvocationCid}`
-      const getCmd = new GetObjectCommand({
-        Bucket: workflowBucketName,
-        Key: encodedAgentMessageArchiveKey,
-      })
-
-      let res
-      try {
-        res = await s3client.send(getCmd)
-      } catch (/** @type {any} */ error) {
-        if (error?.$metadata?.httpStatusCode === 404) {
-          return {
-            error: new RecordNotFound(`agent message archive ${encodedAgentMessageArchiveKey} not found in store`)
-          }
-        }
-        return {
-          error: new StoreOperationFailed(error.message)
-        }
-      }
-      if (!res || !res.Body) {
-        return {
-          error: new RecordNotFound(`agent message archive ${encodedAgentMessageArchiveKey} not found in store`)
-        }
-      }
-
-      // Get invocation from Message Archive
-      const agentMessageBytes = await res.Body.transformToByteArray()
-      const agentMessage = await CAR.request.decode({
-        body: agentMessageBytes,
-        headers: {},
-      })
-
-      const invocation = agentMessage.invocations.find(
-        (invocation) => invocation.cid.toString() === invocationCid.toString()
-      )
-      if (!invocation) {
-        return {
-          error: new RecordNotFound(`agent message archive ${encodedAgentMessageArchiveKey} does not include invocation ${invocationCid.toString()}`)
-        }
-      }
-      return {
-        ok: invocation
-      }
-    },
+    get: (taskCid) =>
+        // @ts-expect-error - need to align RecordNotFoundError
+        Store.getInvocation(store, taskCid),
     has: async (record) => {
       return {
         error: new StoreOperationFailed('no task should checked by storefront')