use clap::{Args, Parser, Subcommand, ValueEnum, ValueHint};
use clap_complete::engine::{ArgValueCompleter, CompletionCandidate};
use std::fmt::Display;
use std::path::PathBuf;

use crate::make_client;
use feldera_rest_api::types::{
    ClusterMonitorEventFieldSelector, CompilationProfile, PipelineMonitorEventFieldSelector,
};

/// Autocompletion for pipeline names by trying to fetch them from the server.
fn pipeline_names(current: &std::ffi::OsStr) -> Vec<CompletionCandidate> {
    let mut completions = vec![];
    // We parse FELDERA_HOST and FELDERA_API_KEY from the environment
    // using the `try_parse_from` method.
    let cli = Cli::try_parse_from(["fda", "pipelines"]);
    if let Ok(cli) = cli {
        let client =
            make_client(cli.host, cli.insecure, cli.tls_cert, cli.auth, cli.timeout).unwrap();

        let r = futures::executor::block_on(async {
            client
                .list_pipelines()
                .send()
                .await
                .map(|r| r.into_inner())
                .unwrap_or_else(|_| vec![])
        });

        let current = current.to_string_lossy();
        for pipeline in r {
            if pipeline.name.starts_with(current.as_ref()) {
                completions.push(CompletionCandidate::new(pipeline.name));
            }
        }
    }

    completions
}

#[derive(Parser)]
#[command(
    name = "fda",
    about = "A CLI to interact with the Feldera REST API.",
    after_help = "Commands marked EXPERIMENTAL may change or be removed at any time.",
    version
)]
pub struct Cli {
    #[command(subcommand)]
    pub command: Commands,
    /// The format in which the outputs from feldera should be displayed.
    ///
    /// Note that this flag may have no effect on some commands in case
    /// the requested output format is not supported for it.
    #[arg(
        long,
        env = "FELDERA_OUTPUT_FORMAT",
        global = true,
        help_heading = "Global Options",
        default_value = "text"
    )]
    pub format: OutputFormat,
    /// The Feldera host to connect to.
    #[arg(
        long,
        env = "FELDERA_HOST",
        value_hint = ValueHint::Url,
        global = true,
        help_heading = "Global Options",
        default_value_t = String::from("https://try.feldera.com")
    )]
    pub host: String,
    /// Accept invalid HTTPS certificates.
    #[arg(
        short = 'k',
        long,
        env = "FELDERA_TLS_INSECURE",
        global = true,
        default_value_t = false,
        help_heading = "Global Options"
    )]
    pub insecure: bool,
    /// Path to a PEM-encoded certificate to trust as an additional root
    /// certificate authority for HTTPS connections.
    ///
    /// Useful when `fda` talks to a Feldera deployment that serves HTTPS with
    /// a self-signed certificate or a certificate signed by a private CA that
    /// is not in the system trust store. The file must be readable by the
    /// current user and contain one or more PEM-encoded certificates.
    #[arg(
        long = "tls-cert",
        env = "FELDERA_HTTPS_TLS_CERT",
        value_hint = ValueHint::FilePath,
        global = true,
        help_heading = "Global Options",
        conflicts_with = "insecure"
    )]
    pub tls_cert: Option<std::path::PathBuf>,
    /// Which API key to use for authentication.
    ///
    /// The provided string should start with "apikey:" followed by the random characters.
    ///
    /// If not specified, a request without authentication will be used.
    #[arg(
        long,
        env = "FELDERA_API_KEY",
        global = true,
        hide_env_values = true,
        help_heading = "Global Options"
    )]
    pub auth: Option<String>,
    /// The client timeout for requests in seconds.
    ///
    /// In almost all cases you should not need to set this value, but it can
    /// be useful to limit the execution of certain commands (e.g., `query` or
    /// `logs`).
    ///
    /// By default, no timeout is set.
    #[arg(
        long,
        env = "FELDERA_REQUEST_TIMEOUT",
        global = true,
        help_heading = "Global Options"
    )]
    pub timeout: Option<u64>,
}

#[derive(ValueEnum, Clone, Copy, Debug, PartialEq)]
#[value(rename_all = "snake_case")]
pub enum OutputFormat {
    /// Return the output in a human-readable text format.
    Text,
    /// Return the output in JSON format.
    ///
    /// This usually corresponds to the exact response returned from the server.
    Json,
    /// Request the output in Arrow IPC format.
    ///
    /// This format can only be specified for SQL queries.
    ArrowIpc,
    /// Return the output in Parquet format.
    ///
    /// This format can only be specified for SQL queries.
    Parquet,
    /// Returns the output in Prometheus format.
    ///
    /// This format can only be specified for the `metrics` command.
    Prometheus,
    /// Returns a hash of the result instead of the result.
    ///
    /// This format can only be specified for ad-hoc SQL queries.
    /// The output in this case is a single string/line containing a SHA256 hash.
    Hash,
}

impl Display for OutputFormat {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let output = match self {
            OutputFormat::Text => "text",
            OutputFormat::Json => "json",
            OutputFormat::ArrowIpc => "arrow_ipc",
            OutputFormat::Parquet => "parquet",
            OutputFormat::Prometheus => "prometheus",
            OutputFormat::Hash => "hash",
        };
        write!(f, "{}", output)
    }
}

#[derive(Subcommand)]
pub enum Commands {
    /// List the available pipelines.
    #[command(next_help_heading = "Pipeline Commands")]
    Pipelines,
    /// Interact with a pipeline.
    ///
    /// If no sub-command is specified retrieves all configuration data for the pipeline.
    #[command(flatten)]
    Pipeline(PipelineAction),
    /// Manage API keys.
    Apikey {
        #[command(subcommand)]
        action: ApiKeyActions,
    },
    /// Cluster information and status.
    Cluster {
        #[command(subcommand)]
        action: ClusterAction,
    },
    /// EXPERIMENTAL: Debugging tools.
    Debug {
        #[command(subcommand)]
        action: DebugActions,
    },
}

#[derive(Subcommand)]
pub enum ApiKeyActions {
    /// List available API keys
    List,
    /// Create a new API key
    Create {
        /// The name of the API key to create
        name: String,
    },
    /// Delete an existing API key
    #[clap(aliases = &["del"])]
    Delete {
        /// The name of the API key to delete
        name: String,
    },
}

#[derive(Subcommand)]
pub enum ClusterAction {
    /// Retrieves all cluster events (status only) and prints them.
    Events,

    /// Retrieve specific cluster event.
    Event {
        /// Identifier (UUID) of the event or `latest`.
        id: String,
        /// Either `all` or `status` (default).
        #[arg(default_value = "status")]
        selector: ClusterMonitorEventFieldSelector,
    },
}

#[derive(Subcommand)]
pub enum DebugActions {
    /// EXPERIMENTAL: Print a MessagePack file, such as `steps.bin` in a checkpoint directory,
    /// to stdout.
    MsgpCat {
        /// The MessagePack file to read.
        #[arg(value_hint = ValueHint::FilePath)]
        path: PathBuf,
    },

    /// EXPERIMENTAL: Reads metrics from a file and prints them in an easier-to-read form.
    Metrics {
        /// The Prometheus metrics file to read.
        #[arg(value_hint = ValueHint::FilePath)]
        path: PathBuf,
    },

    /// EXPERIMENTAL: Re-creates the pipeline(s) found in a support bundle.
    Unbundle {
        /// Support Bundle Zip File.
        #[arg(value_hint = ValueHint::FilePath)]
        path: PathBuf,
        /// Only extract and show pipeline information without trying to create the pipelines.
        #[arg(long)]
        dry_run: bool,
        /// Overwrite pipelines if they already exist.
        #[arg(long)]
        force: bool,
    },
}

/// A list of possible configuration options.
#[derive(ValueEnum, Clone, Copy, Debug)]
#[value(rename_all = "snake_case")]
pub enum RuntimeConfigKey {
    Workers,
    Storage,
    FaultTolerance,
    CheckpointInterval,
    CpuProfiler,
    Tracing,
    TracingEndpointJaeger,
    MinBatchSizeRecords,
    MaxBufferingDelayUsecs,
    CpuCoresMin,
    CpuCoresMax,
    MemoryMbMin,
    MemoryMbMax,
    StorageMbMax,
    StorageClass,
    MinStorageBytes,
    ClockResolutionUsecs,
    Logging,
    HttpWorkers,
    IoWorkers,
    DevTweaks,
}

#[derive(Subcommand)]
pub enum PipelineAction {
    /// Create a new pipeline.
    Create {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// A path to a file containing the SQL code.
        ///
        /// If no path is provided, the pipeline will be created with an empty program.
        /// See the `stdin` flag for reading from stdin instead.
        #[arg(value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        program_path: Option<String>,
        /// A path to a file containing the Rust UDF functions.
        #[arg(short = 'u', long, value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        udf_rs: Option<String>,
        /// A path to the TOML file containing the dependencies for the UDF functions.
        #[arg(short = 't', long, value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        udf_toml: Option<String>,
        /// Override the runtime version of the pipeline.
        ///
        /// If not specified, the default version of the platform will be used.
        ///
        /// Note: This feature needs to be enabled in the platform configuration
        /// and is still in development. Use for testing purposes only.
        #[arg(long, short = 'r', env = "FELDERA_RUNTIME_VERSION")]
        runtime_version: Option<String>,
        /// Whether to use the SQL compiler from the runtime or the platform.
        ///
        /// This should usually be false, which is the default.  It is only meaningful
        /// when the runtime version is set.
        #[arg(long, env = "FELDERA_USE_PLATFORM_COMPILER", default_value_t = false)]
        use_platform_compiler: bool,
        /// The compilation profile to use.
        #[arg(default_value = "optimized")]
        profile: CompilationProfile,
        /// A tag to assign to the pipeline.
        ///
        /// Repeat the flag to assign several tags, e.g. `--tag prod --tag team-billing`.
        /// Tags are deduplicated and stored in sorted order.
        #[arg(long = "tag", value_hint = ValueHint::Other)]
        tags: Vec<String>,
        /// Read the program code from stdin.
        ///
        /// EXAMPLES:
        ///
        /// * cat program.sql | fda create p1 -s
        /// * echo "SELECT 1" | fda create p2 -s
        /// * fda program get p2 | fda create p3 -s
        #[arg(
            verbatim_doc_comment,
            short = 's',
            long,
            default_value_t = false,
            conflicts_with = "program_path"
        )]
        stdin: bool,
    },
    /// Copy a pipeline's program and configuration into a new pipeline.
    #[clap(aliases = &["clone"])]
    Copy {
        /// The name of the pipeline to copy from.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        source: String,
        /// The name of the new pipeline.
        destination: String,
    },
    /// Start a pipeline.
    ///
    /// If the pipeline is compiling it will wait for the compilation to finish.
    Start {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Force the recompilation of the pipeline before starting.
        ///
        /// This is useful for dev purposes in case the Feldera source-code has changed.
        #[arg(long, short = 'r', default_value_t = false)]
        recompile: bool,
        /// Don't wait for pipeline to reach the status before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
        /// Initial desired runtime status once the pipeline is started.
        #[arg(long, short = 'i', default_value = "running")]
        initial: String,
        /// The bootstrap policy to use.
        // TODO: auto-complete
        #[arg(long, short = 'b', default_value = "await_approval")]
        bootstrap_policy: String,
        /// Bootstrap the pipeline with output connectors disabled.
        #[arg(long, default_value_t = false)]
        silent_bootstrap: bool,
        /// Do not dismiss any deployment error before starting.
        #[arg(long, default_value_t = false)]
        no_dismiss_error: bool,
    },
    /// Approve pipeline changes. Called in the AwaitingApproval state to allow
    /// the pipeline to proceed with bootstrapping the modified components.
    Approve {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Bootstrap the pipeline with output connectors disabled.
        #[arg(long, default_value_t = false)]
        silent_bootstrap: bool,
    },

    /// Checkpoint a fault-tolerant pipeline.
    Checkpoint {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Don't wait for pipeline to complete the checkpoint.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
    },
    /// Pause a pipeline.
    Pause {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Don't wait for pipeline to reach the status before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
    },
    /// Resume a pipeline.
    Resume {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Don't wait for pipeline to reach the status before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
    },
    /// Stop a pipeline, then start it again.
    ///
    /// This is a shortcut for calling `fda stop p1 && fda start p1`.
    Restart {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Force the recompilation of the pipeline before starting.
        ///
        /// This is useful for dev purposes in case the Feldera source-code has changed.
        #[arg(long, short = 'r', default_value_t = false)]
        recompile: bool,
        /// Checkpoint the pipeline before restarting it.
        #[arg(long, short = 'c', default_value_t = false)]
        checkpoint: bool,
        /// Don't wait for pipeline to reach the status before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
        /// Initial desired runtime status once the pipeline is restarted.
        #[arg(long, short = 'i', default_value = "running")]
        initial: String,
        /// The bootstrap policy to use.
        #[arg(long, short = 'b', default_value = "await_approval")]
        bootstrap_policy: String,
        /// Bootstrap the pipeline with output connectors disabled.
        #[arg(long, default_value_t = false)]
        silent_bootstrap: bool,
        /// Do not dismiss any deployment error before starting.
        #[arg(long, default_value_t = false)]
        no_dismiss_error: bool,
    },
    /// Stop a pipeline.
    #[clap(aliases = &["shutdown"])]
    Stop {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Checkpoint the pipeline before stopping it.
        #[arg(long, short = 'c', default_value_t = false)]
        checkpoint: bool,
        /// Don't wait for pipeline to reach the status before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
    },
    /// Retrieve the entire state of a pipeline.
    ///
    /// EXAMPLES:
    ///
    /// - `fda status test | jq .program_info.schema`
    ///
    /// - `fda status test | jq .program_info.input_connectors`
    ///
    /// - `fda status test | jq .deployment_config`
    Status {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Retrieve the runtime statistics of a pipeline.
    #[clap(aliases = &["statistics"])]
    Stats {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Retrieve the pipeline metrics.
    ///
    /// Metrics are available in `json` and `prometheus` output formats.
    Metrics {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Retrieve the logs of a pipeline.
    #[clap(aliases = &["log"])]
    Logs {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Watch the log endpoint and emit new pipeline log messages as they are written.
        ///
        /// When `true`, the command will listen to pipeline logs until the pipeline terminates or
        /// the command is interrupted by the user. When `false`, the command outputs pipeline logs
        /// accumulated so far and exits.
        #[arg(long, short = 'w', default_value_t = false)]
        watch: bool,
    },
    /// Interact with the program of the pipeline.
    ///
    /// If no sub-command is specified retrieves the program.
    Program {
        #[command(subcommand)]
        action: ProgramAction,
    },
    /// Retrieve the runtime configuration of a pipeline.
    #[clap(aliases = &["cfg"])]
    Config {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Update the runtime configuration of a pipeline.
    #[clap(aliases = &["set-cfg"])]
    SetConfig {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The key of the configuration to update.
        key: RuntimeConfigKey,
        /// The new value for the configuration.
        value: String,
    },
    /// Retrieve the tags of a pipeline.
    ///
    /// Prints the tags as a comma-separated list, in sorted order.
    Tags {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Replace the tags of a pipeline.
    ///
    /// Takes the new tags as a single comma-separated list, replacing whatever the
    /// pipeline carried before; pass an empty list to clear all tags. To append
    /// instead, include the current tags, e.g.
    /// `fda set-tags my-pipeline $(fda tags my-pipeline),d,e`.
    ///
    /// A tag containing spaces must be quoted, e.g. `"team billing",prod`. Each tag
    /// may be named alone; its color is filled in from the same tag used elsewhere.
    SetTags {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The new tags, as a comma-separated list. Omit to clear all tags.
        #[arg(default_value = "")]
        tags: String,
    },
    /// Recompile a pipeline with the Feldera runtime version included in the
    /// currently installed Feldera platform.
    UpdateRuntime {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Delete a pipeline.
    #[clap(aliases = &["del"])]
    Delete {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Clears any associated storage first to force deletion of the pipeline.
        ///
        /// EXAMPLES:
        ///
        /// - fda delete --force my-pipeline
        ///
        /// Is equivalent to:
        ///
        /// - fda clear my-pipeline && fda delete my-pipeline
        #[arg(long, short = 'f')]
        force: bool,
    },
    /// Control an input connector belonging to a table of a pipeline.
    Connector {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The name of the table or view.
        relation_name: String,
        /// The name of the connector.
        connector_name: String,
        #[command(subcommand)]
        action: ConnectorAction,
    },
    /// Obtains a heap profile for a pipeline.
    ///
    /// By default, or with `--pprof`, this command retrieves the heap profile
    /// to a temporary file and then displays it with `pprof`. With `--output`,
    /// this command instead writes the heap profile to the specified file.
    ///
    /// Get `pprof` from <https://github.com/google/pprof>. There is at least
    /// one other program named `pprof` that is unrelated and will not work.
    HeapProfile {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The `pprof` command to run as a subprocess. The name of the `pprof`
        /// file will be provided as an additional command-line argument.
        #[arg(long, short = 'p', default_value = "pprof -http :")]
        pprof: String,
        /// The file to write the profile to.
        #[arg(value_hint = ValueHint::FilePath, long, short = 'o')]
        output: Option<PathBuf>,
    },
    /// Obtains a circuit profile for a pipeline.
    CircuitProfile {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The ZIP file to write the profile to.
        #[arg(value_hint = ValueHint::FilePath, long, short = 'o')]
        output: Option<PathBuf>,
    },
    /// Download a support bundle which contains debug information about the pipeline.
    SupportBundle {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The ZIP file to write the bundle to.
        #[arg(value_hint = ValueHint::FilePath, long, short = 'o')]
        output: Option<PathBuf>,
        /// Include at most N collections, starting with the most recent.
        /// With the default `collect=true`, `--limit 1` returns only the
        /// bundle gathered just now. Add `--no-collect` to instead return
        /// the N most recent previously stored collections.
        #[arg(long, short = 'n')]
        limit: Option<u64>,
        /// Do not collect fresh data; return only previously stored collections.
        #[arg(long)]
        no_collect: bool,
        /// Skip circuit profile collection.
        #[arg(long)]
        no_circuit_profile: bool,
        /// Skip heap profile collection.
        #[arg(long)]
        no_heap_profile: bool,
        /// Skip metrics collection.
        #[arg(long)]
        no_metrics: bool,
        /// Skip logs collection.
        #[arg(long)]
        no_logs: bool,
        /// Skip stats collection.
        #[arg(long)]
        no_stats: bool,
        /// Skip pipeline configuration collection.
        #[arg(long)]
        no_pipeline_config: bool,
        /// Skip system configuration collection.
        #[arg(long)]
        no_system_config: bool,
        /// Skip dataflow graph collection.
        #[arg(long)]
        no_dataflow_graph: bool,
        /// Skip pipeline monitor event collection.
        #[arg(long)]
        no_pipeline_events: bool,
    },
    /// Enter the ad-hoc SQL shell for a pipeline.
    Shell {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Start the pipeline before entering the shell.
        #[arg(long, short = 's', default_value_t = false)]
        start: bool,
        /// Initial desired runtime status once the pipeline is started
        /// (ignored unless `--start` is provided).
        #[arg(long, short = 'i', default_value = "running")]
        initial: String,
        /// The bootstrap policy to use.
        #[arg(long, short = 'b', default_value = "await_approval")]
        bootstrap_policy: String,
        /// Bootstrap the pipeline with output connectors disabled.
        #[arg(long, default_value_t = false, requires("start"))]
        silent_bootstrap: bool,
        /// Do not dismiss any deployment error before starting.
        #[arg(long, default_value_t = false, requires("start"))]
        no_dismiss_error: bool,
    },
    /// Execute an ad-hoc query against a pipeline and return the result.
    #[clap(aliases = &["exec"])]
    Query {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,

        /// The SQL query to execute against the pipeline.
        ///
        /// EXAMPLES:
        ///
        /// * fda exec p1 "SELECT 1;"
        #[arg(verbatim_doc_comment, conflicts_with = "stdin")]
        sql: Option<String>,

        /// Read the SQL query from stdin.
        ///
        /// EXAMPLES:
        ///
        /// * cat query.sql | fda exec p1 -s
        /// * echo "SELECT 1" | fda exec p1 -s
        #[arg(
            verbatim_doc_comment,
            short = 's',
            long,
            default_value_t = false,
            conflicts_with = "sql"
        )]
        stdin: bool,
    },
    /// Generate a completion token for a SQL table/connector pair in a pipeline.
    #[clap(aliases = &["generate-token", "generate-completion-token"])]
    CompletionToken {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The name of the SQL table to generate the token for.
        table: String,
        /// The name of the connector to generate the token for.
        ///
        /// This can be read from the `name` field in the connector config.
        connector: String,
    },
    /// Check the status of a completion token for a pipeline.
    #[clap(aliases = &["check-token", "check-completion-token"])]
    CompletionStatus {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The token to check the status for.
        ///
        /// A token can be optained by running `fda completion-token`.
        /// Or when ingesting data over HTTP.
        token: String,
    },
    /// Start a new transaction.
    #[clap(aliases = &["transaction-start"])]
    StartTransaction {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Commit the current transaction.
    ///
    /// Commits the currently active transaction for the specified pipeline.
    /// Optionally waits for the commit to complete.
    #[clap(aliases = &["commit", "transaction-commit"])]
    CommitTransaction {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The transaction ID to verify against the current active transaction.
        /// If provided, the function verifies that the currently active transaction matches this ID.
        #[arg(long = "tid", short = 't')]
        transaction_id: Option<u64>,
        /// Don't wait for the transaction to commit before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
    },
    /// Initiate rebalancing.
    Rebalance {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Advance the externally-driven `NOW()` clock by `delta_ms` and
    /// print the new value.
    ///
    /// Requires `dev_tweaks.now_http_driven = true` on the pipeline.
    /// The clock is forward-only; `delta_ms = 0` reads the current value
    /// without moving it; omitting `--delta-ms` advances by one
    /// `clock_resolution` (one configured tick).
    ///
    /// The printed value is the `NOW()` the worker will emit on its
    /// next pipeline step; ad-hoc queries against materialized views
    /// may still observe the previous value until that step completes.
    #[clap(aliases = &["clock-set", "set-clock"])]
    ClockAdvance {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Milliseconds to add to `NOW()`.  Omit to advance by one
        /// `clock_resolution`.
        #[arg(long)]
        delta_ms: Option<u64>,
    },
    /// Initiate compaction.
    StartCompaction {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Clear the storage resources of a pipeline.
    ///
    /// Note that the pipeline must be stopped before clearing its storage resources.
    Clear {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Don't wait for pipeline storage to clear before returning.
        #[arg(long, short = 'n', default_value_t = false)]
        no_wait: bool,
    },
    /// Benchmark the performance of a pipeline.
    ///
    /// This command will perform the following steps
    /// 1. ensure the benchmark is compiled with the latest platform version
    /// 2. run the pipeline & record stats over time
    /// 3. post-process and check recorded statistics
    /// 4. aggregate basic performance metrics and output them in the specified format
    Bench {
        #[command(flatten)]
        args: BenchmarkArgs,
    },
    /// Dismisses the deployment error of a pipeline.
    DismissError {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Retrieves all pipeline events and prints them.
    Events {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Either `all` or `status` (default).
        #[arg(default_value = "status")]
        selector: PipelineMonitorEventFieldSelector,
    },
    /// Retrieve specific pipeline event.
    Event {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Identifier (UUID) of the event or `latest`.
        event_id: String,
        /// Either `all` or `status` (default).
        #[arg(default_value = "status")]
        selector: PipelineMonitorEventFieldSelector,
    },
}

#[derive(Args, Debug)]
pub(crate) struct BenchmarkArgs {
    /// The name of the pipeline.
    #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
    pub name: String,
    /// Duration of the benchmark in seconds.
    ///
    /// If not specified, the benchmark will run until the pipeline indicates it has processed all input.
    #[arg(long, short = 'd')]
    pub duration: Option<u64>,

    /// Do not recompile the pipeline before starting.
    ///
    /// If set to true, one might end up with a pipeline that's not
    /// compiled with the latest feldera runtime.
    #[arg(long, short = 'n', default_value_t = false)]
    pub no_recompile: bool,

    /// Do not wrap the benchmark in a transaction.
    #[arg(long, default_value_t = false)]
    pub no_transaction: bool,

    /// If set upload results to feldera benchmark host.
    ///
    /// For development purposes, you most likely don't want to set this to true.
    ///
    /// Requires `benchmark_token` token to be set.
    #[arg(long, short = 'u', default_value_t = false)]
    pub upload: bool,

    /// Slug or UUID of the project to add results to when uploading
    /// (requires: `--upload`).
    #[arg(long, short = 'p', env = "BENCHER_PROJECT")]
    pub project: String,

    /// Branch name, slug, or UUID. By default it will be set to `main`
    /// (requires: `--upload`).
    #[arg(long, default_value_t = String::from("main"))]
    pub branch: String,

    /// Use the specified branch name as the start point for `branch`
    /// (requires: `--upload`).
    ///
    /// - If `branch` already exists and the start point is different, a new
    ///   branch will be created.
    #[arg(long)]
    pub start_point: Option<String>,

    /// Use the specified full `git` hash as the start point for `branch`
    /// (requires: `--start-point` and `--upload`).
    ///
    /// - If `start_point` already exists and the start point hash is different,
    ///   a new branch will be created
    #[arg(long)]
    pub start_point_hash: Option<String>,

    /// The maximum number of historical branch versions to include
    /// (requires: `--start-point` and `--upload`).
    ///
    /// Versions beyond this number will be omitted.
    #[arg(long, default_value_t = 255)]
    pub start_point_max_versions: u32,

    /// Clone thresholds from the start point branch
    /// (requires: `--branch-start-point` and `--upload`).
    #[arg(long)]
    pub start_point_clone_thresholds: bool,

    /// Reset the branch head to an empty state
    /// (requires: `--branch-start-point` and `--upload`).
    ///
    /// If `start_point` is specified, the new branch head will begin at that start point.
    /// Otherwise, the branch head will be reset to an empty state
    #[arg(long)]
    pub start_point_reset: bool,

    /// Where to upload benchmark results to
    /// (requires: `--upload`).
    #[arg(
        long,
        short = 'b',
        env = "BENCHER_HOST",
        value_hint = ValueHint::Url,
        default_value_t = String::from("https://benchmarks.feldera.io/")
    )]
    pub benchmark_host: String,

    /// Which API key to use for authentication with benchmarks server
    /// (requires: `--upload`).
    #[arg(long, short = 't', env = "BENCHER_API_TOKEN", hide_env_values = true)]
    pub benchmark_token: Option<String>,
}

#[derive(Subcommand)]
pub enum ProgramAction {
    /// Retrieve the program code.
    ///
    /// By default, this returns the SQL code, but you can use the flags to retrieve
    /// the Rust UDF code instead.
    Get {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Retrieve the Rust UDF code.
        #[arg(short = 'u', long, default_value_t = false)]
        udf_rs: bool,
        /// Retrieve the TOML dependencies file for the UDF code.
        #[arg(short = 't', long, default_value_t = false)]
        udf_toml: bool,
    },
    /// Sets a new program.
    Set {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// A path to a file containing the SQL code.
        ///
        /// See the `stdin` flag for reading from stdin instead.
        #[arg(value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        program_path: Option<String>,
        /// A path to a file containing the Rust UDF functions.
        #[arg(short = 'u', long, value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        udf_rs: Option<String>,
        /// A path to the TOML file containing the dependencies for the UDF functions.
        #[arg(short = 't', long, value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        udf_toml: Option<String>,
        /// Read the SQL program code from stdin.
        ///
        /// EXAMPLES:
        ///
        /// * cat program.sql | fda program set p1 -s
        /// * echo "SELECT 1" | fda program set p1 -s
        /// * fda program get p2 | fda program set p1 -s
        #[arg(verbatim_doc_comment, short = 's', long, default_value_t = false)]
        stdin: bool,
    },
    /// Retrieve the configuration of the program.
    #[clap(aliases = &["cfg"])]
    Config {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Set the configuration of the program.
    #[clap(aliases = &["set-cfg"])]
    SetConfig {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// The updated configuration for the pipeline.
        ///
        /// The profile accepts the following values:
        /// `dev`, `unoptimized`, `optimized`
        ///
        /// If not specified, the optimized profile will be used.
        #[arg(short = 'p', long)]
        profile: Option<CompilationProfile>,
        /// Override the runtime version of the pipeline.
        ///
        /// EXPERIMENTAL:
        ///
        /// This feature is still in development and may change in future releases.
        /// Use for testing purposes only. Note that currently no compatibility
        /// guarantees are provided in case the runtime version does not match
        /// the deployed platform version.
        ///
        /// EXAMPLES:
        ///
        ///  - --runtime-version v0.100.0
        ///  - --runtime-version 2880dd6fe206d10c966cc23868ee41a3c9e4e543
        ///    (valid git commit hash of feldera/feldera main branch)
        ///
        /// If not specified, the default version will be used.
        #[arg(verbatim_doc_comment, short = 'r', long)]
        runtime_version: Option<String>,
        /// Whether to use the SQL compiler from the runtime or the platform.
        ///
        /// This should usually be false, which is the default.  It is only meaningful
        /// when the runtime version is set.
        #[arg(long, env = "FELDERA_USE_PLATFORM_COMPILER", default_value_t = false)]
        use_platform_compiler: bool,
    },
    /// Retrieve the compilation status of the program.
    Status {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
    /// Retrieve program compilation errors and warnings.
    Errors {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
    },
}

#[derive(Subcommand)]
pub enum ConnectorAction {
    Start,
    #[clap(aliases = &["stop"])]
    Pause,
    #[clap(aliases = &["status"])]
    Stats,
}

#[cfg(test)]
mod tests {
    use crate::cli::{Cli, Commands, PipelineAction, ProgramAction};
    use clap::Parser;

    /// [clap] will panic inside `try_parse` if it finds anything invalid in the
    /// parser definition, such as a duplicated command name, so this test keeps
    /// really basic errors from passing through CI.
    #[test]
    fn basic_validation() {
        let _ = Cli::try_parse();
    }

    #[test]
    fn parse_program_errors_command() {
        let cli = Cli::try_parse_from(["fda", "program", "errors", "pipeline"])
            .expect("program errors command should parse");

        assert!(matches!(
            cli.command,
            Commands::Pipeline(PipelineAction::Program {
                action: ProgramAction::Errors { name }
            }) if name == "pipeline"
        ));
    }

    #[test]
    fn parse_pipeline_copy_command_and_alias() {
        for command in ["copy", "clone"] {
            let cli = Cli::try_parse_from(["fda", command, "source", "destination"])
                .expect("copy command should parse");

            assert!(matches!(
                cli.command,
                Commands::Pipeline(PipelineAction::Copy {
                    source,
                    destination
                }) if source == "source" && destination == "destination"
            ));
        }
    }
}
