#![deny(clippy::undocumented_unsafe_blocks)]

use cosimo::SyncServerLimits;
use cosimo::sync_http::{
    Request, Response, ServerState, Status, SyncServerPrivacyConfig, SyncServerServiceMode,
    handle_request, http_error_response,
};
use std::collections::BTreeMap;
use std::env;
use std::fs::{self, File, OpenOptions, TryLockError};
use std::io::Write;
use std::net::{IpAddr, TcpListener, TcpStream};
#[cfg(unix)]
use std::os::unix::fs::OpenOptionsExt;
#[cfg(unix)]
use std::os::unix::io::{AsRawFd, FromRawFd};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;

const DEFAULT_BIND_ADDRESS: &str = "127.0.0.1:44791";
const STORAGE_LOCK_FILE_NAME: &str = "cosimo-sync-server.lock";
const REQUEST_READ_TIMEOUT_SECONDS: u64 = 300;
const RESPONSE_WRITE_TIMEOUT_SECONDS: u64 = 60;
const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 64;
const SERVER_BUSY_RETRY_AFTER_SECONDS: &str = "15";

fn main() {
    if let Err(error) = run() {
        eprintln!("error: {error}");
        std::process::exit(1);
    }
}

fn run() -> Result<(), String> {
    let options = Options::parse(env::args().skip(1))?;
    if options.help {
        print_help();
        return Ok(());
    }

    let environment = env::vars().collect::<BTreeMap<_, _>>();
    let limits = sync_limits_from_environment(&environment)?;
    let privacy_config = sync_privacy_config_from_environment(&environment)?;
    let max_concurrent_requests = max_concurrent_requests_from_environment(&environment)?;
    let _storage_lock = match options.storage_root.as_ref() {
        Some(root) => Some(acquire_storage_lock(root, &options.bind_address)?),
        None => None,
    };
    let listener = TcpListener::bind(&options.bind_address)
        .map_err(|error| format!("could not bind {}: {error}", options.bind_address))?;
    eprintln!("Cosimo sync server listening on {}", options.bind_address);
    let state = Arc::new(match options.storage_root {
        Some(root) => {
            eprintln!("Using durable sync storage at {}", root.display());
            ServerState::sqlite_with_limits(root, limits).with_privacy_config(privacy_config)
        }
        None => {
            eprintln!(
                "WARNING: no --storage-root was provided; using volatile in-memory storage. \
                 Uploaded decks and login sessions will be lost when this process exits. \
                 Use --storage-root PATH for any real deployment."
            );
            ServerState::memory_with_limits(limits).with_privacy_config(privacy_config)
        }
    });
    let trusted_client_ip_header = options.trusted_client_ip_header;
    let request_gate = Arc::new(RequestGate::new(max_concurrent_requests));
    for stream in listener.incoming() {
        match stream {
            Ok(stream) => {
                let Some(permit) = request_gate.try_enter() else {
                    reject_busy_connection(stream);
                    continue;
                };
                let state = Arc::clone(&state);
                let trusted_client_ip_header = trusted_client_ip_header.clone();
                std::thread::spawn(move || {
                    let _permit = permit;
                    handle_connection(stream, &state, trusted_client_ip_header.as_deref())
                });
            }
            Err(error) => eprintln!("could not accept connection: {error}"),
        }
    }
    Ok(())
}

#[derive(Debug)]
struct RequestGate {
    max_active: usize,
    active: Mutex<usize>,
}

impl RequestGate {
    fn new(max_active: usize) -> Self {
        Self {
            max_active: max_active.max(1),
            active: Mutex::new(0),
        }
    }

    fn try_enter(self: &Arc<Self>) -> Option<RequestPermit> {
        let mut active = self.active.lock().ok()?;
        if *active >= self.max_active {
            return None;
        }
        *active += 1;
        Some(RequestPermit {
            gate: Arc::clone(self),
        })
    }

    #[cfg(test)]
    fn active_count(&self) -> usize {
        *self.active.lock().unwrap()
    }
}

#[derive(Debug)]
struct RequestPermit {
    gate: Arc<RequestGate>,
}

impl Drop for RequestPermit {
    fn drop(&mut self) {
        if let Ok(mut active) = self.gate.active.lock()
            && *active > 0
        {
            *active -= 1;
        }
    }
}

#[derive(Debug)]
struct StorageLock {
    _file: File,
}

fn acquire_storage_lock(root: &Path, bind_address: &str) -> Result<StorageLock, String> {
    fs::create_dir_all(root).map_err(|error| {
        format!(
            "could not create sync storage directory {}: {error}",
            root.display()
        )
    })?;
    ensure_plain_storage_root(root)?;
    let lock_path = root.join(STORAGE_LOCK_FILE_NAME);
    let mut file = open_storage_lock_file_no_symlink(root, &lock_path)?;
    file.try_lock().map_err(|error| match error {
        TryLockError::WouldBlock => format!(
            "another cosimo-sync-server process is already using storage root {}",
            root.display()
        ),
        TryLockError::Error(error) => format!(
            "could not lock sync storage root {}: {error}",
            root.display()
        ),
    })?;
    file.set_len(0)
        .and_then(|()| {
            writeln!(
                file,
                "Cosimo sync server storage lock\npid={}\nbind={}\n",
                std::process::id(),
                bind_address
            )
        })
        .map_err(|error| {
            format!(
                "could not write sync storage lock {}: {error}",
                lock_path.display()
            )
        })?;
    Ok(StorageLock { _file: file })
}

fn ensure_plain_storage_root(root: &Path) -> Result<(), String> {
    let metadata = fs::symlink_metadata(root).map_err(|error| {
        format!(
            "could not inspect sync storage directory {}: {error}",
            root.display()
        )
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_dir() {
        return Err(format!(
            "sync storage root must be a plain directory: {}",
            root.display()
        ));
    }
    Ok(())
}

fn open_storage_lock_file_no_symlink(root: &Path, path: &Path) -> Result<File, String> {
    #[cfg(unix)]
    {
        open_storage_lock_file_no_symlink_unix(root, path)
    }
    #[cfg(not(unix))]
    {
        let _ = root;
        open_storage_lock_file_no_symlink_fallback(path)
    }
}

#[cfg(unix)]
fn open_storage_lock_file_no_symlink_unix(root: &Path, path: &Path) -> Result<File, String> {
    let dir = OpenOptions::new()
        .read(true)
        .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW | libc::O_CLOEXEC)
        .open(root)
        .map_err(|error| {
            format!(
                "could not open sync storage directory {}: {error}",
                root.display()
            )
        })?;
    let name = std::ffi::CString::new(STORAGE_LOCK_FILE_NAME)
        .map_err(|_| "sync storage lock file name contains an interior NUL byte".to_string())?;
    let fd = unsafe {
        // SAFETY: `dir` is an open storage-root directory descriptor, `name` is
        // a NUL-terminated single-component file name, and the returned file
        // descriptor is immediately wrapped in `File` on success.
        libc::openat(
            dir.as_raw_fd(),
            name.as_ptr(),
            libc::O_RDWR | libc::O_CREAT | libc::O_NOFOLLOW | libc::O_CLOEXEC,
            0o600,
        )
    };
    if fd < 0 {
        return Err(format!(
            "could not open sync storage lock {}: {}",
            path.display(),
            std::io::Error::last_os_error()
        ));
    }
    let file = unsafe {
        // SAFETY: `fd` was returned by a successful `openat` call above and is
        // now owned by this `File`.
        File::from_raw_fd(fd)
    };
    verify_storage_lock_descriptor(&file, path)?;
    Ok(file)
}

#[cfg(not(unix))]
fn open_storage_lock_file_no_symlink_fallback(path: &Path) -> Result<File, String> {
    let mut options = OpenOptions::new();
    options.read(true).write(true).create(true);
    let file = options.open(path).map_err(|error| {
        format!(
            "could not open sync storage lock {}: {error}",
            path.display()
        )
    })?;
    verify_storage_lock_descriptor(&file, path)?;
    Ok(file)
}

fn verify_storage_lock_descriptor(file: &File, path: &Path) -> Result<(), String> {
    let metadata = file.metadata().map_err(|error| {
        format!(
            "could not inspect opened sync storage lock {}: {error}",
            path.display()
        )
    })?;
    if !metadata.is_file() {
        return Err(format!(
            "sync storage lock is not a regular file: {}",
            path.display()
        ));
    }
    Ok(())
}

#[derive(Debug)]
struct Options {
    bind_address: String,
    storage_root: Option<PathBuf>,
    trusted_client_ip_header: Option<String>,
    help: bool,
}

impl Options {
    fn parse(args: impl IntoIterator<Item = String>) -> Result<Self, String> {
        let mut options = Self {
            bind_address: DEFAULT_BIND_ADDRESS.to_string(),
            storage_root: None,
            trusted_client_ip_header: None,
            help: false,
        };
        let mut args = args.into_iter();
        while let Some(arg) = args.next() {
            match arg.as_str() {
                "-h" | "--help" => options.help = true,
                "--bind" => {
                    options.bind_address = args
                        .next()
                        .ok_or_else(|| "--bind requires an address".to_string())?;
                }
                "--storage-root" => {
                    options.storage_root =
                        Some(PathBuf::from(args.next().ok_or_else(|| {
                            "--storage-root requires a path".to_string()
                        })?));
                }
                "--trusted-client-ip-header" => {
                    let header = args.next().ok_or_else(|| {
                        "--trusted-client-ip-header requires a header name".to_string()
                    })?;
                    options.trusted_client_ip_header = Some(normalize_http_header_name(&header)?);
                }
                _ => return Err(format!("unknown argument `{arg}`. Use --help for usage.")),
            }
        }
        Ok(options)
    }
}

fn print_help() {
    println!("{}", help_text());
}

fn help_text() -> &'static str {
    "Usage: cosimo-sync-server [--bind HOST:PORT] [--storage-root PATH] [--trusted-client-ip-header HEADER]\n\
     \n\
     This is the standalone Cosimo sync server. Bind it to loopback\n\
     or another private interface when using it behind a reverse proxy.\n\
     \n\
     Storage:\n\
       --storage-root PATH  Store metadata in SQLite and bundle files below\n\
                            this private storage directory. Use this for any\n\
                            real deployment.\n\
       no --storage-root    Use volatile in-memory storage for development or\n\
                            temporary testing only. Uploaded decks, login\n\
                            sessions, and OAuth state are lost when the process\n\
                            exits.\n\
     \n\
     The server supports Mastodon-compatible browser OAuth.\n\
     Public-service mode is the default. Registration is disabled until\n\
     controller and privacy notice information is supplied, unless\n\
     COSIMO_SYNC_SERVICE_MODE=domestic is set for non-public personal use.\n\
     \n\
     Limit environment variables:\n\
       COSIMO_SYNC_MAX_BUNDLE_BYTES\n\
       COSIMO_SYNC_MAX_BUNDLE_EXPANDED_BYTES\n\
       COSIMO_SYNC_MAX_BUNDLE_ENTRIES\n\
       COSIMO_SYNC_MAX_DECK_FILE_BYTES\n\
       COSIMO_SYNC_MAX_DECKS_PER_OWNER\n\
       COSIMO_SYNC_MAX_STORAGE_BYTES_PER_OWNER\n\
       COSIMO_SYNC_MAX_STORAGE_BYTES_TOTAL\n\
       COSIMO_SYNC_MAX_OAUTH_STARTS_PER_CLIENT_PER_HOUR\n\
       COSIMO_SYNC_MAX_OAUTH_STARTS_PER_INSTANCE_PER_HOUR\n\
       COSIMO_SYNC_MAX_PENDING_OAUTH_LOGINS_PER_INSTANCE\n\
       COSIMO_SYNC_MAX_CONCURRENT_REQUESTS\n\
       COSIMO_SYNC_SERVICE_MODE                 public or domestic\n\
       COSIMO_SYNC_CONTROLLER_NAME\n\
       COSIMO_SYNC_CONTROLLER_CONTACT           email, web form, or postal contact\n\
       COSIMO_SYNC_CONTROLLER_FEDI              optional additional Fediverse contact\n\
       COSIMO_SYNC_DPO_CONTACT                  optional\n\
       COSIMO_SYNC_RIGHTS_CONTACT               defaults to controller contact if omitted\n\
       COSIMO_SYNC_PRIVACY_NOTICE_VERSION\n\
       COSIMO_SYNC_DATA_CATEGORIES\n\
       COSIMO_SYNC_PROCESSING_PURPOSES\n\
       COSIMO_SYNC_LAWFUL_BASIS\n\
       COSIMO_SYNC_RECIPIENTS\n\
       COSIMO_SYNC_INTERNATIONAL_TRANSFERS\n\
       COSIMO_SYNC_RETENTION_POLICY\n\
       COSIMO_SYNC_COMPLAINT_CONTACT\n\
       COSIMO_SYNC_DOMESTIC_REGISTRATION_ALLOWLIST  comma-separated owner identities\n\
     \n\
     --trusted-client-ip-header is only honoured for loopback peers.\n\
     Use it only when a local reverse proxy strips any incoming copy\n\
     of that header and then supplies the real client IP itself."
}

fn handle_connection(
    mut stream: TcpStream,
    state: &ServerState,
    trusted_client_ip_header: Option<&str>,
) {
    let peer_address = stream.peer_addr().ok().map(|address| address.ip());
    let _ = stream.set_read_timeout(Some(Duration::from_secs(REQUEST_READ_TIMEOUT_SECONDS)));
    let _ = stream.set_write_timeout(Some(Duration::from_secs(RESPONSE_WRITE_TIMEOUT_SECONDS)));
    let response = match cosimo::sync_http::read_http_request_with_max_body_bytes(
        &mut stream,
        state.max_request_body_bytes(),
    ) {
        Ok(mut request) => {
            if let Some(client_address) =
                effective_client_address(&request, peer_address, trusted_client_ip_header)
            {
                request
                    .headers
                    .insert("x-cosimo-client-address".to_string(), client_address);
            }
            handle_request(state, request)
        }
        Err(error) => http_error_response(error),
    };
    if let Err(error) = stream.write_all(&response.to_http_bytes()) {
        eprintln!("could not write HTTP response: {error}");
    }
}

fn reject_busy_connection(mut stream: TcpStream) {
    let _ = stream.set_write_timeout(Some(Duration::from_secs(RESPONSE_WRITE_TIMEOUT_SECONDS)));
    let mut response = Response::text(
        Status::ServiceUnavailable,
        "sync web server is temporarily busy; too many requests are already in progress",
    );
    response.headers.insert(
        "retry-after".to_string(),
        SERVER_BUSY_RETRY_AFTER_SECONDS.to_string(),
    );
    if let Err(error) = stream.write_all(&response.to_http_bytes()) {
        eprintln!("could not write busy HTTP response: {error}");
    }
}

fn effective_client_address(
    request: &Request,
    peer_address: Option<IpAddr>,
    trusted_client_ip_header: Option<&str>,
) -> Option<String> {
    if let (Some(peer_address), Some(header_name)) = (peer_address, trusted_client_ip_header)
        && peer_address.is_loopback()
        && let Some(header_value) = request.header(header_name)
        && let Some(forwarded_address) = first_forwarded_ip_address(header_value)
    {
        return Some(forwarded_address);
    }
    peer_address.map(|address| address.to_string())
}

fn first_forwarded_ip_address(value: &str) -> Option<String> {
    let candidate = value.split(',').next()?.trim();
    if candidate.is_empty() {
        return None;
    }
    candidate
        .parse::<IpAddr>()
        .ok()
        .map(|address| address.to_string())
}

fn normalize_http_header_name(value: &str) -> Result<String, String> {
    let name = value.trim().to_ascii_lowercase();
    if name.is_empty() {
        return Err("trusted client IP header name must not be empty".to_string());
    }
    if !name
        .bytes()
        .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
    {
        return Err(format!("invalid trusted client IP header name `{value}`"));
    }
    Ok(name)
}

fn sync_limits_from_environment(
    environment: &BTreeMap<String, String>,
) -> Result<SyncServerLimits, String> {
    let mut limits = SyncServerLimits::default();
    limits.max_bundle_bytes =
        optional_u64_env(environment, "COSIMO_SYNC_MAX_BUNDLE_BYTES")?.or(limits.max_bundle_bytes);
    limits.max_bundle_expanded_bytes =
        optional_u64_env(environment, "COSIMO_SYNC_MAX_BUNDLE_EXPANDED_BYTES")?
            .or(limits.max_bundle_expanded_bytes);
    limits.max_bundle_entries = optional_usize_env(environment, "COSIMO_SYNC_MAX_BUNDLE_ENTRIES")?
        .or(limits.max_bundle_entries);
    limits.max_deck_file_bytes = optional_u64_env(environment, "COSIMO_SYNC_MAX_DECK_FILE_BYTES")?
        .or(limits.max_deck_file_bytes);
    limits.max_decks_per_owner =
        optional_usize_env(environment, "COSIMO_SYNC_MAX_DECKS_PER_OWNER")?;
    limits.max_storage_bytes_per_owner =
        optional_u64_env(environment, "COSIMO_SYNC_MAX_STORAGE_BYTES_PER_OWNER")?;
    limits.max_storage_bytes_total =
        optional_u64_env(environment, "COSIMO_SYNC_MAX_STORAGE_BYTES_TOTAL")?;
    limits.max_oauth_starts_per_client_per_hour = optional_usize_env(
        environment,
        "COSIMO_SYNC_MAX_OAUTH_STARTS_PER_CLIENT_PER_HOUR",
    )?
    .or(limits.max_oauth_starts_per_client_per_hour);
    limits.max_oauth_starts_per_instance_per_hour = optional_usize_env(
        environment,
        "COSIMO_SYNC_MAX_OAUTH_STARTS_PER_INSTANCE_PER_HOUR",
    )?
    .or(limits.max_oauth_starts_per_instance_per_hour);
    limits.max_pending_oauth_logins_per_instance = optional_usize_env(
        environment,
        "COSIMO_SYNC_MAX_PENDING_OAUTH_LOGINS_PER_INSTANCE",
    )?
    .or(limits.max_pending_oauth_logins_per_instance);
    Ok(limits)
}

fn max_concurrent_requests_from_environment(
    environment: &BTreeMap<String, String>,
) -> Result<usize, String> {
    Ok(
        optional_usize_env(environment, "COSIMO_SYNC_MAX_CONCURRENT_REQUESTS")?
            .unwrap_or(DEFAULT_MAX_CONCURRENT_REQUESTS)
            .max(1),
    )
}

fn sync_privacy_config_from_environment(
    environment: &BTreeMap<String, String>,
) -> Result<SyncServerPrivacyConfig, String> {
    let mut config = SyncServerPrivacyConfig::default();
    if let Some(mode) = optional_env_value(environment, "COSIMO_SYNC_SERVICE_MODE")? {
        config.mode = SyncServerServiceMode::from_code(mode)
            .ok_or_else(|| "COSIMO_SYNC_SERVICE_MODE must be public or domestic".to_string())?;
    }
    config.controller_name = optional_single_line_env(environment, "COSIMO_SYNC_CONTROLLER_NAME")?;
    config.controller_contact =
        optional_single_line_env(environment, "COSIMO_SYNC_CONTROLLER_CONTACT")?;
    config.controller_fedi = optional_single_line_env(environment, "COSIMO_SYNC_CONTROLLER_FEDI")?;
    config.dpo_contact = optional_single_line_env(environment, "COSIMO_SYNC_DPO_CONTACT")?;
    config.rights_contact = optional_single_line_env(environment, "COSIMO_SYNC_RIGHTS_CONTACT")?;
    config.notice_version =
        optional_single_line_env(environment, "COSIMO_SYNC_PRIVACY_NOTICE_VERSION")?;
    config.data_categories = optional_single_line_env(environment, "COSIMO_SYNC_DATA_CATEGORIES")?;
    config.purposes = optional_single_line_env(environment, "COSIMO_SYNC_PROCESSING_PURPOSES")?;
    config.lawful_basis = optional_single_line_env(environment, "COSIMO_SYNC_LAWFUL_BASIS")?;
    config.recipients = optional_single_line_env(environment, "COSIMO_SYNC_RECIPIENTS")?;
    config.international_transfers =
        optional_single_line_env(environment, "COSIMO_SYNC_INTERNATIONAL_TRANSFERS")?;
    config.retention = optional_single_line_env(environment, "COSIMO_SYNC_RETENTION_POLICY")?;
    config.complaint_contact =
        optional_single_line_env(environment, "COSIMO_SYNC_COMPLAINT_CONTACT")?;
    if let Some(allowlist) =
        optional_single_line_env(environment, "COSIMO_SYNC_DOMESTIC_REGISTRATION_ALLOWLIST")?
    {
        config.domestic_registration_allowlist = split_allowlist(&allowlist);
    }
    Ok(config)
}

fn optional_single_line_env(
    environment: &BTreeMap<String, String>,
    name: &'static str,
) -> Result<Option<String>, String> {
    let Some(value) = optional_env_value(environment, name)? else {
        return Ok(None);
    };
    if value.contains('\r') || value.contains('\n') {
        return Err(format!("{name} must fit on one line"));
    }
    Ok(Some(value.to_string()))
}

fn split_allowlist(value: &str) -> std::collections::BTreeSet<String> {
    value
        .split(',')
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
        .collect()
}

fn optional_u64_env(
    environment: &BTreeMap<String, String>,
    name: &'static str,
) -> Result<Option<u64>, String> {
    optional_env_value(environment, name)?
        .map(|value| {
            value
                .parse::<u64>()
                .map_err(|_| format!("{name} must be an unsigned integer"))
        })
        .transpose()
}

fn optional_usize_env(
    environment: &BTreeMap<String, String>,
    name: &'static str,
) -> Result<Option<usize>, String> {
    optional_env_value(environment, name)?
        .map(|value| {
            value
                .parse::<usize>()
                .map_err(|_| format!("{name} must be an unsigned integer"))
        })
        .transpose()
}

fn optional_env_value<'a>(
    environment: &'a BTreeMap<String, String>,
    name: &'static str,
) -> Result<Option<&'a str>, String> {
    let Some(value) = environment.get(name) else {
        return Ok(None);
    };
    let value = value.trim();
    if value.is_empty() {
        return Ok(None);
    }
    Ok(Some(value))
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::net::{IpAddr, Ipv4Addr};
    use std::time::{SystemTime, UNIX_EPOCH};

    #[test]
    fn options_parse_accepts_reverse_proxy_header() {
        let options = Options::parse([
            "--bind".to_string(),
            "127.0.0.1:44792".to_string(),
            "--storage-root".to_string(),
            "/tmp/cosimo-sync".to_string(),
            "--trusted-client-ip-header".to_string(),
            "X-Forwarded-For".to_string(),
        ])
        .unwrap();

        assert_eq!(options.bind_address, "127.0.0.1:44792");
        assert_eq!(
            options.storage_root,
            Some(PathBuf::from("/tmp/cosimo-sync"))
        );
        assert_eq!(
            options.trusted_client_ip_header.as_deref(),
            Some("x-forwarded-for")
        );
    }

    #[test]
    fn options_parse_rejects_invalid_proxy_header_names() {
        let error = Options::parse([
            "--trusted-client-ip-header".to_string(),
            "X Forwarded For".to_string(),
        ])
        .unwrap_err();

        assert!(error.contains("invalid trusted client IP header name"));
    }

    #[test]
    fn help_warns_that_missing_storage_root_is_volatile() {
        let text = help_text();

        assert!(text.contains("no --storage-root"));
        assert!(text.contains("volatile in-memory storage"));
        assert!(text.contains("lost when the process"));
        assert!(text.contains("Use this for any"));
    }

    #[test]
    fn sync_limits_are_read_from_environment() {
        let mut environment = BTreeMap::new();
        environment.insert(
            "COSIMO_SYNC_MAX_BUNDLE_BYTES".to_string(),
            "1024".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_BUNDLE_EXPANDED_BYTES".to_string(),
            "4096".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_BUNDLE_ENTRIES".to_string(),
            "12".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_DECK_FILE_BYTES".to_string(),
            "2048".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_DECKS_PER_OWNER".to_string(),
            "3".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_STORAGE_BYTES_PER_OWNER".to_string(),
            "8192".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_STORAGE_BYTES_TOTAL".to_string(),
            "16384".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_OAUTH_STARTS_PER_CLIENT_PER_HOUR".to_string(),
            "5".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_OAUTH_STARTS_PER_INSTANCE_PER_HOUR".to_string(),
            "9".to_string(),
        );
        environment.insert(
            "COSIMO_SYNC_MAX_PENDING_OAUTH_LOGINS_PER_INSTANCE".to_string(),
            "4".to_string(),
        );

        let limits = sync_limits_from_environment(&environment).unwrap();

        assert_eq!(limits.max_bundle_bytes, Some(1024));
        assert_eq!(limits.max_bundle_expanded_bytes, Some(4096));
        assert_eq!(limits.max_bundle_entries, Some(12));
        assert_eq!(limits.max_deck_file_bytes, Some(2048));
        assert_eq!(limits.max_decks_per_owner, Some(3));
        assert_eq!(limits.max_storage_bytes_per_owner, Some(8192));
        assert_eq!(limits.max_storage_bytes_total, Some(16384));
        assert_eq!(limits.max_oauth_starts_per_client_per_hour, Some(5));
        assert_eq!(limits.max_oauth_starts_per_instance_per_hour, Some(9));
        assert_eq!(limits.max_pending_oauth_logins_per_instance, Some(4));
    }

    #[test]
    fn sync_limits_reject_malformed_environment_values() {
        let mut environment = BTreeMap::new();
        environment.insert(
            "COSIMO_SYNC_MAX_DECKS_PER_OWNER".to_string(),
            "many".to_string(),
        );

        let error = sync_limits_from_environment(&environment).unwrap_err();

        assert!(error.contains("COSIMO_SYNC_MAX_DECKS_PER_OWNER"));
    }

    #[test]
    fn max_concurrent_requests_uses_default_and_clamps_zero() {
        let environment = BTreeMap::new();
        assert_eq!(
            max_concurrent_requests_from_environment(&environment).unwrap(),
            DEFAULT_MAX_CONCURRENT_REQUESTS
        );

        let mut environment = BTreeMap::new();
        environment.insert(
            "COSIMO_SYNC_MAX_CONCURRENT_REQUESTS".to_string(),
            "0".to_string(),
        );

        assert_eq!(
            max_concurrent_requests_from_environment(&environment).unwrap(),
            1
        );
    }

    #[test]
    fn request_gate_rejects_above_limit_and_releases_on_drop() {
        let gate = Arc::new(RequestGate::new(1));
        let permit = gate.try_enter().unwrap();

        assert!(gate.try_enter().is_none());
        assert_eq!(gate.active_count(), 1);
        drop(permit);

        assert_eq!(gate.active_count(), 0);
        assert!(gate.try_enter().is_some());
    }

    #[test]
    fn storage_lock_rejects_second_server_for_same_root() {
        let root = unique_temp_dir("sync-server-lock");
        let first = acquire_storage_lock(&root, "127.0.0.1:44791").unwrap();

        let second = acquire_storage_lock(&root, "127.0.0.1:44792");

        assert!(second.unwrap_err().contains("already using storage root"));
        drop(first);
        acquire_storage_lock(&root, "127.0.0.1:44792").unwrap();
    }

    #[cfg(unix)]
    #[test]
    fn storage_lock_rejects_symlink_lock_file() {
        use std::os::unix::fs::symlink;

        let root = unique_temp_dir("sync-server-lock-symlink");
        let outside = root.join("outside");
        fs::write(&outside, "outside").unwrap();
        symlink(&outside, root.join(STORAGE_LOCK_FILE_NAME)).unwrap();

        let error = acquire_storage_lock(&root, "127.0.0.1:44791").unwrap_err();

        assert!(error.contains("Too many levels of symbolic links") || error.contains("symlink"));
    }

    #[test]
    fn effective_client_address_uses_trusted_header_from_loopback_peer() {
        let request = Request {
            method: "GET".to_string(),
            target: "/".to_string(),
            headers: BTreeMap::from([(
                "x-forwarded-for".to_string(),
                "203.0.113.7, 127.0.0.1".to_string(),
            )]),
            body: Vec::new(),
        };

        let address = effective_client_address(
            &request,
            Some(IpAddr::V4(Ipv4Addr::LOCALHOST)),
            Some("x-forwarded-for"),
        );

        assert_eq!(address.as_deref(), Some("203.0.113.7"));
    }

    #[test]
    fn effective_client_address_ignores_trusted_header_from_non_loopback_peer() {
        let request = Request {
            method: "GET".to_string(),
            target: "/".to_string(),
            headers: BTreeMap::from([("x-forwarded-for".to_string(), "203.0.113.7".to_string())]),
            body: Vec::new(),
        };

        let address = effective_client_address(
            &request,
            Some(IpAddr::V4(Ipv4Addr::new(198, 51, 100, 4))),
            Some("x-forwarded-for"),
        );

        assert_eq!(address.as_deref(), Some("198.51.100.4"));
    }

    fn unique_temp_dir(prefix: &str) -> PathBuf {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path =
            std::env::temp_dir().join(format!("cosimo-{prefix}-{}-{nanos}", std::process::id()));
        fs::create_dir_all(&path).unwrap();
        path
    }
}
