check custom filter in persistence

Author: apexearth

src/processors/persistence-filters.ts

@@ -1,21 +1,46 @@
-import { Log, Trace, logFilter, traceFilter } from '@originprotocol/squid-utils'
+import { Context, Log, Trace, logFilter, traceFilter } from '@originprotocol/squid-utils'
 
 type LogFilterInstance = ReturnType<typeof logFilter>
 type TraceFilterInstance = ReturnType<typeof traceFilter>
+type CustomFilter = (ctx: Context, log: Log) => Promise<{ include: boolean }> | { include: boolean }
 
-const logFilters: LogFilterInstance[] = []
+interface RegisteredLogFilter {
+  filter: LogFilterInstance
+  customFilter?: CustomFilter
+}
+
+const logFilters: RegisteredLogFilter[] = []
 const traceFilters: TraceFilterInstance[] = []
 
-export const registerLogFilter = (filter: LogFilterInstance) => {
-  logFilters.push(filter)
+export const registerLogFilter = (filter: LogFilterInstance, customFilter?: CustomFilter) => {
+  logFilters.push({ filter, customFilter })
 }
 
 export const registerTraceFilter = (filter: TraceFilterInstance) => {
   traceFilters.push(filter)
 }
 
+/**
+ * Check if a log matches any registered filter.
+ * Filters with a customFilter require the async version (matchesAnyLogFilterAsync).
+ * This sync version only checks filters without customFilters.
+ */
 export const matchesAnyLogFilter = (log: Log): boolean => {
-  return logFilters.some((f) => f.matches(log))
+  return logFilters.some((f) => f.filter.matches(log))
+}
+
+/**
+ * Check if a log matches any registered filter, including custom filters.
+ * A log matches if any filter's subscription matches AND its customFilter (if present) returns include: true.
+ */
+export const matchesAnyLogFilterAsync = async (ctx: Context, log: Log): Promise<boolean> => {
+  for (const { filter, customFilter } of logFilters) {
+    if (!filter.matches(log)) continue
+    if (!customFilter) return true
+    const result = await customFilter(ctx, log)
+    if (result.include) return true
+  }
+  return false
 }
 
 export const matchesAnyTraceFilter = (trace: Trace): boolean => {

src/processors/persistence.ts

@@ -4,7 +4,7 @@ import { AbiData } from '../model'
 import { EventRecord } from '../model'
 import { TraceRecord } from '../model'
 import { abiRegistry } from '../utils/abi-registry'
-import { matchesAnyLogFilter, matchesAnyTraceFilter } from './persistence-filters'
+import { matchesAnyLogFilterAsync, matchesAnyTraceFilter } from './persistence-filters'
 
 export const persistenceProcessor = defineProcessor({
   name: 'Persistence',
@@ -27,7 +27,7 @@ export const persistenceProcessor = defineProcessor({
       const blockHash = block.header.hash
 
       for (const log of block.logs) {
-        if (!matchesAnyLogFilter(log)) continue
+        if (!(await matchesAnyLogFilterAsync(ctx, log))) continue
         const id = `${ctx.chain.id}:${blockNumber}:${log.transactionIndex}:${log.logIndex}`
         const topic0 = log.topics[0]
         if (!topic0) continue

src/templates/event.ts

@@ -48,7 +48,7 @@ export const createEventProcessor = ({
       topic3,
       transaction: true,
     })
-    registerLogFilter(filter)
+    registerLogFilter(filter, track.customFilter)
     for (const af of additionalFilters ?? []) registerLogFilter(af)
     return { track, filter, entries, additionalFilters }
   })