use crate::api::demo::{read_demos_from_directories, Demo};
use crate::api::endpoints;
use crate::api::support_data_collector::SupportDataCollector;
use crate::auth::JwkCache;
use crate::config::{ApiServerConfig, CommonConfig};
use crate::db::probe::DbProbe;
use crate::db::storage_postgres::StoragePostgres;
use crate::error::ManagerError;
use crate::license::LicenseCheck;
use crate::runner::interaction::RunnerInteraction;
use crate::unstable_features;
use actix_http::body::BoxBody;
use actix_http::StatusCode;
use actix_web::body::MessageBody;
use actix_web::dev::{Service, ServiceResponse};
use actix_web::http::{header, Method};
use actix_web::Scope;
use actix_web::{
    get, middleware,
    web::Data as WebData,
    web::{self},
    App, HttpResponse, HttpServer,
};
use actix_web_httpauth::middleware::HttpAuthentication;
use actix_web_static_files::ResourceFiles;
use anyhow::Result as AnyResult;
use feldera_observability as observability;
use futures_util::FutureExt;
use std::io::Write;
use std::time::Duration;
use std::{env, io, net::TcpListener, sync::Arc};
use termbg::{theme, Theme};
use tokio::signal;
use tokio::sync::watch;
use tokio::sync::{Mutex, RwLock};
use tracing::{error, info, trace, Level};
use utoipa::openapi::security::{HttpAuthScheme, HttpBuilder, SecurityScheme};
use utoipa::{Modify, OpenApi};
use utoipa_swagger_ui::SwaggerUi;

macro_rules! log_with_level {
    ($level:expr, $($arg:tt)+) => {
        match $level {
            tracing::Level::TRACE => tracing::trace!($($arg)+),
            tracing::Level::DEBUG => tracing::debug!($($arg)+),
            tracing::Level::INFO => tracing::info!($($arg)+),
            tracing::Level::WARN => tracing::warn!($($arg)+),
            tracing::Level::ERROR => tracing::error!($($arg)+),
        }
    };
}

#[derive(OpenApi)]
#[openapi(
    modifiers(&SecurityAddon),
    info(
        title = "Feldera API",
        description = r#"
With Feldera, users create data pipelines out of SQL programs.
A SQL program comprises tables and views, and includes as well the definition of
input and output connectors for each respectively. A connector defines a data
source or data sink to feed input data into tables or receive output data
computed by the views respectively.

## Pipeline

The API is centered around the **pipeline**, which most importantly consists
out of the SQL program, but also has accompanying metadata and configuration parameters
(e.g., compilation profile, number of workers, etc.).

* A pipeline is identified and referred to by its user-provided unique name.
* The pipeline program is asynchronously compiled when the pipeline is first created or
  when its program is subsequently updated.
* Pipeline deployment start is only able to proceed to provisioning once the program is successfully
  compiled.
* A pipeline cannot be updated while it is deployed.

## Concurrency

Each pipeline has a version, which is incremented each time its core fields are updated.
The version is monotonically increasing. There is additionally a program version which covers
only the program-related core fields, and is used by the compiler to discern when to recompile.

## Client request handling

### Request outcome expectations

The outcome of a request is that it either fails (e.g., DNS lookup failed) without any response
(no status code nor body), or it succeeds and gets back a response status code and body.

In case of a response, usually it is the Feldera endpoint that generated it:
- If it is success (2xx), it will return whichever body belongs to the success response.
- Otherwise, if it is an error (4xx, 5xx), it will return a Feldera error response JSON body
  which will have an application-level `error_code`.

However, there are two notable exceptions when the response is not generated by the Feldera
endpoint:
- If the HTTP server, to which the endpoint belongs, encountered an issue, it might return
  4xx (e.g., for an unknown endpoint) or 5xx error codes by itself (e.g., when it is initializing).
- If the Feldera API server is behind a (reverse) proxy, the proxy can return error codes by itself,
  for example BAD GATEWAY (502) or GATEWAY TIMEOUT (504).

As such, it is not guaranteed that the (4xx, 5xx) will have a Feldera error response JSON body
in these latter cases.

### Error handling and retrying

The error type returned by the client should distinguish between the error responses generated
by Feldera endpoints themselves (which have a Feldera error response body) and those that are
generated by other sources.

In order for a client operation (e.g., `pipeline.resume()`) to be robust (i.e., not fail due to
a single HTTP request not succeeding) the client should use a retry mechanism if the operation
is idempotent. The retry mechanism must however have a time limit, after which it times out.
This guarantees that the client operation is eventually responsive, which enables the script
it is a part of to not hang indefinitely on Feldera operations and instead be able to decide
by itself whether and how to proceed. If no response is returned, the mechanism should generally
retry. When a response is returned, the decision whether to retry can generally depend on the status
code: especially the status codes 408, 502, 503 and 504 should be considered as transient errors.
Finer grained retry decisions should be made by taking into account the application-level
`error_code` if the response body was indeed a Feldera error response body.

## Feldera client errors (4xx)

_Client behavior:_ clients should generally return with an error when they get back a 4xx status
code, as it usually means the request will likely not succeed even if it is sent again. Certain
requests might make use of a timed retry mechanism when the client error is transient without
requiring any user intervention to overcome, for instance a transaction already being in progress
leading to a temporary CONFLICT (409) error.

- **BAD REQUEST (400)**: invalid user request (general).
  - _Example:_ the new pipeline name `example1@~` contains invalid characters.

- **UNAUTHORIZED (401)**: the user is not authorized to issue the request.
  - _Example:_ an invalid API key is provided.

- **NOT FOUND (404)**: a resource required to exist in order to process the request was not found.
  - _Example:_ a pipeline named `example` does not exist when trying to update it.

- **CONFLICT (409)**: there is a conflict between the request and a relevant resource.
  - _Example:_ a pipeline named `example` already exists.
  - _Example:_ another transaction is already in process.

## Feldera server errors (5xx)

- **INTERNAL SERVER ERROR (500)**: the server is unexpectedly unable to process the request
  (general).
  - _Example:_ unable to reach the database.
  - _Client behavior:_ immediately return with an error.

- **NOT IMPLEMENTED (501)**: the server does not implement functionality required to process the
  request.
  - _Example:_ making a request to an enterprise-only endpoint in the OSS edition.
  - _Client behavior:_ immediately return with an error.

- **SERVICE UNAVAILABLE (503)**: the server is not (yet) able to process the request.
  - _Example:_ pausing a pipeline which is still provisioning.
  - _Client behavior:_ depending on the type of request, client may use a timed retry mechanism.

## Feldera error response body

When the Feldera API returns an HTTP error status code (4xx, 5xx), the body will contain the
following JSON object:

```json
{
  "message": "Human-readable explanation.",
  "error_code": "CodeSpecifyingError",
  "details": {

  }
}
```

It contains the following fields:
- **message (string)**: human-readable explanation of the error that occurred and potentially
  hinting what can be done about it.
- **error_code (string)**: application-level code about the error that occurred, written in CamelCase.
  For example: `UnknownPipelineName`, `DuplicateName`, `PauseWhileNotProvisioned`, ... .
- **details (object)**: JSON object corresponding to the `error_code` with fields that provide
  details relevant to it. For example: if a name is unknown, a field with the unknown name in
  question.
"#
    ),
    paths(
        // Pipeline management endpoints
        endpoints::pipeline_management::list_pipelines,
        endpoints::pipeline_management::get_pipeline,
        endpoints::pipeline_management::post_pipeline,
        endpoints::pipeline_management::put_pipeline,
        endpoints::pipeline_management::patch_pipeline,
        endpoints::pipeline_management::post_pipeline_testing,
        endpoints::pipeline_management::post_update_runtime,
        endpoints::pipeline_management::delete_pipeline,
        endpoints::pipeline_management::post_pipeline_start,
        endpoints::pipeline_management::post_pipeline_stop,
        endpoints::pipeline_management::post_pipeline_dismiss_error,
        endpoints::pipeline_management::post_pipeline_clear,
        endpoints::pipeline_management::get_pipeline_logs,

        // Pipeline interaction endpoints
        endpoints::pipeline_interaction::http_input,
        endpoints::pipeline_interaction::http_output,
        endpoints::pipeline_interaction::post_pipeline_input_connector_action,
        endpoints::pipeline_interaction::get_pipeline_input_connector_status,
        endpoints::pipeline_interaction::get_pipeline_output_connector_status,
        endpoints::pipeline_interaction::get_pipeline_stats,
        endpoints::pipeline_interaction::get_pipeline_metrics,
        endpoints::pipeline_interaction::get_pipeline_circuit_profile,
        endpoints::pipeline_interaction::get_pipeline_circuit_json_profile,
        endpoints::pipeline_interaction::get_pipeline_dataflow_graph,
        endpoints::pipeline_interaction::get_pipeline_heap_profile,
        endpoints::pipeline_interaction::support_bundle::get_pipeline_support_bundle,
        endpoints::pipeline_interaction::pipeline_adhoc_sql,
        endpoints::pipeline_interaction::checkpoint_pipeline,
        endpoints::pipeline_interaction::get_checkpoint_status,
        endpoints::pipeline_interaction::sync_checkpoint,
        endpoints::pipeline_interaction::get_checkpoint_sync_status,
        endpoints::pipeline_interaction::get_checkpoints,
        endpoints::pipeline_interaction::post_pipeline_pause,
        endpoints::pipeline_interaction::post_pipeline_resume,
        endpoints::pipeline_interaction::post_pipeline_activate,
        endpoints::pipeline_interaction::get_pipeline_samply_profile,
        endpoints::pipeline_interaction::start_samply_profile,
        endpoints::pipeline_interaction::post_pipeline_approve,
        endpoints::pipeline_interaction::completion_token,
        endpoints::pipeline_interaction::completion_status,
        endpoints::pipeline_interaction::start_transaction,
        endpoints::pipeline_interaction::commit_transaction,
        endpoints::pipeline_interaction::get_pipeline_time_series,
        endpoints::pipeline_interaction::get_pipeline_time_series_stream,
        endpoints::pipeline_interaction::post_pipeline_rebalance,

        // API keys
        endpoints::api_key::list_api_keys,
        endpoints::api_key::get_api_key,
        endpoints::api_key::post_api_key,
        endpoints::api_key::delete_api_key,

        // Configuration
        endpoints::config::get_config_authentication,
        endpoints::config::get_config_demos,
        endpoints::config::get_config_session,
        endpoints::config::get_config,

        // Metrics
        endpoints::metrics::get_metrics,

        // Cluster
        endpoints::cluster::list_cluster_events,
        endpoints::cluster::get_cluster_event,
        endpoints::cluster::get_cluster_health
    ),
    components(schemas(
        // Authentication
        crate::auth::AuthProvider,
        crate::auth::ProviderAwsCognito,
        crate::auth::ProviderGenericOidc,

        // Common
        crate::db::types::version::Version,
        crate::db::types::tenant::TenantId,
        crate::license::LicenseValidity,
        crate::api::endpoints::config::UpdateInformation,
        crate::api::endpoints::config::Configuration,
        crate::api::endpoints::config::BuildInformation,
        crate::api::endpoints::config::SessionInfo,

        // Pipeline
        crate::db::types::pipeline::PipelineId,
        crate::db::types::combined_status::CombinedStatus,
        crate::db::types::combined_status::CombinedDesiredStatus,
        crate::db::types::resources_status::ResourcesStatus,
        crate::db::types::resources_status::ResourcesDesiredStatus,
        feldera_types::pipeline_diff::ProgramDiff,
        feldera_types::pipeline_diff::PipelineDiff,
        feldera_types::runtime_status::RuntimeStatus,
        feldera_types::runtime_status::RuntimeDesiredStatus,
        feldera_types::runtime_status::BootstrapPolicy,
        crate::api::endpoints::pipeline_management::ConnectorStats,
        crate::api::endpoints::pipeline_management::PipelineInfo,
        crate::api::endpoints::pipeline_management::PipelineSelectedInfo,
        crate::api::endpoints::pipeline_management::PipelineFieldSelector,
        crate::api::endpoints::pipeline_management::GetPipelineParameters,
        crate::api::endpoints::pipeline_management::PostPutPipeline,
        crate::api::endpoints::pipeline_management::PatchPipeline,
        crate::api::endpoints::pipeline_management::PostStopPipelineParameters,

        // Dataflow IR
        feldera_ir::Dataflow,
        feldera_ir::MirNode,
        feldera_ir::CalcitePlan,
        feldera_ir::CalciteId,
        feldera_ir::MirInput,
        feldera_ir::Rel,
        feldera_ir::Operand,
        feldera_ir::Op,
        feldera_ir::Condition,
        feldera_types::config::ProgramIr,

        // Storage
        crate::db::types::storage::StorageStatus,

        // Demo
        crate::api::demo::Demo,

        // Program
        crate::db::types::program::CompilationProfile,
        crate::db::types::program::SqlCompilerMessage,
        crate::db::types::program::ProgramStatus,
        crate::db::types::program::ProgramError,
        crate::db::types::program::SqlCompilationInfo,
        crate::db::types::program::RustCompilationInfo,
        crate::db::types::program::ProgramConfig,
        crate::db::types::program::ProgramInfo,
        crate::api::endpoints::pipeline_management::PartialProgramInfo,

        // API key
        crate::db::types::api_key::ApiKeyId,
        crate::db::types::api_key::ApiPermission,
        crate::db::types::api_key::ApiKeyDescr,
        crate::api::endpoints::api_key::NewApiKeyRequest,
        crate::api::endpoints::api_key::NewApiKeyResponse,

        // Monitor
        crate::db::types::monitor::MonitorStatus,
        crate::db::types::monitor::ClusterMonitorEventId,
        crate::db::types::monitor::ClusterMonitorEvent,
        crate::db::types::monitor::ExtendedClusterMonitorEvent,
        crate::api::endpoints::cluster::ClusterMonitorEventSelectedInfo,
        crate::api::endpoints::cluster::GetClusterEventParameters,
        crate::api::endpoints::cluster::ClusterMonitorEventFieldSelector,
        crate::api::endpoints::cluster::HealthStatus,
        crate::api::endpoints::cluster::ServiceStatus,

        // From the feldera-types crate
        feldera_types::config::PipelineConfig,
        feldera_types::config::MultihostConfig,
        feldera_types::config::StorageConfig,
        feldera_types::config::StorageCacheConfig,
        feldera_types::config::StorageOptions,
        feldera_types::config::StorageBackendConfig,
        feldera_types::config::SyncConfig,
        feldera_types::config::StartFromCheckpoint,
        feldera_types::config::FileBackendConfig,
        feldera_types::config::StorageCompression,
        feldera_types::config::RuntimeConfig,
        feldera_types::config::FtConfig,
        feldera_types::config::InputEndpointConfig,
        feldera_types::config::ConnectorConfig,
        feldera_types::config::OutputBufferConfig,
        feldera_types::config::OutputEndpointConfig,
        feldera_types::config::TransportConfig,
        feldera_types::config::FormatConfig,
        feldera_types::config::ResourceConfig,
        feldera_types::config::ObjectStorageConfig,
        feldera_types::config::FtModel,
        feldera_types::config::PipelineTemplateConfig,
        feldera_types::config::DevTweaks,
        feldera_types::config::dev_tweaks::BufferCacheStrategy,
        feldera_types::config::dev_tweaks::BufferCacheAllocationStrategy,
        feldera_types::config::dev_tweaks::MergerType,
        feldera_types::transport::adhoc::AdHocInputConfig,
        feldera_types::transport::clock::ClockConfig,
        feldera_types::transport::file::FileInputConfig,
        feldera_types::transport::file::FileOutputConfig,
        feldera_types::transport::http::HttpInputConfig,
        feldera_types::transport::url::UrlInputConfig,
        feldera_types::transport::kafka::KafkaHeader,
        feldera_types::transport::kafka::KafkaHeaderValue,
        feldera_types::transport::kafka::KafkaLogLevel,
        feldera_types::transport::kafka::KafkaInputConfig,
        feldera_types::transport::kafka::KafkaOutputConfig,
        feldera_types::transport::kafka::KafkaOutputFtConfig,
        feldera_types::transport::kafka::KafkaStartFromConfig,
        feldera_types::transport::nats::Auth,
        feldera_types::transport::nats::ConnectOptions,
        feldera_types::transport::nats::ConsumerConfig,
        feldera_types::transport::nats::Credentials,
        feldera_types::transport::nats::DeliverPolicy,
        feldera_types::transport::nats::NatsInputConfig,
        feldera_types::transport::nats::ReplayPolicy,
        feldera_types::transport::nats::UserAndPassword,
        feldera_types::transport::pubsub::PubSubInputConfig,
        feldera_types::transport::s3::S3InputConfig,
        feldera_types::transport::datagen::DatagenStrategy,
        feldera_types::transport::datagen::RngFieldSettings,
        feldera_types::transport::datagen::GenerationPlan,
        feldera_types::transport::datagen::DatagenInputConfig,
        feldera_types::transport::nexmark::NexmarkInputConfig,
        feldera_types::transport::nexmark::NexmarkTable,
        feldera_types::transport::nexmark::NexmarkInputOptions,
        feldera_types::transport::delta_table::DeltaTableIngestMode,
        feldera_types::transport::delta_table::DeltaTableTransactionMode,
        feldera_types::transport::delta_table::DeltaTableWriteMode,
        feldera_types::transport::delta_table::DeltaTableReaderConfig,
        feldera_types::transport::delta_table::DeltaTableWriterConfig,
        feldera_types::transport::iceberg::IcebergReaderConfig,
        feldera_types::transport::iceberg::IcebergIngestMode,
        feldera_types::transport::iceberg::IcebergCatalogType,
        feldera_types::transport::iceberg::RestCatalogConfig,
        feldera_types::transport::iceberg::GlueCatalogConfig,
        feldera_types::transport::postgres::PostgresReaderConfig,
        feldera_types::transport::postgres::PostgresWriterConfig,
        feldera_types::transport::postgres::PostgresWriteMode,
        feldera_types::transport::postgres::PostgresTlsConfig,
        feldera_types::transport::redis::RedisOutputConfig,
        feldera_types::transport::http::Chunk,
        feldera_types::transport::clock::ClockConfig,
        feldera_types::query::AdhocQueryArgs,
        feldera_types::query::AdHocResultFormat,
        feldera_types::format::json::JsonUpdateFormat,
        feldera_types::format::json::JsonLines,
        feldera_types::preprocess::PreprocessorConfig,
        feldera_types::program_schema::ProgramSchema,
        feldera_types::program_schema::Relation,
        feldera_types::program_schema::SqlType,
        feldera_types::program_schema::Field,
        feldera_types::program_schema::ColumnType,
        feldera_types::program_schema::IntervalUnit,
        feldera_types::program_schema::SourcePosition,
        feldera_types::program_schema::PropertyValue,
        feldera_types::program_schema::SqlIdentifier,
        feldera_types::query_params::MetricsFormat,
        feldera_types::query_params::MetricsParameters,
        feldera_types::error::ErrorResponse,
        feldera_types::completion_token::CompletionTokenResponse,
        feldera_types::completion_token::CompletionStatusArgs,
        feldera_types::completion_token::CompletionStatus,
        feldera_types::completion_token::CompletionStatusResponse,
        feldera_types::checkpoint::CheckpointStatus,
        feldera_types::checkpoint::CheckpointSyncStatus,
        feldera_types::checkpoint::CheckpointResponse,
        feldera_types::checkpoint::CheckpointFailure,
        feldera_types::checkpoint::CheckpointSyncFailure,
        feldera_types::checkpoint::CheckpointMetadata,
        feldera_types::transaction::StartTransactionResponse,
        feldera_types::transaction::CommitProgressSummary,
        feldera_types::time_series::TimeSeries,
        feldera_types::time_series::SampleStatistics,
        feldera_types::suspend::SuspendError,
        feldera_types::suspend::PermanentSuspendError,
        feldera_types::suspend::TemporarySuspendError,

        // Adapter statistics
        feldera_types::adapter_stats::ExternalControllerStatus,
        feldera_types::memory_pressure::MemoryPressure,
        feldera_types::adapter_stats::ExternalGlobalControllerMetrics,
        feldera_types::adapter_stats::ConnectorError,
        feldera_types::adapter_stats::ConnectorHealth,
        feldera_types::adapter_stats::ConnectorHealthStatus,
        feldera_types::adapter_stats::ExternalInputEndpointStatus,
        feldera_types::adapter_stats::ExternalInputEndpointMetrics,
        feldera_types::adapter_stats::ExternalOutputEndpointStatus,
        feldera_types::adapter_stats::ExternalOutputEndpointMetrics,
        feldera_types::adapter_stats::CompletedWatermark,
        feldera_types::adapter_stats::ExternalTransactionInitiators,
        feldera_types::adapter_stats::ExternalTransactionPhase,
        feldera_types::adapter_stats::ExternalConnectorTransactionPhase,
        feldera_types::adapter_stats::PipelineState,
        feldera_types::adapter_stats::ShortEndpointConfig,
        feldera_types::adapter_stats::TransactionStatus,

        // Telemetry & License
        feldera_cloud1_client::license::DisplaySchedule,
        feldera_cloud1_client::license::LicenseInformation,
    ),),
    tags(
        (name = "Pipeline management", description = "Create, retrieve, update, delete and deploy pipelines."),
        (name = "Pipeline interaction", description = "Interact with deployed pipelines."),
        (name = "Configuration", description = "Retrieve configuration."),
        (name = "API keys", description = "Create, retrieve and delete API keys."),
        (name = "Metrics", description = "Retrieve metrics across pipelines."),
    ),
)]
pub struct ApiDoc;

// `static_files` magic.
include!(concat!(env!("OUT_DIR"), "/generated.rs"));

/// Middleware to add aggressive caching headers for immutable static files
/// (files with hashes in their names from web-console build)
async fn add_cache_headers(
    req: actix_web::dev::ServiceRequest,
    next: middleware::Next<impl MessageBody>,
) -> Result<actix_web::dev::ServiceResponse<impl MessageBody>, actix_web::Error> {
    let path = req.path().to_owned();
    let mut res = next.call(req).await?;

    // SvelteKit generates files like:
    // - _app/immutable/chunks/bundle.abc123.js
    // - _app/immutable/assets/style.def456.css
    // - _app/immutable/entry/app.789abc.js
    if path.contains("/_app/immutable/") || path.contains(".immutable.") {
        // These file names include content hashes and are truly immutable
        res.headers_mut().insert(
            header::CACHE_CONTROL,
            header::HeaderValue::from_static("public, max-age=31536000, immutable"),
        );
        res.headers_mut().insert(
            header::EXPIRES,
            // Far-future expiration date (>10 years) to ensure immutable files
            // are never revalidated. Thu, 31 Dec 2037 is commonly used as it's
            // well beyond any reasonable cache lifetime but before Y2038 issues.
            header::HeaderValue::from_static("Thu, 31 Dec 2037 23:55:55 GMT"),
        );
    } else if path.ends_with(".js")
        || path.ends_with(".css")
        || path.ends_with(".woff2")
        || path.ends_with(".svg")
        || path.ends_with(".png")
    {
        // Other static assets - cache for a shorter period but still aggressive
        res.headers_mut().insert(
            header::CACHE_CONTROL,
            header::HeaderValue::from_static("public, max-age=86400"), // 24 hours
        );
    }

    Ok(res)
}

// The scope for all unauthenticated API endpoints
fn public_scope() -> Scope {
    let openapi = ApiDoc::openapi();

    // Leave this as an empty prefix to load the UI by default. When constructing an
    // app, always attach other scopes without empty prefixes before this one,
    // or route resolution does not work correctly.
    web::scope("")
        .service(endpoints::config::get_config_authentication)
        .service(SwaggerUi::new("/swagger-ui/{_:.*}").url("/api-doc/openapi.json", openapi))
        .service(healthz)
        .service(
            web::scope("") // Nested scope: also matches "/"
                .wrap(middleware::from_fn(add_cache_headers))
                .service(ResourceFiles::new("/", generate()).resolve_not_found_to_root()),
        )
}

// The scope for all authenticated API endpoints
fn api_scope() -> Scope {
    // Make APIs available under the /v0/ prefix
    web::scope("/v0")
        // Pipeline management endpoints
        .service(endpoints::pipeline_management::list_pipelines)
        .service(endpoints::pipeline_management::get_pipeline)
        .service(endpoints::pipeline_management::post_pipeline)
        .service(endpoints::pipeline_management::put_pipeline)
        .service(endpoints::pipeline_management::patch_pipeline)
        .service(endpoints::pipeline_management::post_pipeline_testing)
        .service(endpoints::pipeline_management::post_update_runtime)
        .service(endpoints::pipeline_management::delete_pipeline)
        .service(endpoints::pipeline_management::post_pipeline_start)
        .service(endpoints::pipeline_management::post_pipeline_stop)
        .service(endpoints::pipeline_management::post_pipeline_dismiss_error)
        .service(endpoints::pipeline_management::post_pipeline_clear)
        .service(endpoints::pipeline_management::get_pipeline_logs)
        // Pipeline interaction endpoints
        .service(endpoints::pipeline_interaction::http_input)
        .service(endpoints::pipeline_interaction::http_output)
        .service(endpoints::pipeline_interaction::checkpoint_pipeline)
        .service(endpoints::pipeline_interaction::sync_checkpoint)
        .service(endpoints::pipeline_interaction::get_checkpoint_status)
        .service(endpoints::pipeline_interaction::get_checkpoint_sync_status)
        .service(endpoints::pipeline_interaction::get_checkpoints)
        .service(endpoints::pipeline_interaction::post_pipeline_pause)
        .service(endpoints::pipeline_interaction::post_pipeline_resume)
        .service(endpoints::pipeline_interaction::post_pipeline_activate)
        .service(endpoints::pipeline_interaction::post_pipeline_approve)
        .service(endpoints::pipeline_interaction::post_pipeline_input_connector_action)
        .service(endpoints::pipeline_interaction::get_pipeline_input_connector_status)
        .service(endpoints::pipeline_interaction::get_pipeline_output_connector_status)
        .service(endpoints::pipeline_interaction::get_pipeline_stats)
        .service(endpoints::pipeline_interaction::get_pipeline_metrics)
        .service(endpoints::pipeline_interaction::get_pipeline_time_series)
        .service(endpoints::pipeline_interaction::get_pipeline_time_series_stream)
        .service(endpoints::pipeline_interaction::get_pipeline_circuit_profile)
        .service(endpoints::pipeline_interaction::get_pipeline_circuit_json_profile)
        .service(endpoints::pipeline_interaction::get_pipeline_dataflow_graph)
        .service(endpoints::pipeline_interaction::get_pipeline_heap_profile)
        .service(endpoints::pipeline_interaction::get_pipeline_samply_profile)
        .service(endpoints::pipeline_interaction::start_samply_profile)
        .service(endpoints::pipeline_interaction::support_bundle::get_pipeline_support_bundle)
        .service(endpoints::pipeline_interaction::post_pipeline_rebalance)
        .service(endpoints::pipeline_interaction::pipeline_adhoc_sql)
        .service(endpoints::pipeline_interaction::completion_token)
        .service(endpoints::pipeline_interaction::completion_status)
        .service(endpoints::pipeline_interaction::start_transaction)
        .service(endpoints::pipeline_interaction::commit_transaction)
        // API keys endpoints
        .service(endpoints::api_key::list_api_keys)
        .service(endpoints::api_key::get_api_key)
        .service(endpoints::api_key::post_api_key)
        .service(endpoints::api_key::delete_api_key)
        // Configuration endpoints
        .service(endpoints::config::get_config)
        .service(endpoints::config::get_config_demos)
        .service(endpoints::config::get_config_session)
        // Metrics of all pipelines belonging to this tenant
        .service(endpoints::metrics::get_metrics)
        // Cluster health check
        .service(endpoints::cluster::list_cluster_events)
        .service(endpoints::cluster::get_cluster_event)
        .service(endpoints::cluster::get_cluster_health)
}

struct SecurityAddon;

impl Modify for SecurityAddon {
    fn modify(&self, openapi: &mut utoipa::openapi::OpenApi) {
        if let Some(components) = openapi.components.as_mut() {
            components.add_security_scheme(
                "JSON web token (JWT) or API key",
                SecurityScheme::Http(
                    HttpBuilder::new()
                        .scheme(HttpAuthScheme::Bearer)
                        .bearer_format("JWT")
                        .description(Some(
                            r#"Use a JWT token obtained via an OAuth2/OIDC
                               login workflow or an API key obtained via
                               the `/v0/api-keys` endpoint."#,
                        ))
                        .build(),
                ),
            )
        }
    }
}

// The below types and methods are used for running the api-server

pub(crate) struct ServerState {
    // The server must avoid holding this lock for a long time to avoid blocking concurrent
    // requests.
    pub db: Arc<Mutex<StoragePostgres>>,
    pub runner: RunnerInteraction,
    pub common_config: CommonConfig,
    pub config: ApiServerConfig,
    pub jwk_cache: Arc<Mutex<JwkCache>>,
    probe: Arc<Mutex<DbProbe>>,
    pub demos: Vec<Demo>,
    pub license_check: Arc<RwLock<Option<LicenseCheck>>>,
}

impl ServerState {
    pub async fn new(
        common_config: CommonConfig,
        config: ApiServerConfig,
        db: Arc<Mutex<StoragePostgres>>,
        license_check: Arc<RwLock<Option<LicenseCheck>>>,
    ) -> AnyResult<Self> {
        let runner = RunnerInteraction::new(common_config.clone(), db.clone());
        let db_copy = db.clone();
        let demos = read_demos_from_directories(&config.demos_dir);
        Ok(Self {
            db,
            runner,
            common_config,
            config,
            jwk_cache: Arc::new(Mutex::new(JwkCache::new())),
            probe: DbProbe::new(db_copy).await,
            demos,
            license_check,
        })
    }

    #[cfg(test)]
    pub(crate) async fn test_state(db: Arc<Mutex<StoragePostgres>>) -> Self {
        let common_config = CommonConfig::test_config();
        let api_config = ApiServerConfig::test_config();
        let license_check = Arc::new(RwLock::new(None::<LicenseCheck>));
        Self::new(common_config, api_config, db.clone(), license_check)
            .await
            .unwrap()
    }
}

/// Logs the responses of the HTTP(S) server.
pub fn log_response(
    res: Result<ServiceResponse<BoxBody>, actix_web::Error>,
) -> Result<ServiceResponse<BoxBody>, actix_web::Error> {
    match &res {
        Ok(response) => {
            let req = response.request();
            let level = if response.status().is_success()
                || response.status().is_informational()
                || response.status().is_redirection()
            {
                if req.method() == Method::GET && req.path() == "/healthz" {
                    Level::TRACE
                } else {
                    Level::DEBUG
                }
            } else if response.status().is_client_error()
                || response.status() == StatusCode::SERVICE_UNAVAILABLE
            {
                Level::INFO
            } else {
                Level::ERROR
            };
            log_with_level!(
                level,
                "Response: {} (size: {:?}) to request {} {}",
                response.status(),
                response.response().body().size(),
                req.method(),
                req.path()
            );
        }
        Err(e) => {
            error!("Service response error: {e}");
        }
    }
    res
}

pub async fn run(
    db: Arc<Mutex<StoragePostgres>>,
    common_config: CommonConfig,
    api_config: ApiServerConfig,
    license_check: Arc<RwLock<Option<LicenseCheck>>>,
) -> AnyResult<()> {
    let listener = TcpListener::bind((common_config.bind_address.clone(), common_config.api_port))
        .unwrap_or_else(|_| {
            panic!(
                "API server unable to bind listener to {}:{} -- is the port occupied?",
                common_config.bind_address, common_config.api_port
            )
        });
    let state = WebData::new(
        ServerState::new(common_config.clone(), api_config.clone(), db, license_check).await?,
    );
    let auth_configuration = match api_config.auth_provider {
        crate::config::AuthProviderType::None => None,
        crate::config::AuthProviderType::AwsCognito => Some(crate::auth::aws_auth_config()),
        crate::config::AuthProviderType::GenericOidc => {
            match crate::auth::generic_oidc_auth_config(&api_config).await {
                Ok(config) => Some(config),
                Err(e) => {
                    error!("Failed to configure generic OIDC authentication: {}", e);
                    return Err(anyhow::anyhow!(
                        "Authentication configuration failed: {}",
                        e
                    ));
                }
            }
        }
    };
    let server = match auth_configuration {
        // We instantiate an awc::Client that can be used if the api-server needs to
        // make outgoing calls. This object is not meant to have more than one instance
        // per thread (otherwise, it causes high resource pressure on both CPU and fds).
        Some(auth_configuration) => {
            let common_config_cloned = common_config.clone();
            let api_config = api_config.clone();
            let state = state.clone();
            let server = HttpServer::new(move || {
                let auth_middleware = HttpAuthentication::with_fn(crate::auth::auth_validator);
                let client = WebData::new(common_config_cloned.awc_client());
                App::new()
                    .app_data(state.clone())
                    .app_data(auth_configuration.clone())
                    .app_data(client)
                    .wrap_fn(|req, srv| {
                        let log_level = if req.method() == Method::GET && req.path() == "/healthz" {
                            Level::TRACE
                        } else {
                            Level::DEBUG
                        };

                        log_with_level!(log_level, "Request: {} {}", req.method(), req.path());
                        srv.call(req).map(log_response)
                    })
                    .wrap(middleware::Compress::default())
                    .wrap(api_config.cors())
                    .wrap(observability::actix_middleware())
                    .service(api_scope().wrap(auth_middleware))
                    .service(public_scope())
            })
            .workers(common_config.http_workers)
            .worker_max_blocking_threads(std::cmp::max(512 / common_config.http_workers, 1));
            if let Some(server_config) = common_config.https_server_config() {
                server
                    .listen_rustls_0_23(listener, server_config)
                    .expect("API HTTPS server unable to listen")
                    .run()
            } else {
                server
                    .listen(listener)
                    .expect("API HTTP server unable to listen")
                    .run()
            }
        }
        None => {
            let common_config_cloned = common_config.clone();
            let api_config = api_config.clone();
            let state = state.clone();
            let server = HttpServer::new(move || {
                let client = WebData::new(common_config_cloned.awc_client());
                App::new()
                    .app_data(state.clone())
                    .app_data(client)
                    .wrap_fn(|req, srv| {
                        trace!("Request: {} {}", req.method(), req.path());
                        srv.call(req).map(log_response)
                    })
                    .wrap(middleware::Compress::default())
                    .wrap(api_config.cors())
                    .wrap(observability::actix_middleware())
                    .service(api_scope().wrap_fn(|req, srv| {
                        let req = crate::auth::tag_with_default_tenant_id(req);
                        srv.call(req)
                    }))
                    .service(public_scope())
            })
            .workers(common_config.http_workers)
            .worker_max_blocking_threads(std::cmp::max(512 / common_config.http_workers, 1));
            if let Some(server_config) = common_config.https_server_config() {
                server
                    .listen_rustls_0_23(listener, server_config)
                    .expect("API HTTPS server unable to listen")
                    .run()
            } else {
                server
                    .listen(listener)
                    .expect("API HTTP server unable to listen")
                    .run()
            }
        }
    };
    info!(
        "API {} server: ready on port {} ({} workers)",
        if common_config.enable_https {
            "HTTPS"
        } else {
            "HTTP"
        },
        common_config.api_port,
        common_config.http_workers,
    );

    let banner = if theme(Duration::from_millis(500)).unwrap_or(Theme::Light) == Theme::Dark {
        include_str!("../../light-banner.ascii")
    } else {
        include_str!("../../dark-banner.ascii")
    };
    let addr = env::var("BANNER_ADDR").unwrap_or("127.0.0.1".to_string());
    let url = format!(
        "{}://{}:{}",
        if common_config.enable_https {
            "https"
        } else {
            "http"
        },
        addr,
        common_config.api_port
    );

    // Lock both out streams so that the banner is printed in one go
    // and not interrupted by log messages from other threads.
    let err_lock = io::stderr().lock();
    let mut out_lock = io::stdout().lock();
    let _ = out_lock.write_all(
        format!(
            r"

{banner}

Web console URL: {}
API server URL: {}
Documentation: https://docs.feldera.com/
Version: {} v{}{}
",
            url,
            url,
            if cfg!(feature = "feldera-enterprise") {
                "Enterprise"
            } else {
                "Open source"
            },
            env!("CARGO_PKG_VERSION"),
            if env!("FELDERA_PLATFORM_VERSION_SUFFIX").is_empty() {
                "".to_string()
            } else {
                format!(" ({})", env!("FELDERA_PLATFORM_VERSION_SUFFIX"))
            },
        )
        .as_bytes(),
    );
    if let Some(features) = unstable_features() {
        println!(
            "Unstable Features: {}",
            features.iter().cloned().collect::<Vec<&str>>().join(", ")
        );
    }
    drop(out_lock);
    drop(err_lock);

    // Create a shutdown channel for the support collector
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // Spawn the support collector in a dedicated thread with its own runtime and HTTP client
    let collector_handle = if api_config.support_data_collection_frequency > 0 {
        Some(
            std::thread::Builder::new()
                .name("support-data-collector".to_string())
                .spawn(move || {
                    let rt = tokio::runtime::Builder::new_current_thread()
                        .enable_all()
                        .build()
                        .unwrap();

                    rt.block_on(async {
                        let local = tokio::task::LocalSet::new();
                        local
                            .run_until(async move {
                                let client = common_config.awc_client();
                                let support_collector = SupportDataCollector::new(
                                    state.clone().into_inner(),
                                    client,
                                    api_config.support_data_collection_frequency,
                                    api_config.support_data_retention,
                                    shutdown_rx,
                                );
                                support_collector.run().await;
                            })
                            .await;
                    });
                })
                .unwrap(),
        )
    } else {
        info!("Support data collection is disabled (collection frequency is 0)");
        None
    };

    // Arrange for graceful shutdown on Ctrl+C and SIGTERM
    let server_handle_ctrlc = server.handle();
    tokio::spawn(async move {
        if signal::ctrl_c().await.is_ok() {
            info!("Ctrl+C received, stopping HTTP server...");
            server_handle_ctrlc.stop(true).await;
        }
    });
    #[cfg(unix)]
    {
        use tokio::signal::unix::{signal as unix_signal, SignalKind};
        let mut term_stream = unix_signal(SignalKind::terminate()).expect("SIGTERM handler");
        let server_handle_term = server.handle();
        tokio::spawn(async move {
            term_stream.recv().await;
            info!("SIGTERM received, stopping HTTP server...");
            server_handle_term.stop(true).await;
        });
    }

    // Await HTTP server; when it returns, signal the collector to stop and join
    let server_result = server.await;
    if let Some(collector_handle) = collector_handle {
        let _ = shutdown_tx.send(true);
        let _ = collector_handle.join();
    }

    if let Some(client) = sentry::Hub::current().client() {
        info!("Shutting down sentry");
        client.close(Some(Duration::from_secs(3)));
    }

    server_result?;
    Ok(())
}

/// This is an internal endpoint and as such is not exposed via OpenAPI
#[get("/healthz")]
async fn healthz(state: WebData<ServerState>) -> Result<HttpResponse, ManagerError> {
    let probe = state.probe.lock().await;
    Ok(probe.as_http_response())
}
