feat: integrate latest filecoin-api dependency

storacha · Jul 23, 2024 · 3e1afb9 · 3e1afb9
1 parent 66c9e17
commit 3e1afb9
Show file tree

Hide file tree

Showing 7 changed files with 73 additions and 60 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/packages/core/package.json b/packages/core/package.json
@@ -9,37 +9,37 @@
     "test:all": "ava --serial --no-worker-threads --verbose --timeout=120s test/{*.test.js,**/*.test.js}"
   },
   "dependencies": {
-    "@serverless-stack/node": "^1.18.4",
     "@aws-sdk/client-dynamodb": "^3.363.0",
-    "@aws-sdk/client-sqs": "^3.363.0",
     "@aws-sdk/client-s3": "^3.363.0",
+    "@aws-sdk/client-sqs": "^3.363.0",
     "@aws-sdk/util-dynamodb": "3.363.0",
     "@ipld/dag-json": "10.1.5",
     "@ipld/dag-ucan": "^3.3.2",
+    "@serverless-stack/node": "^1.18.4",
     "@ucanto/client": "^9.0.1",
     "@ucanto/interface": "^10.0.1",
     "@ucanto/principal": "^9.0.1",
     "@ucanto/server": "^10.0.0",
     "@ucanto/transport": "^9.1.1",
     "@web3-storage/capabilities": "^17.1.0",
     "@web3-storage/data-segment": "^5.0.0",
-    "@web3-storage/filecoin-api": "^7.0.0",
+    "@web3-storage/filecoin-api": "^7.2.0",
     "@web3-storage/filecoin-client": "^3.3.3",
     "fzstd": "^0.1.0",
     "multiformats": "12.0.1",
-    "uint8arrays": "^4.0.4",
-    "pretty-ms": "^8.0.0",
     "p-all": "^5.0.0",
     "p-retry": "^5.1.2",
-    "stream-read-all": "^4.0.0"
+    "pretty-ms": "^8.0.0",
+    "stream-read-all": "^4.0.0",
+    "uint8arrays": "^4.0.4"
   },
   "devDependencies": {
     "@ipld/car": "5.1.1",
     "@web-std/blob": "3.0.4",
     "ava": "^5.3.0",
+    "delay": "^6.0.0",
     "nanoid": "^4.0.0",
     "npm-run-all": "^4.1.5",
-    "delay": "^6.0.0",
     "p-defer": "^4.0.0",
     "p-wait-for": "^5.0.2",
     "sqs-consumer": "^7.2.2",
@@ -58,4 +58,4 @@
       "jsdoc/require-param": "off"
     }
   }
-}
+}
diff --git a/packages/core/src/store/aggregator-inclusion-store.js b/packages/core/src/store/aggregator-inclusion-store.js
@@ -167,7 +167,7 @@ export function createClient (conf, context) {
         ok: true
       }
     },
-    query: async (search) => {
+    query: async (search, options) => {
       const queryProps = encodeQueryProps(search)
       if (!queryProps) {
         return {
@@ -178,40 +178,41 @@ export function createClient (conf, context) {
       // @ts-ignore query props partial
       const queryCmd = new QueryCommand({
         TableName: context.tableName,
-        ...queryProps
+        ...queryProps,
+        ExclusiveStartKey: options?.cursor ? JSON.parse(options.cursor) : undefined,
+        Limit: options?.size
       })
 
       let res
       try {
         res = await tableclient.send(queryCmd)
       } catch (/** @type {any} */ error) {
+        console.error(error)
         return {
           error: new StoreOperationFailed(error.message)
         }
       }
 
-      // TODO: handle pulling the entire list. Even with renewals we are far away from this being needed
-      if (!res.Items) {
-        return {
-          ok: /** @type {AggregatorInclusionRecord[]} */ ([])
-        }
-      }
-
       const inclusionRecordsGet = await Promise.all(
-        res.Items.map((item) => getInclusionRecordFromInclusionStoreRecord(
+        (res.Items ?? []).map((item) => getInclusionRecordFromInclusionStoreRecord(
           /** @type {InclusionStoreRecord} */ (unmarshall(item)),
           context.inclusionProofStore
         ))
       )
 
+      const records = []
       for (const get of inclusionRecordsGet) {
         if (get.error) {
           return get
         }
+        records.push(get.ok)
       }
 
       return {
-        ok: /** @type {AggregatorInclusionRecord[]} */ (inclusionRecordsGet.map(get => get.ok))
+        ok: {
+          results: records,
+          ...(res.LastEvaluatedKey ? { cursor: JSON.stringify(res.LastEvaluatedKey) } : {})
+        }
       }
     },
   }

diff --git a/packages/core/src/store/deal-store.js b/packages/core/src/store/deal-store.js
@@ -140,7 +140,7 @@ export function createClient (conf, context) {
         ok: true
       }
     },
-    query: async (search) => {
+    query: async (search, options) => {
       const dealStoreRecordQueryByPiece = encodeQueryByPiece(search)
       const queryCmd = new QueryCommand({
         TableName: context.tableName,
@@ -150,24 +150,28 @@ export function createClient (conf, context) {
             ComparisonOperator: 'EQ',
             AttributeValueList: [{ S: dealStoreRecordQueryByPiece.piece }]
           }
-        }
+        },
+        ExclusiveStartKey: options?.cursor ? JSON.parse(options.cursor) : undefined,
+        Limit: options?.size
       })
 
       let res
       try {
         res = await tableclient.send(queryCmd)
       } catch (/** @type {any} */ error) {
+        console.error(error)
         return {
           error: new StoreOperationFailed(error.message)
         }
       }
 
-      // TODO: handle pulling the entire list. currently we only support 2 providers so
-      // this list should not be longer than the default page size so this is not terribly urgent.
       return {
-        ok: res.Items ? res.Items.map(item => decodeRecord(
-          /** @type {InferStoreRecord} */ (unmarshall(item))
-        )) : []
+        ok: {
+          results: (res.Items ?? []).map(item => decodeRecord(
+            /** @type {InferStoreRecord} */ (unmarshall(item))
+          )),
+          ...(res.LastEvaluatedKey ? { cursor: JSON.stringify(res.LastEvaluatedKey) } : {})
+        }
       }
     }
   }

diff --git a/packages/core/src/store/dealer-aggregate-store.js b/packages/core/src/store/dealer-aggregate-store.js
@@ -243,7 +243,7 @@ export function createClient (conf, context) {
         )
       }
     },
-    query: async (search) => {
+    query: async (search, options) => {
       const queryProps = encodeQueryProps(search)
       if (!queryProps) {
         return {
@@ -254,23 +254,28 @@ export function createClient (conf, context) {
       // @ts-ignore query props partial
       const queryCmd = new QueryCommand({
         TableName: context.tableName,
-        ...queryProps
+        ...queryProps,
+        ExclusiveStartKey: options?.cursor ? JSON.parse(options.cursor) : undefined,
+        Limit: options?.size
       })
 
       let res
       try {
         res = await tableclient.send(queryCmd)
       } catch (/** @type {any} */ error) {
+        console.error(error)
         return {
           error: new StoreOperationFailed(error.message)
         }
       }
 
-      // TODO: handle pulling the entire list. Even with renewals we are far away from this being needed
       return {
-        ok: res.Items ? res.Items.map(item => decodeRecord(
-          /** @type {DealerAggregateStoreRecord} */ (unmarshall(item))
-        )) : []
+        ok: {
+          results: (res.Items ?? []).map(item => decodeRecord(
+            /** @type {DealerAggregateStoreRecord} */ (unmarshall(item))
+          )),
+          ...(res.LastEvaluatedKey ? { cursor: JSON.stringify(res.LastEvaluatedKey) } : {})
+        }
       }
     },
   }

diff --git a/packages/functions/package.json b/packages/functions/package.json
@@ -6,19 +6,19 @@
     "typecheck": "tsc --build"
   },
   "dependencies": {
-    "uint8arrays": "^4.0.6",
     "@ipld/dag-ucan": "^3.4.0",
     "@ucanto/core": "^10.0.1",
     "@ucanto/server": "^10.0.0",
     "@ucanto/transport": "^9.1.1",
+    "@w3filecoin/core": "*",
     "@web3-storage/data-segment": "5.0.0",
-    "@web3-storage/filecoin-api": "^7.0.0",
+    "@web3-storage/filecoin-api": "^7.2.0",
     "@web3-storage/filecoin-client": "3.3.3",
-    "@w3filecoin/core": "*"
+    "uint8arrays": "^4.0.6"
   },
   "devDependencies": {
     "@sentry/serverless": "7.52.1",
     "sst": "^2.8.3",
     "typescript": "^5.0.4"
   }
-}
+}
diff --git a/packages/tools/get-aggregates-pending-deals.js b/packages/tools/get-aggregates-pending-deals.js
@@ -8,16 +8,19 @@ const aggregateStore = createAggregateStoreClient({
   tableName: 'prod-w3filecoin-dealer-aggregate-store'
 })
 
-// Get offered aggregates pending approval/rejection
-const offeredAggregates = await aggregateStore.query({
-  status: 'offered',
-})
-if (offeredAggregates.error) {
-  throw offeredAggregates.error
-}
-
-console.log('Offered aggregates page size:', offeredAggregates.ok.length, '\n')
 console.log('Aggregate offer list:')
-for (const aggregate of offeredAggregates.ok) {
-  console.log(`${aggregate.aggregate.link()} at ${aggregate.insertedAt}`)
-}
+let cursor
+do {
+  // Get offered aggregates pending approval/rejection
+  const offeredAggregates = await aggregateStore.query({
+    status: 'offered',
+  })
+  if (offeredAggregates.error) {
+    throw offeredAggregates.error
+  }
+
+  for (const aggregate of offeredAggregates.ok.results) {
+    console.log(`${aggregate.aggregate.link()} at ${aggregate.insertedAt}`)
+  }
+  cursor = offeredAggregates.ok.cursor
+} while (cursor)