bmc_hub/app/modules/migration_center/backend/router.py

"""HTTP API for the manual migration centre."""
from __future__ import annotations

import json
import re
import hashlib
import math
from html.parser import HTMLParser
from datetime import date, datetime, time as datetime_time, timedelta
from decimal import Decimal
from typing import Any, Dict, Optional

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import Response
from pydantic import BaseModel, Field, field_validator
from psycopg2.extras import Json, RealDictCursor

from app.core.auth_dependencies import require_any_permission, require_permission
from app.core.config import settings
from app.core.database import execute_query, execute_query_single, get_db_connection, release_db_connection
from app.modules.migration_center.backend.service import (
    MUTABLE_LOCK_STATES,
    audit,
    create_session,
    ensure_writable,
    json_value,
    match_item,
    preflight_token,
    report_csv,
    user_id,
    verify_preflight,
    snapshot_hash,
    attach_economic_snapshot,
    EconomicSnapshotRepository,
    subscription_like_item_sql,
    refresh_subscription_relevance,
)
from app.services.simplycrm_service import SimplyCRMService
from app.services.vtiger_service import get_vtiger_service

router = APIRouter()

EXCLUDED_INVOICE_LINE_SQL = """
NOT (
  source_system='economic' AND (
    LOWER(COALESCE(product_name,'')) LIKE '%%gebyr%%'
    OR LOWER(COALESCE(product_name,'')) LIKE '%%fragt%%'
    OR LOWER(COALESCE(product_name,'')) LIKE '%%porto%%'
  )
)
"""


class SessionCreate(BaseModel):
    name: str = Field(min_length=2, max_length=160)


class SessionUpdate(BaseModel):
    status: Optional[str] = None
    read_only: Optional[bool] = None


class CustomerCreate(BaseModel):
    name: str = Field(min_length=1, max_length=255)
    cvr_number: Optional[str] = None
    email: Optional[str] = None
    customer_no: Optional[str] = None


class CaseCreate(BaseModel):
    title: str = Field(min_length=1, max_length=255)
    description: str = ""


class LinkPayload(BaseModel):
    hub_id: int


class ReasonPayload(BaseModel):
    reason: str = Field(min_length=2, max_length=2000)


class NotePayload(BaseModel):
    note: str = Field(max_length=5000)


class CreatePayload(BaseModel):
    preflight_token: str
    idempotency_key: str = Field(min_length=8, max_length=120)


class VtigerCaseImportRequest(BaseModel):
    ticket_number: str = Field(min_length=3, max_length=80)

    @field_validator("ticket_number")
    @classmethod
    def validate_ticket_number(cls, value: str) -> str:
        normalized = str(value or "").strip().upper()
        if not re.fullmatch(r"CC[0-9]{1,12}", normalized):
            raise ValueError("Indtast et sagsnummer i formatet CC12345")
        return normalized


class HubCaseExportRequest(BaseModel):
    sag_id: int = Field(gt=0)


class HubCaseExportConfirmRequest(HubCaseExportRequest):
    source_hash: str = Field(min_length=64, max_length=64)


def _vtiger_reference_id(value: Any) -> Optional[str]:
    if isinstance(value, dict):
        value = value.get("value") or value.get("id")
    value = str(value or "").strip()
    return value if re.fullmatch(r"[A-Za-z0-9_-]+x[0-9]+", value) else None


def _vtiger_case_number(raw: Dict[str, Any]) -> str:
    for key in ("ticket_no", "ticketnumber", "ticket_number", "case_no", "casenumber"):
        value = str(raw.get(key) or "").strip()
        if value:
            return value.upper()
    return ""


def _case_status_from_vtiger(value: Any) -> str:
    status = _vtiger_text(value).lower()
    if status in {"closed", "lukket", "afsluttet", "done", "cancelled", "canceled"}:
        return "Lukket"
    if status in {"resolved", "løst", "lost"}:
        return "Løst"
    if status in {"assigned", "tildelt"}:
        return "Tildelt"
    if status in {"in progress", "in_progress", "igangværende", "open", "åben", "reopened"}:
        return "Igangværende" if status not in {"open", "åben"} else "Ny"
    if status in {"waiting", "waiting_customer", "waiting_internal", "on_hold", "afventer"}:
        return "Afventer"
    return "Ny"


def _case_priority_from_vtiger(value: Any) -> str:
    priority = _vtiger_text(value).lower()
    if priority in {"urgent", "critical", "akut"}:
        return "urgent"
    if priority in {"high", "høj"}:
        return "high"
    if priority in {"low", "lav"}:
        return "low"
    return "normal"


def _vtiger_text(value: Any) -> str:
    if isinstance(value, dict):
        value = value.get("label") or value.get("value") or ""
    return str(value or "").strip()


class _VTigerPlainTextParser(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.parts = []

    def handle_starttag(self, tag, attrs):
        if tag.lower() in {"br", "p", "div", "tr", "h1", "h2", "h3", "h4", "h5", "h6"}:
            self.parts.append("\n")

    def handle_endtag(self, tag):
        if tag.lower() in {"p", "div", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6"}:
            self.parts.append("\n")

    def handle_data(self, data):
        self.parts.append(data)


def _vtiger_plain_text(value: Any) -> str:
    raw = _vtiger_text(value)
    if not raw:
        return ""
    parser = _VTigerPlainTextParser()
    parser.feed(raw)
    text = re.sub(r"\n[ \t]*\n+", "\n\n", "".join(parser.parts))
    return text.strip()


def _vtiger_timestamp(value: Any) -> Optional[datetime]:
    if not value:
        return None
    text = str(value).strip().replace("T", " ")
    try:
        return datetime.fromisoformat(text[:19])
    except ValueError:
        return None


def _timelog_source_id(row: Dict[str, Any]) -> str:
    return str(row.get("id") or "").strip()


def _timelog_description(row: Dict[str, Any], *, legacy: bool = False) -> str:
    # Older imports preferred the generic record name (often "migration").
    keys = ("name", "cf_timelog_description", "description") if legacy else (
        "description", "cf_timelog_description", "name"
    )
    for key in keys:
        value = _vtiger_plain_text(row.get(key))
        if value:
            return value
    return "Importerede vTiger-timer"


async def _resolve_vtiger_assignee_names(timelogs: List[Dict[str, Any]]) -> Dict[str, str]:
    """Resolve time-entry owners in two batch queries instead of one call per row."""
    refs = sorted({
        ref for row in timelogs
        if (ref := _vtiger_reference_id(row.get("assigned_user_id")))
    })
    if not refs:
        return {}
    id_list = "', '".join(refs)
    vtiger = get_vtiger_service()
    names: Dict[str, str] = {}
    users = await vtiger.query(
        "SELECT id, first_name, last_name, user_name FROM Users "
        f"WHERE id IN ('{id_list}');"
    )
    for row in users or []:
        ref = _vtiger_reference_id(row.get("id"))
        if not ref:
            continue
        full_name = " ".join(
            part for part in (_vtiger_text(row.get("first_name")), _vtiger_text(row.get("last_name"))) if part
        ).strip()
        names[ref] = full_name or _vtiger_text(row.get("user_name")) or ref
    groups = await vtiger.query(
        "SELECT id, groupname FROM Groups "
        f"WHERE id IN ('{id_list}');"
    )
    for row in groups or []:
        ref = _vtiger_reference_id(row.get("id"))
        if ref and ref not in names:
            names[ref] = _vtiger_text(row.get("groupname")) or ref
    return names


async def _load_vtiger_case_bundle(ticket_number: str) -> Dict[str, Any]:
    normalized = str(ticket_number or "").strip().upper()
    if not re.fullmatch(r"CC[0-9]{1,12}", normalized):
        raise HTTPException(status_code=400, detail="Indtast et sagsnummer i formatet CC12345")
    vtiger = get_vtiger_service()
    case = None
    for field in ("ticket_no", "ticketnumber", "ticket_number", "case_no"):
        safe_value = normalized.replace("'", "''")
        records = await vtiger.query(f"SELECT * FROM Cases WHERE {field}='{safe_value}' LIMIT 2;")
        if records:
            if len(records) > 1:
                raise HTTPException(status_code=409, detail=f"Flere vTiger-sager matcher {normalized}; overførslen er stoppet")
            case = records[0]
            break
    if not case:
        raise HTTPException(status_code=404, detail=f"Sagen {normalized} blev ikke fundet i vTiger")

    vtiger_case_id = _vtiger_reference_id(case.get("id"))
    if not vtiger_case_id:
        raise HTTPException(status_code=502, detail="vTiger-sagen mangler et gyldigt post-id")
    actual_number = _vtiger_case_number(case) or normalized
    if actual_number != normalized:
        raise HTTPException(status_code=409, detail=f"vTiger returnerede {actual_number} i stedet for {normalized}; overførslen er stoppet")
    account_id = _vtiger_reference_id(case.get("parent_id") or case.get("account_id"))
    if not account_id:
        raise HTTPException(status_code=409, detail="Sagen har ingen entydig vTiger-kundetilknytning")

    customer = execute_query_single(
        "SELECT id, name, vtiger_id FROM customers WHERE vtiger_id = %s AND deleted_at IS NULL",
        (account_id,),
    )
    if not customer:
        account_rows = await vtiger.query(f"SELECT * FROM Accounts WHERE id='{account_id}' LIMIT 1;")
        account = account_rows[0] if account_rows else {}
        account_name = str(account.get("accountname") or "").strip()
        if account_name:
            matches = execute_query(
                "SELECT id, name, vtiger_id FROM customers WHERE LOWER(TRIM(name)) = LOWER(TRIM(%s)) AND deleted_at IS NULL ORDER BY id LIMIT 2",
                (account_name,),
            ) or []
            if len(matches) == 1:
                customer = matches[0]
    if not customer:
        raise HTTPException(
            status_code=409,
            detail=f"Kunden {account_id} er ikke entydigt tilknyttet en Hub-kunde. Link kunden først.",
        )

    timelogs = await vtiger.query(
        f"SELECT * FROM Timelog WHERE relatedto='{vtiger_case_id}' ORDER BY startedon LIMIT 1001;"
    )
    comments = await vtiger.query(
        f"SELECT * FROM ModComments WHERE related_to='{vtiger_case_id}' ORDER BY createdtime LIMIT 1000;"
    )
    # vTiger installations can ignore filters on reference fields; enforce the
    # relationship locally before any source records are shown or imported.
    timelogs = [row for row in (timelogs or []) if _vtiger_reference_id(row.get("relatedto")) == vtiger_case_id]
    comments = [row for row in (comments or []) if _vtiger_reference_id(row.get("related_to")) == vtiger_case_id]
    if len(timelogs) > 1000:
        raise HTTPException(status_code=409, detail="Sagen har mere end 1.000 tidsregistreringer. Overførslen er stoppet for at undgå en ufuldstændig kopi.")
    assignee_names = await _resolve_vtiger_assignee_names(timelogs)
    contact_id = _vtiger_reference_id(case.get("contact_id") or case.get("contactid"))
    contact = execute_query_single(
        "SELECT id, first_name, last_name FROM contacts WHERE vtiger_id = %s",
        (contact_id,),
    ) if contact_id else None
    imported = execute_query_single(
        "SELECT hub_sag_id, vtiger_marked_at FROM vtiger_case_hub_imports WHERE vtiger_case_id = %s",
        (vtiger_case_id,),
    )
    return {
        "case": case,
        "ticket_number": actual_number,
        "vtiger_case_id": vtiger_case_id,
        "account_id": account_id,
        "customer": customer,
        "contact": contact,
        "timelogs": timelogs or [],
        "timelog_assignee_names": assignee_names,
        "comments": comments or [],
        "existing_import": imported,
    }


def _import_vtiger_case_bundle(bundle: Dict[str, Any], current_user: dict) -> Dict[str, Any]:
    """Create a native Hub case and link source time entries atomically."""
    conn = get_db_connection()
    imported_now = False
    vtiger_case_id = bundle["vtiger_case_id"]
    case = bundle["case"]
    customer = bundle["customer"]
    current_user_id = user_id(current_user)
    imported_time_entries = 0
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute(
                "SELECT hub_sag_id, vtiger_marked_at FROM vtiger_case_hub_imports WHERE vtiger_case_id=%s FOR UPDATE",
                (vtiger_case_id,),
            )
            mapped = cursor.fetchone()
            if mapped:
                sag_id = mapped["hub_sag_id"]
                # Repair legacy imports without changing approval or billing state.
                assignee_names = bundle.get("timelog_assignee_names", {})
                for timelog in bundle.get("timelogs", []):
                    source_id = _timelog_source_id(timelog)
                    assignee_ref = _vtiger_reference_id(timelog.get("assigned_user_id"))
                    old_user_name = _vtiger_text(timelog.get("assigned_user_id"))
                    resolved_user_name = assignee_names.get(assignee_ref)
                    if not source_id:
                        continue
                    if assignee_ref and resolved_user_name and resolved_user_name != old_user_name:
                        cursor.execute(
                            """UPDATE tmodule_times SET user_name=%s
                               WHERE vtiger_id=%s AND sag_id=%s AND user_name=%s""",
                            (resolved_user_name, source_id, sag_id, old_user_name),
                        )
                    old_description = _timelog_description(timelog, legacy=True)
                    new_description = _timelog_description(timelog)
                    if old_description != new_description:
                        cursor.execute(
                            """UPDATE tmodule_times SET description=%s
                               WHERE vtiger_id=%s AND sag_id=%s AND description=%s""",
                            (new_description, source_id, sag_id, old_description),
                        )
            else:
                cursor.execute(
                    """INSERT INTO sag_sager
                       (titel, beskrivelse, template_key, type, status, priority, customer_id, created_by_user_id)
                       VALUES (%s,%s,'ticket','ticket',%s,%s,%s,%s) RETURNING id""",
                    (
                        _vtiger_text(case.get("ticket_title") or case.get("subject")) or bundle["ticket_number"],
                        _vtiger_text(case.get("description")),
                        _case_status_from_vtiger(case.get("status")),
                        _case_priority_from_vtiger(case.get("priority")),
                        customer["id"], current_user_id,
                    ),
                )
                sag_id = cursor.fetchone()["id"]

                contact = bundle.get("contact")
                if contact:
                    cursor.execute(
                        "INSERT INTO sag_kontakter (sag_id, contact_id, role) VALUES (%s,%s,'Primær kontakt')",
                        (sag_id, contact["id"]),
                    )

                cursor.execute(
                    """INSERT INTO tmodule_customers (vtiger_id, name, hub_customer_id, vtiger_data)
                       VALUES (%s,%s,%s,%s) ON CONFLICT (vtiger_id) DO UPDATE
                       SET hub_customer_id=COALESCE(tmodule_customers.hub_customer_id, EXCLUDED.hub_customer_id)
                       RETURNING id""",
                    (bundle["account_id"], customer["name"], customer["id"], Json({"id": bundle["account_id"]})),
                )
                tcustomer_id = cursor.fetchone()["id"]
                cursor.execute(
                    """INSERT INTO tmodule_cases
                       (vtiger_id, customer_id, title, description, status, priority, module_type, vtiger_data, sync_hash)
                       VALUES (%s,%s,%s,%s,%s,%s,'Cases',%s,%s)
                       ON CONFLICT (vtiger_id) DO UPDATE SET
                         customer_id=EXCLUDED.customer_id, title=EXCLUDED.title,
                         description=EXCLUDED.description, status=EXCLUDED.status,
                         priority=EXCLUDED.priority, vtiger_data=EXCLUDED.vtiger_data,
                         sync_hash=EXCLUDED.sync_hash, last_synced_at=CURRENT_TIMESTAMP
                       RETURNING id""",
                    (
                        vtiger_case_id, tcustomer_id,
                        _vtiger_text(case.get("ticket_title") or case.get("subject")) or bundle["ticket_number"],
                        _vtiger_text(case.get("description")), _vtiger_text(case.get("status")),
                        _vtiger_text(case.get("priority")), Json(case),
                        hashlib.sha256(json.dumps(case, sort_keys=True, default=str).encode()).hexdigest(),
                    ),
                )
                tcase_id = cursor.fetchone()["id"]

                for timelog in bundle.get("timelogs", []):
                    source_id = _timelog_source_id(timelog)
                    seconds = _amount(timelog.get("duration"))
                    if not source_id or seconds <= 0:
                        continue
                    hours = (seconds / Decimal("3600")).quantize(Decimal("0.01"))
                    if hours <= 0:
                        continue
                    billable = str(timelog.get("isbillable") or "0") == "1"
                    already_billed = any(str(timelog.get(key) or "0") not in {"", "0"} for key in (
                        "cf_timelog_invoiced", "billed_via_thehub_id", "cf_timelog_billedviathehubid",
                        "cf_timelog_rounduptimespent",
                    ))
                    start = _vtiger_timestamp(timelog.get("startedon") or timelog.get("createdtime"))
                    data_hash = hashlib.sha256(json.dumps(timelog, sort_keys=True, default=str).encode()).hexdigest()
                    cursor.execute(
                        """INSERT INTO tmodule_times
                           (vtiger_id, case_id, customer_id, sag_id, description, original_hours,
                            worked_date, user_name, billable, vtiger_data, sync_hash, status,
                            approved_hours, billing_method, faktisk_tid_min, fakturerbar_tid_min,
                            start_tid, entry_status, kilde, entry_type, work_type)
                           VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'api','ukendt','support')
                           ON CONFLICT (vtiger_id) DO UPDATE SET sag_id=EXCLUDED.sag_id
                           WHERE tmodule_times.case_id=EXCLUDED.case_id
                             AND (tmodule_times.sag_id IS NULL OR tmodule_times.sag_id=EXCLUDED.sag_id)
                           RETURNING id""",
                        (
                            source_id, tcase_id, tcustomer_id, sag_id,
                            _timelog_description(timelog),
                            hours, start.date() if start else _safe_date(timelog.get("startedon")),
                            bundle.get("timelog_assignee_names", {}).get(
                                _vtiger_reference_id(timelog.get("assigned_user_id")),
                                _vtiger_text(timelog.get("assigned_user_id")),
                            ), billable, Json(timelog), data_hash,
                            "billed" if already_billed else "pending", hours if already_billed else None,
                            "invoice" if billable else "internal", int((seconds / Decimal("60")).to_integral_value()),
                            int((seconds / Decimal("60")).to_integral_value()) if billable else 0,
                            start, "godkendt" if already_billed else "afventer",
                        ),
                    )
                    if cursor.fetchone():
                        imported_time_entries += 1

                cursor.execute(
                    """INSERT INTO vtiger_case_hub_imports
                       (vtiger_case_id,ticket_number,hub_sag_id,hub_customer_id,imported_by_user_id,source_snapshot)
                       VALUES (%s,%s,%s,%s,%s,%s)""",
                    (vtiger_case_id, bundle["ticket_number"], sag_id, customer["id"], current_user_id,
                    Json({"case": case, "comments_not_imported": len(bundle.get("comments", []))})),
                )
                imported_now = True
            conn.commit()
        return {"hub_sag_id": sag_id, "imported": imported_now,
                "time_entries": imported_time_entries,
                "comments_available_not_imported": len(bundle.get("comments", [])) if imported_now else 0}
    except Exception:
        conn.rollback()
        raise
    finally:
        release_db_connection(conn)


def _safe_date(value: Any) -> Optional[date]:
    if not value:
        return None
    if isinstance(value, date):
        return value
    try:
        return date.fromisoformat(str(value)[:10])
    except ValueError:
        return None


def _amount(value: Any) -> Decimal:
    try:
        return Decimal(str(value or 0).replace(",", "."))
    except Exception:
        return Decimal("0")


def _hub_interval(value: Any) -> str:
    normalized = str(value or "").strip().lower()
    if normalized in {"daily", "biweekly", "monthly", "quarterly", "yearly"}:
        return normalized
    if "quarter" in normalized or "kvart" in normalized or normalized.startswith("3_month"):
        return "quarterly"
    if "year" in normalized or "annual" in normalized or "årlig" in normalized:
        return "yearly"
    if "week" in normalized or "uge" in normalized:
        return "biweekly"
    return "monthly"


def _normalized_crm_record(source: str, raw: Dict[str, Any], service=None) -> Dict[str, Any]:
    if source == "simply" and service:
        data = service.extract_subscription_data(raw)
        record_id = str(data.get("simplycrm_id") or raw.get("id") or "")
        customer_id = str(data.get("account_id") or "")
        product_name = data.get("name") or raw.get("subject") or "Simply abonnement"
        amount = data.get("total_amount") or data.get("subtotal") or 0
        frequency = data.get("billing_frequency")
        start = _safe_date(data.get("start_date"))
        end = _safe_date(data.get("end_date"))
        active = data.get("status") != "cancelled"
    else:
        record_id = str(raw.get("id") or raw.get("subscriptionid") or "")
        customer_id = str(raw.get("account_id") or raw.get("accountid") or "")
        product_name = raw.get("subject") or raw.get("subscriptionname") or "Vtiger abonnement"
        amount = raw.get("total") or raw.get("hdnGrandTotal") or raw.get("amount") or 0
        frequency = str(raw.get("generateinvoiceevery") or raw.get("frequency") or "monthly").lower()
        start = _safe_date(raw.get("startdate") or raw.get("start_period"))
        end = _safe_date(raw.get("enddate") or raw.get("end_period"))
        active = str(raw.get("subscriptionstatus") or "active").lower() not in {"cancelled", "inactive", "expired"}
    normalized = {
        "entity_type": "subscription", "source_system": source, "source_record_id": record_id,
        "source_customer_id": customer_id, "customer_no": raw.get("customer_no") or raw.get("account_no"),
        "customer_name": raw.get("accountname") or raw.get("customer_name") or customer_id,
        "product_code": raw.get("product_code") or raw.get("productid"),
        "product_name": product_name, "amount": _amount(amount), "quantity": _amount(raw.get("quantity") or 1),
        "billing_frequency": frequency, "period_from": start, "period_to": end,
        "invoice_no": None, "invoice_date": None, "source_payload": raw, "active": active,
    }
    normalized["source_hash"] = snapshot_hash(normalized)
    return normalized


def _store_crm_snapshot(session_id: int, item: Dict[str, Any]) -> str:
    if not item["source_record_id"]:
        return "skipped"
    if item.get("source_customer_id"):
        customer_payload = {
            "source_customer_id": item["source_customer_id"],
            "customer_no": item.get("customer_no"),
            "customer_name": item.get("customer_name") or item["source_customer_id"],
        }
        execute_query(
            """
            INSERT INTO migration_center_source_customers
            (source_system, source_customer_id, customer_no, customer_name, raw_payload, snapshot_hash)
            VALUES (%s,%s,%s,%s,%s,%s)
            ON CONFLICT (source_system, source_customer_id) DO UPDATE SET
              customer_no=EXCLUDED.customer_no, customer_name=EXCLUDED.customer_name,
              raw_payload=EXCLUDED.raw_payload, snapshot_hash=EXCLUDED.snapshot_hash,
              updated_at=CURRENT_TIMESTAMP
            """,
            (
                item["source_system"], item["source_customer_id"], item.get("customer_no"),
                customer_payload["customer_name"], Json(customer_payload), snapshot_hash(customer_payload),
            ), fetch=False,
        )
    existing = execute_query_single(
        """
        SELECT id, source_hash, source_payload, lock_status
        FROM migration_center_session_items
        WHERE session_id=%s AND entity_type='subscription' AND source_system=%s AND source_record_id=%s
        """,
        (session_id, item["source_system"], item["source_record_id"]),
    )
    execute_query(
        """
        INSERT INTO migration_center_source_subscriptions
        (source_system, source_record_id, source_customer_id, customer_no, customer_name,
         product_code, product_name, amount, quantity, billing_frequency, start_date, end_date,
         active, raw_payload, snapshot_hash)
        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
        ON CONFLICT (source_system, source_record_id) DO UPDATE SET
          source_customer_id=EXCLUDED.source_customer_id, customer_no=EXCLUDED.customer_no,
          customer_name=EXCLUDED.customer_name, product_code=EXCLUDED.product_code,
          product_name=EXCLUDED.product_name, amount=EXCLUDED.amount, quantity=EXCLUDED.quantity,
          billing_frequency=EXCLUDED.billing_frequency, start_date=EXCLUDED.start_date,
          end_date=EXCLUDED.end_date, active=EXCLUDED.active, raw_payload=EXCLUDED.raw_payload,
          snapshot_hash=EXCLUDED.snapshot_hash, updated_at=CURRENT_TIMESTAMP
        """,
        (
            item["source_system"], item["source_record_id"], item["source_customer_id"], item["customer_no"],
            item["customer_name"], item["product_code"], item["product_name"], item["amount"],
            item["quantity"], item["billing_frequency"], item["period_from"], item["period_to"],
            item["active"], Json(json_value(item["source_payload"])), item["source_hash"],
        ), fetch=False,
    )
    if not existing:
        execute_query(
            """
            INSERT INTO migration_center_session_items
            (session_id, entity_type, source_system, source_record_id, source_customer_id,
             customer_no, customer_name, product_code, product_name, amount, quantity,
             billing_frequency, period_from, period_to, source_payload, source_hash)
            VALUES (%s,'subscription',%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """,
            (
                session_id, item["source_system"], item["source_record_id"], item["source_customer_id"],
                item["customer_no"], item["customer_name"], item["product_code"], item["product_name"],
                item["amount"], item["quantity"], item["billing_frequency"], item["period_from"],
                item["period_to"], Json(json_value(item["source_payload"])), item["source_hash"],
            ), fetch=False,
        )
        return "created"
    if existing["source_hash"] != item["source_hash"] and existing["lock_status"] != "locked":
        execute_query(
            """
            UPDATE migration_center_session_items SET
              previous_source_payload=source_payload, source_payload=%s, source_hash=%s,
              customer_no=%s, customer_name=%s, product_code=%s, product_name=%s, amount=%s,
              quantity=%s, billing_frequency=%s, period_from=%s, period_to=%s,
              match_status='source_changed', approval_status='pending',
              verified_at=NULL, verified_by_user_id=NULL, updated_at=CURRENT_TIMESTAMP
            WHERE id=%s
            """,
            (
                Json(json_value(item["source_payload"])), item["source_hash"], item["customer_no"],
                item["customer_name"], item["product_code"], item["product_name"], item["amount"],
                item["quantity"], item["billing_frequency"], item["period_from"], item["period_to"],
                existing["id"],
            ), fetch=False,
        )
        return "changed"
    return "unchanged"


def _load_item(item_id: int, *, for_update: bool = False, cursor=None) -> Dict[str, Any]:
    suffix = " FOR UPDATE" if for_update else ""
    query = f"SELECT * FROM migration_center_session_items WHERE id = %s{suffix}"
    if cursor:
        cursor.execute(query, (item_id,))
        row = cursor.fetchone()
    else:
        row = execute_query_single(query, (item_id,))
    if not row:
        raise HTTPException(status_code=404, detail="Post blev ikke fundet")
    return dict(row)


def _ensure_item_mutable(item: Dict[str, Any]) -> None:
    ensure_writable(int(item["session_id"]))
    if item["lock_status"] not in MUTABLE_LOCK_STATES:
        raise HTTPException(status_code=423, detail="Posten er låst eller ved at blive låst")


@router.get("/sessions")
async def list_sessions(current_user: dict = Depends(require_permission("migration_center.view"))):
    return execute_query(
        """
        SELECT s.*,
          COUNT(i.id) AS item_count,
          COUNT(i.id) FILTER (WHERE i.approval_status IN ('verified','ignored')) AS treated_count,
          COUNT(i.id) FILTER (WHERE i.lock_status = 'locked') AS locked_count,
          COUNT(i.id) FILTER (WHERE i.match_status IN ('conflict','source_changed')) AS conflict_count
        FROM migration_center_sessions s
        LEFT JOIN migration_center_session_items i ON i.session_id = s.id
        GROUP BY s.id ORDER BY s.created_at DESC
        """
    ) or []


@router.post("/sessions")
async def add_session(
    payload: SessionCreate,
    request: Request,
    current_user: dict = Depends(require_permission("migration_center.sessions")),
):
    return create_session(payload.name, current_user, request)


@router.patch("/sessions/{session_id}")
async def update_session(
    session_id: int,
    payload: SessionUpdate,
    request: Request,
    current_user: dict = Depends(require_permission("migration_center.sessions")),
):
    ensure_writable()
    if payload.status and payload.status not in {"draft", "active", "completed", "archived"}:
        raise HTTPException(status_code=400, detail="Ugyldig sessionsstatus")
    existing = execute_query_single("SELECT * FROM migration_center_sessions WHERE id=%s", (session_id,))
    if not existing:
        raise HTTPException(status_code=404, detail="Kontrolsession blev ikke fundet")
    status = payload.status or existing["status"]
    read_only = payload.read_only if payload.read_only is not None else existing["read_only"]
    if status in {"completed", "archived"}:
        read_only = True
    result = execute_query_single(
        """
        UPDATE migration_center_sessions SET status=%s, read_only=%s, updated_at=CURRENT_TIMESTAMP
        WHERE id=%s RETURNING *
        """,
        (status, read_only, session_id),
    )
    audit(
        request=request, current_user=current_user, action="session_updated", entity_type="session",
        entity_id=session_id, session_id=session_id, old_value=existing, new_value=result,
    )
    return result


@router.post("/sessions/{session_id}/crm-import")
async def import_crm_snapshots(
    session_id: int,
    request: Request,
    source: str = Query(..., pattern="^(vtiger|simply)$"),
    mock: bool = False,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    ensure_writable(session_id)
    if mock:
        records = [{
            "id": f"mock-{source}-1", "account_id": "mock-customer-1",
            "accountname": "Eksempel Kunde A/S", "subject": "Microsoft 365 Business Premium",
            "hdnGrandTotal": "1295.00", "quantity": 5, "recurring_frequency": "Monthly",
            "start_period": date.today().replace(day=1).isoformat(), "enable_recurring": "1",
        }]
        service = SimplyCRMService() if source == "simply" else None
    elif source == "simply":
        async with SimplyCRMService() as service:
            records = await service.fetch_active_subscriptions()
    else:
        service = None
        records = await get_vtiger_service().query("SELECT * FROM Subscription;")
    counts = {"created": 0, "changed": 0, "unchanged": 0, "skipped": 0}
    item_ids = []
    for raw in records:
        item = _normalized_crm_record(source, dict(raw), service)
        outcome = _store_crm_snapshot(session_id, item)
        counts[outcome] += 1
        row = execute_query_single(
            """
            SELECT id FROM migration_center_session_items
            WHERE session_id=%s AND entity_type='subscription' AND source_system=%s AND source_record_id=%s
            """,
            (session_id, source, item["source_record_id"]),
        )
        if row:
            item_ids.append(row["id"])
    for item_id in item_ids:
        row = _load_item(item_id)
        if row["match_status"] != "source_changed":
            match_item(item_id)
    relevance = refresh_subscription_relevance(session_id)
    audit(
        request=request, current_user=current_user, action="crm_imported", entity_type="session",
        entity_id=session_id, session_id=session_id, new_value={"source": source, **counts},
    )
    return {"source": source, "records": len(records), **counts, **relevance}


@router.post("/sessions/{session_id}/economic-snapshot")
async def load_economic_snapshot(
    session_id: int,
    request: Request,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    ensure_writable(session_id)
    run = EconomicSnapshotRepository.latest_run()
    if not run:
        raise HTTPException(
            status_code=409,
            detail="Faktura-fejl-finder har ingen anvendelig import med fakturaer",
        )
    counts = attach_economic_snapshot(session_id, run)
    execute_query(
        """
        UPDATE migration_center_session_items i
        SET hub_customer_id=c.id,
            customer_name=COALESCE(NULLIF(i.customer_name,''),c.name),
            hub_status=CASE WHEN i.hub_status='not_created' THEN 'ready_for_creation' ELSE i.hub_status END,
            match_explanation=CASE
              WHEN NOT (i.match_explanation ? 'e-conomic-kundenummer stemmer')
              THEN i.match_explanation || '["e-conomic-kundenummer stemmer"]'::jsonb
              ELSE i.match_explanation
            END,
            updated_at=CURRENT_TIMESTAMP
        FROM customers c
        WHERE i.session_id=%s AND i.source_system='economic'
          AND i.lock_status IN ('unlocked','lock_failed')
          AND NULLIF(i.customer_no,'') IS NOT NULL
          AND c.economic_customer_number::text=i.customer_no
          AND c.deleted_at IS NULL
        """,
        (session_id,),
        fetch=False,
    )
    mapped_row = execute_query_single(
        """
        SELECT COUNT(*) AS total, COUNT(*) FILTER (WHERE hub_customer_id IS NOT NULL) AS mapped
        FROM migration_center_session_items WHERE session_id=%s AND source_system='economic'
        """,
        (session_id,),
    )
    relevance = refresh_subscription_relevance(session_id)
    result = {
        **counts, "processed": int(mapped_row["total"]), "customers_mapped": int(mapped_row["mapped"]),
        "import_run_id": run["id"], "snapshot_at": run["completed_at"], **relevance,
    }
    audit(
        request=request, current_user=current_user, action="economic_snapshot_attached",
        entity_type="session", entity_id=session_id, session_id=session_id, new_value=result,
    )
    return result


@router.get("/sessions/{session_id}/dashboard")
async def dashboard(
    session_id: int,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    session = execute_query_single(
        """
        SELECT s.*, r.status AS economic_import_status, r.records_imported, r.records_failed
        FROM migration_center_sessions s
        LEFT JOIN invoice_error_finder_import_runs r ON r.id = s.economic_import_run_id
        WHERE s.id = %s
        """,
        (session_id,),
    )
    if not session:
        raise HTTPException(status_code=404, detail="Kontrolsession blev ikke fundet")
    counts = execute_query_single(
        f"""
        SELECT COUNT(*) AS total,
          COUNT(*) FILTER (WHERE entity_type='subscription') AS subscriptions,
          COUNT(*) FILTER (WHERE entity_type='invoice_line') AS invoice_lines,
          COUNT(*) FILTER (WHERE hub_status='ready_for_creation') AS ready,
          COUNT(*) FILTER (WHERE hub_status='created_in_hub') AS created,
          COUNT(*) FILTER (WHERE approval_status='ignored') AS ignored,
          COUNT(*) FILTER (WHERE match_status IN ('conflict','source_changed')) AS conflicts,
          COUNT(*) FILTER (WHERE match_status='source_changed') AS source_changed,
          COUNT(*) FILTER (WHERE lock_status='locked') AS locked
        FROM migration_center_session_items
        WHERE session_id=%s
          AND {EXCLUDED_INVOICE_LINE_SQL}
          AND {subscription_like_item_sql()}
        """,
        (session_id,),
    )
    return {
        "session": session,
        "counts": counts or {},
        "read_only": bool(getattr(settings, "MIGRATION_CENTER_READ_ONLY", False) or session["read_only"]),
        "stale_after_days": int(getattr(settings, "MIGRATION_CENTER_STALE_AFTER_DAYS", 7)),
    }


@router.get("/sessions/{session_id}/company-options")
async def company_options(
    session_id: int,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    return execute_query(
        f"""
        SELECT customer_name, MAX(hub_customer_id) AS hub_customer_id, COUNT(*) AS item_count
        FROM migration_center_session_items
        WHERE session_id=%s AND NULLIF(TRIM(customer_name),'') IS NOT NULL
          AND {EXCLUDED_INVOICE_LINE_SQL}
          AND {subscription_like_item_sql()}
        GROUP BY customer_name
        ORDER BY LOWER(customer_name)
        """,
        (session_id,),
    ) or []


@router.get("/sessions/{session_id}/invoice-history")
async def invoice_history(
    session_id: int,
    only_subscription_candidates: bool = False,
    company_name: Optional[str] = None,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    candidate_filter = """
      WHERE NOT has_hub_subscription AND invoice_count >= 1 AND invoiced_months >= 1
        AND max_amount > 0 AND LOWER(product_name) NOT LIKE '%%gebyr%%'
    """ if only_subscription_candidates else ""
    rows = execute_query(
        f"""
        WITH grouped AS (
          SELECT
            COALESCE(NULLIF(customer_no,''), source_customer_id, customer_name) AS customer_key,
            MAX(customer_name) AS customer_name,
            MAX(customer_no) AS customer_no,
            MAX(hub_customer_id) AS hub_customer_id,
            COALESCE(NULLIF(product_code,''), LOWER(REGEXP_REPLACE(product_name,'\\s+',' ','g'))) AS product_key,
            MAX(product_code) AS product_code,
            MAX(product_name) AS product_name,
            COUNT(DISTINCT invoice_no) AS invoice_count,
            COUNT(DISTINCT DATE_TRUNC('month',invoice_date)) AS invoiced_months,
            MIN(invoice_date) AS first_invoice_date,
            MAX(invoice_date) AS last_invoice_date,
            MIN(amount) AS min_amount,
            MAX(amount) AS max_amount,
            SUM(amount) AS total_amount,
            BOOL_OR(hub_record_id IS NOT NULL OR suggested_hub_record_id IS NOT NULL) AS explicitly_linked,
            MIN(id) AS representative_item_id
          FROM migration_center_session_items mci
          WHERE session_id=%s AND source_system='economic' AND entity_type='invoice_line'
            AND invoice_date >= (DATE_TRUNC('month', CURRENT_DATE) - INTERVAL '12 months')::date
            AND invoice_date <= CURRENT_DATE
            AND {EXCLUDED_INVOICE_LINE_SQL}
            AND {subscription_like_item_sql('mci')}
            AND (%s IS NULL OR LOWER(customer_name)=LOWER(%s))
          GROUP BY
            COALESCE(NULLIF(customer_no,''), source_customer_id, customer_name),
            COALESCE(NULLIF(product_code,''), LOWER(REGEXP_REPLACE(product_name,'\\s+',' ','g')))
        ), matched AS (
          SELECT grouped.*,
            (
              explicitly_linked OR EXISTS (
                SELECT 1 FROM sag_subscriptions s
                WHERE s.customer_id=grouped.hub_customer_id
                  AND s.status <> 'cancelled'
                  AND (
                    LOWER(TRIM(COALESCE(s.product_name,'')))=LOWER(TRIM(grouped.product_name))
                    OR ABS(COALESCE(s.price,0)-COALESCE(grouped.max_amount,0)) <= 0.01
                  )
              )
            ) AS has_hub_subscription
          FROM grouped
        )
        SELECT *,
          CASE
            WHEN invoiced_months >= 10 THEN 'monthly'
            WHEN invoiced_months >= 4 AND
                 (last_invoice_date-first_invoice_date)/GREATEST(invoiced_months-1,1) BETWEEN 60 AND 120
              THEN 'quarterly'
            WHEN invoice_count >= 2 AND last_invoice_date-first_invoice_date >= 300 THEN 'yearly'
            ELSE 'irregular'
          END AS suggested_frequency,
          (
            NOT has_hub_subscription AND invoice_count >= 1 AND invoiced_months >= 1
            AND max_amount > 0 AND LOWER(product_name) NOT LIKE '%%gebyr%%'
          ) AS subscription_candidate
        FROM matched
        {candidate_filter}
        ORDER BY subscription_candidate DESC, invoiced_months DESC, customer_name, product_name
        """,
        (session_id, company_name, company_name),
    ) or []
    return {"months": 13, "groups": rows, "count": len(rows)}


@router.get("/sessions/{session_id}/items")
async def list_items(
    session_id: int,
    q: Optional[str] = None,
    company_name: Optional[str] = None,
    entity_type: Optional[str] = None,
    source_system: Optional[str] = None,
    match_status: Optional[str] = None,
    approval_status: Optional[str] = None,
    hub_status: Optional[str] = None,
    lock_status: Optional[str] = None,
    only_deviations: bool = False,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    sort: str = "invoice_date",
    direction: str = "desc",
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    allowed_sort = {
        "invoice_date": "invoice_date", "customer": "customer_name", "product": "product_name",
        "amount": "amount", "status": "match_status", "created": "created_at",
    }
    clauses = ["session_id = %s"]
    params: list[Any] = [session_id]
    filters = {
        "entity_type": entity_type, "source_system": source_system, "match_status": match_status,
        "approval_status": approval_status, "hub_status": hub_status, "lock_status": lock_status,
    }
    for column, value in filters.items():
        if value:
            clauses.append(f"{column} = %s")
            params.append(value)
    if only_deviations:
        clauses.append("(match_status IN ('no_match','conflict','manual_review','source_changed') OR hub_status='not_created')")
    if q:
        clauses.append(
            "(customer_name ILIKE %s OR customer_no ILIKE %s OR product_name ILIKE %s "
            "OR product_code ILIKE %s OR invoice_no ILIKE %s OR source_record_id ILIKE %s)"
        )
        term = f"%{q.strip()}%"
        params.extend([term] * 6)
    if company_name:
        clauses.append("LOWER(customer_name) = LOWER(%s)")
        params.append(company_name.strip())
    clauses.append(EXCLUDED_INVOICE_LINE_SQL)
    clauses.append(subscription_like_item_sql())
    where = " AND ".join(clauses)
    total = execute_query_single(f"SELECT COUNT(*) AS count FROM migration_center_session_items WHERE {where}", tuple(params))
    order = allowed_sort.get(sort, "invoice_date")
    direction_sql = "ASC" if direction.lower() == "asc" else "DESC"
    rows = execute_query(
        f"""
        SELECT * FROM migration_center_session_items
        WHERE {where}
        ORDER BY {order} {direction_sql} NULLS LAST, id DESC
        LIMIT %s OFFSET %s
        """,
        tuple(params + [page_size, (page - 1) * page_size]),
    ) or []
    return {"items": rows, "total": int((total or {}).get("count", 0)), "page": page, "page_size": page_size}


@router.get("/items/{item_id}")
async def get_item(
    item_id: int,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    return _load_item(item_id)


@router.get("/items/{item_id}/context")
async def get_item_context(
    item_id: int,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    item = _load_item(item_id)
    params: list[Any] = [item["session_id"]]
    relations = []
    if item.get("hub_customer_id"):
        relations.append("related.hub_customer_id=%s")
        params.append(item["hub_customer_id"])
    if item.get("customer_no"):
        relations.append("(NULLIF(related.customer_no,'') IS NOT NULL AND related.customer_no=%s)")
        params.append(item["customer_no"])
    if item.get("source_customer_id"):
        relations.append(
            "(related.source_system=%s AND related.source_customer_id=%s)"
        )
        params.extend([item["source_system"], item["source_customer_id"]])
    relation_sql = " OR ".join(relations) or "related.id=%s"
    if not relations:
        params.append(item_id)
    related = execute_query(
        f"""
        SELECT related.*
        FROM migration_center_session_items related
        WHERE related.session_id=%s AND ({relation_sql})
          AND {EXCLUDED_INVOICE_LINE_SQL}
          AND {subscription_like_item_sql('related')}
        ORDER BY
          CASE related.source_system
            WHEN 'economic' THEN 1 WHEN 'vtiger' THEN 2 WHEN 'simply' THEN 3 ELSE 4
          END,
          related.invoice_date DESC NULLS LAST, related.period_from DESC NULLS LAST, related.id DESC
        LIMIT 500
        """,
        tuple(params),
    ) or []
    by_source = {"economic": [], "vtiger": [], "simply": []}
    for row in related:
        source = row.get("source_system")
        if source in by_source:
            by_source[source].append(row)

    hub_customer = None
    hub_cases = []
    hub_subscriptions = []
    if item.get("hub_customer_id"):
        hub_customer = execute_query_single(
            """
            SELECT id, name, cvr_number, email, phone, economic_customer_number, vtiger_id
            FROM customers WHERE id=%s
            """,
            (item["hub_customer_id"],),
        )
        hub_cases = execute_query(
            """
            SELECT id, titel, status, template_key, created_at
            FROM sag_sager WHERE customer_id=%s
            ORDER BY created_at DESC LIMIT 100
            """,
            (item["hub_customer_id"],),
        ) or []
        hub_subscriptions = execute_query(
            """
            SELECT s.id, s.subscription_number, s.sag_id, s.product_name, s.price,
                   s.billing_interval, s.start_date, s.end_date, s.status,
                   COALESCE(
                     JSONB_AGG(
                       JSONB_BUILD_OBJECT(
                         'id', line.id, 'description', line.description, 'quantity', line.quantity,
                         'unit_price', line.unit_price, 'line_total', line.line_total
                       ) ORDER BY line.line_no
                     ) FILTER (WHERE line.id IS NOT NULL),
                     '[]'::jsonb
                   ) AS lines
            FROM sag_subscriptions s
            LEFT JOIN sag_subscription_items line ON line.subscription_id=s.id
            WHERE s.customer_id=%s
            GROUP BY s.id
            ORDER BY s.updated_at DESC LIMIT 100
            """,
            (item["hub_customer_id"],),
        ) or []
    return {
        "item": item,
        "sources": by_source,
        "hub": {
            "customer": hub_customer,
            "cases": hub_cases,
            "subscriptions": hub_subscriptions,
        },
        "counts": {
            "economic": len(by_source["economic"]),
            "vtiger": len(by_source["vtiger"]),
            "simply": len(by_source["simply"]),
            "hub_subscriptions": len(hub_subscriptions),
        },
    }


@router.get("/sessions/{session_id}/queue/next")
async def next_queue_item(
    session_id: int,
    after_id: Optional[int] = None,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    params: list[Any] = [session_id]
    after = ""
    if after_id:
        after = " AND id > %s"
        params.append(after_id)
    row = execute_query_single(
        f"""
        SELECT * FROM migration_center_session_items
        WHERE session_id=%s AND approval_status='pending'
          AND lock_status IN ('unlocked','lock_failed')
          AND {EXCLUDED_INVOICE_LINE_SQL}
          AND {subscription_like_item_sql()} {after}
        ORDER BY CASE match_status WHEN 'conflict' THEN 0 WHEN 'source_changed' THEN 1
                 WHEN 'manual_review' THEN 2 ELSE 3 END, id
        LIMIT 1
        """,
        tuple(params),
    )
    return {"item": row}


@router.get("/sessions/{session_id}/customers/{customer_id}")
async def customer_overview(
    session_id: int,
    customer_id: int,
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    customer = execute_query_single(
        """
        SELECT id, name, cvr_number, email, economic_customer_number
        FROM customers WHERE id=%s AND deleted_at IS NULL
        """,
        (customer_id,),
    )
    if not customer:
        raise HTTPException(status_code=404, detail="Kunden blev ikke fundet")
    items = execute_query(
        f"""
        SELECT * FROM migration_center_session_items
        WHERE session_id=%s AND hub_customer_id=%s
          AND {EXCLUDED_INVOICE_LINE_SQL}
          AND {subscription_like_item_sql()}
        ORDER BY entity_type, invoice_date DESC NULLS LAST, id
        """,
        (session_id, customer_id),
    ) or []
    subscriptions = execute_query(
        """
        SELECT id, subscription_number, product_name, price, billing_interval, status, start_date, end_date
        FROM sag_subscriptions WHERE customer_id=%s ORDER BY updated_at DESC
        """,
        (customer_id,),
    ) or []
    open_conflicts = [
        row["id"] for row in items if row["match_status"] in {"conflict", "source_changed"}
    ]
    untreated = [
        row["id"] for row in items if row["approval_status"] not in {"verified", "ignored"}
    ]
    return {
        "customer": customer, "items": items, "hub_subscriptions": subscriptions,
        "open_conflicts": open_conflicts, "untreated": untreated,
        "can_lock_all": bool(items) and not open_conflicts and not untreated,
    }


@router.post("/vtiger-cases/preview")
async def preview_vtiger_case_import(
    payload: VtigerCaseImportRequest,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    bundle = await _load_vtiger_case_bundle(payload.ticket_number)
    raw = bundle["case"]
    already_imported = bundle.get("existing_import")
    eligible_timelogs = [row for row in bundle.get("timelogs", []) if _amount(row.get("duration")) > 0]
    return {
        "ticket_number": bundle["ticket_number"],
        "vtiger_case_id": bundle["vtiger_case_id"],
        "title": _vtiger_text(raw.get("ticket_title") or raw.get("subject")) or bundle["ticket_number"],
        "description": _vtiger_plain_text(raw.get("description")),
        "status": _case_status_from_vtiger(raw.get("status")),
        "priority": _case_priority_from_vtiger(raw.get("priority")),
        "customer": bundle["customer"],
        "contact": bundle.get("contact"),
        "time_entries": len(eligible_timelogs),
        "time_hours": str(sum((_amount(row.get("duration")) for row in eligible_timelogs), Decimal("0")) / Decimal("3600")),
        "comments_available_not_imported": len(bundle.get("comments", [])),
        "already_imported": bool(already_imported),
        "hub_sag_id": already_imported.get("hub_sag_id") if already_imported else None,
        "vtiger_marked": bool(already_imported and already_imported.get("vtiger_marked_at")),
    }


@router.post("/vtiger-cases/import")
async def import_vtiger_case(
    payload: VtigerCaseImportRequest,
    request: Request,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    bundle = await _load_vtiger_case_bundle(payload.ticket_number)
    result = _import_vtiger_case_bundle(bundle, current_user)
    marked = bool(bundle.get("existing_import") and bundle["existing_import"].get("vtiger_marked_at"))
    mark_error = None
    if not marked:
        configured_field = getattr(settings, "VTIGER_CASE_HUBBED_FIELD", "cf_cases_hubbet") or "cf_cases_hubbet"
        if not re.fullmatch(r"cf_[A-Za-z0-9_]+", configured_field):
            mark_error = "VTIGER_CASE_HUBBED_FIELD er ikke konfigureret med et gyldigt VTiger-feltnavn"
        else:
            try:
                await get_vtiger_service().update_record_fields("Cases", bundle["vtiger_case_id"], {configured_field: "1"})
                marked = True
                execute_query(
                    "UPDATE vtiger_case_hub_imports SET vtiger_marked_at=CURRENT_TIMESTAMP WHERE vtiger_case_id=%s",
                    (bundle["vtiger_case_id"],), fetch=False,
                )
            except Exception as exc:
                mark_error = f"Sagen blev overført, men vTiger kunne ikke markeres Hubbet: {str(exc)[:300]}"
    audit(
        request=request, current_user=current_user, action="vtiger_case_imported",
        entity_type="case", entity_id=result["hub_sag_id"],
        new_value={"vtiger_case_id": bundle["vtiger_case_id"], "ticket_number": bundle["ticket_number"],
                   "imported": result["imported"], "vtiger_marked": marked},
    )
    return {"success": True, **result, "vtiger_marked": marked, "warning": mark_error}


def _hub_status_to_vtiger(value: Any, allowed: set[str]) -> str:
    key = _vtiger_text(value).strip().lower()
    candidates = {
        "ny": "New", "tildelt": "Assigned", "igangværende": "In Progress",
        "afventer": "Wait for customer", "løst": "Resolved", "lukket": "Closed",
        "kunde svaret": "Open", "åben": "Open",
    }
    candidate = candidates.get(key, "New")
    return candidate if candidate in allowed else ("Open" if "Open" in allowed else next(iter(allowed), "New"))


def _hub_priority_to_vtiger(value: Any, allowed: set[str]) -> str:
    key = _vtiger_text(value).strip().lower()
    candidates = {"urgent": "Urgent", "high": "High", "normal": "Medium", "low": "Low"}
    candidate = candidates.get(key, "Medium")
    return candidate if candidate in allowed else ("Medium" if "Medium" in allowed else next(iter(allowed), candidate))


def _time_export_minutes(row: Dict[str, Any]) -> int:
    actual = row.get("faktisk_tid_min")
    if actual not in (None, "") and int(actual or 0) > 0:
        return int(actual)
    hours = row.get("original_hours") or row.get("approved_hours") or 0
    return max(0, int((_amount(hours) * Decimal("60")).to_integral_value()))


def _time_export_billable(row: Dict[str, Any]) -> bool:
    method = _vtiger_text(row.get("billing_method")).lower()
    return bool(row.get("billable")) and method in {"", "invoice", "faktura"}


def _normalized_person(value: Any) -> str:
    return " ".join(_vtiger_text(value).casefold().split())


async def _load_hub_case_export_bundle(sag_id: int) -> Dict[str, Any]:
    case = execute_query_single(
        """SELECT s.id, s.titel, s.beskrivelse, s.status, s.priority::text AS priority,
                  s.customer_id, c.name AS customer_name, c.vtiger_id AS account_vtiger_id,
                  s.ansvarlig_bruger_id, owner.full_name AS owner_name, owner.email AS owner_email
           FROM sag_sager s
           JOIN customers c ON c.id=s.customer_id
           LEFT JOIN users owner ON owner.user_id=s.ansvarlig_bruger_id
           WHERE s.id=%s AND s.deleted_at IS NULL""",
        (sag_id,),
    )
    if not case:
        raise HTTPException(status_code=404, detail=f"Hub-sag #{sag_id} blev ikke fundet")
    contact = execute_query_single(
        """SELECT c.id, c.first_name, c.last_name, c.email, c.vtiger_id
           FROM sag_kontakter sc JOIN contacts c ON c.id=sc.contact_id
           WHERE sc.sag_id=%s AND sc.deleted_at IS NULL AND c.vtiger_id IS NOT NULL
           ORDER BY sc.is_primary DESC, sc.id LIMIT 1""",
        (sag_id,),
    )
    if not contact:
        contact = execute_query_single(
            """SELECT c.id, c.first_name, c.last_name, c.email, c.vtiger_id
               FROM contact_companies cc JOIN contacts c ON c.id=cc.contact_id
               WHERE cc.customer_id=%s AND cc.is_primary IS TRUE AND c.vtiger_id IS NOT NULL
               ORDER BY cc.id LIMIT 1""",
            (case["customer_id"],),
        )
    times = execute_query(
        """SELECT t.id, t.description, t.worked_date, t.start_tid, t.slut_tid,
                  t.original_hours, t.approved_hours, t.faktisk_tid_min, t.fakturerbar_tid_min,
                  t.billable, t.billing_method, t.status, t.entry_status, t.vtiger_id,
                  t.user_name, t.medarbejder_id, u.full_name AS technician_name,
                  u.email AS technician_email
           FROM tmodule_times t
           LEFT JOIN users u ON u.user_id=t.medarbejder_id
           WHERE t.sag_id=%s AND t.vtiger_id IS NULL
             AND COALESCE(t.entry_status, 'godkendt')='godkendt'
             AND COALESCE(t.status, '') NOT IN ('billed', 'rejected')
             AND t.billed_via_thehub_id IS NULL AND t.economy_settled_at IS NULL
             AND t.economy_order_draft_id IS NULL
             AND NOT EXISTS (SELECT 1 FROM hub_time_vtiger_exports x WHERE x.hub_time_id=t.id)
           ORDER BY COALESCE(t.start_tid, t.worked_date::timestamp, t.created_at), t.id""",
        (sag_id,),
    ) or []
    all_times = execute_query_single(
        """SELECT COUNT(*) AS total,
                  COUNT(*) FILTER (WHERE COALESCE(entry_status, 'godkendt') <> 'godkendt') AS pending
           FROM tmodule_times WHERE sag_id=%s""",
        (sag_id,),
    ) or {"total": 0, "pending": 0}
    export_mapping = execute_query_single(
        "SELECT vtiger_case_id, ticket_number, status FROM hub_case_vtiger_exports WHERE hub_sag_id=%s",
        (sag_id,),
    )
    imported_mapping = execute_query_single(
        "SELECT vtiger_case_id, ticket_number FROM vtiger_case_hub_imports WHERE hub_sag_id=%s",
        (sag_id,),
    )
    return {
        "case": case, "contact": contact, "times": times,
        "pending_time_count": int(all_times.get("pending") or 0),
        "export_mapping": export_mapping, "imported_mapping": imported_mapping,
    }


async def _vtiger_user_records(vtiger) -> list[Dict[str, Any]]:
    records: list[Dict[str, Any]] = []
    for offset in range(0, 500, 100):
        batch = await vtiger.query(
            "SELECT id, first_name, last_name, user_name, email1 FROM Users "
            f"LIMIT {offset}, 100;"
        )
        records.extend(batch or [])
        if len(batch or []) < 100:
            break
    return records


def _match_vtiger_user(row: Dict[str, Any], users: list[Dict[str, Any]]) -> Optional[str]:
    email = _normalized_person(row.get("technician_email"))
    name = _normalized_person(row.get("technician_name") or row.get("user_name"))
    matches = []
    for user in users:
        user_id = _vtiger_reference_id(user.get("id"))
        user_name = " ".join(
            part for part in (_vtiger_text(user.get("first_name")), _vtiger_text(user.get("last_name"))) if part
        ).strip()
        if not user_id:
            continue
        if email and email in {_normalized_person(user.get("email1")), _normalized_person(user.get("user_name"))}:
            return user_id
        if name and name in {_normalized_person(user_name), _normalized_person(user.get("user_name"))}:
            matches.append(user_id)
    return matches[0] if len(set(matches)) == 1 else None


def _export_time_payload(row: Dict[str, Any], sag_id: int, vtiger_case_id: str, vtiger_user_id: str) -> Dict[str, Any]:
    minutes = _time_export_minutes(row)
    start = row.get("start_tid")
    if not isinstance(start, datetime):
        work_date = row.get("worked_date")
        if isinstance(work_date, datetime):
            work_date = work_date.date()
        if not isinstance(work_date, date):
            raise ValueError(f"Tidsregistrering #{row['id']} mangler en gyldig dato")
        start = datetime.combine(work_date, datetime_time(9, 0))
    duration_seconds = minutes * 60
    description = _vtiger_text(row.get("description"))
    name = description[:250] or f"Hub sag #{sag_id} · {start:%Y-%m-%d}"
    method = _vtiger_text(row.get("billing_method")).lower()
    return {
        "name": name,
        "description": description,
        "relatedto": vtiger_case_id,
        "startedon": start.strftime("%Y-%m-%d %H:%M:%S"),
        "endedon": (start + timedelta(seconds=duration_seconds)).strftime("%Y-%m-%d %H:%M:%S"),
        "duration": str(duration_seconds),
        "assigned_user_id": vtiger_user_id,
        "isbillable": "1" if _time_export_billable(row) else "0",
        "timelog_status": "Completed",
        "cf_timelog_invoiced": "0",
    }


async def _prepare_hub_case_export(sag_id: int) -> Dict[str, Any]:
    bundle = await _load_hub_case_export_bundle(sag_id)
    case = bundle["case"]
    blockers = []
    if bundle.get("imported_mapping"):
        blockers.append("Sagen stammer allerede fra vTiger og kan ikke oprettes dér igen.")
    if not case.get("account_vtiger_id"):
        blockers.append("Kunden mangler en VTiger-kontakt/linkning.")
    if not bundle.get("contact") or not bundle["contact"].get("vtiger_id"):
        blockers.append("Sagen skal have en primær kontakt, der er linket til VTiger.")
    vtiger = get_vtiger_service()
    case_meta = await vtiger.describe_module("Cases")
    timelog_meta = await vtiger.describe_module("Timelog")
    fields = {f.get("name"): f for f in case_meta.get("fields", [])}
    time_fields = {f.get("name"): f for f in timelog_meta.get("fields", [])}
    required_case = [name for name, field in fields.items() if field.get("mandatory") and field.get("editable")]
    required_time = [name for name, field in time_fields.items() if field.get("mandatory") and field.get("editable")]
    if set(required_case) - {"title", "description", "casestatus", "casepriority", "parent_id", "contact_id", "cf_cases_hubbet"}:
        blockers.append("VTiger har nye obligatoriske sagsfelter, som endnu ikke er mappet.")
    if set(required_time) - {"name", "description", "relatedto", "startedon", "endedon", "assigned_user_id", "duration", "isbillable", "timelog_status", "cf_timelog_invoiced"}:
        blockers.append("VTiger har nye obligatoriske tidsfelter, som endnu ikke er mappet.")
    if any(name not in fields for name in ("title", "description", "casestatus", "casepriority", "parent_id", "contact_id")):
        blockers.append("VTigers sagsfelter svarer ikke til den understøttede overførsel.")
    if any(name not in time_fields for name in ("name", "description", "relatedto", "startedon", "endedon", "assigned_user_id", "duration", "isbillable", "timelog_status")):
        blockers.append("VTigers tidsregistreringsfelter svarer ikke til den understøttede overførsel.")
    allowed_statuses = {str(p.get("value")) for p in fields.get("casestatus", {}).get("type", {}).get("picklistValues", []) if p.get("value")}
    allowed_priorities = {str(p.get("value")) for p in fields.get("casepriority", {}).get("type", {}).get("picklistValues", []) if p.get("value")}
    users = await _vtiger_user_records(vtiger) if bundle["times"] else []
    enriched_times = []
    for row in bundle["times"]:
        technician_name = _vtiger_text(row.get("technician_name") or row.get("user_name"))
        vtiger_user_id = _match_vtiger_user(row, users)
        if not vtiger_user_id:
            blockers.append(f"Teknikeren på tidsregistrering #{row['id']} matcher ikke en VTiger-bruger.")
        enriched_times.append({
            **row,
            "technician_name": technician_name or "Ukendt tekniker",
            "vtiger_user_id": vtiger_user_id,
            "minutes": _time_export_minutes(row),
            "export_billable": _time_export_billable(row),
        })
    case_hash_data = {
        "case": {key: str(case.get(key) or "") for key in ("id", "titel", "beskrivelse", "status", "priority", "account_vtiger_id")},
        "contact": {key: str(bundle["contact"].get(key) or "") for key in ("vtiger_id", "email")} if bundle.get("contact") else {},
        "times": [
            {key: str(row.get(key) or "") for key in ("id", "description", "worked_date", "start_tid", "original_hours", "faktisk_tid_min", "fakturerbar_tid_min", "billable", "billing_method", "technician_email", "technician_name")}
            for row in enriched_times
        ],
    }
    source_hash = hashlib.sha256(json.dumps(case_hash_data, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
    return {
        **bundle,
        "times": enriched_times,
        "blockers": sorted(set(blockers)),
        "source_hash": source_hash,
        "case_meta": case_meta,
        "allowed_statuses": allowed_statuses,
        "allowed_priorities": allowed_priorities,
        "required_case_fields": required_case,
        "required_time_fields": required_time,
        "status_value": _hub_status_to_vtiger(case.get("status"), allowed_statuses),
        "priority_value": _hub_priority_to_vtiger(case.get("priority"), allowed_priorities),
    }


@router.post("/hub-cases/preview")
async def preview_hub_case_export(
    payload: HubCaseExportRequest,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    data = await _prepare_hub_case_export(payload.sag_id)
    case = data["case"]
    mapping = data.get("export_mapping")
    return {
        "sag_id": case["id"], "title": case["titel"],
        "description": _vtiger_plain_text(case.get("beskrivelse")),
        "customer_name": case.get("customer_name"),
        "contact_name": " ".join(filter(None, [data.get("contact", {}).get("first_name"), data.get("contact", {}).get("last_name")])) if data.get("contact") else None,
        "status": case.get("status"), "vtiger_status": data["status_value"],
        "priority": case.get("priority"), "vtiger_priority": data["priority_value"],
        "time_entries": len(data["times"]),
        "time_hours": str(sum((Decimal(row["minutes"]) / Decimal("60") for row in data["times"]), Decimal("0"))),
        "time_rows": [{"id": row["id"], "date": str(row.get("worked_date") or ""), "description": _vtiger_plain_text(row.get("description")), "technician": row["technician_name"], "hours": str(Decimal(row["minutes"]) / Decimal("60")), "billable": row["export_billable"]} for row in data["times"]],
        "pending_time_count": data["pending_time_count"],
        "already_exported": bool(mapping),
        "vtiger_case_id": mapping.get("vtiger_case_id") if mapping else None,
        "ticket_number": mapping.get("ticket_number") if mapping else None,
        "blockers": data["blockers"], "source_hash": data["source_hash"],
    }


@router.post("/hub-cases/export")
async def export_hub_case_to_vtiger(
    payload: HubCaseExportConfirmRequest,
    request: Request,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    data = await _prepare_hub_case_export(payload.sag_id)
    if data["source_hash"] != payload.source_hash:
        raise HTTPException(status_code=409, detail="Sagen eller tidsregistreringer er ændret siden forhåndsvisningen. Hent en ny forhåndsvisning.")
    if data["blockers"]:
        raise HTTPException(status_code=409, detail={"message": "Sagen kan ikke overføres endnu.", "blockers": data["blockers"]})

    conn = get_db_connection()
    vtiger = get_vtiger_service()
    created_case = False
    created_times = 0
    warnings = []
    mapping = None
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute("SELECT pg_advisory_lock(1046, %s)", (payload.sag_id,))
            cursor.execute("SELECT vtiger_case_id, ticket_number FROM hub_case_vtiger_exports WHERE hub_sag_id=%s FOR UPDATE", (payload.sag_id,))
            mapping = cursor.fetchone()
            if not mapping:
                cursor.execute("SELECT 1 FROM vtiger_case_hub_imports WHERE hub_sag_id=%s", (payload.sag_id,))
                if cursor.fetchone():
                    raise HTTPException(status_code=409, detail="Sagen findes allerede i VTiger fra en tidligere VTiger-overførsel.")
                case = data["case"]
                case_payload = {
                    "title": _vtiger_text(case.get("titel"))[:250],
                    "description": _vtiger_plain_text(case.get("beskrivelse")) or f"Hub sag #{payload.sag_id}",
                    "casestatus": data["status_value"],
                    "casepriority": data["priority_value"],
                    "parent_id": _vtiger_reference_id(case.get("account_vtiger_id")),
                    "contact_id": _vtiger_reference_id(data["contact"].get("vtiger_id")),
                    "cf_cases_hubbet": "1",
                }
                result = await vtiger.create_record("Cases", case_payload)
                vtiger_case_id = _vtiger_reference_id(result.get("id"))
                if not vtiger_case_id:
                    raise RuntimeError("VTiger oprettede sagen, men returnerede ikke et gyldigt sags-id.")
                ticket_number = _vtiger_text(result.get("case_no") or result.get("ticket_no")) or None
                cursor.execute(
                    """INSERT INTO hub_case_vtiger_exports (hub_sag_id, vtiger_case_id, ticket_number, status)
                       VALUES (%s,%s,%s,'partial')""",
                    (payload.sag_id, vtiger_case_id, ticket_number),
                )
                conn.commit()
                mapping = {"vtiger_case_id": vtiger_case_id, "ticket_number": ticket_number}
                created_case = True
            vtiger_case_id = mapping["vtiger_case_id"]
            conn.commit()

            for row in data["times"]:
                cursor.execute("SELECT vtiger_timelog_id FROM hub_time_vtiger_exports WHERE hub_time_id=%s", (row["id"],))
                if cursor.fetchone():
                    continue
                try:
                    time_payload = _export_time_payload(row, payload.sag_id, vtiger_case_id, row["vtiger_user_id"])
                    time_result = await vtiger.create_record("Timelog", time_payload)
                    vtiger_timelog_id = _vtiger_reference_id(time_result.get("id"))
                    if not vtiger_timelog_id:
                        raise RuntimeError("VTiger oprettede tiden, men returnerede ikke et gyldigt tids-id.")
                    cursor.execute(
                        """INSERT INTO hub_time_vtiger_exports (hub_time_id, hub_sag_id, vtiger_timelog_id)
                           VALUES (%s,%s,%s)""",
                        (row["id"], payload.sag_id, vtiger_timelog_id),
                    )
                    conn.commit()
                    created_times += 1
                except Exception as exc:
                    warnings.append(f"Tidsregistrering #{row['id']}: {str(exc)[:250]}")
                    break
            cursor.execute(
                """UPDATE hub_case_vtiger_exports SET status=%s, ticket_number=COALESCE(ticket_number,%s),
                          last_error=%s, updated_at=CURRENT_TIMESTAMP WHERE hub_sag_id=%s""",
                ("complete" if not warnings else "partial", mapping.get("ticket_number"), "; ".join(warnings) or None, payload.sag_id),
            )
            conn.commit()
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        warnings.append(str(exc)[:300])
        execute_query(
            "UPDATE hub_case_vtiger_exports SET status='partial', last_error=%s, updated_at=CURRENT_TIMESTAMP WHERE hub_sag_id=%s",
            ("; ".join(warnings), payload.sag_id), fetch=False,
        )
        if not mapping:
            raise HTTPException(status_code=502, detail=f"VTiger-overførslen fejlede: {str(exc)[:300]}") from exc
    finally:
        try:
            with conn.cursor() as cursor:
                cursor.execute("SELECT pg_advisory_unlock(1046, %s)", (payload.sag_id,))
            conn.commit()
        except Exception:
            conn.rollback()
        release_db_connection(conn)

    result = {
        "success": True, "sag_id": payload.sag_id,
        "vtiger_case_id": vtiger_case_id,
        "ticket_number": mapping.get("ticket_number"),
        "created_case": created_case,
        "created_time_entries": created_times,
        "complete": not warnings,
        "warning": "; ".join(warnings) or None,
    }
    audit(
        request=request, current_user=current_user, action="hub_case_exported_to_vtiger",
        entity_type="case", entity_id=payload.sag_id,
        new_value={key: value for key, value in result.items() if key not in {"warning"}},
    )
    return result


@router.post("/items/{item_id}/rematch")
async def rematch(
    item_id: int, request: Request,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    result = match_item(item_id)
    audit(
        request=request, current_user=current_user, action="rematched", entity_type=item["entity_type"],
        entity_id=item["source_record_id"], session_id=item["session_id"], item_id=item_id,
        old_value=item, new_value=result, source_hash_value=item["source_hash"],
    )
    return result


@router.post("/sessions/{session_id}/rematch")
async def rematch_session(
    session_id: int, request: Request,
    current_user: dict = Depends(require_permission("migration_center.import")),
):
    ensure_writable(session_id)
    rows = execute_query(
        """
        SELECT id FROM migration_center_session_items
        WHERE session_id=%s AND lock_status IN ('unlocked','lock_failed')
        ORDER BY id
        """,
        (session_id,),
    ) or []
    counts = {"processed": 0, "customers_mapped": 0, "matches_found": 0}
    for row in rows:
        result = match_item(int(row["id"]))
        counts["processed"] += 1
        if result.get("hub_customer_id"):
            counts["customers_mapped"] += 1
        if result.get("match_status") == "match_found":
            counts["matches_found"] += 1
    audit(
        request=request, current_user=current_user, action="session_rematched",
        entity_type="session", entity_id=session_id, session_id=session_id, new_value=counts,
    )
    return counts


@router.post("/items/{item_id}/customers")
async def create_customer(
    item_id: int, payload: CustomerCreate, request: Request,
    current_user: dict = Depends(require_permission("migration_center.create")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    duplicate = execute_query_single(
        """
        SELECT id, name FROM customers
        WHERE deleted_at IS NULL AND (
          (%s IS NOT NULL AND NULLIF(TRIM(cvr_number),'') = NULLIF(TRIM(%s),''))
          OR LOWER(name) = LOWER(%s)
        ) LIMIT 1
        """,
        (payload.cvr_number, payload.cvr_number, payload.name),
    )
    if duplicate:
        raise HTTPException(status_code=409, detail={"message": "Mulig dublet fundet", "customer": duplicate})
    email_domain = payload.email.split("@", 1)[1].lower() if payload.email and "@" in payload.email else None
    conn = get_db_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
            cursor.execute(
                """
                INSERT INTO customers (name, cvr_number, email, email_domain, country, is_active, economic_customer_number)
                VALUES (%s,%s,%s,%s,'DK',TRUE,%s) RETURNING id, name, cvr_number, email
                """,
                (payload.name, payload.cvr_number, payload.email, email_domain, payload.customer_no),
            )
            customer = dict(cursor.fetchone())
            cursor.execute(
                "UPDATE migration_center_session_items SET hub_customer_id=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s",
                (customer["id"], item_id),
            )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        release_db_connection(conn)
    audit(
        request=request, current_user=current_user, action="customer_created", entity_type="customer",
        entity_id=customer["id"], session_id=item["session_id"], item_id=item_id, new_value=customer,
        source_hash_value=item["source_hash"],
    )
    return customer


@router.post("/items/{item_id}/customer-link")
async def link_customer(
    item_id: int, payload: LinkPayload, request: Request,
    current_user: dict = Depends(require_permission("migration_center.review")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    customer = execute_query_single("SELECT id, name FROM customers WHERE id=%s AND deleted_at IS NULL", (payload.hub_id,))
    if not customer:
        raise HTTPException(status_code=404, detail="Hub-kunden blev ikke fundet")
    execute_query(
        """
        UPDATE migration_center_session_items SET hub_customer_id=%s,
          hub_sag_id=NULL, hub_record_id=NULL, hub_status='ready_for_creation',
          updated_at=CURRENT_TIMESTAMP WHERE id=%s
        """,
        (payload.hub_id, item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="customer_linked", entity_type=item["entity_type"],
        entity_id=item["source_record_id"], session_id=item["session_id"], item_id=item_id,
        old_value={"hub_customer_id": item.get("hub_customer_id")}, new_value={"hub_customer_id": payload.hub_id},
        source_hash_value=item["source_hash"],
    )
    return {"success": True, "customer": customer}


@router.post("/items/{item_id}/cases")
async def create_case(
    item_id: int, payload: CaseCreate, request: Request,
    current_user: dict = Depends(require_permission("migration_center.create")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    if not item.get("hub_customer_id"):
        raise HTTPException(status_code=409, detail="Kunden skal linkes eller oprettes først")
    row = execute_query_single(
        """
        INSERT INTO sag_sager (titel, beskrivelse, template_key, status, customer_id, created_by_user_id)
        VALUES (%s,%s,'subscription','åben',%s,%s) RETURNING id, titel, customer_id
        """,
        (payload.title, payload.description, item["hub_customer_id"], user_id(current_user)),
    )
    execute_query(
        "UPDATE migration_center_session_items SET hub_sag_id=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s",
        (row["id"], item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="case_created", entity_type="case",
        entity_id=row["id"], session_id=item["session_id"], item_id=item_id, new_value=row,
        source_hash_value=item["source_hash"],
    )
    return row


@router.post("/items/{item_id}/case-link")
async def link_case(
    item_id: int, payload: LinkPayload, request: Request,
    current_user: dict = Depends(require_permission("migration_center.review")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    case = execute_query_single(
        "SELECT id, titel, customer_id FROM sag_sager WHERE id=%s AND customer_id=%s",
        (payload.hub_id, item.get("hub_customer_id")),
    )
    if not case:
        raise HTTPException(status_code=404, detail="Sagen findes ikke på den valgte kunde")
    execute_query(
        "UPDATE migration_center_session_items SET hub_sag_id=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s",
        (payload.hub_id, item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="case_linked", entity_type=item["entity_type"],
        entity_id=item["source_record_id"], session_id=item["session_id"], item_id=item_id,
        new_value={"hub_sag_id": payload.hub_id}, source_hash_value=item["source_hash"],
    )
    return {"success": True, "case": case}


@router.get("/items/{item_id}/preflight")
async def preflight(
    item_id: int,
    current_user: dict = Depends(require_permission("migration_center.create")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    blockers = []
    if not item.get("hub_customer_id"):
        blockers.append("Kunden er ikke linket")
    if not item.get("hub_sag_id"):
        blockers.append("Sagen er ikke linket")
    if item.get("hub_record_id") or item["hub_status"] in {"created_in_hub", "linked_to_existing"}:
        blockers.append("Posten er allerede oprettet eller linket")
    duplicate = execute_query_single(
        "SELECT id FROM sag_subscriptions WHERE migration_source_item_id=%s", (item_id,)
    )
    if duplicate:
        blockers.append(f"Kildeposten er allerede oprettet som abonnement {duplicate['id']}")
    if item.get("hub_sag_id"):
        case = execute_query_single(
            "SELECT id FROM sag_sager WHERE id=%s AND customer_id=%s",
            (item["hub_sag_id"], item.get("hub_customer_id")),
        )
        if not case:
            blockers.append("Den valgte sag tilhører ikke kunden")
    preview = {
        "customer": item.get("customer_name"), "customer_no": item.get("customer_no"),
        "product": item.get("product_name"), "product_code": item.get("product_code"),
        "amount": item.get("amount"), "quantity": item.get("quantity"),
        "frequency": _hub_interval(item.get("billing_frequency")),
        "start_date": item.get("period_from") or item.get("invoice_date") or date.today(),
        "end_date": item.get("period_to"), "source": item.get("source_system"),
        "source_record_id": item.get("source_record_id"),
        "hub_customer_id": item.get("hub_customer_id"), "hub_sag_id": item.get("hub_sag_id"),
    }
    return {
        "ready": not blockers, "blockers": blockers, "preview": preview,
        "token": preflight_token(item, current_user) if not blockers else None,
    }


@router.post("/items/{item_id}/create-in-hub")
async def create_in_hub(
    item_id: int, payload: CreatePayload, request: Request,
    current_user: dict = Depends(require_permission("migration_center.create")),
):
    conn = get_db_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
            item = _load_item(item_id, for_update=True, cursor=cursor)
            _ensure_item_mutable(item)
            verify_preflight(payload.preflight_token, item, current_user)
            if not item.get("hub_customer_id") or not item.get("hub_sag_id"):
                raise HTTPException(status_code=409, detail="Kunde og sag skal være valgt")
            if item.get("hub_record_id") or item["hub_status"] in {"created_in_hub", "linked_to_existing"}:
                return {"success": True, "idempotent": True, "hub_record_id": item.get("hub_record_id")}
            cursor.execute(
                "SELECT id FROM sag_subscriptions WHERE migration_source_item_id=%s", (item_id,)
            )
            duplicate = cursor.fetchone()
            if duplicate:
                cursor.execute(
                    """
                    UPDATE migration_center_session_items SET hub_record_id=%s, hub_status='created_in_hub',
                      updated_at=CURRENT_TIMESTAMP WHERE id=%s
                    """,
                    (duplicate["id"], item_id),
                )
                conn.commit()
                return {"success": True, "idempotent": True, "hub_record_id": duplicate["id"]}
            start_date = item.get("period_from") or item.get("invoice_date") or date.today()
            interval = _hub_interval(item.get("billing_frequency"))
            amount = Decimal(str(item.get("amount") or 0))
            quantity = Decimal(str(item.get("quantity") or 1))
            unit_price = amount / quantity if quantity else amount
            cursor.execute(
                """
                INSERT INTO sag_subscriptions
                (sag_id, customer_id, product_name, billing_interval, billing_day, price,
                 start_date, end_date, period_start, next_invoice_date, status, notes,
                 created_by_user_id, migration_source_item_id)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,'draft',%s,%s,%s)
                RETURNING id, subscription_number
                """,
                (
                    item["hub_sag_id"], item["hub_customer_id"], item["product_name"], interval,
                    min(max(start_date.day, 1), 31), amount, start_date, item.get("period_to"),
                    start_date, start_date, f"Manuelt oprettet via migreringscenter fra {item['source_system']} "
                    f"{item['source_record_id']} (idempotency {payload.idempotency_key})",
                    user_id(current_user), item_id,
                ),
            )
            subscription = dict(cursor.fetchone())
            cursor.execute(
                """
                INSERT INTO sag_subscription_items
                (subscription_id, line_no, description, quantity, unit_price, line_total, period_from, period_to)
                VALUES (%s,1,%s,%s,%s,%s,%s,%s)
                """,
                (
                    subscription["id"], item["product_name"], quantity or Decimal("1"),
                    unit_price, amount, item.get("period_from"), item.get("period_to"),
                ),
            )
            cursor.execute(
                """
                UPDATE migration_center_session_items
                SET hub_record_id=%s, hub_status='created_in_hub', approval_status='approved',
                    creation_idempotency_key=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s
                """,
                (subscription["id"], payload.idempotency_key, item_id),
            )
        conn.commit()
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        audit(
            request=request, current_user=current_user, action="create_in_hub",
            entity_type="session_item", entity_id=item_id, item_id=item_id, success=False,
            error_message=str(exc),
        )
        raise HTTPException(status_code=500, detail=f"Oprettelsen fejlede uden lokal statusændring: {exc}") from exc
    finally:
        release_db_connection(conn)
    audit(
        request=request, current_user=current_user, action="created_in_hub",
        entity_type=item["entity_type"], entity_id=item["source_record_id"], session_id=item["session_id"],
        item_id=item_id, old_value={"hub_status": item["hub_status"]},
        new_value={"hub_status": "created_in_hub", "hub_record_id": subscription["id"]},
        source_hash_value=item["source_hash"],
    )
    return {"success": True, "idempotent": False, "hub_record_id": subscription["id"], "subscription": subscription}


@router.post("/items/{item_id}/hub-link")
async def link_hub_record(
    item_id: int, payload: LinkPayload, request: Request,
    current_user: dict = Depends(require_permission("migration_center.review")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    subscription = execute_query_single(
        "SELECT id, subscription_number, customer_id FROM sag_subscriptions WHERE id=%s AND customer_id=%s",
        (payload.hub_id, item.get("hub_customer_id")),
    )
    if not subscription:
        raise HTTPException(status_code=404, detail="Abonnementet findes ikke på den valgte kunde")
    execute_query(
        """
        UPDATE migration_center_session_items SET hub_record_id=%s, hub_status='linked_to_existing',
          approval_status='approved', updated_at=CURRENT_TIMESTAMP WHERE id=%s
        """,
        (payload.hub_id, item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="linked_to_existing",
        entity_type=item["entity_type"], entity_id=item["source_record_id"], session_id=item["session_id"],
        item_id=item_id, new_value={"hub_record_id": payload.hub_id}, source_hash_value=item["source_hash"],
    )
    return {"success": True, "subscription": subscription}


@router.post("/items/{item_id}/verify")
async def verify_item(
    item_id: int, request: Request,
    current_user: dict = Depends(require_permission("migration_center.review")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    if item["hub_status"] not in {"created_in_hub", "linked_to_existing"}:
        raise HTTPException(status_code=409, detail="Posten skal oprettes eller linkes før kontrol")
    execute_query(
        """
        UPDATE migration_center_session_items
        SET approval_status='verified', hub_status='verified', verified_at=CURRENT_TIMESTAMP,
            verified_by_user_id=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s
        """,
        (user_id(current_user), item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="marked_verified",
        entity_type=item["entity_type"], entity_id=item["source_record_id"], session_id=item["session_id"],
        item_id=item_id, source_hash_value=item["source_hash"],
    )
    return {"success": True}


@router.post("/items/{item_id}/ignore")
async def ignore_item(
    item_id: int, payload: ReasonPayload, request: Request,
    current_user: dict = Depends(require_permission("migration_center.review")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    execute_query(
        """
        UPDATE migration_center_session_items SET approval_status='ignored', ignore_reason=%s,
          updated_at=CURRENT_TIMESTAMP WHERE id=%s
        """,
        (payload.reason, item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="ignored",
        entity_type=item["entity_type"], entity_id=item["source_record_id"], session_id=item["session_id"],
        item_id=item_id, new_value={"reason": payload.reason}, source_hash_value=item["source_hash"],
    )
    return {"success": True}


@router.put("/items/{item_id}/note")
async def update_note(
    item_id: int, payload: NotePayload, request: Request,
    current_user: dict = Depends(require_permission("migration_center.review")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    execute_query(
        "UPDATE migration_center_session_items SET manual_note=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s",
        (payload.note, item_id), fetch=False,
    )
    audit(
        request=request, current_user=current_user, action="note_updated",
        entity_type=item["entity_type"], entity_id=item["source_record_id"], session_id=item["session_id"],
        item_id=item_id, old_value={"note": item.get("manual_note")}, new_value={"note": payload.note},
        source_hash_value=item["source_hash"],
    )
    return {"success": True}


@router.post("/items/{item_id}/lock")
async def lock_item(
    item_id: int, request: Request,
    current_user: dict = Depends(require_any_permission("migration_center.lock", "migration_center.retry")),
):
    item = _load_item(item_id)
    _ensure_item_mutable(item)
    if item["approval_status"] not in {"verified", "ignored"}:
        raise HTTPException(status_code=409, detail="Posten skal være verificeret eller ignoreret før låsning")
    if item["match_status"] in {"conflict", "source_changed"}:
        raise HTTPException(status_code=409, detail="Åben konflikt eller kildeændring blokerer låsning")
    # Economic is read-only and has no external lock target. CRM records require configured custom fields.
    lock_field = None
    if item["source_system"] in {"vtiger", "simply"}:
        lock_field = (
            getattr(settings, "MIGRATION_CENTER_VTIGER_LOCK_FIELD", "")
            if item["source_system"] == "vtiger"
            else getattr(settings, "MIGRATION_CENTER_SIMPLY_LOCK_FIELD", "")
        )
        if not lock_field:
            raise HTTPException(status_code=409, detail="CRM-låsefeltet er ikke konfigureret")
    # Persist pending before any remote call so a crash cannot look like a completed lock.
    pending = execute_query_single(
        """
        INSERT INTO migration_center_lock_operations
        (session_item_id, status, external_system, request_payload, requested_by_user_id)
        VALUES (%s,'pending',%s,%s,%s) RETURNING id
        """,
        (
            item_id, item["source_system"],
            Json({"source_record_id": item["source_record_id"], "field": lock_field}),
            user_id(current_user),
        ),
    )
    execute_query(
        "UPDATE migration_center_session_items SET lock_status='locking_pending', updated_at=CURRENT_TIMESTAMP WHERE id=%s",
        (item_id,), fetch=False,
    )
    if lock_field:
        try:
            lock_values = {
                lock_field: "1",
                "hub_locked_at": date.today().isoformat(),
                "hub_transfer_status": "locked",
            }
            if item["source_system"] == "vtiger":
                await get_vtiger_service().update_subscription(item["source_record_id"], lock_values)
            else:
                async with SimplyCRMService() as service:
                    await service._ensure_session()
                    async with service.session.post(
                        f"{service.base_url}/webservice.php",
                        data={
                            "operation": "update",
                            "sessionName": service.session_name,
                            "elementType": "SalesOrder",
                            "element": json.dumps({"id": item["source_record_id"], **lock_values}),
                        },
                    ) as response:
                        result = await response.json()
                        if not response.ok or not result.get("success"):
                            raise RuntimeError(str(result.get("error") or f"HTTP {response.status}"))
        except Exception as exc:
            execute_query(
                """
                UPDATE migration_center_session_items SET lock_status='lock_failed', updated_at=CURRENT_TIMESTAMP
                WHERE id=%s
                """,
                (item_id,), fetch=False,
            )
            execute_query(
                """
                UPDATE migration_center_lock_operations SET status='failed', error_message=%s,
                  completed_at=CURRENT_TIMESTAMP WHERE id=%s
                """,
                (str(exc), pending["id"]), fetch=False,
            )
            audit(
                request=request, current_user=current_user, action="lock_failed",
                entity_type=item["entity_type"], entity_id=item["source_record_id"],
                session_id=item["session_id"], item_id=item_id, success=False,
                error_message=str(exc), source_hash_value=item["source_hash"],
            )
            raise HTTPException(status_code=502, detail=f"Ekstern låsning fejlede: {exc}") from exc
    conn = get_db_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cursor:
            if item.get("hub_record_id"):
                cursor.execute(
                    """
                    UPDATE sag_subscriptions SET migration_locked=TRUE, migration_locked_at=CURRENT_TIMESTAMP,
                      migration_locked_by_user_id=%s WHERE id=%s
                    """,
                    (user_id(current_user), item["hub_record_id"]),
                )
            cursor.execute(
                """
                UPDATE migration_center_session_items SET lock_status='locked', locked_at=CURRENT_TIMESTAMP,
                  locked_by_user_id=%s, updated_at=CURRENT_TIMESTAMP WHERE id=%s
                """,
                (user_id(current_user), item_id),
            )
            cursor.execute(
                """
                UPDATE migration_center_lock_operations SET status='succeeded', completed_at=CURRENT_TIMESTAMP
                WHERE id=%s
                """,
                (pending["id"],),
            )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        release_db_connection(conn)
    audit(
        request=request, current_user=current_user, action="locked",
        entity_type=item["entity_type"], entity_id=item["source_record_id"], session_id=item["session_id"],
        item_id=item_id, new_value={"lock_status": "locked"}, source_hash_value=item["source_hash"],
    )
    return {"success": True, "lock_status": "locked"}


@router.post("/sessions/{session_id}/customers/{customer_id}/lock")
async def lock_customer_items(
    session_id: int,
    customer_id: int,
    request: Request,
    current_user: dict = Depends(require_any_permission("migration_center.lock", "migration_center.retry")),
):
    ensure_writable(session_id)
    rows = execute_query(
        f"""
        SELECT * FROM migration_center_session_items
        WHERE session_id=%s AND hub_customer_id=%s
          AND {EXCLUDED_INVOICE_LINE_SQL}
          AND {subscription_like_item_sql()}
        ORDER BY id
        """,
        (session_id, customer_id),
    ) or []
    if not rows:
        raise HTTPException(status_code=404, detail="Ingen poster blev fundet for kunden")
    blockers = [
        row["id"] for row in rows
        if row["lock_status"] != "locked" and (
            row["approval_status"] not in {"verified", "ignored"}
            or row["match_status"] in {"conflict", "source_changed"}
        )
    ]
    if blockers:
        raise HTTPException(
            status_code=409,
            detail={"message": "Alle poster skal være behandlet uden åbne konflikter", "item_ids": blockers},
        )
    results = []
    for row in rows:
        if row["lock_status"] == "locked":
            continue
        try:
            result = await lock_item(int(row["id"]), request, current_user)
            results.append({"item_id": row["id"], **result})
        except HTTPException as exc:
            audit(
                request=request, current_user=current_user, action="customer_lock_partial",
                entity_type="customer", entity_id=customer_id, session_id=session_id,
                success=False, error_message=str(exc.detail), new_value={"completed": results},
            )
            raise
    audit(
        request=request, current_user=current_user, action="customer_locked",
        entity_type="customer", entity_id=customer_id, session_id=session_id,
        new_value={"items": [row["item_id"] for row in results]},
    )
    return {"success": True, "locked": results}


@router.get("/sessions/{session_id}/audit")
async def audit_log(
    session_id: int, limit: int = Query(200, ge=1, le=1000),
    current_user: dict = Depends(require_permission("migration_center.view")),
):
    return execute_query(
        """
        SELECT a.*, COALESCE(u.full_name,u.username) AS performed_by
        FROM migration_center_audit_log a
        LEFT JOIN users u ON u.user_id=a.performed_by_user_id
        WHERE a.session_id=%s ORDER BY a.performed_at DESC LIMIT %s
        """,
        (session_id, limit),
    ) or []


@router.get("/sessions/{session_id}/report.csv")
async def export_report(
    session_id: int,
    current_user: dict = Depends(require_permission("migration_center.export")),
):
    content = "\ufeff" + report_csv(session_id)
    return Response(
        content=content, media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": f'attachment; filename="migration-session-{session_id}.csv"'},
    )