// Crate: async-nats = "0.35"
use async_nats::jetstream;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Clone)]
pub struct WorkPackageEvent {
    pub wp_id: String,
    pub event_type: String,
    pub data: serde_json::Value,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to NATS server
    let client = async_nats::connect("nats://localhost:4222").await?;

    // Get JetStream context
    let jetstream = jetstream::new(client);

    // Create or fetch a stream
    let stream = jetstream
        .get_or_create_stream(jetstream::stream::Config {
            name: "wp-events".to_string(),
            subjects: vec!["wp.>".to_string()],
            retention: jetstream::stream::RetentionPolicy::Interest,
            max_age: std::time::Duration::from_secs(7 * 24 * 3600), // 1 week
            storage: jetstream::stream::StorageType::File,
            discard: jetstream::stream::DiscardPolicy::Old,
            ..Default::default()
        })
        .await?;

    // Publish a message
    let event = WorkPackageEvent {
        wp_id: "WP-001".to_string(),
        event_type: "created".to_string(),
        data: serde_json::json!({}),
    };

    jetstream
        .publish("wp.created", serde_json::to_vec(&event)?.into())
        .await?;

    // Create a durable consumer
    let consumer = stream
        .create_consumer(jetstream::consumer::PullConsumer {
            durable_name: Some("wp-processor".to_string()),
            ..Default::default()
        })
        .await?;

    // Consume messages with exactly-once semantics
    let mut messages = consumer.messages().await?;
    while let Ok(msg) = messages.next().await {
        match serde_json::from_slice::<WorkPackageEvent>(&msg.payload) {
            Ok(event) => {
                println!("Processing event: {:?}", event);
                msg.ack().await?;
            }
            Err(e) => {
                eprintln!("Failed to deserialize: {}", e);
                msg.nak().await?; // Negative ack for redelivery
            }
        }
    }

    Ok(())
}

// Exactly-once processing pattern
async fn process_wp_events(jetstream: &jetstream::Context) -> Result<()> {
    let stream = jetstream.get_stream("wp-events").await?;

    let consumer = stream
        .create_consumer(jetstream::consumer::PullConsumer {
            durable_name: Some("wp-processor-1".to_string()),
            deliver_policy: jetstream::consumer::DeliverPolicy::All,
            ack_policy: jetstream::consumer::AckPolicy::Explicit,
            max_deliver: 5, // Retry limit
            backoff: vec![
                std::time::Duration::from_secs(1),
                std::time::Duration::from_secs(5),
                std::time::Duration::from_secs(30),
            ],
            ..Default::default()
        })
        .await?;

    // Pull messages in batches
    let mut messages = consumer.fetch().max_messages(100).await?;
    while let Ok(Some(msg)) = messages.try_next().await {
        // Idempotent processing (store event ID in processed set)
        process_event(&msg).await?;
        msg.ack().await?;
    }

    Ok(())
}

// Crate: redis = "0.25", bb8-redis = "0.16"
use bb8_redis::redis::AsyncCommands;
use bb8_redis::RedisConnectionManager;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create connection manager for Dragonfly
    let manager = RedisConnectionManager::new("redis://localhost:6379")?;
    let pool = bb8::Pool::builder()
        .max_size(16)
        .build(manager)
        .await?;

    // Store snapshot with TTL
    let mut conn = pool.get().await?;
    let snapshot_json = serde_json::to_string(&snapshot)?;
    conn.set_ex(
        format!("snapshot:{}", wp_id),
        snapshot_json,
        Duration::from_secs(3600).as_secs() as usize,
    )
    .await?;

    // Rate limiting: increment counter
    let key = format!("rate_limit:user:{}", user_id);
    let count: i32 = conn.incr(&key, 1).await?;
    if count == 1 {
        conn.expire(&key, 60).await?; // 60-second window
    }
    if count > 100 {
        return Err("Rate limit exceeded".into());
    }

    // Session state storage
    conn.hset(
        format!("session:{}", session_id),
        "user_id",
        user_id,
    )
    .await?;
    conn.hset(
        format!("session:{}", session_id),
        "expires_at",
        chrono::Utc::now() + chrono::Duration::hours(1),
    )
    .await?;
    conn.expire(format!("session:{}", session_id), 3600).await?;

    Ok(())
}

// Crate: neo4rs = "0.7"
use neo4rs::Graph;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Clone)]
pub struct WorkPackage {
    pub id: String,
    pub title: String,
    pub status: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Connect to Neo4j
    let graph = Graph::new("bolt://localhost:7687", "neo4j", "password")
        .await?;

    // Create a WP node
    let query = "CREATE (wp:WorkPackage {
        id: $id,
        title: $title,
        status: $status,
        created_at: timestamp()
    }) RETURN wp";

    let mut result = graph
        .execute(
            neo4rs::query(query)
                .param("id", "WP-001")
                .param("title", "Foundation Infrastructure")
                .param("status", "IN_PROGRESS"),
        )
        .await?;

    while let Ok(Some(row)) = result.next().await {
        let node: neo4rs::Node = row.get("wp")?;
        println!("Created node: {:?}", node);
    }

    Ok(())
}

// Find all dependencies of a WP
async fn get_dependencies(
    graph: &Graph,
    wp_id: &str,
) -> Result<Vec<String>, Box<dyn std::error::Error>> {
    let query = "
        MATCH (wp:WorkPackage {id: $id})-[:DEPENDS_ON]->(dep:WorkPackage)
        RETURN dep.id AS dependent_id
    ";

    let mut result = graph
        .execute(neo4rs::query(query).param("id", wp_id))
        .await?;

    let mut deps = Vec::new();
    while let Ok(Some(row)) = result.next().await {
        deps.push(row.get::<String>("dependent_id")?);
    }
    Ok(deps)
}

// Find shortest path between two WPs (blocking path)
async fn find_blocking_path(
    graph: &Graph,
    from_id: &str,
    to_id: &str,
) -> Result<Option<Vec<String>>, Box<dyn std::error::Error>> {
    let query = "
        MATCH path = shortestPath(
            (from:WorkPackage {id: $from})-[:BLOCKS*]->(to:WorkPackage {id: $to})
        )
        RETURN [node IN nodes(path) | node.id] AS path_ids
    ";

    let mut result = graph
        .execute(
            neo4rs::query(query)
                .param("from", from_id)
                .param("to", to_id),
        )
        .await?;

    if let Ok(Some(row)) = result.next().await {
        Ok(Some(row.get("path_ids")?))
    } else {
        Ok(None)
    }
}

// Transactional WP creation with relationship
async fn create_wp_with_dependency(
    graph: &Graph,
    wp: &WorkPackage,
    depends_on_id: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut txn = graph.start_txn().await?;

    // Create WP node
    txn.run(
        neo4rs::query(
            "CREATE (wp:WorkPackage {id: $id, title: $title, status: $status})"
        )
        .param("id", &wp.id)
        .param("title", &wp.title)
        .param("status", &wp.status),
    )
    .await?;

    // Create relationship
    txn.run(
        neo4rs::query(
            "MATCH (from:WorkPackage {id: $depends_on}),
                    (to:WorkPackage {id: $id})
             CREATE (from)-[:BLOCKED_BY]->(to)"
        )
        .param("depends_on", depends_on_id)
        .param("id", &wp.id),
    )
    .await?;

    txn.commit().await?;
    Ok(())
}

// Using aws-sdk-s3 (recommended for full compatibility)
// Crate: aws-sdk-s3 = "1.x", aws-config = "1.x", tokio = "1"
use aws_sdk_s3::Client;
use std::path::Path;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Configure S3 client for MinIO endpoint
    let config = aws_config::load_from_env().await;
    let mut s3_config = aws_sdk_s3::config::Builder::from(&config)
        .endpoint_url("http://localhost:9000")
        .build();

    let client = Client::from_conf(s3_config);

    // Create bucket
    client
        .create_bucket()
        .bucket("artifacts")
        .create_bucket_configuration(
            aws_sdk_s3::types::CreateBucketConfiguration::builder()
                .location_constraint(
                    aws_sdk_s3::types::BucketLocationConstraint::UsEast1,
                )
                .build(),
        )
        .send()
        .await?;

    // Upload object
    let body = aws_sdk_s3::primitives::ByteStream::from_path(
        Path::new("/path/to/artifact.tar.gz"),
    )
    .await?;

    client
        .put_object()
        .bucket("artifacts")
        .key(format!("wp-001/snapshot-{}.tar.gz", chrono::Utc::now()))
        .body(body)
        .set_metadata(Some({
            let mut map = std::collections::HashMap::new();
            map.insert("wp-id".to_string(), "WP-001".to_string());
            map
        }))
        .send()
        .await?;

    // List objects
    let resp = client
        .list_objects_v2()
        .bucket("artifacts")
        .prefix("wp-001/")
        .send()
        .await?;

    for obj in resp.contents.unwrap_or_default() {
        println!("Object: {}", obj.key.unwrap_or_default());
    }

    // Download object
    let obj = client
        .get_object()
        .bucket("artifacts")
        .key("wp-001/snapshot-latest.tar.gz")
        .send()
        .await?;

    let data = obj.body.collect().await?;
    println!("Downloaded {} bytes", data.len());

    Ok(())
}

// Crate: minio = "0.2" (if available)
// Note: minio-rs is maintained by MinIO but less common in Rust ecosystem
// Recommended to use aws-sdk-s3 instead

// In Rust API server (agileplus-api)
use axum::{routing::get, Router};
use std::sync::Arc;

async fn health_handler() -> (http::StatusCode, &'static str) {
    (http::StatusCode::OK, "OK")
}

async fn readiness_handler(
    State(app): State<Arc<AppState>>,
) -> Result<(http::StatusCode, &'static str), http::StatusCode> {
    // Check critical dependencies
    if !app.nats_ready().await {
        return Err(http::StatusCode::SERVICE_UNAVAILABLE);
    }
    if !app.db_ready().await {
        return Err(http::StatusCode::SERVICE_UNAVAILABLE);
    }
    Ok((http::StatusCode::OK, "Ready"))
}

pub fn health_routes() -> Router {
    Router::new()
        .route("/health", get(health_handler))
        .route("/ready", get(readiness_handler))
}

// Main application initialization (agileplus-api/src/main.rs)
use async_nats::jetstream;
use neo4rs::Graph;
use redis::aio::ConnectionManager;

pub struct AppState {
    nats: async_nats::Client,
    jetstream: jetstream::Context,
    neo4j: Graph,
    redis: ConnectionManager,
    s3: aws_sdk_s3::Client,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt::init();

    // Initialize infrastructure clients
    let nats = async_nats::connect(&env::var("NATS_URL")?).await?;
    let jetstream = jetstream::new(nats.clone());

    let neo4j = Graph::new(
        &env::var("NEO4J_URL")?,
        &env::var("NEO4J_USER")?,
        &env::var("NEO4J_PASSWORD")?,
    )
    .await?;

    let redis_url = env::var("REDIS_URL")?;
    let redis = redis::Client::open(redis_url)?
        .get_tokio_connection_manager()
        .await?;

    let s3_config = aws_config::load_from_env().await;
    let s3 = aws_sdk_s3::Client::new(&s3_config);

    let app_state = AppState {
        nats,
        jetstream,
        neo4j,
        redis,
        s3,
    };

    // Start API server with health routes
    let app = Router::new()
        .merge(health_routes())
        .merge(api_routes())
        .with_state(Arc::new(app_state));

    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
    axum::serve(listener, app).await?;

    Ok(())
}