from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, Iterable, List, Optional, Sequence, Set

from sqlalchemy import select, update
from sqlalchemy.orm import Session

from db.models import ChannelProjectionRow, ChannelProjectionRun, MasterProduct


def _native_connection_id(channel_code: str, connection_id: Optional[int]) -> Optional[int]:
    """Projection cache always keys by native connection id (not Shopify compat offset)."""
    if connection_id is None:
        return None
    from db.compat_connections import resolve_native_connection_id

    channel = str(channel_code or "").strip().lower()
    if channel in {"shopify", "magento"}:
        native = resolve_native_connection_id(channel, connection_id)
        return int(native) if native is not None else int(connection_id)
    return int(connection_id)


def partition_projection_cache(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    skus: Optional[Iterable[str]] = None,
    product_structure: Optional[str] = None,
) -> Dict[str, Any]:
    """Split requested SKUs into fresh cached payloads vs ones that must be rebuilt.

    Fresh requires: projection_status=ready, optional product_structure match, and
    source_row_hash matching MasterProduct.row_hash when a master row exists.
    """
    connection_id = _native_connection_id(channel_code, connection_id)
    wanted = sorted({str(s).strip() for s in (skus or []) if str(s).strip()})
    stmt = select(ChannelProjectionRow).where(ChannelProjectionRow.channel_code == channel_code)
    if connection_id is None:
        stmt = stmt.where(ChannelProjectionRow.connection_id.is_(None))
    else:
        stmt = stmt.where(ChannelProjectionRow.connection_id == connection_id)
    if wanted:
        stmt = stmt.where(ChannelProjectionRow.master_sku.in_(wanted))
    rows = list(session.scalars(stmt.order_by(ChannelProjectionRow.master_sku)).all())
    by_sku = {row.master_sku: row for row in rows}

    hash_by_sku = _master_row_hashes(session, wanted or list(by_sku.keys()))

    fresh: List[Dict[str, Any]] = []
    rebuild: List[str] = []
    target_skus = wanted or sorted(by_sku.keys())
    for sku in target_skus:
        row = by_sku.get(sku)
        if row is None:
            rebuild.append(sku)
            continue
        if row.projection_status != "ready":
            rebuild.append(sku)
            continue
        if product_structure is not None and row.product_structure != product_structure:
            rebuild.append(sku)
            continue
        master_hash = hash_by_sku.get(sku)
        if master_hash is not None and str(row.source_row_hash or "") != str(master_hash):
            rebuild.append(sku)
            continue
        fresh.append(_row_to_payload(row))

    return {
        "fresh": fresh,
        "rebuild_skus": rebuild,
        "fresh_skus": [p["sku"] for p in fresh],
    }


def load_cached_payloads(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    skus: Optional[Iterable[str]] = None,
    product_structure: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """Return cached payloads only when every requested SKU is fresh; otherwise []."""
    wanted = sorted({str(s).strip() for s in (skus or []) if str(s).strip()})
    partitioned = partition_projection_cache(
        session,
        channel_code=channel_code,
        connection_id=connection_id,
        skus=wanted or None,
        product_structure=product_structure,
    )
    if wanted and partitioned["rebuild_skus"]:
        return []
    if wanted and len(partitioned["fresh"]) != len(wanted):
        return []
    return list(partitioned["fresh"])


def store_projection_payloads(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    payloads: List[Dict[str, Any]],
    product_structure: Optional[str] = None,
    scope_key: str = "assigned",
    notes: Optional[str] = None,
    replace_scope: bool = False,
) -> Dict[str, Any]:
    """Upsert projection rows by master_sku. Never wipe sibling SKUs unless replace_scope=True."""
    connection_id = _native_connection_id(channel_code, connection_id)
    now = datetime.utcnow()
    _supersede_ready_runs(session, channel_code=channel_code, connection_id=connection_id, at=now)

    run = ChannelProjectionRun(
        channel_code=channel_code,
        connection_id=connection_id,
        scope_key=scope_key,
        status="ready",
        product_structure=product_structure,
        sku_count=len(payloads),
        started_at=now,
        completed_at=now,
        notes=notes,
    )
    session.add(run)
    session.flush()

    payload_skus = [
        str(payload.get("sku") or "").strip()
        for payload in payloads
        if str(payload.get("sku") or "").strip()
    ]
    existing = _existing_rows(session, channel_code=channel_code, connection_id=connection_id, skus=payload_skus)
    master_hashes = _master_row_hashes(session, payload_skus)

    upserted = 0
    for payload in payloads:
        master_sku = str(payload.get("sku") or "").strip()
        if not master_sku:
            continue
        channel_sku = str(payload.get("channel_sku") or payload.get("sku") or "").strip()
        canonical = dict(payload.get("canonical_fields") or {})
        override_codes = sorted({str(c).strip() for c in (payload.get("override_codes") or []) if str(c).strip()})
        source_hash = str(
            payload.get("source_row_hash")
            or canonical.get("row_hash")
            or master_hashes.get(master_sku)
            or ""
        )
        row = existing.get(master_sku)
        if row is None:
            row = ChannelProjectionRow(
                channel_code=channel_code,
                connection_id=connection_id,
                master_sku=master_sku,
                created_at=now,
            )
            session.add(row)
            existing[master_sku] = row

        row.run_id = run.id
        row.channel_sku = channel_sku
        row.product_role = str(payload.get("product_role") or "standalone")
        row.product_structure = product_structure
        row.projection_status = "ready"
        row.canonical_fields_json = canonical
        row.channel_fields_json = dict(payload.get("fields") or {})
        row.relation_fields_json = dict(payload.get("relation_fields") or {})
        row.association_fields_json = dict(payload.get("association_fields") or {})
        row.taxonomy_targets_json = payload.get("taxonomy_targets")
        row.prices_json = dict(payload.get("prices") or {})
        row.override_codes_json = override_codes
        row.price = payload.get("price")
        row.payload_hash = str(payload.get("payload_hash") or "")
        row.source_row_hash = source_hash
        row.normalized_at = now
        row.updated_at = now
        upserted += 1

    removed = 0
    if replace_scope:
        removed = _delete_scope_rows_not_in(
            session,
            channel_code=channel_code,
            connection_id=connection_id,
            keep_skus=set(payload_skus),
        )

    session.flush()
    return {
        "run_id": run.id,
        "sku_count": upserted,
        "removed_sku_count": removed,
        "status": "ready",
        "replace_scope": replace_scope,
        "connection_id": connection_id,
    }


def mark_projection_rows_stale(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
    skus: Optional[Sequence[str]] = None,
    reason: Optional[str] = None,
) -> int:
    """Mark projection rows stale so the next compose rebuilds them."""
    connection_id = _native_connection_id(channel_code, connection_id)
    now = datetime.utcnow()
    stmt = (
        update(ChannelProjectionRow)
        .where(ChannelProjectionRow.channel_code == channel_code)
        .where(ChannelProjectionRow.projection_status == "ready")
        .values(projection_status="stale", updated_at=now)
    )
    if connection_id is None:
        stmt = stmt.where(ChannelProjectionRow.connection_id.is_(None))
    else:
        stmt = stmt.where(ChannelProjectionRow.connection_id == connection_id)
    wanted = sorted({str(s).strip() for s in (skus or []) if str(s).strip()})
    if wanted:
        stmt = stmt.where(ChannelProjectionRow.master_sku.in_(wanted))
    result = session.execute(stmt)
    _stamp_runs_invalidated(
        session,
        channel_code=channel_code,
        connection_id=connection_id,
        at=now,
        reason=reason,
    )
    return int(result.rowcount or 0)


def _row_to_payload(row: ChannelProjectionRow) -> Dict[str, Any]:
    override_codes = row.override_codes_json
    if isinstance(override_codes, dict):
        codes = sorted(str(k) for k in override_codes.keys())
    elif isinstance(override_codes, list):
        codes = [str(c) for c in override_codes]
    else:
        codes = []
    return {
        "sku": row.master_sku,
        "channel_sku": row.channel_sku,
        "channel_code": row.channel_code,
        "product_role": row.product_role,
        "canonical_fields": dict(row.canonical_fields_json or {}),
        "fields": dict(row.channel_fields_json or {}),
        "relation_fields": dict(row.relation_fields_json or {}),
        "association_fields": dict(row.association_fields_json or {}),
        "taxonomy_targets": row.taxonomy_targets_json,
        "price": float(row.price) if row.price is not None else None,
        "prices": dict(row.prices_json or {}),
        "payload_hash": row.payload_hash,
        "source_row_hash": row.source_row_hash,
        "override_codes": codes,
    }


def _connection_filter(stmt, connection_id: Optional[int]):
    if connection_id is None:
        return stmt.where(ChannelProjectionRow.connection_id.is_(None))
    return stmt.where(ChannelProjectionRow.connection_id == connection_id)


def _existing_rows(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int],
    skus: Sequence[str],
) -> Dict[str, ChannelProjectionRow]:
    if not skus:
        return {}
    stmt = (
        select(ChannelProjectionRow)
        .where(ChannelProjectionRow.channel_code == channel_code)
        .where(ChannelProjectionRow.master_sku.in_(list(skus)))
    )
    stmt = _connection_filter(stmt, connection_id)
    return {row.master_sku: row for row in session.scalars(stmt).all()}


def _master_row_hashes(session: Session, skus: Sequence[str]) -> Dict[str, str]:
    wanted = sorted({str(s).strip() for s in skus if str(s).strip()})
    if not wanted:
        return {}
    rows = session.execute(
        select(MasterProduct.sku, MasterProduct.row_hash).where(MasterProduct.sku.in_(wanted))
    ).all()
    return {str(sku): str(row_hash or "") for sku, row_hash in rows}


def _delete_scope_rows_not_in(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int],
    keep_skus: Set[str],
) -> int:
    stmt = select(ChannelProjectionRow).where(ChannelProjectionRow.channel_code == channel_code)
    stmt = _connection_filter(stmt, connection_id)
    rows = list(session.scalars(stmt).all())
    removed = 0
    for row in rows:
        if row.master_sku not in keep_skus:
            session.delete(row)
            removed += 1
    return removed


def _supersede_ready_runs(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int],
    at: datetime,
) -> None:
    stmt = (
        update(ChannelProjectionRun)
        .where(ChannelProjectionRun.channel_code == channel_code)
        .where(ChannelProjectionRun.status == "ready")
        .values(status="superseded", invalidated_at=at, updated_at=at)
    )
    if connection_id is None:
        stmt = stmt.where(ChannelProjectionRun.connection_id.is_(None))
    else:
        stmt = stmt.where(ChannelProjectionRun.connection_id == connection_id)
    session.execute(stmt)


def _stamp_runs_invalidated(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int],
    at: datetime,
    reason: Optional[str],
) -> None:
    values: Dict[str, Any] = {"invalidated_at": at, "updated_at": at}
    if reason:
        values["notes"] = reason
    stmt = (
        update(ChannelProjectionRun)
        .where(ChannelProjectionRun.channel_code == channel_code)
        .where(ChannelProjectionRun.status == "ready")
        .values(**values)
    )
    if connection_id is None:
        stmt = stmt.where(ChannelProjectionRun.connection_id.is_(None))
    else:
        stmt = stmt.where(ChannelProjectionRun.connection_id == connection_id)
    session.execute(stmt)