# AICL Example: Observer Pattern Event Bus
# Implements a robust publish/subscribe event bus with topic routing, guaranteed message ordering,
# backpressure management, dead letter queue handling, and replay capabilities.

# Level 1: Architecture
Goal: Build a high-throughput event bus implementing the Observer pattern with topic-based routing,
    strict per-topic message ordering, backpressure to protect slow subscribers,
    dead letter queues for poison messages, and replay support for missed events.

Constraint: Publishers must be decoupled from subscribers; no direct references allowed.
Constraint: Each topic maintains FIFO ordering within a partition key.
Constraint: Subscribers must acknowledge messages within a configurable timeout.
Constraint: Backpressure must be applied before memory pressure reaches critical threshold.
Constraint: Dead letter messages must never be silently dropped; always persisted for inspection.

Risk: Slow subscriber causes unbounded buffer growth
Recovery: Apply per-subscriber backpressure with configurable max buffer; overflow goes to DLQ

Risk: Message ordering violation under concurrent publishing
Recovery: Partition by routing key; single-writer per partition with sequencer

Risk: Dead letter queue grows unbounded without reprocessing
Recovery: Scheduled DLQ reprocessor with exponential backoff; alert at queue depth threshold

Risk: Subscriber crash loses in-flight messages
Recovery: Redeliver unacknowledged messages after visibility timeout with at-least-once guarantee

Risk: Topic routing misconfiguration routes messages to wrong subscribers
Recovery: Validate topic subscription patterns at registration; audit log all routing decisions

Risk: Replay floods subscribers with historical volume
Recovery: Throttle replay delivery rate; subscribers opt-in to replay with max throughput setting

Layer: Core
    SubLayer: Publisher
    SubLayer: TopicRouter
    SubLayer: PartitionManager
Layer: Subscription
    SubLayer: SubscriberRegistry
    SubLayer: SubscriptionManager
    SubLayer: AcknowledgmentTracker
Layer: Reliability
    SubLayer: BackpressureController
    SubLayer: DeadLetterQueue
    SubLayer: ReplayEngine

Validation: Topic name must match [a-zA-Z][a-zA-Z0-9_.-] pattern
Validation: Subscriber callback must complete within acknowledgment timeout
Validation: Partition key must be present for ordered topics
Validation: Replay request must specify valid time range and subscriber ID
Validation: Backpressure threshold must be positive and less than system memory limit
Validation: Dead letter message must preserve original headers and failure reason

# Level 2: Entities
Entity Topic
    topicId: string
    topicName: string
    partitionCount: integer
    retentionHours: integer
    isOrdered: boolean
    createdAt: datetime
    messageCount: integer
    byteRate: float
    schema: dict

Entity Subscription
    subscriptionId: string
    subscriberName: string
    topicPattern: string
    callback: string
    filterExpression: string
    maxInFlight: integer
    ackTimeout: integer
    retryPolicy: dict
    createdAt: datetime

Entity Message
    messageId: string
    topic: string
    partitionKey: string
    sequenceNumber: integer
    payload: dict
    headers: dict
    publishedAt: datetime
    expiresAt: datetime
    traceId: string

Entity DeadLetterEntry
    entryId: string
    originalMessage: Message
    subscriberName: string
    failureReason: string
    failureCount: integer
    firstFailedAt: datetime
    lastFailedAt: datetime
    nextRetryAt: datetime
    deadLetteredAt: datetime

Entity BackpressureState
    subscriberId: string
    currentBufferSize: integer
    maxBufferSize: integer
    pressureLevel: string
    throttleRate: float
    droppedCount: integer
    lastAdjustedAt: datetime

Entity SubscriberMetrics
    subscriberId: string
    messagesProcessed: integer
    messagesFailed: integer
    averageLatency: float
    p99Latency: float
    lastMessageAt: datetime
    activeSubscriptions: integer
    dlqDepth: integer

# Level 3: Behaviors
Behavior PublishMessage
    Input:
        topic: string
        partitionKey: string
        payload: dict
        headers: dict
    Output:
        messageId: string
        sequenceNumber: integer
    Action:
        Validate topic exists and schema matches payload
        Determine partition from partition key hash
        Assign monotonic sequence number within partition
        Persist message to topic log
        Route message to all matching subscriptions
        Apply backpressure check per subscriber before delivery
        Return message ID and sequence number

Behavior Subscribe
    Input:
        topicPattern: string
        callback: string
        config: dict
    Output:
        subscriptionId: string
    Action:
        Validate topic pattern syntax and matching topics
        Register subscription in subscriber registry
        Initialize delivery buffer and backpressure controller
        Begin message delivery from current head
        Return subscription identifier

Behavior AcknowledgeMessage
    Input:
        subscriptionId: string
        messageId: string
        status: string
    Output:
        acknowledged: boolean
    Action:
        Verify message belongs to subscription's in-flight set
        If status is success, remove from in-flight and advance checkpoint
        If status is failure, apply retry policy or move to DLQ
        Update subscriber metrics
        Release backpressure slot
        Return acknowledgment confirmation

Behavior RouteToSubscriber
    Input:
        message: Message
        subscription: Subscription
    Output:
        delivered: boolean
    Action:
        Evaluate filter expression against message headers
        Check subscriber backpressure state
        If pressure is critical, route to DLQ with backpressure reason
        If pressure is moderate, apply throttle delay
        If acceptable, deliver to subscriber callback
        Track delivery in acknowledgment tracker
        Return delivery result

Behavior ProcessDeadLetter
    Input:
        entryId: string
        action: string
    Output:
        result: string
    Action:
        Load dead letter entry from DLQ store
        If action is retry, re-inject message to original subscription
        If action is discard, log and permanently remove
        If action is redirect, publish to alternate topic
        Update DLQ metrics and entry status
        Return action result

Behavior ReplayMessages
    Input:
        subscriptionId: string
        fromTimestamp: datetime
        toTimestamp: datetime
        maxThroughput: integer
    Output:
        replayId: string
        messageCount: integer
    Action:
        Validate subscription and time range
        Load messages from topic log for time range
        Apply max throughput throttling
        Deliver messages to subscriber callback
        Skip already-acknowledged messages
        Track replay progress and completion
        Return replay statistics

# Level 4: Conditions
Condition: BackpressureCritical
    When subscriber buffer reaches 90% of max capacity
    Then redirect new messages to DLQ with backpressure flag and alert subscriber owner

Condition: AcknowledgmentTimeout
    When message remains unacknowledged beyond ackTimeout
    Then redeliver message with incremented delivery attempt counter; after max attempts move to DLQ

Condition: InvalidMessageSchema
    When message payload does not conform to topic schema
    Then reject publication with validation error; if already published, route to DLQ

# Level 5: Events
Event: OnSubscriberRegistered
    On new subscription created for a topic pattern
    Action: Warm subscriber connection pool and validate callback endpoint reachability

Event: OnBackpressureThreshold
    On subscriber buffer crosses 75% capacity
    Action: Apply throttle, notify subscriber via health endpoint, and begin buffering upstream

Event: OnDeadLetterCreated
    On message moved to dead letter queue
    Action: Emit metric, send notification to subscriber owner, and log for audit trail

Event: OnReplayCompleted
    On message replay batch fully delivered
    Action: Log replay summary, update subscriber checkpoint, and emit completion metric

# Level 6: Concurrency
Parallel:
    Message publishing to multiple partitions concurrently
    Message delivery to independent subscribers in parallel
    Dead letter queue reprocessing on separate worker pool
    Backpressure monitoring and adjustment across all subscribers
    Metrics aggregation from all subscriber delivery threads

# Level 7: Optimization
Optimize: Message delivery throughput
    Priority: Batch deliveries per subscriber; minimize lock contention on partition sequencer

Optimize: Topic routing latency
    Priority: Cache subscription pattern matching results; pre-compile filter expressions

# Level 8: Learning
Learn: Subscriber processing capacity
    Goal: Predict optimal max-in-flight setting per subscriber to maximize throughput without overflow
    Adapt: Per-subscriber maxInFlight and buffer size
    Based: Subscriber acknowledgment rate and latency distribution over sliding window

Learn: DLQ root cause patterns
    Goal: Categorize dead-lettered messages by failure type to identify systemic issues
    Adapt: Alerting rules and auto-retry thresholds per failure category
    Based: Historical DLQ entry failure reason distribution and retry success rates

# Level 9: Security
Security:
    Encrypt: Message payload in transit via TLS 1.3 with pinned certificates between bus and subscribers
    Encrypt: DLQ entries at rest using AES-256 with per-topic encryption keys
    Protect: Topic publication with IAM-style authorization and topic-level ACL
    Protect: Subscription callbacks from message injection via HMAC message signing
    Protect: Replay endpoints from unauthorized access with admin role requirement and audit logging

# Level 10: Native
Rust
{
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

#[derive(Clone, Debug)]
pub struct Message {
    pub message_id: String,
    pub topic: String,
    pub partition_key: String,
    pub sequence_number: u64,
    pub payload: Vec<u8>,
    pub headers: HashMap<String, String>,
    pub published_at: Instant,
}

#[derive(Clone, Debug)]
pub struct Subscription {
    pub subscription_id: String,
    pub subscriber_name: String,
    pub topic_pattern: String,
    pub max_in_flight: usize,
    pub ack_timeout: Duration,
    pub retry_max: u32,
}

#[derive(Debug)]
pub enum DeliverStatus {
    Acknowledged,
    Failed(String),
    Timeout,
}

pub struct DeadLetterEntry {
    pub entry_id: String,
    pub original_message: Message,
    pub subscriber_name: String,
    pub failure_reason: String,
    pub failure_count: u32,
    pub first_failed_at: Instant,
    pub dead_lettered_at: Instant,
}

pub struct EventBus {
    topics: Arc<Mutex<HashMap<String, Vec<Message>>>>,
    subscriptions: Arc<Mutex<Vec<Subscription>>>,
    in_flight: Arc<Mutex<HashMap<String, (Message, Instant, u32)>>>,
    dead_letter_queue: Arc<Mutex<Vec<DeadLetterEntry>>>,
    max_buffer: usize,
}

impl EventBus {
    pub fn new(max_buffer: usize) -> Self {
        EventBus {
            topics: Arc::new(Mutex::new(HashMap::new())),
            subscriptions: Arc::new(Mutex::new(Vec::new())),
            in_flight: Arc::new(Mutex::new(HashMap::new())),
            dead_letter_queue: Arc::new(Mutex::new(Vec::new())),
            max_buffer,
        }
    }

    pub fn publish(&self, topic: &str, msg: Message) -> u64 {
        let mut topics = self.topics.lock().unwrap();
        let partition = topics.entry(topic.to_string()).or_insert_with(Vec::new);
        let seq = partition.len() as u64;
        let mut msg = msg;
        msg.sequence_number = seq;
        partition.push(msg.clone());
        seq
    }

    pub fn subscribe(&self, sub: Subscription) {
        let mut subs = self.subscriptions.lock().unwrap();
        subs.push(sub);
    }

    pub fn acknowledge(&self, subscription_id: &str, message_id: &str) -> bool {
        let mut in_flight = self.in_flight.lock().unwrap();
        in_flight.remove(&format!("{}:{}", subscription_id, message_id)).is_some()
    }

    pub fn check_timeouts(&self) {
        let mut in_flight = self.in_flight.lock().unwrap();
        let mut dlq = self.dead_letter_queue.lock().unwrap();
        let now = Instant::now();
        let expired: Vec<String> = in_flight.iter()
            .filter(|(_, (msg, sent_at, retries))| {
                now.duration_since(*sent_at) > Duration::from_secs(30) && *retries >= 3
            })
            .map(|(k, _)| k.clone())
            .collect();
        for key in expired {
            if let Some((msg, _, retries)) = in_flight.remove(&key) {
                dlq.push(DeadLetterEntry {
                    entry_id: format!("dlq-{}", msg.message_id),
                    original_message: msg,
                    subscriber_name: String::new(),
                    failure_reason: "ack_timeout_exceeded".to_string(),
                    failure_count: retries,
                    first_failed_at: now,
                    dead_lettered_at: now,
                });
            }
        }
    }

    pub fn dead_letter_count(&self) -> usize {
        self.dead_letter_queue.lock().unwrap().len()
    }
}

fn main() {
    let bus = EventBus::new(10_000);
    let sub = Subscription {
        subscription_id: "sub-1".to_string(),
        subscriber_name: "order-service".to_string(),
        topic_pattern: "orders.*".to_string(),
        max_in_flight: 50,
        ack_timeout: Duration::from_secs(30),
        retry_max: 3,
    };
    bus.subscribe(sub);
    let msg = Message {
        message_id: "msg-1".to_string(),
        topic: "orders.created".to_string(),
        partition_key: "order-123".to_string(),
        sequence_number: 0,
        payload: vec![],
        headers: HashMap::new(),
        published_at: Instant::now(),
    };
    let seq = bus.publish("orders.created", msg);
    println!("Published message with sequence: {}", seq);
}