Add KV cache for D1 index. Closes #306

bad-bits/worker-configuration.d.ts

@@ -465,7 +465,11 @@
     deleteAlarm(options?: DurableObjectSetAlarmOptions): Promise<void>;
     sync(): Promise<void>;
     sql: SqlStorage;
+<<<<<<< HEAD
+    INDEX_CACHE_KV: SyncKvStorage;
+=======
     kv: SyncKvStorage;
+>>>>>>> main
     transactionSync<T>(closure: () => T): T;
     getCurrentBookmark(): Promise<string>;
     getBookmarkForTime(timestamp: number | Date): Promise<string>;

indexer/bin/indexer.js

@@ -25,7 +25,7 @@ import { CID } from 'multiformats/cid'
  *   WALLET_SCREENING_BATCH_SIZE: 1 | 10
  *   WALLET_SCREENING_STALE_THRESHOLD_MS: 86400000 | 21600000
  *   DB: D1Database
- *   KV: KVNamespace
+ *   INDEX_CACHE_KV: KVNamespace
  *   RETRY_QUEUE: Queue
  *   SECRET_HEADER_KEY: string
  *   SECRET_HEADER_VALUE: string

indexer/lib/fwss-handlers.js

@@ -64,7 +64,7 @@ export async function handleFWSSDataSetCreated(
 /**
  * Handle Filecoin Warm Storage Service service termination
  *
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {any} payload
  * @throws {Error}
  */
@@ -78,4 +78,27 @@ export async function handleFWSSServiceTerminated(env, payload) {
   )
     .bind(String(payload.data_set_id))
     .run()
+  await clearDataSetIndexCache(env, payload.data_set_id)
+}
+
+/**
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
+ * @param {string | number} dataSetId
+ */
+async function clearDataSetIndexCache(env, dataSetId) {
+  const { results } = await env.DB.prepare(
+    `
+      SELECT data_sets.payer_address AS payerAddress, pieces.cid AS pieceCID
+      FROM data_sets
+      INNER JOIN pieces ON pieces.data_set_id = data_sets.id
+      WHERE data_sets.id = ?
+    `,
+  )
+    .bind(String(dataSetId))
+    .run()
+  await Promise.all(
+    results.map(async ({ payerAddress, pieceCID }) => {
+      await env.INDEX_CACHE_KV.delete(`${payerAddress}/${pieceCID}`)
+    }),
+  )
 }

indexer/lib/pdp-verifier-handlers.js

@@ -27,20 +27,51 @@ export async function insertDataSetPiece(
 }
 
 /**
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {number | string} dataSetId
  * @param {(number | string)[]} pieceIds
  */
 export async function removeDataSetPieces(env, dataSetId, pieceIds) {
+  await clearDataSetPiecesIndexCache(env, dataSetId, pieceIds)
   await env.DB.prepare(
     `
     DELETE FROM pieces
-    WHERE data_set_id = ? AND id IN (${new Array(pieceIds.length)
-      .fill(null)
-      .map(() => '?')
-      .join(', ')})
+    WHERE data_set_id = ? AND id IN (${sqlPlaceholders(pieceIds.length)})
+    `,
+  )
+    .bind(String(dataSetId), ...pieceIds.map(String))
+    .run()
+}
+
+/**
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
+ * @param {number | string} dataSetId
+ * @param {(number | string)[]} pieceIds
+ */
+async function clearDataSetPiecesIndexCache(env, dataSetId, pieceIds) {
+  const { results } = await env.DB.prepare(
+    `
+      SELECT data_sets.payer_address AS payerAddress, pieces.cid AS pieceCID
+      FROM data_sets
+      INNER JOIN pieces ON pieces.data_set_id = data_sets.id
+      WHERE data_sets.id = ? AND pieces.id IN (${sqlPlaceholders(pieceIds.length)})
     `,
   )
     .bind(String(dataSetId), ...pieceIds.map(String))
     .run()
+  await Promise.all(
+    results.map(async ({ payerAddress, pieceCID }) => {
+      await env.INDEX_CACHE_KV.delete(`${payerAddress}/${pieceCID}`)
+    }),
+  )
 }
+
+/**
+ * @param {number} count
+ * @returns String
+ */
+const sqlPlaceholders = (count) =>
+  new Array(count)
+    .fill(null)
+    .map(() => '?')
+    .join(', ')

indexer/lib/service-provider-registry-handlers.js

@@ -3,7 +3,7 @@ import validator from 'validator'
 const PRODUCT_TYPE_PDP = 0
 
 /**
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {string | number} providerId
  * @param {string | number} productType
  * @param {string} serviceUrl
@@ -34,7 +34,7 @@ export async function handleProductAdded(
 }
 
 /**
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {string | number} providerId
  * @param {string | number} productType
  * @param {string} serviceUrl
@@ -65,7 +65,7 @@ export async function handleProductUpdated(
 }
 
 /**
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {string | number} providerId
  * @param {string | number} productType
  * @returns {Promise<Response>}
@@ -85,6 +85,7 @@ export async function handleProductRemoved(env, providerId, productType) {
     return new Response('OK', { status: 200 })
   }
 
+  await clearServiceProviderIndexCache(env, providerId)
   const result = await env.DB.prepare(
     `
         DELETE FROM service_providers WHERE id = ?
@@ -99,7 +100,7 @@ export async function handleProductRemoved(env, providerId, productType) {
 }
 
 /**
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {string | number} providerId
  * @returns {Promise<Response>}
  */
@@ -111,6 +112,7 @@ export async function handleProviderRemoved(env, providerId) {
     return new Response('Bad Request', { status: 400 })
   }
 
+  await clearServiceProviderIndexCache(env, providerId)
   const result = await env.DB.prepare(
     `
         DELETE FROM service_providers WHERE id = ?
@@ -125,7 +127,7 @@ export async function handleProviderRemoved(env, providerId) {
 }
 
 /**
- * @param {{ DB: D1Database }} env
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
  * @param {string | number} providerId
  * @param {string} serviceUrl
  * @returns {Promise<Response>}
@@ -141,6 +143,7 @@ async function handleProviderServiceUrlUpdate(env, providerId, serviceUrl) {
     `Provider service url updated (providerId=${providerId}, serviceUrl=${serviceUrl})`,
   )
 
+  await clearServiceProviderIndexCache(env, providerId)
   await env.DB.prepare(
     `
         INSERT INTO service_providers (
@@ -156,3 +159,25 @@ async function handleProviderServiceUrlUpdate(env, providerId, serviceUrl) {
     .run()
   return new Response('OK', { status: 200 })
 }
+
+/**
+ * @param {{ DB: D1Database; INDEX_CACHE_KV: KVNamespace }} env
+ * @param {string | number} providerId
+ */
+async function clearServiceProviderIndexCache(env, providerId) {
+  const { results } = await env.DB.prepare(
+    `
+      SELECT data_sets.payer_address AS payerAddress, pieces.cid AS pieceCID
+      FROM data_sets
+      INNER JOIN pieces ON pieces.data_set_id = data_sets.id
+      WHERE data_sets.service_provider_id = ?
+    `,
+  )
+    .bind(String(providerId))
+    .run()
+  await Promise.all(
+    results.map(async ({ payerAddress, pieceCID }) => {
+      await env.INDEX_CACHE_KV.delete(`${payerAddress}/${pieceCID}`)
+    }),
+  )
+}