import uuid
from collections.abc import Generator
from datetime import date
from typing import Any

import pytest
from aws_testing_utils.lambda_handler import LambdaHandler
from test_fixtures.mysql import MySQLConnection
from test_fixtures.snowflake import SnowflakeConnection

from config import UPDATE_DIM_TABLES_FUNCTION_NAME

_ORIGINATING_VENDOR_ID = 6971
_DESTINATION_VENDOR_ID = 7123

# The deployed-QA lambda uses SNOWFLAKE_SCHEMA=QA, so the fully-qualified table names below
# must match. The integration-test Snowflake connection sets no default database/schema, so
# every query has to be fully qualified anyway.
_SNOWFLAKE_SCHEMA = "QA"
_DIM_RELEASE_HISTORY = f"FACTS.{_SNOWFLAKE_SCHEMA}.DIM_RELEASE_HISTORY"
_DIM_RELEASE = f"FACTS.{_SNOWFLAKE_SCHEMA}.DIM_RELEASE"
_STATEMENT_PERIOD = (
    f"ORCHARD_APP_REPORTING_V2.{_SNOWFLAKE_SCHEMA}_ROYALTY_ACCOUNTING_ROYALTY_ACCOUNTING.STATEMENT_PERIOD"
)


def _placeholders(values: list[Any]) -> str:
    """Comma-separated `%s` string for a SQL IN clause (PyMySQL/Snowflake don't expand tuples)."""
    return ",".join(["%s"] * len(values))


def _count_inserted_rows(snowflake_db: SnowflakeConnection, release_id: int, start_statement_period_id: int) -> int:
    """Count DIM_RELEASE_HISTORY rows matching this run's insert signature for a release."""
    row = snowflake_db.fetchone(
        f"SELECT COUNT(*) AS CNT FROM {_DIM_RELEASE_HISTORY} "
        f"WHERE PRODUCT_ID = %s AND LABELID = %s AND START_STATEMENT_PERIOD_ID = %s",
        (release_id, _DESTINATION_VENDOR_ID, start_statement_period_id),
    )
    assert row is not None
    return int(row["CNT"])


@pytest.fixture
def dim_tables_job(
    art_relations_db: MySQLConnection,
    snowflake_db: SnowflakeConnection,
) -> Generator[dict[str, Any], None, None]:
    """Seed a transfer job that drives the full update-dim-tables happy path.

    Coordinates two datastores:
      - MySQL (art_relations): a project_transfer_job with revenue_cutoff_date set, plus
        product_transfer_history rows whose release_ids exist in Snowflake DIM_RELEASE.
      - Snowflake (FACTS.QA): discovers a real STATEMENT_PERIOD and a set of releases that
        exist in DIM_RELEASE, so both the period lookup and the INSERT…SELECT produce rows.

    revenue_cutoff_date is chosen as the 15th of the same month as a real statement period:
    the lambda looks up the period by matching YEAR/MONTH of the cutoff directly, and the
    mid-month day keeps the teardown's reopen filter from colliding with real end-of-month
    close dates.

    The lambda mutates DIM_RELEASE_HISTORY (closes open rows + inserts new ones). Those
    Snowflake writes are intentionally left in place — the test role has only SELECT on
    FACTS.QA and this QA reporting data isn't consumed downstream — so teardown only removes
    the MySQL seed. Each run therefore accumulates a few rows for the chosen releases, but the
    happy-path assertion still holds because every run's close step retires prior open rows
    before inserting the new one.
    """
    # 1. Pick a real statement period.
    period = snowflake_db.fetchone(
        f"SELECT STATEMENT_PERIOD_ID, STATEMENT_YEAR, STATEMENT_MONTH "
        f"FROM {_STATEMENT_PERIOD} "
        f"ORDER BY STATEMENT_YEAR DESC, STATEMENT_MONTH DESC LIMIT 1"
    )
    assert period, f"No STATEMENT_PERIOD rows found in {_STATEMENT_PERIOD}"
    statement_period_id = int(period["STATEMENT_PERIOD_ID"])
    cutoff = date(int(period["STATEMENT_YEAR"]), int(period["STATEMENT_MONTH"]), 15)
    revenue_cutoff_date = cutoff.isoformat()
    expected_start_period = statement_period_id + 1

    # 2. Find a project for the source vendor, then keep only releases that also exist in
    #    Snowflake DIM_RELEASE — the INSERT…SELECT only emits rows for those.
    project = art_relations_db.fetchone(
        """
        SELECT p.project_id
        FROM project p
        WHERE p.vendor_id = %s
          AND (SELECT COUNT(*) FROM releases WHERE project_id = p.project_id) >= 1
        LIMIT 1
        """,
        (_ORIGINATING_VENDOR_ID,),
    )
    assert project, f"No project with releases found for vendor_id {_ORIGINATING_VENDOR_ID}"

    mysql_release_ids = [
        r["release_id"]
        for r in art_relations_db.fetchall(
            "SELECT release_id FROM releases WHERE project_id = %s LIMIT 20",
            (project["project_id"],),
        )
    ]
    assert mysql_release_ids, f"Project {project['project_id']} has no releases"

    dim_release_rows = snowflake_db.fetchall(
        f"SELECT DISTINCT PRODUCT_ID FROM {_DIM_RELEASE} WHERE PRODUCT_ID IN ({_placeholders(mysql_release_ids)})",
        tuple(mysql_release_ids),
    )
    release_ids = [int(r["PRODUCT_ID"]) for r in dim_release_rows][:2]
    assert release_ids, (
        f"None of vendor {_ORIGINATING_VENDOR_ID}'s releases exist in {_DIM_RELEASE}; "
        f"cannot drive the INSERT path. Point the fixture at known-good release_ids."
    )

    # 3. Seed the job (with revenue_cutoff_date) and its product rows in MySQL.
    created_by = str(uuid.uuid4())
    art_relations_db.execute(
        """
        INSERT INTO project_transfer_job
            (project_id, originating_vendor_id, destination_vendor_id,
             revenue_cutoff_date, created_by_identity_id)
        VALUES (%s, %s, %s, %s, %s)
        """,
        (
            project["project_id"],
            _ORIGINATING_VENDOR_ID,
            _DESTINATION_VENDOR_ID,
            revenue_cutoff_date,
            created_by,
        ),
    )
    job = art_relations_db.fetchone("SELECT * FROM project_transfer_job WHERE job_id = LAST_INSERT_ID()")
    assert job is not None

    for release_id in release_ids:
        art_relations_db.execute(
            "INSERT INTO product_transfer_history (job_id, release_id) VALUES (%s, %s)",
            (job["job_id"], release_id),
        )

    yield {
        "job": job,
        "release_ids": release_ids,
        "revenue_cutoff_date": revenue_cutoff_date,
        "statement_period_id": statement_period_id,
        "expected_start_period": expected_start_period,
    }

    # --- Teardown ---
    # The lambda's Snowflake writes (closed + inserted DIM_RELEASE_HISTORY rows) are left in
    # place on purpose: the test role has only SELECT on FACTS.QA, and this QA reporting data
    # isn't consumed downstream. Only the MySQL seed is cleaned up.
    art_relations_db.execute("DELETE FROM product_transfer_history WHERE job_id = %s", (job["job_id"],))
    art_relations_db.execute("DELETE FROM project_transfer_job WHERE job_id = %s", (job["job_id"],))


@pytest.fixture
def job_with_unmatched_cutoff(
    art_relations_db: MySQLConnection,
    snowflake_db: SnowflakeConnection,
) -> Generator[dict[str, Any], None, None]:
    """Seed a job whose revenue_cutoff_date resolves to a month with no STATEMENT_PERIOD.

    Non-destructive: the lambda fails at the statement-period lookup, before any
    DIM_RELEASE_HISTORY writes — so no Snowflake cleanup is needed, only the MySQL job.
    """
    max_year_row = snowflake_db.fetchone(f"SELECT MAX(STATEMENT_YEAR) AS MAX_YEAR FROM {_STATEMENT_PERIOD}")
    assert max_year_row and max_year_row["MAX_YEAR"] is not None, "STATEMENT_PERIOD is empty"
    # Five years past the latest known period → guaranteed no matching period.
    cutoff = date(int(max_year_row["MAX_YEAR"]) + 5, 6, 15)

    project = art_relations_db.fetchone(
        "SELECT project_id FROM project WHERE vendor_id = %s LIMIT 1",
        (_ORIGINATING_VENDOR_ID,),
    )
    assert project, f"No project found for vendor_id {_ORIGINATING_VENDOR_ID}"

    created_by = str(uuid.uuid4())
    art_relations_db.execute(
        """
        INSERT INTO project_transfer_job
            (project_id, originating_vendor_id, destination_vendor_id,
             revenue_cutoff_date, created_by_identity_id)
        VALUES (%s, %s, %s, %s, %s)
        """,
        (
            project["project_id"],
            _ORIGINATING_VENDOR_ID,
            _DESTINATION_VENDOR_ID,
            cutoff.isoformat(),
            created_by,
        ),
    )
    job = art_relations_db.fetchone("SELECT * FROM project_transfer_job WHERE job_id = LAST_INSERT_ID()")
    assert job is not None

    yield job

    art_relations_db.execute("DELETE FROM project_transfer_job WHERE job_id = %s", (job["job_id"],))


pytestmark = pytest.mark.lambda_name("update-dim-tables")


def test_inserts_open_history_rows_for_destination_vendor(
    snowflake_db: SnowflakeConnection,
    dim_tables_job: dict[str, Any],
    lambda_handler: LambdaHandler,
) -> None:
    """End-to-end: the lambda closes any open DIM_RELEASE_HISTORY rows for the job's
    releases and inserts a fresh open row per release carrying the destination vendor
    and the next statement period.

    After the run, every open (END_DATE_INCLUSIVE IS NULL) row for each release must
    belong to this run — proving both the close step (no stale open rows remain) and
    the insert step.
    """
    job_id = dim_tables_job["job"]["job_id"]
    release_ids = dim_tables_job["release_ids"]
    expected_start_period = dim_tables_job["expected_start_period"]

    lambda_handler.invoke(UPDATE_DIM_TABLES_FUNCTION_NAME, {"job_id": job_id})

    for release_id in release_ids:
        open_rows = snowflake_db.fetchall(
            f"SELECT LABELID, SUBACCOUNTID, START_STATEMENT_PERIOD_ID "
            f"FROM {_DIM_RELEASE_HISTORY} "
            f"WHERE PRODUCT_ID = %s AND END_DATE_INCLUSIVE IS NULL",
            (release_id,),
        )
        assert open_rows, f"No open DIM_RELEASE_HISTORY row was inserted for release {release_id}"
        for row in open_rows:
            assert int(row["START_STATEMENT_PERIOD_ID"]) == expected_start_period, (
                f"Release {release_id} has an open row with START_STATEMENT_PERIOD_ID "
                f"{row['START_STATEMENT_PERIOD_ID']}, expected {expected_start_period} "
                f"(a stale open row means the close step didn't run)"
            )
            assert int(row["LABELID"]) == _DESTINATION_VENDOR_ID, (
                f"Release {release_id} open row has LABELID {row['LABELID']}, "
                f"expected destination vendor {_DESTINATION_VENDOR_ID}"
            )
            assert row["SUBACCOUNTID"] is None, (
                f"Release {release_id} open row has SUBACCOUNTID {row['SUBACCOUNTID']}, "
                f"expected NULL (job has no destination_subaccount_id)"
            )


def test_function_error_when_no_statement_period_for_cutoff(
    job_with_unmatched_cutoff: dict[str, Any],
    lambda_handler: LambdaHandler,
) -> None:
    """When revenue_cutoff_date resolves to a month with no STATEMENT_PERIOD, the lambda
    raises before any Snowflake writes — surfaced as a FunctionError so Step Functions
    routes to its error handler.
    """
    job_id = job_with_unmatched_cutoff["job_id"]

    response = lambda_handler.invoke(
        UPDATE_DIM_TABLES_FUNCTION_NAME,
        {"job_id": job_id},
        assertion=False,
    )
    assert response["StatusCode"] == 200
    assert "FunctionError" in response, (
        f"Expected a FunctionError for job {job_id} whose cutoff has no statement period, "
        f"but response had none: { {k: v for k, v in response.items() if k != 'Payload'} }"
    )


def test_function_error_when_revenue_cutoff_date_missing(
    transfer_job: dict[str, Any],
    lambda_handler: LambdaHandler,
) -> None:
    """The shared transfer_job fixture seeds a job with revenue_cutoff_date NULL.
    get_transfer_job rejects that (raises before any Snowflake work), surfaced as a
    FunctionError. Non-destructive — nothing is written to DIM_RELEASE_HISTORY.
    """
    response = lambda_handler.invoke(
        UPDATE_DIM_TABLES_FUNCTION_NAME,
        {"job_id": transfer_job["job_id"]},
        assertion=False,
    )
    assert response["StatusCode"] == 200
    assert "FunctionError" in response, (
        f"Expected a FunctionError for job {transfer_job['job_id']} with NULL "
        f"revenue_cutoff_date, but response had none: "
        f"{ {k: v for k, v in response.items() if k != 'Payload'} }"
    )


def test_function_error_when_job_id_does_not_exist(
    lambda_handler: LambdaHandler,
) -> None:
    """ows-project-manager returns 4xx for a non-existent job_id; the lambda surfaces it
    as a FunctionError before any Snowflake work. No fixture/cleanup needed.
    """
    response = lambda_handler.invoke(
        UPDATE_DIM_TABLES_FUNCTION_NAME,
        {"job_id": 2_000_000_000},
        assertion=False,
    )
    assert response["StatusCode"] == 200
    assert "FunctionError" in response, (
        f"Expected a FunctionError for a nonexistent job, but response had none: "
        f"{ {k: v for k, v in response.items() if k != 'Payload'} }"
    )


def test_re_running_inserts_duplicate_history_rows(
    snowflake_db: SnowflakeConnection,
    dim_tables_job: dict[str, Any],
    lambda_handler: LambdaHandler,
) -> None:
    """update-dim-tables is NOT idempotent. A second invocation closes the row the first
    run inserted (its only open row) and inserts a fresh one — so the insert-signature row
    count grows with each run rather than staying flat.

    Unlike finalize-job / handle-error (where ows-project-manager's terminal-state guard
    blocks re-runs), this lambda writes straight to Snowflake with no such guard. This test
    pins the non-idempotency so adding a guard — or relying on an SFN retry policy that
    could re-invoke this step — is a conscious decision rather than a silent surprise.
    """
    job_id = dim_tables_job["job"]["job_id"]
    release_ids = dim_tables_job["release_ids"]
    expected_start_period = dim_tables_job["expected_start_period"]

    lambda_handler.invoke(UPDATE_DIM_TABLES_FUNCTION_NAME, {"job_id": job_id})
    counts_after_first = {
        release_id: _count_inserted_rows(snowflake_db, release_id, expected_start_period) for release_id in release_ids
    }

    lambda_handler.invoke(UPDATE_DIM_TABLES_FUNCTION_NAME, {"job_id": job_id})

    for release_id in release_ids:
        after_second = _count_inserted_rows(snowflake_db, release_id, expected_start_period)
        assert after_second > counts_after_first[release_id], (
            f"Re-running added no DIM_RELEASE_HISTORY rows for release {release_id} "
            f"(count stayed {after_second}); expected duplicates since the lambda is "
            f"not idempotent"
        )