from __future__ import annotations

import copy
import hashlib
import json
from dataclasses import dataclass, field
from typing import Any, Callable, Iterable, Mapping, Sequence

from sqlalchemy.orm import Session


class ConcurrencyError(RuntimeError):
    """Base class for mutation precondition and compare-and-set failures."""


class MissingPreconditionError(ConcurrencyError):
    def __init__(self, *, resource_type: str, resource_id: str) -> None:
        self.resource_type = resource_type
        self.resource_id = resource_id
        super().__init__("A strong If-Match precondition is required for this mutation.")

    def as_dict(self) -> dict[str, Any]:
        return {
            "code": "precondition_required",
            "resource": {
                "type": self.resource_type,
                "id": self.resource_id,
            },
            "retryable": True,
        }


class RevisionConflictError(ConcurrencyError):
    def __init__(
        self,
        *,
        resource_type: str,
        resource_id: str,
        current_revision: int,
        submitted_base_revision: int,
        refresh_path: str | None = None,
        conflicts: Sequence["MergeConflict"] = (),
        merge_candidate: Any = None,
        current_etag: str | None = None,
    ) -> None:
        self.resource_type = resource_type
        self.resource_id = resource_id
        self.current_revision = int(current_revision)
        self.submitted_base_revision = int(submitted_base_revision)
        self.refresh_path = refresh_path
        self.conflicts = tuple(conflicts)
        self.merge_candidate = merge_candidate
        self.current_etag = current_etag
        super().__init__(
            f"{resource_type} {resource_id} changed from revision "
            f"{submitted_base_revision} to {current_revision}"
        )

    @property
    def safe_merge_available(self) -> bool:
        return self.merge_candidate is not None and not self.conflicts

    def as_dict(
        self,
        *,
        include_values: bool = False,
        include_merge_candidate: bool = False,
    ) -> dict[str, Any]:
        result: dict[str, Any] = {
            "code": "revision_conflict",
            "resource": {
                "type": self.resource_type,
                "id": self.resource_id,
            },
            "current_revision": self.current_revision,
            "submitted_base_revision": self.submitted_base_revision,
            "retryable": True,
            "safe_merge_available": self.safe_merge_available,
            "conflicts": [
                conflict.as_dict(include_values=include_values)
                for conflict in self.conflicts[:100]
            ],
        }
        if self.refresh_path:
            result["refresh_path"] = self.refresh_path
        if self.current_etag:
            result["current_etag"] = self.current_etag
        if include_merge_candidate and self.safe_merge_available:
            result["merge_candidate"] = copy.deepcopy(self.merge_candidate)
        return result


@dataclass(frozen=True, slots=True)
class MergeConflict:
    path: str
    kind: str
    base_value: Any = None
    local_value: Any = None
    current_value: Any = None

    def as_dict(self, *, include_values: bool = False) -> dict[str, Any]:
        result: dict[str, Any] = {
            "path": self.path or "/",
            "kind": self.kind,
        }
        if include_values:
            result.update(
                {
                    "base_value": _bounded_json_value(self.base_value),
                    "local_value": _bounded_json_value(self.local_value),
                    "current_value": _bounded_json_value(self.current_value),
                }
            )
        return result


@dataclass(slots=True)
class ThreeWayMergeResult:
    value: Any
    conflicts: list[MergeConflict] = field(default_factory=list)
    applied_paths: list[str] = field(default_factory=list)

    @property
    def merged(self) -> bool:
        return not self.conflicts


ProtectedPath = str | Callable[[str], bool]


def strong_resource_etag(
    resource_type: str,
    resource_id: str,
    revision: int,
) -> str:
    """Return an opaque strong ETag for one mutable aggregate revision."""

    normalized_revision = int(revision)
    if normalized_revision < 1:
        raise ValueError("Resource revisions must be positive integers")
    digest = hashlib.sha256(
        "\x00".join(
            (
                "govoplan-strong-revision-v1",
                str(resource_type),
                str(resource_id),
                str(normalized_revision),
            )
        ).encode("utf-8")
    ).hexdigest()
    return f'"sha256-{digest}"'


def if_match_matches(header_value: str | None, expected_etag: str) -> bool:
    """Apply strong comparison semantics to an If-Match header."""

    if not header_value:
        return False
    for raw_candidate in header_value.split(","):
        candidate = raw_candidate.strip()
        if candidate == "*":
            return True
        if candidate.startswith("W/"):
            continue
        if candidate == expected_etag:
            return True
    return False


def assert_revision_precondition(
    if_match: str | None,
    *,
    resource_type: str,
    resource_id: str,
    submitted_base_revision: int,
) -> None:
    if not if_match:
        raise MissingPreconditionError(
            resource_type=resource_type,
            resource_id=resource_id,
        )
    submitted_etag = strong_resource_etag(
        resource_type,
        resource_id,
        submitted_base_revision,
    )
    if not if_match_matches(if_match, submitted_etag):
        raise ConcurrencyError(
            "If-Match does not identify the submitted base revision."
        )


def claim_revision(
    session: Session,
    *,
    model: type[Any],
    filters: Iterable[Any],
    revision_attribute: str,
    expected_revision: int,
    resource_type: str,
    resource_id: str,
    refresh_path: str | None = None,
) -> int:
    """Atomically claim the next revision in the caller's transaction."""

    revision_column = getattr(model, revision_attribute)
    expected = int(expected_revision)
    normalized_filters = tuple(filters)
    query = session.query(model).filter(*normalized_filters)
    updated = query.filter(revision_column == expected).update(
        {revision_column: revision_column + 1},
        synchronize_session=False,
    )
    if updated == 1:
        session.flush()
        return expected + 1

    current = (
        session.query(revision_column)
        .filter(*normalized_filters)
        .scalar()
    )
    if current is None:
        raise LookupError(f"{resource_type} {resource_id} was not found")
    raise RevisionConflictError(
        resource_type=resource_type,
        resource_id=resource_id,
        current_revision=int(current),
        submitted_base_revision=expected,
        refresh_path=refresh_path,
        current_etag=strong_resource_etag(
            resource_type,
            resource_id,
            int(current),
        ),
    )


def three_way_merge(
    base: Any,
    local: Any,
    current: Any,
    *,
    protected_paths: Sequence[ProtectedPath] = (),
    stable_id_fields: Sequence[str] = ("id",),
) -> ThreeWayMergeResult:
    """Conservatively merge local changes onto a concurrently changed value."""

    return _merge_value(
        copy.deepcopy(base),
        copy.deepcopy(local),
        copy.deepcopy(current),
        path="",
        protected_paths=tuple(protected_paths),
        stable_id_fields=tuple(stable_id_fields),
    )


def _merge_value(
    base: Any,
    local: Any,
    current: Any,
    *,
    path: str,
    protected_paths: Sequence[ProtectedPath],
    stable_id_fields: Sequence[str],
) -> ThreeWayMergeResult:
    local_changed = local != base
    current_changed = current != base
    if not local_changed:
        return ThreeWayMergeResult(value=current)
    if local == current:
        return ThreeWayMergeResult(value=current)
    if _is_protected(path, protected_paths):
        return _conflict(path, "protected_path", base, local, current)
    if not current_changed:
        return ThreeWayMergeResult(
            value=local,
            applied_paths=[path or "/"],
        )
    if isinstance(base, Mapping) and isinstance(local, Mapping) and isinstance(current, Mapping):
        return _merge_mapping(
            base,
            local,
            current,
            path=path,
            protected_paths=protected_paths,
            stable_id_fields=stable_id_fields,
        )
    if isinstance(base, list) and isinstance(local, list) and isinstance(current, list):
        return _merge_list(
            base,
            local,
            current,
            path=path,
            protected_paths=protected_paths,
            stable_id_fields=stable_id_fields,
        )
    return _conflict(path, "same_path_changed", base, local, current)


def _merge_mapping(
    base: Mapping[str, Any],
    local: Mapping[str, Any],
    current: Mapping[str, Any],
    *,
    path: str,
    protected_paths: Sequence[ProtectedPath],
    stable_id_fields: Sequence[str],
) -> ThreeWayMergeResult:
    missing = object()
    result: dict[str, Any] = {}
    conflicts: list[MergeConflict] = []
    applied_paths: list[str] = []
    keys = list(dict.fromkeys((*current.keys(), *local.keys(), *base.keys())))
    for key in keys:
        child_path = _join_path(path, str(key))
        base_value = base.get(key, missing)
        local_value = local.get(key, missing)
        current_value = current.get(key, missing)
        merged = _merge_presence(
            base_value,
            local_value,
            current_value,
            missing=missing,
            path=child_path,
            protected_paths=protected_paths,
            stable_id_fields=stable_id_fields,
        )
        conflicts.extend(merged.conflicts)
        applied_paths.extend(merged.applied_paths)
        if merged.value is not missing:
            result[str(key)] = merged.value
    return ThreeWayMergeResult(
        value=result,
        conflicts=conflicts,
        applied_paths=applied_paths,
    )


def _merge_presence(
    base: Any,
    local: Any,
    current: Any,
    *,
    missing: object,
    path: str,
    protected_paths: Sequence[ProtectedPath],
    stable_id_fields: Sequence[str],
) -> ThreeWayMergeResult:
    if local is missing and current is missing:
        return ThreeWayMergeResult(value=missing)
    if base is missing:
        if local is missing:
            return ThreeWayMergeResult(value=current)
        if current is missing:
            if _is_protected(path, protected_paths):
                return _conflict(path, "protected_path", None, local, None)
            return ThreeWayMergeResult(value=local, applied_paths=[path])
        if local == current:
            return ThreeWayMergeResult(value=current)
        return _conflict(path, "concurrent_add", None, local, current)
    if local is missing:
        if current == base:
            if _is_protected(path, protected_paths):
                return _conflict(path, "protected_path", base, None, current)
            return ThreeWayMergeResult(value=missing, applied_paths=[path])
        return _conflict(path, "delete_vs_edit", base, None, current)
    if current is missing:
        if local == base:
            return ThreeWayMergeResult(value=missing)
        return _conflict(path, "edit_vs_delete", base, local, None)
    return _merge_value(
        base,
        local,
        current,
        path=path,
        protected_paths=protected_paths,
        stable_id_fields=stable_id_fields,
    )


def _merge_list(
    base: list[Any],
    local: list[Any],
    current: list[Any],
    *,
    path: str,
    protected_paths: Sequence[ProtectedPath],
    stable_id_fields: Sequence[str],
) -> ThreeWayMergeResult:
    identity_field = _stable_identity_field(
        (base, local, current),
        stable_id_fields,
    )
    if identity_field is None:
        return _conflict(path, "unkeyed_collection", base, local, current)

    base_by_id = {str(item[identity_field]): item for item in base}
    local_by_id = {str(item[identity_field]): item for item in local}
    current_by_id = {str(item[identity_field]): item for item in current}
    base_order = list(base_by_id)
    local_order = list(local_by_id)
    current_order = list(current_by_id)

    local_reordered = _common_order(local_order, base_order) != _common_order(
        base_order,
        local_order,
    )
    current_reordered = _common_order(current_order, base_order) != _common_order(
        base_order,
        current_order,
    )
    if local_reordered and current_reordered and local_order != current_order:
        return _conflict(path, "collection_reorder", base_order, local_order, current_order)

    result_by_id: dict[str, Any] = {}
    conflicts: list[MergeConflict] = []
    applied_paths: list[str] = []
    identities = list(dict.fromkeys((*current_order, *local_order, *base_order)))
    missing = object()
    for identity in identities:
        item_path = _join_path(path, f"{identity_field}={identity}")
        merged = _merge_presence(
            base_by_id.get(identity, missing),
            local_by_id.get(identity, missing),
            current_by_id.get(identity, missing),
            missing=missing,
            path=item_path,
            protected_paths=protected_paths,
            stable_id_fields=stable_id_fields,
        )
        conflicts.extend(merged.conflicts)
        applied_paths.extend(merged.applied_paths)
        if merged.value is not missing:
            result_by_id[identity] = merged.value

    order_source = local_order if local_reordered and not current_reordered else current_order
    merged_order = [identity for identity in order_source if identity in result_by_id]
    for identity in identities:
        if identity in result_by_id and identity not in merged_order:
            merged_order.append(identity)
    return ThreeWayMergeResult(
        value=[result_by_id[identity] for identity in merged_order],
        conflicts=conflicts,
        applied_paths=applied_paths,
    )


def _stable_identity_field(
    values: Sequence[list[Any]],
    candidates: Sequence[str],
) -> str | None:
    all_items = [item for value in values for item in value]
    if not all_items or not all(isinstance(item, Mapping) for item in all_items):
        return None
    for candidate in candidates:
        valid = True
        for value in values:
            identities = [
                str(item.get(candidate, "")).strip()
                for item in value
                if isinstance(item, Mapping)
            ]
            if any(not identity for identity in identities) or len(identities) != len(
                set(identities)
            ):
                valid = False
                break
        if valid:
            return candidate
    return None


def _common_order(left: Sequence[str], right: Sequence[str]) -> list[str]:
    right_set = set(right)
    return [item for item in left if item in right_set]


def _is_protected(path: str, protected_paths: Sequence[ProtectedPath]) -> bool:
    normalized = path or "/"
    for protected in protected_paths:
        if callable(protected):
            if protected(normalized):
                return True
            continue
        prefix = protected.rstrip("/") or "/"
        if normalized == prefix or normalized.startswith(f"{prefix}/"):
            return True
    return False


def _join_path(parent: str, segment: str) -> str:
    escaped = segment.replace("~", "~0").replace("/", "~1")
    return f"{parent}/{escaped}" if parent else f"/{escaped}"


def _conflict(
    path: str,
    kind: str,
    base: Any,
    local: Any,
    current: Any,
) -> ThreeWayMergeResult:
    return ThreeWayMergeResult(
        value=current,
        conflicts=[
            MergeConflict(
                path=path or "/",
                kind=kind,
                base_value=base,
                local_value=local,
                current_value=current,
            )
        ],
    )


def _bounded_json_value(value: Any, *, depth: int = 0) -> Any:
    if depth >= 4:
        return {"summary": type(value).__name__}
    if value is None or isinstance(value, (bool, int, float)):
        return value
    if isinstance(value, str):
        return value[:500]
    if isinstance(value, Mapping):
        result = {
            str(key)[:100]: _bounded_json_value(item, depth=depth + 1)
            for key, item in list(value.items())[:20]
        }
        if len(value) > 20:
            result["_truncated_items"] = len(value) - 20
        return result
    if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):
        result = [
            _bounded_json_value(item, depth=depth + 1)
            for item in list(value)[:20]
        ]
        if len(value) > 20:
            result.append({"_truncated_items": len(value) - 20})
        return result
    try:
        return json.loads(json.dumps(value))
    except (TypeError, ValueError):
        return {"summary": type(value).__name__}


__all__ = [
    "ConcurrencyError",
    "MergeConflict",
    "MissingPreconditionError",
    "RevisionConflictError",
    "ThreeWayMergeResult",
    "assert_revision_precondition",
    "claim_revision",
    "if_match_matches",
    "strong_resource_etag",
    "three_way_merge",
]
