"""Module 2 — equipment: inventory, distribution, collection, live stock, ledger."""
from datetime import date

from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from sqlalchemy import case, func, or_, select
from sqlalchemy.orm import Session, selectinload

from app.api.deps import client_ip, get_current_user, require_admin, require_manager, require_write
from app.core.time import range_bounds_utc, utcnow
from app.db.session import get_db
from app.models.enums import (
    DistributionStatus,
    EquipmentCondition,
    EquipmentItemStatus,
    HolderType,
    StockMovementType,
)
from app.models.customer import Customer
from app.models.equipment import (
    EquipmentCollection,
    EquipmentDistribution,
    EquipmentItem,
    EquipmentStockMovement,
)
from app.models.reference import EquipmentType, Location
from app.models.user import User
from app.schemas.common import Message, Page
from app.schemas.equipment import (
    CollectionCreate,
    CollectionOut,
    DistributionCreate,
    DistributionDetail,
    DistributionListItem,
    EquipmentItemCreate,
    EquipmentItemDetail,
    EquipmentItemListItem,
    EquipmentItemUpdate,
    EquipmentSummary,
    HolderOut,
    MovementOut,
    StockAdjust,
    StockByType,
    WriteOff,
)
from app.schemas.reference import EquipmentTypeOut
from app.services import audit
from app.services import equipment as stock

router = APIRouter(prefix="/equipment", tags=["equipment"])


# --- serialization helpers ---
#
# `holder` and `available_quantity` are derived rather than stored, so these rows are
# built field by field instead of via model_validate.


def _as_item(item: EquipmentItem) -> EquipmentItemListItem:
    return EquipmentItemListItem(
        id=item.id,
        equipment_type=EquipmentTypeOut.model_validate(item.equipment_type),
        model=item.model,
        serial_number=item.serial_number,
        condition=item.condition,
        quantity=item.quantity,
        status=item.status,
        holder=HolderOut.from_item(item),
        available_quantity=stock.available_quantity(item),
        created_at=item.created_at,
    )


def _as_item_detail(item: EquipmentItem) -> EquipmentItemDetail:
    history = sorted(
        (d for d in item.distributions if not d.is_deleted),
        key=lambda d: d.issued_at,
        reverse=True,
    )
    return EquipmentItemDetail(
        **_as_item(item).model_dump(),
        purchase_date=item.purchase_date,
        unit_cost=item.unit_cost,
        notes=item.notes,
        updated_at=item.updated_at,
        distributions=[DistributionListItem.model_validate(d) for d in history],
    )


def _as_distribution_detail(distribution: EquipmentDistribution) -> DistributionDetail:
    out = DistributionDetail.model_validate(distribution)
    out.collections = [
        CollectionOut.model_validate(c)
        for c in sorted(distribution.collections, key=lambda c: c.collected_at)
        if not c.is_deleted
    ]
    return out


def _get_item_or_404(db: Session, item_id: int) -> EquipmentItem:
    item = db.scalar(
        select(EquipmentItem)
        .where(EquipmentItem.id == item_id, EquipmentItem.is_deleted.is_(False))
        .options(selectinload(EquipmentItem.distributions))
    )
    if item is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Equipment item not found"
        )
    return item


def _get_distribution_or_404(db: Session, distribution_id: int) -> EquipmentDistribution:
    distribution = db.scalar(
        select(EquipmentDistribution)
        .where(
            EquipmentDistribution.id == distribution_id,
            EquipmentDistribution.is_deleted.is_(False),
        )
        .options(selectinload(EquipmentDistribution.collections))
    )
    if distribution is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Distribution not found"
        )
    return distribution


def _reload_item(db: Session, item_id: int) -> EquipmentItemDetail:
    """Re-read after a write — the session keeps instances alive past commit."""
    db.expire_all()
    return _as_item_detail(_get_item_or_404(db, item_id))


def _reload_distribution(db: Session, distribution_id: int) -> DistributionDetail:
    db.expire_all()
    return _as_distribution_detail(_get_distribution_or_404(db, distribution_id))


# --- live stock ---


@router.get("/summary", response_model=EquipmentSummary)
def equipment_summary(
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
) -> EquipmentSummary:
    return EquipmentSummary(**stock.summary(db))


@router.get("/stock", response_model=list[StockByType])
def stock_balance(
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
    equipment_type_id: int | None = None,
    location_id: int | None = Query(None, description="Narrow on-hand stock to one location"),
    below_reorder: bool = Query(False, description="Only types at or under their reorder level"),
) -> list[StockByType]:
    """Real-time balance per equipment type, broken down by location."""
    rows = stock.stock_by_type(db, equipment_type_id=equipment_type_id, location_id=location_id)
    if below_reorder:
        rows = [r for r in rows if r["below_reorder"]]
    return [StockByType.model_validate(r) for r in rows]


@router.get("/movements", response_model=Page[MovementOut])
def list_movements(
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
    equipment_type_id: int | None = None,
    location_id: int | None = None,
    movement_type: list[StockMovementType] | None = Query(None),
    date_from: date | None = Query(None, description="moved_at >= (Asia/Dhaka)"),
    date_to: date | None = Query(None, description="moved_at <= (Asia/Dhaka)"),
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
) -> Page[MovementOut]:
    """The append-only stock ledger — the source for movement history and charts."""
    stmt = select(EquipmentStockMovement)
    if equipment_type_id is not None:
        stmt = stmt.where(EquipmentStockMovement.equipment_type_id == equipment_type_id)
    if location_id is not None:
        stmt = stmt.where(EquipmentStockMovement.location_id == location_id)
    if movement_type:
        stmt = stmt.where(EquipmentStockMovement.movement_type.in_(movement_type))

    start, end = range_bounds_utc(date_from, date_to)
    if start is not None:
        stmt = stmt.where(EquipmentStockMovement.moved_at >= start)
    if end is not None:
        stmt = stmt.where(EquipmentStockMovement.moved_at < end)

    total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0
    rows = db.scalars(
        stmt.order_by(EquipmentStockMovement.moved_at.desc(), EquipmentStockMovement.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    ).unique().all()
    return Page.build([MovementOut.model_validate(r) for r in rows], total, page, page_size)


# --- inventory ---


@router.get("/items", response_model=Page[EquipmentItemListItem])
def list_items(
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
    search: str | None = Query(None, description="Serial number or model"),
    equipment_type_id: int | None = None,
    condition: list[EquipmentCondition] | None = Query(None),
    status_in: list[EquipmentItemStatus] | None = Query(None, alias="status"),
    holder_type: HolderType | None = None,
    location_id: int | None = None,
    holder_user_id: int | None = None,
    available_only: bool = Query(False, description="Only rows with issuable stock"),
    sort: str = Query(
        "-created_at",
        pattern="^-?(created_at|serial_number|quantity|available_quantity|condition|status|type|holder)$",
    ),
    page: int = Query(1, ge=1),
    page_size: int = Query(25, ge=1, le=200),
) -> Page[EquipmentItemListItem]:
    stmt = select(EquipmentItem).where(EquipmentItem.is_deleted.is_(False))

    if search:
        pattern = f"%{search.strip()}%"
        stmt = stmt.where(
            or_(EquipmentItem.serial_number.ilike(pattern), EquipmentItem.model.ilike(pattern))
        )
    if equipment_type_id is not None:
        stmt = stmt.where(EquipmentItem.equipment_type_id == equipment_type_id)
    if condition:
        stmt = stmt.where(EquipmentItem.condition.in_(condition))
    if status_in:
        stmt = stmt.where(EquipmentItem.status.in_(status_in))
    if holder_type is not None:
        stmt = stmt.where(EquipmentItem.holder_type == holder_type)
    if location_id is not None:
        stmt = stmt.where(EquipmentItem.location_id == location_id)
    if holder_user_id is not None:
        stmt = stmt.where(EquipmentItem.holder_user_id == holder_user_id)
    if available_only:
        stmt = stmt.where(
            EquipmentItem.status == EquipmentItemStatus.IN_STOCK,
            EquipmentItem.holder_type == HolderType.WAREHOUSE,
            EquipmentItem.condition != EquipmentCondition.FAULTY,
            EquipmentItem.quantity > 0,
        )

    total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0

    descending = sort.startswith("-")
    field = sort.lstrip("-")
    # `type` and `holder` are display names on related rows; a correlated subquery
    # keeps the sort working alongside the eager-loaded relationships.
    if field == "type":
        order = (
            select(EquipmentType.name)
            .where(EquipmentType.id == EquipmentItem.equipment_type_id)
            .scalar_subquery()
        )
    elif field == "available_quantity":
        order = case(
            (
                (EquipmentItem.status == EquipmentItemStatus.IN_STOCK)
                & (EquipmentItem.holder_type == HolderType.WAREHOUSE),
                EquipmentItem.quantity,
            ),
            else_=0,
        )
    elif field == "holder":
        order = case(
            (
                EquipmentItem.holder_type == HolderType.TECHNICIAN,
                select(User.full_name)
                .where(User.id == EquipmentItem.holder_user_id)
                .scalar_subquery(),
            ),
            (
                EquipmentItem.holder_type == HolderType.CUSTOMER,
                select(Customer.full_name)
                .where(Customer.id == EquipmentItem.holder_customer_id)
                .scalar_subquery(),
            ),
            else_=select(Location.name)
            .where(Location.id == EquipmentItem.location_id)
            .scalar_subquery(),
        )
    else:
        order = getattr(EquipmentItem, field)
    stmt = stmt.order_by(order.desc() if descending else order.asc(), EquipmentItem.id.desc())

    rows = db.scalars(
        stmt.offset((page - 1) * page_size).limit(page_size)
    ).unique().all()
    return Page.build([_as_item(r) for r in rows], total, page, page_size)


@router.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(
    payload: EquipmentItemCreate,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_write),
):
    """Take new stock into a location. Bulk types merge into the matching stock row."""
    sns = payload.serial_numbers if payload.serial_numbers else ([payload.serial_number] if payload.serial_number else [None])
    
    created_items = []
    
    for sn in sns:
        equipment_type = stock.validate_new_item(
            db,
            equipment_type_id=payload.equipment_type_id,
            location_id=payload.location_id,
            serial_number=sn,
            quantity=payload.quantity,
        )

        item = EquipmentItem(
            equipment_type_id=payload.equipment_type_id,
            model=payload.model,
            serial_number=sn,
            condition=payload.condition,
            quantity=payload.quantity if sn is None else 1,
            holder_type=HolderType.WAREHOUSE,
            location_id=payload.location_id,
            status=EquipmentItemStatus.IN_STOCK,
            purchase_date=payload.purchase_date,
            unit_cost=payload.unit_cost,
            notes=payload.notes,
        )
        db.add(item)
        db.flush()

        stock.record_movement(
            db,
            equipment_type_id=equipment_type.id,
            movement_type=StockMovementType.INBOUND,
            quantity_delta=payload.quantity if sn is None else 1,
            location_id=payload.location_id,
            item_id=item.id,
            user_id=current_user.id,
            notes=payload.notes,
        )
        audit.record(
            db,
            user_id=current_user.id,
            action="create",
            entity_type="equipment_item",
            entity_id=item.id,
            payload={
                "type": equipment_type.code,
                "serial": sn,
                "quantity": payload.quantity if sn is None else 1,
            },
            ip_address=client_ip(request),
        )
        created_items.append(item.id)

    db.commit()
    
    results = [_reload_item(db, item_id) for item_id in created_items]
    return results if len(results) > 1 else results[0]


@router.get("/items/{item_id}", response_model=EquipmentItemDetail)
def get_item(
    item_id: int,
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
) -> EquipmentItemDetail:
    return _as_item_detail(_get_item_or_404(db, item_id))


@router.patch("/items/{item_id}", response_model=EquipmentItemDetail)
def update_item(
    item_id: int,
    payload: EquipmentItemUpdate,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_write),
) -> EquipmentItemDetail:
    item = _get_item_or_404(db, item_id)
    if item.status == EquipmentItemStatus.WRITTEN_OFF:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="A written-off item can no longer be edited",
        )

    changes = payload.model_dump(exclude_unset=True)
    if not changes:
        return _as_item_detail(item)

    if "serial_number" in changes and not item.equipment_type.is_serialized:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"'{item.equipment_type.name}' is a bulk item and cannot carry a serial number",
        )
    if changes.get("condition") == EquipmentCondition.FAULTY and item.status == EquipmentItemStatus.IN_STOCK:
        # Marking on-hand stock faulty pulls it out of the issuable pool immediately.
        item.status = EquipmentItemStatus.RETURNED

    for field, value in changes.items():
        setattr(item, field, value)

    audit.record(
        db,
        user_id=current_user.id,
        action="update",
        entity_type="equipment_item",
        entity_id=item.id,
        payload={k: str(v) for k, v in changes.items()},
        ip_address=client_ip(request),
    )
    db.commit()
    return _reload_item(db, item_id)


@router.delete("/items/{item_id}", response_model=Message)
def delete_item(
    item_id: int,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_admin),
) -> Message:
    """Soft delete a stock row. Anything out in the field must be collected first."""
    item = _get_item_or_404(db, item_id)
    if item.status == EquipmentItemStatus.ISSUED:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="This item is out with a holder — collect it before deleting",
        )
    if any(d.status in stock.OPEN_DISTRIBUTION_STATUSES for d in item.distributions if not d.is_deleted):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="This item has units still outstanding on an open issue",
        )

    if item.quantity:
        stock.record_movement(
            db,
            equipment_type_id=item.equipment_type_id,
            movement_type=StockMovementType.ADJUSTMENT,
            quantity_delta=-item.quantity,
            location_id=item.location_id,
            item_id=item.id,
            user_id=current_user.id,
            notes="Stock row deleted",
        )
    item.is_deleted = True
    item.deleted_at = utcnow()
    audit.record(
        db,
        user_id=current_user.id,
        action="delete",
        entity_type="equipment_item",
        entity_id=item.id,
        ip_address=client_ip(request),
    )
    db.commit()
    return Message(detail="Equipment item deleted")


@router.post("/items/{item_id}/adjust", response_model=EquipmentItemDetail)
def adjust_item(
    item_id: int,
    payload: StockAdjust,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_manager),
) -> EquipmentItemDetail:
    """Correct a bulk count after a physical stock take."""
    item = stock.lock_item(db, item_id)
    stock.adjust(
        db, item, quantity_delta=payload.quantity_delta, reason=payload.reason, actor=current_user
    )
    audit.record(
        db,
        user_id=current_user.id,
        action="adjust_stock",
        entity_type="equipment_item",
        entity_id=item.id,
        payload={"delta": payload.quantity_delta, "reason": payload.reason},
        ip_address=client_ip(request),
    )
    db.commit()
    return _reload_item(db, item_id)


@router.post("/items/{item_id}/write-off", response_model=EquipmentItemDetail)
def write_off_item(
    item_id: int,
    payload: WriteOff,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_manager),
) -> EquipmentItemDetail:
    """Retire units — damaged beyond repair, or lost in the field."""
    item = stock.lock_item(db, item_id)
    stock.write_off(db, item, quantity=payload.quantity, reason=payload.reason, actor=current_user)
    audit.record(
        db,
        user_id=current_user.id,
        action="write_off",
        entity_type="equipment_item",
        entity_id=item.id,
        payload={"quantity": payload.quantity, "reason": payload.reason},
        ip_address=client_ip(request),
    )
    db.commit()
    return _reload_item(db, item_id)


@router.post("/items/{item_id}/release", response_model=EquipmentItemDetail)
def release_item(
    item_id: int,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_write),
    condition: EquipmentCondition = Query(
        EquipmentCondition.USED, description="Condition confirmed by the inspection"
    ),
) -> EquipmentItemDetail:
    """Put quarantined stock back into the issuable pool once it has been inspected."""
    item = stock.lock_item(db, item_id)
    stock.release_from_quarantine(db, item, condition=condition, actor=current_user)
    audit.record(
        db,
        user_id=current_user.id,
        action="release_quarantine",
        entity_type="equipment_item",
        entity_id=item.id,
        payload={"condition": condition.value},
        ip_address=client_ip(request),
    )
    db.commit()
    return _reload_item(db, item_id)


# --- distribution ---


@router.get("/distributions", response_model=Page[DistributionListItem])
def list_distributions(
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
    search: str | None = Query(None, description="Holder name or customer code"),
    equipment_type_id: int | None = None,
    status_in: list[DistributionStatus] | None = Query(None, alias="status"),
    issued_to_type: HolderType | None = None,
    issued_to_user_id: int | None = None,
    issued_to_customer_id: int | None = None,
    connection_id: int | None = None,
    overdue: bool = Query(False, description="Only issues past their expected return date"),
    date_from: date | None = Query(None, description="issued_at >= (Asia/Dhaka)"),
    date_to: date | None = Query(None, description="issued_at <= (Asia/Dhaka)"),
    sort: str = Query(
        "-issued_at",
        pattern="^-?(issued_at|expected_return_date|quantity|outstanding|status|item|holder)$",
    ),
    page: int = Query(1, ge=1),
    page_size: int = Query(25, ge=1, le=200),
) -> Page[DistributionListItem]:
    stmt = select(EquipmentDistribution).where(EquipmentDistribution.is_deleted.is_(False))

    if search:
        stmt = stmt.where(stock.searchable_holder_clause(f"%{search.strip()}%"))
    if equipment_type_id is not None:
        stmt = stmt.where(
            EquipmentDistribution.equipment_item_id.in_(
                select(EquipmentItem.id).where(
                    EquipmentItem.equipment_type_id == equipment_type_id
                )
            )
        )
    if status_in:
        stmt = stmt.where(EquipmentDistribution.status.in_(status_in))
    if issued_to_type is not None:
        stmt = stmt.where(EquipmentDistribution.issued_to_type == issued_to_type)
    if issued_to_user_id is not None:
        stmt = stmt.where(EquipmentDistribution.issued_to_user_id == issued_to_user_id)
    if issued_to_customer_id is not None:
        stmt = stmt.where(EquipmentDistribution.issued_to_customer_id == issued_to_customer_id)
    if connection_id is not None:
        stmt = stmt.where(EquipmentDistribution.connection_id == connection_id)
    if overdue:
        stmt = stmt.where(stock.overdue_filter())

    start, end = range_bounds_utc(date_from, date_to)
    if start is not None:
        stmt = stmt.where(EquipmentDistribution.issued_at >= start)
    if end is not None:
        stmt = stmt.where(EquipmentDistribution.issued_at < end)

    total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0

    descending = sort.startswith("-")
    field = sort.lstrip("-")
    # `item` and `holder` are display names on related rows; a correlated subquery
    # keeps the sort working alongside the eager-loaded relationships.
    if field == "item":
        order = (
            select(EquipmentType.name)
            .join(EquipmentItem, EquipmentItem.equipment_type_id == EquipmentType.id)
            .where(EquipmentItem.id == EquipmentDistribution.equipment_item_id)
            .scalar_subquery()
        )
    elif field == "outstanding":
        order = EquipmentDistribution.quantity - EquipmentDistribution.returned_quantity
    elif field == "holder":
        order = case(
            (
                EquipmentDistribution.issued_to_type == HolderType.TECHNICIAN,
                select(User.full_name)
                .where(User.id == EquipmentDistribution.issued_to_user_id)
                .scalar_subquery(),
            ),
            else_=select(Customer.full_name)
            .where(Customer.id == EquipmentDistribution.issued_to_customer_id)
            .scalar_subquery(),
        )
    else:
        order = getattr(EquipmentDistribution, field)
    stmt = stmt.order_by(
        order.desc() if descending else order.asc(), EquipmentDistribution.id.desc()
    )

    rows = db.scalars(
        stmt.offset((page - 1) * page_size).limit(page_size)
    ).unique().all()
    return Page.build(
        [DistributionListItem.model_validate(r) for r in rows], total, page, page_size
    )


@router.post("/distributions", response_model=DistributionDetail, status_code=status.HTTP_201_CREATED)
def issue_equipment(
    payload: DistributionCreate,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_write),
) -> DistributionDetail:
    """Issue stock to a technician (a loan) or a customer (an install)."""
    distribution = stock.issue(db, payload, current_user)
    audit.record(
        db,
        user_id=current_user.id,
        action="issue_equipment",
        entity_type="equipment_distribution",
        entity_id=distribution.id,
        payload={
            "item_id": payload.equipment_item_id,
            "quantity": payload.quantity,
            "to": payload.issued_to_type.value,
        },
        ip_address=client_ip(request),
    )
    db.commit()
    return _reload_distribution(db, distribution.id)


@router.get("/distributions/{distribution_id}", response_model=DistributionDetail)
def get_distribution(
    distribution_id: int,
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
) -> DistributionDetail:
    return _as_distribution_detail(_get_distribution_or_404(db, distribution_id))


@router.post("/distributions/{distribution_id}/collect", response_model=DistributionDetail)
def collect_equipment(
    distribution_id: int,
    payload: CollectionCreate,
    request: Request,
    db: Session = Depends(get_db),
    current_user: User = Depends(require_write),
) -> DistributionDetail:
    """Take units back into a location. Partial returns leave the issue open."""
    distribution = _get_distribution_or_404(db, distribution_id)
    collection = stock.collect(db, distribution, payload, current_user)
    audit.record(
        db,
        user_id=current_user.id,
        action="collect_equipment",
        entity_type="equipment_distribution",
        entity_id=distribution.id,
        payload={
            "quantity": payload.quantity,
            "condition": payload.condition_on_return.value,
            "collection_id": collection.id,
        },
        ip_address=client_ip(request),
    )
    db.commit()
    return _reload_distribution(db, distribution_id)


@router.get("/collections", response_model=Page[CollectionOut])
def list_collections(
    db: Session = Depends(get_db),
    _: User = Depends(get_current_user),
    equipment_type_id: int | None = None,
    condition: list[EquipmentCondition] | None = Query(None),
    returned_to_location_id: int | None = None,
    date_from: date | None = Query(None, description="collected_at >= (Asia/Dhaka)"),
    date_to: date | None = Query(None, description="collected_at <= (Asia/Dhaka)"),
    page: int = Query(1, ge=1),
    page_size: int = Query(25, ge=1, le=200),
) -> Page[CollectionOut]:
    stmt = select(EquipmentCollection).where(EquipmentCollection.is_deleted.is_(False))

    if equipment_type_id is not None:
        stmt = stmt.where(
            EquipmentCollection.equipment_item_id.in_(
                select(EquipmentItem.id).where(EquipmentItem.equipment_type_id == equipment_type_id)
            )
        )
    if condition:
        stmt = stmt.where(EquipmentCollection.condition_on_return.in_(condition))
    if returned_to_location_id is not None:
        stmt = stmt.where(EquipmentCollection.returned_to_location_id == returned_to_location_id)

    start, end = range_bounds_utc(date_from, date_to)
    if start is not None:
        stmt = stmt.where(EquipmentCollection.collected_at >= start)
    if end is not None:
        stmt = stmt.where(EquipmentCollection.collected_at < end)

    total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0
    rows = db.scalars(
        stmt.order_by(EquipmentCollection.collected_at.desc(), EquipmentCollection.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    ).unique().all()
    return Page.build([CollectionOut.model_validate(r) for r in rows], total, page, page_size)