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

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

import json
from datetime import date
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
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)


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("/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"'},
    )