def get_cutoff_date():
    # get current time in utc
    ts = datetime.now(timezone.utc)
    # round down to 00:00 UTC of the current day
    ts = ts.replace(hour=0, minute=0, second=0, microsecond=0)
    # subtract 1 day
    ts = ts - timedelta(days=1)
    
    return ts

def get_gold_calculation_dates(**kwargs):
    # Cutoff Date
    cutoff_date = get_cutoff_date()
    cutoff_date_str = cutoff_date.strftime('%Y-%m-%d %H:%M:%S')

    # Backfill Date
    dag_run = kwargs.get('dag_run')
    if dag_run is not None:
        start_date_str = dag_run.conf.get('start_date', '1970-01-01 00:00:00')
        end_date_str = dag_run.conf.get('end_date', cutoff_date_str)
        is_backfill = str.lower(dag_run.conf.get('is_backfill', 'false'))
    else:
        raise ValueError('Dag run is None')
    
    # Sanitize the start and end dates
    start_datetime = datetime.strptime(start_date_str, '%Y-%m-%d %H:%M:%S')
    end_datetime = datetime.strptime(end_date_str, '%Y-%m-%d %H:%M:%S')
    cutoff_datetime = datetime.strptime(cutoff_date_str, '%Y-%m-%d %H:%M:%S')

    if start_datetime >= end_datetime:
        raise ValueError('Start date must be before end date')
    
    if end_datetime > cutoff_datetime:
        raise ValueError('End date must be less than or equal to cutoff date')

    # Push to XCom
    kwargs['ti'].xcom_push(key='cutoff_date', value=end_date_str)
    kwargs['ti'].xcom_push(key='rewards_start', value=start_date_str)
    kwargs['ti'].xcom_push(key='is_backfill', value=is_backfill)

SELECT
  lower(coalesce(t.output_data ->> 'depositor', t.output_data ->> 'staker')) as staker,
  lower(t.output_data ->> 'strategy') as strategy,
  (t.output_data ->> 'shares')::numeric(78,0) as shares,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs as t
LEFT JOIN blocks as b ON (t.block_sequence_id = b.id)
WHERE t.address = '{{ var('strategy_manager_address') }}'
AND t.event_name = 'Deposit'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(t.arguments #>> '{0,Value}') AS staker,
  (t.output_data ->> 'sharesDelta')::numeric(78,0) as shares,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('eigen_pod_manager_address') }}'
AND t.event_name = 'PodSharesUpdated'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(coalesce(t.output_data ->> 'depositor', t.output_data ->> 'staker')) as staker,
  lower(t.output_data ->> 'strategy') as strategy,
  (t.output_data ->> 'shares')::numeric(78,0) as shares,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('strategy_manager_address') }}'
AND t.event_name = 'ShareWithdrawalQueued'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'
-- Remove this transaction hash as it is the only withdrawal on m1 that was completed as shares. There is no corresponding deposit event. The withdrawal was completed to the same staker address.
AND t.transaction_hash != '0x62eb0d0865b2636c74ed146e2d161e39e42b09bac7f86b8905fc7a830935dc1e'

WITH migrations AS (
  SELECT 
    (
      SELECT lower(string_agg(lpad(to_hex(elem::int), 2, '0'), ''))
      FROM jsonb_array_elements_text(t.output_data->'oldWithdrawalRoot') AS elem
    ) AS m1_withdrawal_root,
    (
      SELECT lower(string_agg(lpad(to_hex(elem::int), 2, '0'), ''))
      FROM jsonb_array_elements_text(t.output_data->'newWithdrawalRoot') AS elem
    ) AS m2_withdrawal_root
  FROM transaction_logs t
  WHERE t.address = '{{ var('delegation_manager_address') }}'
  AND t.event_name = 'WithdrawalMigrated'
),
full_m2_withdrawals AS (
  SELECT
    lower(t.output_data #>> '{withdrawal}') as withdrawals,
    (
      SELECT lower(string_agg(lpad(to_hex(elem::int), 2, '0'), ''))
      FROM jsonb_array_elements_text(t.output_data ->'withdrawalRoot') AS elem
    ) AS withdrawal_root,
    lower(t.output_data #>> '{withdrawal, staker}') AS staker,
    lower(t_strategy.strategy) AS strategy,
    (t_share.share)::numeric(78,0) AS shares,
    t_strategy.strategy_index,
    t_share.share_index,
    t.transaction_hash,
    t.log_index,
    b.block_time::timestamp(6),
    to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
    t.block_number
  FROM transaction_logs t
  LEFT JOIN blocks b ON t.block_sequence_id = b.id,
  jsonb_array_elements_text(t.output_data #> '{withdrawal, strategies}') WITH ORDINALITY AS t_strategy(strategy, strategy_index),
  jsonb_array_elements_text(t.output_data #> '{withdrawal, shares}') WITH ORDINALITY AS t_share(share, share_index)
  WHERE t.address = '{{ var('delegation_manager_address') }}'
  AND t.event_name = 'WithdrawalQueued'
  AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'
  AND t_strategy.strategy_index = t_share.share_index
)
-- Parse out the m2 withdrawals that were migrated from m1
SELECT 
  full_m2_withdrawals.*
FROM 
  full_m2_withdrawals
LEFT JOIN 
  migrations 
ON 
  full_m2_withdrawals.withdrawal_root = migrations.m2_withdrawal_root
WHERE 
  migrations.m2_withdrawal_root IS NULL

SELECT
  lower(t.arguments #>> '{0,Value}') as operator,
  lower(t.output_data ->> 'strategy') as strategy,
  (t.output_data ->> 'shares')::numeric(78,0) as shares,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('delegation_manager_address') }}'
AND t.event_name = 'OperatorSharesIncreased'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(t.arguments #>> '{0,Value}') as operator,
  lower(t.output_data ->> 'strategy') as strategy,
  (t.output_data ->> 'shares')::numeric(78,0) as shares,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('delegation_manager_address') }}'
AND t.event_name = 'OperatorSharesDecreased'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(t.arguments #>> '{0,Value}') AS staker,
  lower(t.arguments #>> '{1,Value}') AS operator,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('delegation_manager_address') }}'
AND t.event_name = 'StakerDelegated'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(t.arguments #>> '{0,Value}') AS staker,
  lower(t.arguments #>> '{1,Value}') AS operator,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('delegation_manager_address') }}'
AND t.event_name = 'StakerUndelegated'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
    lower(tl.arguments #>> '{0,Value}') AS avs,
    lower(tl.arguments #>> '{2,Value}') AS reward_hash,
    coalesce(lower(tl.output_data #>> '{rewardsSubmission}'), lower(tl.output_data #>> '{rangePayment}')) as rewards_submission,
    coalesce(lower(tl.output_data #>> '{rewardsSubmission, token}'), lower(tl.output_data #>> '{rangePayment, token}')) as token,
    coalesce(tl.output_data #>> '{rewardsSubmission,amount}', tl.output_data #>> '{rangePayment,amount}')::numeric(78,0) as amount,
    to_timestamp(coalesce(tl.output_data #>> '{rewardsSubmission,startTimestamp}', tl.output_data #>> '{rangePayment,startTimestamp}')::bigint)::timestamp(6) as start_timestamp,
    coalesce(tl.output_data #>> '{rewardsSubmission,duration}', tl.output_data #>> '{rangePayment,duration}')::bigint as duration,
    to_timestamp(
            coalesce(tl.output_data #>> '{rewardsSubmission,startTimestamp}', tl.output_data #>> '{rangePayment,startTimestamp}')::bigint
                + coalesce(tl.output_data #>> '{rewardsSubmission,duration}', tl.output_data #>> '{rangePayment,duration}')::bigint
    )::timestamp(6) as end_timestamp,
    lower(t.entry ->> 'strategy') as strategy,
    (t.entry ->> 'multiplier')::numeric(78,0) as multiplier,
    t.strategy_index as strategy_index,
    tl.transaction_hash,
    tl.log_index,
    b.block_time::timestamp(6),
    to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
    tl.block_number
FROM transaction_logs tl
    LEFT JOIN blocks b ON (tl.block_sequence_id = b.id)
    CROSS JOIN LATERAL jsonb_array_elements(
        coalesce(tl.output_data #> '{rewardsSubmission,strategiesAndMultipliers}',tl.output_data #> '{rangePayment,strategiesAndMultipliers}')
    ) WITH ORDINALITY AS t(entry, strategy_index)
WHERE address = '{{ var('rewards_coordinator_address') }}'
AND event_name = 'AVSRewardsSubmissionCreated'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(tl.arguments #>> '{0,Value}') AS avs, -- Keeping as AVS for compatibility with unioning on range_payments.
  lower(tl.arguments #>> '{2,Value}') AS reward_hash,
  coalesce(lower(tl.output_data #>> '{rewardsSubmission}'), lower(tl.output_data #>> '{rangePayment}')) as rewards_submission,
  coalesce(lower(tl.output_data #>> '{rewardsSubmission, token}'), lower(tl.output_data #>> '{rangePayment, token}')) as token,
  coalesce(tl.output_data #>> '{rewardsSubmission,amount}', tl.output_data #>> '{rangePayment,amount}')::numeric(78,0) as amount,
  to_timestamp(coalesce(tl.output_data #>> '{rewardsSubmission,startTimestamp}', tl.output_data #>> '{rangePayment,startTimestamp}')::bigint)::timestamp(6) as start_timestamp,
  coalesce(tl.output_data #>> '{rewardsSubmission,duration}', tl.output_data #>> '{rangePayment,duration}')::bigint as duration,
  to_timestamp(
            coalesce(tl.output_data #>> '{rewardsSubmission,startTimestamp}', tl.output_data #>> '{rangePayment,startTimestamp}')::bigint
                + coalesce(tl.output_data #>> '{rewardsSubmission,duration}', tl.output_data #>> '{rangePayment,duration}')::bigint
    )::timestamp(6) as end_timestamp,
  lower(t.entry ->> 'strategy') as strategy,
  (t.entry ->> 'multiplier')::numeric(78,0) as multiplier,
  t.strategy_index as strategy_index,
  tl.transaction_hash,
  tl.log_index,
  b.block_time::timestamp(6),
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  tl.block_number
FROM transaction_logs tl
LEFT JOIN blocks b ON tl.block_sequence_id = b.id
CROSS JOIN LATERAL jsonb_array_elements(
    coalesce(tl.output_data #> '{rewardsSubmission,strategiesAndMultipliers}',tl.output_data #> '{rangePayment,strategiesAndMultipliers}')
) WITH ORDINALITY AS t(entry, strategy_index)
WHERE address = '{{ var('rewards_coordinator_address') }}'
AND event_name = 'RewardsSubmissionForAllCreated'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

SELECT
  lower(t.arguments #>> '{0,Value}') as operator,
  lower(t.arguments #>> '{1,Value}') as avs,
  (t.output_data -> 'status')::int as status,
  t.transaction_hash,
  t.log_index,
  b.block_time,
  to_char(b.block_time, 'YYYY-MM-DD') AS block_date,
  t.block_number
FROM transaction_logs t
LEFT JOIN blocks b ON t.block_sequence_id = b.id
WHERE t.address = '{{ var('avs_directory_address') }}'
AND t.event_name = 'OperatorAVSRegistrationStatusUpdated'
AND date_trunc('day', b.block_time) < TIMESTAMP '{{ var("cutoff_date") }}'

GENESIS_TIMESTAMP---------------------Day1---------------------Day2
                     ^       ^
                  A=100    B=200

GENESIS_TIMESTAMP---------------------Day1---------------------Day2
                     ^                       ^
                  Register                Deregister

--------------Day1--------------Day2--------------Day3--------------Day4
          ^                                                       ^
        Register                                             Deregister

SUM(shares) OVER (PARTITION BY staker, strategy ORDER BY block_time, log_index) AS shares, 

SELECT
    staker,
    strategy,
    -- Sum each share amount over the window to get total shares for each (staker, strategy) at every timestamp update */
    SUM(shares) OVER (PARTITION BY staker, strategy ORDER BY block_time, log_index) AS shares, 
    transaction_hash,
    log_index,
    strategy_index,
    block_time,
    block_date,
    block_number
FROM (
    SELECT staker, strategy, shares, 0 as strategy_index, transaction_hash, log_index, block_time, block_date, block_number
    FROM {{ ref('staker_deposits') }}

    UNION ALL

    -- Subtract m1 & m2 withdrawals
    SELECT staker, strategy, shares * -1, 0 as strategy_index, transaction_hash, log_index, block_time, block_date, block_number
    FROM {{ ref('m1_staker_withdrawals') }}

    UNION ALL

    SELECT staker, strategy, shares * -1, strategy_index, transaction_hash, log_index, block_time, block_date, block_number
    FROM {{ ref('m2_staker_withdrawals') }}

    UNION all

    -- Shares in eigenpod are positive or negative, so no need to multiply by -1
    SELECT staker, '0xbeac0eeeeeeeeeeeeeeeeeeeeeeeeeeeeeebeac0' as strategy, shares, 0 as strategy_index, transaction_hash, log_index, block_time, block_date, block_number
    FROM {{ ref('eigenpod_shares') }}
) combined_staker_shares

SELECT
  operator,
  strategy,
  -- Sum each share amount over the window to get total shares for each (operator, strategy) at every timestamp update */
  SUM(shares) OVER (PARTITION BY operator, strategy ORDER BY block_time, log_index) AS shares,
  transaction_hash,
  log_index,
  block_time,
  block_date,
  block_number
FROM (
    SELECT operator, strategy, shares, transaction_hash, log_index, block_time, block_date, block_number
    FROM {{ ref('operator_share_increases') }}

    UNION ALL

    SELECT operator, strategy, shares * -1 AS shares, transaction_hash, log_index, block_time, block_date, block_number
    FROM {{ ref('operator_share_decreases') }}
) combined_shares

SELECT 
  staker,
  CASE when src = 'undelegations' THEN '0x0000000000000000000000000000000000000000' ELSE operator END AS operator,
  transaction_hash,
  log_index,
  block_time,
  block_date,
  block_number
FROM (
    SELECT *, 'undelegations' AS src FROM {{ ref('staker_undelegations') }}
    UNION ALL
    SELECT *, 'delegations' AS src FROM {{ ref('staker_delegations') }}
) as delegations_combined

SELECT *, false as reward_for_all from {{ ref('avs_reward_submissions') }}

UNION ALL

SELECT *, true as reward_for_all from {{ ref('reward_submission_for_all') }}

SELECT
    operator,
    avs,
    CASE WHEN status = 1 then true ELSE false END AS registered,
    transaction_hash,
    log_index,
    block_date,
    block_time,
    block_number
FROM {{ ref('operator_avs_registrations') }}

WITH ranked_staker_records as (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY staker, strategy, cast(block_time AS DATE) ORDER BY block_time DESC, log_index DESC) AS rn
    FROM {{ ref('staker_shares') }}
),

 snapshotted_records as (
     SELECT
         staker,
         strategy,
         shares,
         block_time,
         date_trunc('day', block_time) + INTERVAL '1' day AS snapshot_time
     from ranked_staker_records
     where rn = 1
 ),

staker_share_windows as (
     SELECT
     staker, strategy, shares, snapshot_time as start_time,
     CASE
         -- If the range does not have the end, use the current timestamp truncated to 0 UTC
         WHEN LEAD(snapshot_time) OVER (PARTITION BY staker, strategy ORDER BY snapshot_time) is null THEN date_trunc('day', TIMESTAMP '{{ var("cutoff_date") }}')
         ELSE LEAD(snapshot_time) OVER (PARTITION BY staker, strategy ORDER BY snapshot_time)
         END AS end_time
     FROM snapshotted_records
 )
SELECT * from staker_share_windows

WITH ranked_operator_records as (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY operator, strategy, cast(block_time AS DATE) ORDER BY block_time DESC, log_index DESC) AS rn
    FROM {{ ref('operator_shares') }}
),
-- Get the latest record for each day & round up to the snapshot day
snapshotted_records as (
 SELECT
     operator,
     strategy,
     shares,
     block_time,
     date_trunc('day', block_time) + INTERVAL '1' day as snapshot_time
 from ranked_operator_records
 where rn = 1
),
-- Get the range for each operator, strategy pairing
operator_share_windows as (
 SELECT
     operator, strategy, shares, snapshot_time as start_time,
     CASE
         -- If the range does not have the end, use the current timestamp truncated to 0 UTC
         WHEN LEAD(snapshot_time) OVER (PARTITION BY operator, strategy ORDER BY snapshot_time) is null THEN date_trunc('day', TIMESTAMP '{{ var("cutoff_date") }}')
         ELSE LEAD(snapshot_time) OVER (PARTITION BY operator, strategy ORDER BY snapshot_time)
         END AS end_time
 FROM snapshotted_records
)
SELECT * from operator_share_windows

with ranked_delegations as (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY staker, cast(block_time AS DATE) ORDER BY block_time DESC, log_index DESC) AS rn
    FROM {{ ref('staker_delegation_status') }}
),
-- Get the latest record for each day & round up to the snapshot day
 snapshotted_records as (
     SELECT
         staker,
         operator,
         block_time,
         date_trunc('day', block_time) + INTERVAL '1' day AS snapshot_time
     from ranked_delegations
     where rn = 1
 ),
-- Get the range for each staker
 staker_delegation_windows as (
     SELECT
         staker, operator, snapshot_time as start_time,
         CASE
             -- If the range does not have the end, use the cutoff date truncated to 0 UTC
             WHEN LEAD(snapshot_time) OVER (PARTITION BY staker ORDER BY snapshot_time) is null THEN date_trunc('day', TIMESTAMP '{{ var("cutoff_date") }}')
             ELSE LEAD(snapshot_time) OVER (PARTITION BY staker ORDER BY snapshot_time)
             END AS end_time
     FROM snapshotted_records
 )
SELECT * from staker_delegation_windows

WITH marked_statuses AS (
    SELECT
        operator,
        avs,
        registered,
        block_time,
        block_date,
        -- Mark the next action as next_block_time
        LEAD(block_time) OVER (PARTITION BY operator, avs ORDER BY block_time ASC, log_index ASC) AS next_block_time,
        -- The below lead/lag combinations are only used in the next CTE
        -- Get the next row's registered status and block_date
        LEAD(registered) OVER (PARTITION BY operator, avs ORDER BY block_time ASC, log_index ASC) AS next_registration_status,
        LEAD(block_date) OVER (PARTITION BY operator, avs ORDER BY block_time ASC, log_index ASC) AS next_block_date,
        -- Get the previous row's registered status and block_date
        LAG(registered) OVER (PARTITION BY operator, avs ORDER BY block_time ASC, log_index ASC) AS prev_registered,
        LAG(block_date) OVER (PARTITION BY operator, avs ORDER BY block_time ASC, log_index ASC) AS prev_block_date
    FROM {{ ref('operator_avs_status') }}
),

 removed_same_day_deregistrations AS (
     SELECT * from marked_statuses
     WHERE NOT (
         -- Remove the registration part
         (registered = TRUE AND
          COALESCE(next_registration_status = FALSE, false) AND -- default to false if null
          COALESCE(block_date = next_block_date, false)) OR
             -- Remove the deregistration part
         (registered = FALSE AND
          COALESCE(prev_registered = TRUE, false) and
          COALESCE(block_date = prev_block_date, false)
             )
         )
 ),

 registration_periods AS (
     SELECT
         operator,
         avs,
         block_time AS start_time,
         -- Mark the next_block_time as the end_time for the range
         -- Use coalesce because if the next_block_time for a registration is not closed, then we use cutoff_date
         COALESCE(next_block_time, TIMESTAMP '{{ var("cutoff_date") }}') AS end_time,
         registered
     FROM removed_same_day_deregistrations
     WHERE registered = TRUE
 ),

 registration_windows_extra as (
     SELECT
         operator,
         avs,
         date_trunc('day', start_time) + interval '1' day as start_time,
         -- End time is end time non inclusive becuase the operator is not registered on the AVS at the end time OR it is current timestamp rounded up
         date_trunc('day', end_time) as end_time
     FROM registration_periods
 ),

 operator_avs_registration_windows as (
     SELECT * from registration_windows_extra
     WHERE start_time != end_time
 )
select * from operator_avs_registration_windows

with ranked_records AS (
    SELECT
        lower(operator) as operator,
        lower(avs) as avs,
        lower(strategy) as strategy,
        block_time,
        date_trunc('day', CAST(block_time as timestamp(6))) + interval '1' day as start_time,
        ROW_NUMBER() OVER (
            PARTITION BY operator, avs, strategy, date_trunc('day', CAST(block_time as timestamp(6))) + interval '1' day
            ORDER BY block_time DESC -- want latest records to be ranked highest
            ) AS rn
    -- Cannot use ref here because this table is not generated via DBT
    FROM public.operator_restaked_strategies
    -- testnet and holesky all exist together in the blocklake so the avs_directory_address allows us to filter
    WHERE avs_directory_address = lower('{{ var('avs_directory_address') }}')
),

 latest_records AS (
     SELECT
         operator,
         avs,
         strategy,
         start_time,
         block_time
     FROM ranked_records
     WHERE rn = 1
 ),

 grouped_records AS (
     SELECT
         operator,
         avs,
         strategy,
         start_time,
         LEAD(start_time) OVER (
             PARTITION BY operator, avs, strategy
             ORDER BY start_time ASC
             ) AS next_start_time
     FROM latest_records
 ),

 parsed_ranges AS (
     SELECT
         operator,
         avs,
         strategy,
         start_time,
         -- If the next_start_time is not on the consecutive day, close off the end_time
         CASE
             WHEN next_start_time IS NULL OR next_start_time > start_time + INTERVAL '1' DAY THEN start_time
             ELSE next_start_time
             END AS end_time
     FROM grouped_records
 ),

 active_windows as (
     SELECT *
     FROM parsed_ranges
     WHERE start_time != end_time
 ),

 gaps_and_islands AS (
     SELECT
         operator,
         avs,
         strategy,
         start_time,
         end_time,
         LAG(end_time) OVER(PARTITION BY operator, avs, strategy ORDER BY start_time) as prev_end_time
     FROM active_windows
 ),

 island_detection AS (
     SELECT operator, avs, strategy, start_time, end_time, prev_end_time,
            CASE
                -- If the previous end time is equal to the start time, then mark as part of the island, else create a new island
                WHEN prev_end_time = start_time THEN 0
                ELSE 1
                END as new_island
     FROM gaps_and_islands
 ),

 island_groups AS (
     SELECT
         operator,
         avs,
         strategy,
         start_time,
         end_time,
         SUM(new_island) OVER (
             PARTITION BY operator, avs, strategy ORDER BY start_time
             ) AS island_id
     FROM island_detection
 ),

 operator_avs_strategy_windows AS (
     SELECT
         operator,
         avs,
         strategy,
         MIN(start_time) AS start_time,
         MAX(end_time) AS end_time
     FROM island_groups
     GROUP BY operator, avs, strategy, island_id
     ORDER BY operator, avs, strategy, start_time
 )
select * from operator_avs_strategy_windows

WITH cleaned_records as (
  SELECT * FROM {{ ref('staker_share_windows')}}
  WHERE start_time < end_time
)
SELECT
    staker,
    strategy,
    shares,
    cast(day AS DATE) AS snapshot
FROM
    cleaned_records
CROSS JOIN
    generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS day

WITH cleaned_records as (
    SELECT * FROM {{ ref('operator_share_windows')}}
        WHERE start_time < end_time
)
SELECT
    operator,
    strategy,
    shares,
    cast(day AS DATE) AS snapshot
FROM
    cleaned_records
        CROSS JOIN
    generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS day

WITH cleaned_records as (
    SELECT * FROM {{ ref('staker_delegation_windows') }}
        WHERE start_time < end_time
)
SELECT
    staker,
    operator,
    cast(day AS DATE) AS snapshot
FROM
    cleaned_records
        CROSS JOIN
    generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS day

WITH cleaned_records AS (
    SELECT * FROM {{ ref('operator_avs_strategy_windows') }}
        WHERE start_time < end_time
)
SELECT
    operator,
    avs,
    strategy,
    cast(day AS DATE) AS snapshot
FROM
    cleaned_records
        CROSS JOIN
    generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS day

WITH cleaned_records AS (
    SELECT * FROM {{ ref('operator_avs_registration_windows') }}
        WHERE start_time < end_time
)
SELECT
    operator,
    avs,
    day AS snapshot
FROM cleaned_records
CROSS JOIN generate_series(DATE(start_time), DATE(end_time) - interval '1' day, interval '1' day) AS day

Day0------Day1------Day2------Day3------Day4------Day5------Day6------Day7

Day1---------------------Day2
      ^               ^
      A               B

Day1---------------------Day2
      ^               ^
      J               K

--------Day1---------------------Day2---------------------Day3
      ^                                                 ^
     Entry                                             Exit

WITH active_rewards_modified as (
    SELECT *,
           amount/(duration/86400) as tokens_per_day,
           cast('{{ var("cutoff_date") }}' AS TIMESTAMP(6)) as global_end_inclusive -- Inclusive means we DO USE this day as a snapshot
    FROM {{ ref('rewards_combined') }}
        WHERE end_timestamp >= TIMESTAMP '{{ var("rewards_start") }}' and start_timestamp <= TIMESTAMP '{{ var("cutoff_date") }}'
),

 active_rewards_updated_end_timestamps as (
     SELECT
         avs,
         /**
          * Cut the start and end windows to handle
          * A. Retroactive rewards that came recently whose start date is less than start_timestamp
          * B. Don't make any rewards past end_timestamp for this run
          */
         start_timestamp as reward_start_exclusive,
         LEAST(global_end_inclusive, end_timestamp) as reward_end_inclusive,
         tokens_per_day,
         token,
         multiplier,
         strategy,
         reward_hash,
         reward_for_all,
         global_end_inclusive
     FROM active_rewards_modified
),

-- For each reward hash, find the latest snapshot
 active_rewards_updated_start_timestamps as (
     SELECT
         ap.avs,
         CASE
             WHEN '{{ var("is_backfill") }}' = 'true' THEN ap.reward_start_exclusive
             ELSE COALESCE(MAX(g.snapshot), ap.reward_start_exclusive)
             END as reward_start_exclusive,
         ap.reward_end_inclusive,
         ap.token,
         ap.tokens_per_day as tokens_per_day_decimal,
         -- Round down to 15 sigfigs for double precision, ensuring know errouneous round up or down
         ap.tokens_per_day * ((POW(10, 15) - 1)/(POW(10, 15))) as tokens_per_day,
         ap.multiplier,
         ap.strategy,
         ap.reward_hash,
         ap.reward_for_all,
         ap.global_end_inclusive
     FROM active_rewards_updated_end_timestamps ap
              LEFT JOIN {{ var('schema_name') }}.gold_table g 
                        ON g.reward_hash = ap.reward_hash
     GROUP BY ap.avs, ap.reward_end_inclusive, ap.token, ap.tokens_per_day, ap.multiplier, ap.strategy, ap.reward_hash, ap.global_end_inclusive, ap.reward_start_exclusive, ap.reward_for_all
 ),

 active_reward_ranges AS (
     SELECT * from active_rewards_updated_start_timestamps
     /** Take out (reward_start_exclusive, reward_end_inclusive) windows where
      * 1. reward_start_exclusive >= reward_end_inclusive: The reward period is done or we will handle on a subsequent run
     */
     WHERE reward_start_exclusive < reward_end_inclusive
 ),

 exploded_active_range_rewards AS (
     SELECT * FROM active_reward_ranges
     CROSS JOIN generate_series(DATE(reward_start_exclusive), DATE(reward_end_inclusive), INTERVAL '1' DAY) AS day
 ),

 active_rewards_final AS (
     SELECT
         avs,
         cast(day as DATE) as snapshot,
         token,
         tokens_per_day,
         multiplier,
         strategy,
         reward_hash,
         reward_for_all
     FROM exploded_active_range_rewards
     -- Remove snapshots on the start day
     WHERE day != reward_start_exclusive
 )
select * from active_rewards_final

WITH reward_snapshot_operators as (
  SELECT 
    ap.reward_hash,
    ap.snapshot,
    ap.token,
    ap.tokens_per_day,
    ap.avs,
    ap.strategy,
    ap.multiplier,
    ap.reward_for_all,
    oar.operator
  FROM {{ ref('active_rewards') }} ap  
  JOIN {{ ref('operator_avs_registration_snapshots') }} oar
  ON ap.avs = oar.avs and ap.snapshot = oar.snapshot
  WHERE ap.reward_for_all = false
),

_operator_restaked_strategies AS (
  SELECT
    rso.*
  FROM reward_snapshot_operators rso
  JOIN {{ ref('operator_avs_strategy_snapshots') }} oas
  ON
    rso.operator = oas.operator AND
    rso.avs = oas.avs AND
    rso.strategy = oas.strategy AND
    rso.snapshot = oas.snapshot
),

staker_delegated_operators AS (
  SELECT
    ors.*,
    sds.staker
  FROM _operator_restaked_strategies ors
  JOIN {{ ref('staker_delegation_snapshots') }} sds
  ON
    ors.operator = sds.operator AND
    ors.snapshot = sds.snapshot
),

staker_avs_strategy_shares AS (
  SELECT
    sdo.*,
    sss.shares
  FROM staker_delegated_operators sdo
  JOIN {{ ref('staker_share_snapshots') }} sss
  ON
    sdo.staker = sss.staker AND
    sdo.snapshot = sss.snapshot AND
    sdo.strategy = sss.strategy
  -- Parse out negative shares and zero multiplier so there is no division by zero case
  WHERE sss.shares > 0 and sdo.multiplier != 0
),

staker_weights AS (
  SELECT *,
    SUM(multiplier * shares) OVER (PARTITION BY staker, reward_hash, snapshot) AS staker_weight
  FROM staker_avs_strategy_shares
),

distinct_stakers AS (
  SELECT *
  FROM (
      SELECT *,
        -- We can use an arbitrary order here since the staker_weight is the same for each (staker, strategy, hash, snapshot)
        -- We use strategy ASC for better debuggability
        ROW_NUMBER() OVER (PARTITION BY reward_hash, snapshot, staker ORDER BY strategy ASC) as rn
      FROM staker_weights
  ) t
  WHERE rn = 1
  ORDER BY reward_hash, snapshot, staker
),

staker_weight_sum AS (
  SELECT *,
    SUM(staker_weight) OVER (PARTITION BY reward_hash, snapshot) as total_weight
  FROM distinct_stakers
),

staker_proportion AS (
  SELECT *,
    FLOOR((staker_weight / total_weight) * 1000000000000000) / 1000000000000000 AS staker_proportion
  FROM staker_weight_sum
),

staker_operator_total_tokens AS (
  SELECT *,
    (staker_proportion * tokens_per_day)::text::decimal(38,0) as total_staker_operator_payout
  FROM staker_proportion
),

token_breakdowns AS (
  SELECT *,
    (total_staker_operator_payout * 0.10)::text::decimal(38,0) as operator_tokens,
    total_staker_operator_payout - ((total_staker_operator_payout * 0.10)::text::decimal(38,0)) as staker_tokens
  FROM staker_operator_total_tokens
)
SELECT * from token_breakdowns
ORDER BY reward_hash, snapshot, staker, operator

WITH operator_token_sums AS (
  SELECT
    reward_hash,
    snapshot,
    token,
    tokens_per_day,
    avs,
    strategy,
    multiplier,
    reward_for_all,
    operator,
    SUM(operator_tokens) OVER (PARTITION BY operator, reward_hash, snapshot) AS operator_tokens
  FROM {{ ref('2_staker_reward_amounts') }}
),

distinct_operators AS (
  SELECT *
  FROM (
      SELECT *,
        -- We can use an arbitrary order here since the staker_weight is the same for each (operator, strategy, hash, snapshot)
        -- We use strategy ASC for better debuggability
        ROW_NUMBER() OVER (PARTITION BY reward_hash, snapshot, operator ORDER BY strategy ASC) as rn
      FROM operator_token_sums
  ) t
  WHERE rn = 1
)
SELECT * FROM distinct_operators

WITH reward_snapshot_stakers AS (
  SELECT
    ap.reward_hash,
    ap.snapshot,
    ap.token,
    ap.tokens_per_day,
    ap.avs,
    ap.strategy,
    ap.multiplier,
    ap.reward_for_all,
    sss.staker,
    sss.shares
  FROM {{ ref('1_active_rewards') }} ap
  JOIN {{ ref('staker_share_snapshots') }} as sss
  ON ap.strategy = sss.strategy and ap.snapshot = sss.snapshot
  WHERE ap.reward_for_all = true
  -- Parse out negative shares and zero multiplier so there is no division by zero case
  AND sss.shares > 0 and ap.multiplier != 0
),

-- Calculate the weight of a staker
staker_weights AS (
  SELECT *,
    SUM(multiplier * shares) OVER (PARTITION BY staker, reward_hash, snapshot) AS staker_weight
  FROM reward_snapshot_stakers
),
-- Get distinct stakers since their weights are already calculated
distinct_stakers AS (
  SELECT *
  FROM (
      SELECT *,
        -- We can use an arbitrary order here since the staker_weight is the same for each (staker, strategy, hash, snapshot)
        -- We use strategy ASC for better debuggability
        ROW_NUMBER() OVER (PARTITION BY reward_hash, snapshot, staker ORDER BY strategy ASC) as rn
      FROM staker_weights
  ) t
  WHERE rn = 1
  ORDER BY reward_hash, snapshot, staker
),
-- Calculate sum of all staker weights
staker_weight_sum AS (
  SELECT *,
    SUM(staker_weight) OVER (PARTITION BY reward_hash, snapshot) as total_staker_weight
  FROM distinct_stakers
),
-- Calculate staker token proportion
staker_proportion AS (
  SELECT *,
    FLOOR((staker_weight / total_staker_weight) * 1000000000000000) / 1000000000000000 AS staker_proportion
  FROM staker_weight_sum
),
-- Calculate total tokens to staker
staker_tokens AS (
  SELECT *,
  (tokens_per_day * staker_proportion)::text::decimal(38,0) as staker_tokens
  FROM staker_proportion
)
SELECT * from staker_tokens

WITH avs_opted_operators AS (
  SELECT DISTINCT
    snapshot,
    operator
  FROM {{ ref('operator_avs_registration_snapshots') }}
),

-- Get the operators who will earn rewards for the reward submission at the given snapshot
reward_snapshot_operators as (
  SELECT
    ap.reward_hash,
    ap.snapshot,
    ap.token,
    ap.tokens_per_day_decimal,
    ap.avs,
    ap.strategy,
    ap.multiplier,
    ap.reward_type,
    ap.reward_submission_date,
    aoo.operator
  FROM {{ ref('1_active_rewards') }} ap
  JOIN avs_opted_operators aoo
  ON ap.snapshot = aoo.snapshot
  WHERE ap.reward_type = 'all_earners'
),

-- Get the stakers that were delegated to the operator for the snapshot 
staker_delegated_operators AS (
  SELECT
    rso.*,
    sds.staker
  FROM reward_snapshot_operators rso
  JOIN {{ ref('staker_delegation_snapshots') }} sds
  ON
    rso.operator = sds.operator AND
    rso.snapshot = sds.snapshot
),

-- Get the shares of each strategy the staker has delegated to the operator
staker_strategy_shares AS (
  SELECT
    sdo.*,
    sss.shares
  FROM staker_delegated_operators sdo
  JOIN {{ ref('staker_share_snapshots') }} sss
  ON
    sdo.staker = sss.staker AND
    sdo.snapshot = sss.snapshot AND
    sdo.strategy = sss.strategy
  -- Parse out negative shares and zero multiplier so there is no division by zero case
  WHERE sss.shares > 0 and sdo.multiplier != 0
),
-- Calculate the weight of a staker
staker_weights AS (
  SELECT *,
    SUM(multiplier * shares) OVER (PARTITION BY staker, reward_hash, snapshot) AS staker_weight
  FROM staker_strategy_shares
),
-- Get distinct stakers since their weights are already calculated
distinct_stakers AS (
  SELECT *
  FROM (
      SELECT *,
        -- We can use an arbitrary order here since the staker_weight is the same for each (staker, strategy, hash, snapshot)
        -- We use strategy ASC for better debuggability
        ROW_NUMBER() OVER (PARTITION BY reward_hash, snapshot, staker ORDER BY strategy ASC) as rn
      FROM staker_weights
  ) t
  WHERE rn = 1
  ORDER BY reward_hash, snapshot, staker
),
-- Calculate sum of all staker weights for each reward and snapshot
staker_weight_sum AS (
  SELECT *,
    SUM(staker_weight) OVER (PARTITION BY reward_hash, snapshot) as total_weight
  FROM distinct_stakers
),
-- Calculate staker proportion of tokens for each reward and snapshot
staker_proportion AS (
  SELECT *,
    FLOOR((staker_weight / total_weight) * 1000000000000000) / 1000000000000000 AS staker_proportion
  FROM staker_weight_sum
),
-- Calculate total tokens to the (staker, operator) pair
staker_operator_total_tokens AS (
  SELECT *,
    FLOOR(staker_proportion * tokens_per_day_decimal) as total_staker_operator_payout
  FROM staker_proportion
),
-- Calculate the token breakdown for each (staker, operator) pair
token_breakdowns AS (
  SELECT *,
    floor(total_staker_operator_payout * 0.10) as operator_tokens,
    total_staker_operator_payout - floor(total_staker_operator_payout * 0.10) as staker_tokens
  FROM staker_operator_total_tokens
)
SELECT * from token_breakdowns
ORDER BY reward_hash, snapshot, staker, operator

-- Sum up the operator tokens across stakers for each operator, reward hash, and snapshot
WITH operator_token_sums AS (
  SELECT
    reward_hash,
    snapshot,
    token,
    tokens_per_day_decimal,
    avs,
    strategy,
    multiplier,
    reward_type,
    operator,
    SUM(operator_tokens) OVER (PARTITION BY operator, reward_hash, snapshot) AS operator_tokens
  FROM {{ ref('5_rfae_stakers') }}
),
-- Dedupe the operator tokens across strategies for each operator, reward hash, and snapshot
distinct_operators AS (
  SELECT *
  FROM (
      SELECT *,
        -- We can use an arbitrary order here since the staker_weight is the same for each (operator, strategy, hash, snapshot)
        -- We use strategy ASC for better debuggability
        ROW_NUMBER() OVER (PARTITION BY reward_hash, snapshot, operator ORDER BY strategy ASC) as rn
      FROM operator_token_sums
  ) t
  WHERE rn = 1
)
SELECT * FROM distinct_operators

WITH staker_rewards AS (
  -- We can select DISTINCT here because the staker's tokens are the same for each strategy in the reward hash
  SELECT DISTINCT
    staker as earner,
    snapshot,
    reward_hash,
    token,
    staker_tokens as amount
  FROM {{ ref('2_staker_reward_amounts') }}
),
operator_rewards AS (
  SELECT DISTINCT
    -- We can select DISTINCT here because the operator's tokens are the same for each strategy in the reward hash
    operator as earner,
    snapshot,
    reward_hash,
    token,
    operator_tokens as amount
  FROM {{ ref('3_operator_reward_amounts') }}
),
rewards_for_all AS (
  SELECT DISTINCT
    staker as earner,
    snapshot,
    reward_hash,
    token,
    staker_tokens as amount
  FROM {{ ref('4_deprecated_rewards_for_all') }}
),
rewards_for_all_earners_stakers AS (
  SELECT DISTINCT
    staker as earner,
    snapshot,
    reward_hash,
    token,
    staker_tokens as amounts
  FROM {{ ref('5_rfae_stakers.sql')}}
),
rewards_for_all_earners_operators AS (
  SELECT DISTINCT
    operator as earner,
    snapshot,
    reward_hash,
    token,
    operator_tokens as amount
  FROM {{ ref('6_rfae_operators.sql')}}
)
combined_rewards AS (
  SELECT * FROM operator_rewards
  UNION ALL
  SELECT * FROM staker_rewards
  UNION ALL
  SELECT * FROM rewards_for_all
  UNION ALL
  SELECT * FROM rewards_for_all_earners_stakers
  UNION ALL
  SELECT * FROM rewards_for_all_earners_operators
),

-- Dedupe earners, primarily operators who are also their own staker.
deduped_earners AS (
  SELECT
    earner,
    snapshot,
    reward_hash,
    token,
    SUM(amount) as amount
  FROM combined_rewards
  GROUP BY
    earner,
    snapshot,
    reward_hash,
    token
)
SELECT *
FROM deduped_earners

SELECT
    earner,
    snapshot,
    reward_hash,
    token,
    amount
FROM {{ ref('gold_staging') }}

SELECT
    earner,
    snapshot,
    reward_hash,
    token,
    amount
FROM {{ ref('gold_staging') }}