Deduplicate and Order Batch and Streaming Logs

Problem Statement

You are tasked with building a log processing component used across both batch ingestion and real-time streaming pipelines.

Each log entry is represented as:

class LogEntry:
    timestamp: int      # Unix timestamp in milliseconds
    event_id: str       # Unique identifier of the underlying event
    payload: str        # Content / message

Duplicate entries can arise due to network retries and distributed producer retries. Duplicates share the same event_id, but may have differing payloads or timestamps.

Part 1: Batch Ingestion

Implement a function deduplicate_batch(logs: list[LogEntry], policy: str) -> list[LogEntry]:

• policy can be either "KEEP_FIRST" or "KEEP_LATEST".
• Under "KEEP_FIRST", keep the record with the smallest timestamp. If there is a tie, keep the one that appeared first in the input list.
• Under "KEEP_LATEST", keep the record with the largest timestamp. If there is a tie, keep the one that appeared latest in the input list.
• The returned list must be sorted in ascending order of timestamp. If two deduplicated records have the exact same timestamp, break ties lexicographically by event_id.

Part 2: Streaming Processor

Implement a streaming class StreamDeduplicator:

• __init__(self, policy: str, allowed_lateness_ms: int): Initializes the processor. allowed_lateness_ms defines the watermark delay. A record arriving with timestamp None: Ingests an incoming log record.
• advance_watermark(self, current_time_ms: int) -> list[LogEntry]: Advances the processor time to current_time_ms. Any events whose timestamp is now strictly older than current_time_ms - allowed_lateness_ms are finalized, emitted in ascending timestamp order (ties broken by event_id), and evicted from internal memory to prevent unbounded growth.

Example

# Batch Example
logs = [
    LogEntry(100, "A", "v1"),
    LogEntry(150, "B", "v1"),
    LogEntry(90,  "A", "v2"),
    LogEntry(150, "C", "v1"),
]

# Under KEEP_FIRST: 'A' keeps timestamp 90 ("v2")
# Result: [LogEntry(90, "A", "v2"), LogEntry(150, "B", "v1"), LogEntry(150, "C", "v1")]

# Under KEEP_LATEST: 'A' keeps timestamp 100 ("v1")
# Result: [LogEntry(100, "A", "v1"), LogEntry(150, "B", "v1"), LogEntry(150, "C", "v1")]

Constraints

• For batch: 0 <= len(logs) <= 10^5.
• timestamp is a non-negative 64-bit integer.
• event_id is a non-empty string of length up to 64 ASCII characters.
• Streaming state must evict emitted entries completely to maintain O(W) space complexity, where W is the number of active entries within the watermark window.