-- ============================================================================
-- CloudFront Access Logs -> ClickHouse Cloud via S3Queue
-- ============================================================================
-- This schema is for CloudFront Standard Logging (Legacy) format.
-- Legacy logs are tab-separated W3C extended format with 33 fields.
--
-- If using Standard Logging v2 with JSON or Parquet output, you'll need
-- different format settings (JSONEachRow or Parquet respectively).
-- ============================================================================

-- ============================================================================
-- 1. DESTINATION TABLE (MergeTree)
-- ============================================================================
-- This table stores the parsed CloudFront logs for querying.
-- Adjust ORDER BY and PARTITION BY based on your query patterns.

CREATE TABLE IF NOT EXISTS cloudfront_logs
(
    -- Timestamp: combined from date + time fields
    timestamp DateTime64(3) DEFAULT now64(),

    -- Edge location that served the request (e.g., 'DFW3', 'IAD89-C1')
    x_edge_location LowCardinality(String),

    -- Bytes sent from server to client
    sc_bytes UInt64 DEFAULT 0,

    -- Client IP address
    c_ip String,

    -- HTTP method (GET, POST, HEAD, etc.)
    cs_method LowCardinality(String),

    -- CloudFront distribution domain name
    cs_host String,

    -- URI path (e.g., '/images/logo.png')
    cs_uri_stem String,

    -- HTTP status code returned to viewer
    sc_status UInt16 DEFAULT 0,

    -- Referer header
    cs_referer String DEFAULT '',

    -- User-Agent header
    cs_user_agent String DEFAULT '',

    -- Query string (without leading '?')
    cs_uri_query String DEFAULT '',

    -- Cookie header (if cookie logging enabled)
    cs_cookie String DEFAULT '',

    -- Edge result type: Hit, Miss, Error, RefreshHit, etc.
    x_edge_result_type LowCardinality(String),

    -- Unique request identifier
    x_edge_request_id String,

    -- Host header from viewer request (may differ from cs_host for CNAMEs)
    x_host_header String DEFAULT '',

    -- Protocol: http or https
    cs_protocol LowCardinality(String),

    -- Bytes received from client
    cs_bytes UInt64 DEFAULT 0,

    -- Time to process request (seconds)
    time_taken Float64 DEFAULT 0,

    -- X-Forwarded-For header (client IP if behind proxy)
    x_forwarded_for String DEFAULT '',

    -- SSL/TLS protocol version (e.g., 'TLSv1.2', 'TLSv1.3')
    ssl_protocol LowCardinality(String) DEFAULT '',

    -- SSL/TLS cipher suite
    ssl_cipher LowCardinality(String) DEFAULT '',

    -- How CloudFront classified response before returning to viewer
    x_edge_response_result_type LowCardinality(String) DEFAULT '',

    -- HTTP protocol version (e.g., 'HTTP/1.1', 'HTTP/2.0')
    cs_protocol_version LowCardinality(String) DEFAULT '',

    -- Field-level encryption status
    fle_status LowCardinality(String) DEFAULT '',

    -- Number of fields encrypted with FLE
    fle_encrypted_fields UInt16 DEFAULT 0,

    -- Client port number
    c_port UInt16 DEFAULT 0,

    -- Time to first byte (seconds)
    time_to_first_byte Float64 DEFAULT 0,

    -- Detailed result type for error diagnosis
    x_edge_detailed_result_type LowCardinality(String) DEFAULT '',

    -- Content-Type of response
    sc_content_type LowCardinality(String) DEFAULT '',

    -- Content-Length of response
    sc_content_len UInt64 DEFAULT 0,

    -- Range request start byte
    sc_range_start Int64 DEFAULT -1,

    -- Range request end byte
    sc_range_end Int64 DEFAULT -1,

    -- Materialized columns for common query patterns
    date Date MATERIALIZED toDate(timestamp),
    hour UInt8 MATERIALIZED toHour(timestamp),
    is_cache_hit UInt8 MATERIALIZED x_edge_result_type = 'Hit',
    is_error UInt8 MATERIALIZED sc_status >= 400
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(timestamp)
ORDER BY (timestamp, c_ip)
TTL timestamp + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;


-- ============================================================================
-- 2. S3QUEUE TABLE (Streaming Source)
-- ============================================================================
-- This table continuously polls S3 for new CloudFront log files.
--
-- IMPORTANT: Replace the following placeholders:
--   - YOUR_BUCKET_NAME: Your S3 bucket name
--   - YOUR_PREFIX: Optional prefix/path to logs
--   - YOUR_REGION: AWS region (e.g., us-east-1)
--
-- Authentication: Uses ClickHouse Cloud's IAM role (no credentials needed).
-- This role will assume a IAM role in your account and this new Role must
-- have access to read the S3 bucket.
--
-- Path pattern for legacy logging (no date folders):
--   s3://bucket/prefix/*.gz
--
-- Path pattern for nested folders
--   s3://bucket/prefix/**.gz

CREATE TABLE IF NOT EXISTS cloudfront_logs_queue
(
    -- Raw fields as they appear in CloudFront logs (tab-separated)
    date String,
    time String,
    x_edge_location String,
    sc_bytes String,
    c_ip String,
    cs_method String,
    cs_host String,
    cs_uri_stem String,
    sc_status String,
    cs_referer String,
    cs_user_agent String,
    cs_uri_query String,
    cs_cookie String,
    x_edge_result_type String,
    x_edge_request_id String,
    x_host_header String,
    cs_protocol String,
    cs_bytes String,
    time_taken String,
    x_forwarded_for String,
    ssl_protocol String,
    ssl_cipher String,
    x_edge_response_result_type String,
    cs_protocol_version String,
    fle_status String,
    fle_encrypted_fields String,
    c_port String,
    time_to_first_byte String,
    x_edge_detailed_result_type String,
    sc_content_type String,
    sc_content_len String,
    sc_range_start String,
    sc_range_end String
)
ENGINE = S3Queue(
    'https://<BUCKET_NAME>.s3.<REGION>.amazonaws.com/cloudfront/**.gz',
    'TabSeparated',
    'gzip',
    extra_credentials(role_arn = 'arn:aws:iam::012345678901:role/ClickHouseAccess-foobar')
)
SETTINGS
    mode = 'ordered',
    s3queue_loading_retries = 3,
    -- CloudFront delivers logs up to once per hour (delays up to 24h possible)
    -- so aggressive polling is unnecessary and wastes LIST API calls
    s3queue_polling_min_timeout_ms = 30000,   -- 30 seconds
    s3queue_polling_max_timeout_ms = 300000,  -- 5 minutes
    s3queue_enable_logging_to_s3queue_log = 1,
    -- Skip the two header lines (#Version and #Fields)
    input_format_tsv_skip_first_lines = 2,
    -- Treat '-' as empty/null value
    input_format_tsv_empty_as_default = 1;


-- ============================================================================
-- 3. MATERIALIZED VIEW (Connects Queue to Destination)
-- ============================================================================
-- This view transforms raw string data to proper types and inserts into
-- the destination MergeTree table.
--
-- Key transformations:
--   - Combines date + time into DateTime64 timestamp
--   - Converts numeric strings to proper integer/float types
--   - Handles '-' null markers with defaults
--   - URL-decodes URI components (optional, commented out)

CREATE MATERIALIZED VIEW IF NOT EXISTS cloudfront_logs_mv
TO cloudfront_logs
AS SELECT
    -- Combine date and time into timestamp
    parseDateTime64BestEffortOrZero(concat(date, ' ', time), 3, 'UTC') AS timestamp,

    x_edge_location,

    toUInt64OrZero(sc_bytes) AS sc_bytes,

    c_ip,

    cs_method,

    cs_host,

    -- Optionally URL-decode: decodeURLComponent(cs_uri_stem)
    cs_uri_stem,

    toUInt16OrZero(sc_status) AS sc_status,

    if(cs_referer = '-', '', cs_referer) AS cs_referer,

    if(cs_user_agent = '-', '', cs_user_agent) AS cs_user_agent,

    -- Optionally URL-decode: decodeURLComponent(cs_uri_query)
    if(cs_uri_query = '-', '', cs_uri_query) AS cs_uri_query,

    if(cs_cookie = '-', '', cs_cookie) AS cs_cookie,

    x_edge_result_type,

    x_edge_request_id,

    if(x_host_header = '-', '', x_host_header) AS x_host_header,

    cs_protocol,

    toUInt64OrZero(cs_bytes) AS cs_bytes,

    toFloat64OrZero(time_taken) AS time_taken,

    if(x_forwarded_for = '-', '', x_forwarded_for) AS x_forwarded_for,

    if(ssl_protocol = '-', '', ssl_protocol) AS ssl_protocol,

    if(ssl_cipher = '-', '', ssl_cipher) AS ssl_cipher,

    if(x_edge_response_result_type = '-', '', x_edge_response_result_type) AS x_edge_response_result_type,

    if(cs_protocol_version = '-', '', cs_protocol_version) AS cs_protocol_version,

    if(fle_status = '-', '', fle_status) AS fle_status,

    toUInt16OrZero(fle_encrypted_fields) AS fle_encrypted_fields,

    toUInt16OrZero(c_port) AS c_port,

    toFloat64OrZero(time_to_first_byte) AS time_to_first_byte,

    if(x_edge_detailed_result_type = '-', '', x_edge_detailed_result_type) AS x_edge_detailed_result_type,

    if(sc_content_type = '-', '', sc_content_type) AS sc_content_type,

    toUInt64OrZero(sc_content_len) AS sc_content_len,

    if(sc_range_start = '-', -1, toInt64OrZero(sc_range_start)) AS sc_range_start,

    if(sc_range_end = '-', -1, toInt64OrZero(sc_range_end)) AS sc_range_end

FROM cloudfront_logs_queue;


-- ============================================================================
-- 4. USEFUL QUERIES
-- ============================================================================

-- Check queue processing status
-- SELECT * FROM system.s3queue_log ORDER BY event_time DESC LIMIT 20;

-- Request count by hour
-- SELECT
--     toStartOfHour(timestamp) AS hour,
--     count() AS requests,
--     sum(is_cache_hit) AS cache_hits,
--     round(sum(is_cache_hit) / count() * 100, 2) AS cache_hit_rate
-- FROM cloudfront_logs
-- WHERE date = today()
-- GROUP BY hour
-- ORDER BY hour;

-- Top paths by request count
-- SELECT
--     cs_uri_stem,
--     count() AS requests,
--     round(avg(time_taken) * 1000, 2) AS avg_latency_ms
-- FROM cloudfront_logs
-- WHERE date = today()
-- GROUP BY cs_uri_stem
-- ORDER BY requests DESC
-- LIMIT 20;

-- Error rate by status code
-- SELECT
--     sc_status,
--     count() AS count,
--     round(count() / (SELECT count() FROM cloudfront_logs WHERE date = today()) * 100, 4) AS pct
-- FROM cloudfront_logs
-- WHERE date = today() AND sc_status >= 400
-- GROUP BY sc_status
-- ORDER BY count DESC;

-- Latency percentiles
-- SELECT
--     quantile(0.50)(time_taken) AS p50,
--     quantile(0.90)(time_taken) AS p90,
--     quantile(0.95)(time_taken) AS p95,
--     quantile(0.99)(time_taken) AS p99
-- FROM cloudfront_logs
-- WHERE date = today();

-- Bandwidth by edge location
-- SELECT
--     x_edge_location,
--     formatReadableSize(sum(sc_bytes)) AS bandwidth,
--     count() AS requests
-- FROM cloudfront_logs
-- WHERE date = today()
-- GROUP BY x_edge_location
-- ORDER BY sum(sc_bytes) DESC
-- LIMIT 20;

-- Cache hit ratio by path prefix
-- SELECT
--     arrayElement(splitByChar('/', cs_uri_stem), 2) AS path_prefix,
--     count() AS requests,
--     round(sum(is_cache_hit) / count() * 100, 2) AS cache_hit_rate
-- FROM cloudfront_logs
-- WHERE date = today() AND cs_uri_stem != '/'
-- GROUP BY path_prefix
-- HAVING requests > 100
-- ORDER BY requests DESC;
