from __future__ import annotations

from datetime import datetime
from typing import Any, Iterable, Sequence

from sqlalchemy import BigInteger, DateTime, Index, Integer, JSON, String, UniqueConstraint, func
from sqlalchemy.orm import Mapped, Session, mapped_column

from govoplan_core.core.events import EventActorRef, EventObjectRef, EventTenantRef, PlatformEvent, emit_platform_event
from govoplan_core.db.base import Base, utcnow

WATERMARK_PREFIX = "seq:"


class ChangeSequenceEntry(Base):
    __tablename__ = "core_change_sequence"
    __table_args__ = (
        Index("ix_core_change_sequence_tenant_module_id", "tenant_id", "module_id", "id"),
        Index("ix_core_change_sequence_collection_id", "collection", "id"),
        Index("ix_core_change_sequence_resource", "module_id", "resource_type", "resource_id"),
    )

    id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
    tenant_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
    module_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
    collection: Mapped[str] = mapped_column(String(150), nullable=False, index=True)
    resource_type: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
    resource_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
    operation: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
    actor_type: Mapped[str | None] = mapped_column(String(30), nullable=True, index=True)
    actor_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
    payload: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, nullable=False, index=True)


class ChangeSequenceRetentionFloor(Base):
    __tablename__ = "core_change_sequence_retention_floor"
    __table_args__ = (
        UniqueConstraint("tenant_key", "module_id", "collection", name="uq_core_change_sequence_retention_scope"),
        Index("ix_core_change_sequence_retention_scope", "tenant_key", "module_id", "collection"),
    )

    id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
    tenant_key: Mapped[str] = mapped_column(String(36), nullable=False, default="")
    module_id: Mapped[str] = mapped_column(String(100), nullable=False)
    collection: Mapped[str] = mapped_column(String(150), nullable=False)
    min_valid_sequence: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), nullable=False, default=0)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False)


def _tenant_key(tenant_id: str | None) -> str:
    return tenant_id or ""


def encode_sequence_watermark(sequence_id: int | None) -> str:
    return f"{WATERMARK_PREFIX}{int(sequence_id or 0)}"


def decode_sequence_watermark(value: str | None) -> int:
    if value is None or not value.strip():
        return 0
    raw = value.strip()
    if raw.startswith(WATERMARK_PREFIX):
        raw = raw[len(WATERMARK_PREFIX):]
    try:
        sequence_id = int(raw)
    except ValueError as exc:
        raise ValueError("Invalid delta watermark") from exc
    if sequence_id < 0:
        raise ValueError("Invalid delta watermark")
    return sequence_id


def record_change(
    session: Session,
    *,
    module_id: str,
    collection: str,
    resource_type: str,
    resource_id: str,
    operation: str,
    tenant_id: str | None = None,
    actor_type: str | None = None,
    actor_id: str | None = None,
    payload: dict[str, Any] | None = None,
) -> ChangeSequenceEntry:
    entry = ChangeSequenceEntry(
        tenant_id=tenant_id,
        module_id=module_id,
        collection=collection,
        resource_type=resource_type,
        resource_id=resource_id,
        operation=operation,
        actor_type=actor_type,
        actor_id=actor_id,
        payload=payload or {},
    )
    session.add(entry)
    _publish_change_event(session, entry)
    return entry


def _publish_change_event(session: Session, entry: ChangeSequenceEntry) -> None:
    event_type = _change_event_type(entry.module_id, entry.resource_type, entry.operation)
    emit_platform_event(
        session,
        PlatformEvent(
            type=event_type,
            module_id=entry.module_id,
            payload={
                "collection": entry.collection,
                "operation": entry.operation,
                "payload": dict(entry.payload or {}),
            },
            actor=EventActorRef(type=entry.actor_type, id=entry.actor_id) if entry.actor_type else None,
            tenant=EventTenantRef(id=entry.tenant_id) if entry.tenant_id else None,
            resource=EventObjectRef(type=entry.resource_type, id=entry.resource_id),
            classification="internal",
        )
    )


def _change_event_type(module_id: str, resource_type: str, operation: str) -> str:
    resource = _event_segment(resource_type)
    module = _event_segment(module_id)
    if resource.startswith(f"{module}."):
        resource = resource[len(module) + 1:]
    return ".".join(item for item in (module, resource, _event_segment(operation)) if item)


def _event_segment(value: str) -> str:
    compact = "".join(character if character.isalnum() else "." for character in value.casefold())
    return ".".join(part for part in compact.split(".") if part)


def max_sequence_id(
    session: Session,
    *,
    tenant_id: str | None = None,
    module_id: str | None = None,
    collections: Sequence[str] | None = None,
) -> int:
    query = session.query(func.max(ChangeSequenceEntry.id))
    if tenant_id is not None:
        query = query.filter(ChangeSequenceEntry.tenant_id == tenant_id)
    if module_id is not None:
        query = query.filter(ChangeSequenceEntry.module_id == module_id)
    if collections:
        query = query.filter(ChangeSequenceEntry.collection.in_(tuple(collections)))
    return int(query.scalar() or 0)


def min_sequence_id(
    session: Session,
    *,
    tenant_id: str | None = None,
    module_id: str | None = None,
    collections: Sequence[str] | None = None,
) -> int:
    query = session.query(func.min(ChangeSequenceEntry.id))
    if tenant_id is not None:
        query = query.filter(ChangeSequenceEntry.tenant_id == tenant_id)
    if module_id is not None:
        query = query.filter(ChangeSequenceEntry.module_id == module_id)
    if collections:
        query = query.filter(ChangeSequenceEntry.collection.in_(tuple(collections)))
    return int(query.scalar() or 0)


def sequence_watermark_is_expired(
    session: Session,
    *,
    since: int,
    tenant_id: str | None = None,
    module_id: str | None = None,
    collections: Sequence[str] | None = None,
) -> bool:
    floor = retained_sequence_floor(
        session,
        tenant_id=tenant_id,
        module_id=module_id,
        collections=collections,
    )
    return floor > 0 and since < floor


def retained_sequence_floor(
    session: Session,
    *,
    tenant_id: str | None = None,
    module_id: str | None = None,
    collections: Sequence[str] | None = None,
) -> int:
    query = session.query(func.max(ChangeSequenceRetentionFloor.min_valid_sequence))
    query = query.filter(ChangeSequenceRetentionFloor.tenant_key == _tenant_key(tenant_id))
    if module_id is not None:
        query = query.filter(ChangeSequenceRetentionFloor.module_id == module_id)
    if collections:
        query = query.filter(ChangeSequenceRetentionFloor.collection.in_(tuple(collections)))
    return int(query.scalar() or 0)


def prune_sequence_entries(
    session: Session,
    *,
    before_or_equal_id: int,
    tenant_id: str | None = None,
    module_id: str | None = None,
    collections: Sequence[str] | None = None,
) -> int:
    """Delete old sequence rows and record the new minimum safe watermark.

    After pruning rows up to `before_or_equal_id`, clients with watermarks below
    that value cannot be replayed safely and should receive a full snapshot.
    """

    cutoff = int(before_or_equal_id)
    if cutoff <= 0:
        return 0
    query = session.query(ChangeSequenceEntry).filter(ChangeSequenceEntry.id <= cutoff)
    if tenant_id is not None:
        query = query.filter(ChangeSequenceEntry.tenant_id == tenant_id)
    if module_id is not None:
        query = query.filter(ChangeSequenceEntry.module_id == module_id)
    collection_values = tuple(collections or ())
    if collection_values:
        query = query.filter(ChangeSequenceEntry.collection.in_(collection_values))
    scopes = {
        (_tenant_key(entry.tenant_id), entry.module_id, entry.collection)
        for entry in query.with_entities(
            ChangeSequenceEntry.tenant_id,
            ChangeSequenceEntry.module_id,
            ChangeSequenceEntry.collection,
        ).distinct()
    }
    deleted = query.delete(synchronize_session=False)
    for tenant_key, scoped_module_id, collection in scopes:
        floor = session.query(ChangeSequenceRetentionFloor).filter(
            ChangeSequenceRetentionFloor.tenant_key == tenant_key,
            ChangeSequenceRetentionFloor.module_id == scoped_module_id,
            ChangeSequenceRetentionFloor.collection == collection,
        ).one_or_none()
        if floor is None:
            floor = ChangeSequenceRetentionFloor(
                tenant_key=tenant_key,
                module_id=scoped_module_id,
                collection=collection,
                min_valid_sequence=cutoff,
            )
        else:
            floor.min_valid_sequence = max(int(floor.min_valid_sequence or 0), cutoff)
        session.add(floor)
    return int(deleted)


def sequence_entries_since(
    session: Session,
    *,
    since: int,
    tenant_id: str | None = None,
    module_id: str | None = None,
    collections: Iterable[str] | None = None,
    limit: int = 500,
) -> list[ChangeSequenceEntry]:
    query = session.query(ChangeSequenceEntry).filter(ChangeSequenceEntry.id > since)
    if tenant_id is not None:
        query = query.filter(ChangeSequenceEntry.tenant_id == tenant_id)
    if module_id is not None:
        query = query.filter(ChangeSequenceEntry.module_id == module_id)
    collection_values = tuple(collections or ())
    if collection_values:
        query = query.filter(ChangeSequenceEntry.collection.in_(collection_values))
    return query.order_by(ChangeSequenceEntry.id.asc()).limit(max(1, limit)).all()


__all__ = [
    "ChangeSequenceEntry",
    "ChangeSequenceRetentionFloor",
    "decode_sequence_watermark",
    "encode_sequence_watermark",
    "max_sequence_id",
    "min_sequence_id",
    "prune_sequence_entries",
    "record_change",
    "retained_sequence_floor",
    "sequence_entries_since",
    "sequence_watermark_is_expired",
]
