fn append_search_values(value: &Value, output: &mut String) {
    match value {
        Value::Null => {}
        Value::Bool(value) => output.push_str(if *value { " true" } else { " false" }),
        Value::Number(value) => {
            output.push(' ');
            output.push_str(&value.to_string());
        }
        Value::String(value) => {
            output.push(' ');
            output.push_str(&value.to_ascii_lowercase());
        }
        Value::Array(values) => values
            .iter()
            .for_each(|value| append_search_values(value, output)),
        Value::Object(values) => values
            .values()
            .for_each(|value| append_search_values(value, output)),
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum FileLogSearchClause {
    Any {
        value: String,
        negated: bool,
    },
    Field {
        field: String,
        value: String,
        negated: bool,
    },
}

fn tokenize_file_log_search(input: &str) -> Result<Vec<String>, ApiError> {
    let mut tokens = Vec::new();
    let mut current = String::new();
    let mut quoted = false;
    let mut escaped = false;
    for ch in input.chars() {
        if escaped {
            current.push(ch);
            escaped = false;
            continue;
        }
        if quoted && ch == '\\' {
            escaped = true;
            continue;
        }
        if ch == '"' {
            quoted = !quoted;
            continue;
        }
        if ch.is_whitespace() && !quoted {
            if !current.is_empty() {
                tokens.push(std::mem::take(&mut current));
            }
            continue;
        }
        current.push(ch);
    }
    if quoted || escaped {
        return Err(ApiError(
            StatusCode::BAD_REQUEST,
            "invalid file log search syntax".into(),
        ));
    }
    if !current.is_empty() {
        tokens.push(current);
    }
    Ok(tokens)
}

fn parse_file_log_search(input: &str) -> Result<Vec<FileLogSearchClause>, ApiError> {
    let tokens = tokenize_file_log_search(input)?;
    let mut clauses = Vec::new();
    let mut index = 0usize;
    while index < tokens.len() {
        let token = &tokens[index];
        if token.eq_ignore_ascii_case("AND") {
            index += 1;
            continue;
        }
        if token.eq_ignore_ascii_case("OR") {
            return Err(ApiError(
                StatusCode::BAD_REQUEST,
                "file log search supports AND and negation, not OR".into(),
            ));
        }
        let (negated, term) = token
            .strip_prefix('!')
            .map_or((false, token.as_str()), |value| (true, value));
        if term.is_empty() {
            return Err(ApiError(
                StatusCode::BAD_REQUEST,
                "invalid file log search syntax".into(),
            ));
        }
        if let Some((field, inline_value)) = term.split_once(':') {
            if field.is_empty()
                || field.len() > 128
                || !field
                    .chars()
                    .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.'))
            {
                return Err(ApiError(
                    StatusCode::BAD_REQUEST,
                    "invalid file log search field".into(),
                ));
            }
            let value = if inline_value.is_empty() {
                index += 1;
                let Some(next) = tokens.get(index) else {
                    return Err(ApiError(
                        StatusCode::BAD_REQUEST,
                        "missing file log search value".into(),
                    ));
                };
                if next.eq_ignore_ascii_case("AND") || next.eq_ignore_ascii_case("OR") {
                    return Err(ApiError(
                        StatusCode::BAD_REQUEST,
                        "missing file log search value".into(),
                    ));
                }
                next.clone()
            } else {
                inline_value.to_owned()
            };
            if value.is_empty() || value.len() > 256 {
                return Err(ApiError(
                    StatusCode::BAD_REQUEST,
                    "invalid file log search value".into(),
                ));
            }
            clauses.push(FileLogSearchClause::Field {
                field: field.to_ascii_lowercase(),
                value: value.to_ascii_lowercase(),
                negated,
            });
        } else {
            clauses.push(FileLogSearchClause::Any {
                value: term.to_ascii_lowercase(),
                negated,
            });
        }
        if clauses.len() > 32 {
            return Err(ApiError(
                StatusCode::BAD_REQUEST,
                "too many file log search clauses".into(),
            ));
        }
        index += 1;
    }
    Ok(clauses)
}

fn file_log_field_alias(field: &str) -> &str {
    match field {
        "req" | "request" => "method",
        "ip" | "client" => "client_ip",
        "source" => "_source",
        "time" => "timestamp",
        other => other,
    }
}

fn file_log_field<'a>(value: &'a Value, field: &str) -> Option<&'a Value> {
    let mut current = value;
    for part in file_log_field_alias(field).split('.') {
        let object = current.as_object()?;
        current = object.get(part).or_else(|| {
            object
                .iter()
                .find(|(key, _)| key.eq_ignore_ascii_case(part))
                .map(|(_, value)| value)
        })?;
    }
    Some(current)
}

fn wildcard_match_ci(candidate: &str, pattern: &str) -> bool {
    if !pattern.contains('*') {
        return candidate.contains(pattern);
    }
    if pattern == "*" {
        return true;
    }
    let starts_wild = pattern.starts_with('*');
    let ends_wild = pattern.ends_with('*');
    let parts = pattern
        .split('*')
        .filter(|part| !part.is_empty())
        .collect::<Vec<_>>();
    if parts.is_empty() {
        return true;
    }
    let mut cursor = 0usize;
    for (index, part) in parts.iter().enumerate() {
        let Some(found) = candidate[cursor..].find(part) else {
            return false;
        };
        if index == 0 && !starts_wild && found != 0 {
            return false;
        }
        cursor += found + part.len();
    }
    ends_wild || cursor == candidate.len()
}

fn numeric_match(candidate: f64, pattern: &str) -> Option<bool> {
    if let Some(raw) = pattern.strip_prefix(">=") {
        return raw
            .parse::<f64>()
            .ok()
            .map(|expected| candidate >= expected);
    }
    if let Some(raw) = pattern.strip_prefix("<=") {
        return raw
            .parse::<f64>()
            .ok()
            .map(|expected| candidate <= expected);
    }
    if let Some(raw) = pattern.strip_prefix('>') {
        return raw.parse::<f64>().ok().map(|expected| candidate > expected);
    }
    if let Some(raw) = pattern.strip_prefix('<') {
        return raw.parse::<f64>().ok().map(|expected| candidate < expected);
    }
    None
}

fn file_log_value_matches(value: &Value, pattern: &str) -> bool {
    match value {
        Value::Null => pattern == "null",
        Value::Bool(value) => wildcard_match_ci(if *value { "true" } else { "false" }, pattern),
        Value::Number(value) => {
            let rendered = value.to_string();
            if pattern.len() == 3 && pattern.ends_with("xx") {
                return rendered.as_bytes().first() == pattern.as_bytes().first();
            }
            numeric_match(value.as_f64().unwrap_or_default(), pattern)
                .unwrap_or_else(|| wildcard_match_ci(&rendered, pattern))
        }
        Value::String(value) => {
            let candidate = value.to_ascii_lowercase();
            if pattern.ends_with("xx") && pattern.len() == 3 && candidate.len() == 3 {
                return candidate.as_bytes().first() == pattern.as_bytes().first();
            }
            if let Ok(candidate_number) = candidate.parse::<f64>() {
                if let Some(result) = numeric_match(candidate_number, pattern) {
                    return result;
                }
            }
            wildcard_match_ci(&candidate, pattern)
        }
        Value::Array(values) => values
            .iter()
            .any(|value| file_log_value_matches(value, pattern)),
        Value::Object(_) => {
            let mut rendered = String::new();
            append_search_values(value, &mut rendered);
            wildcard_match_ci(rendered.trim(), pattern)
        }
    }
}

fn record_matches_search(value: &Value, clauses: &[FileLogSearchClause]) -> bool {
    clauses.iter().all(|clause| {
        let (matched, negated) = match clause {
            FileLogSearchClause::Any {
                value: needle,
                negated,
            } => {
                let mut searchable = String::new();
                append_search_values(value, &mut searchable);
                (wildcard_match_ci(searchable.trim(), needle), *negated)
            }
            FileLogSearchClause::Field {
                field,
                value: needle,
                negated,
            } => (
                file_log_field(value, field)
                    .is_some_and(|value| file_log_value_matches(value, needle)),
                *negated,
            ),
        };
        matched != negated
    })
}

fn valid_raw_field_key(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 128
        && value
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.'))
}

fn parse_raw_error_record(line: &str) -> Value {
    let mut parts = line.split_whitespace();
    let Some(timestamp) = parts.next() else {
        return json!({"raw": line});
    };
    let Some(level) = parts.next() else {
        return json!({"raw": line});
    };
    let Some(target) = parts.next() else {
        return json!({"raw": line});
    };
    let rest = parts.collect::<Vec<_>>();
    let first_field = rest.iter().position(|token| {
        token
            .split_once('=')
            .is_some_and(|(key, _)| valid_raw_field_key(key))
    });
    let message_end = first_field.unwrap_or(rest.len());
    let mut object = serde_json::Map::new();
    object.insert("timestamp".into(), Value::String(timestamp.into()));
    object.insert("level".into(), Value::String(level.into()));
    object.insert("target".into(), Value::String(target.into()));
    if message_end > 0 {
        object.insert(
            "message".into(),
            Value::String(rest[..message_end].join(" ")),
        );
    }
    let mut index = message_end;
    while index < rest.len() {
        let Some((key, start)) = rest[index].split_once('=') else {
            index += 1;
            continue;
        };
        if !valid_raw_field_key(key) {
            index += 1;
            continue;
        }
        let mut pieces = vec![start];
        index += 1;
        while index < rest.len()
            && !rest[index]
                .split_once('=')
                .is_some_and(|(next_key, _)| valid_raw_field_key(next_key))
        {
            pieces.push(rest[index]);
            index += 1;
        }
        object.insert(key.into(), Value::String(pieces.join(" ")));
    }
    object.insert("raw".into(), Value::String(line.into()));
    Value::Object(object)
}

fn collect_file_log_fields(value: &Value, output: &mut std::collections::BTreeSet<String>) {
    fn walk(prefix: &str, value: &Value, output: &mut std::collections::BTreeSet<String>) {
        if output.len() >= 64 {
            return;
        }
        match value {
            Value::Object(object) => {
                for (key, child) in object {
                    if key == "_source_kind" || output.len() >= 64 {
                        continue;
                    }
                    let path = if prefix.is_empty() {
                        key.clone()
                    } else {
                        format!("{prefix}.{key}")
                    };
                    if child.is_object() {
                        walk(&path, child, output);
                    } else {
                        output.insert(path);
                    }
                }
            }
            _ if !prefix.is_empty() => {
                output.insert(prefix.into());
            }
            _ => {}
        }
    }
    walk("", value, output);
}