tapped 0.3.1

//! Configuration types for tap process and client.

use std::time::Duration;
use url::Url;

/// Log level for tap process.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum LogLevel {
    Debug,
    #[default]
    Info,
    Warn,
    Error,
}

impl LogLevel {
    /// Convert to the string value expected by tap.
    pub fn as_str(&self) -> &'static str {
        match self {
            LogLevel::Debug => "debug",
            LogLevel::Info => "info",
            LogLevel::Warn => "warn",
            LogLevel::Error => "error",
        }
    }
}

/// Configuration for a tap instance.
///
/// All fields are optional. When spawning a tap process, unset fields will use
/// tap's built-in defaults. When connecting to an existing instance, only
/// client-side options (timeouts, auth) are relevant.
///
/// Use [`TapConfig::builder()`] for ergonomic construction.
#[derive(Debug, Clone, Default)]
pub struct TapConfig {
    // Database
    /// Database connection string (sqlite://path or postgres://...)
    pub database_url: Option<String>,
    /// Maximum number of database connections
    pub max_db_conns: Option<u32>,

    // Server
    /// HTTP server bind address (e.g., `":2480"`, `"127.0.0.1:2480"`, or `"[::1]:2480"`)
    pub bind: Option<String>,
    /// Basic auth admin password for all requests
    pub admin_password: Option<String>,
    /// Address for metrics/pprof server
    pub metrics_listen: Option<String>,
    /// Log verbosity level
    pub log_level: Option<LogLevel>,

    // AT Protocol
    /// PLC directory URL
    pub plc_url: Option<Url>,
    /// AT Protocol relay URL
    pub relay_url: Option<Url>,

    // Processing
    /// Number of parallel firehose event processors
    pub firehose_parallelism: Option<u32>,
    /// Number of parallel resync workers
    pub resync_parallelism: Option<u32>,
    /// Number of parallel outbox workers
    pub outbox_parallelism: Option<u32>,
    /// How often to save firehose cursor
    pub cursor_save_interval: Option<Duration>,
    /// Timeout for fetching repo CARs from PDS
    pub repo_fetch_timeout: Option<Duration>,
    /// Size of in-process identity cache
    pub ident_cache_size: Option<u32>,
    /// Size of outbox before back pressure
    pub outbox_capacity: Option<u32>,
    /// Timeout before retrying unacked events
    pub retry_timeout: Option<Duration>,

    // Network boundary
    /// Track all repos on the network
    pub full_network: Option<bool>,
    /// Track repos with records in this collection
    pub signal_collection: Option<String>,

    // Filtering
    /// Filter output records by collection (supports wildcards)
    pub collection_filters: Option<Vec<String>>,

    // Delivery mode
    /// Enable fire-and-forget mode (no client acks)
    pub disable_acks: Option<bool>,
    /// Webhook URL for event delivery
    pub webhook_url: Option<Url>,
    /// Run in outbox-only mode
    pub outbox_only: Option<bool>,

    // Client-side options (not sent to tap process)
    /// Forward tap's stdout/stderr to this process (default: false)
    pub inherit_stdio: Option<bool>,
    /// Graceful shutdown timeout (default: 5s)
    pub shutdown_timeout: Option<Duration>,
    /// HTTP request timeout (default: 30s)
    pub request_timeout: Option<Duration>,
    /// Max wait for tap to become healthy (default: 30s)
    pub startup_timeout: Option<Duration>,
}

impl TapConfig {
    /// Create a new empty configuration.
    pub fn new() -> Self {
        Self::default()
    }

    /// Create a builder for ergonomic configuration.
    pub fn builder() -> TapConfigBuilder {
        TapConfigBuilder::default()
    }

    /// Get the shutdown timeout, or the default (5 seconds).
    pub fn shutdown_timeout(&self) -> Duration {
        self.shutdown_timeout.unwrap_or(Duration::from_secs(5))
    }

    /// Get the request timeout, or the default (30 seconds).
    pub fn request_timeout(&self) -> Duration {
        self.request_timeout.unwrap_or(Duration::from_secs(30))
    }

    /// Get the startup timeout, or the default (30 seconds).
    pub fn startup_timeout(&self) -> Duration {
        self.startup_timeout.unwrap_or(Duration::from_secs(30))
    }

    /// Whether to inherit stdio from the parent process (default: false).
    pub fn inherit_stdio(&self) -> bool {
        self.inherit_stdio.unwrap_or(false)
    }

    /// Convert configuration to environment variables for subprocess.
    pub fn to_env_vars(&self) -> Vec<(String, String)> {
        let mut vars = Vec::new();

        /// Helper macro to push an env var if the field is Some.
        macro_rules! push_env {
            ($field:expr, $name:literal, clone) => {
                if let Some(ref v) = $field {
                    vars.push(($name.into(), v.clone()));
                }
            };
            ($field:expr, $name:literal, string) => {
                if let Some(v) = $field {
                    vars.push(($name.into(), v.to_string()));
                }
            };
            ($field:expr, $name:literal, ref_string) => {
                if let Some(ref v) = $field {
                    vars.push(($name.into(), v.to_string()));
                }
            };
            ($field:expr, $name:literal, as_str) => {
                if let Some(ref v) = $field {
                    vars.push(($name.into(), v.as_str().into()));
                }
            };
            ($field:expr, $name:literal, duration) => {
                if let Some(v) = $field {
                    vars.push(($name.into(), format_duration(v)));
                }
            };
        }

        push_env!(self.database_url, "TAP_DATABASE_URL", clone);
        push_env!(self.max_db_conns, "TAP_MAX_DB_CONNS", string);
        push_env!(self.bind, "TAP_BIND", clone);
        push_env!(self.admin_password, "TAP_ADMIN_PASSWORD", clone);
        push_env!(self.metrics_listen, "TAP_METRICS_LISTEN", clone);
        push_env!(self.log_level, "TAP_LOG_LEVEL", as_str);
        push_env!(self.plc_url, "TAP_PLC_URL", ref_string);
        push_env!(self.relay_url, "TAP_RELAY_URL", ref_string);
        push_env!(
            self.firehose_parallelism,
            "TAP_FIREHOSE_PARALLELISM",
            string
        );
        push_env!(self.resync_parallelism, "TAP_RESYNC_PARALLELISM", string);
        push_env!(self.outbox_parallelism, "TAP_OUTBOX_PARALLELISM", string);
        push_env!(
            self.cursor_save_interval,
            "TAP_CURSOR_SAVE_INTERVAL",
            duration
        );
        push_env!(self.repo_fetch_timeout, "TAP_REPO_FETCH_TIMEOUT", duration);
        push_env!(self.ident_cache_size, "RELAY_IDENT_CACHE_SIZE", string);
        push_env!(self.outbox_capacity, "TAP_OUTBOX_CAPACITY", string);
        push_env!(self.retry_timeout, "TAP_RETRY_TIMEOUT", duration);
        push_env!(self.full_network, "TAP_FULL_NETWORK", string);
        push_env!(self.signal_collection, "TAP_SIGNAL_COLLECTION", clone);
        push_env!(self.disable_acks, "TAP_DISABLE_ACKS", string);
        push_env!(self.webhook_url, "TAP_WEBHOOK_URL", ref_string);
        push_env!(self.outbox_only, "TAP_OUTBOX_ONLY", string);

        if let Some(ref v) = self.collection_filters {
            vars.push(("TAP_COLLECTION_FILTERS".into(), v.join(",")));
        }

        vars
    }
}

/// Format a Duration as a Go-style duration string (e.g., "30s", "5m").
fn format_duration(d: Duration) -> String {
    let secs = d.as_secs();
    let millis = d.subsec_millis();

    if millis == 0 {
        if secs > 0 && secs.is_multiple_of(3600) {
            format!("{}h", secs / 3600)
        } else if secs > 0 && secs.is_multiple_of(60) {
            format!("{}m", secs / 60)
        } else {
            format!("{}s", secs)
        }
    } else {
        format!("{}ms", d.as_millis())
    }
}

/// Builder for [`TapConfig`].
#[derive(Debug, Clone, Default)]
pub struct TapConfigBuilder {
    config: TapConfig,
}

impl TapConfigBuilder {
    /// Set the database URL.
    pub fn database_url(mut self, url: impl Into<String>) -> Self {
        self.config.database_url = Some(url.into());
        self
    }

    /// Set the maximum number of database connections.
    pub fn max_db_conns(mut self, n: u32) -> Self {
        self.config.max_db_conns = Some(n);
        self
    }

    /// Set the HTTP server bind address.
    pub fn bind(mut self, addr: impl Into<String>) -> Self {
        self.config.bind = Some(addr.into());
        self
    }

    /// Set the admin password for Basic auth.
    pub fn admin_password(mut self, password: impl Into<String>) -> Self {
        self.config.admin_password = Some(password.into());
        self
    }

    /// Set the metrics server listen address.
    pub fn metrics_listen(mut self, addr: impl Into<String>) -> Self {
        self.config.metrics_listen = Some(addr.into());
        self
    }

    /// Set the log level.
    pub fn log_level(mut self, level: LogLevel) -> Self {
        self.config.log_level = Some(level);
        self
    }

    /// Set the PLC directory URL.
    pub fn plc_url(mut self, url: Url) -> Self {
        self.config.plc_url = Some(url);
        self
    }

    /// Set the relay URL.
    pub fn relay_url(mut self, url: Url) -> Self {
        self.config.relay_url = Some(url);
        self
    }

    /// Set the firehose parallelism.
    pub fn firehose_parallelism(mut self, n: u32) -> Self {
        self.config.firehose_parallelism = Some(n);
        self
    }

    /// Set the resync parallelism.
    pub fn resync_parallelism(mut self, n: u32) -> Self {
        self.config.resync_parallelism = Some(n);
        self
    }

    /// Set the outbox parallelism.
    pub fn outbox_parallelism(mut self, n: u32) -> Self {
        self.config.outbox_parallelism = Some(n);
        self
    }

    /// Set how often to save the firehose cursor.
    pub fn cursor_save_interval(mut self, d: Duration) -> Self {
        self.config.cursor_save_interval = Some(d);
        self
    }

    /// Set the repo fetch timeout.
    pub fn repo_fetch_timeout(mut self, d: Duration) -> Self {
        self.config.repo_fetch_timeout = Some(d);
        self
    }

    /// Set the identity cache size.
    pub fn ident_cache_size(mut self, n: u32) -> Self {
        self.config.ident_cache_size = Some(n);
        self
    }

    /// Set the outbox capacity.
    pub fn outbox_capacity(mut self, n: u32) -> Self {
        self.config.outbox_capacity = Some(n);
        self
    }

    /// Set the retry timeout for unacked events.
    pub fn retry_timeout(mut self, d: Duration) -> Self {
        self.config.retry_timeout = Some(d);
        self
    }

    /// Enable full network mode.
    pub fn full_network(mut self, enabled: bool) -> Self {
        self.config.full_network = Some(enabled);
        self
    }

    /// Set the signal collection for repo discovery.
    pub fn signal_collection(mut self, collection: impl Into<String>) -> Self {
        self.config.signal_collection = Some(collection.into());
        self
    }

    /// Add a collection filter.
    ///
    /// This can be called multiple times to add multiple filters.
    /// Supports wildcards (e.g., "app.bsky.feed.*").
    pub fn collection_filter(mut self, filter: impl Into<String>) -> Self {
        self.config
            .collection_filters
            .get_or_insert_with(Vec::new)
            .push(filter.into());
        self
    }

    /// Set collection filters.
    pub fn collection_filters(mut self, filters: Vec<String>) -> Self {
        self.config.collection_filters = Some(filters);
        self
    }

    /// Disable acknowledgments (fire-and-forget mode).
    pub fn disable_acks(mut self, disabled: bool) -> Self {
        self.config.disable_acks = Some(disabled);
        self
    }

    /// Set the webhook URL for event delivery.
    pub fn webhook_url(mut self, url: Url) -> Self {
        self.config.webhook_url = Some(url);
        self
    }

    /// Enable outbox-only mode.
    pub fn outbox_only(mut self, enabled: bool) -> Self {
        self.config.outbox_only = Some(enabled);
        self
    }

    /// Set the graceful shutdown timeout.
    pub fn shutdown_timeout(mut self, d: Duration) -> Self {
        self.config.shutdown_timeout = Some(d);
        self
    }

    /// Set the HTTP request timeout.
    pub fn request_timeout(mut self, d: Duration) -> Self {
        self.config.request_timeout = Some(d);
        self
    }

    /// Set the startup health check timeout.
    pub fn startup_timeout(mut self, d: Duration) -> Self {
        self.config.startup_timeout = Some(d);
        self
    }

    /// Forward tap's stdout/stderr to this process.
    ///
    /// When enabled, tap's output will be visible in the terminal.
    /// When disabled (default), tap's output is discarded.
    pub fn inherit_stdio(mut self, inherit: bool) -> Self {
        self.config.inherit_stdio = Some(inherit);
        self
    }

    /// Build the configuration.
    pub fn build(self) -> TapConfig {
        self.config
    }
}

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

    #[test]
    fn test_format_duration() {
        assert_eq!(format_duration(Duration::from_secs(30)), "30s");
        assert_eq!(format_duration(Duration::from_secs(60)), "1m");
        assert_eq!(format_duration(Duration::from_secs(3600)), "1h");
        assert_eq!(format_duration(Duration::from_secs(90)), "90s");
        assert_eq!(format_duration(Duration::from_millis(500)), "500ms");
    }

    #[test]
    fn test_config_to_env_vars() {
        let config = TapConfig::builder()
            .database_url("sqlite://./test.db")
            .bind(":3000")
            .signal_collection("app.bsky.feed.post")
            .collection_filters(vec!["app.bsky.feed.post".into(), "app.bsky.graph.*".into()])
            .disable_acks(true)
            .build();

        let vars = config.to_env_vars();
        assert!(vars.contains(&("TAP_DATABASE_URL".into(), "sqlite://./test.db".into())));
        assert!(vars.contains(&("TAP_BIND".into(), ":3000".into())));
        assert!(vars.contains(&("TAP_SIGNAL_COLLECTION".into(), "app.bsky.feed.post".into())));
        assert!(vars.contains(&(
            "TAP_COLLECTION_FILTERS".into(),
            "app.bsky.feed.post,app.bsky.graph.*".into()
        )));
        assert!(vars.contains(&("TAP_DISABLE_ACKS".into(), "true".into())));
    }
}