"""Artist overview + paged images.

Images are linked to an artist through the provenance chain
Source(artist_id) -> ImageProvenance(source_id) -> ImageRecord. DISTINCT
guards against an image having several provenance rows for one artist.
Dates come from Post.post_date via ImageProvenance.post_id.
"""

from dataclasses import dataclass

from sqlalchemy import and_, case, func, or_, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import (
    Artist,
    ArtistVisit,
    ImageProvenance,
    ImageRecord,
    Post,
    Source,
    Tag,
)
from ..models.tag import image_tag
from ..utils.slug import slugify
from .db_helpers import get_or_create
from .gallery_service import thumbnail_url
from .pagination import decode_cursor, encode_cursor


@dataclass(frozen=True)
class ArtistImagesPage:
    images: list[dict]
    next_cursor: str | None


class ArtistService:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def _artist_by_slug(self, slug: str) -> Artist | None:
        return (
            await self.session.execute(
                select(Artist).where(Artist.slug == slug)
            )
        ).scalar_one_or_none()

    async def overview(self, slug: str) -> dict | None:
        artist = await self._artist_by_slug(slug)
        if artist is None:
            return None
        aid = artist.id

        img_join = (
            select(ImageRecord.id).where(ImageRecord.artist_id == aid)
        )
        image_count = (
            await self.session.execute(
                select(func.count(ImageRecord.id))
                .where(ImageRecord.artist_id == aid)
            )
        ).scalar_one()

        # Posts under this artist that have at least one image attached.
        # Use Post.artist_id (alembic 0030) for the artist filter; keep
        # the ImageProvenance JOIN so date bounds reflect only image-
        # bearing posts (matches the original semantic). NULL-source
        # posts now surface too.
        date_row = (
            await self.session.execute(
                select(func.min(Post.post_date), func.max(Post.post_date))
                .select_from(Post)
                .join(ImageProvenance, ImageProvenance.post_id == Post.id)
                .where(Post.artist_id == aid)
            )
        ).first()
        dmin, dmax = date_row if date_row else (None, None)

        cooccurring = (
            await self.session.execute(
                select(
                    Tag.id, Tag.name, Tag.kind,
                    func.count(image_tag.c.image_record_id).label("cnt"),
                )
                .select_from(Tag)
                .join(image_tag, image_tag.c.tag_id == Tag.id)
                .where(image_tag.c.image_record_id.in_(img_join))
                .group_by(Tag.id, Tag.name, Tag.kind)
                .order_by(func.count(image_tag.c.image_record_id).desc())
                .limit(20)
            )
        ).all()

        sources = (
            await self.session.execute(
                select(
                    Source.id, Source.platform, Source.url,
                    func.count(func.distinct(ImageProvenance.image_record_id)).label("cnt"),
                )
                .select_from(Source)
                .outerjoin(ImageProvenance, ImageProvenance.source_id == Source.id)
                .where(Source.artist_id == aid)
                .group_by(Source.id, Source.platform, Source.url)
                .order_by(Source.id)
            )
        ).all()

        # Same Post.artist_id direct filter — counts NULL-source posts too.
        month = func.date_trunc("month", Post.post_date).label("m")
        activity = (
            await self.session.execute(
                select(month, func.count(func.distinct(ImageProvenance.image_record_id)))
                .select_from(Post)
                .join(ImageProvenance, ImageProvenance.post_id == Post.id)
                .where(and_(Post.artist_id == aid, Post.post_date.isnot(None)))
                .group_by(month)
                .order_by(month)
            )
        ).all()

        post_count = (
            await self.session.execute(
                select(func.count(func.distinct(Post.id)))
                .where(Post.artist_id == aid)
            )
        ).scalar_one()

        # Mark this artist as "visited now"; the returned count is what
        # the operator should see in the banner ("N new since last
        # visit"). Done LAST so the read aggregates above all see the
        # pre-visit state (cosmetic — none depend on visit data).
        unseen_at_visit = await self._mark_visited_returning_unseen(aid)

        return {
            "id": artist.id,
            "name": artist.name,
            "slug": artist.slug,
            "is_subscription": bool(artist.is_subscription),
            "image_count": int(image_count),
            "post_count": int(post_count),
            "unseen_count_at_visit": unseen_at_visit,
            "date_range": {
                "min": dmin.isoformat() if dmin else None,
                "max": dmax.isoformat() if dmax else None,
            },
            "cooccurring_tags": [
                {
                    "id": tid,
                    "name": name,
                    "kind": kind.value if hasattr(kind, "value") else kind,
                    "count": int(cnt),
                }
                for tid, name, kind, cnt in cooccurring
            ],
            "sources": [
                {
                    "id": sid,
                    "platform": platform,
                    "url": url,
                    "image_count": int(cnt),
                }
                for sid, platform, url, cnt in sources
            ],
            "activity": [
                {"month": m.isoformat(), "count": int(cnt)}
                for m, cnt in activity
            ],
        }

    async def _mark_visited_returning_unseen(self, artist_id: int) -> int:
        """Read pre-visit `last_viewed_at`, count images added since,
        then upsert `last_viewed_at = NOW()`. Returns the count BEFORE
        the upsert so the banner has data to render.

        Postgres UPSERT (`ON CONFLICT DO UPDATE`) keeps the write
        atomic — no SELECT-then-INSERT race per
        `reference_scalar_one_or_none_duplicates`.
        """
        prev = (
            await self.session.execute(
                select(ArtistVisit.last_viewed_at).where(
                    ArtistVisit.artist_id == artist_id
                )
            )
        ).scalar_one_or_none()

        count_stmt = select(func.count(ImageRecord.id)).where(
            ImageRecord.artist_id == artist_id
        )
        if prev is not None:
            count_stmt = count_stmt.where(ImageRecord.created_at > prev)
        unseen = (await self.session.execute(count_stmt)).scalar_one()

        upsert = pg_insert(ArtistVisit.__table__).values(artist_id=artist_id)
        upsert = upsert.on_conflict_do_update(
            index_elements=["artist_id"],
            set_={"last_viewed_at": func.now()},
        )
        await self.session.execute(upsert)
        await self.session.commit()
        return int(unseen)

    async def images(
        self, slug: str, cursor: str | None, limit: int = 60
    ) -> ArtistImagesPage | None:
        if limit < 1 or limit > 200:
            raise ValueError("limit must be between 1 and 200")
        artist = await self._artist_by_slug(slug)
        if artist is None:
            return None

        # Dedupe via IN-subquery rather than JOIN + DISTINCT: an image can
        # have several provenance rows for one artist, and SELECT DISTINCT
        # over ImageRecord fails in Postgres because its json columns have
        # no equality operator.
        stmt = select(ImageRecord).where(
            ImageRecord.artist_id == artist.id
        )
        if cursor:
            cur_ts, cur_id = decode_cursor(cursor)
            stmt = stmt.where(
                or_(
                    ImageRecord.created_at < cur_ts,
                    and_(ImageRecord.created_at == cur_ts, ImageRecord.id < cur_id),
                )
            )
        stmt = stmt.order_by(
            ImageRecord.created_at.desc(), ImageRecord.id.desc()
        ).limit(limit + 1)
        rows = (await self.session.execute(stmt)).scalars().all()

        next_cursor = None
        if len(rows) > limit:
            last = rows[limit - 1]
            next_cursor = encode_cursor(last.created_at, last.id)
            rows = rows[:limit]

        return ArtistImagesPage(
            images=[
                {
                    "id": r.id,
                    "sha256": r.sha256,
                    "mime": r.mime,
                    "width": r.width,
                    "height": r.height,
                    "thumbnail_url": thumbnail_url(r.thumbnail_path, r.sha256, r.mime),
                }
                for r in rows
            ],
            next_cursor=next_cursor,
        )

    async def find_or_create(self, name: str) -> tuple[Artist, bool]:
        """Return (artist, created). Slug-keyed; idempotent under races via the
        shared race-safe db_helpers.get_or_create (savepoint + IntegrityError
        recovery). A new artist also seeds an ArtistVisit so the directory's
        `+N new` badge starts at 0.
        """
        cleaned = (name or "").strip()
        if not cleaned:
            raise ValueError("artist name must not be empty")
        slug = slugify(cleaned)

        select_existing = select(Artist).where(Artist.slug == slug)

        async def _create() -> Artist:
            artist = Artist(name=cleaned, slug=slug)
            self.session.add(artist)
            await self.session.flush()
            # New artist starts "caught up" — seed ArtistVisit so the
            # directory's `+N new` badge stays at 0 until real new
            # content arrives. Without this, the unseen-count query
            # treats NULL last_viewed_at as "never visited" and would
            # count every image imported in the same session.
            self.session.add(ArtistVisit(artist_id=artist.id))
            await self.session.flush()
            return artist

        artist, created = await get_or_create(
            self.session, select_existing, _create
        )
        if created:
            await self.session.commit()
        return artist, created

    async def rename(self, artist_id: int, name: str) -> Artist | None:
        """Change the display NAME only (#130). The slug — and every on-disk path
        keyed off it — is IMMUTABLE, so a rename never moves files or risks a path
        collision. Name is free text and NON-unique (migration 0077). Returns the
        updated Artist, None if not found; raises ValueError on empty name."""
        cleaned = (name or "").strip()
        if not cleaned:
            raise ValueError("artist name must not be empty")
        artist = (await self.session.execute(
            select(Artist).where(Artist.id == artist_id)
        )).scalar_one_or_none()
        if artist is None:
            return None
        artist.name = cleaned
        await self.session.commit()
        return artist

    async def autocomplete(self, prefix: str, limit: int = 20) -> list[Artist]:
        cleaned = (prefix or "").strip()
        if not cleaned:
            return []
        like = f"%{cleaned.lower()}%"
        prefix_like = f"{cleaned.lower()}%"
        # Rank: exact (0) < prefix (1) < substring (2).
        rank = case(
            (func.lower(Artist.name) == cleaned.lower(), 0),
            (func.lower(Artist.name).like(prefix_like), 1),
            else_=2,
        ).label("rank")
        rows = (await self.session.execute(
            select(Artist, rank)
            .where(func.lower(Artist.name).like(like))
            .order_by(rank, Artist.name.asc())
            .limit(limit)
        )).all()
        return [a for a, _ in rows]