from __future__ import annotations

import hashlib
import json
import re
from collections import Counter
from collections.abc import Mapping, Sequence
from datetime import datetime, timezone
from typing import Any
from urllib.parse import quote

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.external_references import (
    INTEGRATION_MATURITY_ORDER,
    ExternalObjectReference,
)
from govoplan_core.core.runtime import get_registry
from govoplan_core.core.search import search_index_writer
from govoplan_core.security.credential_envelopes import (
    CredentialAccessContext,
    CredentialEnvelopeError,
    resolve_credential_envelope,
)
from govoplan_connectors.backend.db.models import (
    ConnectorConfiguration,
    ConnectorDefinition,
    ConnectorKnowledgeObject,
    ConnectorKnowledgeProfile,
    ConnectorKnowledgeSyncRun,
)
from govoplan_connectors.backend.knowledge_schemas import (
    KnowledgeDiagnostic,
    KnowledgeDiscoveryResponse,
    KnowledgeExternalReferenceResponse,
    KnowledgeMigrationDryRunRequest,
    KnowledgeMigrationDryRunResponse,
    KnowledgeObjectItem,
    KnowledgeProfileCreateRequest,
    KnowledgeProfileItem,
    KnowledgeProfileUpdateRequest,
    KnowledgePublishRequest,
    KnowledgePublishResponse,
    KnowledgeSyncRequest,
    KnowledgeSyncRunItem,
)
from govoplan_connectors.backend.mediawiki_transport import (
    HttpMediaWikiTransport,
    MediaWikiPublishResult,
    MediaWikiTransport,
    MediaWikiTransportError,
)
from govoplan_connectors.backend.recovery import (
    ConnectorRecoveryError,
    begin_connector_external_mutation,
)


KNOWLEDGE_READ_SCOPE = "connectors:knowledge:read"
KNOWLEDGE_ADMIN_SCOPE = "connectors:knowledge:admin"
KNOWLEDGE_SYNC_SCOPE = "connectors:knowledge:sync"
KNOWLEDGE_PUBLISH_SCOPE = "connectors:knowledge:publish"
KNOWLEDGE_MIGRATE_SCOPE = "connectors:knowledge:migrate"
KNOWLEDGE_CAPABILITY = "connectors.external_knowledge"
KNOWLEDGE_INTERFACE_VERSION = "1.0.0"
KNOWLEDGE_PROVIDER_ID = "connectors.mediawiki.pages"
KNOWLEDGE_RESOURCE_TYPE = "external_knowledge_page"

_MACRO = re.compile(r"\{\{\s*([^{}|#]+)", re.MULTILINE)
_REDIRECT = re.compile(r"^\s*#redirect\s*\[\[([^\]]+)\]\]", re.IGNORECASE)


class KnowledgeConnectorError(RuntimeError):
    def __init__(self, code: str, message: str, *, retryable: bool = False) -> None:
        super().__init__(message)
        self.code = code
        self.retryable = retryable


def create_profile(
    session: Session,
    principal: ApiPrincipal,
    payload: KnowledgeProfileCreateRequest,
) -> KnowledgeProfileItem:
    _require_scope(principal, KNOWLEDGE_ADMIN_SCOPE)
    configuration = _configuration(
        session, principal.tenant_id, payload.configuration_id, active=False
    )
    _assert_knowledge_configuration(session, configuration)
    item = ConnectorKnowledgeProfile(
        tenant_id=principal.tenant_id,
        configuration_id=configuration.id,
        desired_maturity=payload.desired_maturity,
        source_authority_mode=payload.source_authority_mode,
        default_visibility=payload.default_visibility,
        default_acl_tokens=list(payload.default_acl_tokens),
        namespace_mappings=[item.model_dump(mode="json") for item in payload.namespace_mappings],
        updated_by=_actor_id(principal),
    )
    session.add(item)
    try:
        session.flush()
    except IntegrityError as exc:
        raise KnowledgeConnectorError(
            "profile_conflict",
            "This connector configuration already has a knowledge profile.",
        ) from exc
    _audit(
        session,
        principal,
        action="profile.created",
        object_type="connector_knowledge_profile",
        object_id=item.id,
        details={
            "configuration_id": configuration.id,
            "desired_maturity": item.desired_maturity,
            "source_authority_mode": item.source_authority_mode,
        },
    )
    response = _profile_item(item, configuration)
    session.commit()
    return response


def update_profile(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str,
    payload: KnowledgeProfileUpdateRequest,
    registry: object | None = None,
) -> KnowledgeProfileItem:
    _require_scope(principal, KNOWLEDGE_ADMIN_SCOPE)
    item = _profile(session, principal.tenant_id, profile_id)
    if item.resource_revision != payload.expected_resource_revision:
        raise KnowledgeConnectorError(
            "profile_conflict", "The profile changed; reload it before saving."
        )
    for field_name in (
        "status",
        "desired_maturity",
        "source_authority_mode",
        "default_visibility",
    ):
        value = getattr(payload, field_name)
        if value is not None:
            setattr(item, field_name, value)
    if payload.default_acl_tokens is not None:
        item.default_acl_tokens = list(payload.default_acl_tokens)
    if payload.namespace_mappings is not None:
        item.namespace_mappings = [
            mapping.model_dump(mode="json") for mapping in payload.namespace_mappings
        ]
    _validate_effective_acl(item.default_visibility, item.default_acl_tokens)
    affected = tuple(
        session.scalars(
            select(ConnectorKnowledgeObject).where(
                ConnectorKnowledgeObject.tenant_id == principal.tenant_id,
                ConnectorKnowledgeObject.profile_id == item.id,
                ConnectorKnowledgeObject.object_type == "page",
                ConnectorKnowledgeObject.status != "deleted",
            )
        )
    )
    for row in affected:
        mapping = _namespace_mapping(item, row.namespace_id)
        if mapping is None or not bool(mapping.get("include", True)):
            continue
        data = dict(row.mapped_data or {})
        data["target_space_ref"] = mapping.get("target_space_ref")
        data["target_path"] = _target_path(mapping, row.title)
        if data.get("permission_source") == "configured_fallback":
            visibility = str(mapping.get("visibility") or item.default_visibility)
            tokens = list(mapping.get("acl_tokens") or item.default_acl_tokens)
            _validate_effective_acl(visibility, tokens)
            row.visibility = visibility
            row.acl_tokens = tokens
        row.mapped_data = data
        row.content_hash = _mapped_content_hash(
            data,
            visibility=row.visibility,
            acl_tokens=row.acl_tokens,
        )
        row.resource_revision += 1
        row.observed_at = _now()
    item.resource_revision += 1
    item.updated_by = _actor_id(principal)
    session.flush()
    configuration = _configuration(
        session, principal.tenant_id, item.configuration_id, active=False
    )
    _audit(
        session,
        principal,
        action="profile.updated",
        object_type="connector_knowledge_profile",
        object_id=item.id,
        details={
            "resource_revision": item.resource_revision,
            "status": item.status,
            "desired_maturity": item.desired_maturity,
        },
    )
    search_diagnostics: list[dict[str, Any]] = []
    _update_search(
        session,
        principal,
        affected,
        registry=registry or get_registry(),
        diagnostics=search_diagnostics,
        source_active=item.status == "active",
    )
    if search_diagnostics:
        item.health_details = {
            **dict(item.health_details or {}),
            "search_diagnostics": search_diagnostics,
        }
    response = _profile_item(item, configuration)
    session.commit()
    return response


def list_profiles(
    session: Session, principal: ApiPrincipal
) -> tuple[KnowledgeProfileItem, ...]:
    _require_any_scope(principal, KNOWLEDGE_READ_SCOPE, KNOWLEDGE_ADMIN_SCOPE)
    rows = tuple(
        session.scalars(
            select(ConnectorKnowledgeProfile)
            .where(ConnectorKnowledgeProfile.tenant_id == principal.tenant_id)
            .order_by(ConnectorKnowledgeProfile.created_at, ConnectorKnowledgeProfile.id)
        )
    )
    configurations = _configurations(session, principal.tenant_id, rows)
    return tuple(_profile_item(row, configurations[row.configuration_id]) for row in rows)


def discover_profile(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str,
    transport: MediaWikiTransport | None = None,
) -> KnowledgeDiscoveryResponse:
    _require_any_scope(principal, KNOWLEDGE_ADMIN_SCOPE, KNOWLEDGE_SYNC_SCOPE)
    item = _profile(session, principal.tenant_id, profile_id)
    configuration = _configuration(
        session, principal.tenant_id, item.configuration_id, active=True
    )
    _assert_knowledge_configuration(session, configuration)
    endpoint = _endpoint(configuration)
    credential = _credential(session, principal, configuration)
    try:
        payload = (transport or HttpMediaWikiTransport()).discover(
            endpoint_url=endpoint,
            credential=credential,
        )
    except MediaWikiTransportError as exc:
        item.health_status = "unavailable"
        item.health_details = {
            "code": exc.code,
            "retryable": exc.retryable,
            "summary": str(exc),
        }
        session.commit()
        raise KnowledgeConnectorError(
            "discovery_unavailable", str(exc), retryable=exc.retryable
        ) from exc
    discovery = _discovery(payload, credential_present=bool(configuration.credential_ref))
    item.product = discovery["product"]
    item.product_version = discovery["product_version"]
    item.capabilities = discovery["capabilities"]
    item.discovered_maturity = discovery["maturity"]
    item.discovery_revision = discovery["revision"]
    item.discovery_evidence = {
        "api_version": discovery["api_version"],
        "extensions": discovery["extensions"],
        "namespaces": discovery["namespaces"],
        "capability_count": len(discovery["capabilities"]),
    }
    item.health_status = discovery["health_status"]
    item.health_details = {
        "diagnostics": [value.model_dump(mode="json") for value in discovery["diagnostics"]],
        "checked_at": _now().isoformat(),
    }
    item.discovered_at = _now()
    item.resource_revision += 1
    session.flush()
    _audit(
        session,
        principal,
        action="profile.discovered",
        object_type="connector_knowledge_profile",
        object_id=item.id,
        details={
            "product": item.product,
            "product_version": item.product_version,
            "maturity": item.discovered_maturity,
            "health": item.health_status,
            "discovery_revision": item.discovery_revision,
        },
    )
    response = KnowledgeDiscoveryResponse(
        profile=_profile_item(item, configuration),
        product=discovery["product"],
        product_version=discovery["product_version"],
        api_version=discovery["api_version"],
        capabilities=discovery["capabilities"],
        namespaces=discovery["namespaces"],
        extensions=discovery["extensions"],
        maturity=discovery["maturity"],
        health_status=discovery["health_status"],
        diagnostics=discovery["diagnostics"],
        revision=discovery["revision"],
    )
    session.commit()
    return response


def synchronize_profile(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str,
    payload: KnowledgeSyncRequest,
    transport: MediaWikiTransport | None = None,
    registry: object | None = None,
) -> KnowledgeSyncRunItem:
    _require_scope(principal, KNOWLEDGE_SYNC_SCOPE)
    item = _profile(session, principal.tenant_id, profile_id)
    if item.status != "active":
        raise KnowledgeConnectorError("profile_paused", "The knowledge profile is paused.")
    configuration = _configuration(
        session, principal.tenant_id, item.configuration_id, active=True
    )
    _assert_knowledge_configuration(session, configuration)
    if item.discovered_at is None:
        raise KnowledgeConnectorError(
            "profile_not_discovered", "Discover the provider before synchronization."
        )
    if not _supports(item.discovered_maturity, "synchronize") or not _supports(
        item.desired_maturity, "synchronize"
    ):
        raise KnowledgeConnectorError(
            "synchronization_unsupported",
            "The discovered provider maturity does not support synchronization.",
        )
    cursor = None if payload.force_full else payload.cursor or item.last_sync_cursor
    mode = "backfill" if payload.force_full else "delta"
    request_hash = _hash(
        {
            "profile_id": item.id,
            "mode": mode,
            "cursor": cursor,
            "limit": payload.limit,
            "discovery_revision": item.discovery_revision,
        }
    )
    replay = _run_replay(
        session,
        tenant_id=principal.tenant_id,
        profile_id=item.id,
        mode=mode,
        idempotency_key=payload.idempotency_key,
        request_hash=request_hash,
    )
    if replay is not None:
        return _run_item(replay)
    run = ConnectorKnowledgeSyncRun(
        tenant_id=principal.tenant_id,
        profile_id=item.id,
        mode=mode,
        idempotency_key=payload.idempotency_key,
        request_hash=request_hash,
        status="running",
        cursor_before=cursor,
        counts={},
        effects=[],
        diagnostics=[],
        provenance={
            "configuration_revision": configuration.resource_revision,
            "configuration_hash": configuration.effective_hash,
            "discovery_revision": item.discovery_revision,
        },
        created_by=_actor_id(principal),
        started_at=_now(),
    )
    session.add(run)
    session.flush()
    credential = _credential(session, principal, configuration)
    try:
        batch = (transport or HttpMediaWikiTransport()).changes(
            endpoint_url=_endpoint(configuration),
            credential=credential,
            cursor=cursor,
            limit=payload.limit,
            force_full=payload.force_full,
        )
    except MediaWikiTransportError as exc:
        run.status = "failed"
        run.diagnostics = [
            _diagnostic(
                "error",
                exc.code,
                str(exc),
                retryable=exc.retryable,
            )
        ]
        run.finished_at = _now()
        item.health_status = "unavailable"
        item.health_details = {"code": exc.code, "retryable": exc.retryable}
        session.commit()
        raise KnowledgeConnectorError(
            "synchronization_unavailable", str(exc), retryable=exc.retryable
        ) from exc

    effects: list[dict[str, Any]] = []
    diagnostics: list[dict[str, Any]] = []
    changed: list[ConnectorKnowledgeObject] = []
    for raw in batch.changes:
        mapping = _namespace_mapping(item, raw.get("ns"))
        change_kind = str(raw.get("change_kind") or "upsert")
        external_id = (
            _optional_text(raw.get("pageid"))
            or _optional_text(raw.get("title"))
            or "unknown"
            if change_kind == "delete"
            else _external_page_id(raw)
        )
        if mapping is None or not bool(mapping.get("include", True)):
            effects.append(
                {
                    "effect": "ignored",
                    "external_id": external_id,
                    "title": _optional_text(raw.get("title")) or "Unmapped page",
                    "reason_code": "namespace_not_mapped",
                }
            )
            diagnostics.append(
                _diagnostic(
                    "warning",
                    "namespace_not_mapped",
                    "The source namespace is excluded or has no target mapping.",
                    object_ref=external_id,
                )
            )
            continue
        if change_kind == "delete":
            stored, effect = _delete_object(
                session,
                item,
                raw,
                mapping=mapping,
                cursor=raw.get("change_cursor") or batch.next_cursor,
            )
            effects.append(effect)
            if effect.get("reason_code"):
                diagnostics.append(
                    _diagnostic(
                        "warning",
                        str(effect["reason_code"]),
                        "The deletion could not be matched to a synchronized stable page id; a separate tombstone was retained.",
                        object_ref=str(effect["external_id"]),
                    )
                )
            if stored is not None:
                changed.append(stored)
            continue
        mapped, mapped_diagnostics = _map_page(item, raw, mapping=mapping)
        stored, effect = _upsert_object(
            session,
            item,
            mapped,
            cursor=raw.get("change_cursor") or batch.next_cursor,
        )
        effects.append(effect)
        diagnostics.extend(mapped_diagnostics)
        if effect["effect"] != "unchanged":
            changed.append(stored)

    session.flush()
    _update_search(
        session,
        principal,
        changed,
        registry=registry or get_registry(),
        diagnostics=diagnostics,
    )
    counts = dict(Counter(str(effect["effect"]) for effect in effects))
    run.status = "completed"
    run.cursor_after = batch.next_cursor
    run.high_watermark = batch.high_watermark
    run.counts = counts
    run.effects = effects
    run.diagnostics = diagnostics
    run.provenance = {**run.provenance, "transport": dict(batch.evidence)}
    run.finished_at = _now()
    item.last_sync_cursor = batch.next_cursor
    item.last_high_watermark = batch.high_watermark
    item.health_status = "healthy"
    item.health_details = {
        "last_sync_at": run.finished_at.isoformat(),
        "complete": batch.complete,
        "counts": counts,
    }
    item.resource_revision += 1
    session.flush()
    _audit(
        session,
        principal,
        action="profile.synchronized",
        object_type="connector_knowledge_sync_run",
        object_id=run.id,
        details={
            "profile_id": item.id,
            "mode": mode,
            "counts": counts,
            "complete": batch.complete,
            "high_watermark": batch.high_watermark,
        },
    )
    response = _run_item(run)
    session.commit()
    return response


def list_objects(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str,
    cursor: str | None = None,
    limit: int = 100,
) -> tuple[tuple[KnowledgeObjectItem, ...], str | None]:
    _require_any_scope(principal, KNOWLEDGE_READ_SCOPE, KNOWLEDGE_ADMIN_SCOPE)
    item = _profile(session, principal.tenant_id, profile_id)
    statement = select(ConnectorKnowledgeObject).where(
        ConnectorKnowledgeObject.tenant_id == principal.tenant_id,
        ConnectorKnowledgeObject.profile_id == item.id,
    )
    if cursor:
        statement = statement.where(ConnectorKnowledgeObject.id > cursor)
    rows = tuple(
        session.scalars(
            statement.order_by(ConnectorKnowledgeObject.id.asc()).limit(limit + 1)
        )
    )
    selected = rows[:limit]
    next_cursor = selected[-1].id if len(rows) > limit and selected else None
    return tuple(_object_item(row, item) for row in selected), next_cursor


def list_runs(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str | None = None,
    limit: int = 100,
) -> tuple[KnowledgeSyncRunItem, ...]:
    _require_any_scope(principal, KNOWLEDGE_READ_SCOPE, KNOWLEDGE_ADMIN_SCOPE)
    statement = select(ConnectorKnowledgeSyncRun).where(
        ConnectorKnowledgeSyncRun.tenant_id == principal.tenant_id
    )
    if profile_id:
        _profile(session, principal.tenant_id, profile_id)
        statement = statement.where(ConnectorKnowledgeSyncRun.profile_id == profile_id)
    return tuple(
        _run_item(row)
        for row in session.scalars(
            statement.order_by(
                ConnectorKnowledgeSyncRun.started_at.desc(),
                ConnectorKnowledgeSyncRun.id.desc(),
            ).limit(limit)
        )
    )


def migration_dry_run(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str,
    payload: KnowledgeMigrationDryRunRequest,
) -> KnowledgeMigrationDryRunResponse:
    _require_scope(principal, KNOWLEDGE_MIGRATE_SCOPE)
    profile = _profile(session, principal.tenant_id, profile_id)
    if (
        profile.discovered_at is None
        or not _supports(profile.discovered_maturity, "migrate")
        or not _supports(profile.desired_maturity, "migrate")
    ):
        raise KnowledgeConnectorError(
            "migration_unsupported",
            "The discovered provider maturity does not support migration.",
        )
    request_hash = _hash(payload.model_dump(mode="json"))
    replay = _run_replay(
        session,
        tenant_id=principal.tenant_id,
        profile_id=profile.id,
        mode="migration_dry_run",
        idempotency_key=payload.idempotency_key,
        request_hash=request_hash,
    )
    if replay is not None:
        return _migration_response(replay, payload.target_space_ref)
    rows = tuple(
        session.scalars(
            select(ConnectorKnowledgeObject)
            .where(
                ConnectorKnowledgeObject.tenant_id == principal.tenant_id,
                ConnectorKnowledgeObject.profile_id == profile.id,
                ConnectorKnowledgeObject.object_type == "page",
                ConnectorKnowledgeObject.status != "deleted",
            )
            .order_by(ConnectorKnowledgeObject.id.asc())
            .limit(payload.max_items + 1)
        )
    )
    selected = rows[: payload.max_items]
    truncated = len(rows) > payload.max_items
    targets = {item.path: item for item in payload.existing_targets}
    supported_macros = {item.casefold() for item in payload.supported_macros}
    effects: list[dict[str, Any]] = []
    diagnostics: list[dict[str, Any]] = []
    for row in selected:
        data = dict(row.mapped_data or {})
        if data.get("target_space_ref") != payload.target_space_ref:
            continue
        path = str(data.get("target_path") or row.title)
        target = targets.get(path)
        effect = "create"
        reason = None
        if target is not None:
            if target.source_external_id == row.external_id:
                effect = "update"
            else:
                effect = "conflict"
                reason = "target_path_conflict"
                diagnostics.append(
                    _diagnostic(
                        "error",
                        "target_path_conflict",
                        "Another target page already uses the mapped path.",
                        object_ref=row.external_id,
                        field="path",
                        details={"target_path": path},
                    )
                )
        unsupported = sorted(
            macro
            for macro in set(data.get("macros") or ())
            if str(macro).casefold() not in supported_macros
        )
        for macro in unsupported:
            diagnostics.append(
                _diagnostic(
                    "warning",
                    "unsupported_macro",
                    "The source page uses a macro that is not declared as supported.",
                    object_ref=row.external_id,
                    field="body",
                    details={"macro": macro},
                )
            )
        attachment_names = {
            str(value.get("name") or "")
            for value in data.get("files") or ()
            if isinstance(value, Mapping)
        }
        conflicts = sorted(
            attachment_names.intersection(
                set(target.attachment_names if target is not None else ())
            )
        )
        for name in conflicts:
            diagnostics.append(
                _diagnostic(
                    "error",
                    "attachment_name_conflict",
                    "A target attachment already uses this source filename.",
                    object_ref=row.external_id,
                    field="files",
                    details={"name": name, "target_path": path},
                )
            )
            effect = "conflict"
            reason = "attachment_name_conflict"
        effects.append(
            {
                "effect": effect,
                "external_id": row.external_id,
                "source_revision": row.source_revision,
                "target_space_ref": payload.target_space_ref,
                "target_path": path,
                "title": row.title,
                "unsupported_macros": unsupported,
                "attachment_count": len(attachment_names),
                "reason_code": reason,
            }
        )
    if truncated:
        diagnostics.append(
            _diagnostic(
                "warning",
                "migration_truncated",
                "The migration preview reached its configured item limit.",
            )
        )
    counts = dict(Counter(str(effect["effect"]) for effect in effects))
    fingerprint = _hash(
        [(row.external_id, row.source_revision, row.content_hash) for row in selected]
    )
    run = ConnectorKnowledgeSyncRun(
        tenant_id=principal.tenant_id,
        profile_id=profile.id,
        mode="migration_dry_run",
        idempotency_key=payload.idempotency_key,
        request_hash=request_hash,
        status="completed",
        cursor_before=None,
        cursor_after=None,
        high_watermark=profile.last_high_watermark,
        counts=counts,
        effects=effects,
        diagnostics=diagnostics,
        provenance={
            "target_space_ref": payload.target_space_ref,
            "source_revision": profile.last_high_watermark or profile.discovery_revision,
            "source_fingerprint": fingerprint,
            "truncated": truncated,
        },
        created_by=_actor_id(principal),
        started_at=_now(),
        finished_at=_now(),
    )
    session.add(run)
    session.flush()
    _audit(
        session,
        principal,
        action="migration.previewed",
        object_type="connector_knowledge_sync_run",
        object_id=run.id,
        details={
            "profile_id": profile.id,
            "target_space_ref": payload.target_space_ref,
            "counts": counts,
            "truncated": truncated,
        },
    )
    response = _migration_response(run, payload.target_space_ref)
    session.commit()
    return response


def publish_page(
    session: Session,
    principal: ApiPrincipal,
    *,
    profile_id: str,
    external_page_id: str,
    payload: KnowledgePublishRequest,
    transport: MediaWikiTransport | None = None,
    registry: object | None = None,
    durable_recovery: bool = True,
) -> KnowledgePublishResponse:
    _require_scope(principal, KNOWLEDGE_PUBLISH_SCOPE)
    profile = _profile(session, principal.tenant_id, profile_id)
    if profile.status != "active":
        raise KnowledgeConnectorError("profile_paused", "The knowledge profile is paused.")
    if (
        profile.discovered_at is None
        or "publish" not in profile.capabilities
        or not _supports(profile.desired_maturity, "publish")
    ):
        raise KnowledgeConnectorError(
            "publication_unsupported",
            "The discovered provider maturity does not support publication.",
        )
    configuration = _configuration(
        session, principal.tenant_id, profile.configuration_id, active=True
    )
    request_hash = _hash(
        {
            "profile_id": profile.id,
            "external_page_id": external_page_id,
            **payload.model_dump(mode="json"),
        }
    )
    replay = _run_replay(
        session,
        tenant_id=principal.tenant_id,
        profile_id=profile.id,
        mode="publish",
        idempotency_key=payload.idempotency_key,
        request_hash=request_hash,
    )
    if replay is not None:
        stored = _object_by_external_id(
            session, profile.id, "page", external_page_id
        )
        if stored is None:
            raise KnowledgeConnectorError(
                "publication_reconciliation_required",
                "The publication evidence exists but the local reference is unavailable.",
            )
        return KnowledgePublishResponse(
            run=_run_item(replay),
            external_reference=_external_reference_response(stored, profile),
            accepted=replay.status == "completed",
            outcome_unknown=replay.status == "outcome_unknown",
        )
    run = ConnectorKnowledgeSyncRun(
        tenant_id=principal.tenant_id,
        profile_id=profile.id,
        mode="publish",
        idempotency_key=payload.idempotency_key,
        request_hash=request_hash,
        status="prepared",
        counts={},
        effects=[],
        diagnostics=[],
        provenance={
            "configuration_revision": configuration.resource_revision,
            "configuration_hash": configuration.effective_hash,
            "expected_external_revision": payload.expected_external_revision,
        },
        created_by=_actor_id(principal),
        started_at=_now(),
    )
    session.add(run)
    session.flush()
    recovery = None
    if durable_recovery:
        try:
            recovery = begin_connector_external_mutation(
                session,
                tenant_id=principal.tenant_id,
                provider_id=f"mediawiki:{profile.id}",
                idempotency_key=payload.idempotency_key,
                request_sha256=request_hash,
                source_revision=payload.expected_external_revision,
                cursor=profile.last_sync_cursor,
                dry_run_evidence={
                    "profile_revision": profile.resource_revision,
                    "configuration_revision": configuration.resource_revision,
                    "title": payload.title,
                },
                resource_type="external_knowledge_page",
                resource_id=external_page_id,
            )
        except ConnectorRecoveryError as exc:
            session.rollback()
            raise KnowledgeConnectorError("recovery_unavailable", str(exc)) from exc
    credential = _credential(session, principal, configuration)
    try:
        result = (transport or HttpMediaWikiTransport()).publish(
            endpoint_url=_endpoint(configuration),
            credential=credential,
            title=payload.title,
            body=payload.body,
            summary=payload.summary,
            expected_revision=payload.expected_external_revision,
            minor=payload.minor,
        )
    except MediaWikiTransportError as exc:
        run.status = "outcome_unknown" if exc.outcome_unknown else "rejected"
        run.diagnostics = [
            _diagnostic(
                "error", exc.code, str(exc), retryable=exc.retryable
            )
        ]
        run.finished_at = _now()
        session.commit()
        if recovery is not None:
            if exc.outcome_unknown:
                recovery.outcome_unknown(summary=str(exc), provider_code=exc.code)
            else:
                recovery.reject(summary=str(exc), provider_code=exc.code)
        if exc.outcome_unknown:
            raise KnowledgeConnectorError(
                "publication_outcome_unknown",
                "Publication outcome is unknown; inspect the provider before retrying.",
            ) from exc
        raise KnowledgeConnectorError("publication_rejected", str(exc)) from exc
    stored = _store_published_result(
        session,
        profile,
        external_page_id=external_page_id,
        payload=payload,
        result=result,
    )
    run.status = "completed"
    run.counts = {"update": 1}
    run.effects = [
        {
            "effect": "update",
            "external_id": stored.external_id,
            "source_revision": stored.source_revision,
            "title": stored.title,
        }
    ]
    run.provenance = {**run.provenance, "provider": dict(result.evidence)}
    run.finished_at = _now()
    session.flush()
    diagnostics: list[dict[str, Any]] = []
    _update_search(
        session,
        principal,
        [stored],
        registry=registry or get_registry(),
        diagnostics=diagnostics,
    )
    if diagnostics:
        run.diagnostics = diagnostics
    _audit(
        session,
        principal,
        action="page.published",
        object_type="external_knowledge_page",
        object_id=stored.id,
        details={
            "profile_id": profile.id,
            "external_page_id": stored.external_id,
            "external_revision_id": stored.external_revision_id,
            "recovery_operation_id": recovery.operation_id if recovery else None,
        },
    )
    response = KnowledgePublishResponse(
        run=_run_item(run),
        external_reference=_external_reference_response(stored, profile),
        accepted=True,
        outcome_unknown=False,
    )
    if recovery is not None:
        recovery.commit_success(
            session,
            provider_evidence={
                "verified": True,
                "checks": {
                    "provider_result": "Success",
                    "stable_page_id": result.page_id,
                    "accepted_revision_id": result.revision_id,
                },
                "page_id": result.page_id,
                "revision_id": result.revision_id,
                "canonical_url": result.canonical_url,
            },
        )
    else:
        session.commit()
    return response


class ExternalKnowledgeCapability:
    def list_references(
        self,
        session: object,
        principal: object,
        *,
        profile_id: str,
        limit: int = 100,
    ) -> Sequence[ExternalObjectReference]:
        if not isinstance(session, Session) or not isinstance(principal, ApiPrincipal):
            raise TypeError("External knowledge references require platform context.")
        rows, _cursor = list_objects(
            session, principal, profile_id=profile_id, limit=min(limit, 500)
        )
        profile = _profile(session, principal.tenant_id, profile_id)
        return tuple(
            ExternalObjectReference(
                system=profile.product,
                object_type=row.object_type,
                object_id=row.external_id,
                maturity=profile.discovered_maturity,
                authority_mode=profile.source_authority_mode,
                connector_id=profile.id,
                canonical_url=row.canonical_url,
                version=row.source_revision,
                metadata={"title": row.title, "status": row.status},
            )
            for row in rows
        )


def _discovery(
    payload: Mapping[str, Any], *, credential_present: bool
) -> dict[str, Any]:
    query = payload.get("query")
    if not isinstance(query, Mapping):
        raise KnowledgeConnectorError(
            "invalid_discovery", "MediaWiki site discovery returned no query metadata."
        )
    general = query.get("general")
    if not isinstance(general, Mapping):
        raise KnowledgeConnectorError(
            "invalid_discovery", "MediaWiki site discovery returned no general metadata."
        )
    extensions = tuple(
        _safe_extension(item)
        for item in query.get("extensions") or ()
        if isinstance(item, Mapping)
    )
    extension_names = {
        str(item.get("name") or "").casefold() for item in extensions
    }
    blue_spice = any("bluespice" in name for name in extension_names)
    generator = _optional_text(general.get("generator")) or "MediaWiki"
    product = "bluespice" if blue_spice else "mediawiki"
    product_version = _version(
        next(
            (
                str(item.get("version"))
                for item in extensions
                if "bluespice" in str(item.get("name") or "").casefold()
                and item.get("version")
            ),
            None,
        )
        or generator
    )
    namespaces = tuple(
        sorted(
            (
                {
                    "id": int(value.get("id", key)),
                    "name": _optional_text(value.get("name") or value.get("*")) or "",
                    "canonical": _optional_text(value.get("canonical")),
                    "content": bool(value.get("content")),
                }
                for key, value in (query.get("namespaces") or {}).items()
                if isinstance(value, Mapping)
            ),
            key=lambda value: value["id"],
        )
    )
    userinfo = query.get("userinfo")
    rights = {
        str(value)
        for value in (
            userinfo.get("rights")
            if isinstance(userinfo, Mapping)
            else ()
        )
    }
    capabilities = [
        "discover",
        "link",
        "search",
        "read",
        "synchronize",
        "migrate",
        "namespaces",
        "pages",
        "revisions",
        "users",
        "categories",
        "links",
        "files",
        "redirects",
        "deletion_tombstones",
        "permission_fallback",
    ]
    diagnostics: list[KnowledgeDiagnostic] = []
    if credential_present and (not rights or "edit" in rights):
        capabilities.append("publish")
    elif credential_present:
        diagnostics.append(
            KnowledgeDiagnostic(
                severity="warning",
                code="publish_right_missing",
                message="The configured principal did not advertise the edit right.",
            )
        )
    else:
        diagnostics.append(
            KnowledgeDiagnostic(
                severity="info",
                code="publish_credential_missing",
                message="Read and synchronization are available; publication requires a governed credential.",
            )
        )
    if blue_spice:
        capabilities.extend(("bluespice", "permission_metadata", "discussions"))
    else:
        diagnostics.extend(
            (
                KnowledgeDiagnostic(
                    severity="warning",
                    code="permission_mapping_defaulted",
                    message="The standard Action API does not expose complete page ACLs; configured namespace ACLs are used and every Search result is rechecked.",
                ),
                KnowledgeDiagnostic(
                    severity="warning",
                    code="discussion_mapping_lossy",
                    message="Discussion threads are mapped only when the provider supplies a supported discussion payload.",
                ),
            )
        )
    maturity = max(
        (value for value in capabilities if value in INTEGRATION_MATURITY_ORDER),
        key=INTEGRATION_MATURITY_ORDER.index,
    )
    safe_payload = {
        "generator": generator,
        "product": product,
        "product_version": product_version,
        "api_version": _optional_text(general.get("phpversion")),
        "namespaces": namespaces,
        "extensions": extensions,
        "capabilities": sorted(set(capabilities)),
    }
    return {
        **safe_payload,
        "maturity": maturity,
        "health_status": "healthy",
        "diagnostics": diagnostics,
        "revision": _hash(safe_payload),
    }


def _map_page(
    profile: ConnectorKnowledgeProfile,
    raw: Mapping[str, Any],
    *,
    mapping: Mapping[str, Any],
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    external_id = _external_page_id(raw)
    title = _optional_text(raw.get("title")) or f"Page {external_id}"
    revisions = tuple(
        item for item in raw.get("revisions") or () if isinstance(item, Mapping)
    )
    revision = revisions[0] if revisions else {}
    revision_id = _optional_text(
        revision.get("revid") or raw.get("lastrevid") or raw.get("revid")
    ) or _optional_text(raw.get("touched")) or "unknown"
    body = _revision_content(revision)
    timestamp = _parse_datetime(
        revision.get("timestamp") or raw.get("touched") or raw.get("timestamp")
    )
    visibility, acl_tokens, permission_diagnostics, permission_source = _mapped_acl(
        profile, mapping=mapping, raw=raw, external_id=external_id
    )
    categories = tuple(
        _without_prefix(str(item.get("title") or ""), "Category:")
        for item in raw.get("categories") or ()
        if isinstance(item, Mapping) and item.get("title")
    )
    links = tuple(
        {
            "title": str(item.get("title") or ""),
            "namespace_id": item.get("ns"),
        }
        for item in raw.get("links") or ()
        if isinstance(item, Mapping) and item.get("title")
    )
    files = tuple(
        _mapped_file(profile, item, observed_at=_now())
        for item in raw.get("images") or ()
        if isinstance(item, Mapping) and item.get("title")
    )
    discussions = tuple(
        _mapped_discussion(profile, item, external_id=external_id)
        for item in raw.get("discussions") or ()
        if isinstance(item, Mapping)
    )
    redirect_match = _REDIRECT.match(body)
    redirect_target = (
        _optional_text(raw.get("redirect_target_pageid"))
        or (redirect_match.group(1).strip() if redirect_match else None)
    )
    target_path = _target_path(mapping, title)
    canonical_url = _optional_text(raw.get("fullurl"))
    macros = tuple(sorted(set(match.strip() for match in _MACRO.findall(body))))
    revision_author = _optional_text(revision.get("user"))
    revision_author_reference = (
        ExternalObjectReference(
            system=profile.product if profile.product != "unknown" else "mediawiki",
            object_type="user",
            object_id=revision_author,
            maturity=profile.discovered_maturity,
            authority_mode="linked_reference",
            connector_id=profile.id,
            observed_at=_now(),
            metadata={"display_name": revision_author},
        ).to_dict()
        if revision_author
        else None
    )
    reference = ExternalObjectReference(
        system=profile.product if profile.product != "unknown" else "mediawiki",
        object_type="page",
        object_id=external_id,
        maturity=profile.discovered_maturity,
        authority_mode=profile.source_authority_mode,
        connector_id=profile.id,
        canonical_url=canonical_url,
        version=revision_id,
        etag=_optional_text(revision.get("sha1")),
        observed_at=_now(),
        metadata={
            "namespace_id": raw.get("ns"),
            "title": title,
            "redirect": bool(redirect_target),
        },
    )
    mapped = {
        "object_type": "page",
        "external_id": external_id,
        "external_page_id": external_id,
        "external_revision_id": revision_id,
        "namespace_id": _integer(raw.get("ns")),
        "title": title,
        "canonical_url": canonical_url,
        "status": "redirect" if redirect_target else "active",
        "redirect_target_external_id": redirect_target,
        "source_revision": revision_id,
        "source_updated_at": timestamp,
        "visibility": visibility,
        "acl_tokens": acl_tokens,
        "mapped_data": {
            "title": title,
            "summary": _optional_text(revision.get("comment")) or "",
            "body": body,
            "content_model": _optional_text(revision.get("contentmodel")) or "wikitext",
            "target_space_ref": mapping.get("target_space_ref"),
            "target_path": target_path,
            "namespace_id": _integer(raw.get("ns")),
            "categories": list(categories),
            "links": list(links),
            "files": list(files),
            "discussions": list(discussions),
            "macros": list(macros),
            "permission_source": permission_source,
            "revision_author": revision_author,
            "revision_author_reference": revision_author_reference,
            "revision_minor": bool(revision.get("minor")),
            "redirect_target_external_id": redirect_target,
            "external_reference": reference.to_dict(),
        },
        "provenance": {
            "connector_profile_id": profile.id,
            "source_system": reference.system,
            "source_revision": revision_id,
            "source_updated_at": timestamp.isoformat() if timestamp else None,
            "observed_at": _now().isoformat(),
        },
    }
    mapped["content_hash"] = _mapped_content_hash(
        mapped["mapped_data"],
        visibility=visibility,
        acl_tokens=acl_tokens,
    )
    diagnostics = permission_diagnostics
    if not discussions and "discussions" not in profile.capabilities:
        diagnostics.append(
            _diagnostic(
                "warning",
                "discussions_unavailable",
                "No supported discussion payload was available for this page.",
                object_ref=external_id,
                field="discussions",
            )
        )
    if files:
        diagnostics.append(
            _diagnostic(
                "info",
                "attachments_reference_only",
                "Source files are stable external references; binary migration requires Files or DMS conflict resolution.",
                object_ref=external_id,
                field="files",
            )
        )
    return mapped, diagnostics


def _upsert_object(
    session: Session,
    profile: ConnectorKnowledgeProfile,
    mapped: Mapping[str, Any],
    *,
    cursor: object,
) -> tuple[ConnectorKnowledgeObject, dict[str, Any]]:
    row = _object_by_external_id(
        session,
        profile.id,
        str(mapped["object_type"]),
        str(mapped["external_id"]),
    )
    effect = "create" if row is None else "update"
    if row is None:
        row = ConnectorKnowledgeObject(
            tenant_id=profile.tenant_id,
            profile_id=profile.id,
            object_type=str(mapped["object_type"]),
            external_id=str(mapped["external_id"]),
            external_page_id=_optional_text(mapped.get("external_page_id")),
            external_revision_id=_optional_text(mapped.get("external_revision_id")),
            namespace_id=_integer(mapped.get("namespace_id")),
            title=str(mapped["title"]),
            canonical_url=_optional_text(mapped.get("canonical_url")),
            status=str(mapped["status"]),
            redirect_target_external_id=_optional_text(
                mapped.get("redirect_target_external_id")
            ),
            source_revision=str(mapped["source_revision"]),
            content_hash=str(mapped["content_hash"]),
            visibility=str(mapped["visibility"]),
            acl_tokens=list(mapped["acl_tokens"]),
            mapped_data=dict(mapped["mapped_data"]),
            provenance=dict(mapped["provenance"]),
            change_cursor=_optional_text(cursor),
            source_updated_at=mapped.get("source_updated_at"),
            observed_at=_now(),
        )
        session.add(row)
    elif (
        row.source_revision == mapped["source_revision"]
        and row.content_hash == mapped["content_hash"]
        and row.status == mapped["status"]
    ):
        effect = "unchanged"
        row.change_cursor = _optional_text(cursor) or row.change_cursor
        row.observed_at = _now()
    else:
        for field_name in (
            "external_page_id",
            "external_revision_id",
            "namespace_id",
            "title",
            "canonical_url",
            "status",
            "redirect_target_external_id",
            "source_revision",
            "content_hash",
            "visibility",
            "source_updated_at",
        ):
            setattr(row, field_name, mapped.get(field_name))
        row.acl_tokens = list(mapped["acl_tokens"])
        row.mapped_data = dict(mapped["mapped_data"])
        row.provenance = dict(mapped["provenance"])
        row.change_cursor = _optional_text(cursor)
        row.observed_at = _now()
        row.resource_revision += 1
    return row, {
        "effect": effect,
        "external_id": row.external_id,
        "title": row.title,
        "source_revision": row.source_revision,
        "status": row.status,
    }


def _delete_object(
    session: Session,
    profile: ConnectorKnowledgeProfile,
    raw: Mapping[str, Any],
    *,
    mapping: Mapping[str, Any],
    cursor: object,
) -> tuple[ConnectorKnowledgeObject | None, dict[str, Any]]:
    supplied_id = _optional_text(raw.get("pageid"))
    title = _optional_text(raw.get("title")) or "Deleted page"
    external_id = supplied_id if supplied_id not in {None, "0"} else None
    row = (
        _object_by_external_id(session, profile.id, "page", external_id)
        if external_id
        else session.scalar(
            select(ConnectorKnowledgeObject).where(
                ConnectorKnowledgeObject.profile_id == profile.id,
                ConnectorKnowledgeObject.object_type == "page",
                ConnectorKnowledgeObject.namespace_id == _integer(raw.get("ns")),
                ConnectorKnowledgeObject.title == title,
            )
        )
    )
    matched_by_title = row is not None and external_id is None
    if row is not None:
        external_id = row.external_id
    if external_id is None:
        external_id = f"deleted:{_hash({'title': title, 'ns': raw.get('ns')})[:32]}"
    if row is None:
        row = ConnectorKnowledgeObject(
            tenant_id=profile.tenant_id,
            profile_id=profile.id,
            object_type="page",
            external_id=external_id,
            external_page_id=external_id,
            namespace_id=_integer(raw.get("ns")),
            title=title,
            status="deleted",
            source_revision=f"deleted:{_optional_text(raw.get('logid')) or cursor or 'unknown'}",
            content_hash=_hash({"deleted": external_id, "cursor": cursor}),
            visibility=profile.default_visibility,
            acl_tokens=list(profile.default_acl_tokens),
            mapped_data={
                "title": title,
                "target_space_ref": mapping.get("target_space_ref"),
                "target_path": _target_path(mapping, title),
                "deleted": True,
            },
            provenance={
                "connector_profile_id": profile.id,
                "deletion_log_id": raw.get("logid"),
            },
            change_cursor=_optional_text(cursor),
            source_updated_at=_parse_datetime(raw.get("timestamp")),
            observed_at=_now(),
        )
        session.add(row)
        effect = "delete"
    elif row.status == "deleted":
        effect = "unchanged"
    else:
        row.status = "deleted"
        row.source_revision = (
            f"deleted:{_optional_text(raw.get('logid')) or cursor or row.source_revision}"
        )
        row.content_hash = _hash({"deleted": external_id, "cursor": cursor})
        row.mapped_data = {
            **dict(row.mapped_data or {}),
            "body": "",
            "deleted": True,
        }
        row.change_cursor = _optional_text(cursor)
        row.observed_at = _now()
        row.resource_revision += 1
        effect = "delete"
    return row, {
        "effect": effect,
        "external_id": external_id,
        "title": row.title,
        "source_revision": row.source_revision,
        "status": "deleted",
        "matched_by": "title" if matched_by_title else "page_id",
        "reason_code": (
            "unmatched_deletion_tombstone"
            if row.external_id.startswith("deleted:")
            else None
        ),
    }


def _store_published_result(
    session: Session,
    profile: ConnectorKnowledgeProfile,
    *,
    external_page_id: str,
    payload: KnowledgePublishRequest,
    result: MediaWikiPublishResult,
) -> ConnectorKnowledgeObject:
    mapping = _namespace_mapping(profile, 0) or profile.namespace_mappings[0]
    mapped = {
        "object_type": "page",
        "external_id": result.page_id or external_page_id,
        "external_page_id": result.page_id or external_page_id,
        "external_revision_id": result.revision_id,
        "namespace_id": 0,
        "title": result.title,
        "canonical_url": result.canonical_url,
        "status": "active",
        "redirect_target_external_id": None,
        "source_revision": result.revision_id,
        "source_updated_at": _now(),
        "visibility": profile.default_visibility,
        "acl_tokens": list(profile.default_acl_tokens),
        "mapped_data": {
            "title": result.title,
            "summary": payload.summary,
            "body": payload.body,
            "content_model": "wikitext",
            "target_space_ref": mapping.get("target_space_ref"),
            "target_path": _target_path(mapping, result.title),
            "namespace_id": 0,
            "categories": [],
            "links": [],
            "files": [],
            "discussions": [],
            "macros": sorted(set(match.strip() for match in _MACRO.findall(payload.body))),
            "permission_source": "configured_fallback",
        },
        "provenance": {
            "connector_profile_id": profile.id,
            "source_system": profile.product,
            "source_revision": result.revision_id,
            "published_by_govoplan": True,
            "observed_at": _now().isoformat(),
        },
    }
    mapped["content_hash"] = _mapped_content_hash(
        mapped["mapped_data"],
        visibility=profile.default_visibility,
        acl_tokens=profile.default_acl_tokens,
    )
    row, _effect = _upsert_object(
        session,
        profile,
        mapped,
        cursor=result.revision_id,
    )
    return row


def _mapped_acl(
    profile: ConnectorKnowledgeProfile,
    *,
    mapping: Mapping[str, Any],
    raw: Mapping[str, Any],
    external_id: str,
) -> tuple[str, list[str], list[dict[str, Any]], str]:
    permissions = raw.get("permissions")
    diagnostics: list[dict[str, Any]] = []
    if isinstance(permissions, Mapping):
        visibility = str(permissions.get("visibility") or "").strip()
        tokens = [
            str(item).strip()
            for item in permissions.get("acl_tokens") or ()
            if str(item).strip()
        ]
        if visibility in {"tenant", "restricted"}:
            try:
                _validate_effective_acl(visibility, tokens)
            except KnowledgeConnectorError:
                diagnostics.append(
                    _diagnostic(
                        "error",
                        "invalid_remote_acl",
                        "The remote permission mapping is invalid; the configured fallback ACL was applied.",
                        object_ref=external_id,
                        field="permissions",
                    )
                )
            else:
                return visibility, tokens, diagnostics, "remote"
    visibility = str(mapping.get("visibility") or profile.default_visibility)
    tokens = list(mapping.get("acl_tokens") or profile.default_acl_tokens)
    _validate_effective_acl(visibility, tokens)
    diagnostics.append(
        _diagnostic(
            "warning",
            "permission_mapping_defaulted",
            "The configured namespace visibility and ACL were applied because authoritative page permissions were unavailable.",
            object_ref=external_id,
            field="permissions",
        )
    )
    return visibility, tokens, diagnostics, "configured_fallback"


def _mapped_content_hash(
    data: Mapping[str, Any],
    *,
    visibility: str,
    acl_tokens: Sequence[str],
) -> str:
    return _hash(
        {
            "title": data.get("title"),
            "body": data.get("body"),
            "categories": data.get("categories") or (),
            "links": data.get("links") or (),
            "files": data.get("files") or (),
            "discussions": data.get("discussions") or (),
            "redirect_target": data.get("redirect_target_external_id"),
            "visibility": visibility,
            "acl_tokens": tuple(acl_tokens),
        }
    )


def _mapped_file(
    profile: ConnectorKnowledgeProfile,
    raw: Mapping[str, Any],
    *,
    observed_at: datetime,
) -> dict[str, Any]:
    title = str(raw.get("title") or "")
    external_id = _optional_text(raw.get("pageid")) or title
    return {
        "name": _without_prefix(title, "File:"),
        "external_id": external_id,
        "reference": ExternalObjectReference(
            system=profile.product if profile.product != "unknown" else "mediawiki",
            object_type="file",
            object_id=external_id,
            maturity=profile.discovered_maturity,
            authority_mode=profile.source_authority_mode,
            connector_id=profile.id,
            canonical_url=_optional_text(raw.get("url")),
            version=_optional_text(raw.get("sha1") or raw.get("timestamp")),
            observed_at=observed_at,
            metadata={"title": title},
        ).to_dict(),
    }


def _mapped_discussion(
    profile: ConnectorKnowledgeProfile,
    raw: Mapping[str, Any],
    *,
    external_id: str,
) -> dict[str, Any]:
    discussion_id = _optional_text(raw.get("id")) or _hash(raw)[:32]
    return {
        "external_id": discussion_id,
        "author": _optional_text(raw.get("author")),
        "body": str(raw.get("body") or "")[:20_000],
        "created_at": _optional_text(raw.get("created_at")),
        "reference": ExternalObjectReference(
            system=profile.product if profile.product != "unknown" else "mediawiki",
            object_type="discussion",
            object_id=discussion_id,
            maturity=profile.discovered_maturity,
            authority_mode=profile.source_authority_mode,
            connector_id=profile.id,
            version=_optional_text(raw.get("revision")),
            observed_at=_now(),
            metadata={"page_id": external_id},
        ).to_dict(),
    }


def _update_search(
    session: Session,
    principal: ApiPrincipal,
    rows: Sequence[ConnectorKnowledgeObject],
    *,
    registry: object | None,
    diagnostics: list[dict[str, Any]],
    source_active: bool = True,
) -> None:
    writer = search_index_writer(registry)
    if writer is None:
        diagnostics.append(
            _diagnostic(
                "info",
                "search_unavailable",
                "Search is not installed; synchronized pages remain available through the connector.",
            )
        )
        return
    from govoplan_connectors.backend.knowledge_search import search_document

    for row in rows:
        try:
            if row.status == "deleted" or not source_active:
                writer.delete_document(
                    session,
                    principal,
                    tenant_id=row.tenant_id,
                    module_id="connectors",
                    resource_type=KNOWLEDGE_RESOURCE_TYPE,
                    resource_id=row.id,
                )
            else:
                document = search_document(session, row)
                if document is not None:
                    writer.upsert_document(session, principal, document=document)
        except Exception:
            diagnostics.append(
                _diagnostic(
                    "warning",
                    "search_update_deferred",
                    "The Search projection could not be updated immediately; authorization rechecks still fail closed and the next rebuild will reconcile it.",
                    object_ref=row.external_id,
                    retryable=True,
                )
            )


def _profile_item(
    item: ConnectorKnowledgeProfile, configuration: ConnectorConfiguration
) -> KnowledgeProfileItem:
    return KnowledgeProfileItem(
        id=item.id,
        tenant_id=item.tenant_id,
        configuration_id=item.configuration_id,
        status=item.status,
        product=item.product,
        product_version=item.product_version,
        desired_maturity=item.desired_maturity,
        discovered_maturity=item.discovered_maturity,
        source_authority_mode=item.source_authority_mode,
        default_visibility=item.default_visibility,
        default_acl_tokens=list(item.default_acl_tokens or ()),
        namespace_mappings=list(item.namespace_mappings or ()),
        capabilities=list(item.capabilities or ()),
        discovery_revision=item.discovery_revision,
        health_status=item.health_status,
        health_details=dict(item.health_details or {}),
        discovered_at=item.discovered_at,
        last_sync_cursor=item.last_sync_cursor,
        last_high_watermark=item.last_high_watermark,
        resource_revision=item.resource_revision,
        credential_reference_present=bool(configuration.credential_ref),
        endpoint_configured=bool(configuration.endpoint_url),
        created_at=item.created_at,
        updated_at=item.updated_at,
    )


def _object_item(
    row: ConnectorKnowledgeObject, profile: ConnectorKnowledgeProfile
) -> KnowledgeObjectItem:
    return KnowledgeObjectItem(
        id=row.id,
        profile_id=row.profile_id,
        object_type=row.object_type,
        external_id=row.external_id,
        external_page_id=row.external_page_id,
        external_revision_id=row.external_revision_id,
        namespace_id=row.namespace_id,
        title=row.title,
        canonical_url=row.canonical_url,
        status=row.status,
        redirect_target_external_id=row.redirect_target_external_id,
        source_revision=row.source_revision,
        visibility=row.visibility,
        acl_tokens=list(row.acl_tokens or ()),
        mapped_data=dict(row.mapped_data or {}),
        external_reference=_external_reference_response(row, profile),
        source_updated_at=row.source_updated_at,
        observed_at=row.observed_at,
        resource_revision=row.resource_revision,
    )


def _external_reference_response(
    row: ConnectorKnowledgeObject, profile: ConnectorKnowledgeProfile
) -> KnowledgeExternalReferenceResponse:
    return KnowledgeExternalReferenceResponse.model_validate(
        ExternalObjectReference(
            system=profile.product if profile.product != "unknown" else "mediawiki",
            object_type=row.object_type,
            object_id=row.external_id,
            maturity=profile.discovered_maturity,
            authority_mode=profile.source_authority_mode,
            connector_id=profile.id,
            canonical_url=row.canonical_url,
            version=row.source_revision,
            etag=row.content_hash,
            observed_at=row.observed_at,
            metadata={
                "title": row.title,
                "status": row.status,
                "namespace_id": row.namespace_id,
                "external_revision_id": row.external_revision_id,
            },
        ).to_dict()
    )


def _run_item(row: ConnectorKnowledgeSyncRun) -> KnowledgeSyncRunItem:
    return KnowledgeSyncRunItem(
        id=row.id,
        tenant_id=row.tenant_id,
        profile_id=row.profile_id,
        mode=row.mode,
        idempotency_key=row.idempotency_key,
        status=row.status,
        cursor_before=row.cursor_before,
        cursor_after=row.cursor_after,
        high_watermark=row.high_watermark,
        counts=dict(row.counts or {}),
        effects=list(row.effects or ()),
        diagnostics=[
            KnowledgeDiagnostic.model_validate(item)
            for item in row.diagnostics or ()
        ],
        provenance=dict(row.provenance or {}),
        started_at=row.started_at,
        finished_at=row.finished_at,
        created_at=row.created_at,
    )


def _migration_response(
    row: ConnectorKnowledgeSyncRun, target_space_ref: str
) -> KnowledgeMigrationDryRunResponse:
    provenance = dict(row.provenance or {})
    diagnostics = [
        KnowledgeDiagnostic.model_validate(item) for item in row.diagnostics or ()
    ]
    return KnowledgeMigrationDryRunResponse(
        run=_run_item(row),
        target_space_ref=target_space_ref,
        source_revision=str(provenance.get("source_revision") or "unknown"),
        source_fingerprint=str(provenance.get("source_fingerprint") or "unknown"),
        summary={key: int(value) for key, value in dict(row.counts or {}).items()},
        effects=list(row.effects or ()),
        diagnostics=diagnostics,
        truncated=bool(provenance.get("truncated")),
        can_apply=(
            not bool(provenance.get("truncated"))
            and not any(item.severity == "error" for item in diagnostics)
            and int(dict(row.counts or {}).get("conflict", 0)) == 0
        ),
    )


def _configuration(
    session: Session, tenant_id: str, configuration_id: str, *, active: bool
) -> ConnectorConfiguration:
    row = session.scalar(
        select(ConnectorConfiguration).where(
            ConnectorConfiguration.tenant_id == tenant_id,
            ConnectorConfiguration.id == configuration_id,
        )
    )
    if row is None:
        raise KnowledgeConnectorError(
            "configuration_not_found", "Connector configuration not found."
        )
    if active and row.status != "active":
        raise KnowledgeConnectorError(
            "configuration_inactive", "Connector configuration is not active."
        )
    return row


def _profile(
    session: Session, tenant_id: str, profile_id: str
) -> ConnectorKnowledgeProfile:
    row = session.scalar(
        select(ConnectorKnowledgeProfile).where(
            ConnectorKnowledgeProfile.tenant_id == tenant_id,
            ConnectorKnowledgeProfile.id == profile_id,
        )
    )
    if row is None:
        raise KnowledgeConnectorError(
            "profile_not_found", "Knowledge connector profile not found."
        )
    return row


def _configurations(
    session: Session,
    tenant_id: str,
    profiles: Sequence[ConnectorKnowledgeProfile],
) -> dict[str, ConnectorConfiguration]:
    ids = tuple(dict.fromkeys(item.configuration_id for item in profiles))
    if not ids:
        return {}
    return {
        row.id: row
        for row in session.scalars(
            select(ConnectorConfiguration).where(
                ConnectorConfiguration.tenant_id == tenant_id,
                ConnectorConfiguration.id.in_(ids),
            )
        )
    }


def _assert_knowledge_configuration(
    session: Session, configuration: ConnectorConfiguration
) -> None:
    definition = session.scalar(
        select(ConnectorDefinition).where(
            ConnectorDefinition.id == configuration.definition_id,
            ConnectorDefinition.tenant_id == configuration.tenant_id,
        )
    )
    specification = dict(configuration.effective_configuration or {})
    provider = str(specification.get("provider") or "").casefold()
    protocol = str(specification.get("protocol") or "").casefold()
    if provider not in {"mediawiki", "bluespice", "knowledge"} or protocol not in {
        "mediawiki",
        "action_api",
        "mediawiki_action_api",
    }:
        raise KnowledgeConnectorError(
            "configuration_type_invalid",
            "The selected configuration is not a MediaWiki/BlueSpice Action API connector.",
        )
    if definition is None or definition.status != "active":
        raise KnowledgeConnectorError(
            "definition_unavailable", "The connector definition is unavailable."
        )


def _credential(
    session: Session,
    principal: ApiPrincipal,
    configuration: ConnectorConfiguration,
) -> dict[str, Any] | None:
    if not configuration.credential_ref:
        return None
    try:
        value = resolve_credential_envelope(
            session,
            credential_id=configuration.credential_ref,
            context=CredentialAccessContext(
                tenant_id=principal.tenant_id,
                user_id=_actor_id(principal),
                group_ids=frozenset(principal.principal.group_ids),
                target_scope_type="tenant",
                target_scope_id=principal.tenant_id,
                module_id="connectors",
                server_ref=_endpoint(configuration),
            ),
        )
    except CredentialEnvelopeError as exc:
        raise KnowledgeConnectorError(
            "credential_unavailable",
            "The connector credential is inactive, unavailable, or outside its approved scope.",
        ) from exc
    return {**dict(value.public_data), **dict(value.secret_data)}


def _run_replay(
    session: Session,
    *,
    tenant_id: str,
    profile_id: str,
    mode: str,
    idempotency_key: str,
    request_hash: str,
) -> ConnectorKnowledgeSyncRun | None:
    row = session.scalar(
        select(ConnectorKnowledgeSyncRun).where(
            ConnectorKnowledgeSyncRun.tenant_id == tenant_id,
            ConnectorKnowledgeSyncRun.profile_id == profile_id,
            ConnectorKnowledgeSyncRun.mode == mode,
            ConnectorKnowledgeSyncRun.idempotency_key == idempotency_key,
        )
    )
    if row is None:
        return None
    if row.request_hash != request_hash:
        raise KnowledgeConnectorError(
            "idempotency_conflict",
            "This idempotency key was already used with a different request.",
        )
    if row.status in {"running", "prepared", "outcome_unknown"}:
        raise KnowledgeConnectorError(
            "operation_unresolved",
            "This connector operation is active or requires reconciliation.",
        )
    return row


def _object_by_external_id(
    session: Session, profile_id: str, object_type: str, external_id: str
) -> ConnectorKnowledgeObject | None:
    return session.scalar(
        select(ConnectorKnowledgeObject).where(
            ConnectorKnowledgeObject.profile_id == profile_id,
            ConnectorKnowledgeObject.object_type == object_type,
            ConnectorKnowledgeObject.external_id == external_id,
        )
    )


def _namespace_mapping(
    profile: ConnectorKnowledgeProfile, namespace_id: object
) -> Mapping[str, Any] | None:
    expected = _integer(namespace_id)
    return next(
        (
            item
            for item in profile.namespace_mappings or ()
            if isinstance(item, Mapping)
            and _integer(item.get("source_namespace_id")) == expected
        ),
        None,
    )


def _target_path(mapping: Mapping[str, Any], title: str) -> str:
    source_name = str(mapping.get("source_name") or "").strip()
    unqualified = _without_prefix(title, f"{source_name}:") if source_name else title
    prefix = str(mapping.get("target_path_prefix") or "").strip(" /_")
    slug = "/".join(
        quote(part.strip().replace(" ", "-"), safe="-._~")
        for part in unqualified.split("/")
        if part.strip()
    )
    return "/".join(value for value in (prefix, slug) if value)


def _validate_effective_acl(visibility: str, tokens: Sequence[str]) -> None:
    if visibility not in {"tenant", "restricted"}:
        raise KnowledgeConnectorError(
            "visibility_invalid", "Knowledge visibility is invalid."
        )
    if visibility == "restricted" and not tokens:
        raise KnowledgeConnectorError(
            "acl_required", "Restricted knowledge requires at least one ACL token."
        )


def _require_scope(principal: ApiPrincipal, scope: str) -> None:
    if not has_scope(principal, scope):
        raise KnowledgeConnectorError("access_denied", f"Missing required scope: {scope}")


def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None:
    if not any(has_scope(principal, scope) for scope in scopes):
        raise KnowledgeConnectorError(
            "access_denied", f"Missing one of the required scopes: {', '.join(scopes)}"
        )


def _endpoint(configuration: ConnectorConfiguration) -> str:
    value = _optional_text(configuration.endpoint_url)
    if not value:
        raise KnowledgeConnectorError(
            "endpoint_missing", "The connector configuration has no endpoint URL."
        )
    return value


def _external_page_id(raw: Mapping[str, Any]) -> str:
    value = _optional_text(raw.get("pageid"))
    if not value:
        raise KnowledgeConnectorError(
            "external_identity_missing", "A MediaWiki page change omitted its stable page id."
        )
    return value


def _revision_content(revision: Mapping[str, Any]) -> str:
    slots = revision.get("slots")
    if isinstance(slots, Mapping):
        main = slots.get("main")
        if isinstance(main, Mapping):
            for name in ("content", "*", "text"):
                if main.get(name) is not None:
                    return str(main[name])[:200_000]
    for name in ("content", "*", "text"):
        if revision.get(name) is not None:
            return str(revision[name])[:200_000]
    return ""


def _safe_extension(value: Mapping[str, Any]) -> dict[str, Any]:
    return {
        key: value[key]
        for key in ("name", "version", "type", "license-name")
        if key in value and value[key] is not None
    }


def _diagnostic(
    severity: str,
    code: str,
    message: str,
    *,
    object_ref: str | None = None,
    field: str | None = None,
    retryable: bool = False,
    details: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    return {
        "severity": severity,
        "code": code,
        "message": message,
        "object_ref": object_ref,
        "field": field,
        "retryable": retryable,
        "details": dict(details or {}),
    }


def _audit(
    session: Session,
    principal: ApiPrincipal,
    *,
    action: str,
    object_type: str,
    object_id: str,
    details: Mapping[str, Any],
) -> None:
    audit_event(
        session,
        tenant_id=principal.tenant_id,
        user_id=getattr(principal.user, "id", None),
        api_key_id=principal.api_key_id,
        action=f"connectors.knowledge.{action}",
        object_type=object_type,
        object_id=object_id,
        details=dict(details),
    )


def _actor_id(principal: ApiPrincipal) -> str | None:
    return _optional_text(
        getattr(principal.user, "id", None)
        or getattr(principal.account, "id", None)
        or principal.principal.account_id
    )


def _supports(current: str, required: str) -> bool:
    try:
        return INTEGRATION_MATURITY_ORDER.index(current) >= INTEGRATION_MATURITY_ORDER.index(
            required  # type: ignore[arg-type]
        )
    except ValueError:
        return False


def _parse_datetime(value: object) -> datetime | None:
    text = _optional_text(value)
    if not text:
        return None
    try:
        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)


def _version(value: object) -> str | None:
    match = re.search(r"\d+(?:\.\d+){1,3}(?:[-+._a-zA-Z0-9]*)?", str(value or ""))
    return match.group(0) if match else None


def _without_prefix(value: str, prefix: str) -> str:
    return value[len(prefix) :] if value.casefold().startswith(prefix.casefold()) else value


def _hash(value: object) -> str:
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode()
    ).hexdigest()


def _optional_text(value: object) -> str | None:
    normalized = str(value or "").strip()
    return normalized or None


def _integer(value: object) -> int | None:
    try:
        return int(value) if value is not None else None
    except (TypeError, ValueError):
        return None


def _now() -> datetime:
    return datetime.now(timezone.utc)


__all__ = [
    "ExternalKnowledgeCapability",
    "KNOWLEDGE_ADMIN_SCOPE",
    "KNOWLEDGE_CAPABILITY",
    "KNOWLEDGE_INTERFACE_VERSION",
    "KNOWLEDGE_MIGRATE_SCOPE",
    "KNOWLEDGE_PROVIDER_ID",
    "KNOWLEDGE_PUBLISH_SCOPE",
    "KNOWLEDGE_READ_SCOPE",
    "KNOWLEDGE_RESOURCE_TYPE",
    "KNOWLEDGE_SYNC_SCOPE",
    "KnowledgeConnectorError",
    "create_profile",
    "discover_profile",
    "list_objects",
    "list_profiles",
    "list_runs",
    "migration_dry_run",
    "publish_page",
    "synchronize_profile",
    "update_profile",
]