"""FastAPI sub-app for domain term update endpoint (Airflow 3.x plugin).

Mounted on the Airflow api-server via AirflowPlugin.fastapi_apps at
url_prefix="/api/v1/domain-terms". Full path:
  PATCH /api/v1/domain-terms/{term_id}

NOTE: This module runs inside the Airflow api-server process.
Do NOT import from src.*.
"""

import logging
import time
from uuid import uuid4

from fastapi import FastAPI, Request
from pydantic import BaseModel, Field, model_validator
from sqlalchemy import text

from api.config import get_docstore_namespace
from api.db import get_engine
from api.metrics import domain_term_endpoint_duration_seconds, domain_term_op_total, record_outcome
from api.middleware import (
    CorrelationIdMiddleware,
    HMACVerificationMiddleware,
    RequestIdMiddleware,
)
from api.milvus import batch_upsert_chunks, query_chunks_with_term, upsert_chunk_field
from api.responses import error_response, success_response
from api.rls import rls_connection

log = logging.getLogger(__name__)

domain_terms_app = FastAPI(title="TextIQ Domain Terms API")

domain_terms_app.add_middleware(HMACVerificationMiddleware)
domain_terms_app.add_middleware(CorrelationIdMiddleware)
domain_terms_app.add_middleware(RequestIdMiddleware)

NAMESPACE = get_docstore_namespace()

# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------


# Human-created terms default to confidence=1.0 per tag-spec precedent
# (_planning/.../extract_tag_function_spec.md: "Human = 1.0").
USER_CREATED_CONFIDENCE = 1.0


class DomainTermCreate(BaseModel):
    term_original: str = Field(..., min_length=1, max_length=255)
    source_language: str = Field(..., min_length=1, max_length=10)
    term_ja: str | None = Field(None, max_length=255)
    term_en: str | None = Field(None, max_length=255)
    term_vi: str | None = Field(None, max_length=255)
    domain_category: str | None = Field(None, min_length=1, max_length=100)
    confidence: float | None = Field(None, ge=0, le=1)


class DomainTermBulkCreate(BaseModel):
    terms: list[DomainTermCreate] = Field(..., min_length=1, max_length=500)
    skip_duplicates: bool = Field(default=False)


class DomainTermUpdate(BaseModel):
    term_original: str | None = Field(None, min_length=1, max_length=255)
    term_ja: str | None = Field(None, max_length=255)
    term_en: str | None = Field(None, max_length=255)
    term_vi: str | None = Field(None, max_length=255)
    domain_category: str | None = Field(None, min_length=1, max_length=100)
    confidence: float | None = Field(None, ge=0, le=1)

    @model_validator(mode="after")
    def at_least_one_field(self):
        if not any([
            self.term_original, self.term_ja, self.term_en, self.term_vi,
            self.domain_category, self.confidence is not None,
        ]):
            raise ValueError("At least one field must be provided")
        return self


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


class _EarlyReturn(Exception):
    """Signal to exit the rls_connection block and return an error response."""

    def __init__(self, response):
        self.response = response


def _check_global_uniqueness(term_original: str, term_id: str) -> None:
    """Check term_original uniqueness globally (outside RLS scope).

    The unique constraint on term_original is global across tenants.
    RLS-scoped queries would miss cross-tenant conflicts.
    """
    engine = get_engine()
    conn = engine.connect()
    try:
        conflict = conn.execute(
            text(
                "SELECT id FROM domain_term_dictionary "
                "WHERE term_original = :term AND id != :tid"
            ),
            {"term": term_original, "tid": term_id},
        ).fetchone()
        if conflict is not None:
            raise _EarlyReturn(
                error_response("CONFLICT", f"Term '{term_original}' already exists", status_code=409)
            )
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


@domain_terms_app.post("/")
def create_domain_term(body: DomainTermCreate, request: Request):
    """Create a single domain term."""
    ctx = request.state.hmac_ctx
    tenant_id = ctx["tenant_id"]
    node_id = ctx["node_id"]
    _t0 = time.perf_counter()
    _op = "create"

    try:
        with rls_connection(tenant_id) as conn:
            term_id = str(uuid4())
            domain_category = body.domain_category or "general"
            confidence = body.confidence if body.confidence is not None else USER_CREATED_CONFIDENCE
            result = conn.execute(
                text(
                    "INSERT INTO domain_term_dictionary "
                    "(id, term_original, source_language, term_ja, term_en, term_vi, "
                    "domain_category, confidence, tenant_id, node_id) "
                    "VALUES (:id, :term_original, :source_language, :term_ja, :term_en, "
                    ":term_vi, :domain_category, :confidence, :tenant_id, :node_id) "
                    "ON CONFLICT (term_original, node_id) DO NOTHING"
                ),
                {
                    "id": term_id,
                    "term_original": body.term_original,
                    "source_language": body.source_language,
                    "term_ja": body.term_ja,
                    "term_en": body.term_en,
                    "term_vi": body.term_vi,
                    "domain_category": domain_category,
                    "confidence": confidence,
                    "tenant_id": tenant_id,
                    "node_id": node_id,
                },
            )

            if result.rowcount == 0:
                raise _EarlyReturn(
                    error_response(
                        "DUPLICATE_TERM",
                        f"Term '{body.term_original}' already exists",
                        status_code=409,
                    )
                )

        resp = success_response(
            {
                "term_id": term_id,
                "term_original": body.term_original,
                "source_language": body.source_language,
                "term_ja": body.term_ja,
                "term_en": body.term_en,
                "term_vi": body.term_vi,
                "domain_category": domain_category,
                "confidence": confidence,
            },
            status_code=201,
        )
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp

    except _EarlyReturn as e:
        record_outcome(domain_term_op_total, e.response.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return e.response
    except Exception as e:
        log.exception("Domain term creation failed: %s", e)
        resp = error_response("INTERNAL_ERROR", "Domain term creation failed", status_code=500)
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp


@domain_terms_app.post("/bulk")
def create_domain_terms_bulk(body: DomainTermBulkCreate, request: Request):
    """Bulk create domain terms."""
    ctx = request.state.hmac_ctx
    tenant_id = ctx["tenant_id"]
    node_id = ctx["node_id"]
    _t0 = time.perf_counter()
    _op = "bulk_create"

    try:
        # Deduplicate within the batch itself
        seen = set()
        unique_terms = []
        for term in body.terms:
            if term.term_original not in seen:
                seen.add(term.term_original)
                unique_terms.append(term)

        with rls_connection(tenant_id) as conn:
            results = []
            for term in unique_terms:
                term_id = str(uuid4())
                domain_category = term.domain_category or "general"
                confidence = term.confidence if term.confidence is not None else USER_CREATED_CONFIDENCE
                result = conn.execute(
                    text(
                        "INSERT INTO domain_term_dictionary "
                        "(id, term_original, source_language, term_ja, term_en, term_vi, "
                        "domain_category, confidence, tenant_id, node_id) "
                        "VALUES (:id, :term_original, :source_language, :term_ja, :term_en, "
                        ":term_vi, :domain_category, :confidence, :tenant_id, :node_id) "
                        "ON CONFLICT (term_original, node_id) DO NOTHING"
                    ),
                    {
                        "id": term_id,
                        "term_original": term.term_original,
                        "source_language": term.source_language,
                        "term_ja": term.term_ja,
                        "term_en": term.term_en,
                        "term_vi": term.term_vi,
                        "domain_category": domain_category,
                        "confidence": confidence,
                        "tenant_id": tenant_id,
                        "node_id": node_id,
                    },
                )
                if result.rowcount > 0:
                    results.append(
                        {"term_id": term_id, "term_original": term.term_original, "confidence": confidence, "status": "created"}
                    )
                else:
                    results.append(
                        {"term_id": None, "term_original": term.term_original, "status": "skipped"}
                    )

            # Reject entire batch if duplicates found and skip_duplicates is off
            skipped = [r for r in results if r["status"] == "skipped"]
            if skipped and not body.skip_duplicates:
                raise _EarlyReturn(
                    error_response(
                        "DUPLICATE_TERMS",
                        f"Duplicate terms found: {', '.join(r['term_original'] for r in skipped)}",
                        status_code=409,
                    )
                )

        created_count = sum(1 for r in results if r["status"] == "created")
        skipped_count = sum(1 for r in results if r["status"] == "skipped")

        resp = success_response(
            {
                "total_created": created_count,
                "total_skipped": skipped_count,
                "total_failed": 0,
                "results": results,
            },
            status_code=201,
        )
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp

    except _EarlyReturn as e:
        record_outcome(domain_term_op_total, e.response.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return e.response
    except Exception as e:
        log.exception("Bulk domain term creation failed: %s", e)
        resp = error_response("INTERNAL_ERROR", "Bulk domain term creation failed", status_code=500)
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp


@domain_terms_app.delete("/{term_id}", status_code=204)
def delete_domain_term(term_id: str, request: Request):
    """Delete a domain term and cascade-remove it from chunk metadata."""
    ctx = request.state.hmac_ctx
    tenant_id = ctx["tenant_id"]
    _t0 = time.perf_counter()
    _op = "delete"

    try:
        with rls_connection(tenant_id) as conn:
            row = conn.execute(
                text("SELECT id, term_original FROM domain_term_dictionary WHERE id = :tid"),
                {"tid": term_id},
            ).fetchone()
            if row is None:
                raise _EarlyReturn(
                    error_response("NOT_FOUND", "Domain term not found", status_code=404)
                )

            term_original = row[1]

            # Cascade: remove term from PG chunk metadata
            conn.execute(
                text(
                    "UPDATE data_hierarchical_nodes "
                    "SET value = jsonb_set(value, '{__data__,metadata,domain_terms}', "
                    "  (SELECT COALESCE(jsonb_agg(elem), '[]'::jsonb) "
                    "   FROM jsonb_array_elements(value->'__data__'->'metadata'->'domain_terms') AS elem "
                    "   WHERE elem #>> '{}' != :term)) "
                    "WHERE namespace = :ns "
                    "  AND value->'__data__'->'metadata'->'domain_terms' @> to_jsonb(CAST(:term AS text))"
                ),
                {"term": term_original, "ns": NAMESPACE},
            )

            # Cascade: remove term from Milvus chunk metadata
            affected_chunks = query_chunks_with_term(term_original, tenant_id)
            if affected_chunks:
                batch_data = []
                for chunk in affected_chunks:
                    domain_terms = chunk.get("domain_terms") or []
                    updated_terms = [t for t in domain_terms if t != term_original]
                    batch_data.append({"id": chunk["id"], "domain_terms": updated_terms})
                batch_upsert_chunks(batch_data)

            # Delete the dictionary row
            conn.execute(
                text("DELETE FROM domain_term_dictionary WHERE id = :tid"),
                {"tid": term_id},
            )

        record_outcome(domain_term_op_total, 204, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)

    except _EarlyReturn as e:
        record_outcome(domain_term_op_total, e.response.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return e.response
    except Exception as e:
        log.exception("Domain term deletion failed: %s", e)
        resp = error_response("INTERNAL_ERROR", "Domain term deletion failed", status_code=500)
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp


@domain_terms_app.patch("/{term_id}")
def update_domain_term(term_id: str, body: DomainTermUpdate, request: Request):
    """Update a domain term. If term_original changes, cascade to all chunks."""
    ctx = request.state.hmac_ctx
    tenant_id = ctx["tenant_id"]
    _t0 = time.perf_counter()
    _op = "update"

    try:
        with rls_connection(tenant_id) as conn:
            # Fetch existing term
            row = conn.execute(
                text("SELECT id, term_original FROM domain_term_dictionary WHERE id = :tid"),
                {"tid": term_id},
            ).fetchone()
            if row is None:
                raise _EarlyReturn(
                    error_response("NOT_FOUND", "Domain term not found", status_code=404)
                )

            old_term_original = row[1]
            term_original_changed = body.term_original is not None and body.term_original != old_term_original

            # Check global uniqueness if term_original is being changed
            # (outside RLS scope so cross-tenant conflicts are detected)
            if term_original_changed:
                _check_global_uniqueness(body.term_original, term_id)

            # Build dynamic UPDATE
            set_clauses = []
            params = {"tid": term_id}
            for field_name in ["term_original", "term_ja", "term_en", "term_vi", "domain_category", "confidence"]:
                value = getattr(body, field_name)
                if value is not None:
                    set_clauses.append(f"{field_name} = :{field_name}")
                    params[field_name] = value

            if set_clauses:
                conn.execute(
                    text(f"UPDATE domain_term_dictionary SET {', '.join(set_clauses)} WHERE id = :tid"),
                    params,
                )

            # Cascade term_original rename to chunks in PG and Milvus
            if term_original_changed:
                new_term = body.term_original

                # Update all chunks in PG that contain the old term
                conn.execute(
                    text(
                        "UPDATE data_hierarchical_nodes "
                        "SET value = jsonb_set(value, '{__data__,metadata,domain_terms}', "
                        "  (SELECT jsonb_agg("
                        "    CASE WHEN elem #>> '{}' = :old_term THEN to_jsonb(CAST(:new_term AS text)) ELSE elem END"
                        "  ) FROM jsonb_array_elements(value->'__data__'->'metadata'->'domain_terms') AS elem)) "
                        "WHERE namespace = :ns "
                        "  AND value->'__data__'->'metadata'->'domain_terms' @> to_jsonb(CAST(:old_term AS text))"
                    ),
                    {"old_term": old_term_original, "new_term": new_term, "ns": NAMESPACE},
                )

                # Cascade to Milvus: find affected chunks and batch update
                affected_chunks = query_chunks_with_term(old_term_original, tenant_id)
                if affected_chunks:
                    batch_data = []
                    for chunk in affected_chunks:
                        domain_terms = chunk.get("domain_terms") or []
                        updated_terms = [new_term if t == old_term_original else t for t in domain_terms]
                        batch_data.append({"id": chunk["id"], "domain_terms": updated_terms})
                    batch_upsert_chunks(batch_data)

        resp = success_response({
            "term_id": term_id,
            "updated_fields": [
                f for f in ["term_original", "term_ja", "term_en", "term_vi", "domain_category", "confidence"]
                if getattr(body, f) is not None
            ],
            "cascaded": term_original_changed,
        })
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp

    except _EarlyReturn as e:
        record_outcome(domain_term_op_total, e.response.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return e.response
    except Exception as e:
        log.exception("Domain term update failed: %s", e)
        resp = error_response("INTERNAL_ERROR", "Domain term update failed", status_code=500)
        record_outcome(domain_term_op_total, resp.status_code, operation=_op, tenant_id=tenant_id)
        domain_term_endpoint_duration_seconds.labels(operation=_op).observe(time.perf_counter() - _t0)
        return resp