Add KV cache for D1 index. Closes #306

indexer/lib/fwss-handlers.js

@@ -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 {Env} 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.KV.delete(`${payerAddress}/${pieceCID}`)
+    }),
+  )
 }

indexer/lib/pdp-verifier-handlers.js

@@ -32,6 +32,7 @@ export async function insertDataSetPiece(
  * @param {(number | string)[]} pieceIds
  */
 export async function removeDataSetPieces(env, dataSetId, pieceIds) {
+  await clearDataSetPiecesIndexCache(env, dataSetId, pieceIds)
   await env.DB.prepare(
     `
     DELETE FROM pieces
@@ -44,3 +45,29 @@ export async function removeDataSetPieces(env, dataSetId, pieceIds) {
     .bind(String(dataSetId), ...pieceIds.map(String))
     .run()
 }
+
+/**
+ * @param {Env} 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 (${new Array(pieceIds.length)
+        .fill(null)
+        .map(() => '?')
+        .join(', ')})
+    `,
+  )
+    .bind(String(dataSetId), ...pieceIds.map(String))
+    .run()
+  await Promise.all(
+    results.map(async ({ payerAddress, pieceCID }) => {
+      await env.KV.delete(`${payerAddress}/${pieceCID}`)
+    }),
+  )
+}

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

@@ -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 = ?
@@ -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 = ?
@@ -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 {Env} 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.KV.delete(`${payerAddress}/${pieceCID}`)
+    }),
+  )
+}