Design a News Subscription Processing Engine

Problem Description

Design an in-memory news subscription processing engine that receives incoming news items, matches them against registered subscriber filter predicates, and manages delivery state.

In high-frequency trading and financial market environments, news feeds must be disseminated to internal consumers with strict delivery semantics, deduplication guarantees, and low processing overhead.

Implement the NewsSubscriptionEngine class with the following capabilities:

1. register_subscriber(subscriber_id: str, topic: str, min_urgency: int) -> None

   • Registers a subscriber. If subscriber_id already exists, update its subscription rule to the new topic and min_urgency.
   • topic can be a specific string (e.g., "EQUITIES") or '*' to match all topics.
   • A subscriber matches an item if the topic matches and the item's urgency is greater than or equal to min_urgency.

2. unregister_subscriber(subscriber_id: str) -> bool

   • Removes the subscriber. Purges all pending unacknowledged deliveries for this subscriber.
   • Returns True if the subscriber was registered, or False if not found.

3. publish_item(news_id: str, topic: str, urgency: int) -> List[str]

   • Publishes an item to all matching registered subscribers.
   • Deduplication: If news_id has already been processed, the engine must ignore the publication and return an empty list [].
   • When matched, the item is appended to each matched subscriber's pending queue in the order published.
   • Returns a sorted list of subscriber_ids that matched and were queued.

4. ack_delivery(subscriber_id: str, news_id: str) -> bool

   • Marks a pending item as acknowledged by the given subscriber and removes it from their pending list.
   • Returns True if the item was in the subscriber's pending list and successfully acknowledged; otherwise False.

5. get_pending(subscriber_id: str) -> List[str]

   • Returns a list of news_ids currently pending for subscriber_id in FIFO order of publication.
   • Returns an empty list if the subscriber does not exist or has no pending items.

Class Interface

class NewsSubscriptionEngine:
    def __init__(self):
        pass

    def register_subscriber(self, subscriber_id: str, topic: str, min_urgency: int) -> None:
        pass

    def unregister_subscriber(self, subscriber_id: str) -> bool:
        pass

    def publish_item(self, news_id: str, topic: str, urgency: int) -> List[str]:
        pass

    def ack_delivery(self, subscriber_id: str, news_id: str) -> bool:
        pass

    def get_pending(self, subscriber_id: str) -> List[str]:
        pass

Example

engine = NewsSubscriptionEngine()

engine.register_subscriber("sub1", "EQUITIES", 2)
engine.register_subscriber("sub2", "*", 4)
engine.register_subscriber("sub3", "CRYPTO", 1)

# Item matching sub1 (topic EQUITIES, urgency 3 >= 2)
engine.publish_item("N1", "EQUITIES", 3)  # Returns: ["sub1"]

# Item matching sub1 and sub2 (urgency 4 >= 2 and >= 4)
engine.publish_item("N2", "EQUITIES", 5)  # Returns: ["sub1", "sub2"]

# Duplicate publication of N1 should be ignored
engine.publish_item("N1", "EQUITIES", 5)  # Returns: []

# Check pending
engine.get_pending("sub1")  # Returns: ["N1", "N2"]
engine.get_pending("sub2")  # Returns: ["N2"]

# Acknowledge delivery
engine.ack_delivery("sub1", "N1")  # Returns: True
engine.get_pending("sub1")         # Returns: ["N2"]

# Unregistering subscriber clears their pending queue
engine.unregister_subscriber("sub1")  # Returns: True
engine.get_pending("sub1")            # Returns: []

Constraints

• Number of total calls: N <= 100,000
• subscriber_id, news_id, topic are non-empty strings of alphanumeric characters and underscores, length 1 <= len(s) <= 64.
• urgency, min_urgency are integers where 1 <= urgency <= 10.
• Time complexity per operation should be O(M) or O(M log M) where M is the number of matching subscribers for publish_item, and O(1) amortized for registration, unregistration, and acknowledgment.