#!/usr/bin/env python3
import json
import os
import time
import uuid
from datetime import date, datetime, timezone
from decimal import Decimal

import boto3
import pymysql

from hubspot_sync import refresh_hubspot_snapshot


TENANT_COLUMNS = """
  id,
  created_at,
  updated_at,
  name,
  email,
  country,
  time_zone,
  status,
  kind,
  plan,
  alias,
  country_region,
  sales_owner_email,
  custom_id,
  segmentation,
  sales_force_region,
  sales_force_kind,
  tam_name
"""

DEFAULT_TENANT_SELECT_SQL = f"""
SELECT
{TENANT_COLUMNS}
FROM tenants
WHERE status = 'active'
  AND kind = 'external'
  AND sales_force_region = 'APAC'
  AND country NOT IN ('JP')
  AND plan NOT IN ('free_trial', 'new_user', 'closing')
"""

CST_CURRENT_DAY_START_UTC_SQL = (
    "DATE_SUB(DATE(DATE_ADD(UTC_TIMESTAMP(), INTERVAL 8 HOUR)), INTERVAL 8 HOUR)"
)

DEFAULT_NEW_SIGNUP_SELECT_SQL = f"""
SELECT
{TENANT_COLUMNS}
FROM tenants
WHERE created_at >= DATE_SUB({CST_CURRENT_DAY_START_UTC_SQL}, INTERVAL 1 DAY)
  AND created_at < {CST_CURRENT_DAY_START_UTC_SQL}
  AND country_region = 'APAC'
  AND country NOT IN ('JP')
"""

_sqs_client = None
_lambda_client = None


def env(name, default=None, required=False):
    value = os.environ.get(name, default)
    if required and (value is None or str(value).strip() == ""):
        raise RuntimeError(f"{name} is required")
    return value


def json_default(value):
    if isinstance(value, datetime):
        if value.tzinfo is not None:
            value = value.astimezone(timezone.utc).replace(tzinfo=None)
        return value.strftime("%Y-%m-%d %H:%M:%S")
    if isinstance(value, date):
        return value.isoformat()
    if isinstance(value, Decimal):
        return str(value)
    raise TypeError(f"{type(value).__name__} is not JSON serializable")


def log_event(event_name, **fields):
    print(
        json.dumps(
            {"event": event_name, **fields},
            default=json_default,
            separators=(",", ":"),
            sort_keys=True,
        )
    )


def sqs_client():
    global _sqs_client
    if _sqs_client is None:
        _sqs_client = boto3.client("sqs")
    return _sqs_client


def lambda_client():
    global _lambda_client
    if _lambda_client is None:
        from botocore.config import Config

        timeout = int(env("EFFECTIVE_OWNER_SYNC_INVOKE_TIMEOUT_SECONDS", "720"))
        _lambda_client = boto3.client(
            "lambda",
            config=Config(
                connect_timeout=10,
                read_timeout=timeout,
                retries={"max_attempts": 0},
            ),
        )
    return _lambda_client


def mysql_connect():
    return pymysql.connect(
        host=env("DEVOPS_DB_HOST", required=True),
        port=int(env("DEVOPS_DB_PORT", "3306")),
        user=env("DEVOPS_DB_USER", required=True),
        password=env("DEVOPS_DB_PASSWORD", required=True),
        database=env("DEVOPS_DB_NAME", required=True),
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
        connect_timeout=10,
        read_timeout=60,
        write_timeout=60,
    )


def select_tenants(connection, sql):
    with connection.cursor() as cursor:
        cursor.execute(sql)
        return cursor.fetchall()


def message_body(tenant, source_type, run_id, snapshot_metadata):
    return json.dumps(
        {
            "source_type": source_type,
            "tenant": tenant,
            "tenant_id": str(tenant.get("id") or ""),
            "run_id": run_id,
            **snapshot_metadata,
        },
        default=json_default,
        separators=(",", ":"),
    )


def send_batch(client, queue_url, batch, failed):
    response = client.send_message_batch(QueueUrl=queue_url, Entries=batch)
    failed.extend(response.get("Failed") or [])
    return len(response.get("Successful") or [])


def enqueue_tenants(queue_url, tenants, source_type, run_id, snapshot_metadata):
    sent = 0
    failed = []
    batch = []
    client = sqs_client()
    total = len(tenants)

    for tenant in tenants:
        batch.append(
            {
                "Id": f"m{len(batch)}",
                "MessageBody": message_body(
                    tenant, source_type, run_id, snapshot_metadata
                ),
            }
        )
        if len(batch) == 10:
            sent += send_batch(client, queue_url, batch, failed)
            log_event(
                "dispatcher_enqueue_progress",
                run_id=run_id,
                source_type=source_type,
                enqueued=sent,
                total=total,
                failed=len(failed),
            )
            batch = []

    if batch:
        sent += send_batch(client, queue_url, batch, failed)
        log_event(
            "dispatcher_enqueue_progress",
            run_id=run_id,
            source_type=source_type,
            enqueued=sent,
            total=total,
            failed=len(failed),
        )
    if failed:
        raise RuntimeError(f"failed to enqueue {len(failed)} {source_type} monitor messages")
    return sent


def bool_event_value(event, name, default):
    value = (event or {}).get(name, default)
    if isinstance(value, bool):
        return value
    if value is None:
        return default
    return str(value).strip().lower() not in {"0", "false", "no", "off"}


def sync_effective_owners(tenants, sync_id):
    tenant_ids = list(
        dict.fromkeys(str(tenant.get("id") or "") for tenant in tenants)
    )
    tenant_ids = [tenant_id for tenant_id in tenant_ids if tenant_id]
    if not tenant_ids:
        return {
            "effective_owner_sync_id": None,
            "effective_owner_rule_version": None,
            "effective_owner_tenant_count": 0,
        }

    response = lambda_client().invoke(
        FunctionName=env("EFFECTIVE_OWNER_SYNC_FUNCTION_NAME", required=True),
        InvocationType="RequestResponse",
        Payload=json.dumps(
            {
                "sync_id": sync_id,
                "tenant_ids": tenant_ids,
                "dry_run": False,
            },
            separators=(",", ":"),
        ).encode("utf-8"),
    )
    raw_payload = response["Payload"].read()
    try:
        result = json.loads(raw_payload or b"{}")
    except json.JSONDecodeError as exc:
        raise RuntimeError("Effective Owner Lambda returned invalid JSON") from exc
    if response.get("FunctionError"):
        error_type = str(result.get("errorType") or "LambdaFunctionError")
        raise RuntimeError(f"Effective Owner sync failed: {error_type}")
    if int(response.get("StatusCode") or 0) != 200:
        raise RuntimeError("Effective Owner Lambda invocation failed")
    if result.get("sync_id") != sync_id:
        raise RuntimeError("Effective Owner Lambda returned a mismatched sync_id")
    if int(result.get("tenant_count") or 0) != len(tenant_ids):
        raise RuntimeError("Effective Owner Lambda returned a mismatched tenant count")
    return {
        "effective_owner_sync_id": sync_id,
        "effective_owner_rule_version": result.get("rule_version"),
        "effective_owner_tenant_count": len(tenant_ids),
    }


def lambda_handler(event, context):
    queue_url = env("MONITOR_QUEUE_URL", required=True)
    now = datetime.now(timezone.utc)
    run_id = str((event or {}).get("run_id") or now.strftime("%Y%m%dT%H%M%SZ"))
    limit = int((event or {}).get("limit") or 0)
    include_active = bool_event_value(event, "include_active", True)
    include_new_signup = bool_event_value(event, "include_new_signup", True)
    started = time.time()
    log_event(
        "dispatcher_started",
        run_id=run_id,
        include_active=include_active,
        include_new_signup=include_new_signup,
        limit=limit,
    )

    snapshot_metadata = refresh_hubspot_snapshot(run_id, now)

    devops_conn = mysql_connect()
    try:
        active_tenants = (
            select_tenants(devops_conn, DEFAULT_TENANT_SELECT_SQL)
            if include_active
            else []
        )
        new_signup_tenants = (
            select_tenants(devops_conn, DEFAULT_NEW_SIGNUP_SELECT_SQL)
            if include_new_signup
            else []
        )
    finally:
        devops_conn.close()

    if limit > 0:
        active_tenants = active_tenants[:limit]
        new_signup_tenants = new_signup_tenants[:limit]
    log_event(
        "dispatcher_tenants_selected",
        run_id=run_id,
        active_tenant_count=len(active_tenants),
        new_signup_tenant_count=len(new_signup_tenants),
        hubspot_snapshot_status=snapshot_metadata["hubspot_snapshot_status"],
    )

    effective_owner_metadata = sync_effective_owners(
        [*active_tenants, *new_signup_tenants], str(uuid.uuid4())
    )
    message_metadata = {**snapshot_metadata, **effective_owner_metadata}
    log_event(
        "dispatcher_effective_owner_synced",
        run_id=run_id,
        effective_owner_sync_id=effective_owner_metadata["effective_owner_sync_id"],
        effective_owner_tenant_count=effective_owner_metadata[
            "effective_owner_tenant_count"
        ],
        effective_owner_rule_version=effective_owner_metadata[
            "effective_owner_rule_version"
        ],
    )

    active_enqueued = enqueue_tenants(
        queue_url, active_tenants, "active_candidate", run_id, message_metadata
    )
    new_signup_enqueued = enqueue_tenants(
        queue_url, new_signup_tenants, "new_signup", run_id, message_metadata
    )
    elapsed = round(time.time() - started, 3)
    log_event(
        "dispatcher_completed",
        run_id=run_id,
        active_enqueued=active_enqueued,
        new_signup_enqueued=new_signup_enqueued,
        total_enqueued=active_enqueued + new_signup_enqueued,
        hubspot_snapshot_status=snapshot_metadata["hubspot_snapshot_status"],
        elapsed_seconds=elapsed,
    )
    return {
        "run_id": run_id,
        "active_tenant_count": len(active_tenants),
        "active_enqueued": active_enqueued,
        "new_signup_tenant_count": len(new_signup_tenants),
        "new_signup_enqueued": new_signup_enqueued,
        "hubspot_snapshot": snapshot_metadata,
        "effective_owner_sync": effective_owner_metadata,
        "elapsed_seconds": elapsed,
    }