WITH abc AS (
    SELECT 
        DISTINCT 
        app.application_key,
        CAST(app.application_timestamp AS TIMESTAMP) AS application_timestamp,
        LOWER(app.application_email_address) AS application_email_address,
        app.application_is_rejected,
        NULLIF(app.application_ssn, '') AS application_ssn, 
        NULLIF(app.application_bank_account_number, '') AS application_bank_account_number,
        loans.originated_loan_is_fraud,
        loans.originated_loan_is_never_paid
    FROM public.applications AS app
    LEFT JOIN public.bridge_funding_lifecycle AS master
        ON app.application_key = master.application_key
    LEFT JOIN public.originated_loan AS loans
        ON master.originated_loan_key = loans.originated_loan_key
        -- WHERE app.application_customer_type = 'Direct New'
),

current_app AS (
    SELECT
        application_key AS current_app_key,
        application_timestamp AS current_app_created_at,
        $hd_score_m1 AS hd_score_m1
    FROM abc
    WHERE application_key = $application_key  
),

connected_apps AS (
    SELECT
        a.*,
        a.application_key AS connected_app_key
    FROM abc a
    WHERE application_key = ANY(ARRAY$connected_application_keys::text[])
),

clusters_alt AS (
    SELECT
        a.*,
        b.current_app_key,
        b.current_app_created_at,
        b.hd_score_m1,
        CASE
            WHEN LOWER(CAST(a.originated_loan_is_fraud AS TEXT)) != 'true'
                AND a.originated_loan_is_never_paid IS NULL THEN NULL
            WHEN ABS(EXTRACT(DAY FROM AGE(b.current_app_created_at, a.application_timestamp))) > 29
                AND (LOWER(CAST(a.originated_loan_is_fraud AS TEXT)) = 'true'
                    OR LOWER(CAST(a.originated_loan_is_never_paid AS TEXT)) = 'true')
                THEN 1
            ELSE 0
        END AS target_connected_30,
            CONCAT(a.application_ssn, '_', a.application_bank_account_number) AS user_connected_v2,
        a.application_timestamp AS connected_date,
        a.application_email_address AS connected_email,
        a.application_is_rejected AS application_is_rejected_connected
    FROM connected_apps a
    CROSS JOIN current_app b
),

full_apps_past AS (
    SELECT
        current_app_key,
        SUM(target_connected_30) AS target_connected_30_sum,
        COUNT(DISTINCT CASE
            WHEN LOWER(CAST(application_is_rejected_connected AS TEXT)) = 'true'
            THEN connected_app_key
            ELSE NULL
        END) AS rejected_app_count
    FROM clusters_alt
    WHERE current_app_key != connected_app_key
    GROUP BY current_app_key
),

clusters AS (
    SELECT
        a.current_app_key AS application_key,
        CAST(MAX(hd_score_m1) AS FLOAT) AS hd_score_m1,
        COUNT(DISTINCT a.user_connected_v2) AS cluster_size_users_v2,
        MAX(b.target_connected_30_sum) AS target_connected_30_sum,
        COUNT(DISTINCT a.connected_email) AS email_cnt,
        MAX(b.rejected_app_count) AS rejected_app_count,
        COUNT(DISTINCT EXTRACT(DAY FROM DATE(a.connected_date))) AS app_dt_day_cnt,
        COUNT(*) AS cluster_size
    FROM clusters_alt a
    LEFT JOIN full_apps_past b
        ON a.current_app_key = b.current_app_key
    GROUP BY 1
)

SELECT
	COALESCE(MAX(hd_score_m1), $hd_score_m1)     AS hd_score_m1,
	COALESCE(MAX(cluster_size_users_v2), 0)      AS cluster_size_users_v2,
	COALESCE(MAX(target_connected_30_sum), 0)    AS target_connected_30_sum,
	COALESCE(MAX(email_cnt), 0)                  AS email_cnt,
	COALESCE(MAX(rejected_app_count), 0)         AS rejected_app_count,
	COALESCE(MAX(app_dt_day_cnt), 0)             AS app_dt_day_cnt,
	COALESCE(MAX(cluster_size), 0)               AS cluster_size
FROM clusters;