"""Lambda hive_ai_detection function module."""

import config
import json
import logging
import sentry_sdk
from httpx import HTTPStatusError
from kafka_utils.consumer.deserializer.simple_json import JSONDeserializer
from kafka_utils.consumer.source.mapping import EventSourceMessage
from lambdacommon.common_config import logger
from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration
from src.connectors import hive
from src.connectors import ows_assets

# httpx logs an INFO "HTTP Request: ..." line for every call (~3 per asset);
# we already log our own per-asset outcome and errors, so silence that noise.
logging.getLogger("httpx").setLevel(logging.WARNING)

# initialize sentry
if config.SENTRY_DSN:
    logger.info("Initializing with sentry")
    sentry_sdk.init(config.SENTRY_DSN, integrations=[AwsLambdaIntegration()])
else:
    logger.info("Initializing without sentry")


def handler(event, context) -> dict:
    """Lambda entry point."""
    try:
        for asset_final_id, duration, overwrite in decode_records(event):
            # check for unprocessable assets
            if duration >= config.MAX_ASSET_DURATION_SECONDS * 1000:
                logger.info(
                    f"Skipped asset_final_id {asset_final_id} due to duration of {duration}ms"
                )
                continue

            # get presigned url
            try:
                url = ows_assets.get_presigned_url(asset_final_id)
            except HTTPStatusError as e:
                if e.response.status_code == 404:
                    logger.info(
                        f"Skipping asset_final_id {asset_final_id}, asset deleted before Hive scan started"
                    )
                    continue
                raise e

            # run hive task
            hive_results = hive.run_task(url)

            # save results
            try:
                ows_assets.post_results(
                    asset_final_id, hive_results, overwrite=overwrite
                )
            except HTTPStatusError as e:
                # A first-scan-wins 409 is benign, but a 409 when overwrite was
                # requested means the endpoint didn't honor the flag -> let it raise.
                if e.response.status_code == 409 and not overwrite:
                    logger.info(
                        f"Skipping asset_final_id {asset_final_id}, results exist"
                    )
                    continue
                if e.response.status_code == 404:
                    logger.info(
                        f"Skipping asset_final_id {asset_final_id}, asset deleted, Hive results discarded"
                    )
                    continue
                raise e

            logger.info(f"Processed asset_final_id {asset_final_id}")
        return {"status": "OK"}
    except HTTPStatusError as e:
        if e.response.status_code == 400:
            logger.error(e.response.json())
        logger.exception(str(e))
        raise e
    except Exception as e:
        logger.exception(str(e))
        raise e


def decode_records(event) -> list[tuple[int, int, bool]]:
    if event.get("eventSource") == "aws:kafka":
        deserializer = JSONDeserializer()
        data = [
            deserializer.deserialize(msk_message.value)
            for _, msk_message in EventSourceMessage(event)
        ]
    elif event.get("eventSource") == "custom":
        data = event.get("assets", [])
    elif event.get("Records", [{}])[0].get("eventSource") == "aws:sqs":
        data = [json.loads(record["body"]) for record in event["Records"]]
    else:
        raise Exception("Unrecognized event format!")
    # OVERWRITE is an optional per-message flag (default first-scan-wins). The
    # enqueue-hive-backfill tool in scripts-assets sets it so a backfill re-scan
    # replaces an asset's existing scan; the field name is a cross-repo contract
    # with that producer. Only an exact boolean True counts as a request -- no
    # truthy coercion, so a stray or renamed value can't silently overwrite.
    return [
        (
            int(x["ASSET_FINAL_ID"]),
            int(x["DURATION_MS"]),
            x.get("OVERWRITE") is True,
        )
        for x in data
    ]