drasi_source_http/

lib.rs

// Copyright 2025 The Drasi Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#![allow(unexpected_cfgs)]

//! HTTP Source Plugin for Drasi
//!
//! This plugin exposes HTTP endpoints for receiving data change events. It supports
//! two mutually exclusive modes:
//!
//! - **Standard Mode**: Fixed-format `HttpSourceChange` endpoints with adaptive batching
//! - **Webhook Mode**: Configurable routes with template-based payload transformation
//!
//! # Standard Mode
//!
//! When no `webhooks` configuration is present, the source operates in standard mode
//! with the following endpoints:
//!
//! - **`POST /sources/{source_id}/events`** - Submit a single event
//! - **`POST /sources/{source_id}/events/batch`** - Submit multiple events
//! - **`GET /health`** - Health check endpoint
//!
//! ## Data Format
//!
//! Events are submitted as JSON using the `HttpSourceChange` format:
//!
//! ### Insert Operation
//!
//! ```json
//! {
//!     "operation": "insert",
//!     "element": {
//!         "type": "node",
//!         "id": "user-123",
//!         "labels": ["User"],
//!         "properties": {
//!             "name": "Alice",
//!             "email": "alice@example.com"
//!         }
//!     },
//!     "timestamp": 1699900000000000000
//! }
//! ```
//!
//! ### Update Operation
//!
//! ```json
//! {
//!     "operation": "update",
//!     "element": {
//!         "type": "node",
//!         "id": "user-123",
//!         "labels": ["User"],
//!         "properties": {
//!             "name": "Alice Updated"
//!         }
//!     }
//! }
//! ```
//!
//! ### Delete Operation
//!
//! ```json
//! {
//!     "operation": "delete",
//!     "id": "user-123",
//!     "labels": ["User"]
//! }
//! ```
//!
//! # Webhook Mode
//!
//! When a `webhooks` section is present in the configuration, the source operates
//! in webhook mode. This enables:
//!
//! - Custom routes with path parameters (e.g., `/github/events`, `/users/:id/hooks`)
//! - Multiple HTTP methods per route (POST, PUT, PATCH, DELETE, GET)
//! - Handlebars template-based payload transformation
//! - HMAC signature verification (GitHub, Shopify style)
//! - Bearer token authentication
//! - Support for JSON, XML, YAML, and plain text payloads
//!
//! ## Webhook Configuration Example
//!
//! ```yaml
//! webhooks:
//!   error_behavior: accept_and_log
//!   routes:
//!     - path: "/github/events"
//!       methods: ["POST"]
//!       auth:
//!         signature:
//!           type: hmac-sha256
//!           secret_env: GITHUB_WEBHOOK_SECRET
//!           header: X-Hub-Signature-256
//!           prefix: "sha256="
//!       error_behavior: reject
//!       mappings:
//!         - when:
//!             header: X-GitHub-Event
//!             equals: push
//!           operation: insert
//!           element_type: node
//!           effective_from: "{{payload.head_commit.timestamp}}"
//!           template:
//!             id: "commit-{{payload.head_commit.id}}"
//!             labels: ["Commit"]
//!             properties:
//!               message: "{{payload.head_commit.message}}"
//!               author: "{{payload.head_commit.author.name}}"
//! ```
//!
//! ## Relation Element
//!
//! ```json
//! {
//!     "operation": "insert",
//!     "element": {
//!         "type": "relation",
//!         "id": "follows-1",
//!         "labels": ["FOLLOWS"],
//!         "from": "user-123",
//!         "to": "user-456",
//!         "properties": {}
//!     }
//! }
//! ```
//!
//! # Batch Submission
//!
//! ```json
//! {
//!     "events": [
//!         { "operation": "insert", ... },
//!         { "operation": "update", ... }
//!     ]
//! }
//! ```
//!
//! # Adaptive Batching
//!
//! The HTTP source includes adaptive batching to optimize throughput. Events are
//! buffered and dispatched in batches, with batch size and timing adjusted based
//! on throughput patterns.
//!
//! | Parameter | Default | Description |
//! |-----------|---------|-------------|
//! | `adaptive_enabled` | `true` | Enable/disable adaptive batching |
//! | `adaptive_max_batch_size` | `1000` | Maximum events per batch |
//! | `adaptive_min_batch_size` | `1` | Minimum events per batch |
//! | `adaptive_max_wait_ms` | `100` | Maximum wait time before dispatching |
//! | `adaptive_min_wait_ms` | `10` | Minimum wait time between batches |
//!
//! # Configuration
//!
//! | Field | Type | Default | Description |
//! |-------|------|---------|-------------|
//! | `host` | string | *required* | Host address to bind to |
//! | `port` | u16 | `8080` | Port to listen on |
//! | `endpoint` | string | None | Optional custom path prefix |
//! | `timeout_ms` | u64 | `10000` | Request timeout in milliseconds |
//!
//! # Example Configuration (YAML)
//!
//! ```yaml
//! source_type: http
//! properties:
//!   host: "0.0.0.0"
//!   port: 8080
//!   adaptive_enabled: true
//!   adaptive_max_batch_size: 500
//! ```
//!
//! # Usage Examples
//!
//! ## Rust
//!
//! ```rust,no_run
//! use drasi_source_http::{HttpSource, HttpSourceBuilder};
//! use std::sync::Arc;
//!
//! # async fn example() -> anyhow::Result<()> {
//! let source = HttpSource::builder("http-source")
//!     .with_host("0.0.0.0")
//!     .with_port(8080)
//!     .with_adaptive_enabled(true)
//!     .build()?;
//! # Ok(())
//! # }
//! ```
//!
//! ## curl (Single Event)
//!
//! ```bash
//! curl -X POST http://localhost:8080/sources/my-source/events \
//!   -H "Content-Type: application/json" \
//!   -d '{"operation":"insert","element":{"type":"node","id":"1","labels":["Test"],"properties":{}}}'
//! ```
//!
//! ## curl (Batch)
//!
//! ```bash
//! curl -X POST http://localhost:8080/sources/my-source/events/batch \
//!   -H "Content-Type: application/json" \
//!   -d '{"events":[...]}'
//! ```

pub mod config;
pub mod descriptor;
pub use config::HttpSourceConfig;

mod adaptive_batcher;
mod models;
mod time;

// Webhook support modules
pub mod auth;
pub mod content_parser;
pub mod route_matcher;
pub mod template_engine;

// Export HTTP source models and conversion
pub use models::{convert_http_to_source_change, HttpElement, HttpSourceChange};

use anyhow::Result;
use async_trait::async_trait;
use axum::{
    body::Bytes,
    extract::{Path, State},
    http::{header, Method, StatusCode},
    response::IntoResponse,
    routing::{delete, get, post, put},
    Json, Router,
};
use log::{debug, error, info, trace, warn};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::timeout;
use tower_http::cors::{Any, CorsLayer};

use drasi_lib::channels::{ComponentType, *};
use drasi_lib::schema::{NodeSchema, PropertySchema, RelationSchema, SourceSchema};
use drasi_lib::sources::base::{SourceBase, SourceBaseParams};
use drasi_lib::wal::{WalError, WalProvider};
use drasi_lib::Source;
use tracing::Instrument;

use crate::adaptive_batcher::{AdaptiveBatchConfig, AdaptiveBatcher};
use crate::auth::{verify_auth, AuthResult};
use crate::config::{CorsConfig, ErrorBehavior, WebhookConfig};
use crate::content_parser::{parse_content, ContentType};
use crate::route_matcher::{convert_method, find_matching_mappings, headers_to_map, RouteMatcher};
use crate::template_engine::{TemplateContext, TemplateEngine};

/// Response for event submission
#[derive(Debug, Serialize, Deserialize)]
pub struct EventResponse {
    pub success: bool,
    pub message: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// HTTP source with configurable adaptive batching.
///
/// This source exposes HTTP endpoints for receiving data change events.
/// It supports both single-event and batch submission modes, with adaptive
/// batching for optimized throughput.
///
/// # Fields
///
/// - `base`: Common source functionality (dispatchers, status, lifecycle)
/// - `config`: HTTP-specific configuration (host, port, timeout)
/// - `adaptive_config`: Adaptive batching settings for throughput optimization
pub struct HttpSource {
    /// Base source implementation providing common functionality
    base: SourceBase,
    /// HTTP source configuration
    config: HttpSourceConfig,
    /// Adaptive batching configuration for throughput optimization
    adaptive_config: AdaptiveBatchConfig,
    /// WAL provider for durable event persistence (if durability is enabled)
    wal: tokio::sync::RwLock<Option<Arc<dyn WalProvider>>>,
    /// Handle to the WAL pruning background task (if running)
    prune_task: tokio::sync::RwLock<Option<tokio::task::JoinHandle<()>>>,
}

/// Batch event request that can accept multiple events
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventRequest {
    pub events: Vec<HttpSourceChange>,
}

/// HTTP source app state with batching channel.
///
/// Shared state passed to Axum route handlers.
#[derive(Clone)]
struct HttpAppState {
    /// The source ID for validation against incoming requests
    source_id: String,
    /// Channel for sending events to the adaptive batcher
    batch_tx: mpsc::Sender<SourceChangeEvent>,
    /// Webhook configuration (if in webhook mode)
    webhook_config: Option<Arc<WebhookState>>,
    /// WAL provider for durable persistence (if durability is enabled)
    wal: Option<Arc<dyn WalProvider>>,
}

/// State for webhook mode processing
struct WebhookState {
    /// Webhook configuration
    config: WebhookConfig,
    /// Route matcher for incoming requests
    route_matcher: RouteMatcher,
    /// Template engine for payload transformation
    template_engine: TemplateEngine,
}

fn extract_property_schemas(properties: Option<&serde_json::Value>) -> Vec<PropertySchema> {
    match properties {
        Some(serde_json::Value::Object(map)) => map
            .keys()
            .map(|key| PropertySchema::new(key.clone()))
            .collect(),
        _ => Vec::new(),
    }
}

fn derive_schema_from_webhooks(webhooks: &WebhookConfig) -> Option<SourceSchema> {
    let mut node_map: HashMap<String, Vec<PropertySchema>> = HashMap::new();
    let mut relation_map: HashMap<String, Vec<PropertySchema>> = HashMap::new();

    for route in &webhooks.routes {
        for mapping in &route.mappings {
            let Some(label) = mapping.template.labels.first().cloned() else {
                continue;
            };

            let properties = extract_property_schemas(mapping.template.properties.as_ref());

            match mapping.element_type {
                crate::config::ElementType::Node => {
                    let entry = node_map.entry(label).or_default();
                    for prop in properties {
                        if !entry.iter().any(|p| p.name == prop.name) {
                            entry.push(prop);
                        }
                    }
                }
                crate::config::ElementType::Relation => {
                    let entry = relation_map.entry(label).or_default();
                    for prop in properties {
                        if !entry.iter().any(|p| p.name == prop.name) {
                            entry.push(prop);
                        }
                    }
                }
            }
        }
    }

    let nodes: Vec<_> = node_map
        .into_iter()
        .map(|(label, properties)| NodeSchema { label, properties })
        .collect();
    let relations: Vec<_> = relation_map
        .into_iter()
        .map(|(label, properties)| RelationSchema {
            label,
            from: None,
            to: None,
            properties,
        })
        .collect();

    if nodes.is_empty() && relations.is_empty() {
        None
    } else {
        Some(SourceSchema { nodes, relations })
    }
}

impl HttpSource {
    /// Create a new HTTP source.
    ///
    /// The event channel is automatically injected when the source is added
    /// to DrasiLib via `add_source()`.
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for this source instance
    /// * `config` - HTTP source configuration
    ///
    /// # Returns
    ///
    /// A new `HttpSource` instance, or an error if construction fails.
    ///
    /// # Errors
    ///
    /// Returns an error if the base source cannot be initialized.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use drasi_source_http::HttpSource;
    ///
    /// # fn example() -> anyhow::Result<()> {
    /// let source = HttpSource::builder("my-http-source")
    ///     .with_host("0.0.0.0")
    ///     .with_port(8080)
    ///     .build()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn new(id: impl Into<String>, config: HttpSourceConfig) -> Result<Self> {
        let id = id.into();
        let params = SourceBaseParams::new(id);

        // Configure adaptive batching
        let mut adaptive_config = AdaptiveBatchConfig::default();

        // Allow overriding adaptive parameters from config
        if let Some(max_batch) = config.adaptive_max_batch_size {
            adaptive_config.max_batch_size = max_batch;
        }
        if let Some(min_batch) = config.adaptive_min_batch_size {
            adaptive_config.min_batch_size = min_batch;
        }
        if let Some(max_wait_ms) = config.adaptive_max_wait_ms {
            adaptive_config.max_wait_time = Duration::from_millis(max_wait_ms);
        }
        if let Some(min_wait_ms) = config.adaptive_min_wait_ms {
            adaptive_config.min_wait_time = Duration::from_millis(min_wait_ms);
        }
        if let Some(window_secs) = config.adaptive_window_secs {
            adaptive_config.throughput_window = Duration::from_secs(window_secs);
        }
        if let Some(enabled) = config.adaptive_enabled {
            adaptive_config.adaptive_enabled = enabled;
        }

        Ok(Self {
            base: SourceBase::new(params)?,
            config,
            adaptive_config,
            wal: tokio::sync::RwLock::new(None),
            prune_task: tokio::sync::RwLock::new(None),
        })
    }

    /// Create a new HTTP source with custom dispatch settings.
    ///
    /// The event channel is automatically injected when the source is added
    /// to DrasiLib via `add_source()`.
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for this source instance
    /// * `config` - HTTP source configuration
    /// * `dispatch_mode` - Optional dispatch mode (Channel, Direct, etc.)
    /// * `dispatch_buffer_capacity` - Optional buffer capacity for channel dispatch
    ///
    /// # Returns
    ///
    /// A new `HttpSource` instance with custom dispatch settings.
    ///
    /// # Errors
    ///
    /// Returns an error if the base source cannot be initialized.
    pub fn with_dispatch(
        id: impl Into<String>,
        config: HttpSourceConfig,
        dispatch_mode: Option<DispatchMode>,
        dispatch_buffer_capacity: Option<usize>,
    ) -> Result<Self> {
        let id = id.into();
        let mut params = SourceBaseParams::new(id);
        if let Some(mode) = dispatch_mode {
            params = params.with_dispatch_mode(mode);
        }
        if let Some(capacity) = dispatch_buffer_capacity {
            params = params.with_dispatch_buffer_capacity(capacity);
        }

        let mut adaptive_config = AdaptiveBatchConfig::default();

        if let Some(max_batch) = config.adaptive_max_batch_size {
            adaptive_config.max_batch_size = max_batch;
        }
        if let Some(min_batch) = config.adaptive_min_batch_size {
            adaptive_config.min_batch_size = min_batch;
        }
        if let Some(max_wait_ms) = config.adaptive_max_wait_ms {
            adaptive_config.max_wait_time = Duration::from_millis(max_wait_ms);
        }
        if let Some(min_wait_ms) = config.adaptive_min_wait_ms {
            adaptive_config.min_wait_time = Duration::from_millis(min_wait_ms);
        }
        if let Some(window_secs) = config.adaptive_window_secs {
            adaptive_config.throughput_window = Duration::from_secs(window_secs);
        }
        if let Some(enabled) = config.adaptive_enabled {
            adaptive_config.adaptive_enabled = enabled;
        }

        Ok(Self {
            base: SourceBase::new(params)?,
            config,
            adaptive_config,
            wal: tokio::sync::RwLock::new(None),
            prune_task: tokio::sync::RwLock::new(None),
        })
    }

    /// Handle a single event submission from `POST /sources/{source_id}/events`.
    ///
    /// Validates the source ID matches this source and converts the HTTP event
    /// to a source change before sending to the adaptive batcher.
    async fn handle_single_event(
        Path(source_id): Path<String>,
        State(state): State<HttpAppState>,
        Json(event): Json<HttpSourceChange>,
    ) -> Result<impl IntoResponse, (StatusCode, Json<EventResponse>)> {
        debug!("[{source_id}] HTTP endpoint received single event: {event:?}");
        Self::process_events(&source_id, &state, vec![event]).await
    }

    /// Handle a batch event submission from `POST /sources/{source_id}/events/batch`.
    ///
    /// Validates the source ID and processes all events in the batch,
    /// returning partial success if some events fail.
    async fn handle_batch_events(
        Path(source_id): Path<String>,
        State(state): State<HttpAppState>,
        Json(batch): Json<BatchEventRequest>,
    ) -> Result<impl IntoResponse, (StatusCode, Json<EventResponse>)> {
        debug!(
            "[{}] HTTP endpoint received batch of {} events",
            source_id,
            batch.events.len()
        );
        Self::process_events(&source_id, &state, batch.events).await
    }

    /// Process a list of events, converting and sending them to the batcher.
    ///
    /// # Arguments
    ///
    /// * `source_id` - Source ID from the request path
    /// * `state` - Shared app state containing the batch channel
    /// * `events` - List of HTTP source changes to process
    ///
    /// # Returns
    ///
    /// Success response with count of processed events, or error if all fail.
    async fn process_events(
        source_id: &str,
        state: &HttpAppState,
        events: Vec<HttpSourceChange>,
    ) -> Result<impl IntoResponse, (StatusCode, Json<EventResponse>)> {
        trace!("[{}] Processing {} events", source_id, events.len());

        if source_id != state.source_id {
            error!(
                "[{}] Source name mismatch. Expected '{}', got '{}'",
                state.source_id, state.source_id, source_id
            );
            return Err((
                StatusCode::BAD_REQUEST,
                Json(EventResponse {
                    success: false,
                    message: "Source name mismatch".to_string(),
                    error: Some(format!(
                        "Expected source '{}', got '{}'",
                        state.source_id, source_id
                    )),
                }),
            ));
        }

        let mut success_count = 0;
        let mut error_count = 0;
        let mut last_error = None;

        for (idx, event) in events.iter().enumerate() {
            match convert_http_to_source_change(event, source_id) {
                Ok(source_change) => {
                    // WAL append before ACK (if durability enabled)
                    let sequence = if let Some(ref wal) = state.wal {
                        match wal.append(&state.source_id, &source_change).await {
                            Ok(seq) => {
                                trace!(
                                    "[{}] WAL append succeeded: sequence={}",
                                    state.source_id,
                                    seq
                                );
                                Some(seq)
                            }
                            Err(WalError::CapacityExhausted(_)) => {
                                warn!(
                                    "[{}] WAL capacity exhausted, rejecting event",
                                    state.source_id
                                );
                                return Err((
                                    StatusCode::SERVICE_UNAVAILABLE,
                                    Json(EventResponse {
                                        success: false,
                                        message: "WAL capacity exhausted".to_string(),
                                        error: Some("Source durability buffer is full".to_string()),
                                    }),
                                ));
                            }
                            Err(e) => {
                                error!(
                                    "[{}] WAL append failed for event {}: {}",
                                    state.source_id,
                                    idx + 1,
                                    e
                                );
                                // WAL durability failure is a server-side error — reject entire batch
                                return Err((
                                    StatusCode::SERVICE_UNAVAILABLE,
                                    Json(EventResponse {
                                        success: false,
                                        message: format!(
                                            "WAL durability failure at event {}: {e}",
                                            idx + 1
                                        ),
                                        error: Some(format!("WAL error: {e}")),
                                    }),
                                ));
                            }
                        }
                    } else {
                        None
                    };

                    let change_event = SourceChangeEvent {
                        source_id: source_id.to_string(),
                        change: source_change,
                        timestamp: chrono::Utc::now(),
                        sequence,
                    };

                    if let Err(e) = state.batch_tx.send(change_event).await {
                        error!(
                            "[{}] Failed to send event {} to batch channel: {}",
                            state.source_id,
                            idx + 1,
                            e
                        );
                        error_count += 1;
                        last_error = Some("Internal channel error".to_string());
                    } else {
                        success_count += 1;
                    }
                }
                Err(e) => {
                    error!(
                        "[{}] Failed to convert event {}: {}",
                        state.source_id,
                        idx + 1,
                        e
                    );
                    error_count += 1;
                    last_error = Some(e.to_string());
                }
            }
        }

        debug!(
            "[{source_id}] Event processing complete: {success_count} succeeded, {error_count} failed"
        );

        if error_count > 0 && success_count == 0 {
            Err((
                StatusCode::BAD_REQUEST,
                Json(EventResponse {
                    success: false,
                    message: format!("All {error_count} events failed"),
                    error: last_error,
                }),
            ))
        } else if error_count > 0 {
            Ok(Json(EventResponse {
                success: true,
                message: format!(
                    "Processed {success_count} events successfully, {error_count} failed"
                ),
                error: last_error,
            }))
        } else {
            Ok(Json(EventResponse {
                success: true,
                message: format!("All {success_count} events processed successfully"),
                error: None,
            }))
        }
    }

    async fn health_check() -> impl IntoResponse {
        Json(serde_json::json!({
            "status": "healthy",
            "service": "http-source",
            "features": ["adaptive-batching", "batch-endpoint", "webhooks"]
        }))
    }

    /// Handle webhook requests
    ///
    /// This handler processes requests in webhook mode, matching against
    /// configured routes and transforming payloads using templates.
    async fn handle_webhook(
        method: axum::http::Method,
        uri: axum::http::Uri,
        headers: axum::http::HeaderMap,
        State(state): State<HttpAppState>,
        body: Bytes,
    ) -> impl IntoResponse {
        let path = uri.path();
        let source_id = &state.source_id;

        debug!("[{source_id}] Webhook received: {method} {path}");

        // Get webhook state - should always be present in webhook mode
        let webhook_state = match &state.webhook_config {
            Some(ws) => ws,
            None => {
                error!("[{source_id}] Webhook handler called but no webhook config present");
                return (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    Json(EventResponse {
                        success: false,
                        message: "Internal configuration error".to_string(),
                        error: Some("Webhook mode not properly configured".to_string()),
                    }),
                );
            }
        };

        // Convert method
        let http_method = match convert_method(&method) {
            Some(m) => m,
            None => {
                return handle_error(
                    &webhook_state.config.error_behavior,
                    source_id,
                    StatusCode::METHOD_NOT_ALLOWED,
                    "Method not supported",
                    None,
                );
            }
        };

        // Match route
        let route_match = match webhook_state.route_matcher.match_route(
            path,
            &http_method,
            &webhook_state.config.routes,
        ) {
            Some(rm) => rm,
            None => {
                debug!("[{source_id}] No matching route for {method} {path}");
                return handle_error(
                    &webhook_state.config.error_behavior,
                    source_id,
                    StatusCode::NOT_FOUND,
                    "No matching route",
                    None,
                );
            }
        };

        let route = route_match.route;
        let error_behavior = route
            .error_behavior
            .as_ref()
            .unwrap_or(&webhook_state.config.error_behavior);

        // Verify authentication
        let auth_result = verify_auth(route.auth.as_ref(), &headers, &body);
        if let AuthResult::Failed(reason) = auth_result {
            warn!("[{source_id}] Authentication failed for {path}: {reason}");
            return handle_error(
                error_behavior,
                source_id,
                StatusCode::UNAUTHORIZED,
                "Authentication failed",
                Some(&reason),
            );
        }

        // Parse content
        let content_type = ContentType::from_header(
            headers
                .get(axum::http::header::CONTENT_TYPE)
                .and_then(|v| v.to_str().ok()),
        );

        let payload = match parse_content(&body, content_type) {
            Ok(p) => p,
            Err(e) => {
                warn!("[{source_id}] Failed to parse payload: {e}");
                return handle_error(
                    error_behavior,
                    source_id,
                    StatusCode::BAD_REQUEST,
                    "Failed to parse payload",
                    Some(&e.to_string()),
                );
            }
        };

        // Build template context
        let headers_map = headers_to_map(&headers);
        let query_map = parse_query_string(uri.query());

        let context = TemplateContext {
            payload: payload.clone(),
            route: route_match.path_params,
            query: query_map,
            headers: headers_map.clone(),
            method: method.to_string(),
            path: path.to_string(),
            source_id: source_id.clone(),
        };

        // Find matching mappings
        let matching_mappings = find_matching_mappings(&route.mappings, &headers_map, &payload);

        if matching_mappings.is_empty() {
            debug!("[{source_id}] No matching mappings for request");
            return handle_error(
                error_behavior,
                source_id,
                StatusCode::BAD_REQUEST,
                "No matching mapping for request",
                None,
            );
        }

        // Process each matching mapping
        let mut success_count = 0;
        let mut error_count = 0;
        let mut last_error = None;

        for mapping in matching_mappings {
            match webhook_state
                .template_engine
                .process_mapping(mapping, &context, source_id)
            {
                Ok(source_change) => {
                    // WAL append before ACK (if durability enabled)
                    let sequence = if let Some(ref wal) = state.wal {
                        match wal.append(&state.source_id, &source_change).await {
                            Ok(seq) => Some(seq),
                            Err(WalError::CapacityExhausted(_)) => {
                                return handle_error(
                                    error_behavior,
                                    source_id,
                                    StatusCode::SERVICE_UNAVAILABLE,
                                    "WAL capacity exhausted",
                                    None,
                                );
                            }
                            Err(e) => {
                                error!("[{source_id}] WAL append failed: {e}");
                                // WAL durability failure — reject to maintain at-least-once guarantee
                                return handle_error(
                                    error_behavior,
                                    source_id,
                                    StatusCode::SERVICE_UNAVAILABLE,
                                    &format!("WAL durability failure: {e}"),
                                    None,
                                );
                            }
                        }
                    } else {
                        None
                    };

                    let event = SourceChangeEvent {
                        source_id: source_id.clone(),
                        change: source_change,
                        timestamp: chrono::Utc::now(),
                        sequence,
                    };

                    if let Err(e) = state.batch_tx.send(event).await {
                        error!("[{source_id}] Failed to send event to batcher: {e}");
                        error_count += 1;
                        last_error = Some(format!("Failed to queue event: {e}"));
                    } else {
                        success_count += 1;
                    }
                }
                Err(e) => {
                    warn!("[{source_id}] Failed to process mapping: {e}");
                    error_count += 1;
                    last_error = Some(e.to_string());
                }
            }
        }

        debug!("[{source_id}] Webhook processing complete: {success_count} succeeded, {error_count} failed");

        if error_count > 0 && success_count == 0 {
            handle_error(
                error_behavior,
                source_id,
                StatusCode::BAD_REQUEST,
                &format!("All {error_count} mappings failed"),
                last_error.as_deref(),
            )
        } else if error_count > 0 {
            (
                StatusCode::OK,
                Json(EventResponse {
                    success: true,
                    message: format!("Processed {success_count} events, {error_count} failed"),
                    error: last_error,
                }),
            )
        } else {
            (
                StatusCode::OK,
                Json(EventResponse {
                    success: true,
                    message: format!("Processed {success_count} events successfully"),
                    error: None,
                }),
            )
        }
    }

    async fn run_adaptive_batcher(
        batch_rx: mpsc::Receiver<SourceChangeEvent>,
        dispatchers: Arc<
            tokio::sync::RwLock<
                Vec<
                    Box<
                        dyn drasi_lib::channels::ChangeDispatcher<SourceEventWrapper> + Send + Sync,
                    >,
                >,
            >,
        >,
        adaptive_config: AdaptiveBatchConfig,
        source_id: String,
    ) {
        let mut batcher = AdaptiveBatcher::new(batch_rx, adaptive_config.clone());
        let mut total_events = 0u64;
        let mut total_batches = 0u64;

        info!("[{source_id}] Adaptive HTTP batcher started with config: {adaptive_config:?}");

        while let Some(batch) = batcher.next_batch().await {
            if batch.is_empty() {
                debug!("[{source_id}] Batcher received empty batch, skipping");
                continue;
            }

            let batch_size = batch.len();
            total_events += batch_size as u64;
            total_batches += 1;

            debug!(
                "[{source_id}] Batcher forwarding batch #{total_batches} with {batch_size} events to dispatchers"
            );

            let mut sent_count = 0;
            let mut failed_count = 0;
            for (idx, event) in batch.into_iter().enumerate() {
                debug!(
                    "[{}] Batch #{}, dispatching event {}/{}",
                    source_id,
                    total_batches,
                    idx + 1,
                    batch_size
                );

                let mut profiling = drasi_lib::profiling::ProfilingMetadata::new();
                profiling.source_send_ns = Some(drasi_lib::profiling::timestamp_ns());

                let mut wrapper = SourceEventWrapper::with_profiling(
                    event.source_id.clone(),
                    SourceEvent::Change(event.change),
                    event.timestamp,
                    profiling,
                );

                // Carry WAL-assigned sequence through to the wrapper
                if let Some(seq) = event.sequence {
                    wrapper.sequence = Some(seq);
                    wrapper.source_position = Some(bytes::Bytes::from(seq.to_be_bytes().to_vec()));
                }

                if let Err(e) =
                    SourceBase::dispatch_from_task(dispatchers.clone(), wrapper.clone(), &source_id)
                        .await
                {
                    error!(
                        "[{}] Batch #{}, failed to dispatch event {}/{} (no subscribers): {}",
                        source_id,
                        total_batches,
                        idx + 1,
                        batch_size,
                        e
                    );
                    failed_count += 1;
                } else {
                    debug!(
                        "[{}] Batch #{}, successfully dispatched event {}/{}",
                        source_id,
                        total_batches,
                        idx + 1,
                        batch_size
                    );
                    sent_count += 1;
                }
            }

            debug!(
                "[{source_id}] Batch #{total_batches} complete: {sent_count} dispatched, {failed_count} failed"
            );

            if total_batches.is_multiple_of(100) {
                info!(
                    "[{}] Adaptive HTTP metrics - Batches: {}, Events: {}, Avg batch size: {:.1}",
                    source_id,
                    total_batches,
                    total_events,
                    total_events as f64 / total_batches as f64
                );
            }
        }

        info!(
            "[{source_id}] Adaptive HTTP batcher stopped - Total batches: {total_batches}, Total events: {total_events}"
        );
    }
}

#[async_trait]
impl Source for HttpSource {
    fn id(&self) -> &str {
        &self.base.id
    }

    fn type_name(&self) -> &str {
        "http"
    }

    fn properties(&self) -> HashMap<String, serde_json::Value> {
        use crate::descriptor::HttpSourceConfigDto;

        self.base
            .properties_or_serialize(&HttpSourceConfigDto::from(&self.config))
    }

    fn auto_start(&self) -> bool {
        self.base.get_auto_start()
    }

    fn describe_schema(&self) -> Option<SourceSchema> {
        self.config
            .webhooks
            .as_ref()
            .and_then(derive_schema_from_webhooks)
    }

    async fn start(&self) -> Result<()> {
        info!("[{}] Starting adaptive HTTP source", self.base.id);

        self.base
            .set_status(
                ComponentStatus::Starting,
                Some("Starting adaptive HTTP source".to_string()),
            )
            .await;

        // Initialize WAL if durability is enabled
        let wal_ref: Option<Arc<dyn WalProvider>> =
            if self.config.durability.as_ref().is_some_and(|d| d.enabled) {
                let ctx = self
                    .base
                    .context()
                    .await
                    .ok_or_else(|| anyhow::anyhow!("Context not initialized"))?;
                let wal = ctx.wal_provider.clone().ok_or_else(|| {
                    anyhow::anyhow!("Durability enabled but no WAL provider configured on DrasiLib")
                })?;
                let wal_config = self
                    .config
                    .durability
                    .as_ref()
                    .expect("durability checked above")
                    .to_wal_config();
                wal.register(&self.base.id, wal_config.clone())
                    .await
                    .map_err(|e| {
                        anyhow::anyhow!(
                            "Failed to register WAL for source '{}': {}",
                            self.base.id,
                            e
                        )
                    })?;

                info!(
                    "[{}] WAL registered: max_events={}, policy={:?}",
                    self.base.id, wal_config.max_events, wal_config.capacity_policy
                );

                // Resume sequence counter from WAL head
                let head = wal.head_sequence(&self.base.id).await.unwrap_or(0);
                if head > 0 {
                    self.base.set_next_sequence(head);
                    info!(
                        "[{}] WAL resumed from persisted state: head={}, next_sequence={}",
                        self.base.id,
                        head,
                        head + 1
                    );
                }

                *self.wal.write().await = Some(wal.clone());
                Some(wal)
            } else {
                None
            };

        let host = self.config.host.clone();
        let port = self.config.port;

        // Create batch channel with capacity based on batch configuration
        let batch_channel_capacity = self.adaptive_config.recommended_channel_capacity();
        let (batch_tx, batch_rx) = mpsc::channel(batch_channel_capacity);
        info!(
            "[{}] HttpSource using batch channel capacity: {} (max_batch_size: {} x 5)",
            self.base.id, batch_channel_capacity, self.adaptive_config.max_batch_size
        );

        // Start adaptive batcher task
        let adaptive_config = self.adaptive_config.clone();
        let source_id = self.base.id.clone();
        let dispatchers = self.base.dispatchers.clone();

        // Get instance_id from context for log routing isolation
        let instance_id = self
            .base
            .context()
            .await
            .map(|c| c.instance_id)
            .unwrap_or_default();

        info!("[{source_id}] Starting adaptive batcher task");
        let source_id_for_span = source_id.clone();
        let span = tracing::info_span!(
            "http_adaptive_batcher",
            instance_id = %instance_id,
            component_id = %source_id_for_span,
            component_type = "source"
        );
        tokio::spawn(
            async move {
                Self::run_adaptive_batcher(
                    batch_rx,
                    dispatchers,
                    adaptive_config,
                    source_id.clone(),
                )
                .await
            }
            .instrument(span),
        );

        // Create app state
        let webhook_state = if let Some(ref webhook_config) = self.config.webhooks {
            info!(
                "[{}] Webhook mode enabled with {} routes",
                self.base.id,
                webhook_config.routes.len()
            );
            Some(Arc::new(WebhookState {
                config: webhook_config.clone(),
                route_matcher: RouteMatcher::new(&webhook_config.routes),
                template_engine: TemplateEngine::new(),
            }))
        } else {
            info!("[{}] Standard mode enabled", self.base.id);
            None
        };

        let state = HttpAppState {
            source_id: self.base.id.clone(),
            batch_tx,
            webhook_config: webhook_state,
            wal: wal_ref.clone(),
        };

        // Build router based on mode
        let app = if self.config.is_webhook_mode() {
            // Webhook mode: only health check + catch-all webhook handler
            let router = Router::new()
                .route("/health", get(Self::health_check))
                .fallback(Self::handle_webhook)
                .with_state(state);

            // Apply CORS if configured
            if let Some(ref webhooks) = self.config.webhooks {
                if let Some(ref cors_config) = webhooks.cors {
                    if cors_config.enabled {
                        info!("[{}] CORS enabled for webhook endpoints", self.base.id);
                        router.layer(build_cors_layer(cors_config))
                    } else {
                        router
                    }
                } else {
                    router
                }
            } else {
                router
            }
        } else {
            // Standard mode: original endpoints
            Router::new()
                .route("/health", get(Self::health_check))
                .route(
                    "/sources/:source_id/events",
                    post(Self::handle_single_event),
                )
                .route(
                    "/sources/:source_id/events/batch",
                    post(Self::handle_batch_events),
                )
                .with_state(state)
        };

        // Create shutdown channel
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();

        let host_clone = host.clone();

        // Start server
        let (error_tx, error_rx) = tokio::sync::oneshot::channel();
        let source_id = self.base.id.clone();
        let source_id_for_span = source_id.clone();
        let span = tracing::info_span!(
            "http_source_server",
            instance_id = %instance_id,
            component_id = %source_id_for_span,
            component_type = "source"
        );
        let server_handle = tokio::spawn(
            async move {
                let addr = format!("{host}:{port}");
                info!("[{source_id}] Adaptive HTTP source attempting to bind to {addr}");

                let listener = match tokio::net::TcpListener::bind(&addr).await {
                    Ok(listener) => {
                        info!("[{source_id}] Adaptive HTTP source successfully listening on {addr}");
                        listener
                    }
                    Err(e) => {
                        error!("[{source_id}] Failed to bind HTTP server to {addr}: {e}");
                        let _ = error_tx.send(format!(
                        "Failed to bind HTTP server to {addr}: {e}. Common causes: port already in use, insufficient permissions"
                    ));
                    return;
                }
            };

                if let Err(e) = axum::serve(listener, app)
                    .with_graceful_shutdown(async move {
                        let _ = shutdown_rx.await;
                    })
                    .await
                {
                    error!("[{source_id}] HTTP server error: {e}");
                }
            }
            .instrument(span),
        );

        *self.base.task_handle.write().await = Some(server_handle);
        *self.base.shutdown_tx.write().await = Some(shutdown_tx);

        // Check for startup errors with a short timeout
        match timeout(Duration::from_millis(500), error_rx).await {
            Ok(Ok(error_msg)) => {
                self.base.set_status(ComponentStatus::Error, None).await;
                return Err(anyhow::anyhow!("{error_msg}"));
            }
            _ => {
                self.base
                    .set_status(
                        ComponentStatus::Running,
                        Some(format!(
                    "Adaptive HTTP source running on {host_clone}:{port} with batch support"
                )),
                    )
                    .await;
            }
        }

        // Spawn WAL pruning task if durability is enabled
        if let Some(wal) = wal_ref {
            let base = self.base.clone_shared();
            let source_id = self.base.id.clone();
            let prune_handle = tokio::spawn(async move {
                let mut interval = tokio::time::interval(Duration::from_secs(30));
                loop {
                    interval.tick().await;
                    if let Some(confirmed) = base.compute_confirmed_position().await {
                        if confirmed > 0 {
                            match wal.prune_up_to(&source_id, confirmed).await {
                                Ok(pruned) => {
                                    if pruned > 0 {
                                        let remaining =
                                            wal.event_count(&source_id).await.unwrap_or(0);
                                        debug!(
                                            "[{source_id}] WAL pruned: count={pruned}, confirmed_seq={confirmed}, remaining={remaining}"
                                        );
                                    }
                                }
                                Err(e) => {
                                    warn!("[{source_id}] WAL prune failed: {e}");
                                }
                            }
                        }
                    }
                }
            });
            *self.prune_task.write().await = Some(prune_handle);
        }

        Ok(())
    }

    async fn stop(&self) -> Result<()> {
        info!("[{}] Stopping adaptive HTTP source", self.base.id);

        self.base
            .set_status(
                ComponentStatus::Stopping,
                Some("Stopping adaptive HTTP source".to_string()),
            )
            .await;

        // Cancel WAL pruning task
        if let Some(handle) = self.prune_task.write().await.take() {
            handle.abort();
        }

        if let Some(tx) = self.base.shutdown_tx.write().await.take() {
            let _ = tx.send(());
        }

        if let Some(handle) = self.base.task_handle.write().await.take() {
            let _ = timeout(Duration::from_secs(5), handle).await;
        }

        self.base
            .set_status(
                ComponentStatus::Stopped,
                Some("Adaptive HTTP source stopped".to_string()),
            )
            .await;

        Ok(())
    }

    async fn status(&self) -> ComponentStatus {
        self.base.get_status().await
    }

    async fn subscribe(
        &self,
        settings: drasi_lib::config::SourceSubscriptionSettings,
    ) -> Result<SubscriptionResponse> {
        // If WAL is enabled and subscriber is resuming, use WAL replay
        let wal_guard = self.wal.read().await;
        if let (Some(wal), Some(ref resume_from)) = (wal_guard.as_ref(), &settings.resume_from) {
            // Decode resume_from as big-endian u64 sequence
            if resume_from.len() >= 8 {
                let resume_seq =
                    u64::from_be_bytes(resume_from[..8].try_into().unwrap_or_default());
                let wal_clone = wal.clone();
                drop(wal_guard);
                return self
                    .base
                    .subscribe_with_replay(&settings, wal_clone.as_ref(), resume_seq, "HTTP")
                    .await;
            } else {
                drop(wal_guard);
                return Err(anyhow::anyhow!(
                    "Invalid resume_from position: expected at least 8 bytes, got {}",
                    resume_from.len()
                ));
            }
        }
        drop(wal_guard);
        self.base.subscribe_with_bootstrap(&settings, "HTTP").await
    }

    fn supports_replay(&self) -> bool {
        self.config.durability.as_ref().is_some_and(|d| d.enabled)
    }

    async fn deprovision(&self) -> Result<()> {
        // Delete WAL data if durability was enabled
        let wal_guard = self.wal.read().await;
        if let Some(ref wal) = *wal_guard {
            info!("[{}] Deprovisioning: deleting WAL data", self.base.id);
            if let Err(e) = wal.delete_wal(&self.base.id).await {
                warn!(
                    "[{}] Failed to delete WAL during deprovision: {}",
                    self.base.id, e
                );
            }
        }
        drop(wal_guard);
        Ok(())
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    async fn initialize(&self, context: drasi_lib::context::SourceRuntimeContext) {
        self.base.initialize(context).await;
    }

    async fn set_bootstrap_provider(
        &self,
        provider: Box<dyn drasi_lib::bootstrap::BootstrapProvider + 'static>,
    ) {
        self.base.set_bootstrap_provider(provider).await;
    }
}

/// Builder for HttpSource instances.
///
/// Provides a fluent API for constructing HTTP sources with sensible defaults
/// and adaptive batching settings. The builder takes the source ID at construction
/// and returns a fully constructed `HttpSource` from `build()`.
///
/// # Example
///
/// ```rust,no_run
/// use drasi_source_http::HttpSource;
///
/// # fn example() -> anyhow::Result<()> {
/// let source = HttpSource::builder("my-source")
///     .with_host("0.0.0.0")
///     .with_port(8080)
///     .with_adaptive_enabled(true)
///     .build()?;
/// # Ok(())
/// # }
/// ```
pub struct HttpSourceBuilder {
    id: String,
    host: String,
    port: u16,
    endpoint: Option<String>,
    timeout_ms: u64,
    adaptive_max_batch_size: Option<usize>,
    adaptive_min_batch_size: Option<usize>,
    adaptive_max_wait_ms: Option<u64>,
    adaptive_min_wait_ms: Option<u64>,
    adaptive_window_secs: Option<u64>,
    adaptive_enabled: Option<bool>,
    webhooks: Option<WebhookConfig>,
    durability: Option<drasi_lib::DurabilityConfig>,
    dispatch_mode: Option<DispatchMode>,
    dispatch_buffer_capacity: Option<usize>,
    bootstrap_provider: Option<Box<dyn drasi_lib::bootstrap::BootstrapProvider + 'static>>,
    auto_start: bool,
}

impl HttpSourceBuilder {
    /// Create a new HTTP source builder with the given source ID.
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for the source instance
    pub fn new(id: impl Into<String>) -> Self {
        Self {
            id: id.into(),
            host: String::new(),
            port: 8080,
            endpoint: None,
            timeout_ms: 10000,
            adaptive_max_batch_size: None,
            adaptive_min_batch_size: None,
            adaptive_max_wait_ms: None,
            adaptive_min_wait_ms: None,
            adaptive_window_secs: None,
            adaptive_enabled: None,
            webhooks: None,
            durability: None,
            dispatch_mode: None,
            dispatch_buffer_capacity: None,
            bootstrap_provider: None,
            auto_start: true,
        }
    }

    /// Set the HTTP host
    pub fn with_host(mut self, host: impl Into<String>) -> Self {
        self.host = host.into();
        self
    }

    /// Set the HTTP port
    pub fn with_port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }

    /// Set the endpoint path
    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
        self.endpoint = Some(endpoint.into());
        self
    }

    /// Set the request timeout in milliseconds
    pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Self {
        self.timeout_ms = timeout_ms;
        self
    }

    /// Set the adaptive batching maximum batch size
    pub fn with_adaptive_max_batch_size(mut self, size: usize) -> Self {
        self.adaptive_max_batch_size = Some(size);
        self
    }

    /// Set the adaptive batching minimum batch size
    pub fn with_adaptive_min_batch_size(mut self, size: usize) -> Self {
        self.adaptive_min_batch_size = Some(size);
        self
    }

    /// Set the adaptive batching maximum wait time in milliseconds
    pub fn with_adaptive_max_wait_ms(mut self, wait_ms: u64) -> Self {
        self.adaptive_max_wait_ms = Some(wait_ms);
        self
    }

    /// Set the adaptive batching minimum wait time in milliseconds
    pub fn with_adaptive_min_wait_ms(mut self, wait_ms: u64) -> Self {
        self.adaptive_min_wait_ms = Some(wait_ms);
        self
    }

    /// Set the adaptive batching throughput window in seconds
    pub fn with_adaptive_window_secs(mut self, secs: u64) -> Self {
        self.adaptive_window_secs = Some(secs);
        self
    }

    /// Enable or disable adaptive batching
    pub fn with_adaptive_enabled(mut self, enabled: bool) -> Self {
        self.adaptive_enabled = Some(enabled);
        self
    }

    /// Set the dispatch mode for event routing.
    pub fn with_dispatch_mode(mut self, mode: DispatchMode) -> Self {
        self.dispatch_mode = Some(mode);
        self
    }

    /// Set the dispatch buffer capacity.
    pub fn with_dispatch_buffer_capacity(mut self, capacity: usize) -> Self {
        self.dispatch_buffer_capacity = Some(capacity);
        self
    }

    /// Set the bootstrap provider for initial data delivery.
    pub fn with_bootstrap_provider(
        mut self,
        provider: impl drasi_lib::bootstrap::BootstrapProvider + 'static,
    ) -> Self {
        self.bootstrap_provider = Some(Box::new(provider));
        self
    }

    /// Set whether this source should auto-start when DrasiLib starts.
    ///
    /// Default is `true`. Set to `false` if this source should only be
    /// started manually via `start_source()`.
    pub fn with_auto_start(mut self, auto_start: bool) -> Self {
        self.auto_start = auto_start;
        self
    }

    /// Set the webhook configuration to enable webhook mode.
    ///
    /// When webhook mode is enabled, the standard `HttpSourceChange` endpoints
    /// are disabled and custom webhook routes are used instead.
    pub fn with_webhooks(mut self, webhooks: WebhookConfig) -> Self {
        self.webhooks = Some(webhooks);
        self
    }

    /// Set the WAL durability configuration.
    ///
    /// When enabled, events are persisted to a Write-Ahead Log before
    /// acknowledging the caller, enabling crash recovery and replay.
    pub fn with_durability(mut self, config: drasi_lib::DurabilityConfig) -> Self {
        self.durability = Some(config);
        self
    }

    /// Set the full configuration at once
    pub fn with_config(mut self, config: HttpSourceConfig) -> Self {
        self.host = config.host;
        self.port = config.port;
        self.endpoint = config.endpoint;
        self.timeout_ms = config.timeout_ms;
        self.adaptive_max_batch_size = config.adaptive_max_batch_size;
        self.adaptive_min_batch_size = config.adaptive_min_batch_size;
        self.adaptive_max_wait_ms = config.adaptive_max_wait_ms;
        self.adaptive_min_wait_ms = config.adaptive_min_wait_ms;
        self.adaptive_window_secs = config.adaptive_window_secs;
        self.adaptive_enabled = config.adaptive_enabled;
        self.webhooks = config.webhooks;
        self.durability = config.durability;
        self
    }

    /// Build the HttpSource instance.
    ///
    /// # Returns
    ///
    /// A fully constructed `HttpSource`, or an error if construction fails.
    pub fn build(self) -> Result<HttpSource> {
        let config = HttpSourceConfig {
            host: self.host,
            port: self.port,
            endpoint: self.endpoint,
            timeout_ms: self.timeout_ms,
            adaptive_max_batch_size: self.adaptive_max_batch_size,
            adaptive_min_batch_size: self.adaptive_min_batch_size,
            adaptive_max_wait_ms: self.adaptive_max_wait_ms,
            adaptive_min_wait_ms: self.adaptive_min_wait_ms,
            adaptive_window_secs: self.adaptive_window_secs,
            adaptive_enabled: self.adaptive_enabled,
            webhooks: self.webhooks,
            durability: self.durability,
        };

        // Build SourceBaseParams with all settings
        let mut params = SourceBaseParams::new(&self.id).with_auto_start(self.auto_start);
        if let Some(mode) = self.dispatch_mode {
            params = params.with_dispatch_mode(mode);
        }
        if let Some(capacity) = self.dispatch_buffer_capacity {
            params = params.with_dispatch_buffer_capacity(capacity);
        }
        if let Some(provider) = self.bootstrap_provider {
            params = params.with_bootstrap_provider(provider);
        }

        // Configure adaptive batching
        let mut adaptive_config = AdaptiveBatchConfig::default();
        if let Some(max_batch) = config.adaptive_max_batch_size {
            adaptive_config.max_batch_size = max_batch;
        }
        if let Some(min_batch) = config.adaptive_min_batch_size {
            adaptive_config.min_batch_size = min_batch;
        }
        if let Some(max_wait_ms) = config.adaptive_max_wait_ms {
            adaptive_config.max_wait_time = Duration::from_millis(max_wait_ms);
        }
        if let Some(min_wait_ms) = config.adaptive_min_wait_ms {
            adaptive_config.min_wait_time = Duration::from_millis(min_wait_ms);
        }
        if let Some(window_secs) = config.adaptive_window_secs {
            adaptive_config.throughput_window = Duration::from_secs(window_secs);
        }
        if let Some(enabled) = config.adaptive_enabled {
            adaptive_config.adaptive_enabled = enabled;
        }

        Ok(HttpSource {
            base: SourceBase::new(params)?,
            config,
            adaptive_config,
            wal: tokio::sync::RwLock::new(None),
            prune_task: tokio::sync::RwLock::new(None),
        })
    }
}

impl HttpSource {
    /// Create a builder for HttpSource with the given ID.
    ///
    /// This is the recommended way to construct an HttpSource.
    ///
    /// # Arguments
    ///
    /// * `id` - Unique identifier for the source instance
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use drasi_source_http::HttpSource;
    ///
    /// # fn example() -> anyhow::Result<()> {
    /// let source = HttpSource::builder("my-source")
    ///     .with_host("0.0.0.0")
    ///     .with_port(8080)
    ///     .build()?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn builder(id: impl Into<String>) -> HttpSourceBuilder {
        HttpSourceBuilder::new(id)
    }
}

/// Handle errors according to configured error behavior
fn handle_error(
    behavior: &ErrorBehavior,
    source_id: &str,
    status: StatusCode,
    message: &str,
    detail: Option<&str>,
) -> (StatusCode, Json<EventResponse>) {
    match behavior {
        ErrorBehavior::Reject => {
            debug!("[{source_id}] Rejecting request: {message}");
            (
                status,
                Json(EventResponse {
                    success: false,
                    message: message.to_string(),
                    error: detail.map(String::from),
                }),
            )
        }
        ErrorBehavior::AcceptAndLog => {
            warn!("[{source_id}] Accepting with error (logged): {message}");
            (
                StatusCode::OK,
                Json(EventResponse {
                    success: true,
                    message: format!("Accepted with warning: {message}"),
                    error: detail.map(String::from),
                }),
            )
        }
        ErrorBehavior::AcceptAndSkip => {
            trace!("[{source_id}] Accepting silently: {message}");
            (
                StatusCode::OK,
                Json(EventResponse {
                    success: true,
                    message: "Accepted".to_string(),
                    error: None,
                }),
            )
        }
    }
}

/// Parse query string into a HashMap
fn parse_query_string(query: Option<&str>) -> HashMap<String, String> {
    query
        .map(|q| {
            q.split('&')
                .filter_map(|pair| {
                    let mut parts = pair.splitn(2, '=');
                    let key = parts.next()?;
                    let value = parts.next().unwrap_or("");
                    Some((urlencoding_decode(key), urlencoding_decode(value)))
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Simple URL decoding (handles %XX sequences) with proper UTF-8 handling
fn urlencoding_decode(s: &str) -> String {
    // Collect decoded bytes first, then convert to String to properly handle UTF-8
    let mut decoded: Vec<u8> = Vec::with_capacity(s.len());
    let mut chars = s.chars();

    while let Some(c) = chars.next() {
        if c == '%' {
            let mut hex = String::new();
            if let Some(c1) = chars.next() {
                hex.push(c1);
            }
            if let Some(c2) = chars.next() {
                hex.push(c2);
            }

            if hex.len() == 2 {
                if let Ok(byte) = u8::from_str_radix(&hex, 16) {
                    decoded.push(byte);
                    continue;
                }
            }

            // If we couldn't decode a valid %XX sequence, keep the original text
            decoded.extend_from_slice(b"%");
            decoded.extend_from_slice(hex.as_bytes());
        } else if c == '+' {
            decoded.push(b' ');
        } else {
            // Encode the character as UTF-8 and append its bytes
            let mut buf = [0u8; 4];
            let encoded = c.encode_utf8(&mut buf);
            decoded.extend_from_slice(encoded.as_bytes());
        }
    }

    // Convert bytes to string, replacing invalid UTF-8 sequences
    String::from_utf8_lossy(&decoded).into_owned()
}

/// Build a CORS layer from configuration
fn build_cors_layer(cors_config: &CorsConfig) -> CorsLayer {
    let mut cors = CorsLayer::new();

    // Configure allowed origins
    if cors_config.allow_origins.len() == 1 && cors_config.allow_origins[0] == "*" {
        cors = cors.allow_origin(Any);
    } else {
        let origins: Vec<_> = cors_config
            .allow_origins
            .iter()
            .filter_map(|o| o.parse().ok())
            .collect();
        cors = cors.allow_origin(origins);
    }

    // Configure allowed methods
    let methods: Vec<Method> = cors_config
        .allow_methods
        .iter()
        .filter_map(|m| m.parse().ok())
        .collect();
    cors = cors.allow_methods(methods);

    // Configure allowed headers
    if cors_config.allow_headers.len() == 1 && cors_config.allow_headers[0] == "*" {
        cors = cors.allow_headers(Any);
    } else {
        let headers: Vec<header::HeaderName> = cors_config
            .allow_headers
            .iter()
            .filter_map(|h| h.parse().ok())
            .collect();
        cors = cors.allow_headers(headers);
    }

    // Configure exposed headers
    if !cors_config.expose_headers.is_empty() {
        let exposed: Vec<header::HeaderName> = cors_config
            .expose_headers
            .iter()
            .filter_map(|h| h.parse().ok())
            .collect();
        cors = cors.expose_headers(exposed);
    }

    // Configure credentials
    if cors_config.allow_credentials {
        cors = cors.allow_credentials(true);
    }

    // Configure max age
    cors = cors.max_age(Duration::from_secs(cors_config.max_age));

    cors
}

#[cfg(test)]
mod tests {
    use super::*;

    mod construction {
        use super::*;

        #[test]
        fn test_builder_with_valid_config() {
            let source = HttpSourceBuilder::new("test-source")
                .with_host("localhost")
                .with_port(8080)
                .build();
            assert!(source.is_ok());
        }

        #[test]
        fn test_builder_with_custom_config() {
            let source = HttpSourceBuilder::new("http-source")
                .with_host("0.0.0.0")
                .with_port(9000)
                .with_endpoint("/events")
                .build()
                .unwrap();
            assert_eq!(source.id(), "http-source");
        }

        #[test]
        fn test_with_dispatch_creates_source() {
            let config = HttpSourceConfig {
                host: "localhost".to_string(),
                port: 8080,
                endpoint: None,
                timeout_ms: 10000,
                adaptive_max_batch_size: None,
                adaptive_min_batch_size: None,
                adaptive_max_wait_ms: None,
                adaptive_min_wait_ms: None,
                adaptive_window_secs: None,
                adaptive_enabled: None,
                webhooks: None,
                durability: None,
            };
            let source = HttpSource::with_dispatch(
                "dispatch-source",
                config,
                Some(DispatchMode::Channel),
                Some(1000),
            );
            assert!(source.is_ok());
            assert_eq!(source.unwrap().id(), "dispatch-source");
        }
    }

    mod properties {
        use super::*;

        #[test]
        fn test_id_returns_correct_value() {
            let source = HttpSourceBuilder::new("my-http-source")
                .with_host("localhost")
                .build()
                .unwrap();
            assert_eq!(source.id(), "my-http-source");
        }

        #[test]
        fn test_type_name_returns_http() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .build()
                .unwrap();
            assert_eq!(source.type_name(), "http");
        }

        #[test]
        fn test_properties_contains_host_and_port() {
            let source = HttpSourceBuilder::new("test")
                .with_host("192.168.1.1")
                .with_port(9000)
                .build()
                .unwrap();
            let props = source.properties();

            assert_eq!(
                props.get("host"),
                Some(&serde_json::Value::String("192.168.1.1".to_string()))
            );
            assert_eq!(
                props.get("port"),
                Some(&serde_json::Value::Number(9000.into()))
            );
        }

        #[test]
        fn test_properties_includes_endpoint_when_set() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .with_endpoint("/api/v1")
                .build()
                .unwrap();
            let props = source.properties();

            assert_eq!(
                props.get("endpoint"),
                Some(&serde_json::Value::String("/api/v1".to_string()))
            );
        }

        #[test]
        fn test_properties_excludes_endpoint_when_none() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .build()
                .unwrap();
            let props = source.properties();

            assert!(!props.contains_key("endpoint"));
        }

        #[test]
        fn test_describe_schema_uses_webhook_mappings() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .with_webhooks(WebhookConfig {
                    error_behavior: ErrorBehavior::AcceptAndLog,
                    cors: None,
                    routes: vec![crate::config::WebhookRoute {
                        path: "/events".to_string(),
                        methods: vec![crate::config::HttpMethod::Post],
                        auth: None,
                        error_behavior: None,
                        mappings: vec![crate::config::WebhookMapping {
                            when: None,
                            operation: Some(crate::config::OperationType::Insert),
                            operation_from: None,
                            operation_map: None,
                            element_type: crate::config::ElementType::Node,
                            effective_from: None,
                            template: crate::config::ElementTemplate {
                                id: "{{payload.id}}".to_string(),
                                labels: vec!["Order".to_string()],
                                properties: Some(serde_json::json!({
                                    "total": "{{payload.total}}",
                                    "status": "{{payload.status}}"
                                })),
                                from: None,
                                to: None,
                            },
                        }],
                    }],
                })
                .build()
                .unwrap();

            let schema = source
                .describe_schema()
                .expect("webhook-configured HTTP source should expose schema");

            assert_eq!(schema.nodes.len(), 1);
            let node = &schema.nodes[0];
            assert_eq!(node.label, "Order");
            assert!(node
                .properties
                .iter()
                .any(|property| property.name == "total"));
            assert!(node
                .properties
                .iter()
                .any(|property| property.name == "status"));
        }

        #[test]
        fn test_describe_schema_includes_relation_mappings() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .with_webhooks(WebhookConfig {
                    error_behavior: ErrorBehavior::AcceptAndLog,
                    cors: None,
                    routes: vec![crate::config::WebhookRoute {
                        path: "/events".to_string(),
                        methods: vec![crate::config::HttpMethod::Post],
                        auth: None,
                        error_behavior: None,
                        mappings: vec![crate::config::WebhookMapping {
                            when: None,
                            operation: Some(crate::config::OperationType::Insert),
                            operation_from: None,
                            operation_map: None,
                            element_type: crate::config::ElementType::Relation,
                            effective_from: None,
                            template: crate::config::ElementTemplate {
                                id: "{{payload.id}}".to_string(),
                                labels: vec!["PLACED_BY".to_string()],
                                properties: Some(serde_json::json!({
                                    "placed_at": "{{payload.timestamp}}"
                                })),
                                from: None,
                                to: None,
                            },
                        }],
                    }],
                })
                .build()
                .unwrap();

            let schema = source
                .describe_schema()
                .expect("webhook-configured HTTP source should expose schema for relations");

            assert_eq!(schema.relations.len(), 1);
            let relation = &schema.relations[0];
            assert_eq!(relation.label, "PLACED_BY");
            assert!(relation
                .properties
                .iter()
                .any(|property| property.name == "placed_at"));
            // Static derivation cannot infer endpoints
            assert_eq!(relation.from, None);
            assert_eq!(relation.to, None);
        }
    }

    mod lifecycle {
        use super::*;

        #[tokio::test]
        async fn test_initial_status_is_stopped() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .build()
                .unwrap();
            assert_eq!(source.status().await, ComponentStatus::Stopped);
        }
    }

    mod builder {
        use super::*;

        #[test]
        fn test_http_builder_defaults() {
            let source = HttpSourceBuilder::new("test").build().unwrap();
            assert_eq!(source.config.port, 8080);
            assert_eq!(source.config.timeout_ms, 10000);
            assert_eq!(source.config.endpoint, None);
        }

        #[test]
        fn test_http_builder_custom_values() {
            let source = HttpSourceBuilder::new("test")
                .with_host("api.example.com")
                .with_port(9000)
                .with_endpoint("/webhook")
                .with_timeout_ms(5000)
                .build()
                .unwrap();

            assert_eq!(source.config.host, "api.example.com");
            assert_eq!(source.config.port, 9000);
            assert_eq!(source.config.endpoint, Some("/webhook".to_string()));
            assert_eq!(source.config.timeout_ms, 5000);
        }

        #[test]
        fn test_http_builder_adaptive_batching() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .with_adaptive_max_batch_size(1000)
                .with_adaptive_min_batch_size(10)
                .with_adaptive_max_wait_ms(500)
                .with_adaptive_min_wait_ms(50)
                .with_adaptive_window_secs(60)
                .with_adaptive_enabled(true)
                .build()
                .unwrap();

            assert_eq!(source.config.adaptive_max_batch_size, Some(1000));
            assert_eq!(source.config.adaptive_min_batch_size, Some(10));
            assert_eq!(source.config.adaptive_max_wait_ms, Some(500));
            assert_eq!(source.config.adaptive_min_wait_ms, Some(50));
            assert_eq!(source.config.adaptive_window_secs, Some(60));
            assert_eq!(source.config.adaptive_enabled, Some(true));
        }

        #[test]
        fn test_builder_id() {
            let source = HttpSource::builder("my-http-source")
                .with_host("localhost")
                .build()
                .unwrap();

            assert_eq!(source.base.id, "my-http-source");
        }
    }

    mod event_conversion {
        use super::*;

        #[test]
        fn test_convert_node_insert() {
            let mut props = serde_json::Map::new();
            props.insert(
                "name".to_string(),
                serde_json::Value::String("Alice".to_string()),
            );
            props.insert("age".to_string(), serde_json::Value::Number(30.into()));

            let http_change = HttpSourceChange::Insert {
                element: HttpElement::Node {
                    id: "user-1".to_string(),
                    labels: vec!["User".to_string()],
                    properties: props,
                },
                timestamp: Some(1234567890000000000),
            };

            let result = convert_http_to_source_change(&http_change, "test-source");
            assert!(result.is_ok());

            match result.unwrap() {
                drasi_core::models::SourceChange::Insert { element } => match element {
                    drasi_core::models::Element::Node {
                        metadata,
                        properties,
                    } => {
                        assert_eq!(metadata.reference.element_id.as_ref(), "user-1");
                        assert_eq!(metadata.labels.len(), 1);
                        assert_eq!(metadata.effective_from, 1234567890000);
                        assert!(properties.get("name").is_some());
                        assert!(properties.get("age").is_some());
                    }
                    _ => panic!("Expected Node element"),
                },
                _ => panic!("Expected Insert operation"),
            }
        }

        #[test]
        fn test_convert_relation_insert() {
            let http_change = HttpSourceChange::Insert {
                element: HttpElement::Relation {
                    id: "follows-1".to_string(),
                    labels: vec!["FOLLOWS".to_string()],
                    from: "user-1".to_string(),
                    to: "user-2".to_string(),
                    properties: serde_json::Map::new(),
                },
                timestamp: None,
            };

            let result = convert_http_to_source_change(&http_change, "test-source");
            assert!(result.is_ok());

            match result.unwrap() {
                drasi_core::models::SourceChange::Insert { element } => match element {
                    drasi_core::models::Element::Relation {
                        metadata,
                        out_node,
                        in_node,
                        ..
                    } => {
                        assert_eq!(metadata.reference.element_id.as_ref(), "follows-1");
                        assert_eq!(in_node.element_id.as_ref(), "user-1");
                        assert_eq!(out_node.element_id.as_ref(), "user-2");
                    }
                    _ => panic!("Expected Relation element"),
                },
                _ => panic!("Expected Insert operation"),
            }
        }

        #[test]
        fn test_convert_delete() {
            let http_change = HttpSourceChange::Delete {
                id: "user-1".to_string(),
                labels: Some(vec!["User".to_string()]),
                timestamp: Some(9999999999),
            };

            let result = convert_http_to_source_change(&http_change, "test-source");
            assert!(result.is_ok());

            match result.unwrap() {
                drasi_core::models::SourceChange::Delete { metadata } => {
                    assert_eq!(metadata.reference.element_id.as_ref(), "user-1");
                    assert_eq!(metadata.labels.len(), 1);
                }
                _ => panic!("Expected Delete operation"),
            }
        }

        #[test]
        fn test_convert_update() {
            let http_change = HttpSourceChange::Update {
                element: HttpElement::Node {
                    id: "user-1".to_string(),
                    labels: vec!["User".to_string()],
                    properties: serde_json::Map::new(),
                },
                timestamp: None,
            };

            let result = convert_http_to_source_change(&http_change, "test-source");
            assert!(result.is_ok());

            match result.unwrap() {
                drasi_core::models::SourceChange::Update { .. } => {
                    // Success
                }
                _ => panic!("Expected Update operation"),
            }
        }
    }

    mod adaptive_config {
        use super::*;

        #[test]
        fn test_adaptive_config_from_http_config() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .with_adaptive_max_batch_size(500)
                .with_adaptive_enabled(true)
                .build()
                .unwrap();

            // The adaptive config should be initialized from the http config
            assert_eq!(source.adaptive_config.max_batch_size, 500);
            assert!(source.adaptive_config.adaptive_enabled);
        }

        #[test]
        fn test_adaptive_config_uses_defaults_when_not_specified() {
            let source = HttpSourceBuilder::new("test")
                .with_host("localhost")
                .build()
                .unwrap();

            // Should use AdaptiveBatchConfig defaults
            let default_config = AdaptiveBatchConfig::default();
            assert_eq!(
                source.adaptive_config.max_batch_size,
                default_config.max_batch_size
            );
            assert_eq!(
                source.adaptive_config.min_batch_size,
                default_config.min_batch_size
            );
        }
    }
}

#[cfg(test)]
mod fallback_tests {
    use super::*;
    use drasi_lib::sources::Source;

    #[test]
    fn test_builder_fallback_produces_camel_case() {
        let source = HttpSourceBuilder::new("http-fallback")
            .with_host("0.0.0.0")
            .with_port(9090)
            .with_endpoint("/ingest")
            .with_timeout_ms(5000)
            .with_adaptive_max_batch_size(500)
            .with_adaptive_min_batch_size(10)
            .with_adaptive_max_wait_ms(2000)
            .with_adaptive_min_wait_ms(100)
            .build()
            .unwrap();

        let props = source.properties();

        // Must use camelCase keys (DTO serialization)
        assert!(
            props.contains_key("timeoutMs"),
            "expected camelCase 'timeoutMs', got keys: {:?}",
            props.keys().collect::<Vec<_>>()
        );
        assert!(
            props.contains_key("adaptiveMaxBatchSize"),
            "expected camelCase 'adaptiveMaxBatchSize'"
        );
        assert!(
            props.contains_key("adaptiveMinBatchSize"),
            "expected camelCase 'adaptiveMinBatchSize'"
        );
        assert!(
            props.contains_key("adaptiveMaxWaitMs"),
            "expected camelCase 'adaptiveMaxWaitMs'"
        );
        assert!(
            props.contains_key("adaptiveMinWaitMs"),
            "expected camelCase 'adaptiveMinWaitMs'"
        );

        // Must NOT have snake_case keys
        assert!(
            !props.contains_key("timeout_ms"),
            "should not have snake_case 'timeout_ms'"
        );
        assert!(
            !props.contains_key("adaptive_max_batch_size"),
            "should not have snake_case 'adaptive_max_batch_size'"
        );

        // Values should be correct
        assert_eq!(props.get("host").and_then(|v| v.as_str()), Some("0.0.0.0"));
        assert_eq!(props.get("port").and_then(|v| v.as_u64()), Some(9090));
        assert_eq!(
            props.get("endpoint").and_then(|v| v.as_str()),
            Some("/ingest")
        );
        assert_eq!(props.get("timeoutMs").and_then(|v| v.as_u64()), Some(5000));
        assert_eq!(
            props.get("adaptiveMaxBatchSize").and_then(|v| v.as_u64()),
            Some(500)
        );
    }
}

/// Dynamic plugin entry point.
///
/// Dynamic plugin entry point.
#[cfg(feature = "dynamic-plugin")]
drasi_plugin_sdk::export_plugin!(
    plugin_id = "http-source",
    core_version = env!("CARGO_PKG_VERSION"),
    lib_version = env!("CARGO_PKG_VERSION"),
    plugin_version = env!("CARGO_PKG_VERSION"),
    source_descriptors = [descriptor::HttpSourceDescriptor],
    reaction_descriptors = [],
    bootstrap_descriptors = [],
);