Implement Event Filtering and Queue Routing

Problem Statement

Design and implement an in-memory event routing system, EventRouter, modeled after notification systems like Amazon SNS. Subscribers register interest in events using a destination queue identifier and a filter policy. When an event is published, the router evaluates the event against all registered policies and returns a list of target queue_ids in the order they were subscribed.

Requirements

Implement the EventRouter class with the following methods:

1. subscribe(queue_id: str, filter_policy: dict) -> None:

   • Registers a queue with a given filter policy.
   • If the queue_id was already registered, update its filter policy without changing its position in subscription order.

2. route(event: dict) -> list[str]:

   • Evaluates the event against all active subscriptions.
   • Returns the queue_ids of all matching subscriptions, preserving original subscription order.

Filter Policy Specification

A filter policy is a dictionary where keys are field paths and values are conditions. For an event to match a policy, all field conditions in the policy must be satisfied (logical AND).

• Field Paths: Keys can use dot-notation to access nested JSON fields (e.g., "user.address.city"). If a field path does not exist in the event, the condition fails.
• Conditions: A condition on a field can be one of:
  • Exact Value List (logical OR): A list of primitive values (str, int, float, bool). The field matches if its value equals any item in the list.
  • Numeric Comparison: A dict with one or more operators from "$gt", "$gte", "$lt", "$lte". The event value must be numeric (int or float) and satisfy all specified operators for that field.
  • Prefix Match: A dict {"$prefix": "value"}. The event value must be a string starting with the prefix.

If there is a type mismatch during comparison (e.g., performing a numeric comparison on a string value), the condition evaluates to False.

Example

router = EventRouter()

# Queue 1: Matches orders in US or CA with total >= 100
router.subscribe("queue-1", {
    "customer.country": ["US", "CA"],
    "order.total": {"$gte": 100}
})

# Queue 2: Matches payment events starting with 'txn_'
router.subscribe("queue-2", {
    "event_id": {"$prefix": "txn_"}
})

event_1 = {
    "event_id": "txn_98721",
    "customer": {"country": "US"},
    "order": {"total": 150.50}
}
print(router.route(event_1))  # Output: ["queue-1", "queue-2"]

event_2 = {
    "event_id": "audit_001",
    "customer": {"country": "UK"},
    "order": {"total": 200}
}
print(router.route(event_2))  # Output: []

Constraints

• Number of subscriptions: N <= 1,000
• Number of fields in an event: <= 50
• Max nesting depth of an event: <= 5
• String lengths: <= 100 characters
• Types are standard JSON: dict, list, str, int, float, bool, None