chore: address review

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

@@ -18,6 +18,7 @@ import {
 import {
   Store,
   UpdatableAndQueryableStore,
+  StreammableStore,
   Queue,
   ServiceConfig,
 } from '../types.js'
@@ -29,7 +30,7 @@ export type PieceStore = UpdatableAndQueryableStore<
 >
 export type FilecoinSubmitQueue = Queue<FilecoinSubmitMessage>
 export type PieceOfferQueue = Queue<PieceOfferMessage>
-export type DataStore = Store<UnknownLink, AsyncIterable<Uint8Array>>
+export type DataStore = StreammableStore<UnknownLink, Uint8Array>
 export type TaskStore = Store<UnknownLink, Invocation>
 export type ReceiptStore = Store<UnknownLink, Receipt>
 
@@ -123,8 +124,7 @@ export interface ClaimsClientContext {
    */
   claimsService: {
     invocationConfig: ClaimsInvocationConfig
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    connection: ConnectionView<any>
+    connection: ConnectionView<import('@web3-storage/content-claims/server/service/api').Service>
   }
 }
 

packages/filecoin-api/src/storefront/events.js

@@ -40,12 +40,12 @@ export const handleFilecoinSubmitMessage = async (context, message) => {
 
   // read and compute piece for content
   // TODO: needs to be hooked with location claims
-  const contentGetRes = await context.dataStore.get(message.content)
-  if (contentGetRes.error) {
-    return { error: new BlobNotFound(contentGetRes.error.message) }
+  const contentStreamRes = await context.dataStore.stream(message.content)
+  if (contentStreamRes.error) {
+    return { error: new BlobNotFound(contentStreamRes.error.message) }
   }
 
-  const computedPieceCid = await computePieceCid(contentGetRes.ok)
+  const computedPieceCid = await computePieceCid(contentStreamRes.ok)
   if (computedPieceCid.error) {
     return computedPieceCid
   }

packages/filecoin-api/src/types.ts

@@ -52,6 +52,17 @@ export interface UpdatableStore<RecKey, Rec> extends Store<RecKey, Rec> {
   ) => Promise<Result<Rec, StoreGetError>>
 }
 
+export interface StreammableStore<RecKey, Rec> {
+  /**
+   * Puts a record in the store.
+   */
+  put: (record: Rec) => Promise<Result<Unit, StorePutError>>
+  /**
+   * Gets a record from the store.
+   */
+  stream: (key: RecKey) => Promise<Result<AsyncIterable<Rec>, StoreGetError>>
+}
+
 export interface QueryableStore<RecKey, Rec, Query> extends Store<RecKey, Rec> {
   /**
    * Queries for record matching a given criterium.

packages/filecoin-api/test/context/store-implementations.js

@@ -1,4 +1,4 @@
-import { UpdatableStore } from './store.js'
+import { UpdatableStore, StreammableStore } from './store.js'
 
 /**
  * @typedef {import('@ucanto/interface').Link} Link
@@ -18,7 +18,8 @@ import { UpdatableStore } from './store.js'
  * @typedef {import('../../src/deal-tracker/api.js').DealRecordKey} DealRecordKey
  */
 export const getStoreImplementations = (
-  StoreImplementation = UpdatableStore
+  StoreImplementation = UpdatableStore,
+  StreammableStoreImplementation = StreammableStore
 ) => ({
   storefront: {
     pieceStore: new StoreImplementation({
@@ -76,12 +77,24 @@ export const getStoreImplementations = (
         return Array.from(items).find((i) => i.ran.link().equals(record))
       },
     }),
-    dataStore: new StoreImplementation({
-      getFn: (
-        /** @type {Set<AsyncIterable<Uint8Array>>} */ items,
+    dataStore: new StreammableStore({
+      streamFn: (
+        /** @type {Set<Uint8Array>} */ items,
         /** @type {import('@ucanto/interface').UnknownLink} */ record
       ) => {
-        return Array.from(items).pop()
+        const item = Array.from(items).pop()
+        if (!item) {
+          return undefined
+        }
+        const asyncIterableRes = {
+          [Symbol.asyncIterator]: async function* () {
+            // Yield the Uint8Array asynchronously
+            if (item) {
+              yield item
+            }
+          },
+        }
+        return asyncIterableRes
       },
     }),
   },

packages/filecoin-api/test/context/store.js

@@ -94,6 +94,53 @@ export class Store {
   }
 }
 
+/**
+ * @template K
+ * @template V
+ * @implements {API.StreammableStore<K,V>}
+ */
+export class StreammableStore {
+  /**
+   * @param {import('./types.js').StreammableStoreOptions<K, V>} options
+   */
+  constructor(options) {
+    /** @type {Set<V>} */
+    this.items = new Set()
+    this.streamFn = options.streamFn
+  }
+
+  /**
+   * @param {V} record
+   * @returns {Promise<import('@ucanto/interface').Result<{}, StorePutError>>}
+   */
+  async put(record) {
+    this.items.add(record)
+
+    return Promise.resolve({
+      ok: {},
+    })
+  }
+
+  /**
+   * @param {K} item
+   * @returns {Promise<import('@ucanto/interface').Result<AsyncIterable<V>, StoreGetError>>}
+   */
+  async stream(item) {
+    if (!this.streamFn) {
+      throw new Error('get not supported')
+    }
+    const t = this.streamFn(this.items, item)
+    if (!t) {
+      return {
+        error: new RecordNotFound('not found'),
+      }
+    }
+    return {
+      ok: t,
+    }
+  }
+}
+
 /**
  * @template K
  * @template V

packages/filecoin-api/test/context/types.ts

@@ -6,3 +6,7 @@ export interface StoreOptions<K, V> {
 export interface UpdatableStoreOptions<K, V> extends StoreOptions<K, V> {
   updateFn?: (items: Set<V>, key: K, item: Partial<V>) => V
 }
+
+export interface StreammableStoreOptions<K, V> extends StoreOptions<K, V> {
+  streamFn?: (items: Set<V>, item: K) => AsyncIterable<V> | undefined
+}

packages/filecoin-api/test/events/storefront.js

@@ -39,13 +39,7 @@ export const test = {
     }
 
     // Store bytes on datastore
-    const asyncIterableMock = {
-      [Symbol.asyncIterator]: async function* () {
-        // Yield the Uint8Array asynchronously
-        yield cargo.bytes
-      },
-    }
-    await context.dataStore.put(asyncIterableMock)
+    await context.dataStore.put(cargo.bytes)
 
     // Handle message
     const handledMessageRes =