servers/http/

event.rs

// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::borrow::Cow;
use std::collections::BTreeMap;
use std::fmt::Display;
use std::io::BufRead;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Instant;

use api::helper::pb_value_to_value_ref;
use async_trait::async_trait;
use axum::body::Bytes;
use axum::extract::{FromRequest, Multipart, Path, Query, Request, State};
use axum::http::header::CONTENT_TYPE;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::{Extension, Json};
use axum_extra::TypedHeader;
use common_catalog::consts::default_engine;
use common_error::ext::{BoxedError, ErrorExt};
use common_query::{Output, OutputData};
use common_telemetry::{error, warn};
use headers::ContentType;
use lazy_static::lazy_static;
use mime_guess::mime;
use operator::expr_helper::{create_table_expr_by_column_schemas, expr_to_create};
use pipeline::util::to_pipeline_version;
use pipeline::{ContextReq, GreptimePipelineParams, PipelineContext, PipelineDefinition};
use prometheus::{HistogramVec, IntCounterVec};
use serde::{Deserialize, Serialize};
use serde_json::{Deserializer, Map, Value as JsonValue, json};
use session::context::{Channel, QueryContext, QueryContextRef};
use simd_json::Buffers;
use snafu::{OptionExt, ResultExt, ensure};
use store_api::mito_engine_options::APPEND_MODE_KEY;
use strum::{EnumIter, IntoEnumIterator};
use table::table_reference::TableReference;
use vrl::value::{KeyString, Value as VrlValue};

use crate::error::{
    Error, InvalidParameterSnafu, OtherSnafu, ParseJsonSnafu, PipelineSnafu, Result,
    status_code_to_http_status,
};
use crate::http::HttpResponse;
use crate::http::header::constants::{
    GREPTIME_LOG_PIPELINE_NAME_HEADER_NAME, GREPTIME_PIPELINE_NAME_HEADER_NAME,
    GREPTIME_PIPELINE_PARAMS_HEADER,
};
use crate::http::header::{
    CONTENT_TYPE_JSON_STREAM_STR, CONTENT_TYPE_JSON_STREAM_SUBTYPE_STR, CONTENT_TYPE_NDJSON_STR,
    CONTENT_TYPE_NDJSON_SUBTYPE_STR, CONTENT_TYPE_PROTOBUF_STR,
};
use crate::http::result::greptime_manage_resp::{GreptimedbManageResponse, SqlOutput};
use crate::http::result::greptime_result_v1::GreptimedbV1Response;
use crate::interceptor::{LogIngestInterceptor, LogIngestInterceptorRef};
use crate::metrics::{
    METRIC_FAILURE_VALUE, METRIC_HTTP_LOGS_INGESTION_COUNTER, METRIC_HTTP_LOGS_INGESTION_ELAPSED,
    METRIC_HTTP_LOGS_UTF8_RETRIES_TOTAL, METRIC_SUCCESS_VALUE,
};
use crate::pipeline::run_pipeline;
use crate::query_handler::PipelineHandlerRef;

const GREPTIME_INTERNAL_PIPELINE_NAME_PREFIX: &str = "greptime_";
const GREPTIME_PIPELINE_SKIP_ERROR_KEY: &str = "skip_error";

const CREATE_TABLE_SQL_SUFFIX_EXISTS: &str =
    "the pipeline has dispatcher or table_suffix, the table name may not be fixed";
const CREATE_TABLE_SQL_TABLE_EXISTS: &str =
    "table already exists, the CREATE TABLE SQL may be different";

lazy_static! {
    pub static ref JSON_CONTENT_TYPE: ContentType = ContentType::json();
    pub static ref TEXT_CONTENT_TYPE: ContentType = ContentType::text();
    pub static ref TEXT_UTF8_CONTENT_TYPE: ContentType = ContentType::text_utf8();
    pub static ref PB_CONTENT_TYPE: ContentType =
        ContentType::from_str(CONTENT_TYPE_PROTOBUF_STR).unwrap();
    pub static ref NDJSON_CONTENT_TYPE: ContentType =
        ContentType::from_str(CONTENT_TYPE_NDJSON_STR).unwrap();
    pub static ref JSON_STREAM_CONTENT_TYPE: ContentType =
        ContentType::from_str(CONTENT_TYPE_JSON_STREAM_STR).unwrap();
}

/// LogIngesterQueryParams is used for query params of log ingester API.
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct LogIngesterQueryParams {
    /// The database where log data will be written to.
    pub db: Option<String>,
    /// The table where log data will be written to.
    pub table: Option<String>,
    /// The pipeline that will be used for log ingestion.
    pub pipeline_name: Option<String>,
    /// The version of the pipeline to be used for log ingestion.
    pub version: Option<String>,
    /// Whether to ignore payload parsing and custom timestamp errors during log ingestion.
    /// Defaults to the server's `pipeline.ignore_errors` setting when omitted.
    pub ignore_errors: Option<bool>,
    /// The source of the log data.
    pub source: Option<String>,
    /// The JSON field name of the log message. If not provided, it will take the whole log as the message.
    /// The field must be at the top level of the JSON structure.
    pub msg_field: Option<String>,
    /// Specify a custom time index from the input data rather than server's arrival time.
    /// Valid formats:
    /// - <field_name>;epoch;<resolution>
    /// - <field_name>;datestr;<format>
    ///
    /// If an error occurs while parsing the config, the error will be returned in the response.
    /// If an error occurs while ingesting the data, the `ignore_errors` will be used to determine if the error should be ignored.
    /// If so, use the current server's timestamp as the event time.
    pub custom_time_index: Option<String>,
    /// Whether to skip errors during log ingestion.
    /// If set to true, the ingestion will continue even if there are errors in the data.
    /// If set to false, the ingestion will stop at the first error.
    /// This controls pipeline execution errors, separately from `ignore_errors`.
    /// The priority of query params is lower than that headers of x-greptime-pipeline-params.
    pub skip_error: Option<bool>,
}

/// LogIngestRequest is the internal request for log ingestion. The raw log input can be transformed into multiple LogIngestRequests.
/// Multiple LogIngestRequests will be ingested into the same database with the same pipeline.
#[derive(Debug, PartialEq)]
pub(crate) struct PipelineIngestRequest {
    /// The table where the log data will be written to.
    pub table: String,
    /// The log data to be ingested.
    pub values: Vec<VrlValue>,
}

pub struct PipelineContent(String);

impl<S> FromRequest<S> for PipelineContent
where
    S: Send + Sync,
{
    type Rejection = Response;

    async fn from_request(req: Request, state: &S) -> Result<Self, Self::Rejection> {
        let content_type_header = req.headers().get(CONTENT_TYPE);
        let content_type = content_type_header.and_then(|value| value.to_str().ok());
        if let Some(content_type) = content_type {
            if content_type.ends_with("yaml") {
                let payload = String::from_request(req, state)
                    .await
                    .map_err(IntoResponse::into_response)?;
                return Ok(Self(payload));
            }

            if content_type.starts_with("multipart/form-data") {
                let mut payload: Multipart = Multipart::from_request(req, state)
                    .await
                    .map_err(IntoResponse::into_response)?;
                let file = payload
                    .next_field()
                    .await
                    .map_err(IntoResponse::into_response)?;
                let payload = file
                    .ok_or(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())?
                    .text()
                    .await
                    .map_err(IntoResponse::into_response)?;
                return Ok(Self(payload));
            }
        }

        Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response())
    }
}

#[axum_macros::debug_handler]
pub async fn query_pipeline(
    State(state): State<LogState>,
    Extension(mut query_ctx): Extension<QueryContext>,
    Query(query_params): Query<LogIngesterQueryParams>,
    Path(pipeline_name): Path<String>,
) -> Result<GreptimedbManageResponse> {
    let start = Instant::now();
    let handler = state.log_handler;
    ensure!(
        !pipeline_name.is_empty(),
        InvalidParameterSnafu {
            reason: "pipeline_name is required in path",
        }
    );

    let version = to_pipeline_version(query_params.version.as_deref()).context(PipelineSnafu)?;

    query_ctx.set_channel(Channel::Log);
    let query_ctx = Arc::new(query_ctx);

    let (pipeline, pipeline_version) = handler
        .get_pipeline_str(&pipeline_name, version, query_ctx)
        .await?;

    Ok(GreptimedbManageResponse::from_pipeline(
        pipeline_name,
        query_params
            .version
            .unwrap_or(pipeline_version.0.to_timezone_aware_string(None)),
        start.elapsed().as_millis() as u64,
        Some(pipeline),
    ))
}

/// Generate DDL from pipeline definition.
#[axum_macros::debug_handler]
pub async fn query_pipeline_ddl(
    State(state): State<LogState>,
    Extension(mut query_ctx): Extension<QueryContext>,
    Query(query_params): Query<LogIngesterQueryParams>,
    Path(pipeline_name): Path<String>,
) -> Result<GreptimedbManageResponse> {
    let start = Instant::now();
    let handler = state.log_handler;
    ensure!(
        !pipeline_name.is_empty(),
        InvalidParameterSnafu {
            reason: "pipeline_name is required in path",
        }
    );
    ensure!(
        !pipeline_name.starts_with(GREPTIME_INTERNAL_PIPELINE_NAME_PREFIX),
        InvalidParameterSnafu {
            reason: "built-in pipelines don't have fixed table schema",
        }
    );
    let table_name = query_params.table.context(InvalidParameterSnafu {
        reason: "table name is required",
    })?;

    let version = to_pipeline_version(query_params.version.as_deref()).context(PipelineSnafu)?;

    query_ctx.set_channel(Channel::Log);
    let query_ctx = Arc::new(query_ctx);

    handler.check_pipeline_query_permission(&query_ctx)?;
    let pipeline = handler
        .get_pipeline(&pipeline_name, version, query_ctx.clone())
        .await?;

    let schemas_def = pipeline.schemas().context(InvalidParameterSnafu {
        reason: "auto transform doesn't have fixed table schema",
    })?;

    let schema = query_ctx.current_schema();
    let table_name_ref = TableReference {
        catalog: query_ctx.current_catalog(),
        schema: &schema,
        table: &table_name,
    };

    let mut create_table_expr =
        create_table_expr_by_column_schemas(&table_name_ref, schemas_def, default_engine(), None)
            .map_err(BoxedError::new)
            .context(OtherSnafu)?;

    // manually set the append_mode to true
    create_table_expr
        .table_options
        .insert(APPEND_MODE_KEY.to_string(), "true".to_string());

    let expr = expr_to_create(&create_table_expr, None)
        .map_err(BoxedError::new)
        .context(OtherSnafu)?;

    let message = if handler.get_table(&table_name, &query_ctx).await?.is_some() {
        Some(CREATE_TABLE_SQL_TABLE_EXISTS.to_string())
    } else if pipeline.is_variant_table_name() {
        Some(CREATE_TABLE_SQL_SUFFIX_EXISTS.to_string())
    } else {
        None
    };

    let sql = SqlOutput {
        sql: format!("{:#}", expr),
        message,
    };

    Ok(GreptimedbManageResponse::from_sql(
        sql,
        start.elapsed().as_millis() as u64,
    ))
}

#[axum_macros::debug_handler]
pub async fn add_pipeline(
    State(state): State<LogState>,
    Path(pipeline_name): Path<String>,
    Extension(mut query_ctx): Extension<QueryContext>,
    PipelineContent(payload): PipelineContent,
) -> Result<GreptimedbManageResponse> {
    let start = Instant::now();
    let handler = state.log_handler;
    ensure!(
        !pipeline_name.is_empty(),
        InvalidParameterSnafu {
            reason: "pipeline_name is required in path",
        }
    );
    ensure!(
        !pipeline_name.starts_with(GREPTIME_INTERNAL_PIPELINE_NAME_PREFIX),
        InvalidParameterSnafu {
            reason: "pipeline_name cannot start with greptime_",
        }
    );
    ensure!(
        !payload.is_empty(),
        InvalidParameterSnafu {
            reason: "pipeline is required in body",
        }
    );

    query_ctx.set_channel(Channel::Log);
    let query_ctx = Arc::new(query_ctx);

    let content_type = "yaml";
    let result = handler
        .insert_pipeline(&pipeline_name, content_type, &payload, query_ctx)
        .await;

    result
        .map(|pipeline| {
            GreptimedbManageResponse::from_pipeline(
                pipeline_name,
                pipeline.0.to_timezone_aware_string(None),
                start.elapsed().as_millis() as u64,
                None,
            )
        })
        .map_err(|e| {
            error!(e; "failed to insert pipeline");
            e
        })
}

#[axum_macros::debug_handler]
pub async fn delete_pipeline(
    State(state): State<LogState>,
    Extension(mut query_ctx): Extension<QueryContext>,
    Query(query_params): Query<LogIngesterQueryParams>,
    Path(pipeline_name): Path<String>,
) -> Result<GreptimedbManageResponse> {
    let start = Instant::now();
    let handler = state.log_handler;
    ensure!(
        !pipeline_name.is_empty(),
        InvalidParameterSnafu {
            reason: "pipeline_name is required",
        }
    );

    let version_str = query_params.version.context(InvalidParameterSnafu {
        reason: "version is required",
    })?;

    let version = to_pipeline_version(Some(&version_str)).context(PipelineSnafu)?;

    query_ctx.set_channel(Channel::Log);
    let query_ctx = Arc::new(query_ctx);

    handler
        .delete_pipeline(&pipeline_name, version, query_ctx)
        .await
        .map(|v| {
            if v.is_some() {
                GreptimedbManageResponse::from_pipeline(
                    pipeline_name,
                    version_str,
                    start.elapsed().as_millis() as u64,
                    None,
                )
            } else {
                GreptimedbManageResponse::from_pipelines(vec![], start.elapsed().as_millis() as u64)
            }
        })
        .map_err(|e| {
            error!(e; "failed to delete pipeline");
            e
        })
}

/// Transform NDJSON array into a single array
/// always return an array
pub(crate) fn transform_ndjson_array_factory(
    values: impl IntoIterator<Item = Result<VrlValue, serde_json::Error>>,
    ignore_error: bool,
) -> Result<Vec<VrlValue>> {
    values
        .into_iter()
        .try_fold(Vec::with_capacity(100), |mut acc_array, item| match item {
            Ok(item_value) => {
                match item_value {
                    VrlValue::Array(item_array) => {
                        acc_array.extend(item_array);
                    }
                    VrlValue::Object(_) => {
                        acc_array.push(item_value);
                    }
                    _ => {
                        if !ignore_error {
                            warn!("invalid item in array: {:?}", item_value);
                            return InvalidParameterSnafu {
                                reason: format!("invalid item: {} in array", item_value),
                            }
                            .fail();
                        }
                    }
                }
                Ok(acc_array)
            }
            Err(_) if !ignore_error => item.map(|x| vec![x]).context(ParseJsonSnafu),
            Err(_) => {
                warn!("invalid item in array: {:?}", item);
                Ok(acc_array)
            }
        })
}

/// Dryrun pipeline with given data
async fn dryrun_pipeline_inner(
    value: Vec<VrlValue>,
    pipeline: Arc<pipeline::Pipeline>,
    pipeline_handler: PipelineHandlerRef,
    query_ctx: &QueryContextRef,
) -> Result<Response> {
    let params = GreptimePipelineParams::default();

    let pipeline_def = PipelineDefinition::Resolved(pipeline);
    let pipeline_ctx = PipelineContext::new(&pipeline_def, &params, query_ctx.channel());
    let results = run_pipeline(
        &pipeline_handler,
        &pipeline_ctx,
        PipelineIngestRequest {
            table: "dry_run".to_owned(),
            values: value,
        },
        query_ctx,
        true,
    )
    .await?;

    let column_type_key = "column_type";
    let data_type_key = "data_type";
    let name_key = "name";

    let results = results
        .all_req()
        .filter_map(|row| {
            if let Some(rows) = row.rows {
                let table_name = row.table_name;
                let result_schema = rows.schema;

                let schema = result_schema
                    .iter()
                    .map(|cs| {
                        let mut map = Map::new();
                        map.insert(
                            name_key.to_string(),
                            JsonValue::String(cs.column_name.clone()),
                        );
                        map.insert(
                            data_type_key.to_string(),
                            JsonValue::String(cs.datatype().as_str_name().to_string()),
                        );
                        map.insert(
                            column_type_key.to_string(),
                            JsonValue::String(cs.semantic_type().as_str_name().to_string()),
                        );
                        map.insert(
                            "fulltext".to_string(),
                            JsonValue::Bool(
                                cs.options
                                    .clone()
                                    .is_some_and(|x| x.options.contains_key("fulltext")),
                            ),
                        );
                        JsonValue::Object(map)
                    })
                    .collect::<Vec<_>>();

                let rows = rows
                    .rows
                    .into_iter()
                    .map(|row| {
                        row.values
                            .into_iter()
                            .enumerate()
                            .map(|(idx, v)| {
                                let mut map = Map::new();
                                let value_ref = pb_value_to_value_ref(
                                    &v,
                                    result_schema[idx].datatype_extension.as_ref(),
                                );
                                let greptime_value: datatypes::value::Value = value_ref.into();
                                let serde_json_value =
                                    serde_json::Value::try_from(greptime_value).unwrap();
                                map.insert("value".to_string(), serde_json_value);
                                map.insert("key".to_string(), schema[idx][name_key].clone());
                                map.insert(
                                    "semantic_type".to_string(),
                                    schema[idx][column_type_key].clone(),
                                );
                                map.insert(
                                    "data_type".to_string(),
                                    schema[idx][data_type_key].clone(),
                                );
                                JsonValue::Object(map)
                            })
                            .collect()
                    })
                    .collect();

                let mut result = Map::new();
                result.insert("schema".to_string(), JsonValue::Array(schema));
                result.insert("rows".to_string(), JsonValue::Array(rows));
                result.insert("table_name".to_string(), JsonValue::String(table_name));
                let result = JsonValue::Object(result);
                Some(result)
            } else {
                None
            }
        })
        .collect();
    Ok(Json(JsonValue::Array(results)).into_response())
}

/// Dryrun pipeline with given data
/// pipeline_name and pipeline_version to specify pipeline stored in db
/// pipeline to specify pipeline raw content
/// data to specify data
/// data maght be list of string or list of object
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct PipelineDryrunParams {
    pub pipeline_name: Option<String>,
    pub pipeline_version: Option<String>,
    pub pipeline: Option<String>,
    pub data_type: Option<String>,
    pub data: String,
}

/// Check if the payload is valid json
/// Check if the payload contains pipeline or pipeline_name and data
/// Return Some if valid, None if invalid
fn check_pipeline_dryrun_params_valid(payload: &Bytes) -> Option<PipelineDryrunParams> {
    match serde_json::from_slice::<PipelineDryrunParams>(payload) {
        // payload with pipeline or pipeline_name and data is array
        Ok(params) if params.pipeline.is_some() || params.pipeline_name.is_some() => Some(params),
        // because of the pipeline_name or pipeline is required
        Ok(_) => None,
        // invalid json
        Err(_) => None,
    }
}

/// Check if the pipeline_name exists
fn check_pipeline_name_exists(pipeline_name: Option<String>) -> Result<String> {
    pipeline_name.context(InvalidParameterSnafu {
        reason: "pipeline_name is required",
    })
}

/// Check if the data length less than 10
fn check_data_valid(data_len: usize) -> Result<()> {
    ensure!(
        data_len <= 10,
        InvalidParameterSnafu {
            reason: "data is required",
        }
    );
    Ok(())
}

fn add_step_info_for_pipeline_dryrun_error(step_msg: &str, e: Error) -> Response {
    let body = Json(json!({
        "error": format!("{}: {}", step_msg,e.output_msg()),
    }));

    (status_code_to_http_status(&e.status_code()), body).into_response()
}

/// Parse the data with given content type
/// If the content type is invalid, return error
/// content type is one of application/json, text/plain, application/x-ndjson
fn parse_dryrun_data(
    data_type: String,
    data: String,
    ignore_errors: bool,
) -> Result<Vec<VrlValue>> {
    if let Ok(content_type) = ContentType::from_str(&data_type) {
        extract_pipeline_value_by_content_type(content_type, Bytes::from(data), ignore_errors)
    } else {
        InvalidParameterSnafu {
            reason: format!(
                "invalid content type: {}, expected: one of {}",
                data_type,
                EventPayloadResolver::support_content_type_list().join(", ")
            ),
        }
        .fail()
    }
}

#[axum_macros::debug_handler]
pub async fn pipeline_dryrun(
    State(log_state): State<LogState>,
    Query(query_params): Query<LogIngesterQueryParams>,
    Extension(mut query_ctx): Extension<QueryContext>,
    TypedHeader(content_type): TypedHeader<ContentType>,
    payload: Bytes,
) -> Result<Response> {
    let handler = log_state.log_handler;

    query_ctx.set_channel(Channel::Log);
    let query_ctx = Arc::new(query_ctx);
    handler.check_pipeline_query_permission(&query_ctx)?;

    let ignore_errors = query_params
        .ignore_errors
        .unwrap_or(handler.pipeline_options().ignore_errors);

    match check_pipeline_dryrun_params_valid(&payload) {
        Some(params) => {
            let data = parse_dryrun_data(
                params.data_type.unwrap_or("application/json".to_string()),
                params.data,
                ignore_errors,
            )?;

            check_data_valid(data.len())?;

            match params.pipeline {
                None => {
                    let version = to_pipeline_version(params.pipeline_version.as_deref())
                        .context(PipelineSnafu)?;
                    let pipeline_name = check_pipeline_name_exists(params.pipeline_name)?;
                    let pipeline = handler
                        .get_pipeline(&pipeline_name, version, query_ctx.clone())
                        .await?;
                    dryrun_pipeline_inner(data, pipeline, handler, &query_ctx).await
                }
                Some(pipeline) => {
                    let pipeline = handler.build_pipeline(&pipeline);
                    match pipeline {
                        Ok(pipeline) => {
                            match dryrun_pipeline_inner(
                                data,
                                Arc::new(pipeline),
                                handler,
                                &query_ctx,
                            )
                            .await
                            {
                                Ok(response) => Ok(response),
                                Err(e) => Ok(add_step_info_for_pipeline_dryrun_error(
                                    "Failed to exec pipeline",
                                    e,
                                )),
                            }
                        }
                        Err(e) => Ok(add_step_info_for_pipeline_dryrun_error(
                            "Failed to build pipeline",
                            e,
                        )),
                    }
                }
            }
        }
        None => {
            // This path is for back compatibility with the previous dry run code
            // where the payload is just data (JSON or plain text) and the pipeline name
            // is specified using query param.
            let pipeline_name = check_pipeline_name_exists(query_params.pipeline_name)?;

            let version =
                to_pipeline_version(query_params.version.as_deref()).context(PipelineSnafu)?;

            let value =
                extract_pipeline_value_by_content_type(content_type, payload, ignore_errors)?;

            check_data_valid(value.len())?;

            let pipeline = handler
                .get_pipeline(&pipeline_name, version, query_ctx.clone())
                .await?;

            dryrun_pipeline_inner(value, pipeline, handler, &query_ctx).await
        }
    }
}

pub(crate) fn extract_pipeline_params_map_from_headers(
    headers: &HeaderMap,
) -> ahash::HashMap<String, String> {
    GreptimePipelineParams::parse_header_str_to_map(
        headers
            .get(GREPTIME_PIPELINE_PARAMS_HEADER)
            .and_then(|v| v.to_str().ok()),
    )
}

/// Extracts the pipeline name from the request headers.
///
/// Both `x-greptime-pipeline-name` and the deprecated `x-greptime-log-pipeline-name`
/// are accepted, matching the headers already honored by the OTLP/Elasticsearch/Splunk
/// log ingestion endpoints. If both are present, the non-deprecated
/// `x-greptime-pipeline-name` takes precedence. Empty header values are ignored so that
/// they fall back to the `pipeline_name` query parameter.
fn pipeline_name_from_headers(headers: &HeaderMap) -> Option<String> {
    [
        GREPTIME_PIPELINE_NAME_HEADER_NAME,
        GREPTIME_LOG_PIPELINE_NAME_HEADER_NAME,
    ]
    .iter()
    .find_map(|name| {
        headers
            .get(*name)
            .and_then(|value| value.to_str().ok())
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToString::to_string)
    })
}

#[axum_macros::debug_handler]
pub async fn log_ingester(
    State(log_state): State<LogState>,
    Query(query_params): Query<LogIngesterQueryParams>,
    Extension(mut query_ctx): Extension<QueryContext>,
    TypedHeader(content_type): TypedHeader<ContentType>,
    headers: HeaderMap,
    payload: Bytes,
) -> Result<HttpResponse> {
    // validate source and payload
    let source = query_params.source.as_deref();
    let response = match &log_state.log_validator {
        Some(validator) => validator.validate(source, &payload).await,
        None => None,
    };
    if let Some(response) = response {
        return response;
    }

    let handler = log_state.log_handler;

    let table_name = query_params.table.context(InvalidParameterSnafu {
        reason: "table is required",
    })?;

    let ignore_errors = query_params
        .ignore_errors
        .unwrap_or(handler.pipeline_options().ignore_errors);

    // A pipeline name supplied via header takes precedence over the query parameter,
    // consistent with how other pipeline options (e.g. `x-greptime-pipeline-params`)
    // outrank their query-parameter counterparts.
    let pipeline_name = pipeline_name_from_headers(&headers)
        .or(query_params.pipeline_name)
        .context(InvalidParameterSnafu {
            reason: "pipeline_name is required",
        })?;
    let skip_error = query_params.skip_error.unwrap_or(false);
    let version = to_pipeline_version(query_params.version.as_deref()).context(PipelineSnafu)?;
    let pipeline = PipelineDefinition::from_name(
        &pipeline_name,
        version,
        query_params.custom_time_index.map(|s| (s, ignore_errors)),
    )
    .context(PipelineSnafu)?;

    let value = extract_pipeline_value_by_content_type(content_type, payload, ignore_errors)?;

    query_ctx.set_channel(Channel::Log);
    let query_ctx = Arc::new(query_ctx);

    let value = log_state
        .ingest_interceptor
        .as_ref()
        .pre_pipeline(value, query_ctx.clone())?;

    let mut pipeline_params_map = extract_pipeline_params_map_from_headers(&headers);
    if !pipeline_params_map.contains_key(GREPTIME_PIPELINE_SKIP_ERROR_KEY) && skip_error {
        pipeline_params_map.insert(GREPTIME_PIPELINE_SKIP_ERROR_KEY.to_string(), "true".into());
    }
    let pipeline_params = GreptimePipelineParams::from_map(pipeline_params_map);

    ingest_logs_inner(
        handler,
        pipeline,
        vec![PipelineIngestRequest {
            table: table_name,
            values: value,
        }],
        query_ctx,
        pipeline_params,
    )
    .await
}

#[derive(Debug, EnumIter)]
enum EventPayloadResolverInner {
    Json,
    /// `application/x-json-stream`: concatenated JSON values without any
    /// separator between them, e.g. `{"a":1}{"b":2}` (the `json_stream`
    /// format of fluentbit).
    JsonStream,
    Ndjson,
    Text,
}

impl Display for EventPayloadResolverInner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EventPayloadResolverInner::Json => write!(f, "{}", *JSON_CONTENT_TYPE),
            EventPayloadResolverInner::JsonStream => write!(f, "{}", *JSON_STREAM_CONTENT_TYPE),
            EventPayloadResolverInner::Ndjson => write!(f, "{}", *NDJSON_CONTENT_TYPE),
            EventPayloadResolverInner::Text => write!(f, "{}", *TEXT_CONTENT_TYPE),
        }
    }
}

impl TryFrom<&ContentType> for EventPayloadResolverInner {
    type Error = Error;

    fn try_from(content_type: &ContentType) -> Result<Self> {
        let mime: mime_guess::Mime = content_type.clone().into();
        match (mime.type_(), mime.subtype()) {
            (mime::APPLICATION, mime::JSON) => Ok(EventPayloadResolverInner::Json),
            (mime::APPLICATION, subtype) if subtype == CONTENT_TYPE_JSON_STREAM_SUBTYPE_STR => {
                Ok(EventPayloadResolverInner::JsonStream)
            }
            (mime::APPLICATION, subtype) if subtype == CONTENT_TYPE_NDJSON_SUBTYPE_STR => {
                Ok(EventPayloadResolverInner::Ndjson)
            }
            (mime::TEXT, mime::PLAIN) => Ok(EventPayloadResolverInner::Text),
            _ => InvalidParameterSnafu {
                reason: format!(
                    "invalid content type: {}, expected: one of {}",
                    content_type,
                    EventPayloadResolver::support_content_type_list().join(", ")
                ),
            }
            .fail(),
        }
    }
}

#[derive(Debug)]
struct EventPayloadResolver<'a> {
    inner: EventPayloadResolverInner,
    /// The content type of the payload.
    /// keep it for logging original content type
    #[allow(dead_code)]
    content_type: &'a ContentType,
}

impl EventPayloadResolver<'_> {
    pub(super) fn support_content_type_list() -> Vec<String> {
        EventPayloadResolverInner::iter()
            .map(|x| x.to_string())
            .collect()
    }
}

impl<'a> TryFrom<&'a ContentType> for EventPayloadResolver<'a> {
    type Error = Error;

    fn try_from(content_type: &'a ContentType) -> Result<Self> {
        let inner = EventPayloadResolverInner::try_from(content_type)?;
        Ok(EventPayloadResolver {
            inner,
            content_type,
        })
    }
}

impl EventPayloadResolver<'_> {
    fn parse_payload(
        &self,
        payload: Bytes,
        ignore_errors: bool,
        utf8_retries: &IntCounterVec,
    ) -> Result<Vec<VrlValue>> {
        match self.inner {
            EventPayloadResolverInner::Json | EventPayloadResolverInner::JsonStream => {
                // Detect UTF-8 failures even when ignore_errors would discard them.
                let mut saw_invalid_utf8 = false;
                let result = transform_ndjson_array_factory(
                    Deserializer::from_slice(&payload)
                        .into_iter()
                        .inspect(|item| {
                            // serde_json keeps its error codes private.
                            saw_invalid_utf8 |= item.as_ref().is_err_and(|error| {
                                error.is_syntax()
                                    && error
                                        .to_string()
                                        .starts_with("invalid unicode code point at line ")
                            });
                        }),
                    ignore_errors,
                );
                if saw_invalid_utf8 && let Cow::Owned(text) = String::from_utf8_lossy(&payload) {
                    // Release partially parsed records before retrying the entire payload.
                    drop(result);
                    let format = if matches!(self.inner, EventPayloadResolverInner::Json) {
                        "json"
                    } else {
                        "json_stream"
                    };
                    utf8_retries.with_label_values(&[format]).inc();
                    return transform_ndjson_array_factory(
                        Deserializer::from_str(&text).into_iter(),
                        ignore_errors,
                    );
                }
                result
            }
            EventPayloadResolverInner::Ndjson => {
                let mut result = Vec::with_capacity(1000);
                let mut buffer = Buffers::new(1000);
                let mut remaining = payload.as_ref();
                let mut line_bytes = Vec::new();
                // read_until keeps the optimized newline search without validating UTF-8.
                for index in 0usize.. {
                    let original = remaining;
                    line_bytes.clear();
                    let read = remaining.read_until(b'\n', &mut line_bytes)?;
                    if read == 0 {
                        break;
                    }

                    let line = &original[..read];
                    let line = if let Some(line) = line.strip_suffix(b"\n") {
                        line.strip_suffix(b"\r").unwrap_or(line)
                    } else {
                        line
                    };
                    if line.is_empty() {
                        continue;
                    }

                    line_bytes.truncate(line.len());
                    let mut parsed =
                        simd_json::serde::from_slice_with_buffers(&mut line_bytes, &mut buffer);
                    // Retry from the original line because SIMD parsing can rewrite its input.
                    if parsed.as_ref().is_err_and(|error| {
                        matches!(error.error(), simd_json::ErrorType::InvalidUtf8)
                    }) && let Cow::Owned(text) = String::from_utf8_lossy(line)
                    {
                        line_bytes = text.into_bytes();
                        utf8_retries.with_label_values(&["ndjson"]).inc();
                        parsed =
                            simd_json::serde::from_slice_with_buffers(&mut line_bytes, &mut buffer);
                    }

                    if let Ok(v) = parsed {
                        result.push(v);
                    } else if !ignore_errors {
                        warn!(
                            "invalid JSON at index: {}, content: {:?}",
                            index,
                            String::from_utf8_lossy(line)
                        );
                        return InvalidParameterSnafu {
                            reason: format!("invalid JSON at index: {}", index),
                        }
                        .fail();
                    }
                }
                Ok(result)
            }
            EventPayloadResolverInner::Text => {
                let result = payload
                    .lines()
                    .filter_map(|line| line.ok().filter(|line| !line.is_empty()))
                    .map(|line| {
                        let mut map = BTreeMap::new();
                        map.insert(
                            KeyString::from("message"),
                            VrlValue::Bytes(Bytes::from(line)),
                        );
                        VrlValue::Object(map)
                    })
                    .collect::<Vec<_>>();
                Ok(result)
            }
        }
    }
}

fn extract_pipeline_value_by_content_type(
    content_type: ContentType,
    payload: Bytes,
    ignore_errors: bool,
) -> Result<Vec<VrlValue>> {
    EventPayloadResolver::try_from(&content_type).and_then(|resolver| {
        resolver
            .parse_payload(payload, ignore_errors, &METRIC_HTTP_LOGS_UTF8_RETRIES_TOTAL)
            .map_err(|e| match &e {
                Error::InvalidParameter { reason, .. } if content_type == *JSON_CONTENT_TYPE => {
                    if reason.contains("invalid item:") {
                        InvalidParameterSnafu {
                            reason: "json format error, please check the date is valid JSON.",
                        }
                        .build()
                    } else {
                        e
                    }
                }
                _ => e,
            })
    })
}

pub(crate) async fn ingest_logs_inner(
    handler: PipelineHandlerRef,
    pipeline: PipelineDefinition,
    log_ingest_requests: Vec<PipelineIngestRequest>,
    query_ctx: QueryContextRef,
    pipeline_params: GreptimePipelineParams,
) -> Result<HttpResponse> {
    // Keep the timer boundary before pipeline execution to preserve existing
    // ingestion elapsed metrics.
    let exec_timer = Instant::now();
    let mut req = ContextReq::default();

    let pipeline_ctx = PipelineContext::new(&pipeline, &pipeline_params, query_ctx.channel());
    for pipeline_req in log_ingest_requests {
        let requests =
            run_pipeline(&handler, &pipeline_ctx, pipeline_req, &query_ctx, true).await?;

        req.merge(requests);
    }

    execute_log_context_req(
        handler,
        req,
        query_ctx,
        exec_timer,
        &METRIC_HTTP_LOGS_INGESTION_COUNTER,
        &METRIC_HTTP_LOGS_INGESTION_ELAPSED,
    )
    .await
}

pub(crate) async fn execute_log_context_req(
    handler: PipelineHandlerRef,
    ctx_req: ContextReq,
    query_ctx: QueryContextRef,
    exec_timer: Instant,
    counter: &IntCounterVec,
    elapsed: &HistogramVec,
) -> Result<HttpResponse> {
    let db = query_ctx.get_db_string();

    let mut total_rows: u64 = 0;
    let mut fail = false;
    let batches = ctx_req.as_req_iter(query_ctx).collect::<Vec<_>>();
    let outputs = handler.insert_all(batches).await?;
    for output in &outputs {
        if let Ok(Output {
            data: OutputData::AffectedRows(rows),
            meta: _,
        }) = &output
        {
            total_rows += *rows as u64;
        } else {
            fail = true;
        }
    }

    // Record one aggregate metric sample for the whole ingestion request.
    if total_rows > 0 {
        counter.with_label_values(&[db.as_str()]).inc_by(total_rows);
        elapsed
            .with_label_values(&[db.as_str(), METRIC_SUCCESS_VALUE])
            .observe(exec_timer.elapsed().as_secs_f64());
    }
    if fail {
        elapsed
            .with_label_values(&[db.as_str(), METRIC_FAILURE_VALUE])
            .observe(exec_timer.elapsed().as_secs_f64());
    }

    let response = GreptimedbV1Response::from_output(outputs)
        .await
        .with_execution_time(exec_timer.elapsed().as_millis() as u64);
    Ok(response)
}

#[async_trait]
pub trait LogValidator: Send + Sync {
    /// validate payload by source before processing
    /// Return a `Some` result to indicate validation failure.
    async fn validate(&self, source: Option<&str>, payload: &Bytes)
    -> Option<Result<HttpResponse>>;
}

pub type LogValidatorRef = Arc<dyn LogValidator + 'static>;

/// axum state struct to hold log handler and validator
#[derive(Clone)]
pub struct LogState {
    pub log_handler: PipelineHandlerRef,
    pub log_validator: Option<LogValidatorRef>,
    pub ingest_interceptor: Option<LogIngestInterceptorRef<Error>>,
}

#[cfg(test)]
mod tests {

    use prometheus::Opts;

    use super::*;

    #[test]
    fn test_transform_ndjson() {
        let s = "{\"a\": 1}\n{\"b\": 2}";
        let a = serde_json::to_string(
            &transform_ndjson_array_factory(Deserializer::from_str(s).into_iter(), false).unwrap(),
        )
        .unwrap();
        assert_eq!(a, "[{\"a\":1},{\"b\":2}]");

        let s = "{\"a\": 1}";
        let a = serde_json::to_string(
            &transform_ndjson_array_factory(Deserializer::from_str(s).into_iter(), false).unwrap(),
        )
        .unwrap();
        assert_eq!(a, "[{\"a\":1}]");

        let s = "[{\"a\": 1}]";
        let a = serde_json::to_string(
            &transform_ndjson_array_factory(Deserializer::from_str(s).into_iter(), false).unwrap(),
        )
        .unwrap();
        assert_eq!(a, "[{\"a\":1}]");

        let s = "[{\"a\": 1}, {\"b\": 2}]";
        let a = serde_json::to_string(
            &transform_ndjson_array_factory(Deserializer::from_str(s).into_iter(), false).unwrap(),
        )
        .unwrap();
        assert_eq!(a, "[{\"a\":1},{\"b\":2}]");
    }

    #[test]
    fn test_extract_by_json_stream_content_type() {
        // `application/x-json-stream` accepts concatenated JSON values, with
        // or without line breaks between them.
        let json_stream = ContentType::from_str("application/x-json-stream").unwrap();
        let payload = r#"{"a": 1}{"b": 2}
{"c": 1}"#
            .as_bytes();
        let payload = Bytes::from_static(payload);

        let values =
            extract_pipeline_value_by_content_type(json_stream.clone(), payload, false).unwrap();
        assert_eq!(
            values,
            vec![
                json!({"a": 1}).into(),
                json!({"b": 2}).into(),
                json!({"c": 1}).into()
            ]
        );

        assert!(
            EventPayloadResolver::support_content_type_list().contains(&json_stream.to_string())
        );
    }

    #[test]
    fn test_json_stream_matches_json_for_concatenated_values() {
        // a concatenated payload is parsed exactly like `application/json`
        let payload = Bytes::from_static(r#"{"a": 1}{"b": 2}"#.as_bytes());
        let via_json =
            extract_pipeline_value_by_content_type(ContentType::json(), payload.clone(), false)
                .unwrap();
        let via_stream = extract_pipeline_value_by_content_type(
            ContentType::from_str("application/x-json-stream").unwrap(),
            payload,
            false,
        )
        .unwrap();
        assert_eq!(via_json, via_stream);
        assert_eq!(via_stream.len(), 2);
    }

    #[test]
    fn test_json_parser_invalid_utf8_errors() {
        for payload in [
            b"{\"message\":\"bad\xff\"}".as_slice(),
            b"{\"message\":\"bad\xc1\x89\xc8\"}".as_slice(),
            b"{\"message\":\"bad\xe2\x82\"}".as_slice(),
            b"{\"message\":\"line\\nbad\xff\"}".as_slice(),
            b"{\"bad\xff\":1}".as_slice(),
        ] {
            let error = Deserializer::from_slice(payload)
                .into_iter::<VrlValue>()
                .next()
                .unwrap()
                .unwrap_err();
            assert!(error.is_syntax(), "{error:?}");
            assert!(
                error
                    .to_string()
                    .starts_with("invalid unicode code point at line "),
                "{error:?}"
            );

            let mut line = payload.to_vec();
            let error = simd_json::serde::from_slice_with_buffers::<VrlValue>(
                &mut line,
                &mut Buffers::new(payload.len()),
            )
            .unwrap_err();
            assert_eq!(error.error(), &simd_json::ErrorType::InvalidUtf8);
        }
    }

    #[test]
    fn test_json_does_not_retry_other_errors() {
        for payload in [
            b"{\"message\":\"bad\xff\\q\"}".as_slice(),
            b"{\"message\":\"bad\xff".as_slice(),
        ] {
            let original_error = Deserializer::from_slice(payload)
                .into_iter::<VrlValue>()
                .next()
                .unwrap()
                .unwrap_err();
            assert!(
                !original_error
                    .to_string()
                    .starts_with("invalid unicode code point at line ")
            );

            for content_type in [ContentType::json(), JSON_STREAM_CONTENT_TYPE.clone()] {
                let error = extract_pipeline_value_by_content_type(
                    content_type,
                    Bytes::from_static(payload),
                    false,
                )
                .unwrap_err();
                let Error::ParseJson { error: source, .. } = error else {
                    panic!("expected a JSON parse error, got {error}");
                };
                assert_eq!(source.to_string(), original_error.to_string());
            }
        }
    }

    #[test]
    fn test_utf8_retry_counter() {
        // Keep exact retry counts independent of other parser tests.
        let utf8_retries = IntCounterVec::new(
            Opts::new("test_utf8_retries", "UTF-8 retry attempts"),
            &["format"],
        )
        .unwrap();
        for (content_type, format, multi_record_retries) in [
            (ContentType::json(), "json", 1),
            (JSON_STREAM_CONTENT_TYPE.clone(), "json_stream", 1),
            (NDJSON_CONTENT_TYPE.clone(), "ndjson", 2),
        ] {
            let resolver = EventPayloadResolver::try_from(&content_type).unwrap();
            let counter = utf8_retries.with_label_values(&[format]);
            for ignore_errors in [false, true] {
                for (payload, retries, valid_after_retry) in [
                    (br#"{"message":"ok"}"#.as_slice(), 0, true),
                    (
                        b"{\"message\":\"bad\xff\"}\n{\"message\":\"bad\xfe\"}".as_slice(),
                        multi_record_retries,
                        true,
                    ),
                    (br#"{"broken":}"#.as_slice(), 0, false),
                    (br#"{"message":"\uDC00"}"#.as_slice(), 0, false),
                    (
                        b"{\"message\":\"bad\xff\",\"broken\":}".as_slice(),
                        1,
                        false,
                    ),
                    // SIMD validates UTF-8 before decoding the invalid escape.
                    (
                        b"{\"message\":\"bad\xff\\q\"}".as_slice(),
                        u64::from(format == "ndjson"),
                        false,
                    ),
                ] {
                    let before = counter.get();
                    let result = resolver.parse_payload(
                        Bytes::from_static(payload),
                        ignore_errors,
                        &utf8_retries,
                    );
                    assert_eq!(result.is_ok(), valid_after_retry || ignore_errors);
                    assert_eq!(
                        counter.get() - before,
                        retries,
                        "format={format}, ignore_errors={ignore_errors}, payload={payload:?}"
                    );
                }
            }
        }
    }

    #[test]
    fn test_json_invalid_utf8_retry() {
        let expected = vec![
            json!({"message": "before"}).into(),
            json!({"message": "bad\u{fffd}\u{fffd}\u{fffd}\u{7f}\u{7f}"}).into(),
            json!({"message": "after"}).into(),
        ];

        for payload in [
            b"[{\"message\":\"before\"},{\"message\":\"bad\xc1\x89\xc8\x7f\x7f\"},{\"message\":\"after\"}]".as_slice(),
            b"{\"message\":\"before\"}\n{\"message\":\"bad\xc1\x89\xc8\x7f\x7f\"}\n{\"message\":\"after\"}".as_slice(),
        ] {
            for content_type in [ContentType::json(), JSON_STREAM_CONTENT_TYPE.clone()] {
                for ignore_errors in [false, true] {
                    let actual = extract_pipeline_value_by_content_type(
                        content_type.clone(),
                        Bytes::from_static(payload),
                        ignore_errors,
                    )
                    .unwrap();
                    assert_eq!(actual, expected);
                }
            }
        }
    }

    #[test]
    fn test_json_invalid_utf8_retry_preserves_parse_errors() {
        let cases = [
            (
                br#"{"message":"before"}{"message":"\uD800"}{"message":"after"}"#.as_slice(),
                vec![json!({"message": "before"}).into()],
            ),
            (
                b"{\"message\":\"before\"}{\"message\":\"bad\xff\"}{\"broken\":}".as_slice(),
                vec![
                    json!({"message": "before"}).into(),
                    json!({"message": "bad\u{fffd}"}).into(),
                ],
            ),
        ];

        for (payload, expected) in cases {
            for content_type in [ContentType::json(), JSON_STREAM_CONTENT_TYPE.clone()] {
                for ignore_errors in [false, true] {
                    let result = extract_pipeline_value_by_content_type(
                        content_type.clone(),
                        Bytes::from_static(payload),
                        ignore_errors,
                    );
                    if ignore_errors {
                        assert_eq!(result.unwrap(), expected);
                    } else {
                        assert!(matches!(result, Err(Error::ParseJson { .. })));
                    }
                }
            }
        }
    }

    #[test]
    fn test_json_invalid_utf8_retry_preserves_ignored_items() {
        let payload = Bytes::from_static(b"0{\"message\":\"bad\xff\"}false{\"message\":\"after\"}");
        for content_type in [ContentType::json(), JSON_STREAM_CONTENT_TYPE.clone()] {
            let result = extract_pipeline_value_by_content_type(
                content_type.clone(),
                payload.clone(),
                false,
            );
            assert!(matches!(result, Err(Error::InvalidParameter { .. })));

            let actual =
                extract_pipeline_value_by_content_type(content_type, payload.clone(), true)
                    .unwrap();
            assert_eq!(
                actual,
                vec![
                    json!({"message": "bad\u{fffd}"}).into(),
                    json!({"message": "after"}).into(),
                ],
            );
        }
    }

    #[test]
    fn test_ndjson_invalid_utf8_retry() {
        for invalid_bytes in [1, 8192] {
            let mut payload = b"\n{\"message\":\"before\"}\r\n\r\n{\"escaped\":\"line\\nquote\\\"slash\\\\\",\"message\":\"bad".to_vec();
            payload.extend(std::iter::repeat_n(0xff, invalid_bytes));
            payload.extend_from_slice(b"\"}\r\n{\"message\":\"after\"}");
            let payload = Bytes::from(payload);
            let expected = vec![
                json!({"message": "before"}).into(),
                json!({
                    "escaped": "line\nquote\"slash\\",
                    "message": format!("bad{}", "\u{fffd}".repeat(invalid_bytes)),
                })
                .into(),
                json!({"message": "after"}).into(),
            ];

            for ignore_errors in [false, true] {
                let actual = extract_pipeline_value_by_content_type(
                    NDJSON_CONTENT_TYPE.clone(),
                    payload.clone(),
                    ignore_errors,
                )
                .unwrap();
                assert_eq!(actual, expected);
            }
        }
    }

    #[test]
    fn test_ndjson_invalid_utf8_retry_preserves_parse_errors() {
        let too_deep = format!("{}0{}", "[".repeat(1025), "]".repeat(1025));
        for bad_line in [
            br#"{"escaped":"line\n","broken":}"#.as_slice(),
            b"{\"escaped\":\"line\\n\",\"message\":\"bad\xff\",\"broken\":}".as_slice(),
            br#"{"escaped":"line\n","message":"\uDC00"}"#.as_slice(),
            br#"{"message":"\uD800"}"#.as_slice(),
            br#"{"value":1e4294967296}"#.as_slice(),
            b"{\"value\":123\0}".as_slice(),
            b"{\"message\":\"bad\xe2\x82".as_slice(),
            too_deep.as_bytes(),
        ] {
            let mut payload = b"{\"message\":\"before\"}\n".to_vec();
            payload.extend_from_slice(bad_line);
            payload.extend_from_slice(b"\n{\"message\":\"after\"}\n");
            let payload = Bytes::from(payload);

            for ignore_errors in [false, true] {
                let result = extract_pipeline_value_by_content_type(
                    NDJSON_CONTENT_TYPE.clone(),
                    payload.clone(),
                    ignore_errors,
                );
                if ignore_errors {
                    assert_eq!(
                        result.unwrap(),
                        vec![
                            json!({"message": "before"}).into(),
                            json!({"message": "after"}).into(),
                        ],
                    );
                } else {
                    assert!(matches!(
                        result,
                        Err(Error::InvalidParameter { reason, .. })
                            if reason.ends_with("at index: 1")
                    ));
                }
            }
        }
    }

    #[test]
    fn test_ndjson_large_exponents() {
        let payload = Bytes::from_static(br#"{"zero":0e99999999999,"underflow":1e-4294967295}"#);
        for ignore_errors in [false, true] {
            let actual = extract_pipeline_value_by_content_type(
                NDJSON_CONTENT_TYPE.clone(),
                payload.clone(),
                ignore_errors,
            )
            .unwrap();
            assert_eq!(actual, vec![json!({"zero": 0.0, "underflow": 0.0}).into()]);
        }
    }

    #[test]
    fn test_ndjson_empty_and_whitespace_lines() {
        for ignore_errors in [false, true] {
            for payload in [b"".as_slice(), b"\n\r\n".as_slice()] {
                let actual = extract_pipeline_value_by_content_type(
                    NDJSON_CONTENT_TYPE.clone(),
                    Bytes::from_static(payload),
                    ignore_errors,
                )
                .unwrap();
                assert!(actual.is_empty());
            }

            for payload in [b"\r".as_slice(), b" \n".as_slice(), b"\r\r\n".as_slice()] {
                let result = extract_pipeline_value_by_content_type(
                    NDJSON_CONTENT_TYPE.clone(),
                    Bytes::from_static(payload),
                    ignore_errors,
                );
                if ignore_errors {
                    assert!(result.unwrap().is_empty());
                } else {
                    assert!(matches!(result, Err(Error::InvalidParameter { .. })));
                }
            }
        }
    }

    #[test]
    fn test_extract_by_content() {
        let payload = r#"
        {"a": 1}
        {"b": 2"}
        {"c": 1}
"#
        .as_bytes();
        let payload = Bytes::from_static(payload);

        let fail_rest =
            extract_pipeline_value_by_content_type(ContentType::json(), payload.clone(), true);
        assert!(fail_rest.is_ok());
        assert_eq!(fail_rest.unwrap(), vec![json!({"a": 1}).into()]);

        let fail_only_wrong =
            extract_pipeline_value_by_content_type(NDJSON_CONTENT_TYPE.clone(), payload, true);
        assert!(fail_only_wrong.is_ok());

        let mut map1 = BTreeMap::new();
        map1.insert(KeyString::from("a"), VrlValue::Integer(1));
        let map1 = VrlValue::Object(map1);
        let mut map2 = BTreeMap::new();
        map2.insert(KeyString::from("c"), VrlValue::Integer(1));
        let map2 = VrlValue::Object(map2);
        assert_eq!(fail_only_wrong.unwrap(), vec![map1, map2]);
    }
}