import warnings
from datetime import datetime

import dagster as dg
from dagster import EnvVar
from dagster_snowflake import SnowflakeResource

# from src.assets.dbt.analytics.dbt_analytics import hourly_partitioned_dbt_models
# Backend events are now pulled from individual staging tables
# Mobile events are now pulled from individual staging tables
from src.assets.snowflake.fact.fact_hook_play.assets import (
    FACT_HOOK_PLAY_START_DATE,
    FACT_HOOK_PLAY_TABLE_NAME,
    fact_hook_play,
)
from src.utils.automation_conditions import hourly_cron_with_eager_historical_backfill_condition
from src.utils.dbt import get_dbt_schema_name
from src.utils.snowflake.constants import TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H, PartitionExpr, Team, Warehouse
from src.utils.snowflake.query import JinjaSQLFormatter, PythonStringSQLFormatter

warnings.filterwarnings("ignore", category=dg.BetaWarning)

# Hook session data starts when we have both backend events and hook play data
HOOK_SESSION_TABLE_NAME = "HOOK_SESSION"
HOOK_SESSION_START_DATE = FACT_HOOK_PLAY_START_DATE
HOOK_SESSION_BACKEND_EVENTS_START_DATE = datetime.strptime('2025-07-31', '%Y-%m-%d')
HOOK_SESSION_SEGMENT_EVENTS_START_DATE = datetime.strptime('2025-08-14', '%Y-%m-%d')


class HookBackendEventsConfig(dg.Config):
    # Like/unlike events
    hook_like_table_name: str = "STG_BACKEND_EVENTS__HOOK_LIKE"
    hook_undo_like_table_name: str = "STG_BACKEND_EVENTS__HOOK_UNDO_LIKE"
    hook_dislike_table_name: str = "STG_BACKEND_EVENTS__HOOK_DISLIKE"
    hook_undo_dislike_table_name: str = "STG_BACKEND_EVENTS__HOOK_UNDO_DISLIKE"

    # Comment events
    hook_comment_table_name: str = "STG_BACKEND_EVENTS__HOOK_COMMENT"

    # Share events
    hook_share_table_name: str = "STG_BACKEND_EVENTS__HOOK_SHARE"

    # Profile and follow/unfollow events
    profile_follow_hooks_table_name: str = "STG_BACKEND_EVENTS__PROFILE_FOLLOW_HOOKS"
    profile_unfollow_hooks_table_name: str = "STG_BACKEND_EVENTS__PROFILE_UNFOLLOW_HOOKS"

    # Playlist events
    playlist_add_clip_hooks_table_name: str = "STG_BACKEND_EVENTS__PLAYLIST_ADD_CLIP_HOOKS"

    # Create hook events
    hook_create_submit_table_name: str = "STG_BACKEND_EVENTS__HOOK_CREATE_SUBMIT"
    hook_create_success_table_name: str = "STG_BACKEND_EVENTS__HOOK_CREATE_SUCCESS"

    # Report/hide creator events
    hook_report_table_name: str = "STG_BACKEND_EVENTS__HOOK_REPORT"
    hook_hide_creator_table_name: str = "STG_BACKEND_EVENTS__HOOK_HIDE_CREATOR"

    # Target table
    stg_hook_session_backend_events_agg_table_name: str = "STG_HOOK_SESSION_BACKEND_EVENTS"

    warehouse: str = Warehouse.HOOK_SESSION_X_SMALL.value

@dg.asset(
    name="stg_hook_session_backend_events",
    description="Staging table for backend hook session events",
    group_name="hooks_events_backend",
    partitions_def=dg.HourlyPartitionsDefinition(start_date=HOOK_SESSION_BACKEND_EVENTS_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep('stg_backend_events__hook_like', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_undo_like', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_dislike', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_undo_dislike', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_comment', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_report', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__profile_follow_hooks', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__profile_unfollow_hooks', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__playlist_add_clip_hooks', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_hide_creator', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_backend_events__hook_share', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": "stg_hook_session_backend_events",
        "data_start_date": HOOK_SESSION_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 120,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def stg_hook_session_backend_events(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: HookBackendEventsConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Because hook sessions can span multiple hour partitions, we use a buffered window to fetch data.
    # We then use the actual partition window to filter and insert data that belongs to the target partition.
    is_multi_partition_range = context.has_partition_key_range
    fetch_window = context.asset_partitions_time_window_for_input('stg_backend_events__hook_like')
    fetch_start_ts = fetch_window.start
    fetch_end_ts = fetch_window.end

    fetch_params = {
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "fact_hook_play_table_name": FACT_HOOK_PLAY_TABLE_NAME,
        "stg_hook_session_backend_events_agg_table_name": config.stg_hook_session_backend_events_agg_table_name,
        "dbt_stg_schema": get_dbt_schema_name("stg"),
    }

    logger.info(f"Processing target partition(s): {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")
    logger.info(f"Fetch params: {fetch_params}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.stg_hook_session_backend_events_agg_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}.")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={**fetch_params, "delete_partition_table_name": config.stg_hook_session_backend_events_agg_table_name}, logger=logger)
        cursor.execute(delete_query)

        # 2. Insert new hook session data into the target table
        logger.info(f"Inserting new hook session data into {config.stg_hook_session_backend_events_agg_table_name}.")
        insert_query = jinja_formatter.load("src/assets/snowflake/session/hook_session/queries/stg_hook_session_backend_events_insert.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_query)

        rows_inserted = cursor.rowcount

    logger.info(f"Successfully processed partition(s). Inserted {rows_inserted} rows.")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.stg_hook_session_backend_events_agg_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": rows_inserted if not is_multi_partition_range else 0,
        },
    )


class HookSegmentEventsConfig(dg.Config):
    # Comment events
    comment_tapped_android_table_name: str = "COMMENT_TAPPED_ANDROID"
    comment_tapped_ios_table_name: str = "COMMENT_TAPPED_IOS"
    comment_begin_write_android_table_name: str = "COMMENT_BEGIN_WRITE_ANDROID"
    # comment_begin_write_ios_table_name: str = "COMMENT_BEGIN_WRITE_IOS" # TODO: Does not exist yet

    # Share events
    share_tapped_android_table_name: str = "HOOK_SHARE_TAPPED_ANDROID"
    share_tapped_ios_table_name: str = "HOOK_SHARE_TAPPED_IOS"
    share_clicked_web_table_name: str = "HOOK_SHARE_CLICKED_WEB"

    # Profile and follow/unfollow events
    hook_creator_clicked_web_table_name: str = "HOOK_CREATOR_CLICKED_WEB" # TODO: Does not exist yet

    # Playlist events
    add_hook_song_to_playlist_tapped_android_table_name: str = "ADD_HOOK_SONG_TO_PLAYLIST_TAPPED_ANDROID"
    add_hook_song_to_playlist_tapped_ios_table_name: str = "ADD_HOOK_SONG_TO_PLAYLIST_TAPPED_IOS"
    add_hook_song_to_playlist_clicked_web_table_name: str = "ADD_HOOK_SONG_TO_PLAYLIST_CLICKED_WEB"

    # Create hook events
    # create_hook_tapped_android_table_name: str = "CREATE_HOOK_TAPPED_ANDROID" # TODO: Does not exist yet
    create_hook_tapped_ios_table_name: str = "CREATE_HOOK_TAPPED_IOS"
    create_hook_clicked_web_table_name: str = "CREATE_HOOK_CLICKED_WEB"
    remix_hook_song_tapped_android_table_name: str = "REMIX_HOOK_SONG_TAPPED_ANDROID"
    remix_hook_song_tapped_ios_table_name: str = "REMIX_HOOK_SONG_TAPPED_IOS"
    remix_clicked_web_table_name: str = "REMIX_HOOK_SONG_CLICKED_WEB"

    # Misc button taps & event fire counts
    song_pill_tapped_android_table_name: str = "SONG_PILL_TAPPED_ANDROID"
    song_pill_tapped_ios_table_name: str = "SONG_PILL_TAPPED_IOS"
    go_to_full_song_tapped_android_table_name: str = "GO_TO_FULL_SONG_TAPPED_ANDROID"
    go_to_full_song_tapped_ios_table_name: str = "GO_TO_FULL_SONG_TAPPED_IOS"

    # Target table
    stg_hook_session_segment_events_table_name: str = "STG_HOOK_SESSION_SEGMENT_EVENTS"

    warehouse: str = Warehouse.HOOK_SESSION_X_SMALL.value

@dg.asset(
    name="stg_hook_session_segment_events",
    description="Staging table for segment hook session events",
    group_name="hooks_events_segment",
    partitions_def=dg.HourlyPartitionsDefinition(start_date=HOOK_SESSION_SEGMENT_EVENTS_START_DATE, end_offset=-1),
    deps=[
        # Frontend hook events - Android
        dg.AssetDep('stg_segment_events__hook_comment_tapped_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_comment_begin_write_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__add_hook_song_to_playlist_tapped_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_share_tapped_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_song_pill_tapped_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_go_to_full_song_tapped_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__remix_hook_song_tapped_android', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        # Frontend hook events - iOS
        dg.AssetDep('stg_segment_events__hook_comment_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__add_hook_song_to_playlist_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_share_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_song_pill_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_go_to_full_song_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__create_hook_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__remix_hook_song_tapped_ios', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        # Frontend hook events - Web
        dg.AssetDep('stg_segment_events__add_hook_song_to_playlist_clicked_web', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_share_clicked_web', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_creator_clicked_web', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__create_hook_clicked_web', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__hook_song_title_clicked_web', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep('stg_segment_events__remix_hook_song_clicked_web', partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1, allow_nonexistent_upstream_partitions=True)),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": "stg_hook_session_segment_events",
        "data_start_date": HOOK_SESSION_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 120,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def stg_hook_session_segment_events(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: HookSegmentEventsConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Use buffered window for events to capture events that might span partitions
    # Since all staging models use the same partition window, we can use any one to get the buffered window
    staging_events_window = context.asset_partitions_time_window_for_input('stg_segment_events__hook_comment_tapped_android')
    is_multi_partition_range = context.has_partition_key_range

    # Use the staging window for fetching events (already has 1-hour buffer from end_offset=1)
    fetch_start_ts = staging_events_window.start
    fetch_end_ts = staging_events_window.end

    fetch_params = {
        # Buffered window for fetching events
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        # Target partition window
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        "fact_hook_play_table_name": FACT_HOOK_PLAY_TABLE_NAME,
        "stg_hook_session_segment_events_table_name": config.stg_hook_session_segment_events_table_name,
        "dbt_stg_schema": get_dbt_schema_name("stg"),
    }

    logger.info(f"Processing target partition(s): {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")
    logger.info(f"Fetch params: {fetch_params}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.stg_hook_session_segment_events_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}.")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={**fetch_params, "delete_partition_table_name": config.stg_hook_session_segment_events_table_name}, logger=logger)
        cursor.execute(delete_query)

        # 2. Insert new hook session data into the target table
        logger.info(f"Inserting new hook session data into {config.stg_hook_session_segment_events_table_name}.")
        insert_query = jinja_formatter.load("src/assets/snowflake/session/hook_session/queries/stg_hook_session_segment_events_insert.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_query)

        rows_inserted = cursor.rowcount

    logger.info(f"Successfully processed partition(s). Inserted {rows_inserted} rows.")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.stg_hook_session_segment_events_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": rows_inserted if not is_multi_partition_range else 0,
        },
    )

class HookSessionConfig(dg.Config):
    hook_session_table_name: str = HOOK_SESSION_TABLE_NAME
    warehouse: str = Warehouse.HOOK_SESSION_X_SMALL.value


@dg.asset(
    name="hook_session",
    description="Hook session table aggregating engagement events (likes, comments, taps, etc) per hook play session",
    group_name="hooks",
    partitions_def=dg.HourlyPartitionsDefinition(start_date=HOOK_SESSION_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(fact_hook_play, partition_mapping=dg.TimeWindowPartitionMapping()),
        dg.AssetDep(stg_hook_session_segment_events, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_hook_session_backend_events, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": "hook_session",
        "data_start_date": HOOK_SESSION_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
        "transient": True,
        "sla_minutes": 180,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
    freshness_policy=TIME_WINDOW_FRESHNESS_POLICY_WARN_1H_FAIL_2H,
)
def hook_session(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: HookSessionConfig) -> dg.MaterializeResult:
    run_id = context.run.run_id
    logger = dg.get_dagster_logger()
    jinja_formatter = JinjaSQLFormatter()
    python_formatter = PythonStringSQLFormatter()

    # Use the staging tables' partition window since they already have the processed data
    # The staging tables handle the buffered window internally
    is_multi_partition_range = context.has_partition_key_range
    fetch_start_ts = context.partition_time_window.start
    fetch_end_ts = context.partition_time_window.end


    fetch_params = {
        # Buffered window for fetching events
        "buffered_partition_start_date": fetch_start_ts.strftime("%Y-%m-%d"),
        "buffered_partition_end_date": fetch_end_ts.strftime("%Y-%m-%d"),
        "buffered_partition_start_hour": fetch_start_ts.hour,
        "buffered_partition_end_hour": fetch_end_ts.hour,
        # Target partition window
        "partition_start_date": context.partition_time_window.start.strftime("%Y-%m-%d"),
        "partition_end_date": context.partition_time_window.end.strftime("%Y-%m-%d"),
        "partition_start_hour": context.partition_time_window.start.hour,
        "partition_end_hour": context.partition_time_window.end.hour,
        # Table names
        "fact_hook_play_table_name": FACT_HOOK_PLAY_TABLE_NAME,
        "hook_session_table_name": config.hook_session_table_name,
        "stg_hook_session_backend_events_agg_table_name": "STG_HOOK_SESSION_BACKEND_EVENTS",
        "stg_hook_session_segment_events_table_name": "STG_HOOK_SESSION_SEGMENT_EVENTS",
        # dbt schema for frontend staging tables
        "dbt_stg_schema": get_dbt_schema_name("stg")
    }

    logger.info(f"Processing target partition(s): {context.partition_time_window.start} to {context.partition_time_window.end}")
    logger.info(f"Fetching raw data from buffered window: {fetch_start_ts} to {fetch_end_ts}")
    logger.info(f"Fetch params: {fetch_params}")

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        logger.info(f"Using warehouse {config.warehouse}")
        warehouse_query = python_formatter.load("src/utils/snowflake/queries/use_warehouse.sql", params={"warehouse": config.warehouse}, logger=logger)
        cursor.execute(warehouse_query)

        # 1. Delete existing data from the target table for the partition window
        logger.info(f"Deleting existing data from {config.hook_session_table_name} for partition window {context.partition_time_window.start} to {context.partition_time_window.end}.")
        delete_query = python_formatter.load("src/utils/snowflake/queries/delete_hourly_partitions.sql", params={**fetch_params, "delete_partition_table_name": config.hook_session_table_name}, logger=logger)
        cursor.execute(delete_query)

        # 2. Insert new hook session data into the target table
        logger.info(f"Inserting new hook session data into {config.hook_session_table_name}.")
        insert_query = jinja_formatter.load("src/assets/snowflake/session/hook_session/queries/insert_stg_to_final_table.sql", params=fetch_params, logger=logger)
        cursor.execute(insert_query)

        rows_inserted = cursor.rowcount

    logger.info(f"Successfully processed partition(s). Inserted {rows_inserted} rows.")

    return dg.MaterializeResult(
        metadata={
            "run_id": dg.MetadataValue.text(run_id),
            "table_name": config.hook_session_table_name,
            "partition_time_window_start": dg.MetadataValue.text(context.partition_time_window.start.isoformat()),
            "partition_time_window_end": dg.MetadataValue.text(context.partition_time_window.end.isoformat()),
            "dagster/row_count": rows_inserted if not is_multi_partition_range else 0,
        },
    )