CREATE MATERIALIZED VIEW ja4_logs.mv_http_logs
TO ja4_logs.http_logs
AS
WITH
    coalesce(JSONExtractString(raw_json, 'header_User-Agent'), '')                         AS _ua,
    toIPv6(toIPv4(coalesce(JSONExtractString(raw_json, 'src_ip'), '0.0.0.0')))            AS _ip,
    toUInt32(dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'asn', _ip, toUInt32(0)))  AS _asn,
    dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'country_code', _ip, '')            AS _cc
SELECT
    parseDateTimeBestEffort(coalesce(JSONExtractString(raw_json, 'time'), '1970-01-01T00:00:00Z')) AS time,
    toDate(time)                                                                     AS log_date,
    toIPv4(coalesce(JSONExtractString(raw_json, 'src_ip'), '0.0.0.0'))              AS src_ip,
    toUInt16(coalesce(JSONExtractUInt(raw_json, 'src_port'), 0))                    AS src_port,
    _asn                                                                             AS src_asn,
    _cc                                                                              AS src_country_code,
    toIPv4(coalesce(JSONExtractString(raw_json, 'dst_ip'), '0.0.0.0'))              AS dst_ip,
    toUInt16(coalesce(JSONExtractUInt(raw_json, 'dst_port'), 0))                    AS dst_port,
    dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'name',   _ip, '')           AS src_as_name,
    dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'org',    _ip, '')           AS src_org,
    dictGetOrDefault('ja4_processing.dict_iplocate_asn', 'domain', _ip, '')           AS src_domain,
    coalesce(JSONExtractString(raw_json, 'method'),       '') AS method,
    coalesce(JSONExtractString(raw_json, 'scheme'),       '') AS scheme,
    coalesce(JSONExtractString(raw_json, 'host'),         '') AS host,
    coalesce(JSONExtractString(raw_json, 'path'),         '') AS path,
    coalesce(JSONExtractString(raw_json, 'query'),        '') AS query,
    coalesce(JSONExtractString(raw_json, 'http_version'), '') AS http_version,
    coalesce(JSONExtractString(raw_json, 'orphan_side'),  '') AS orphan_side,
    toUInt8(coalesce(JSONExtractBool(raw_json, 'correlated'),   0)) AS correlated,
    toUInt16(coalesce(JSONExtractUInt(raw_json, 'keepalives'),  0)) AS keepalives,
    coalesce(JSONExtractUInt(raw_json, 'a_timestamp'), 0)           AS a_timestamp,
    coalesce(JSONExtractUInt(raw_json, 'b_timestamp'), 0)           AS b_timestamp,
    coalesce(JSONExtractString(raw_json, 'conn_id'), '')             AS conn_id,
    toUInt8(coalesce(JSONExtractBool(raw_json, 'ip_meta_df'),        0)) AS ip_meta_df,
    toUInt16(coalesce(JSONExtractUInt(raw_json, 'ip_meta_id'),       0)) AS ip_meta_id,
    toUInt16(coalesce(JSONExtractUInt(raw_json, 'ip_meta_total_length'), 0)) AS ip_meta_total_length,
    toUInt8(coalesce(JSONExtractUInt(raw_json, 'ip_meta_ttl'),       0)) AS ip_meta_ttl,
    coalesce(JSONExtractString(raw_json, 'tcp_meta_options'), '')         AS tcp_meta_options,
    toUInt32(coalesce(JSONExtractUInt(raw_json, 'tcp_meta_window_size'), 0)) AS tcp_meta_window_size,
    toUInt16(coalesce(JSONExtractUInt(raw_json, 'tcp_meta_mss'),     0)) AS tcp_meta_mss,
    toUInt8(coalesce(JSONExtractUInt(raw_json, 'tcp_meta_window_scale'), 0)) AS tcp_meta_window_scale,
    toInt32(coalesce(JSONExtractInt(raw_json, 'syn_to_clienthello_ms'), 0)) AS syn_to_clienthello_ms,
    coalesce(JSONExtractString(raw_json, 'tls_version'), '') AS tls_version,
    coalesce(JSONExtractString(raw_json, 'tls_sni'),     '') AS tls_sni,
    coalesce(JSONExtractString(raw_json, 'tls_alpn'),    '') AS tls_alpn,
    coalesce(JSONExtractString(raw_json, 'ja3'),         '') AS ja3,
    coalesce(JSONExtractString(raw_json, 'ja3_hash'),    '') AS ja3_hash,
    coalesce(JSONExtractString(raw_json, 'ja4'),         '') AS ja4,
    coalesce(JSONExtractString(raw_json, 'client_headers'), '')                  AS client_headers,
    coalesce(JSONExtractString(raw_json, 'header_User-Agent'), '')               AS header_user_agent,
    coalesce(JSONExtractString(raw_json, 'header_Accept'), '')                   AS header_accept,
    coalesce(JSONExtractString(raw_json, 'header_Accept-Encoding'), '')          AS header_accept_encoding,
    coalesce(JSONExtractString(raw_json, 'header_Accept-Language'), '')          AS header_accept_language,
    coalesce(JSONExtractString(raw_json, 'header_Content-Type'), '')             AS header_content_type,
    coalesce(JSONExtractString(raw_json, 'header_X-Request-Id'), '')             AS header_x_request_id,
    coalesce(JSONExtractString(raw_json, 'header_X-Trace-Id'), '')               AS header_x_trace_id,
    coalesce(JSONExtractString(raw_json, 'header_X-Forwarded-For'), '')          AS header_x_forwarded_for,
    coalesce(JSONExtractString(raw_json, 'header_Sec-CH-UA'), '')                AS header_sec_ch_ua,
    coalesce(JSONExtractString(raw_json, 'header_Sec-CH-UA-Mobile'), '')         AS header_sec_ch_ua_mobile,
    coalesce(JSONExtractString(raw_json, 'header_Sec-CH-UA-Platform'), '')       AS header_sec_ch_ua_platform,
    coalesce(JSONExtractString(raw_json, 'header_Sec-Fetch-Dest'), '')           AS header_sec_fetch_dest,
    coalesce(JSONExtractString(raw_json, 'header_Sec-Fetch-Mode'), '')           AS header_sec_fetch_mode,
    coalesce(JSONExtractString(raw_json, 'header_Sec-Fetch-Site'), '')           AS header_sec_fetch_site,

    -- Anubis enrichment : logique de correspondance combinée UA+IP
    -- Priorité : (1) UA+IP [même rule_id] > (2) UA seul > (3) IP seul > (4) ASN > (5) Pays
    CASE
        WHEN dictGet('ja4_processing.dict_anubis_ua', 'has_ip', _ua) = '1'
             AND dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua) != ''
             AND dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '') != ''
             AND toUInt64OrZero(dictGet('ja4_processing.dict_anubis_ua', 'rule_id', _ua))
                 = dictGetOrDefault('ja4_processing.dict_anubis_ip', 'rule_id', _ip, toUInt64(0))
            THEN dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua)
        WHEN dictGet('ja4_processing.dict_anubis_ua', 'has_ip', _ua) = '0'
             AND dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua) != ''
            THEN dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua)
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_ip', 'has_ua', _ip, toUInt8(0)) = 0
             AND dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '')
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_asn', 'bot_name', _asn, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_asn', 'bot_name', _asn, '')
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_country', 'bot_name', _cc, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_country', 'bot_name', _cc, '')
        ELSE ''
    END AS anubis_bot_name,

    CASE
        WHEN dictGet('ja4_processing.dict_anubis_ua', 'has_ip', _ua) = '1'
             AND dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua) != ''
             AND dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '') != ''
             AND toUInt64OrZero(dictGet('ja4_processing.dict_anubis_ua', 'rule_id', _ua))
                 = dictGetOrDefault('ja4_processing.dict_anubis_ip', 'rule_id', _ip, toUInt64(0))
            THEN dictGet('ja4_processing.dict_anubis_ua', 'action', _ua)
        WHEN dictGet('ja4_processing.dict_anubis_ua', 'has_ip', _ua) = '0'
             AND dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua) != ''
            THEN dictGet('ja4_processing.dict_anubis_ua', 'action', _ua)
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_ip', 'has_ua', _ip, toUInt8(0)) = 0
             AND dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_ip', 'action', _ip, '')
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_asn', 'bot_name', _asn, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_asn', 'action', _asn, '')
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_country', 'bot_name', _cc, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_country', 'action', _cc, '')
        ELSE ''
    END AS anubis_bot_action,

    CASE
        WHEN dictGet('ja4_processing.dict_anubis_ua', 'has_ip', _ua) = '1'
             AND dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua) != ''
             AND dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '') != ''
             AND toUInt64OrZero(dictGet('ja4_processing.dict_anubis_ua', 'rule_id', _ua))
                 = dictGetOrDefault('ja4_processing.dict_anubis_ip', 'rule_id', _ip, toUInt64(0))
            THEN dictGet('ja4_processing.dict_anubis_ua', 'category', _ua)
        WHEN dictGet('ja4_processing.dict_anubis_ua', 'has_ip', _ua) = '0'
             AND dictGet('ja4_processing.dict_anubis_ua', 'bot_name', _ua) != ''
            THEN dictGet('ja4_processing.dict_anubis_ua', 'category', _ua)
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_ip', 'has_ua', _ip, toUInt8(0)) = 0
             AND dictGetOrDefault('ja4_processing.dict_anubis_ip', 'bot_name', _ip, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_ip', 'category', _ip, '')
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_asn', 'bot_name', _asn, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_asn', 'category', _asn, '')
        WHEN dictGetOrDefault('ja4_processing.dict_anubis_country', 'bot_name', _cc, '') != ''
            THEN dictGetOrDefault('ja4_processing.dict_anubis_country', 'category', _cc, '')
        ELSE ''
    END AS anubis_bot_category

FROM ja4_logs.http_logs_raw