Intermediate

// Cargo.toml:
// sqlx = { version = "0.8", features = ["postgres", "runtime-tokio", "macros"] }
// axum = "0.8"
// tokio = { version = "1", features = ["full"] }

use axum::{extract::State, routing::get, Json, Router};
use sqlx::{PgPool, postgres::PgPoolOptions};  // => PostgreSQL pool
use std::sync::Arc;

// AppState holds the connection pool - shared across all handlers
#[derive(Clone)]
struct AppState {
    db: PgPool,  // => PgPool is already Arc-wrapped internally; Clone is cheap
}

// Simple row structure for a SELECT query
#[derive(sqlx::FromRow, serde::Serialize)]  // => FromRow: map DB rows to struct
struct User {
    id: i64,         // => PostgreSQL BIGINT or SERIAL
    username: String, // => PostgreSQL TEXT or VARCHAR
    email: String,
}

async fn list_users(
    State(state): State<Arc<AppState>>,  // => Inject pool from state
) -> Result<Json<Vec<User>>, axum::http::StatusCode> {
    let users = sqlx::query_as::<_, User>(
        "SELECT id, username, email FROM users ORDER BY id LIMIT 50"
        // => query_as maps each row to User via FromRow derive
        // => Compile-time checks if SQLX_OFFLINE=true (requires sqlx-data.json)
    )
    .fetch_all(&state.db)              // => Execute query, get Vec<User>
    .await
    .map_err(|_| axum::http::StatusCode::INTERNAL_SERVER_ERROR)?;
    // => DB error → 500 Internal Server Error

    Ok(Json(users))                    // => 200 OK with JSON array
}

#[tokio::main]
async fn main() {
    let database_url = std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "postgres://user:pass@localhost/mydb".into());
    // => DATABASE_URL from environment - never hardcode credentials

    let pool = PgPoolOptions::new()
        .max_connections(10)           // => Max 10 concurrent DB connections
        .min_connections(2)            // => Keep 2 connections warm
        .connect(&database_url)
        .await
        .expect("Failed to connect to database");
    // => Panics on startup if DB unreachable - fail fast is correct here

    let state = Arc::new(AppState { db: pool });

    let app = Router::new()
        .route("/users", get(list_users))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

use axum::{
    extract::{Path, State},
    http::StatusCode,
    routing::{get, post, put, delete},
    Json, Router,
};
use sqlx::PgPool;
use serde::{Deserialize, Serialize};
use std::sync::Arc;

#[derive(Clone)]
struct AppState { db: PgPool }

#[derive(Serialize, sqlx::FromRow)]
struct Post {
    id: i64,
    title: String,
    body: String,
}

#[derive(Deserialize)]
struct CreatePost {
    title: String,  // => Required: title of the post
    body: String,   // => Required: post content
}

#[derive(Deserialize)]
struct UpdatePost {
    title: Option<String>,  // => Optional partial update
    body: Option<String>,   // => Only provided fields are updated
}

// CREATE
async fn create_post(
    State(state): State<Arc<AppState>>,
    Json(payload): Json<CreatePost>,
) -> Result<(StatusCode, Json<Post>), StatusCode> {
    let post = sqlx::query_as::<_, Post>(
        "INSERT INTO posts (title, body) VALUES ($1, $2) RETURNING id, title, body"
        // => $1, $2 are positional placeholders (PostgreSQL syntax)
        // => RETURNING fetches the inserted row in one query
    )
    .bind(&payload.title)   // => Bind $1 = title (prevents SQL injection)
    .bind(&payload.body)    // => Bind $2 = body
    .fetch_one(&state.db)   // => Execute and return single row
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok((StatusCode::CREATED, Json(post)))  // => 201 Created + new post
}

// READ
async fn get_post(
    State(state): State<Arc<AppState>>,
    Path(id): Path<i64>,
) -> Result<Json<Post>, StatusCode> {
    let post = sqlx::query_as::<_, Post>("SELECT id, title, body FROM posts WHERE id = $1")
        .bind(id)               // => Safe parameterized query
        .fetch_optional(&state.db)  // => Returns Option<Post>
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
        .ok_or(StatusCode::NOT_FOUND)?;  // => None → 404 Not Found

    Ok(Json(post))
}

// UPDATE
async fn update_post(
    State(state): State<Arc<AppState>>,
    Path(id): Path<i64>,
    Json(payload): Json<UpdatePost>,
) -> Result<Json<Post>, StatusCode> {
    // Build SET clause dynamically for partial updates
    let post = sqlx::query_as::<_, Post>(
        "UPDATE posts SET
            title = COALESCE($1, title),
            body  = COALESCE($2, body)
         WHERE id = $3 RETURNING id, title, body"
        // => COALESCE($1, title): use $1 if not null, else keep existing value
        // => Implements partial update without multiple queries
    )
    .bind(&payload.title)    // => $1: new title or NULL
    .bind(&payload.body)     // => $2: new body or NULL
    .bind(id)                // => $3: row ID
    .fetch_optional(&state.db)
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
    .ok_or(StatusCode::NOT_FOUND)?;

    Ok(Json(post))
}

// DELETE
async fn delete_post(
    State(state): State<Arc<AppState>>,
    Path(id): Path<i64>,
) -> Result<StatusCode, StatusCode> {
    let result = sqlx::query("DELETE FROM posts WHERE id = $1")
        .bind(id)
        .execute(&state.db)    // => execute() for queries with no return rows
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    if result.rows_affected() == 0 {  // => rows_affected() = 0 means no row matched
        return Err(StatusCode::NOT_FOUND);
    }
    Ok(StatusCode::NO_CONTENT)  // => 204 No Content - successful delete
}

#[tokio::main]
async fn main() {
    let pool = PgPool::connect("postgres://user:pass@localhost/mydb").await.unwrap();
    let state = Arc::new(AppState { db: pool });
    let app = Router::new()
        .route("/posts", post(create_post))
        .route("/posts/:id", get(get_post).put(update_post).delete(delete_post));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

use axum::{
    extract::{Json, State},
    http::StatusCode,
    routing::post,
    Router,
};
use sqlx::PgPool;
use serde::Deserialize;
use std::sync::Arc;

#[derive(Clone)]
struct AppState { db: PgPool }

#[derive(Deserialize)]
struct TransferRequest {
    from_account: i64,   // => Debit this account
    to_account: i64,     // => Credit this account
    amount: i64,         // => Amount in cents (use integers for money)
}

async fn transfer_funds(
    State(state): State<Arc<AppState>>,
    Json(req): Json<TransferRequest>,
) -> Result<StatusCode, StatusCode> {
    // Start a database transaction
    let mut tx = state.db.begin()     // => Acquires a connection and begins transaction
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    // Step 1: Debit the sender
    let rows = sqlx::query(
        "UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1"
        // => Atomic debit: only succeeds if sufficient funds
        // => balance >= $1 prevents overdraft in a single query
    )
    .bind(req.amount)
    .bind(req.from_account)
    .execute(&mut *tx)           // => Execute within the transaction
    .await
    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    if rows.rows_affected() == 0 {
        tx.rollback().await.ok();   // => Insufficient funds - roll back
        return Err(StatusCode::UNPROCESSABLE_ENTITY);  // => 422
    }

    // Step 2: Credit the receiver (within same transaction)
    sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
        .bind(req.amount)
        .bind(req.to_account)
        .execute(&mut *tx)          // => Same transaction - both succeed or both fail
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    // Commit: both operations succeeded
    tx.commit()                     // => Makes both updates permanent atomically
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(StatusCode::OK)              // => 200: transfer successful
    // => If handler returns Err before commit, tx drops and auto-rollbacks
}

#[tokio::main]
async fn main() {
    let pool = PgPool::connect("postgres://user:pass@localhost/mydb").await.unwrap();
    let state = Arc::new(AppState { db: pool });
    let app = Router::new().route("/transfer", post(transfer_funds));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

// Cargo.toml:
// argon2 = "0.5"
// axum = "0.8"

use argon2::{
    password_hash::{rand_core::OsRng, PasswordHash, PasswordHasher, PasswordVerifier, SaltString},
    Argon2,  // => Argon2 hasher (Argon2id variant is the default, recommended)
};

// Hash a password during user registration
fn hash_password(password: &str) -> Result<String, argon2::password_hash::Error> {
    let salt = SaltString::generate(&mut OsRng);  // => Cryptographically random salt
                                                    // => Unique per password - prevents rainbow tables
    let argon2 = Argon2::default();               // => Argon2id with secure default parameters
    let hash = argon2
        .hash_password(password.as_bytes(), &salt)?  // => Compute hash
        .to_string();                                  // => Store this string in DB
    // => hash format: "$argon2id$v=19$m=65536,t=3,p=4$salt$hash"
    // => Embeds algorithm, parameters, salt, and hash in one string
    Ok(hash)
}

// Verify a password during login
fn verify_password(password: &str, hash: &str) -> bool {
    let parsed_hash = PasswordHash::new(hash)       // => Parse stored hash string
        .expect("Invalid hash format in database");  // => DB corruption if this panics
    Argon2::default()
        .verify_password(password.as_bytes(), &parsed_hash)  // => Recompute and compare
        .is_ok()                                              // => true if match
    // => Constant-time comparison prevents timing attacks
}

// Example usage in a registration handler
async fn register_example() {
    let password = "user-secret-password";
    let hash = hash_password(password).unwrap();
    println!("Hash: {}", &hash[..40]);  // => "$argon2id$v=19$m=65536,t=3,p=4..." (truncated)

    let valid = verify_password(password, &hash);
    println!("Valid: {}", valid);    // => "Valid: true"

    let invalid = verify_password("wrong-password", &hash);
    println!("Invalid: {}", invalid); // => "Invalid: false"
}

#[tokio::main]
async fn main() {
    register_example().await;
}

// Cargo.toml:
// jsonwebtoken = "9"
// serde = { version = "1", features = ["derive"] }

use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation};
use serde::{Deserialize, Serialize};
use std::time::{SystemTime, UNIX_EPOCH};

// JWT claims - the payload embedded in the token
#[derive(Serialize, Deserialize)]
struct Claims {
    sub: String,   // => Subject: user ID or email (standard JWT field)
    role: String,  // => Custom claim: user role for authorization
    exp: u64,      // => Expiration: Unix timestamp (standard JWT field)
                    // => JWT library rejects tokens past this time automatically
}

const SECRET: &[u8] = b"my-256-bit-secret-do-not-hardcode-in-production";
// => In production: read from environment variable or secrets manager

// Generate a JWT token (call this after successful login)
fn generate_token(user_id: &str, role: &str) -> Result<String, jsonwebtoken::errors::Error> {
    let expiry = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs() + 3600;  // => Token valid for 1 hour from now

    let claims = Claims {
        sub: user_id.to_string(),  // => User identifier
        role: role.to_string(),    // => User role
        exp: expiry,               // => Expiration timestamp
    };

    encode(
        &Header::default(),                       // => HS256 algorithm (HMAC-SHA256)
        &claims,                                  // => Claims payload
        &EncodingKey::from_secret(SECRET),        // => Sign with secret key
    )
    // => Returns base64url-encoded header.payload.signature
}

// Validate a JWT token (call this in auth middleware)
fn validate_token(token: &str) -> Result<Claims, jsonwebtoken::errors::Error> {
    let token_data = decode::<Claims>(
        token,
        &DecodingKey::from_secret(SECRET),        // => Same secret used for signing
        &Validation::default(),                    // => Validates exp, alg, signature
                                                    // => Returns error if expired or invalid
    )?;
    Ok(token_data.claims)                          // => Validated claims
}

fn main() {
    let token = generate_token("user-42", "admin").unwrap();
    println!("Token: {}...", &token[..30]);  // => "eyJ0eXAiOiJKV1QiLCJhbGc..." (truncated)

    match validate_token(&token) {
        Ok(claims) => println!("Valid! sub={}, role={}", claims.sub, claims.role),
        // => "Valid! sub=user-42, role=admin"
        Err(e) => println!("Invalid: {}", e),
    }
}

use axum::{
    extract::Extension,
    http::{Request, StatusCode},
    middleware::{self, Next},
    response::Response,
    routing::get,
    Router,
};
use jsonwebtoken::{decode, DecodingKey, Validation};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Clone, Debug)]
struct Claims {
    sub: String,
    role: String,
    exp: u64,
}

const SECRET: &[u8] = b"my-secret-key";

// Middleware: validate JWT and insert Claims into request extensions
async fn jwt_middleware(
    mut request: Request,
    next: Next,
) -> Result<Response, StatusCode> {
    // Extract Bearer token from Authorization header
    let auth_header = request
        .headers()
        .get("Authorization")
        .and_then(|v| v.to_str().ok())
        .ok_or(StatusCode::UNAUTHORIZED)?;  // => 401 if header missing

    let token = auth_header
        .strip_prefix("Bearer ")             // => Remove "Bearer " prefix
        .ok_or(StatusCode::UNAUTHORIZED)?;   // => 401 if wrong scheme

    // Validate JWT and extract claims
    let claims = decode::<Claims>(
        token,
        &DecodingKey::from_secret(SECRET),
        &Validation::default(),              // => Checks signature + expiry
    )
    .map_err(|_| StatusCode::UNAUTHORIZED)?  // => 401 if invalid/expired
    .claims;

    // Insert validated claims into request extensions
    request.extensions_mut().insert(claims);  // => Keyed by type Claims
    Ok(next.run(request).await)               // => Continue to handler
}

// Protected handler: extract Claims set by middleware
async fn profile(Extension(claims): Extension<Claims>) -> String {
    format!("User: {}, Role: {}", claims.sub, claims.role)
    // => "User: user-42, Role: admin"
}

// Admin-only handler
async fn admin_only(Extension(claims): Extension<Claims>) -> Result<String, StatusCode> {
    if claims.role != "admin" {
        return Err(StatusCode::FORBIDDEN);  // => 403 if not admin
    }
    Ok("Admin panel".to_string())
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/profile", get(profile))
        .route("/admin", get(admin_only))
        .layer(middleware::from_fn(jwt_middleware));
        // => All routes require valid JWT

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

// Cargo.toml:
// tower-http = { version = "0.6", features = ["cors"] }

use axum::{routing::get, Router};
use tower_http::cors::{Any, CorsLayer};  // => CORS layer and helpers
use axum::http::{HeaderName, Method};

#[tokio::main]
async fn main() {
    // Development: allow all origins (use in dev only!)
    let dev_cors = CorsLayer::permissive();
    // => Allows all origins, methods, and headers
    // => NEVER use in production

    // Production: strict CORS for an API used by a React frontend
    let prod_cors = CorsLayer::new()
        .allow_origin([
            "https://app.example.com".parse().unwrap(),  // => Main frontend
            "https://staging.example.com".parse().unwrap(), // => Staging env
        ])
        .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
        // => Only allow these HTTP methods
        .allow_headers([
            axum::http::header::CONTENT_TYPE,        // => Required for JSON APIs
            axum::http::header::AUTHORIZATION,       // => JWT tokens
            HeaderName::from_static("x-request-id"), // => Custom header
        ])
        .allow_credentials(true);
        // => Allow cookies and Authorization headers in cross-origin requests
        // => Requires specific origins (not `Any`) when using credentials

    let app = Router::new()
        .route("/api/data", get(data_handler))
        .layer(prod_cors);  // => Apply CORS to all routes

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn data_handler() -> &'static str { "API data" }

// Cargo.toml:
// tower_governor = "0.4"
// axum = "0.8"

use axum::{routing::get, Router};
use tower_governor::{
    governor::GovernorConfigBuilder,
    GovernorLayer,  // => Tower layer for rate limiting
};
use std::sync::Arc;

#[tokio::main]
async fn main() {
    // Configure rate limiter: 5 requests per second per IP
    let governor_conf = Arc::new(
        GovernorConfigBuilder::default()
            .per_second(5)          // => 5 requests per second (refill rate)
            .burst_size(10)         // => Allow burst of up to 10 requests
                                     // => Token bucket: starts with 10, refills at 5/sec
            .finish()
            .unwrap(),
    );

    let app = Router::new()
        .route("/api/limited", get(rate_limited_handler))
        .route("/api/unlimited", get(unlimited_handler))
        .layer(GovernorLayer {
            config: governor_conf,  // => Apply rate limiting to all routes above
        });
        // => Requests exceeding limit get 429 Too Many Requests
        // => X-RateLimit-Limit and Retry-After headers are set automatically

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn rate_limited_handler() -> &'static str {
    "Rate limited response"  // => Rejected with 429 if rate exceeded
}

async fn unlimited_handler() -> &'static str {
    "No rate limit"          // => Always served (layer applied before this route? NO)
    // => IMPORTANT: layer() applies to routes defined BEFORE it in Axum 0.7+
    // => To exclude a route from rate limiting, add it to a separate Router
}

// Cargo.toml:
// axum = { version = "0.8", features = ["multipart"] }

use axum::{
    extract::Multipart,          // => Multipart form data extractor
    http::StatusCode,
    routing::post,
    Router,
};

async fn upload_file(
    mut multipart: Multipart,    // => Iterator over multipart fields
) -> Result<String, StatusCode> {
    let mut file_names = Vec::new();

    // Iterate over each field in the multipart form
    while let Some(field) = multipart.next_field().await
        .map_err(|_| StatusCode::BAD_REQUEST)?  // => Malformed multipart → 400
    {
        let name = field.name()
            .unwrap_or("unknown")       // => Form field name (e.g., "file")
            .to_string();

        let file_name = field.file_name()
            .unwrap_or("unnamed")       // => Original filename from browser
            .to_string();

        // Read file bytes
        let data = field.bytes()        // => Read entire field as Bytes
            .await
            .map_err(|_| StatusCode::BAD_REQUEST)?;

        // In production: validate file type, size, and save to storage
        println!("Field: {}, File: {}, Size: {} bytes", name, file_name, data.len());
        // => "Field: avatar, File: profile.jpg, Size: 24576 bytes"

        // Security: validate file type by magic bytes, not extension
        let is_image = data.starts_with(b"\xFF\xD8") // => JPEG magic bytes
            || data.starts_with(b"\x89PNG");          // => PNG magic bytes

        if !is_image {
            return Err(StatusCode::UNPROCESSABLE_ENTITY);  // => 422 wrong file type
        }

        file_names.push(file_name);
        // => In production: save to S3/GCS/local disk here
    }

    Ok(format!("Uploaded: {:?}", file_names))
    // => "Uploaded: [\"profile.jpg\"]"
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/upload", post(upload_file));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

// Cargo.toml:
// axum = { version = "0.8", features = ["ws"] }

use axum::{
    extract::ws::{Message, WebSocket, WebSocketUpgrade},  // => WS types
    response::Response,
    routing::get,
    Router,
};
use futures::{sink::SinkExt, stream::StreamExt};  // => send/recv on split socket

#[tokio::main]
async fn main() {
    let app = Router::new().route("/ws", get(ws_handler));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

// The handler receives a WebSocketUpgrade extractor
async fn ws_handler(ws: WebSocketUpgrade) -> Response {
    ws.on_upgrade(handle_socket)  // => Upgrades HTTP → WebSocket
                                   // => handle_socket runs in a new task
}

// The actual WebSocket logic runs after upgrade
async fn handle_socket(mut socket: WebSocket) {
    // Echo server: receive message and send it back
    while let Some(msg) = socket.recv().await {  // => Wait for next message
        let msg = match msg {
            Ok(m) => m,                           // => Message received
            Err(_) => break,                      // => Connection error - close
        };

        match msg {
            Message::Text(text) => {
                println!("Received: {}", text);   // => "Received: hello"
                // Echo the message back to the client
                if socket.send(Message::Text(format!("Echo: {}", text))).await.is_err() {
                    break;  // => Client disconnected - stop the loop
                }
                // => Client receives: "Echo: hello"
            }
            Message::Close(_) => {
                println!("Client disconnected");  // => Graceful close
                break;
            }
            _ => {}  // => Binary, Ping, Pong - ignore for this echo server
        }
    }
}

use axum::{
    extract::{
        ws::{Message, WebSocket, WebSocketUpgrade},
        State,
    },
    response::Response,
    routing::get,
    Router,
};
use futures::{sink::SinkExt, stream::StreamExt};
use std::sync::Arc;
use tokio::sync::broadcast;  // => Multi-producer multi-consumer channel

#[derive(Clone)]
struct AppState {
    // broadcast::Sender can be cloned; each clone sends to all Receivers
    tx: broadcast::Sender<String>,  // => Channel for broadcasting messages
}

#[tokio::main]
async fn main() {
    let (tx, _rx) = broadcast::channel(100);  // => Buffer up to 100 messages
                                                // => _rx: initial receiver (dropped immediately)
    let state = Arc::new(AppState { tx });

    let app = Router::new()
        .route("/ws", get(ws_handler))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

async fn ws_handler(ws: WebSocketUpgrade, State(state): State<Arc<AppState>>) -> Response {
    ws.on_upgrade(move |socket| handle_socket(socket, state))
}

async fn handle_socket(socket: WebSocket, state: Arc<AppState>) {
    // Subscribe to the broadcast channel (each client gets its own receiver)
    let mut rx = state.tx.subscribe();  // => Creates a new receiver for this client
                                         // => Receives all messages sent AFTER this call

    // Split socket into writer and reader halves
    let (mut sender, mut receiver) = socket.split();
    // => sender: write messages to this client
    // => receiver: read messages from this client

    // Spawn a task to forward broadcast messages to this client
    let mut send_task = tokio::spawn(async move {
        while let Ok(msg) = rx.recv().await {      // => Wait for broadcast messages
            if sender.send(Message::Text(msg)).await.is_err() {
                break;  // => Client disconnected - stop forwarding
            }
        }
    });

    // Read messages from this client and broadcast to all others
    let tx = state.tx.clone();
    let mut recv_task = tokio::spawn(async move {
        while let Some(Ok(Message::Text(text))) = receiver.next().await {
            let _ = tx.send(text);  // => Broadcast to all subscribers
                                     // => Ignored if no receivers (all disconnected)
        }
    });

    // Wait for either task to finish (means client disconnected)
    tokio::select! {
        _ = &mut send_task => recv_task.abort(),   // => Send failed: abort recv
        _ = &mut recv_task => send_task.abort(),   // => Recv failed: abort send
    }
}

use axum::{
    response::sse::{Event, Sse},  // => SSE response type
    routing::get,
    Router,
};
use futures::stream::{self, Stream};
use std::{convert::Infallible, time::Duration};
use tokio_stream::StreamExt as _;  // => .throttle() for rate limiting the stream

#[tokio::main]
async fn main() {
    let app = Router::new().route("/sse", get(sse_handler));
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

// Returns an SSE stream that sends an event every second
async fn sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    // Create a stream of SSE events
    let stream = stream::iter(1u64..)          // => Infinite counter: 1, 2, 3, ...
        .map(|counter| {
            Event::default()                   // => Create an SSE event
                .event("update")               // => event: update (client uses this to filter)
                .data(format!("count={}", counter))  // => data: count=1
                                                      // => Client receives: event:update\ndata:count=1\n\n
        })
        .map(Ok::<_, Infallible>)              // => Wrap in Ok (stream never errors)
        .throttle(Duration::from_secs(1));     // => Emit one event per second

    Sse::new(stream)                           // => Wrap stream in Sse response
        .keep_alive(                           // => Sends : keep-alive comments
            axum::response::sse::KeepAlive::new()
                .interval(Duration::from_secs(15))  // => Every 15 seconds
                .text("keep-alive"),                 // => Comment text (client ignores)
        )
    // => Content-Type: text/event-stream
    // => Cache-Control: no-cache
    // => Connection: keep-alive
}

use axum::{routing::get, Router};
use tokio::signal;  // => Signal handling utilities

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/", get(handler))
        .route("/slow", get(slow_handler));

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();

    println!("Server started. Send SIGINT (Ctrl+C) or SIGTERM to shut down gracefully.");

    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal())  // => Stop accepting new connections on signal
                                                     // => Waits for in-flight requests to complete
        .await
        .unwrap();

    println!("Server shutdown complete.");  // => Only printed after all requests finish
}

// Signal handler: resolves when SIGINT (Ctrl+C) or SIGTERM is received
async fn shutdown_signal() {
    let ctrl_c = async {
        signal::ctrl_c()                             // => Listen for Ctrl+C (SIGINT)
            .await
            .expect("Failed to install CTRL+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        signal::unix::signal(signal::unix::SignalKind::terminate())
            // => Listen for SIGTERM (sent by Docker, Kubernetes on shutdown)
            .expect("Failed to install signal handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();  // => Windows has no SIGTERM

    // Wait for either signal
    tokio::select! {
        _ = ctrl_c => {},     // => SIGINT received
        _ = terminate => {},  // => SIGTERM received
    }

    println!("Shutdown signal received, draining connections...");
}

async fn handler() -> &'static str { "OK" }

async fn slow_handler() -> &'static str {
    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
    // => Simulates a slow request; graceful shutdown waits for this to complete
    "Slow response"
}

// Cargo.toml:
// tracing = "0.1"
// tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }

use axum::{extract::Path, routing::get, Router};
use tower_http::trace::TraceLayer;
use tracing::{info, instrument, warn};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};

#[tokio::main]
async fn main() {
    // Initialize structured JSON logging for production
    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| "my_app=debug,tower_http=debug".into()),
        )
        .with(tracing_subscriber::fmt::layer().json())
        // => .json(): Output logs as JSON lines (easy to ingest by log aggregators)
        // => Remove .json() for human-readable output in development
        .init();

    let app = Router::new()
        .route("/users/:id", get(get_user))
        .layer(TraceLayer::new_for_http());
        // => Each request gets a tracing span with method, uri, status, latency

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

// #[instrument] creates a span for this function, automatically propagating context
#[instrument(fields(user_id = %id))]  // => Adds user_id field to the span
async fn get_user(Path(id): Path<u64>) -> String {
    info!("Fetching user");  // => {"level":"INFO","span":{"user_id":"42"},"message":"Fetching user"}

    if id == 0 {
        warn!(id = id, "Invalid user ID requested");
        // => {"level":"WARN","span":{"user_id":"0"},"id":0,"message":"Invalid user ID"}
        return "Invalid".to_string();
    }

    let user = format!("User {}", id);
    info!(found = true, "User found");  // => Logs with additional structured field
    user
}