/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */

use std::fs;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::OnceLock;

use flow_common::cli_overrides::CliOverrides;
use flow_common::options::Options;
use flow_common::options::SavedStateFetcher;
use flow_common::slow_to_check_logging::SlowToCheckLogging;
use flow_common::verbose::Verbose;
use flow_common_exit_status::FlowExitStatus;
use flow_daemon::ChannelPair;
use flow_daemon::Entry;
use flow_daemon::Handle;
use flow_daemon::StdioFd;
use flow_event_logger::LoggingContext;
use flow_server_files::server_files_js;

/// The fixed daemon entry name for the server master child process.
pub const SERVER_ENTRY_NAME: &str = "main_1";

pub type ServerOptionsBuilder = fn(&ServerDaemonArgs) -> Arc<Options>;

static SERVER_ENTRY: OnceLock<Entry<ServerEntryParam, (), ()>> = OnceLock::new();
static SERVER_OPTIONS_BUILDER: OnceLock<ServerOptionsBuilder> = OnceLock::new();

/// Serializable subset of `Options` needed to reconstruct the server child.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct ServerDaemonArgs {
    pub flowconfig_name: String,
    pub no_flowlib: bool,
    pub ignore_version: bool,
    pub include_suppressions: bool,
    pub all: bool,
    pub debug: bool,
    pub quiet: bool,
    pub lazy_mode: Option<String>,
    pub long_lived_workers: Option<bool>,
    pub max_workers: Option<i32>,
    pub wait_for_recheck: Option<bool>,
    pub profile: bool,
    pub verbose: Option<Verbose>,
    pub saved_state_fetcher: Option<SavedStateFetcher>,
    pub saved_state_force_recheck: bool,
    pub saved_state_no_fallback: bool,
    pub saved_state_skip_version_check: bool,
    pub saved_state_verify: bool,
    pub strip_root: bool,
    pub distributed: bool,
    pub estimate_recheck_time: Option<bool>,
    pub include_warnings: bool,
    pub merge_timeout: Option<i32>,
    pub munge_underscore_members: bool,
    pub slow_to_check_logging: SlowToCheckLogging,
    pub vpn_less: Option<bool>,
    pub cli_overrides: CliOverrides,
    pub root: PathBuf,
    pub temp_dir: String,
}

impl ServerDaemonArgs {
    pub fn of_options(
        options: &Options,
        lazy_mode: Option<String>,
        no_flowlib: bool,
        ignore_version: bool,
        cli_overrides: &CliOverrides,
    ) -> Self {
        Self {
            flowconfig_name: options.flowconfig_name.to_string(),
            no_flowlib,
            ignore_version,
            include_suppressions: options.include_suppressions,
            all: options.all,
            debug: options.debug,
            quiet: options.quiet,
            lazy_mode,
            long_lived_workers: Some(options.long_lived_workers),
            max_workers: Some(options.max_workers),
            wait_for_recheck: Some(options.wait_for_recheck),
            profile: options.profile,
            verbose: options
                .verbose
                .as_ref()
                .map(|verbose| verbose.as_ref().clone()),
            saved_state_fetcher: Some(options.saved_state_fetcher),
            saved_state_force_recheck: options.saved_state_force_recheck,
            saved_state_no_fallback: options.saved_state_no_fallback,
            saved_state_skip_version_check: options.saved_state_skip_version_check,
            saved_state_verify: options.saved_state_verify,
            strip_root: options.strip_root,
            distributed: options.distributed,
            estimate_recheck_time: Some(options.estimate_recheck_time),
            include_warnings: options.include_warnings,
            merge_timeout: options
                .merge_timeout
                .map(|merge_timeout| merge_timeout as i32),
            munge_underscore_members: options.munge_underscores,
            slow_to_check_logging: options.slow_to_check_logging,
            vpn_less: Some(options.vpn_less),
            cli_overrides: cli_overrides.clone(),
            root: (*options.root).clone(),
            temp_dir: options.temp_dir.to_string(),
        }
    }
}

/// Serializable arguments shipped to the server master daemon child.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct ServerEntryParam {
    pub daemon_args: ServerDaemonArgs,
    pub init_id: String,
    pub argv: Vec<String>,
    pub log_file: String,
    pub file_watcher_pid: Option<u32>,
    pub parent_pid: u32,
    pub logging_context: LoggingContext,
    pub start_cause: flow_server_env::server_status::StartCause,
}

pub fn entry_point(build: ServerOptionsBuilder) -> &'static Entry<ServerEntryParam, (), ()> {
    let _ = SERVER_OPTIONS_BUILDER.set(build);
    SERVER_ENTRY
        .get_or_init(|| flow_daemon::register_entry_point(SERVER_ENTRY_NAME, server_entry_handler))
}

pub fn register(build: ServerOptionsBuilder) {
    entry_point(build);
}

fn registered_entry_point() -> &'static Entry<ServerEntryParam, (), ()> {
    SERVER_ENTRY.get().unwrap_or_else(|| {
        panic!(
            "server entry point not registered; call flow_server::server_daemon::register before \
             daemon startup"
        )
    })
}

fn server_entry_handler(param: ServerEntryParam, pair: ChannelPair<(), ()>) {
    let ServerEntryParam {
        daemon_args,
        init_id,
        argv,
        log_file: _log_file,
        file_watcher_pid,
        parent_pid,
        logging_context,
        start_cause,
    } = param;
    let ChannelPair(in_chan, out_chan) = pair;

    flow_event_logger::restore_context(logging_context);
    flow_event_logger::set_command(Some("server".to_string()));
    flow_event_logger::init_flow_command(&init_id);

    let build = *SERVER_OPTIONS_BUILDER.get().unwrap_or_else(|| {
        panic!(
            "server options builder not registered; call flow_server::server_daemon::register \
             before daemon startup"
        )
    });
    let options = build(&daemon_args);

    flow_logging_utils::set_hh_logger_min_level(None, &options);
    flow_hh_logger::info!("argv={}", argv.join(" "));
    dump_server_options(&options);

    let pids_file = flow_server_files::server_files_js::pids_file(
        &options.flowconfig_name,
        options.temp_dir.as_str(),
        options.root.as_path(),
    );
    if let Err(e) = flow_daemon::pid_log::init(std::path::Path::new(&pids_file)) {
        flow_hh_logger::warn!("server: pid_log init failed: {}", e);
    }
    flow_daemon::pid_log::log(Some("monitor"), false, parent_pid);
    if let Some(pid) = file_watcher_pid {
        flow_daemon::pid_log::log(Some("file_watcher"), false, pid);
    }
    flow_daemon::pid_log::log(Some("main"), false, std::process::id());
    if let Some(pid) = flow_event_logger_common::event_logger::logger_pid() {
        flow_daemon::pid_log::log(Some("main_logger"), false, pid as u32);
    }

    let in_stream = flow_daemon::into_in_stream(in_chan);
    let out_stream = flow_daemon::into_out_stream(out_chan);
    let channels: flow_server_env::monitor_rpc::Channels = (in_stream, out_stream);

    crate::server::run_from_daemonize(&init_id, options, Some(channels), start_cause);
}

pub fn dump_server_options(options: &Options) {
    let lazy_mode = if options.lazy_mode { "on" } else { "off" };
    flow_hh_logger::info!("lazy_mode={}", lazy_mode);
    flow_hh_logger::info!("max_workers={}", options.max_workers);
    flow_hh_logger::info!("long_lived_workers={}", options.long_lived_workers);
    flow_hh_logger::info!("debug={}", options.debug);
    for (method_name, log_saving) in options.log_saving.iter() {
        let limit_str = match log_saving.limit {
            None => "null".to_string(),
            Some(limit) => format!("{}", limit),
        };
        flow_hh_logger::info!(
            "{} threshold_time_ms={} limit={} rate={}",
            method_name,
            log_saving.threshold_time_ms,
            limit_str,
            log_saving.rate
        );
    }
    for (r, g) in options.enabled_rollouts.iter() {
        flow_hh_logger::info!("Rollout {:?} set to {:?}", r, g);
    }
}

pub fn try_open_log_file(file: &str) -> Result<fs::File, String> {
    if Path::new(file).exists() {
        let old_file = format!("{}.old", file);
        if let Err(e) = (|| -> std::io::Result<()> {
            if Path::new(&old_file).exists() {
                fs::remove_file(&old_file)?;
            }
            fs::rename(file, &old_file)?;
            Ok(())
        })() {
            eprintln!(
                "Log rotate: failed to move '{}' to '{}'\n{}",
                file, old_file, e
            );
        }
    }
    fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(file)
        .map_err(|e| format!("Failed to open log file '{}': {}", file, e))
}

pub fn open_log_file(file: &str) -> fs::File {
    try_open_log_file(file).unwrap_or_else(|e| panic!("{}", e))
}
pub fn daemonize(
    init_id: &str,
    log_file: &str,
    argv: &[String],
    lazy_mode: Option<String>,
    no_flowlib: bool,
    ignore_version: bool,
    options: Arc<Options>,
    file_watcher_pid: Option<u32>,
    start_cause: flow_server_env::server_status::StartCause,
    cli_overrides: &CliOverrides,
) -> Result<Handle<(), ()>, String> {
    let entry = registered_entry_point();

    let root = &options.root;
    let tmp_dir = &options.temp_dir;
    let flowconfig_name = &options.flowconfig_name;
    let lock = server_files_js::lock_file(flowconfig_name, tmp_dir, root);
    if !flow_common::lock::check(&lock) {
        let msg = format!(
            "Error: There is already a server running for {}",
            root.display()
        );
        flow_common_exit_status::exit_with_msg(FlowExitStatus::LockStolen, &msg);
    }

    let name = format!("server master process watching {}", root.display());
    let param = ServerEntryParam {
        daemon_args: ServerDaemonArgs::of_options(
            &options,
            lazy_mode,
            no_flowlib,
            ignore_version,
            cli_overrides,
        ),
        init_id: init_id.to_string(),
        argv: argv.to_vec(),
        log_file: log_file.to_string(),
        file_watcher_pid,
        parent_pid: std::process::id(),
        logging_context: flow_event_logger::get_context(),
        start_cause,
    };

    let stdio = (
        StdioFd::Owned(flow_daemon::null_fd()),
        StdioFd::Owned(try_open_log_file(log_file)?),
        StdioFd::Owned(try_open_log_file(log_file)?),
    );

    flow_daemon::spawn(None, Some(&name), stdio, entry, param)
        .map_err(|e| format!("Failed to spawn server daemon '{}': {}", name, e))
}
