feat(filecoin-api): paginated queries

packages/filecoin-api/package.json

@@ -149,7 +149,8 @@
     "check": "tsc --build",
     "lint": "tsc --build && eslint '**/*.{js,ts}'",
     "test": "mocha --bail --timeout 10s -n no-warnings -n experimental-vm-modules -n experimental-fetch test/**/*.spec.js",
-    "test-watch": "pnpm build && mocha --bail --timeout 10s --watch --parallel -n no-warnings -n experimental-vm-modules -n experimental-fetch --watch-files src,test"
+    "test-watch": "pnpm build && mocha --bail --timeout 10s --watch --parallel -n no-warnings -n experimental-vm-modules -n experimental-fetch --watch-files src,test",
+    "coverage": "c8 -r text -r html npm run test"
   },
   "dependencies": {
     "@ipld/dag-ucan": "^3.4.0",
@@ -172,6 +173,7 @@
     "@web-std/blob": "^3.0.5",
     "@web3-storage/eslint-config-w3up": "workspace:^",
     "@web3-storage/filecoin-client": "workspace:^",
+    "c8": "^10.1.2",
     "mocha": "^10.2.0",
     "multiformats": "^12.1.2",
     "one-webcrypto": "git://github.com/web3-storage/one-webcrypto",

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

@@ -13,17 +13,15 @@ import {
   ServiceConfig,
 } from '../types.js'
 
-export type PieceStore = UpdatableStore<PieceRecordKey, PieceRecord>
+export type PieceStore = Store<PieceRecordKey, PieceRecord> &
+  UpdatableStore<PieceRecordKey, PieceRecord>
 export type PieceQueue = Queue<PieceMessage>
 export type BufferQueue = Queue<BufferMessage>
 export type BufferStore = Store<Link, BufferRecord>
 export type AggregateStore = Store<AggregateRecordKey, AggregateRecord>
 export type PieceAcceptQueue = Queue<PieceAcceptMessage>
-export type InclusionStore = QueryableStore<
-  InclusionRecordKey,
-  InclusionRecord,
-  InclusionRecordQueryByGroup
->
+export type InclusionStore = Store<InclusionRecordKey, InclusionRecord> &
+  QueryableStore<InclusionRecordQueryByGroup, InclusionRecord>
 export type AggregateOfferQueue = Queue<AggregateOfferMessage>
 
 export interface ServiceContext {

packages/filecoin-api/src/aggregator/service.js

@@ -82,7 +82,7 @@ export const pieceAccept = async ({ capability }, context) => {
       error: new StoreOperationFailed(getInclusionRes.error?.message),
     }
   }
-  if (!getInclusionRes.ok.length) {
+  if (!getInclusionRes.ok.results.length) {
     return {
       error: new UnexpectedState(
         `no inclusion proof found for pair {${piece}, ${group}}`
@@ -91,7 +91,7 @@ export const pieceAccept = async ({ capability }, context) => {
   }
 
   // Get buffered pieces
-  const [{ aggregate, inclusion }] = getInclusionRes.ok
+  const [{ aggregate, inclusion }] = getInclusionRes.ok.results
   const getAggregateRes = await context.aggregateStore.get({ aggregate })
   if (getAggregateRes.error) {
     return {

packages/filecoin-api/src/deal-tracker/api.ts

@@ -1,12 +1,9 @@
 import type { Signer } from '@ucanto/interface'
 import { PieceLink } from '@web3-storage/data-segment'
-import { QueryableStore } from '../types.js'
+import { Store, QueryableStore } from '../types.js'
 
-export type DealStore = QueryableStore<
-  DealRecordKey,
-  DealRecord,
-  DealRecordQueryByPiece
->
+export type DealStore = Store<DealRecordKey, DealRecord> &
+  QueryableStore<DealRecordQueryByPiece, DealRecord>
 
 export interface ServiceContext {
   /**

packages/filecoin-api/src/deal-tracker/service.js

@@ -19,16 +19,22 @@ import { StoreOperationFailed } from '../errors.js'
 export const dealInfo = async ({ capability }, context) => {
   const { piece } = capability.nb
 
-  const storeGet = await context.dealStore.query({ piece })
-  if (storeGet.error) {
-    return {
-      error: new StoreOperationFailed(storeGet.error.message),
+  const records = []
+  /** @type {string|undefined} */
+  let cursor
+  do {
+    const storeQuery = await context.dealStore.query({ piece }, { cursor })
+    if (storeQuery.error) {
+      return { error: new StoreOperationFailed(storeQuery.error.message) }
     }
-  }
+
+    records.push(...storeQuery.ok.results)
+    cursor = storeQuery.ok.cursor
+  } while (cursor)
 
   return {
     ok: {
-      deals: storeGet.ok.reduce((acc, curr) => {
+      deals: records.reduce((acc, curr) => {
         acc[`${curr.dealId}`] = {
           provider: curr.provider,
         }

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

@@ -5,17 +5,17 @@ import {
   DealTrackerService,
 } from '@web3-storage/filecoin-client/types'
 import {
+  Store,
   UpdatableStore,
-  UpdatableAndQueryableStore,
+  QueryableStore,
   ServiceConfig,
 } from '../types.js'
 
-export type OfferStore<OfferDoc> = UpdatableStore<string, OfferDoc>
-export type AggregateStore = UpdatableAndQueryableStore<
-  AggregateRecordKey,
-  AggregateRecord,
-  Pick<AggregateRecord, 'status'>
->
+export type OfferStore<OfferDoc> = Store<string, OfferDoc> &
+  UpdatableStore<string, OfferDoc>
+export type AggregateStore = Store<AggregateRecordKey, AggregateRecord> &
+  UpdatableStore<AggregateRecordKey, AggregateRecord> &
+  QueryableStore<Pick<AggregateRecord, 'status'>, AggregateRecord>
 
 export interface ServiceContext<OfferDoc = OfferDocument> {
   id: Signer

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

@@ -7,6 +7,9 @@ import { StoreOperationFailed } from '../errors.js'
  * @typedef {import('./api.js').AggregateRecordKey} AggregateRecordKey
  */
 
+/** Max items per page of query. */
+const MAX_PAGE_SIZE = 20
+
 /**
  * On aggregate insert event, update offer key with date to be retrievable by broker.
  *
@@ -55,45 +58,57 @@ export const handleAggregateUpdatedStatus = async (context, record) => {
  * @param {import('./api.js').CronContext} context
  */
 export const handleCronTick = async (context) => {
-  // Get offered deals pending approval/rejection
-  const offeredDeals = await context.aggregateStore.query({
-    status: 'offered',
-  })
-  if (offeredDeals.error) {
-    return {
-      error: offeredDeals.error,
+  let totalDealsCount = 0
+  let updatedDealsCount = 0
+  /** @type {string|undefined} */
+  let cursor
+  do {
+    // Get offered deals pending approval/rejection
+    const offeredDeals = await context.aggregateStore.query(
+      {
+        status: 'offered',
+      },
+      { cursor, size: MAX_PAGE_SIZE }
+    )
+    if (offeredDeals.error) {
+      return {
+        error: offeredDeals.error,
+      }
     }
-  }
-
-  // Update approved deals from the ones resolved
-  const updatedResponses = await Promise.all(
-    offeredDeals.ok.map((deal) =>
-      updateApprovedDeals({
-        deal,
-        aggregateStore: context.aggregateStore,
-        dealTrackerServiceConnection: context.dealTrackerService.connection,
-        dealTrackerInvocationConfig:
-          context.dealTrackerService.invocationConfig,
-      })
+    totalDealsCount += offeredDeals.ok.results.length
+
+    // Update approved deals from the ones resolved
+    const updatedResponses = await Promise.all(
+      offeredDeals.ok.results.map((deal) =>
+        updateApprovedDeals({
+          deal,
+          aggregateStore: context.aggregateStore,
+          dealTrackerServiceConnection: context.dealTrackerService.connection,
+          dealTrackerInvocationConfig:
+            context.dealTrackerService.invocationConfig,
+        })
+      )
     )
-  )
 
-  // Fail if one or more update operations did not succeed.
-  // The successful ones are still valid, but we should keep track of errors for monitoring/alerting.
-  const updateErrorResponse = updatedResponses.find((r) => r.error)
-  if (updateErrorResponse) {
-    return {
-      error: updateErrorResponse.error,
+    // Fail if one or more update operations did not succeed.
+    // The successful ones are still valid, but we should keep track of errors for monitoring/alerting.
+    const updateErrorResponse = updatedResponses.find((r) => r.error)
+    if (updateErrorResponse) {
+      return {
+        error: updateErrorResponse.error,
+      }
     }
-  }
+
+    updatedDealsCount += updatedResponses.filter((r) => r.ok?.updated).length
+    cursor = offeredDeals.ok.cursor
+  } while (cursor)
 
   // Return successful update operation
   // Include in response the ones that were Updated, and the ones still pending response.
-  const updatedDealsCount = updatedResponses.filter((r) => r.ok?.updated).length
   return {
     ok: {
       updatedCount: updatedDealsCount,
-      pendingCount: updatedResponses.length - updatedDealsCount,
+      pendingCount: totalDealsCount - updatedDealsCount,
     },
   }
 }
@@ -103,7 +118,7 @@ export const handleCronTick = async (context) => {
  *
  * @param {object} context
  * @param {AggregateRecord} context.deal
- * @param {import('../types.js').UpdatableAndQueryableStore<AggregateRecordKey, AggregateRecord, Pick<AggregateRecord, 'status'>>} context.aggregateStore
+ * @param {import('../types.js').UpdatableStore<AggregateRecordKey, AggregateRecord>} context.aggregateStore
  * @param {import('@ucanto/interface').ConnectionView<any>} context.dealTrackerServiceConnection
  * @param {import('@web3-storage/filecoin-client/types').InvocationConfig} context.dealTrackerInvocationConfig
  */

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

@@ -18,17 +18,16 @@ import {
 } from '@web3-storage/filecoin-client/types'
 import {
   Store,
-  UpdatableAndQueryableStore,
+  UpdatableStore,
+  QueryableStore,
   Queue,
   ServiceConfig,
   StoreGetError,
 } from '../types.js'
 
-export type PieceStore = UpdatableAndQueryableStore<
-  PieceRecordKey,
-  PieceRecord,
-  Pick<PieceRecord, 'status'>
->
+export type PieceStore = Store<PieceRecordKey, PieceRecord> &
+  UpdatableStore<PieceRecordKey, PieceRecord> &
+  QueryableStore<Pick<PieceRecord, 'status'>, PieceRecord>
 export type FilecoinSubmitQueue = Queue<FilecoinSubmitMessage>
 export type PieceOfferQueue = Queue<PieceOfferMessage>
 export type TaskStore = Store<UnknownLink, Invocation>

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

@@ -1,4 +1,3 @@
-import pMap from 'p-map'
 import { Storefront, Aggregator } from '@web3-storage/filecoin-client'
 import * as AggregatorCaps from '@web3-storage/capabilities/filecoin/aggregator'
 import { Assert } from '@web3-storage/content-claims/capability'
@@ -17,9 +16,12 @@ import {
 /**
  * @typedef {import('./api.js').PieceRecord} PieceRecord
  * @typedef {import('./api.js').PieceRecordKey} PieceRecordKey
- * @typedef {import('../types.js').UpdatableAndQueryableStore<PieceRecordKey, PieceRecord, Pick<PieceRecord, 'status'>>} PieceStore
+ * @typedef {import('./api.js').PieceStore} PieceStore
  */
 
+/** Max items per page of query. */
+const MAX_PAGE_SIZE = 20
+
 /**
  * On filecoin submit queue messages, validate piece for given content and store it in store.
  *
@@ -185,49 +187,57 @@ export const handlePieceStatusUpdate = async (context, record) => {
  * @param {import('./api.js').CronContext} context
  */
 export const handleCronTick = async (context) => {
-  const submittedPieces = await context.pieceStore.query({
-    status: 'submitted',
-  })
-  if (submittedPieces.error) {
-    return {
-      error: submittedPieces.error,
-    }
-  }
-  // Update approved pieces from the ones resolved
-  const updatedResponses = await pMap(
-    submittedPieces.ok,
-    (pieceRecord) =>
-      updatePiecesWithDeal({
-        id: context.id,
-        aggregatorId: context.aggregatorId,
-        pieceRecord,
-        pieceStore: context.pieceStore,
-        taskStore: context.taskStore,
-        receiptStore: context.receiptStore,
-      }),
-    {
-      concurrency: 20,
+  let totalPiecesCount = 0
+  let updatedPiecesCount = 0
+  /** @type {string|undefined} */
+  let cursor
+  do {
+    const submittedPieces = await context.pieceStore.query(
+      {
+        status: 'submitted',
+      },
+      { cursor, size: MAX_PAGE_SIZE }
+    )
+    if (submittedPieces.error) {
+      return {
+        error: submittedPieces.error,
+      }
     }
-  )
+    totalPiecesCount += submittedPieces.ok.results.length
 
-  // Fail if one or more update operations did not succeed.
-  // The successful ones are still valid, but we should keep track of errors for monitoring/alerting.
-  const updateErrorResponse = updatedResponses.find((r) => r.error)
-  if (updateErrorResponse) {
-    return {
-      error: updateErrorResponse.error,
+    // Update approved pieces from the ones resolved
+    const updatedResponses = await Promise.all(
+      submittedPieces.ok.results.map((pieceRecord) =>
+        updatePiecesWithDeal({
+          id: context.id,
+          aggregatorId: context.aggregatorId,
+          pieceRecord,
+          pieceStore: context.pieceStore,
+          taskStore: context.taskStore,
+          receiptStore: context.receiptStore,
+        })
+      )
+    )
+
+    // Fail if one or more update operations did not succeed.
+    // The successful ones are still valid, but we should keep track of errors for monitoring/alerting.
+    const updateErrorResponse = updatedResponses.find((r) => r.error)
+    if (updateErrorResponse) {
+      return {
+        error: updateErrorResponse.error,
+      }
     }
-  }
+
+    updatedPiecesCount += updatedResponses.filter((r) => r.ok?.updated).length
+    cursor = submittedPieces.ok.cursor
+  } while (cursor)
 
   // Return successful update operation
   // Include in response the ones that were Updated, and the ones still pending response.
-  const updatedPiecesCount = updatedResponses.filter(
-    (r) => r.ok?.updated
-  ).length
   return {
     ok: {
       updatedCount: updatedPiecesCount,
-      pendingCount: updatedResponses.length - updatedPiecesCount,
+      pendingCount: totalPiecesCount - updatedPiecesCount,
     },
   }
 }