index escrow events

src/@types/Escrow.ts

@@ -14,3 +14,22 @@ export interface EscrowLock {
   expiry: BigInt
   token: string
 }
+
+export interface EscrowEvent {
+  id: string
+  eventType: string
+  chainId: number
+  contract: string
+  block: number
+  txHash: string
+  payer?: string
+  payee?: string
+  token?: string
+  jobId?: string
+  amount?: string
+  expiry?: string
+  proof?: string
+  maxLockedAmount?: string
+  maxLockSeconds?: string
+  maxLockCounts?: string
+}

src/@types/commands.ts

@@ -106,6 +106,18 @@ export interface QueryCommand extends Command {
   maxResultsPerPage?: number
   pageNumber?: number
 }
+
+export interface GetEscrowEventsCommand extends Command {
+  chainId?: number
+  eventType?: string
+  payer?: string
+  payee?: string
+  token?: string
+  jobId?: string
+  txId?: string
+  maxResultsPerPage?: number
+  pageNumber?: number
+}
 export interface ReindexCommand extends Command {
   txId: string
   chainId: number

src/components/Indexer/processor.ts

@@ -20,6 +20,7 @@ import {
   NewAccessListEventProcessor,
   AddressAddedEventProcessor,
   AddressRemovedEventProcessor,
+  EscrowEventProcessor,
   ProcessorConstructor
 } from './processors/index.js'
 import { findEventByKey } from './utils.js'
@@ -42,7 +43,13 @@ const EVENT_PROCESSOR_MAP: Record<string, ProcessorConstructor> = {
   [EVENTS.EXCHANGE_RATE_CHANGED]: ExchangeRateChangedEventProcessor,
   [EVENTS.NEW_ACCESS_LIST]: NewAccessListEventProcessor,
   [EVENTS.ADDRESS_ADDED]: AddressAddedEventProcessor,
-  [EVENTS.ADDRESS_REMOVED]: AddressRemovedEventProcessor
+  [EVENTS.ADDRESS_REMOVED]: AddressRemovedEventProcessor,
+  [EVENTS.ESCROW_AUTH]: EscrowEventProcessor,
+  [EVENTS.ESCROW_LOCK]: EscrowEventProcessor,
+  [EVENTS.ESCROW_CLAIMED]: EscrowEventProcessor,
+  [EVENTS.ESCROW_CANCELED]: EscrowEventProcessor,
+  [EVENTS.ESCROW_DEPOSIT]: EscrowEventProcessor,
+  [EVENTS.ESCROW_WITHDRAW]: EscrowEventProcessor
 }
 
 const processorInstances = new Map<string, BaseEventProcessor>()

src/components/Indexer/processors/EscrowEventProcessor.ts

@@ -0,0 +1,112 @@
+import { ethers, Signer, FallbackProvider, Interface } from 'ethers'
+import { INDEXER_LOGGER } from '../../../utils/logging/common.js'
+import { LOG_LEVELS_STR } from '../../../utils/logging/Logger.js'
+import { BaseEventProcessor } from './BaseProcessor.js'
+import { getContractAddress } from '../utils.js'
+import { EVENTS } from '../../../utils/constants.js'
+import { EscrowEvent } from '../../../@types/Escrow.js'
+import { OceanNodeConfig } from '../../../@types/OceanNode.js'
+import EscrowJson from '@oceanprotocol/contracts/artifacts/contracts/escrow/Escrow.sol/Escrow.json' with { type: 'json' }
+
+const escrowInterface = new Interface(EscrowJson.abi)
+
+const addr = (v: any): string => v?.toString().toLowerCase()
+const num = (v: any): string => v?.toString()
+
+export class EscrowEventProcessor extends BaseEventProcessor {
+  private readonly escrowAddress: string
+
+  constructor(chainId: number, config: OceanNodeConfig) {
+    super(chainId, config)
+    this.escrowAddress = getContractAddress(chainId, 'Escrow')
+  }
+
+  async processEvent(
+    event: ethers.Log,
+    chainId: number,
+    signer: Signer,
+    provider: FallbackProvider,
+    eventName?: string
+  ): Promise<any> {
+    try {
+      if (
+        !this.escrowAddress ||
+        event.address.toLowerCase() !== this.escrowAddress.toLowerCase()
+      ) {
+        return null
+      }
+
+      const decoded = escrowInterface.parseLog({
+        topics: Array.from(event.topics),
+        data: event.data
+      })
+      if (!decoded) return null
+
+      const { args } = decoded
+      const record: EscrowEvent = {
+        id: `${event.transactionHash}-${event.index}`,
+        eventType: eventName,
+        chainId,
+        contract: event.address.toLowerCase(),
+        block: event.blockNumber,
+        txHash: event.transactionHash
+      }
+
+      switch (eventName) {
+        case EVENTS.ESCROW_AUTH:
+          record.payer = addr(args.payer)
+          record.payee = addr(args.payee)
+          record.maxLockedAmount = num(args.maxLockedAmount)
+          record.maxLockSeconds = num(args.maxLockSeconds)
+          record.maxLockCounts = num(args.maxLockCounts)
+          break
+        case EVENTS.ESCROW_LOCK:
+          record.payer = addr(args.payer)
+          record.payee = addr(args.payee)
+          record.jobId = num(args.jobId)
+          record.amount = num(args.amount)
+          record.expiry = num(args.expiry)
+          record.token = addr(args.token)
+          break
+        case EVENTS.ESCROW_CLAIMED:
+          record.payee = addr(args.payee)
+          record.jobId = num(args.jobId)
+          record.token = addr(args.token)
+          record.payer = addr(args.payer)
+          record.amount = num(args.amount)
+          record.proof = args.proof?.toString()
+          break
+        case EVENTS.ESCROW_CANCELED:
+          record.payee = addr(args.payee)
+          record.jobId = num(args.jobId)
+          record.token = addr(args.token)
+          record.payer = addr(args.payer)
+          record.amount = num(args.amount)
+          break
+        case EVENTS.ESCROW_DEPOSIT:
+        case EVENTS.ESCROW_WITHDRAW:
+          record.payer = addr(args.payer)
+          record.token = addr(args.token)
+          record.amount = num(args.amount)
+          break
+        default:
+          return null
+      }
+
+      const { escrow } = await this.getDatabase()
+      if (!escrow) return null
+      const result = await escrow.create(record)
+      INDEXER_LOGGER.logMessage(
+        `[Escrow] ${eventName} indexed for tx ${event.transactionHash} on chain ${chainId}`
+      )
+      return result
+    } catch (err) {
+      INDEXER_LOGGER.log(
+        LOG_LEVELS_STR.LEVEL_ERROR,
+        `Error processing Escrow ${eventName} event: ${err.message}`,
+        true
+      )
+      return null
+    }
+  }
+}

src/components/Indexer/processors/index.ts

@@ -15,6 +15,7 @@ export * from './OrderStartedEventProcessor.js'
 export * from './NewAccessListEventProcessor.js'
 export * from './AddressAddedEventProcessor.js'
 export * from './AddressRemovedEventProcessor.js'
+export * from './EscrowEventProcessor.js'
 export * from './BaseProcessor.js'
 
 export type ProcessorConstructor = new (

src/components/core/handler/coreHandlersRegistry.ts

@@ -56,6 +56,7 @@ import {
   PersistentStorageUploadFileHandler
 } from './persistentStorage.js'
 import { GetAccessListHandler, SearchAccessListHandler } from './accessListHandler.js'
+import { EscrowEventsHandler } from './escrowHandler.js'
 
 export type HandlerRegistry = {
   handlerName: string // name of the handler
@@ -208,6 +209,10 @@ export class CoreHandlersRegistry {
       PROTOCOL_COMMANDS.SEARCH_ACCESS_LIST,
       new SearchAccessListHandler(node)
     )
+    this.registerCoreHandler(
+      PROTOCOL_COMMANDS.GET_ESCROW_EVENTS,
+      new EscrowEventsHandler(node)
+    )
   }
 
   public static getInstance(

src/components/core/handler/escrowHandler.ts

@@ -0,0 +1,61 @@
+import { CommandHandler } from './handler.js'
+import { GetEscrowEventsCommand } from '../../../@types/commands.js'
+import { P2PCommandResponse } from '../../../@types/OceanNode.js'
+import { Readable } from 'stream'
+import {
+  ValidateParams,
+  validateCommandParameters
+} from '../../httpRoutes/validateCommands.js'
+import { CORE_LOGGER } from '../../../utils/logging/common.js'
+
+export class EscrowEventsHandler extends CommandHandler {
+  validate(command: GetEscrowEventsCommand): ValidateParams {
+    return validateCommandParameters(command, [])
+  }
+
+  async handle(task: GetEscrowEventsCommand): Promise<P2PCommandResponse> {
+    const validationResponse = await this.verifyParamsAndRateLimits(task)
+    if (this.shouldDenyTaskHandling(validationResponse)) {
+      return validationResponse
+    }
+    try {
+      const database = await this.getOceanNode().getDatabase()
+      if (!database || !database.escrow) {
+        CORE_LOGGER.error('Escrow database is not available')
+        return {
+          stream: null,
+          status: { httpStatus: 503, error: 'Escrow database is not available' }
+        }
+      }
+
+      const filters: Record<string, any> = {
+        chainId: task.chainId,
+        eventType: task.eventType,
+        payer: task.payer ? task.payer.toLowerCase() : undefined,
+        payee: task.payee ? task.payee.toLowerCase() : undefined,
+        token: task.token ? task.token.toLowerCase() : undefined,
+        jobId: task.jobId,
+        txHash: task.txId
+      }
+
+      let result = await database.escrow.search(
+        filters,
+        task.maxResultsPerPage,
+        task.pageNumber
+      )
+      if (!result) {
+        result = []
+      }
+      return {
+        stream: Readable.from(JSON.stringify(result)),
+        status: { httpStatus: 200 }
+      }
+    } catch (error) {
+      CORE_LOGGER.error(`Error in EscrowEventsHandler: ${error.message}`)
+      return {
+        stream: null,
+        status: { httpStatus: 500, error: 'Unknown error: ' + error.message }
+      }
+    }
+  }
+}

src/components/database/BaseDatabase.ts

@@ -1,6 +1,7 @@
 import { Schema } from '.'
 import { OceanNodeDBConfig } from '../../@types'
 import { AccessListUser } from '../../@types/AccessList.js'
+import { EscrowEvent } from '../../@types/Escrow.js'
 import { GENERIC_EMOJIS, LOG_LEVELS_STR } from '../../utils/logging/Logger.js'
 import { DATABASE_LOGGER } from '../../utils/logging/common.js'
 import { ElasticsearchSchema } from './ElasticSchemas.js'
@@ -152,6 +153,28 @@ export abstract class AbstractOrderDatabase {
   abstract delete(orderId: string): Promise<any>
 }
 
+export abstract class AbstractEscrowDatabase {
+  protected config: OceanNodeDBConfig
+  protected schema: Schema
+
+  constructor(config: OceanNodeDBConfig, schema: Schema) {
+    this.config = config
+    this.schema = schema
+  }
+
+  abstract create(event: EscrowEvent): Promise<any>
+
+  abstract retrieve(id: string): Promise<Record<string, any> | null>
+
+  abstract search(
+    filters: Record<string, any>,
+    maxResultsPerPage?: number,
+    pageNumber?: number
+  ): Promise<Record<string, any>[] | null>
+
+  abstract delete(id: string): Promise<any>
+}
+
 export abstract class AbstractDdoDatabase {
   protected config: OceanNodeDBConfig
   protected schemas: Schema[]

src/components/database/DatabaseFactory.ts

@@ -3,6 +3,7 @@ import {
   AbstractAccessListDatabase,
   AbstractDdoDatabase,
   AbstractDdoStateDatabase,
+  AbstractEscrowDatabase,
   AbstractIndexerDatabase,
   AbstractLogDatabase,
   AbstractOrderDatabase
@@ -11,6 +12,7 @@ import {
   ElasticsearchAccessListDatabase,
   ElasticsearchDdoDatabase,
   ElasticsearchDdoStateDatabase,
+  ElasticsearchEscrowDatabase,
   ElasticsearchIndexerDatabase,
   ElasticsearchLogDatabase,
   ElasticsearchOrderDatabase
@@ -20,6 +22,7 @@ import {
   TypesenseAccessListDatabase,
   TypesenseDdoDatabase,
   TypesenseDdoStateDatabase,
+  TypesenseEscrowDatabase,
   TypesenseIndexerDatabase,
   TypesenseLogDatabase,
   TypesenseOrderDatabase
@@ -50,7 +53,9 @@ export class DatabaseFactory {
       ddoStateQuery: () => new ElasticSearchDdoStateQuery(),
       metadataQuery: () => new ElasticSearchMetadataQuery(),
       accessList: (config: OceanNodeDBConfig) =>
-        new ElasticsearchAccessListDatabase(config)
+        new ElasticsearchAccessListDatabase(config),
+      escrow: (config: OceanNodeDBConfig) =>
+        new ElasticsearchEscrowDatabase(config, elasticSchemas.escrowSchema)
     },
     typesense: {
       ddo: (config: OceanNodeDBConfig) =>
@@ -66,7 +71,9 @@ export class DatabaseFactory {
       ddoStateQuery: () => new TypesenseDdoStateQuery(),
       metadataQuery: () => new TypesenseMetadataQuery(),
       accessList: (config: OceanNodeDBConfig) =>
-        new TypesenseAccessListDatabase(config, typesenseSchemas.accessListSchema)
+        new TypesenseAccessListDatabase(config, typesenseSchemas.accessListSchema),
+      escrow: (config: OceanNodeDBConfig) =>
+        new TypesenseEscrowDatabase(config, typesenseSchemas.escrowSchema)
     }
   }
 
@@ -142,4 +149,10 @@ export class DatabaseFactory {
   ): Promise<AbstractAccessListDatabase> {
     return this.createDatabase('accessList', config)
   }
+
+  static createEscrowDatabase(
+    config: OceanNodeDBConfig
+  ): Promise<AbstractEscrowDatabase> {
+    return this.createDatabase('escrow', config)
+  }
 }

src/components/database/ElasticSchemas.ts

@@ -22,6 +22,7 @@ export type ElasticsearchSchemas = {
   orderSchema: ElasticsearchSchema
   ddoStateSchema: ElasticsearchSchema
   accessListSchema: ElasticsearchSchema
+  escrowSchema: ElasticsearchSchema
 }
 
 // "op_ddo_short" is a node-side index for deprecated DDOs (state !== 0).
@@ -143,5 +144,30 @@ export const elasticSchemas: ElasticsearchSchemas = {
         }
       }
     }
+  },
+  escrowSchema: {
+    index: 'escrow',
+    body: {
+      mappings: {
+        properties: {
+          id: { type: 'keyword' },
+          eventType: { type: 'keyword' },
+          chainId: { type: 'long' },
+          contract: { type: 'keyword' },
+          block: { type: 'long' },
+          txHash: { type: 'keyword' },
+          payer: { type: 'keyword' },
+          payee: { type: 'keyword' },
+          token: { type: 'keyword' },
+          jobId: { type: 'keyword' },
+          amount: { type: 'text' },
+          expiry: { type: 'text' },
+          proof: { type: 'text' },
+          maxLockedAmount: { type: 'text' },
+          maxLockSeconds: { type: 'text' },
+          maxLockCounts: { type: 'text' }
+        }
+      }
+    }
   }
 }