"""Marketplace service layer.

Business logic for creating and reading listings. Routes stay thin and call
into these functions. Image files are persisted via
:func:`app.marketplace.utils.save_listing_image`.
"""

from __future__ import annotations

import hashlib
import uuid
from datetime import datetime, timedelta
from math import ceil

from flask import current_app, session
from sqlalchemy import func, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import selectinload
from werkzeug.datastructures import FileStorage

from app.extensions import db
from app.marketplace.utils import generate_sku, generate_slug, save_listing_image
from app.models import (
    Category,
    CompletedTransaction,
    Conversation,
    Favorite,
    Listing,
    ListingImage,
    ListingView,
    Notification,
    NotificationType,
    Offer,
    RelatedEntityType,
    Subscription,
    SubscriptionStatus,
    User,
    UserProfile,
    UserReputation,
)
from app.models.base import utcnow
from app.models.enums import (
    ListingCondition,
    ListingStatus,
    ListingVisibility,
    OfferStatus,
    ReviewStatus,
)
from app.reviews.badges import badge_filter_expr, trusted_seller_filter_expr
from app.notifications.services import create_notification


def create_listing(
    seller: User,
    *,
    title: str,
    description: str,
    category_id: int,
    price,
    condition: str,
    negotiable: bool,
    location: str | None,
    images: list[FileStorage],
) -> Listing:
    """Create and publish a listing with its images in one transaction.

    The MVP publishes immediately (``available`` / ``public`` / ``approved``).
    Images are saved to disk first, then the rows are committed together.
    """
    listing = Listing(
        seller_id=seller.id,
        category_id=category_id,
        title=title.strip(),
        slug=generate_slug(title),
        sku=generate_sku(),
        description=description.strip(),
        price=price,
        condition=ListingCondition(condition),
        negotiable=negotiable,
        location=(location or None),
        status=ListingStatus.AVAILABLE,
        visibility=ListingVisibility.PUBLIC,
        review_status=ReviewStatus.APPROVED,
        published_at=utcnow(),
    )

    for order, file in enumerate([f for f in images if f and f.filename], start=1):
        stored_name = save_listing_image(file)
        listing.images.append(
            ListingImage(image_path=stored_name, display_order=order)
        )

    db.session.add(listing)
    db.session.commit()
    _publish_activity(seller, "listing_created", "Listing created",
                      message=f"Your listing “{listing.title}” is now live.",
                      entity_type=RelatedEntityType.LISTING, entity_id=listing.id)
    return listing


def _publish_activity(user, event_type, title, *, message=None, entity_type=None, entity_id=None):
    """Thin wrapper so marketplace events flow through the single activity API."""
    from app.activity.services import publish_activity
    publish_activity(user, event_type, title, message=message,
                     related_entity_type=entity_type, related_entity_id=entity_id)


def _active_premium_seller_ids_subquery():
    """A scalar subquery of user ids with an active, non-expired subscription.

    Used inside ORDER BY to boost premium sellers without an N+1 — the premium
    check happens once, in the database, as part of the listing query.
    """
    return db.select(Subscription.user_id).where(
        Subscription.status == SubscriptionStatus.ACTIVE,
        Subscription.end_date > utcnow(),
    )


def _catalog_order_by(sort: str) -> list:
    """Ordering for the public catalog.

    Featured listings always come first (most-recently featured first). Premium
    sellers are boosted *without changing relevance*: for the default "newest"
    view they form a tier above free sellers; for an explicit sort (price/oldest)
    the requested order is preserved and premium only breaks ties. A final
    ``created_at`` key keeps results deterministic.
    """
    premium_seller = Listing.seller_id.in_(_active_premium_seller_ids_subquery())
    # Only *currently* featured (live period) listings get featured priority.
    order = [_currently_featured_expr().desc(), Listing.featured_at.desc()]
    if sort == "newest":
        order += [premium_seller.desc(), Listing.created_at.desc()]
    else:
        order += [
            _LISTING_SORTS.get(sort, _LISTING_SORTS["newest"]),
            premium_seller.desc(),
            Listing.created_at.desc(),
        ]
    return order


def get_featured_listings(limit: int = 3) -> list[Listing]:
    """Return listings for the landing page — genuinely featured ones first."""
    return list(
        db.session.scalars(
            db.select(Listing)
            .where(
                Listing.status == ListingStatus.AVAILABLE,
                Listing.visibility == ListingVisibility.PUBLIC,
                Listing.deleted_at.is_(None),
            )
            .order_by(
                _currently_featured_expr().desc(),
                Listing.featured_at.desc(),
                Listing.created_at.desc(),
            )
            .limit(limit)
        )
    )


# --- Featured listings (Phase 2B) -----------------------------------------

# A featured listing is promoted for this many days, after which it silently
# returns to normal visibility. Defined once so the duration is easy to change.
FEATURE_DURATION_DAYS = 7


class FeatureListingError(Exception):
    """Raised when a listing cannot be featured."""


def _currently_featured_expr():
    """SQL predicate: featured AND still within its featured period.

    Used in ORDER BY / queries so expired featured listings immediately lose
    priority even before their flag is cleaned up — never relying on
    ``is_featured`` alone.
    """
    return db.and_(
        Listing.is_featured.is_(True),
        Listing.featured_until.isnot(None),
        Listing.featured_until > utcnow(),
    )


def is_listing_featured(listing: Listing) -> bool:
    """Raw featured flag (ignores expiry). Prefer
    :func:`is_listing_currently_featured` for display and priority decisions."""
    return bool(listing.is_featured)


def is_listing_currently_featured(listing: Listing) -> bool:
    """Whether the listing is featured *and* its period is still live.

    Pure, O(1) check over already-loaded columns (no query) — safe for templates
    and ordering-correctness. An expired listing reads as not featured even if
    its flag hasn't been cleaned up yet.
    """
    return bool(
        listing.is_featured
        and listing.featured_until is not None
        and listing.featured_until > utcnow()
    )


def featured_days_remaining(listing: Listing) -> int | None:
    """Whole days left in the featured period, or None when not currently
    featured. Rounded up so a fresh feature reads as the full duration."""
    if not is_listing_currently_featured(listing):
        return None
    seconds = (listing.featured_until - utcnow()).total_seconds()
    return max(int(ceil(seconds / 86400)), 0)


def get_featured_listing(user: User) -> Listing | None:
    """Return the user's currently-featured (live, available, non-deleted) listing."""
    return db.session.scalar(
        db.select(Listing).where(
            Listing.seller_id == user.id,
            Listing.deleted_at.is_(None),
            _currently_featured_expr(),
        )
    )


def expire_featured_listing(listing: Listing, *, commit: bool = True) -> None:
    """Clear a single listing's featured state (used for expiry or un-feature)."""
    listing.is_featured = False
    listing.featured_at = None
    listing.featured_until = None
    if commit:
        db.session.commit()


# Manual "un-feature" is the same clearing operation; keep the familiar name.
unfeature_listing = expire_featured_listing


def cleanup_expired_featured_listings() -> int:
    """Clear the featured flag from every listing whose period has ended.

    One bulk UPDATE (O(1), no N+1), run transparently whenever listings are read
    through the marketplace services — no scheduled job required. Returns the
    number of listings expired.
    """
    result = db.session.execute(
        db.update(Listing)
        .where(
            Listing.is_featured.is_(True),
            Listing.featured_until.isnot(None),
            Listing.featured_until <= utcnow(),
        )
        .values(is_featured=False, featured_at=None, featured_until=None)
    )
    if result.rowcount:
        db.session.commit()
    return result.rowcount or 0


def feature_listing(listing: Listing, user: User) -> Listing | None:
    """Feature ``listing`` for ``user`` and return the replaced listing (or None).

    Enforces server-side: ownership, the premium capability, and that the listing
    is available (not deleted / sold / inactive). Applies the one-featured-listing
    rule via :func:`app.premium.permissions.max_featured_listings` — other
    featured listings over the allowance (oldest first) are unfeatured to make
    room, so a premium seller always ends up with exactly one featured listing.
    Sets a fresh ``FEATURE_DURATION_DAYS`` window. Raises
    :class:`FeatureListingError` when the listing may not be featured.
    """
    from app.premium.permissions import can_feature_listing, max_featured_listings

    if listing.seller_id != user.id:
        raise FeatureListingError("You can only feature your own listings.")
    if not can_feature_listing(user):
        raise FeatureListingError("Featuring listings is a Premium feature.")
    if listing.deleted_at is not None:
        raise FeatureListingError("This listing has been removed and can't be featured.")
    if listing.status != ListingStatus.AVAILABLE:
        raise FeatureListingError("Only available listings can be featured.")
    if is_listing_currently_featured(listing):
        return None  # already actively featured — nothing to do

    limit = max_featured_listings(user)
    # Make room: keep the (limit - 1) most-recently-featured OTHER listings,
    # unfeature the rest (oldest first). Excludes this listing so re-featuring an
    # expired-but-still-flagged listing isn't mistaken for a replacement.
    current = list(
        db.session.scalars(
            db.select(Listing).where(
                Listing.seller_id == user.id,
                Listing.is_featured.is_(True),
                Listing.deleted_at.is_(None),
                Listing.id != listing.id,
            )
        )
    )
    current.sort(key=lambda l: l.featured_at or datetime.min)  # oldest first
    replaced = None
    overflow = len(current) - max(limit - 1, 0)
    for old in current[: max(overflow, 0)]:
        old.is_featured = False
        old.featured_at = None
        old.featured_until = None
        replaced = old

    now = utcnow()
    listing.is_featured = True
    listing.featured_at = now
    listing.featured_until = now + timedelta(days=FEATURE_DURATION_DAYS)
    db.session.commit()
    _publish_activity(user, "listing_featured", "Listing featured",
                      message=f"“{listing.title}” is now featured and boosted in discovery.",
                      entity_type=RelatedEntityType.LISTING, entity_id=listing.id)
    return replaced


def get_featurable_listing_cards(user: User) -> list[dict]:
    """Read model for the "Choose Featured Listing" page.

    Returns the user's active (available, non-deleted) listings that are *not*
    already featured — the only ones eligible to be featured — each as a small
    display dict with views/favorites (cached columns) and an offer count. Offers
    are resolved with a single GROUP BY over the page's listing ids (no N+1); no
    business rule lives here, this is purely a read for the picker UI.
    """
    listings = db.session.scalars(
        db.select(Listing)
        .options(selectinload(Listing.images), selectinload(Listing.category))
        .where(
            Listing.seller_id == user.id,
            Listing.deleted_at.is_(None),
            Listing.status == ListingStatus.AVAILABLE,
            db.not_(_currently_featured_expr()),
        )
        .order_by(Listing.created_at.desc())
    ).all()
    if not listings:
        return []
    ids = [l.id for l in listings]
    offers = {
        lid: int(c) for lid, c in db.session.execute(
            db.select(Offer.listing_id, func.count()).where(Offer.listing_id.in_(ids)).group_by(Offer.listing_id)
        ).all()
    }
    return [{
        "public_id": l.public_id,
        "title": l.title,
        "price": l.price,
        "image": l.images[0].image_path if l.images else None,
        "views": l.view_count or 0,
        "favorites": l.favorite_count or 0,
        "offers": offers.get(l.id, 0),
    } for l in listings]


# Public catalog filter/sort option lists. Exposed for the marketplace browse
# view so the template and the query stay in sync (values are validated below).
CATALOG_SORT_OPTIONS = [
    ("newest", "Newest"),
    ("oldest", "Oldest"),
    ("price_low", "Price: Low to High"),
    ("price_high", "Price: High to Low"),
    ("most_viewed", "Most Viewed"),
    ("most_favorited", "Most Favorited"),
    ("best_rated_seller", "Best Rated Seller"),
    ("most_reviews", "Most Reviews"),
    ("recently_updated", "Recently Updated"),
]
CATALOG_CONDITION_OPTIONS = [
    (member.value, member.value.title()) for member in ListingCondition
]
# Seller-rating filter options (all read the cached UserReputation, never recompute).
SELLER_RATING_OPTIONS = [
    ("", "Any rating"),
    ("4", "4+ stars"),
    ("4.5", "4.5+ stars"),
    ("5", "5 stars"),
    ("new", "New sellers"),
]


def _escape_like(term: str) -> str:
    """Escape LIKE wildcards so '%' and '_' match literally, not as patterns."""
    return term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def search_listings(
    page: int,
    *,
    q: str | None = None,
    category_id: int | None = None,
    condition: str | None = None,
    min_price=None,
    max_price=None,
    sort: str = "newest",
    seller_rating: str | None = None,
    trusted_only: bool = False,
    badge: str | None = None,
):
    """Smart catalog search: a pagination of public, available listings.

    All filtering is pushed into SQL for correct pagination:
        q            - case-insensitive match on title, description, category name
                       or seller name (joins Category / User / UserProfile).
        category_id  - restrict to a single category.
        condition    - a :class:`ListingCondition` value (None/invalid = all).
        min_price / max_price - inclusive price bounds (swapped if inverted).
        sort         - one of ``_LISTING_SORTS`` keys (defaults to newest).
        seller_rating- '4' / '4.5' / '5' (cached avg rating with ≥1 review) or
                       'new' (no reviews). Reads UserReputation — never recomputed.
        trusted_only - only Trusted/Top Seller badge holders (badge service).
        badge        - a specific badge key (badge service filter expression).

    Seller-quality filters reuse the cached ``UserReputation`` and the badge
    service's thresholds, so eligibility is never recalculated here.
    """
    per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20)
    # Retire lapsed featured listings so they lose featured priority (no cron).
    cleanup_expired_featured_listings()

    query = db.select(Listing).where(
        Listing.status == ListingStatus.AVAILABLE,
        Listing.visibility == ListingVisibility.PUBLIC,
        Listing.deleted_at.is_(None),
    )

    # Joins are to at-most-one row per listing (category/seller/reputation), so
    # they never multiply rows — pagination counts stay correct without DISTINCT.
    if q:
        query = (
            query.outerjoin(Category, Listing.category_id == Category.id)
            .outerjoin(User, Listing.seller_id == User.id)
            .outerjoin(UserProfile, UserProfile.user_id == User.id)
        )
    need_reputation = bool(seller_rating or trusted_only or badge) or sort in _REPUTATION_SORTS
    if need_reputation:
        query = query.outerjoin(UserReputation, UserReputation.user_id == Listing.seller_id)

    if q:
        like = f"%{_escape_like(q)}%"
        query = query.where(or_(
            Listing.title.ilike(like, escape="\\"),
            Listing.description.ilike(like, escape="\\"),
            Category.name.ilike(like, escape="\\"),
            UserProfile.full_name.ilike(like, escape="\\"),
            User.email.ilike(like, escape="\\"),
        ))
    if category_id:
        query = query.where(Listing.category_id == category_id)
    if condition in {member.value for member in ListingCondition}:
        query = query.where(Listing.condition == ListingCondition(condition))

    if min_price is not None and max_price is not None and min_price > max_price:
        min_price, max_price = max_price, min_price  # prevent an inverted range
    if min_price is not None:
        query = query.where(Listing.price >= min_price)
    if max_price is not None:
        query = query.where(Listing.price <= max_price)

    # --- Seller-quality filters (cached reputation + badge service only) ---
    if seller_rating == "new":
        query = query.where(or_(UserReputation.review_count.is_(None), UserReputation.review_count == 0))
    elif seller_rating:
        try:
            threshold = float(seller_rating)
        except (TypeError, ValueError):
            threshold = None
        if threshold is not None:
            query = query.where(UserReputation.review_count > 0, UserReputation.average_rating >= threshold)
    if trusted_only:
        query = query.where(trusted_seller_filter_expr())
    if badge:
        expr = badge_filter_expr(badge)
        if expr is not None:
            query = query.where(expr)

    query = query.order_by(*_catalog_order_by(sort))
    return db.paginate(query, page=page, per_page=per_page, error_out=False)


def search_published_listings(page: int, **kwargs):
    """Backwards-compatible alias for the basic catalog search."""
    return search_listings(page, **kwargs)


def get_search_suggestions(q: str | None, limit: int = 5) -> list[str]:
    """Up to ``limit`` distinct listing-title suggestions for a typed query.

    Prefix matches rank first (``mac`` → MacBook, MacBook Air, MacBook Pro), then
    'contains' matches. Uses only existing listing titles — no external index.
    """
    term = (q or "").strip()
    if len(term) < 2:
        return []
    like = f"%{_escape_like(term)}%"
    titles = db.session.execute(
        db.select(Listing.title)
        .where(
            Listing.status == ListingStatus.AVAILABLE,
            Listing.visibility == ListingVisibility.PUBLIC,
            Listing.deleted_at.is_(None),
            Listing.title.ilike(like, escape="\\"),
        )
        .distinct()
        .limit(limit * 5)
    ).scalars().all()

    lower = term.lower()
    seen: set[str] = set()
    ordered = sorted(titles, key=lambda t: (not t.lower().startswith(lower), t.lower()))
    suggestions: list[str] = []
    for title in ordered:
        key = title.lower()
        if key in seen:
            continue
        seen.add(key)
        suggestions.append(title)
        if len(suggestions) >= limit:
            break
    return suggestions


def get_visible_listing_or_404(public_id: str) -> Listing:
    """Return a non-deleted listing by public id, or abort with 404."""
    return db.first_or_404(
        db.select(Listing).where(
            Listing.public_id == public_id,
            Listing.deleted_at.is_(None),
        )
    )


_LISTING_SORTS = {
    "newest": Listing.created_at.desc(),
    "oldest": Listing.created_at.asc(),
    "price_low": Listing.price.asc(),
    "price_high": Listing.price.desc(),
    # Cached listing counters — no aggregation needed.
    "most_viewed": Listing.view_count.desc(),
    "most_favorited": Listing.favorite_count.desc(),
    "recently_updated": Listing.updated_at.desc(),
    # Cached seller reputation (query joins UserReputation for these).
    "best_rated_seller": func.coalesce(UserReputation.average_rating, 0).desc(),
    "most_reviews": func.coalesce(UserReputation.review_count, 0).desc(),
}

# Sorts that need the UserReputation join.
_REPUTATION_SORTS = {"best_rated_seller", "most_reviews"}


def get_user_listings(
    user: User,
    page: int,
    *,
    q: str | None = None,
    status: str | None = None,
    sort: str = "newest",
):
    """Return a pagination of a user's own (non-deleted) listings.

    Server-side search/filter/sort over real columns only:
        q       - case-insensitive match on title or description.
        status  - a :class:`ListingStatus` value to filter by (None = all).
        sort    - one of ``_LISTING_SORTS`` keys (defaults to newest).
    """
    per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20)
    # Expire lapsed featured listings so My Listings shows "Feature Listing"
    # again and the featured badge/timer reflect reality.
    cleanup_expired_featured_listings()
    query = db.select(Listing).where(
        Listing.seller_id == user.id,
        Listing.deleted_at.is_(None),
    )

    if q:
        # Treat user input literally: escape LIKE wildcards so '%' and '_'
        # match themselves rather than acting as patterns.
        escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        like = f"%{escaped}%"
        query = query.where(
            or_(
                Listing.title.ilike(like, escape="\\"),
                Listing.description.ilike(like, escape="\\"),
            )
        )
    if status:
        query = query.where(Listing.status == ListingStatus(status))

    query = query.order_by(_LISTING_SORTS.get(sort, _LISTING_SORTS["newest"]))
    return db.paginate(query, page=page, per_page=per_page, error_out=False)


def user_has_listings(user: User) -> bool:
    """True if the user has at least one non-deleted listing (ignoring filters)."""
    count = db.session.scalar(
        db.select(db.func.count())
        .select_from(Listing)
        .where(Listing.seller_id == user.id, Listing.deleted_at.is_(None))
    )
    return bool(count)


def get_user_listing_status_counts(user: User) -> dict[str, int]:
    """Return aggregated counts for a user's non-deleted listings by status."""
    rows = db.session.execute(
        db.select(Listing.status, db.func.count())
        .where(Listing.seller_id == user.id, Listing.deleted_at.is_(None))
        .group_by(Listing.status)
    ).all()

    status_counts = {status.value: count for status, count in rows}
    return {
        "total": sum(status_counts.values()),
        "available": status_counts.get("available", 0),
        "reserved": status_counts.get("reserved", 0),
        "sold": status_counts.get("sold", 0),
        # Per-status counts (draft/expired/removed included) for the filter chips.
        "by_status": status_counts,
    }


def get_seller_overview(user: User) -> dict:
    """Read model for the My Listings control-center KPIs and weekly insights.

    A handful of bounded aggregation queries (COUNT/SUM over the seller's
    non-deleted listings, plus 7-day windows) — never one-per-listing. Views and
    favorites totals come from the cached counter columns; the weekly figures use
    the event tables. Read-only; no business rule lives here.
    """
    now = utcnow()
    week_ago = now - timedelta(days=7)
    listing_ids = db.select(Listing.id).where(
        Listing.seller_id == user.id, Listing.deleted_at.is_(None)
    )

    total_listings, total_views, total_favorites = db.session.execute(
        db.select(
            func.count(Listing.id),
            func.coalesce(func.sum(Listing.view_count), 0),
            func.coalesce(func.sum(Listing.favorite_count), 0),
        ).where(Listing.seller_id == user.id, Listing.deleted_at.is_(None))
    ).one()
    total_conversations = db.session.scalar(
        db.select(func.count(Conversation.id)).where(Conversation.seller_id == user.id)
    ) or 0
    total_offers = db.session.scalar(
        db.select(func.count(Offer.id)).where(Offer.seller_id == user.id)
    ) or 0

    views_7d = db.session.scalar(
        db.select(func.count(ListingView.id)).where(
            ListingView.listing_id.in_(listing_ids), ListingView.viewed_at >= week_ago)
    ) or 0
    favorites_7d = db.session.scalar(
        db.select(func.count(Favorite.id))
        .where(Favorite.listing_id.in_(listing_ids), Favorite.created_at >= week_ago)
    ) or 0
    conversations_7d = db.session.scalar(
        db.select(func.count(Conversation.id))
        .where(Conversation.seller_id == user.id, Conversation.created_at >= week_ago)
    ) or 0

    top = db.session.scalar(
        db.select(Listing)
        .where(Listing.seller_id == user.id, Listing.deleted_at.is_(None), Listing.view_count > 0)
        .order_by(Listing.view_count.desc(), Listing.created_at.desc())
        .limit(1)
    )
    return {
        "total_listings": int(total_listings or 0),
        "total_views": int(total_views or 0),
        "total_favorites": int(total_favorites or 0),
        "total_conversations": int(total_conversations),
        "total_offers": int(total_offers),
        "views_7d": int(views_7d),
        "favorites_7d": int(favorites_7d),
        "conversations_7d": int(conversations_7d),
        "top_listing": {"title": top.title, "views": top.view_count or 0,
                        "public_id": top.public_id} if top else None,
    }


def get_offers_by_listing(listing_ids) -> dict[int, int]:
    """Offer counts keyed by listing id for a page of listings (one GROUP BY)."""
    ids = list(listing_ids)
    if not ids:
        return {}
    return {
        lid: int(c) for lid, c in db.session.execute(
            db.select(Offer.listing_id, func.count()).where(Offer.listing_id.in_(ids)).group_by(Offer.listing_id)
        ).all()
    }


def get_related_listings(listing: Listing, limit: int = 3) -> list[Listing]:
    """Return other available listings related to ``listing`` (same category).

    Simple "same category, newest first" query — not a recommendation engine.
    Excludes the listing itself and anything not publicly available. Falls back
    to filling remaining slots with other recent public listings so the section
    is never awkwardly empty when a category has few items.
    """
    base = (
        db.select(Listing)
        .where(
            Listing.id != listing.id,
            Listing.status == ListingStatus.AVAILABLE,
            Listing.visibility == ListingVisibility.PUBLIC,
            Listing.deleted_at.is_(None),
        )
        .options(selectinload(Listing.images), selectinload(Listing.category))
    )

    related = list(
        db.session.scalars(
            base.where(Listing.category_id == listing.category_id)
            .order_by(Listing.created_at.desc())
            .limit(limit)
        )
    )

    if len(related) < limit:
        seen = {l.id for l in related}
        extra = db.session.scalars(
            base.where(Listing.category_id != listing.category_id)
            .order_by(Listing.created_at.desc())
            .limit(limit)
        )
        for candidate in extra:
            if candidate.id not in seen:
                related.append(candidate)
            if len(related) >= limit:
                break

    return related[:limit]


def get_public_seller_listings(user: User, limit: int | None = None) -> list[Listing]:
    """Return a seller's publicly visible, available listings for their profile.

    Only surfaces items a buyer may legitimately see (available + public +
    approved + not deleted). Eager-loads images and category so the listing
    cards render without an N+1.
    """
    query = (
        db.select(Listing)
        .where(
            Listing.seller_id == user.id,
            Listing.status == ListingStatus.AVAILABLE,
            Listing.visibility == ListingVisibility.PUBLIC,
            Listing.review_status == ReviewStatus.APPROVED,
            Listing.deleted_at.is_(None),
        )
        .options(
            selectinload(Listing.images),
            selectinload(Listing.category),
        )
        .order_by(Listing.created_at.desc())
    )
    if limit:
        query = query.limit(limit)
    return list(db.session.scalars(query))


def get_owned_listing_or_404(public_id: str, user: User) -> Listing:
    """Return the user's own non-deleted listing by public id, or abort 404.

    Returning 404 (not 403) for someone else's listing avoids disclosing that
    the listing exists. Ownership is enforced server-side only.
    """
    return db.first_or_404(
        db.select(Listing).where(
            Listing.public_id == public_id,
            Listing.seller_id == user.id,
            Listing.deleted_at.is_(None),
        )
    )


def update_listing(
    listing: Listing,
    editor: User,
    *,
    title: str,
    description: str,
    category_id: int,
    price,
    condition: str,
    negotiable: bool,
) -> Listing:
    """Apply edits to an existing listing (text-only fields; images untouched)."""
    listing.title = title.strip()
    listing.description = description.strip()
    listing.category_id = category_id
    listing.price = price
    listing.condition = ListingCondition(condition)
    listing.negotiable = negotiable
    listing.updated_by = editor.id
    db.session.commit()
    return listing


def soft_delete_listing(listing: Listing, editor: User) -> None:
    """Soft-delete a listing: set ``deleted_at`` and mark it removed.

    Soft-deleted listings are excluded from the marketplace, search and
    My Listings (all queries filter ``deleted_at IS NULL``).
    """
    was_featured_visible = listing.is_featured and _listing_is_visible(listing)
    listing.deleted_at = utcnow()
    listing.status = ListingStatus.REMOVED
    listing.updated_by = editor.id
    # A removed listing can no longer be featured (keeps "featured => available").
    listing.is_featured = False
    listing.featured_at = None
    listing.featured_until = None
    db.session.commit()
    if was_featured_visible:
        notify_featured_listing_hidden(listing, "removed")


def _listing_is_visible(listing: Listing) -> bool:
    """Whether a listing currently qualifies for the public discovery pool
    (the same rule the Featured/discovery queries enforce)."""
    return (
        listing.status == ListingStatus.AVAILABLE
        and listing.visibility == ListingVisibility.PUBLIC
        and listing.deleted_at is None
    )


# Reason-specific copy so the seller immediately understands *why* their featured
# listing dropped out of the Featured section (each ``{title}`` gets the listing
# name). ``reason`` is one of the keys below; unknown reasons fall back to generic.
_FEATURED_HIDDEN_COPY = {
    "reserved": (
        "Featured listing reserved",
        "Your featured listing '{title}' is no longer displayed because it has been reserved.",
    ),
    "sold": (
        "Featured listing sold",
        "Your featured listing '{title}' has been removed from Featured because it has been sold.",
    ),
    "removed": (
        "Featured listing removed",
        "Your featured listing '{title}' is no longer visible because it was removed.",
    ),
    "expired": (
        "Featured listing expired",
        "Your featured listing '{title}' has expired and is no longer displayed.",
    ),
}
# Shared, actionable next step appended to every featured-hidden notice.
_FEATURED_HIDDEN_CTA = " Feature another available listing to appear in the Featured section again."


def notify_featured_listing_hidden(listing: Listing, reason: str) -> None:
    """Tell a Premium seller their featured listing left the Featured section.

    Called only at a visible → non-visible transition (reserved / sold / removed /
    expired), never on read, with reason-specific copy so the seller learns exactly
    why their promotion vanished and how to restore it.
    """
    seller = listing.seller
    if seller is None:
        return
    title, body = _FEATURED_HIDDEN_COPY.get(
        reason,
        ("Featured listing no longer displayed",
         "Your featured listing '{title}' is no longer displayed."),
    )
    create_notification(
        user=seller,
        notification_type=NotificationType.LISTING,
        title=title,
        message=body.format(title=listing.title) + _FEATURED_HIDDEN_CTA,
        related_entity_type=RelatedEntityType.LISTING,
        related_entity_id=listing.id,
    )


class ListingLifecycleError(Exception):
    """Raised when a listing transitions outside the supported lifecycle."""


def _notify_listing_status_change(
    listing: Listing,
    *,
    new_status: ListingStatus,
    recipient: User,
    title: str,
    message: str,
) -> None:
    existing = db.session.scalar(
        db.select(Notification)
        .where(
            Notification.user_id == recipient.id,
            Notification.notification_type == NotificationType.LISTING,
            Notification.related_entity_type == RelatedEntityType.LISTING,
            Notification.related_entity_id == listing.id,
            Notification.read_at.is_(None),
        )
        .limit(1)
    )
    if existing is None:
        create_notification(
            user=recipient,
            notification_type=NotificationType.LISTING,
            title=title,
            message=message,
            related_entity_type=RelatedEntityType.LISTING,
            related_entity_id=listing.id,
        )


def reserve_listing_for_offer(offer, *, commit: bool = True) -> Listing:
    """Reserve a listing when an offer is accepted."""
    listing = offer.listing
    if listing is None:
        raise ListingLifecycleError("This listing is no longer available.")
    if listing.status == ListingStatus.SOLD:
        raise ListingLifecycleError("This listing has already been sold.")
    if listing.status == ListingStatus.RESERVED:
        if commit:
            db.session.commit()
        return listing

    # Capture visibility before the transition so we notify the seller exactly
    # once, only when a *displayed* featured listing drops out of the pool.
    was_featured_visible = listing.is_featured and _listing_is_visible(listing)

    listing.status = ListingStatus.RESERVED
    listing.updated_by = offer.seller_id
    if commit:
        db.session.commit()

    _notify_listing_status_change(
        listing,
        new_status=ListingStatus.RESERVED,
        recipient=listing.seller,
        title="Listing reserved",
        message=(
            f"The listing '{listing.title}' is now reserved for the accepted offer."
        ),
    )
    if was_featured_visible:
        notify_featured_listing_hidden(listing, "reserved")
    return listing


def mark_listing_as_sold(listing: Listing, seller: User, *, commit: bool = True, notify: bool = True) -> Listing:
    """Move a reserved listing to sold state for the listing owner.

    ``notify=False`` skips the generic "Listing sold" notifications — used by
    :func:`complete_transaction`, which sends its own transaction-completed
    notifications so the two are never duplicated.
    """
    if listing.seller_id != seller.id:
        raise ListingLifecycleError("Only the listing owner can mark it as sold.")
    if listing.status == ListingStatus.SOLD:
        return listing
    if listing.status != ListingStatus.RESERVED:
        raise ListingLifecycleError("Listings must be reserved before they can be marked as sold.")

    listing.status = ListingStatus.SOLD
    listing.updated_by = seller.id
    # Sold listings drop out of the catalog, so they can't stay featured.
    listing.is_featured = False
    listing.featured_at = None
    listing.featured_until = None
    if commit:
        db.session.commit()

    if notify:
        for recipient in {listing.seller, listing.conversations[0].buyer if listing.conversations else None}:
            if recipient is None:
                continue
            _notify_listing_status_change(
                listing,
                new_status=ListingStatus.SOLD,
                recipient=recipient,
                title="Listing sold",
                message=(f"The listing '{listing.title}' has been marked as sold."),
            )
    return listing


# ---------------------------------------------------------------------------
# Completed transactions (Trust System — Phase 1)
#
# Completing a transaction is the single authoritative way a sale is finalised:
# it records who bought the item (from the accepted offer), flips the listing to
# SOLD and the offer to COMPLETED, and notifies both parties — all in one commit.
# ---------------------------------------------------------------------------

class TransactionError(Exception):
    """Raised when a transaction cannot be completed for the current state."""


def _accepted_offer_for_listing(listing: Listing) -> Offer | None:
    """The accepted offer that reserved this listing (most recent if several)."""
    return db.session.scalar(
        db.select(Offer)
        .where(Offer.listing_id == listing.id, Offer.status == OfferStatus.ACCEPTED)
        .order_by(Offer.created_at.desc())
        .limit(1)
    )


def get_transaction_for_listing(listing: Listing) -> CompletedTransaction | None:
    """Return the completed transaction for a listing, if the sale is finalised."""
    return db.session.scalar(
        db.select(CompletedTransaction).where(CompletedTransaction.listing_id == listing.id)
    )


def complete_transaction(listing: Listing, seller: User) -> CompletedTransaction:
    """Finalise a sale: record a :class:`CompletedTransaction`, mark the listing
    SOLD and the offer COMPLETED, and notify both parties — in one commit.

    Idempotency is guaranteed at two levels: this service refuses to run twice
    (raising :class:`TransactionError`), and the unique constraints on
    ``completed_transactions`` are the last line of defence against a race.

    The buyer/seller are taken from the accepted offer's conversation, so the
    record is trustworthy regardless of whether a buyer offer or a seller
    counter-offer was the one accepted.
    """
    if listing.seller_id != seller.id:
        raise TransactionError("Only the listing owner can complete this transaction.")
    if get_transaction_for_listing(listing) is not None:
        raise TransactionError("This transaction has already been completed.")
    if listing.status == ListingStatus.SOLD:
        raise TransactionError("This listing has already been sold.")
    if listing.status != ListingStatus.RESERVED:
        raise TransactionError("A buyer must reserve the listing (accepted offer) before completion.")

    offer = _accepted_offer_for_listing(listing)
    if offer is None or offer.conversation is None:
        raise TransactionError("No accepted offer was found for this listing.")

    conversation = offer.conversation
    buyer_id = conversation.buyer_id            # canonical buyer, not the offer maker
    seller_id = listing.seller_id

    transaction = CompletedTransaction(
        listing_id=listing.id,
        offer_id=offer.id,
        buyer_id=buyer_id,
        seller_id=seller_id,
        completed_at=utcnow(),
    )
    db.session.add(transaction)

    # Reuse the existing lifecycle transition (without its generic notifications).
    mark_listing_as_sold(listing, seller, commit=False, notify=False)
    offer.status = OfferStatus.COMPLETED

    try:
        db.session.commit()
    except IntegrityError as exc:  # unique(listing_id)/unique(offer_id) — concurrent double submit
        db.session.rollback()
        raise TransactionError("This transaction has already been completed.") from exc

    # Refresh both parties' cached sale/purchase counts synchronously.
    from app.reviews.reputation import update_reputation_after_transaction
    update_reputation_after_transaction(transaction)

    _notify_transaction_completed(transaction, listing)
    _publish_activity(seller, "listing_sold", "Listing sold",
                      message=f"You completed the sale of “{listing.title}”.",
                      entity_type=RelatedEntityType.LISTING, entity_id=listing.id)
    return transaction


def _notify_transaction_completed(transaction: CompletedTransaction, listing: Listing) -> None:
    """Notify both parties that the sale is finalised (reuses the notification
    service). A delivery failure never rolls back the committed transaction."""
    recipients = [
        (transaction.seller_id, "Transaction Completed", "You successfully completed the sale."),
        (transaction.buyer_id, "Transaction Completed", "The seller has marked the transaction as complete."),
    ]
    for user_id, title, message in recipients:
        user = db.session.get(User, user_id)
        if user is None:
            continue
        try:
            create_notification(
                user=user,
                notification_type=NotificationType.LISTING,
                title=title,
                message=message,
                related_entity_type=RelatedEntityType.LISTING,
                related_entity_id=listing.id,
            )
        except Exception:  # pragma: no cover - notifications must never break the sale
            db.session.rollback()
            current_app.logger.exception(
                "Failed to send transaction-completed notification to user %s", user_id
            )


# ---------------------------------------------------------------------------
# Favorites
#
# All favorite business logic is centralized here. Routes must never touch the
# Favorite rows or ``favorite_count`` directly — they call these functions.
# ---------------------------------------------------------------------------


class FavoriteError(Exception):
    """Base error for favorite operations that violate a business rule."""


class SelfFavoriteError(FavoriteError):
    """Raised when a user tries to favorite their own listing."""


def is_favorited(user: User, listing: Listing) -> bool:
    """Return True if ``user`` has already favorited ``listing``."""
    if user is None or getattr(user, "id", None) is None:
        return False
    return db.session.scalar(
        db.select(Favorite.id).filter_by(user_id=user.id, listing_id=listing.id)
    ) is not None


def add_favorite(user: User, listing: Listing) -> Favorite:
    """Save ``listing`` to ``user``'s favorites.

    Enforces the business rules: a user cannot favorite their own listing, and
    a listing can be favorited only once. Idempotent — re-saving an existing
    favorite returns it without changing ``favorite_count``. Keeps
    ``listing.favorite_count`` in sync. Returns the Favorite record.
    """
    if listing.seller_id == user.id:
        raise SelfFavoriteError("You cannot save your own listing.")

    existing = db.session.scalar(
        db.select(Favorite).filter_by(user_id=user.id, listing_id=listing.id)
    )
    if existing is not None:
        return existing

    favorite = Favorite(user_id=user.id, listing_id=listing.id)
    db.session.add(favorite)
    listing.favorite_count = (listing.favorite_count or 0) + 1
    try:
        db.session.commit()
    except IntegrityError:
        # Concurrent double-submit hit the UNIQUE(user_id, listing_id) guard:
        # roll back and return the row the other request created.
        db.session.rollback()
        return db.session.scalar(
            db.select(Favorite).filter_by(user_id=user.id, listing_id=listing.id)
        )
    # Tell the seller a buyer saved their listing (grouped in the feed).
    if listing.seller_id != user.id:
        seller = db.session.get(User, listing.seller_id)
        _publish_activity(seller, "favorite_received", "Listing saved",
                          message=f"Someone saved your listing “{listing.title}”.",
                          entity_type=RelatedEntityType.LISTING, entity_id=listing.id)
    return favorite


def remove_favorite(user: User, listing: Listing) -> bool:
    """Remove ``listing`` from ``user``'s favorites.

    Deletes only the Favorite record and decrements ``favorite_count``
    (never below zero). Returns True if a favorite was removed, False if there
    was nothing to remove.
    """
    favorite = db.session.scalar(
        db.select(Favorite).filter_by(user_id=user.id, listing_id=listing.id)
    )
    if favorite is None:
        return False

    db.session.delete(favorite)
    listing.favorite_count = max(0, (listing.favorite_count or 0) - 1)
    db.session.commit()
    return True


def get_user_favorites(user: User, page: int):
    """Return a pagination of the user's favorites whose listing is not deleted.

    Ordered by most recently saved first. Items are :class:`Favorite` records
    (each exposes ``.listing`` and the ``created_at`` it was saved).
    """
    per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20)
    query = (
        db.select(Favorite)
        .join(Listing, Favorite.listing_id == Listing.id)
        .where(Favorite.user_id == user.id, Listing.deleted_at.is_(None))
        .order_by(Favorite.created_at.desc())
    )
    return db.paginate(query, page=page, per_page=per_page, error_out=False)


# --- Listing view tracking (Phase 4.1) -------------------------------------
#
# One de-duplicated view is recorded per genuine visit. Duplicate detection uses
# an EXISTS query over an indexed 30-minute window; a legitimate view inserts one
# ``ListingView`` row and increments the cached ``listings.view_count`` (the
# count displayed throughout the app). Raw IPs / user agents are never stored —
# only salted SHA-256 hashes. All logic lives here; routes only call
# ``record_listing_view``.

VIEW_DEDUP_WINDOW_MINUTES = 30


def _hash_value(value: str | None) -> str | None:
    """Salted SHA-256 of ``value`` (or None). Never store the raw value.

    Salted with the app ``SECRET_KEY`` so hashes can't be reversed via a rainbow
    table of common IPs.
    """
    if not value:
        return None
    secret = current_app.config.get("SECRET_KEY", "") or ""
    return hashlib.sha256(f"{value}::{secret}".encode("utf-8")).hexdigest()


def _client_ip(request) -> str | None:
    """Best-effort client IP (first hop of X-Forwarded-For, else remote_addr)."""
    if request is None:
        return None
    forwarded = request.headers.get("X-Forwarded-For", "")
    if forwarded:
        return forwarded.split(",")[0].strip() or None
    return request.remote_addr


def _view_identity(current_user, request) -> dict:
    """Resolve the (hashed) identity used to de-duplicate a view.

    Returns viewer_id (logged-in), a stable per-browser ``session_id`` (for
    anonymous dedup) and salted IP / user-agent hashes.
    """
    authenticated = getattr(current_user, "is_authenticated", False)
    session_id = None
    try:
        session_id = session.get("view_sid")
        if session_id is None:
            session_id = uuid.uuid4().hex
            session["view_sid"] = session_id
    except RuntimeError:
        session_id = None  # outside a request/session context

    return {
        "viewer_id": current_user.id if authenticated else None,
        "session_id": session_id,
        "ip_hash": _hash_value(_client_ip(request)),
        "user_agent_hash": _hash_value(request.headers.get("User-Agent") if request else None),
    }


def _recent_view_condition(listing: Listing, identity: dict):
    """Build the WHERE for the duplicate-detection EXISTS, or None when the
    visitor can't be identified at all (so the view simply counts)."""
    cutoff = utcnow() - timedelta(minutes=VIEW_DEDUP_WINDOW_MINUTES)
    base = db.and_(ListingView.listing_id == listing.id, ListingView.viewed_at >= cutoff)

    if identity["viewer_id"] is not None:
        return db.and_(base, ListingView.viewer_id == identity["viewer_id"])

    # Anonymous: same session, OR same hashed IP + user agent.
    anon_clauses = []
    if identity["session_id"]:
        anon_clauses.append(ListingView.session_id == identity["session_id"])
    if identity["ip_hash"] and identity["user_agent_hash"]:
        anon_clauses.append(
            db.and_(
                ListingView.ip_hash == identity["ip_hash"],
                ListingView.user_agent_hash == identity["user_agent_hash"],
            )
        )
    if not anon_clauses:
        return None
    return db.and_(base, ListingView.viewer_id.is_(None), or_(*anon_clauses))


def has_recent_view(listing: Listing, current_user, request) -> bool:
    """Whether this visitor already viewed ``listing`` within the dedup window.

    A single indexed EXISTS query — loads no rows.
    """
    condition = _recent_view_condition(listing, _view_identity(current_user, request))
    if condition is None:
        return False
    return bool(db.session.scalar(db.select(db.exists().where(condition))))


def record_listing_view(listing: Listing, current_user, request) -> bool:
    """Record a genuine listing view (de-duplicated). Returns True when a new
    view was counted, False when it was suppressed.

    Suppressed when the viewer is the listing owner or has viewed within the last
    ``VIEW_DEDUP_WINDOW_MINUTES``. On a genuine view: inserts one ``ListingView``
    row and increments the cached ``listing.view_count`` — the single writer of
    that column.
    """
    # The owner viewing their own listing is not a genuine view.
    if getattr(current_user, "is_authenticated", False) and current_user.id == listing.seller_id:
        return False

    identity = _view_identity(current_user, request)
    condition = _recent_view_condition(listing, identity)
    if condition is not None and db.session.scalar(db.select(db.exists().where(condition))):
        return False

    db.session.add(ListingView(
        listing_id=listing.id,
        viewer_id=identity["viewer_id"],
        session_id=identity["session_id"],
        ip_hash=identity["ip_hash"],
        user_agent_hash=identity["user_agent_hash"],
    ))
    listing.view_count = (listing.view_count or 0) + 1
    db.session.commit()
    return True


def listing_view_statistics(listing: Listing) -> dict:
    """Aggregate view stats for a listing (for future trend analytics).

    Bounded aggregation queries — no rows loaded. The cached ``view_count`` stays
    the displayed figure; these read the ``listing_views`` history.
    """
    cutoff_7d = utcnow() - timedelta(days=7)
    listing_scope = ListingView.listing_id == listing.id
    tracked = db.session.scalar(
        db.select(func.count()).select_from(ListingView).where(listing_scope)
    ) or 0
    unique_users = db.session.scalar(
        db.select(func.count(func.distinct(ListingView.viewer_id)))
        .where(listing_scope, ListingView.viewer_id.isnot(None))
    ) or 0
    anonymous = db.session.scalar(
        db.select(func.count()).select_from(ListingView)
        .where(listing_scope, ListingView.viewer_id.is_(None))
    ) or 0
    last_7_days = db.session.scalar(
        db.select(func.count()).select_from(ListingView)
        .where(listing_scope, ListingView.viewed_at >= cutoff_7d)
    ) or 0
    return {
        "cached_view_count": listing.view_count or 0,
        "tracked_views": int(tracked),
        "unique_logged_in_viewers": int(unique_users),
        "anonymous_views": int(anonymous),
        "views_last_7_days": int(last_7_days),
    }