Feat/673 add events filters to the postman (#683)

* feat: add event filters options to the postman * fix: update package lock file * fix: update postman params and event filtering * fix: update filters format * fix: update .env.sample * fix: clean return condition in utility functions regarding events filtering * fix: change calldata function interface for testing
2026-01-09 04:08:01 -05:00 · 2025-02-17 11:41:26 +01:00
parent d50e8141b3
commit 12a5c33c82
13 changed files with 561 additions and 45 deletions
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -266,8 +266,6 @@ importers:
        specifier: 17.7.2
        version: 17.7.2

-  contracts/lib/forge-std: {}
-
  e2e:
    devDependencies:
      '@jest/globals':
@@ -367,6 +365,9 @@ importers:
      ethers:
        specifier: 6.13.4
        version: 6.13.4(bufferutil@4.0.8)(utf-8-validate@5.0.10)
+      filtrex:
+        specifier: 3.1.0
+        version: 3.1.0
      pg:
        specifier: 8.13.1
        version: 8.13.1
@@ -1207,10 +1208,6 @@ packages:
    peerDependencies:
      '@babel/core': ^7.0.0-0

-  '@babel/runtime@7.25.7':
-    resolution: {integrity: sha512-FjoyLe754PMiYsFaN5C94ttGiOmBNYTf6pLr4xXHAT5uctHb092PBszndLDR5XA/jghQvn4n7JMHl7dmTgbm9w==}
-    engines: {node: '>=6.9.0'}
-
  '@babel/runtime@7.26.0':
    resolution: {integrity: sha512-FDSOghenHTiToteC/QRlv2q3DhPZ/oOXTBoirfWNx1Cx3TMVcGWQtMMmQcSvb/JjpNeGzx8Pq/b4fKEJuWm1sw==}
    engines: {node: '>=6.9.0'}
@@ -5397,6 +5394,9 @@ packages:
    resolution: {integrity: sha512-8rXg1ZnX7xzy2NGDVkBVaAy+lSlPNwad13BtgSlLuxfIslyt5Vg64U7tFcCt4WS1R0hvtnQybT/IyCkGZ3DpXQ==}
    engines: {node: '>=0.10.0'}

+  filtrex@3.1.0:
+    resolution: {integrity: sha512-mHzZ2wUISETF1OaEcNRiGz1ljuIV8c/C9td9qyAZ+wTwigkAk5RO9YrCxQKk5H9v7joDRFIBik9U5RTK9eXZ/A==}
+
  finalhandler@1.1.2:
    resolution: {integrity: sha512-aAWcW57uxVNrQZqFXjITpW3sIUQmHGG3qSb9mUah9MgMC4NeWhNOlNjXEYq3HjRAvL6arUviZGGJsBg6z0zsWA==}
    engines: {node: '>= 0.8'}
@@ -10737,10 +10737,6 @@ snapshots:
      pirates: 4.0.6
      source-map-support: 0.5.21

-  '@babel/runtime@7.25.7':
-    dependencies:
-      regenerator-runtime: 0.14.1
-
  '@babel/runtime@7.26.0':
    dependencies:
      regenerator-runtime: 0.14.1
@@ -11933,8 +11929,8 @@ snapshots:
    dependencies:
      '@ethereumjs/tx': 4.2.0
      '@metamask/superstruct': 3.1.0
-      '@noble/hashes': 1.5.0
-      '@scure/base': 1.1.9
+      '@noble/hashes': 1.6.1
+      '@scure/base': 1.2.1
      '@types/debug': 4.1.12
      debug: 4.3.7(supports-color@8.1.1)
      pony-cause: 2.1.11
@@ -15656,7 +15652,7 @@ snapshots:

  date-fns@2.30.0:
    dependencies:
-      '@babel/runtime': 7.25.7
+      '@babel/runtime': 7.26.0

  date-fns@4.1.0: {}

@@ -16819,6 +16815,8 @@ snapshots:

  filter-obj@1.1.0: {}

+  filtrex@3.1.0: {}
+
  finalhandler@1.1.2:
    dependencies:
      debug: 2.6.9
@@ -17534,11 +17532,11 @@ snapshots:

  i18next-browser-languagedetector@7.1.0:
    dependencies:
-      '@babel/runtime': 7.25.7
+      '@babel/runtime': 7.26.0

  i18next@23.11.5:
    dependencies:
-      '@babel/runtime': 7.25.7
+      '@babel/runtime': 7.26.0

  iconv-lite@0.4.24:
    dependencies:
@@ -20053,7 +20051,7 @@ snapshots:

  regenerator-transform@0.15.2:
    dependencies:
-      '@babel/runtime': 7.25.7
+      '@babel/runtime': 7.26.0

  regexp.prototype.flags@1.5.3:
    dependencies:
--- a/postman/.env.sample
+++ b/postman/.env.sample
@@ -41,4 +41,15 @@ POSTGRES_DB=postman_db
 DB_CLEANER_ENABLED=false
 DB_CLEANING_INTERVAL=10000
 DB_DAYS_BEFORE_NOW_TO_DELETE=1
-ENABLE_LINEA_ESTIMATE_GAS=false
+ENABLE_LINEA_ESTIMATE_GAS=false
+
+# Optional event filter params
+L1_EVENT_FILTER_FROM_ADDRESS=<FROM_ADDRESS>
+L1_EVENT_FILTER_TO_ADDRESS=<TO_ADDRESS>
+L1_EVENT_FILTER_CALLDATA=<criteria>
+L1_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE=<FUNCTION_INTERFACE>
+
+L2_EVENT_FILTER_FROM_ADDRESS=<FROM_ADDRESS>
+L2_EVENT_FILTER_TO_ADDRESS=<TO_ADDRESS>
+L2_EVENT_FILTER_CALLDATA=<criteria>
+L2_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE=<FUNCTION_INTERFACE>
--- a/postman/README.md
+++ b/postman/README.md
@@ -28,6 +28,16 @@ All messages are stored in a configurable Postgres DB.
 - `L1_LISTENER_BLOCK_CONFIRMATION`: Required block confirmations
 - `L1_MAX_BLOCKS_TO_FETCH_LOGS`: Maximum blocks to fetch in one request
 - `L1_MAX_GAS_FEE_ENFORCED`: Enable/disable gas fee enforcement
+- `L1_EVENT_FILTER_FROM_ADDRESS`: Filter events using a from address
+- `L1_EVENT_FILTER_TO_ADDRESS`: Filter events using a to address
+- `L1_EVENT_FILTER_CALLDATA`: MessageSent event calldata filtering criteria expression. See [Filtrex repo](https://github.com/joewalnes/filtrex/tree/master).
+    <br>
+    You can filter by the calldata field:
+    <br>
+    
+    Example:
+    `calldata.funcSignature == "0x6463fb2a" and calldata.params.messageNumber == 85804`,
+- `L1_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE`: Calldata data function interface following this format: `"function transfer(address to, uint256 amount)"`. Make sure you specify parameters names in order to use syntax like `calldata.params.messageNumber`.

 #### L2 Configuration
 - `L2_RPC_URL`: Linea node RPC endpoint
@@ -39,6 +49,16 @@ All messages are stored in a configurable Postgres DB.
 - `L2_MAX_BLOCKS_TO_FETCH_LOGS`: Maximum blocks to fetch in one request
 - `L2_MAX_GAS_FEE_ENFORCED`: Enable/disable gas fee enforcement
 - `L2_MESSAGE_TREE_DEPTH`: Depth of the message Merkle tree
+- `L2_EVENT_FILTER_FROM_ADDRESS`: Filter events using a from address
+- `L2_EVENT_FILTER_TO_ADDRESS`: Filter events using a to address
+- `L2_EVENT_FILTER_CALLDATA`: MessageSent event calldata filtering criteria expression. See [Filtrex repo](https://github.com/joewalnes/filtrex/tree/master).
+    <br>
+    You can filter by the calldata field:
+    <br>
+    
+    Example:
+    `calldata.funcSignature == "0x6463fb2a" and calldata.params.messageNumber == 85804`,
+- `L2_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE`: Calldata data function interface following this format: `"function transfer(address to, uint256 amount)"`. Make sure you specify parameters names in order to use syntax like `calldata.params.messageNumber`.

 #### Message Processing
 - `MESSAGE_SUBMISSION_TIMEOUT`: Timeout for message submission (ms)
@@ -80,7 +100,7 @@ All messages are stored in a configurable Postgres DB.

 From the root folder, run the following command:
 ```bash
-make fresh-start-all
+make start-env-with-tracing-v2
 ```

 Stop the postman docker container manually.
--- a/postman/package.json
+++ b/postman/package.json
@@ -24,6 +24,7 @@
    "class-validator": "0.14.1",
    "dotenv": "16.4.5",
    "ethers": "6.13.4",
+    "filtrex": "3.1.0",
    "pg": "8.13.1",
    "typeorm": "0.3.20",
    "typeorm-naming-strategies": "4.1.0",
--- a/postman/scripts/runPostman.ts
+++ b/postman/scripts/runPostman.ts
@@ -25,6 +25,24 @@ async function main() {
        ...(parseInt(process.env.L1_LISTENER_BLOCK_CONFIRMATION ?? "") >= 0
          ? { blockConfirmation: parseInt(process.env.L1_LISTENER_BLOCK_CONFIRMATION ?? "") }
          : {}),
+        ...(process.env.L1_EVENT_FILTER_FROM_ADDRESS ||
+        process.env.L1_EVENT_FILTER_TO_ADDRESS ||
+        (process.env.L1_EVENT_FILTER_CALLDATA && process.env.L1_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE)
+          ? {
+              eventFilters: {
+                fromAddressFilter: process.env.L1_EVENT_FILTER_FROM_ADDRESS,
+                toAddressFilter: process.env.L1_EVENT_FILTER_TO_ADDRESS,
+                ...(process.env.L1_EVENT_FILTER_CALLDATA && process.env.L1_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE
+                  ? {
+                      calldataFilter: {
+                        criteriaExpression: process.env.L1_EVENT_FILTER_CALLDATA,
+                        calldataFunctionInterface: process.env.L1_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE,
+                      },
+                    }
+                  : {}),
+              },
+            }
+          : {}),
      },
      claiming: {
        signerPrivateKey: process.env.L1_SIGNER_PRIVATE_KEY ?? "",
@@ -65,6 +83,24 @@ async function main() {
        ...(parseInt(process.env.L2_LISTENER_BLOCK_CONFIRMATION ?? "") >= 0
          ? { blockConfirmation: parseInt(process.env.L2_LISTENER_BLOCK_CONFIRMATION ?? "") }
          : {}),
+        ...(process.env.L2_EVENT_FILTER_FROM_ADDRESS ||
+        process.env.L2_EVENT_FILTER_TO_ADDRESS ||
+        (process.env.L2_EVENT_FILTER_CALLDATA && process.env.L2_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE)
+          ? {
+              eventFilters: {
+                fromAddressFilter: process.env.L2_EVENT_FILTER_FROM_ADDRESS,
+                toAddressFilter: process.env.L2_EVENT_FILTER_TO_ADDRESS,
+                ...(process.env.L2_EVENT_FILTER_CALLDATA && process.env.L2_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE
+                  ? {
+                      calldataFilter: {
+                        criteriaExpression: process.env.L2_EVENT_FILTER_CALLDATA,
+                        calldataFunctionInterface: process.env.L2_EVENT_FILTER_CALLDATA_FUNCTION_INTERFACE,
+                      },
+                    }
+                  : {}),
+              },
+            }
+          : {}),
      },
      claiming: {
        signerPrivateKey: process.env.L2_SIGNER_PRIVATE_KEY ?? "",
--- a/postman/src/application/postman/app/PostmanServiceClient.ts
+++ b/postman/src/application/postman/app/PostmanServiceClient.ts
@@ -120,6 +120,7 @@ export class PostmanServiceClient {
        blockConfirmation: config.l1Config.listener.blockConfirmation,
        isEOAEnabled: config.l1Config.isEOAEnabled,
        isCalldataEnabled: config.l1Config.isCalldataEnabled,
+        eventFilters: config.l1Config.listener.eventFilters,
      },
      new WinstonLogger(`L1${MessageSentEventProcessor.name}`, config.loggerOptions),
    );
@@ -245,6 +246,7 @@ export class PostmanServiceClient {
        blockConfirmation: config.l2Config.listener.blockConfirmation,
        isEOAEnabled: config.l2Config.isEOAEnabled,
        isCalldataEnabled: config.l2Config.isCalldataEnabled,
+        eventFilters: config.l2Config.listener.eventFilters,
      },
      new WinstonLogger(`L2${MessageSentEventProcessor.name}`, config.loggerOptions),
    );
--- a/postman/src/application/postman/app/tests/PostmanServiceClient.test.ts
+++ b/postman/src/application/postman/app/tests/PostmanServiceClient.test.ts
@@ -135,6 +135,36 @@ describe("PostmanServiceClient", () => {
        new Error("Something went wrong when trying to generate Wallet. Please check your private key."),
      );
    });
+
+    it("should throw an error when events filters are not valid", () => {
+      const postmanServiceClientOptionsWithInvalidPrivateKey: PostmanOptions = {
+        ...postmanServiceClientOptions,
+        l1Options: {
+          ...postmanServiceClientOptions.l1Options,
+          listener: {
+            ...postmanServiceClientOptions.l1Options.listener,
+            eventFilters: {
+              fromAddressFilter: "0x",
+            },
+          },
+          claiming: {
+            ...postmanServiceClientOptions.l1Options.claiming,
+            signerPrivateKey: "0x",
+          },
+        },
+        l2Options: {
+          ...postmanServiceClientOptions.l2Options,
+          claiming: {
+            ...postmanServiceClientOptions.l2Options.claiming,
+            signerPrivateKey: "0x",
+          },
+        },
+      };
+
+      expect(() => new PostmanServiceClient(postmanServiceClientOptionsWithInvalidPrivateKey)).toThrow(
+        new Error("Invalid fromAddressFilter: 0x"),
+      );
+    });
  });

  describe("connectDatabase", () => {
--- a/postman/src/application/postman/app/config/tests/utils.test.ts
+++ b/postman/src/application/postman/app/config/tests/utils.test.ts
@@ -1,5 +1,5 @@
 import { describe } from "@jest/globals";
-import { getConfig } from "../utils";
+import { getConfig, validateEventsFiltersConfig } from "../utils";
 import {
  TEST_ADDRESS_1,
  TEST_ADDRESS_2,
@@ -235,4 +235,61 @@ describe("Config utils", () => {
      });
    });
  });
+
+  describe("validateEventsFiltersConfig", () => {
+    it("should throw an error when the from address event filter is not valid", () => {
+      expect(() =>
+        validateEventsFiltersConfig({
+          fromAddressFilter: "0x123",
+        }),
+      ).toThrow("Invalid fromAddressFilter: 0x123");
+    });
+
+    it("should throw an error when the to address event filter is not valid", () => {
+      expect(() =>
+        validateEventsFiltersConfig({
+          toAddressFilter: "0x123",
+        }),
+      ).toThrow("Invalid toAddressFilter: 0x123");
+    });
+
+    it("should not throw an error when filters are valid", () => {
+      expect(() =>
+        validateEventsFiltersConfig({
+          fromAddressFilter: "0xc59d8de7f984AbC4913f0177bfb7BBdaFaC41fA6",
+          toAddressFilter: "0xc59d8de7f984AbC4913f0177bfb7BBdaFaC41fA6",
+          calldataFilter: {
+            criteriaExpression: `calldata.funcSignature == "0x26dfbc20" and calldata.amount > 0`,
+            calldataFunctionInterface: "function receiveFromOtherLayer(address recipient, uint256 amount)",
+          },
+        }),
+      ).not.toThrow();
+    });
+
+    it("should throw an error when calldataFilter filter expression is invalid", () => {
+      expect(() =>
+        validateEventsFiltersConfig({
+          fromAddressFilter: "0xc59d8de7f984AbC4913f0177bfb7BBdaFaC41fA6",
+          toAddressFilter: "0xc59d8de7f984AbC4913f0177bfb7BBdaFaC41fA6",
+          calldataFilter: {
+            criteriaExpression: `calldata.funcSignature == "0x26dfbc20" and calldata.amount = 0`,
+            calldataFunctionInterface: "function receiveFromOtherLayer(address recipient, uint256 amount)",
+          },
+        }),
+      ).toThrow('Invalid calldataFilter expression: calldata.funcSignature == "0x26dfbc20" and calldata.amount = 0');
+    });
+
+    it("should throw an error when calldataFunctionInterface is invalid", () => {
+      expect(() =>
+        validateEventsFiltersConfig({
+          fromAddressFilter: "0xc59d8de7f984AbC4913f0177bfb7BBdaFaC41fA6",
+          toAddressFilter: "0xc59d8de7f984AbC4913f0177bfb7BBdaFaC41fA6",
+          calldataFilter: {
+            criteriaExpression: `calldata.funcSignature == "0x26dfbc20" and calldata.amount > 0`,
+            calldataFunctionInterface: "function receiveFromOtherLayer(address recipient uint256 amount)",
+          },
+        }),
+      ).toThrow("Invalid calldataFunctionInterface: function receiveFromOtherLayer(address recipient uint256 amount)");
+    });
+  });
 });
--- a/postman/src/application/postman/app/config/config.ts
+++ b/postman/src/application/postman/app/config/config.ts
@@ -99,6 +99,15 @@ export type ListenerOptions = {
  blockConfirmation?: number;
  maxFetchMessagesFromDb?: number;
  maxBlocksToFetchLogs?: number;
+  eventFilters?: {
+    fromAddressFilter?: string;
+    toAddressFilter?: string;
+    calldataFilter?: {
+      criteriaExpression: string;
+      calldataFunctionInterface: string;
+    };
+  };
 };

-export type ListenerConfig = Required<ListenerOptions>;
+export type ListenerConfig = Required<Omit<ListenerOptions, "eventFilters">> &
+  Partial<Pick<ListenerOptions, "eventFilters">>;
--- a/postman/src/application/postman/app/config/utils.ts
+++ b/postman/src/application/postman/app/config/utils.ts
@@ -1,3 +1,5 @@
+import { Interface, isAddress } from "ethers";
+import { compileExpression, useDotAccessOperator } from "filtrex";
 import {
  DEFAULT_CALLDATA_ENABLED,
  DEFAULT_EOA_ENABLED,
@@ -17,7 +19,7 @@ import {
  DEFAULT_PROFIT_MARGIN,
  DEFAULT_RETRY_DELAY_IN_SECONDS,
 } from "../../../../core/constants";
-import { PostmanConfig, PostmanOptions } from "./config";
+import { ListenerConfig, PostmanConfig, PostmanOptions } from "./config";

 /**
 * @notice Generates the configuration for the Postman service based on provided options.
@@ -36,6 +38,14 @@ export function getConfig(postmanOptions: PostmanOptions): PostmanConfig {
    loggerOptions,
  } = postmanOptions;

+  if (l1Options.listener.eventFilters) {
+    validateEventsFiltersConfig(l1Options.listener.eventFilters);
+  }
+
+  if (l2Options.listener.eventFilters) {
+    validateEventsFiltersConfig(l2Options.listener.eventFilters);
+  }
+
  return {
    l1Config: {
      rpcUrl: l1Options.rpcUrl,
@@ -48,6 +58,7 @@ export function getConfig(postmanOptions: PostmanOptions): PostmanConfig {
        maxBlocksToFetchLogs: l1Options.listener.maxBlocksToFetchLogs ?? DEFAULT_MAX_BLOCKS_TO_FETCH_LOGS,
        initialFromBlock: l1Options.listener.initialFromBlock ?? DEFAULT_INITIAL_FROM_BLOCK,
        blockConfirmation: l1Options.listener.blockConfirmation ?? DEFAULT_LISTENER_BLOCK_CONFIRMATIONS,
+        ...(l1Options.listener.eventFilters ? { eventFilters: l1Options.listener.eventFilters } : {}),
      },
      claiming: {
        signerPrivateKey: l1Options.claiming.signerPrivateKey,
@@ -77,6 +88,7 @@ export function getConfig(postmanOptions: PostmanOptions): PostmanConfig {
        maxBlocksToFetchLogs: l2Options.listener.maxBlocksToFetchLogs ?? DEFAULT_MAX_BLOCKS_TO_FETCH_LOGS,
        initialFromBlock: l2Options.listener.initialFromBlock ?? DEFAULT_INITIAL_FROM_BLOCK,
        blockConfirmation: l2Options.listener.blockConfirmation ?? DEFAULT_LISTENER_BLOCK_CONFIRMATIONS,
+        ...(l2Options.listener.eventFilters ? { eventFilters: l2Options.listener.eventFilters } : {}),
      },
      claiming: {
        signerPrivateKey: l2Options.claiming.signerPrivateKey,
@@ -104,3 +116,46 @@ export function getConfig(postmanOptions: PostmanOptions): PostmanConfig {
    loggerOptions,
  };
 }
+
+export function validateEventsFiltersConfig(eventFilters: ListenerConfig["eventFilters"]): void {
+  if (eventFilters?.fromAddressFilter && !isAddress(eventFilters.fromAddressFilter)) {
+    throw new Error(`Invalid fromAddressFilter: ${eventFilters.fromAddressFilter}`);
+  }
+
+  if (eventFilters?.toAddressFilter && !isAddress(eventFilters.toAddressFilter)) {
+    throw new Error(`Invalid toAddressFilter: ${eventFilters.toAddressFilter}`);
+  }
+
+  if (
+    eventFilters?.calldataFilter?.criteriaExpression &&
+    !isValidFiltrexExpression(eventFilters?.calldataFilter?.criteriaExpression)
+  ) {
+    throw new Error(`Invalid calldataFilter expression: ${eventFilters.calldataFilter.criteriaExpression}`);
+  }
+
+  if (
+    eventFilters?.calldataFilter?.calldataFunctionInterface &&
+    !isFunctionInterfaceValid(eventFilters?.calldataFilter?.calldataFunctionInterface)
+  ) {
+    throw new Error(`Invalid calldataFunctionInterface: ${eventFilters?.calldataFilter?.calldataFunctionInterface}`);
+  }
+}
+
+export function isFunctionInterfaceValid(functionInterface: string): boolean {
+  try {
+    const i = new Interface([functionInterface]);
+
+    return i.fragments.length !== 0;
+  } catch (error) {
+    return false;
+  }
+}
+
+export function isValidFiltrexExpression(expression: string): boolean {
+  try {
+    compileExpression(expression, { customProp: useDotAccessOperator });
+    return true;
+  } catch (error) {
+    return false;
+  }
+}
--- a/postman/src/core/services/processors/IMessageSentEventProcessor.ts
+++ b/postman/src/core/services/processors/IMessageSentEventProcessor.ts
@@ -12,4 +12,12 @@ export type MessageSentEventProcessorConfig = {
  blockConfirmation: number;
  isEOAEnabled: boolean;
  isCalldataEnabled: boolean;
+  eventFilters?: {
+    fromAddressFilter?: string;
+    toAddressFilter?: string;
+    calldataFilter?: {
+      criteriaExpression: string;
+      calldataFunctionInterface: string;
+    };
+  };
 };
--- a/postman/src/services/processors/MessageSentEventProcessor.ts
+++ b/postman/src/services/processors/MessageSentEventProcessor.ts
@@ -1,12 +1,15 @@
 import {
  Block,
  ContractTransactionResponse,
+  dataSlice,
+  Interface,
  JsonRpcProvider,
  TransactionReceipt,
  TransactionRequest,
  TransactionResponse,
 } from "ethers";
-import { serialize, isEmptyBytes } from "@consensys/linea-sdk";
+import { compileExpression, useDotAccessOperator } from "filtrex";
+import { serialize, isEmptyBytes, MessageSent } from "@consensys/linea-sdk";
 import { ILineaRollupLogClient } from "../../core/clients/blockchain/ethereum/ILineaRollupLogClient";
 import { IProvider } from "../../core/clients/blockchain/IProvider";
 import { MessageFactory } from "../../core/entities/MessageFactory";
@@ -41,7 +44,7 @@ export class MessageSentEventProcessor implements IMessageSentEventProcessor {
      TransactionResponse,
      JsonRpcProvider
    >,
-    private readonly config: MessageSentEventProcessorConfig,
+    protected readonly config: MessageSentEventProcessorConfig,
    private readonly logger: ILogger,
  ) {
    this.maxBlocksToFetchLogs = Math.max(config.maxBlocksToFetchLogs, 0);
@@ -80,6 +83,10 @@ export class MessageSentEventProcessor implements IMessageSentEventProcessor {
    this.logger.info("Getting events fromBlock=%s toBlock=%s", fromBlock, toBlock);

    const events = await this.logClient.getMessageSentEvents({
+      filters: {
+        from: this.config.eventFilters?.fromAddressFilter,
+        to: this.config.eventFilters?.toAddressFilter,
+      },
      fromBlock,
      toBlock,
      fromBlockLogIndex,
@@ -88,7 +95,11 @@ export class MessageSentEventProcessor implements IMessageSentEventProcessor {
    this.logger.info("Number of fetched MessageSent events: %s", events.length);

    for (const event of events) {
-      const shouldBeProcessed = this.shouldProcessMessage(event.calldata, event.messageHash);
+      const shouldBeProcessed = this.shouldProcessMessage(
+        event,
+        event.messageHash,
+        this.config.eventFilters?.calldataFilter,
+      );
      const messageStatusToInsert = shouldBeProcessed ? MessageStatus.SENT : MessageStatus.EXCLUDED;

      const message = MessageFactory.createMessage({
@@ -109,25 +120,106 @@ export class MessageSentEventProcessor implements IMessageSentEventProcessor {
  /**
   * Determines whether a message should be processed based on its calldata and the configuration.
   *
-   * @param {string} messageCalldata - The calldata of the message.
+   * @param {string} event - The message event.
   * @param {string} messageHash - The hash of the message.
   * @returns {boolean} `true` if the message should be processed, `false` otherwise.
   */
-  private shouldProcessMessage(messageCalldata: string, messageHash: string): boolean {
-    if (isEmptyBytes(messageCalldata)) {
-      if (this.config.isEOAEnabled) {
-        return true;
-      }
+  protected shouldProcessMessage(
+    event: MessageSent,
+    messageHash: string,
+    filters?: {
+      criteriaExpression: string;
+      calldataFunctionInterface: string;
+    },
+  ): boolean {
+    const hasEmptyCalldata = isEmptyBytes(event.calldata);
+    let basicProcess = false;
+
+    if (hasEmptyCalldata) {
+      basicProcess = this.config.isEOAEnabled;
    } else {
-      if (this.config.isCalldataEnabled) {
-        return true;
-      }
+      basicProcess = this.config.isCalldataEnabled;
    }

-    this.logger.debug(
-      "Message has been excluded because target address is not an EOA or calldata is not empty: messageHash=%s",
-      messageHash,
-    );
-    return false;
+    if (!basicProcess) {
+      this.logger.debug(
+        "Message has been excluded because target address is not an EOA or calldata is not empty: messageHash=%s",
+        messageHash,
+      );
+      return false;
+    }
+
+    if (!hasEmptyCalldata && this.config.isCalldataEnabled && !this.isMessageMatchingCriteria(event, filters)) {
+      return false;
+    }
+
+    return true;
+  }
+
+  private isMessageMatchingCriteria(
+    event: MessageSent,
+    filters?: { criteriaExpression: string; calldataFunctionInterface: string },
+  ) {
+    if (!filters) {
+      return true;
+    }
+
+    const iface = new Interface([filters.calldataFunctionInterface]);
+    const decodedCalldata = iface.decodeFunctionData(filters.calldataFunctionInterface, event.calldata);
+
+    const context = {
+      calldata: {
+        funcSignature: dataSlice(event.calldata, 0, 4),
+        ...this.convertBigInts(decodedCalldata.toObject(true)),
+      },
+    };
+
+    const passesFilter = this.evaluateExpression(filters.criteriaExpression, context);
+
+    if (!passesFilter) {
+      this.logger.debug(
+        "Message has been excluded because it does not match the criteria: criteria=%s messageHash=%s transactionHash=%s",
+        filters.criteriaExpression,
+        event.messageHash,
+        event.transactionHash,
+      );
+      return false;
+    }
+
+    return true;
+  }
+
+  private evaluateExpression(expression: string, context: unknown): boolean {
+    try {
+      const compiledFilter = compileExpression(expression, { customProp: useDotAccessOperator });
+      const passesFilter = compiledFilter(context);
+      return passesFilter === true;
+    } catch (error) {
+      return false;
+    }
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  private convertBigInts(data: any): any {
+    if (typeof data === "bigint") {
+      return Number(data);
+    }
+
+    if (Array.isArray(data)) {
+      return data.map((item) => this.convertBigInts(item));
+    }
+
+    if (data !== null && typeof data === "object") {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      const result: Record<string, any> = {};
+      for (const key in data) {
+        if (Object.prototype.hasOwnProperty.call(data, key)) {
+          result[key] = this.convertBigInts(data[key]);
+        }
+      }
+      return result;
+    }
+
+    return data;
  }
 }
--- a/postman/src/services/processors/tests/MessageSentEventProcessor.test.ts
+++ b/postman/src/services/processors/tests/MessageSentEventProcessor.test.ts
@@ -1,23 +1,56 @@
 import { describe, it, beforeEach } from "@jest/globals";
 import { mock } from "jest-mock-extended";
 import { TestLogger } from "../../../utils/testing/helpers";
-import { Direction } from "@consensys/linea-sdk";
+import { Direction, MessageSent } from "@consensys/linea-sdk";
 import { MessageStatus } from "../../../core/enums";
 import {
+  TEST_ADDRESS_1,
+  TEST_ADDRESS_2,
  testL1NetworkConfig,
  testMessageSentEvent,
  testMessageSentEventWithCallData,
 } from "../../../utils/testing/constants";
 import { IProvider } from "../../../core/clients/blockchain/IProvider";
-import { IMessageSentEventProcessor } from "../../../core/services/processors/IMessageSentEventProcessor";
+import { MessageSentEventProcessorConfig } from "../../../core/services/processors/IMessageSentEventProcessor";
 import { MessageSentEventProcessor } from "../MessageSentEventProcessor";
 import { ILineaRollupLogClient } from "../../../core/clients/blockchain/ethereum/ILineaRollupLogClient";
 import { MessageFactory } from "../../../core/entities/MessageFactory";
-import { Block, JsonRpcProvider, TransactionReceipt, TransactionRequest, TransactionResponse } from "ethers";
+import {
+  Block,
+  ContractTransactionResponse,
+  Interface,
+  JsonRpcProvider,
+  TransactionReceipt,
+  TransactionRequest,
+  TransactionResponse,
+} from "ethers";
 import { EthereumMessageDBService } from "../../persistence/EthereumMessageDBService";
+import { IMessageDBService } from "../../../core/persistence/IMessageDBService";
+import { ILogger } from "../../../core/utils/logging/ILogger";
+import { IL2MessageServiceLogClient } from "../../../core/clients/blockchain/linea/IL2MessageServiceLogClient";
+
+class TestMessageSentEventProcessor extends MessageSentEventProcessor {
+  constructor(
+    databaseService: IMessageDBService<ContractTransactionResponse>,
+    logClient: ILineaRollupLogClient | IL2MessageServiceLogClient,
+    provider: IProvider<TransactionReceipt, Block, TransactionRequest, TransactionResponse, JsonRpcProvider>,
+    public readonly config: MessageSentEventProcessorConfig,
+    logger: ILogger,
+  ) {
+    super(databaseService, logClient, provider, config, logger);
+  }
+
+  public shouldProcessMessage(
+    message: MessageSent,
+    messageHash: string,
+    filters?: { criteriaExpression: string; calldataFunctionInterface: string },
+  ): boolean {
+    return super.shouldProcessMessage(message, messageHash, filters);
+  }
+}

 describe("TestMessageSentEventProcessor", () => {
-  let messageSentEventProcessor: IMessageSentEventProcessor;
+  let messageSentEventProcessor: TestMessageSentEventProcessor;
  const databaseService = mock<EthereumMessageDBService>();
  const l1LogClientMock = mock<ILineaRollupLogClient>();
  const provider =
@@ -25,7 +58,7 @@ describe("TestMessageSentEventProcessor", () => {
  const logger = new TestLogger(MessageSentEventProcessor.name);

  beforeEach(() => {
-    messageSentEventProcessor = new MessageSentEventProcessor(
+    messageSentEventProcessor = new TestMessageSentEventProcessor(
      databaseService,
      l1LogClientMock,
      provider,
@@ -66,7 +99,7 @@ describe("TestMessageSentEventProcessor", () => {
    });

    it("Should insert message with status as excluded into repository if the message is excluded", async () => {
-      messageSentEventProcessor = new MessageSentEventProcessor(
+      messageSentEventProcessor = new TestMessageSentEventProcessor(
        databaseService,
        l1LogClientMock,
        provider,
@@ -98,8 +131,69 @@ describe("TestMessageSentEventProcessor", () => {
      expect(messageRepositoryInsertSpy).toHaveBeenCalledWith(expectedMessageToInsert);
    });

+    it("Should insert message with status as excluded into repository if the message is excluded becuase of events filters", async () => {
+      messageSentEventProcessor = new TestMessageSentEventProcessor(
+        databaseService,
+        l1LogClientMock,
+        provider,
+        {
+          direction: Direction.L1_TO_L2,
+          maxBlocksToFetchLogs: testL1NetworkConfig.listener.maxBlocksToFetchLogs,
+          blockConfirmation: testL1NetworkConfig.listener.blockConfirmation,
+          isEOAEnabled: testL1NetworkConfig.isEOAEnabled,
+          isCalldataEnabled: true,
+          eventFilters: {
+            fromAddressFilter: TEST_ADDRESS_1,
+            toAddressFilter: TEST_ADDRESS_2,
+            calldataFilter: {
+              criteriaExpression: `calldata.funcSignature == "0x26dfbc20" and calldata.amount == 0`,
+              calldataFunctionInterface: "function receiveFromOtherLayer(address recipient, uint256 amount)",
+            },
+          },
+        },
+        logger,
+      );
+      const loggerInfoSpy = jest.spyOn(logger, "info");
+      const messageRepositoryInsertSpy = jest.spyOn(databaseService, "insertMessage");
+      jest.spyOn(provider, "getBlockNumber").mockResolvedValue(100);
+      jest.spyOn(l1LogClientMock, "getMessageSentEvents").mockResolvedValue([
+        testMessageSentEvent,
+        {
+          ...testMessageSentEvent,
+          calldata:
+            "0x26dfbc200000000000000000000000005eeea0e70ffe4f5419477056023c4b0aca01656200000000000000000000000000000000000000000000000000000000000186a0",
+        },
+      ]);
+      const expectedMessage1ToInsert = MessageFactory.createMessage({
+        ...testMessageSentEvent,
+        sentBlockNumber: testMessageSentEvent.blockNumber,
+        direction: Direction.L1_TO_L2,
+        status: MessageStatus.SENT,
+        claimNumberOfRetry: 0,
+      });
+
+      const expectedMessage2ToInsert = MessageFactory.createMessage({
+        ...{
+          ...testMessageSentEvent,
+          calldata:
+            "0x26dfbc200000000000000000000000005eeea0e70ffe4f5419477056023c4b0aca01656200000000000000000000000000000000000000000000000000000000000186a0",
+        },
+        sentBlockNumber: testMessageSentEvent.blockNumber,
+        direction: Direction.L1_TO_L2,
+        status: MessageStatus.EXCLUDED,
+        claimNumberOfRetry: 0,
+      });
+
+      await messageSentEventProcessor.process(0, 0);
+
+      expect(loggerInfoSpy).toHaveBeenCalledTimes(3);
+      expect(messageRepositoryInsertSpy).toHaveBeenCalledTimes(2);
+      expect(messageRepositoryInsertSpy).toHaveBeenNthCalledWith(1, expectedMessage1ToInsert);
+      expect(messageRepositoryInsertSpy).toHaveBeenNthCalledWith(2, expectedMessage2ToInsert);
+    });
+
    it("Should insert message with calldata with status as sent into repository if calldata is enabled", async () => {
-      messageSentEventProcessor = new MessageSentEventProcessor(
+      messageSentEventProcessor = new TestMessageSentEventProcessor(
        databaseService,
        l1LogClientMock,
        provider,
@@ -131,4 +225,107 @@ describe("TestMessageSentEventProcessor", () => {
      expect(messageRepositoryInsertSpy).toHaveBeenCalledWith(expectedMessageToInsert);
    });
  });
+
+  describe("shouldProcessMessage", () => {
+    const funcFragment = "function receiveFromOtherLayer(address recipient, uint256 amount)";
+
+    const encodedCalldata = new Interface([funcFragment]).encodeFunctionData(funcFragment, [
+      "0x5eeea0e70ffe4f5419477056023c4b0aca016562",
+      100000n,
+    ]);
+
+    it("Should return true if calldata is empty and EOA is enabled", () => {
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        testMessageSentEvent,
+        testMessageSentEvent.messageHash,
+      );
+      expect(result).toBeTruthy();
+    });
+
+    it("Should return false if calldata is empty and EOA is disabled", () => {
+      messageSentEventProcessor.config.isEOAEnabled = false;
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        testMessageSentEvent,
+        testMessageSentEvent.messageHash,
+      );
+      expect(result).toBeFalsy();
+    });
+
+    it("Should return true if calldata is not empty and calldata option is enabled", () => {
+      messageSentEventProcessor.config.isCalldataEnabled = true;
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        { ...testMessageSentEvent, calldata: "0x1111111111" },
+        testMessageSentEvent.messageHash,
+      );
+      expect(result).toBeTruthy();
+    });
+
+    it("Should return false if calldata is not empty and calldata option is disabled", () => {
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        { ...testMessageSentEvent, calldata: "0x1111111111" },
+        testMessageSentEvent.messageHash,
+      );
+      expect(result).toBeFalsy();
+    });
+
+    it("Should return false if EOA and calldata options are disabled", () => {
+      messageSentEventProcessor.config.isEOAEnabled = false;
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        { ...testMessageSentEvent, calldata: "0x1111111111" },
+        testMessageSentEvent.messageHash,
+      );
+      expect(result).toBeFalsy();
+    });
+
+    it("Should return false if event filter criteria is not correctly formatted", () => {
+      messageSentEventProcessor.config.isCalldataEnabled = true;
+
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        {
+          ...testMessageSentEvent,
+          calldata: encodedCalldata,
+        },
+        testMessageSentEvent.messageHash,
+        {
+          criteriaExpression: `calldata.funcSignature == 0x26dfbc20 and calldata.amount > 0`,
+          calldataFunctionInterface: funcFragment,
+        },
+      );
+      expect(result).toBeFalsy();
+    });
+
+    it("Should return false if event filter criteria is false", () => {
+      messageSentEventProcessor.config.isCalldataEnabled = true;
+
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        {
+          ...testMessageSentEvent,
+          calldata: encodedCalldata,
+        },
+        testMessageSentEvent.messageHash,
+        {
+          criteriaExpression: `calldata.funcSignature == "0x26dfbc20" and calldata.amount == 0`,
+          calldataFunctionInterface: funcFragment,
+        },
+      );
+      expect(result).toBeFalsy();
+    });
+
+    it("Should return true if event filter criteria is true", () => {
+      messageSentEventProcessor.config.isCalldataEnabled = true;
+
+      const result = messageSentEventProcessor.shouldProcessMessage(
+        {
+          ...testMessageSentEvent,
+          calldata: encodedCalldata,
+        },
+        testMessageSentEvent.messageHash,
+        {
+          criteriaExpression: `calldata.funcSignature == "0x26dfbc20" and calldata.amount > 0`,
+          calldataFunctionInterface: funcFragment,
+        },
+      );
+      expect(result).toBeTruthy();
+    });
+  });
 });