feat: add filter-based index cleanup for L2 data items

docker-compose.yaml

@@ -216,6 +216,12 @@ services:
       - WRITE_ANS104_DATA_ITEM_DB_SIGNATURES=${WRITE_ANS104_DATA_ITEM_DB_SIGNATURES:-}
       - WRITE_TRANSACTION_DB_SIGNATURES=${WRITE_TRANSACTION_DB_SIGNATURES:-}
       - ENABLE_DATA_DB_WAL_CLEANUP=${ENABLE_DATA_DB_WAL_CLEANUP:-}
+      - ENABLE_INDEX_CLEANUP_WORKER=${ENABLE_INDEX_CLEANUP_WORKER:-}
+      - INDEX_CLEANUP_INTERVAL_SECONDS=${INDEX_CLEANUP_INTERVAL_SECONDS:-}
+      - INDEX_CLEANUP_BATCH_SIZE=${INDEX_CLEANUP_BATCH_SIZE:-}
+      - INDEX_CLEANUP_FILTER=${INDEX_CLEANUP_FILTER:-}
+      - INDEX_CLEANUP_MIN_AGE_SECONDS=${INDEX_CLEANUP_MIN_AGE_SECONDS:-}
+      - INDEX_CLEANUP_DRY_RUN=${INDEX_CLEANUP_DRY_RUN:-}
       - MAX_DATA_ITEM_QUEUE_SIZE=${MAX_DATA_ITEM_QUEUE_SIZE:-}
       - TAG_SELECTIVITY=${TAG_SELECTIVITY:-}
       - MAX_EXPECTED_DATA_ITEM_INDEXING_INTERVAL_SECONDS=${MAX_EXPECTED_DATA_ITEM_INDEXING_INTERVAL_SECONDS:-}

src/config.ts

@@ -1360,6 +1360,44 @@ export const PREFERRED_ARNS_BASE_NAMES = new Set(
   env.varOrDefault('PREFERRED_ARNS_BASE_NAMES', '').split(','),
 );
 
+//
+// Index Cleanup
+//
+
+// Enable periodic index cleanup worker
+export const ENABLE_INDEX_CLEANUP_WORKER =
+  env.varOrDefault('ENABLE_INDEX_CLEANUP_WORKER', 'false') === 'true';
+
+// Interval in seconds between cleanup cycles (default: 24 hours)
+export const INDEX_CLEANUP_INTERVAL_SECONDS = env.positiveIntOrDefault(
+  'INDEX_CLEANUP_INTERVAL_SECONDS',
+  60 * 60 * 24,
+);
+
+// Number of data items to process per batch
+export const INDEX_CLEANUP_BATCH_SIZE = env.positiveIntOrDefault(
+  'INDEX_CLEANUP_BATCH_SIZE',
+  1000,
+);
+
+// Filter determining which data items to clean up (flat JSON format)
+// Example: {"owners":["addr1"],"tags":[{"name":"App-Name","values":["ArDrive"]}]}
+export const INDEX_CLEANUP_FILTER = env.varOrDefault(
+  'INDEX_CLEANUP_FILTER',
+  '{}',
+);
+
+// How old data items must be (in seconds) before they are eligible for cleanup
+// Default: 30 days
+export const INDEX_CLEANUP_MIN_AGE_SECONDS = env.positiveIntOrDefault(
+  'INDEX_CLEANUP_MIN_AGE_SECONDS',
+  60 * 60 * 24 * 30,
+);
+
+// When true, only logs what would be deleted without deleting
+export const INDEX_CLEANUP_DRY_RUN =
+  env.varOrDefault('INDEX_CLEANUP_DRY_RUN', 'true') === 'true';
+
 //
 // Webhooks
 //

src/database/composite-clickhouse.ts

@@ -16,7 +16,11 @@ import {
   hexToB64Url,
   utf8ToB64Url,
 } from '../lib/encoding.js';
-import { GqlTransactionsResult, GqlQueryable } from '../types.js';
+import {
+  GqlTransactionsResult,
+  GqlQueryable,
+  ClickHouseIndexCleanup,
+} from '../types.js';
 
 export function encodeTransactionGqlCursor({
   height,
@@ -87,7 +91,9 @@ function inB64UrlStrings(xs: string[]) {
   return sql(xs.map((x) => `unhex('${b64UrlToHex(x)}')`).join(', '));
 }
 
-export class CompositeClickHouseDatabase implements GqlQueryable {
+export class CompositeClickHouseDatabase
+  implements GqlQueryable, ClickHouseIndexCleanup
+{
   private log: winston.Logger;
   private clickhouseClient: ClickHouseClient;
   private gqlQueryable: GqlQueryable;
@@ -509,4 +515,48 @@ export class CompositeClickHouseDatabase implements GqlQueryable {
   }) {
     return this.gqlQueryable.getGqlBlocks(args);
   }
+
+  async deleteDataItemsByIds(ids: string[]): Promise<void> {
+    if (ids.length === 0) return;
+
+    const tables = [
+      'transactions',
+      'id_transactions',
+      'owner_transactions',
+      'target_transactions',
+    ];
+
+    const SUB_BATCH_SIZE = 500;
+    const failures: { table: string; error: string }[] = [];
+
+    for (let i = 0; i < ids.length; i += SUB_BATCH_SIZE) {
+      const batch = ids.slice(i, i + SUB_BATCH_SIZE);
+      const idList = batch.map((id) => `unhex('${b64UrlToHex(id)}')`).join(',');
+
+      for (const table of tables) {
+        try {
+          await this.clickhouseClient.command({
+            query: `ALTER TABLE ${table} DELETE WHERE is_data_item = true AND id IN (${idList})`,
+          });
+        } catch (error: any) {
+          this.log.error('ClickHouse index cleanup error', {
+            table,
+            batchSize: batch.length,
+            error: error?.message,
+          });
+          failures.push({ table, error: error?.message });
+        }
+      }
+    }
+
+    if (failures.length > 0) {
+      throw new Error(
+        `ClickHouse index cleanup failed for ${failures.length} table(s): ${failures.map((f) => `${f.table}: ${f.error}`).join('; ')}`,
+      );
+    }
+
+    this.log.info('ClickHouse index cleanup mutations submitted', {
+      totalIds: ids.length,
+    });
+  }
 }