-- Batched, idempotent backfill from gateway_request_log into ip_addresses and
-- ip_addresses_statistics. Safe to run repeatedly (cron / manual).
--
-- Examples (tune chunk / batch size for load):
--   SELECT * FROM apply_ip_backfill_gateway_log_chunk(5000);
--   SELECT * FROM reconcile_ip_addresses_statistics_keyset(100);
--
-- If ip_addresses_statistics was already fully populated by the additive DO blocks in
-- add_ipv4_address_and_ip_statistics.sql, starting apply_ip_backfill_gateway_log_chunk with an
-- empty ip_backfill_gateway_log_chunk_applied ledger will double-count. Use only
-- reconcile_ip_addresses_statistics_keyset (after resetting stats if needed), or backfill
-- ip_addresses only by calling the chunk function after truncating statistics and the chunk
-- ledger (destructive — only when rebuilding).
--
-- 1) normalize_gateway_client_ip — same resolution order as add_ipv4_address_and_ip_statistics.sql
-- 2) apply_ip_backfill_gateway_log_chunk — processes the next [start, end) slice of gateway_request_log
--    rows; statistics use additive ON CONFLICT only once per chunk (recorded in ledger).
-- 3) reconcile_ip_addresses_statistics_keyset — optional: replaces stats for up to N ip_addresses rows
--    with totals recomputed from the full gateway_request_log (idempotent; use after migrations
--    or to fix drift). Cursor in ip_backfill_job_state; reset with stats_reconcile_after_ip_id = 0.

CREATE OR REPLACE FUNCTION public.normalize_gateway_client_ip(
    p_ipv4_address text,
    p_ipv4 text,
    p_headers jsonb
)
RETURNS text
LANGUAGE sql
IMMUTABLE
PARALLEL SAFE
AS $$
    SELECT
        NULLIF(
            NULLIF(
                trim(
                    COALESCE(
                        NULLIF(trim(p_ipv4_address), ''),
                        NULLIF(trim(p_ipv4), ''),
                        trim(
                            split_part(
                                COALESCE(
                                    p_headers ->> 'x-forwarded-for',
                                    p_headers ->> 'X-Forwarded-For',
                                    ''
                                ),
                                ',',
                                1
                            )
                        )
                    )
                ),
                ''
            ),
            'NOT_FOUND'
        );
$$;

CREATE TABLE IF NOT EXISTS ip_backfill_gateway_log_chunk_applied (
    chunk_start_inclusive bigint NOT NULL,
    chunk_end_exclusive bigint NOT NULL,
    gateway_log_rows bigint NOT NULL,
    applied_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT ip_backfill_gateway_log_chunk_applied_pkey PRIMARY KEY (chunk_start_inclusive, chunk_end_exclusive)
);

CREATE INDEX IF NOT EXISTS idx_ip_backfill_chunk_applied_at ON ip_backfill_gateway_log_chunk_applied (applied_at DESC);

CREATE TABLE IF NOT EXISTS ip_backfill_job_state (
    singleton smallint NOT NULL DEFAULT 1 PRIMARY KEY,
    stats_reconcile_after_ip_id bigint NOT NULL DEFAULT 0,
    CONSTRAINT ip_backfill_job_state_singleton_chk CHECK (singleton = 1)
);

INSERT INTO ip_backfill_job_state (singleton, stats_reconcile_after_ip_id)
VALUES (1, 0)
ON CONFLICT (singleton) DO NOTHING;

-- Next contiguous chunk of gateway_request_log by primary key; merges ip_addresses and adds
-- per-chunk statistics. Idempotent via ip_backfill_gateway_log_chunk_applied.
CREATE OR REPLACE FUNCTION public.apply_ip_backfill_gateway_log_chunk(p_chunk_size bigint)
RETURNS TABLE (
    chunk_start_inclusive bigint,
    chunk_end_exclusive bigint,
    gateway_log_rows_in_chunk bigint,
    distinct_ips_in_chunk bigint,
    already_applied boolean
)
LANGUAGE plpgsql
AS $$
DECLARE
    v_min_id bigint;
    v_max_id bigint;
    v_start bigint;
    v_end bigint;
    v_rows bigint;
    v_distinct bigint;
BEGIN
    IF p_chunk_size IS NULL OR p_chunk_size < 1 THEN
        RAISE EXCEPTION 'p_chunk_size must be >= 1';
    END IF;

    PERFORM pg_advisory_xact_lock(87261431);

    SELECT
        min(id),
        max(id)
    INTO
        v_min_id,
        v_max_id
    FROM
        gateway_request_log;

    IF v_min_id IS NULL OR v_max_id IS NULL THEN
        chunk_start_inclusive := NULL;
        chunk_end_exclusive := NULL;
        gateway_log_rows_in_chunk := 0;
        distinct_ips_in_chunk := 0;
        already_applied := FALSE;
        RETURN NEXT;
        RETURN;
    END IF;

    SELECT
        COALESCE(
            (
                SELECT
                    max(a.chunk_end_exclusive)
                FROM
                    ip_backfill_gateway_log_chunk_applied a
            ),
            v_min_id
        )
    INTO v_start;

    IF v_start > v_max_id THEN
        chunk_start_inclusive := NULL;
        chunk_end_exclusive := NULL;
        gateway_log_rows_in_chunk := 0;
        distinct_ips_in_chunk := 0;
        already_applied := FALSE;
        RETURN NEXT;
        RETURN;
    END IF;

    v_end := v_start + p_chunk_size;

    IF EXISTS (
        SELECT
            1
        FROM
            ip_backfill_gateway_log_chunk_applied c
        WHERE
            c.chunk_start_inclusive = v_start
            AND c.chunk_end_exclusive = v_end
    ) THEN
        chunk_start_inclusive := v_start;
        chunk_end_exclusive := v_end;
        gateway_log_rows_in_chunk := 0;
        distinct_ips_in_chunk := 0;
        already_applied := TRUE;
        RETURN NEXT;
        RETURN;
    END IF;

    SELECT
        COUNT(*)
    INTO v_rows
    FROM
        gateway_request_log g
    WHERE
        g.id >= v_start
        AND g.id < v_end;

    INSERT INTO ip_addresses (ipv4_address, first_seen_at, last_seen_at)
    SELECT
        v.ip_norm,
        MIN(v.created_at),
        MAX(v.created_at)
    FROM (
        SELECT
            g.created_at,
            public.normalize_gateway_client_ip(g.ipv4_address, g.ipv4, g.headers) AS ip_norm
        FROM
            gateway_request_log g
        WHERE
            g.id >= v_start
            AND g.id < v_end
    ) v
    WHERE
        v.ip_norm IS NOT NULL
    GROUP BY
        v.ip_norm
    ON CONFLICT (ipv4_address) DO UPDATE
    SET
        first_seen_at = LEAST(ip_addresses.first_seen_at, EXCLUDED.first_seen_at),
        last_seen_at = GREATEST(ip_addresses.last_seen_at, EXCLUDED.last_seen_at);

    INSERT INTO ip_addresses_statistics (
        ip_address_id,
        total_requests,
        successful_requests,
        failed_requests,
        total_cached_requests,
        last_request_at,
        updated_at
    )
    SELECT
        a.id,
        COUNT(*)::bigint,
        SUM(
            CASE
                WHEN g.status_code >= 200
                AND g.status_code < 400 THEN 1
                ELSE 0
            END
        )::bigint,
        SUM(
            CASE
                WHEN g.status_code >= 400 THEN 1
                ELSE 0
            END
        )::bigint,
        SUM(
            CASE
                WHEN g.cached IS TRUE THEN 1
                ELSE 0
            END
        )::bigint,
        MAX(g.created_at),
        now()
    FROM
        gateway_request_log g
        INNER JOIN ip_addresses a ON a.ipv4_address = public.normalize_gateway_client_ip(
            g.ipv4_address,
            g.ipv4,
            g.headers
        )
    WHERE
        g.id >= v_start
        AND g.id < v_end
    GROUP BY
        a.id
    ON CONFLICT (ip_address_id) DO UPDATE
    SET
        total_requests = ip_addresses_statistics.total_requests + EXCLUDED.total_requests,
        successful_requests = ip_addresses_statistics.successful_requests
        + EXCLUDED.successful_requests,
        failed_requests = ip_addresses_statistics.failed_requests + EXCLUDED.failed_requests,
        total_cached_requests = ip_addresses_statistics.total_cached_requests
        + EXCLUDED.total_cached_requests,
        last_request_at = GREATEST(
            ip_addresses_statistics.last_request_at,
            EXCLUDED.last_request_at
        ),
        updated_at = now();

    SELECT
        COUNT(DISTINCT v.ip_norm)
    INTO v_distinct
    FROM (
        SELECT
            public.normalize_gateway_client_ip(g.ipv4_address, g.ipv4, g.headers) AS ip_norm
        FROM
            gateway_request_log g
        WHERE
            g.id >= v_start
            AND g.id < v_end
    ) v
    WHERE
        v.ip_norm IS NOT NULL;

    INSERT INTO ip_backfill_gateway_log_chunk_applied (
        chunk_start_inclusive,
        chunk_end_exclusive,
        gateway_log_rows
    )
    VALUES (v_start, v_end, v_rows);

    chunk_start_inclusive := v_start;
    chunk_end_exclusive := v_end;
    gateway_log_rows_in_chunk := v_rows;
    distinct_ips_in_chunk := COALESCE(v_distinct, 0);
    already_applied := FALSE;
    RETURN NEXT;
END;
$$;

-- Recompute statistics from the entire gateway_request_log for up to p_max_ips ip_addresses
-- rows (keyset on ip_addresses.id). Idempotent: repeated runs for the same data yield the same totals.
-- Advance cursor automatically; set ip_backfill_job_state.stats_reconcile_after_ip_id = 0 to rescan all.
CREATE OR REPLACE FUNCTION public.reconcile_ip_addresses_statistics_keyset(p_max_ips integer)
RETURNS TABLE (
    batch_min_ip_id bigint,
    batch_max_ip_id bigint,
    ip_rows_in_batch integer,
    statistics_upserted bigint
)
LANGUAGE plpgsql
AS $$
DECLARE
    v_after bigint;
    v_min_batch bigint;
    v_max_batch bigint;
    v_count integer;
    v_upserted bigint;
BEGIN
    IF p_max_ips IS NULL OR p_max_ips < 1 THEN
        RAISE EXCEPTION 'p_max_ips must be >= 1';
    END IF;

    PERFORM pg_advisory_xact_lock(87261432);

    SELECT
        s.stats_reconcile_after_ip_id
    INTO v_after
    FROM
        ip_backfill_job_state s
    WHERE
        s.singleton = 1
    FOR UPDATE;

    IF v_after IS NULL THEN
        INSERT INTO ip_backfill_job_state (singleton, stats_reconcile_after_ip_id)
        VALUES (1, 0)
        ON CONFLICT (singleton) DO NOTHING;

        SELECT
            s.stats_reconcile_after_ip_id
        INTO v_after
        FROM
            ip_backfill_job_state s
        WHERE
            s.singleton = 1
        FOR UPDATE;
    END IF;

    WITH batch AS (
        SELECT
            ia.id,
            ia.ipv4_address
        FROM
            ip_addresses ia
        WHERE
            ia.id > v_after
        ORDER BY
            ia.id
        LIMIT p_max_ips
    ),
    agg AS (
        SELECT
            b.id AS ip_address_id,
            COUNT(g.id)::bigint AS total_requests,
            SUM(
                CASE
                    WHEN g.status_code >= 200
                    AND g.status_code < 400 THEN 1
                    ELSE 0
                END
            )::bigint AS successful_requests,
            SUM(
                CASE
                    WHEN g.status_code >= 400 THEN 1
                    ELSE 0
                END
            )::bigint AS failed_requests,
            SUM(
                CASE
                    WHEN g.cached IS TRUE THEN 1
                    ELSE 0
                END
            )::bigint AS total_cached_requests,
            MAX(g.created_at) AS last_request_at
        FROM
            batch b
            LEFT JOIN gateway_request_log g ON b.ipv4_address = public.normalize_gateway_client_ip(
                g.ipv4_address,
                g.ipv4,
                g.headers
            )
        GROUP BY
            b.id
    )
    INSERT INTO ip_addresses_statistics (
        ip_address_id,
        total_requests,
        successful_requests,
        failed_requests,
        total_cached_requests,
        last_request_at,
        updated_at
    )
    SELECT
        a.ip_address_id,
        COALESCE(a.total_requests, 0),
        COALESCE(a.successful_requests, 0),
        COALESCE(a.failed_requests, 0),
        COALESCE(a.total_cached_requests, 0),
        a.last_request_at,
        now()
    FROM
        agg a
    ON CONFLICT (ip_address_id) DO UPDATE
    SET
        total_requests = EXCLUDED.total_requests,
        successful_requests = EXCLUDED.successful_requests,
        failed_requests = EXCLUDED.failed_requests,
        total_cached_requests = EXCLUDED.total_cached_requests,
        last_request_at = EXCLUDED.last_request_at,
        updated_at = now();

    GET DIAGNOSTICS v_upserted = ROW_COUNT;

    SELECT
        COUNT(*)::integer,
        MIN(b.id),
        MAX(b.id)
    INTO v_count, v_min_batch, v_max_batch
    FROM (
        SELECT
            ia.id
        FROM
            ip_addresses ia
        WHERE
            ia.id > v_after
        ORDER BY
            ia.id
        LIMIT p_max_ips
    ) b;

    IF v_count > 0 AND v_max_batch IS NOT NULL THEN
        UPDATE
            ip_backfill_job_state s
        SET
            stats_reconcile_after_ip_id = v_max_batch
        WHERE
            s.singleton = 1;
    END IF;

    batch_min_ip_id := v_min_batch;
    batch_max_ip_id := v_max_batch;
    ip_rows_in_batch := COALESCE(v_count, 0);
    statistics_upserted := COALESCE(v_upserted, 0);
    RETURN NEXT;
END;
$$;
