Distributed systems.
Watch them come alive.

// 1. Primary Coordinator: OpenRaft Consensus, Shard Orchestrator & Admin Console
use aaron::{
    admin::AdminService,
    control_plane::ControlPlaneService,
    membership::MembershipService,
    shard::ShardService,
    tracing::TracingService,
    Node,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize cluster subsystems on documented default ports
    let (membership_svc, membership_handle) = MembershipService::pair();
    let (control_svc, control_handle) = ControlPlaneService::pair();
    let (shard_svc, shard_handle) = ShardService::coordinator(control_handle.clone());

    let node = Node::new("orders-coordinator")
        .with(membership_svc)                                              // SWIM Gossip on default port 7946
        .with(TracingService::new())                                       // Dynamic log filter reload
        .with(control_svc)                                                 // OpenRaft consensus RPC on port 18946
        .with(shard_svc)                                                   // Shard placement coordinator
        .with(
            AdminService::new()                                            // Embedded Admin Web UI on port 8080
                .with_membership_handle(membership_handle)
                .with_control_plane_handle(control_handle)
                .with_shard_handle(shard_handle),
        );

    println!("Orders coordinator online. Admin console available at http://127.0.0.1:8080");
    node.run().await?;
    Ok(())
}

// 2. Application Worker: Receives ShardEvents from Control Plane & streams telemetry back
use aaron::{
    membership::{MembershipConfig, MembershipService},
    service_fn,
    shard::{MemberRole, ShardEvent, ShardService},
    tracing::TracingService,
    Context, Node,
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Join SWIM gossip mesh (port 7947 to avoid collision on localhost)
    let config = MembershipConfig::lan()
        .bind_addr("0.0.0.0:7947")
        .add_seed("127.0.0.1:7946");

    let (membership_svc, _membership_handle) = MembershipService::pair_with_config(config);

    // ShardService::worker() binds QUIC (18946), receives commands from Control Plane,
    // publishes ShardEvents to EventHub, and streams telemetry back to Control Plane
    let (shard_svc, shard_handle) = ShardService::worker();

    let node = Node::new("inventory-worker")
        .with(membership_svc)                                              // Gossip mesh discovery
        .with(TracingService::new())                                       // Observability service
        .with(shard_svc)                                                   // Shard listener & telemetry reporter
        .with(service_fn("inventory-storage", move |ctx: Context| {
            let shards = shard_handle.clone();
            async move {
                // Subscribe to reactive shard lifecycle events dispatched by ShardService
                let mut sub = ctx.event_hub.subscribe::<ShardEvent>().await;

                while let Ok(event) = sub.recv().await {
                    match event {
                        ShardEvent::Join { shard_id, members, role } => {
                            // Initialize local LSM keyspace partition for assigned shard
                            let store = ctx.store.keyspace(&format!("inventory_shard_{shard_id}"))?;
                            store.insert(b"status", b"active_partition")?;
                            ctx.store.persist()?;

                            // Notify local state & acknowledge role in shard quorum
                            shards.announce_role(shard_id, role).await;
                            println!("Assigned shard {shard_id} as {role:?} (Quorum: {} members)", members.len());
                        }
                        ShardEvent::RoleChanged { shard_id, role } => {
                            shards.announce_role(shard_id, role).await;
                            println!("Shard {shard_id} role updated -> {role:?}");
                        }
                        ShardEvent::Leave { shard_id } => {
                            println!("Shard {shard_id} decommissioned. Detaching keyspace.");
                        }
                        ShardEvent::Bootstrap { shards: batch } => {
                            println!("Bootstrapping {} partitions from Control Plane.", batch.len());
                        }
                    }
                }
                Ok(())
            }
        }));

    println!("Worker running. QUIC listener on :18946 • Telemetry streaming to Control Plane");
    node.run().await?;
    Ok(())
}

// 3. Standalone Supervisor: Local actor lifecycle & embedded LSM storage (No network required)
use aaron::{service_fn, Context, Node};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Supervise local services with isolated LSM storage and graceful drain on SIGINT
    let node = Node::new("local-runner")
        .with(service_fn("metrics-collector", |ctx: Context| async move {
            let stats = ctx.store.keyspace("telemetry")?;
            let mut seq: u64 = 0;

            // Service loop drains cleanly when supervisor receives SIGINT/SIGTERM
            while !ctx.cancellation_token.is_cancelled() {
                tokio::time::sleep(Duration::from_millis(500)).await;
                seq += 1;
                stats.insert("heartbeat:ticks", seq.to_le_bytes())?;
                println!("Supervised collector heartbeat [seq={seq}]");
            }
            Ok(())
        }));

    println!("Standalone supervisor running. Press Ctrl+C for graceful drain.");
    node.run().await?;
    Ok(())
}

// 4. EventHub: Decoupled in-memory async pub/sub between supervised services
use aaron::{service_fn, Context, Node};

#[derive(Clone, Debug)]
struct OrderEvent {
    order_id: &'static str,
    amount_cents: u64,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let node = Node::new("event-driven-node")
        // Service 1: Asynchronous subscriber reacting to OrderEvent broadcast
        .with(service_fn("billing-worker", |ctx: Context| async move {
            let mut stream = ctx.event_hub.subscribe::<OrderEvent>().await;
            tokio::spawn(async move {
                while let Ok(ev) = stream.recv().await {
                    println!("Billing processed order {} for ${:.2}", ev.order_id, ev.amount_cents as f64 / 100.0);
                }
            });
            Ok(())
        }))
        // Service 2: Publisher emitting strongly-typed domain events
        .with(service_fn("order-gateway", |ctx: Context| async move {
            let hub = ctx.event_hub.clone();
            hub.publish(OrderEvent { order_id: "ORD-9901", amount_cents: 9950 }).await;
            Ok(())
        }));

    println!("EventHub node online. Services communicating over lockless bus.");
    node.run().await?;
    Ok(())
}