fda 0.331.0

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

use crate::make_client;
use feldera_rest_api::types::{
    ClusterMonitorEventFieldSelector, CompilationProfile, MemberRole,
    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![];
    // Parse FELDERA_HOST / FELDERA_API_KEY from the environment.
    let Ok(cli) = Cli::try_parse_from(["fda", "pipelines"]) else {
        return completions;
    };
    // Never resolve `--auth-token-command` for completion: it would execute the
    // user's token command (e.g. `gcloud auth print-access-token`) on every Tab
    // press. Static API keys are cheap, so pass those through; token-command
    // users simply get no pipeline-name completion (a failed/anonymous list
    // returns nothing rather than panicking).
    let Ok(client) = make_client(
        cli.host,
        cli.insecure,
        cli.tls_cert,
        cli.auth,
        None,
        cli.timeout,
        cli.tenant,
    ) else {
        return completions;
    };

    let r = futures::executor::block_on(async {
        client
            .list_pipelines()
            .send()
            .await
            .map(|r| r.into_inner())
            .unwrap_or_default()
    });

    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>,
    /// Shell command that prints a bearer token on stdout.
    ///
    /// Run once per `fda` invocation; the trimmed stdout becomes the
    /// `Authorization: Bearer <token>` header for every request that
    /// invocation makes. Use for OIDC workload-identity flows with short-lived
    /// tokens: because the command runs on every invocation, a rotated token is
    /// picked up automatically. Conflicts with `--auth`.
    ///
    /// Examples:
    ///
    /// Kubernetes projected service-account token:
    /// `--auth-token-command 'cat /var/run/secrets/kubernetes.io/serviceaccount/token'`
    ///
    /// AWS EKS, IAM roles for service accounts:
    /// `--auth-token-command 'cat $AWS_WEB_IDENTITY_TOKEN_FILE'`
    ///
    /// Google Cloud, where an ID token is a JWT and an access token is not:
    /// `--auth-token-command 'gcloud auth print-identity-token'`
    #[arg(
        long,
        env = "FELDERA_AUTH_TOKEN_COMMAND",
        global = true,
        help_heading = "Global Options",
        conflicts_with = "auth"
    )]
    pub auth_token_command: 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>,
    /// The tenant to act in, by name or id, sent as the `Feldera-Tenant`
    /// header on every request.
    ///
    /// Needed when the credential may act in several tenants: a platform
    /// owner, or a user who belongs to more than one tenant. An API key is
    /// tenant-scoped and needs no selection.
    #[arg(
        long,
        env = "FELDERA_TENANT",
        global = true,
        help_heading = "Global Options"
    )]
    pub tenant: Option<String>,
}

#[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),
    /// Validate a SQL program by compiling it, without creating a pipeline.
    ///
    /// Prints, as a JSON object, whether the program is valid, along with any
    /// SQL compiler warnings and errors and the derived schema and connectors.
    /// A SQL compilation error is reported in the output, not as a command
    /// failure; the command fails only on a system error (e.g. an invalid
    /// runtime version or an unavailable compiler).
    ValidateProgram {
        /// Path to a file with the SQL program to validate. If omitted, read
        /// the program from stdin (pass `--stdin`).
        #[arg(value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        program_path: Option<String>,
        /// Read the SQL program from stdin.
        #[arg(long, default_value_t = false, conflicts_with = "program_path")]
        stdin: bool,
        /// Runtime version to compile with: a version tag (`vX.Y.Z`) or a
        /// 40-character git SHA. If omitted, the platform's default runtime is used.
        #[arg(long)]
        runtime_version: Option<String>,
        /// Also return the program IR (dataflow) in the output.
        #[arg(long, default_value_t = false)]
        ir: bool,
    },
    /// Manage API keys.
    Apikey {
        #[command(subcommand)]
        action: ApiKeyActions,
    },
    /// Manage OIDC trust relationships (workload identity federation).
    OidcTrust {
        #[command(subcommand)]
        action: OidcTrustActions,
    },
    /// Manage tenants across the installation (platform owner only).
    Tenant {
        #[command(subcommand)]
        action: TenantActions,
    },
    /// Manage the members of the acting tenant and their roles.
    ///
    /// Acts in the tenant named by `--tenant`, or in the one your credential
    /// resolves to without it.
    Member {
        #[command(subcommand)]
        action: MemberActions,
    },
    /// 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,
        /// Role the key carries: `read` (default) or `write`. The role may not
        /// exceed the caller's own role.
        #[arg(long, default_value = "read")]
        role: ApiKeyRole,
    },
    /// Delete an existing API key
    #[clap(aliases = &["del"])]
    Delete {
        /// The name of the API key to delete
        name: String,
    },
}

/// The roles an API key may carry.
#[derive(Clone, Copy, Debug, ValueEnum)]
pub enum ApiKeyRole {
    Read,
    Write,
}

#[derive(Subcommand)]
pub enum OidcTrustActions {
    /// List configured OIDC trust relationships.
    List,
    /// Register a new OIDC trust relationship.
    ///
    /// JWTs from `--issuer` whose `sub` claim matches `--subject` and (if
    /// specified) `aud` claim matches `--audience` are authorized as the
    /// current tenant. `*` is a wildcard.
    Create {
        /// Unique name for the trust relationship.
        name: String,
        /// OIDC issuer URL (must match the `iss` claim exactly).
        #[arg(long)]
        issuer: String,
        /// Pattern for the `sub` claim. `*` matches any sequence.
        #[arg(long)]
        subject: String,
        /// Pattern for the `aud` claim. `*` matches any sequence. Optional.
        #[arg(long)]
        audience: Option<String>,
        /// What this trust is for, e.g. the workload or CI job it authorizes.
        /// Shown when listing trust relationships.
        #[arg(long)]
        description: Option<String>,
        /// Role granted to a matching token: `read` (default), `write` or
        /// `admin`, capped at the caller's role.
        #[arg(long)]
        role: Option<TrustRole>,
    },
    /// Delete an OIDC trust relationship.
    #[clap(aliases = &["del"])]
    Delete {
        /// Name of the trust relationship to delete.
        name: String,
    },
}

/// The roles an OIDC trust relationship may grant.
#[derive(Clone, Copy, Debug, ValueEnum)]
pub enum TrustRole {
    Read,
    Write,
    Admin,
}

/// The roles a tenant membership may carry. `owner` is platform-wide rather
/// than a membership, so it is configured at deploy time and never assigned
/// here, which is why this cannot be a total mapping from the API's role type.
#[derive(Clone, Copy, Debug, ValueEnum)]
pub enum TenantMemberRole {
    Read,
    Write,
    Admin,
}

impl From<TenantMemberRole> for MemberRole {
    fn from(role: TenantMemberRole) -> Self {
        match role {
            TenantMemberRole::Read => MemberRole::Read,
            TenantMemberRole::Write => MemberRole::Write,
            TenantMemberRole::Admin => MemberRole::Admin,
        }
    }
}

/// Spelled as the API spells it, so printed output matches what the server
/// stores and what `fda member list` reads back.
impl Display for TenantMemberRole {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            TenantMemberRole::Read => "read",
            TenantMemberRole::Write => "write",
            TenantMemberRole::Admin => "admin",
        })
    }
}

#[derive(Subcommand)]
pub enum MemberActions {
    /// List the members of the acting tenant and their roles.
    List,
    /// Grant a user access to the acting tenant, by identity.
    ///
    /// Works before the user's first login: the membership authorizes as soon
    /// as that identity authenticates through the platform's identity
    /// provider. The role is capped at your own.
    Add {
        /// OIDC subject of the user, matching the `sub` claim their identity
        /// provider issues.
        subject: String,
        /// Role to grant: `read`, `write` or `admin`.
        #[arg(long)]
        role: TenantMemberRole,
        /// Email, shown in the member list. Optional.
        #[arg(long)]
        email: Option<String>,
    },
    /// Change a member's role in the acting tenant.
    SetRole {
        /// Identifier of the user, as shown by `fda member list`.
        user_id: Uuid,
        /// New role: `read`, `write` or `admin`.
        role: TenantMemberRole,
    },
    /// Remove a member from the acting tenant.
    ///
    /// Whether this alone revokes access depends on the deployment: where a
    /// login provisions memberships, the user is re-added at the default role
    /// on their next login unless their identity provider stops resolving this
    /// tenant for them.
    #[clap(aliases = &["rm"])]
    Remove {
        /// Identifier of the user, as shown by `fda member list`.
        user_id: Uuid,
    },
}

#[derive(Subcommand)]
pub enum TenantActions {
    /// List every tenant in the installation.
    List,
    /// Retrieve a single tenant by name or identifier.
    Get {
        /// The tenant's name, or its identifier as shown by `fda tenant list`.
        tenant: String,
    },
    /// Create a tenant, or return it if one with this name already exists.
    ///
    /// A login resolves its tenant by name, so a user whose identity provider
    /// asserts this name lands in the tenant created here.
    Create {
        /// The name of the tenant.
        name: String,
    },
    /// Rename a tenant.
    ///
    /// A login resolves its tenant by name, so the new name decides which users
    /// arrive in this tenant.
    Rename {
        /// Identifier of the tenant to rename, as shown by `fda tenant list`.
        tenant_id: Uuid,
        /// The new name.
        name: String,
        /// Take the name from the tenant that currently holds it, which is
        /// renamed to `<name> (<id>)` and keeps everything it had. Makes the
        /// rename atomic, so no request handled in between can re-create the
        /// name as a new tenant.
        #[arg(long)]
        displace_existing: bool,
    },
    /// Delete a tenant that holds no pipelines, API keys or OIDC trusts.
    #[clap(aliases = &["del"])]
    Delete {
        /// Identifier of the tenant to delete, as shown by `fda tenant list`.
        tenant_id: Uuid,
    },
}

#[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,
        /// Bootstrap new and modified views concurrently: keep serving the
        /// existing views while the changed views backfill in the background.
        ///
        /// Mutually exclusive with `--silent-bootstrap`.
        #[arg(long, default_value_t = false)]
        concurrent_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,
        /// Bootstrap new and modified views concurrently: keep serving the
        /// existing views while the changed views backfill in the background.
        ///
        /// Mutually exclusive with `--silent-bootstrap`.
        #[arg(long, default_value_t = false)]
        concurrent_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,
        /// Bootstrap new and modified views concurrently: keep serving the
        /// existing views while the changed views backfill in the background.
        ///
        /// Mutually exclusive with `--silent-bootstrap`.
        #[arg(long, default_value_t = false)]
        concurrent_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,
    },
    /// Compute the diff between the pipeline's current program and a proposed
    /// new version, without modifying or restarting the pipeline.
    ///
    /// Prints, as a JSON object, the tables, views, and connectors that would
    /// be added, removed, or modified. This is the same diff shown when
    /// approving changes during bootstrapping.
    Diff {
        /// The name of the pipeline.
        #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))]
        name: String,
        /// Path to a file with the new SQL program to compare against.
        /// If omitted (and `--stdin` is not set), the pipeline's current
        /// program code is used.
        #[arg(value_hint = ValueHint::FilePath, conflicts_with = "stdin")]
        program_path: Option<String>,
        /// Read the new SQL program from stdin.
        #[arg(long, default_value_t = false, conflicts_with = "program_path")]
        stdin: bool,
        /// Runtime version to compile the new program with: a version tag
        /// (`vX.Y.Z`) or a 40-character git SHA. If omitted, the platform's
        /// default runtime is used.
        #[arg(long)]
        runtime_version: Option<String>,
    },
    /// 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,
        /// Bootstrap new and modified views concurrently: keep serving the
        /// existing views while the changed views backfill in the background.
        ///
        /// Mutually exclusive with `--silent-bootstrap`.
        #[arg(long, default_value_t = false, requires("start"))]
        concurrent_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"
            ));
        }
    }
}