FabledCurator/backend/app/services/source_service.py

"""FC-3a: CRUD over Source rows, with platform/url/config validation and
is_subscription auto-flip on first add / last delete.
"""

from dataclasses import dataclass

from sqlalchemy import func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import (
    Artist,
    DownloadEvent,
    ImageProvenance,
    ImageRecord,
    ImportSettings,
    Post,
    Source,
)
from .db_helpers import failing_sources_clause
from .gallery_dl import ErrorType
from .platforms import known_platform_keys
from .scheduler_service import compute_next_check_at

# Re-exported for back-compat with FC-3a; the registry in
# services/platforms.py is the single source of truth.
KNOWN_PLATFORMS = known_platform_keys()


# --- Errors -----------------------------------------------------------------

class SourceServiceError(Exception):
    """Base."""


class ArtistNotFoundError(SourceServiceError):
    pass


class UnknownPlatformError(SourceServiceError):
    def __init__(self, platform: str):
        super().__init__(f"unknown platform: {platform!r}")
        self.platform = platform


class InvalidConfigError(SourceServiceError):
    pass


class EmptyUrlError(SourceServiceError):
    pass


class DuplicateSourceError(SourceServiceError):
    def __init__(self, existing_id: int):
        super().__init__(f"duplicate source (existing id={existing_id})")
        self.existing_id = existing_id


@dataclass(frozen=True)
class SourceRecord:
    id: int
    artist_id: int
    artist_name: str
    artist_slug: str
    platform: str
    url: str
    enabled: bool
    config_overrides: dict | None
    last_checked_at: str | None
    last_error: str | None
    error_type: str | None
    check_interval_override: int | None
    consecutive_failures: int
    next_check_at: str | None
    backfill_runs_remaining: int
    # plan #693: derived from config_overrides for the UI badge.
    backfill_state: str | None  # "running" | "complete" | "stalled" | None (idle)
    backfill_chunks: int
    # plan #697: a running deep-walk that bypasses the Patreon seen-ledger
    # (recovery) vs. a normal backfill. Lets the badge label it "Recovering".
    backfill_bypass_seen: bool
    # #830: a running deep-walk in RECAPTURE mode (re-grab post bodies/links +
    # localize on-disk inline images, no media re-download). Lets the badge label
    # it "Recapturing".
    backfill_recapture: bool
    # plan #704: cumulative posts processed across the walk's chunks — live
    # progress for the badge.
    backfill_posts: int
    # Milestone #387 A3: posts the last walk skipped because the account can't
    # view them. Lives on the EVENT (run_stats.tier_gated_count), not the
    # source, so it is joined in by `list()` only — None everywhere else, which
    # the UI renders as the bare no-access state with no fabricated number.
    tier_gated_count: int | None = None

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "artist_id": self.artist_id,
            "artist_name": self.artist_name,
            "artist_slug": self.artist_slug,
            "platform": self.platform,
            "url": self.url,
            "enabled": self.enabled,
            "config_overrides": self.config_overrides,
            "last_checked_at": self.last_checked_at,
            "last_error": self.last_error,
            "error_type": self.error_type,
            "check_interval_override": self.check_interval_override,
            "consecutive_failures": self.consecutive_failures,
            "next_check_at": self.next_check_at,
            "backfill_runs_remaining": self.backfill_runs_remaining,
            "backfill_state": self.backfill_state,
            "backfill_chunks": self.backfill_chunks,
            "backfill_bypass_seen": self.backfill_bypass_seen,
            "backfill_recapture": self.backfill_recapture,
            "backfill_posts": self.backfill_posts,
            "tier_gated_count": self.tier_gated_count,
        }


# --- Service ----------------------------------------------------------------

_EDITABLE = {"enabled", "url", "config_overrides", "check_interval_override", "platform"}

# `config_overrides` carries two unrelated things under one column: the
# operator's per-source download settings, and state FC writes for ITSELF. An
# operator edit replaces the first wholesale — removing a key has to be able to
# remove it — but it must never take the second with it.
#
# Two families, both matching data already on disk:
#   `_*`             the #693 backfill state machine (_backfill_state, _cursor,
#                    _cursor_stalls, _chunks, _posts, _bypass_seen, _recapture)
#   `*_campaign_id`  the resolved platform identity cache, written by
#                    `download_service._phase3_persist`
_CAMPAIGN_ID_SUFFIX = "_campaign_id"


def _is_app_managed(key: str) -> bool:
    """Is this a key FC maintains, rather than one the operator edits?"""
    return key.startswith("_") or key.endswith(_CAMPAIGN_ID_SUFFIX)


# Plan #693: backfill safety cap. "Start backfill" (and a newly created
# enabled source) arms a run-until-done walk; this caps how many time-boxed
# chunks it may spend before pausing as "stalled", so a pathological walk that
# never reaches the bottom can't run forever. Generous on purpose — at
# BACKFILL_CHUNK_SECONDS (600s) per chunk this is ~33h of cumulative walk, far
# beyond any real catalog; the cursor stall-guard is the real terminator.
BACKFILL_MAX_CHUNKS = 200


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

    async def _artist_or_raise(self, artist_id: int) -> Artist:
        row = (await self.session.execute(
            select(Artist).where(Artist.id == artist_id)
        )).scalar_one_or_none()
        if row is None:
            raise ArtistNotFoundError(f"artist id={artist_id} not found")
        return row

    @staticmethod
    def _validate_platform(platform: str) -> str:
        if platform not in KNOWN_PLATFORMS:
            raise UnknownPlatformError(platform)
        return platform

    @staticmethod
    def _validate_url(url: str) -> str:
        cleaned = (url or "").strip()
        if not cleaned:
            raise EmptyUrlError("url must not be empty")
        return cleaned

    @staticmethod
    def _validate_config(config: object) -> dict | None:
        if config is None:
            return None
        if not isinstance(config, dict):
            raise InvalidConfigError("config_overrides must be a JSON object")
        return config

    @staticmethod
    def _merged_config(source: Source, incoming: dict | None) -> dict | None:
        """The operator's keys replace wholesale; FC's own keys survive.

        Without this, one edit in the Subscriptions dialog silently discarded
        the resolved campaign id AND the entire backfill position — the dialog
        posts the whole object back (`SourceFormDialog`), so anything absent
        from its JSON box was simply gone.

        FC's keys are applied LAST so they win: the dialog round-trips whatever
        it last read, and a client echoing a stale `_backfill_state` must not be
        able to overwrite what the walk has since written.
        """
        managed = {
            k: v for k, v in (source.config_overrides or {}).items()
            if _is_app_managed(k)
        }
        if incoming is None:
            # An explicit null clears the operator's settings. It is not a
            # request to forget where a backfill had got to.
            return managed or None
        operator = {k: v for k, v in incoming.items() if not _is_app_managed(k)}
        return {**operator, **managed}

    async def _load_settings(self) -> ImportSettings:
        return await ImportSettings.load(self.session)

    async def _tier_gated_counts(self, source_ids: list[int]) -> dict[int, int]:
        """Latest walk's tier-gated post count, per source, in ONE query.

        Selects the `run_stats` sub-object rather than whole `metadata` blobs:
        those carry truncated stdout/stderr up to 500KB each, and pulling one
        per source to read a single integer would make the subscriptions list
        pay for the Logs view. DISTINCT ON + ORDER BY takes the newest event per
        source (Postgres-only, like the rest of this codebase).

        Callers pass only the sources that actually need it — the count is
        meaningless for a source that isn't tier-gated.
        """
        if not source_ids:
            return {}
        rows = (await self.session.execute(
            select(
                DownloadEvent.source_id,
                DownloadEvent.metadata_["run_stats"],
            )
            .where(DownloadEvent.source_id.in_(source_ids))
            .distinct(DownloadEvent.source_id)
            .order_by(DownloadEvent.source_id, DownloadEvent.started_at.desc())
        )).all()
        counts: dict[int, int] = {}
        for source_id, run_stats in rows:
            n = (run_stats or {}).get("tier_gated_count") or 0
            if n:
                counts[source_id] = int(n)
        return counts

    def _build_record(
        self, source: Source, artist: Artist, settings: ImportSettings,
        gated_counts: dict[int, int] | None = None,
    ) -> SourceRecord:
        nxt = compute_next_check_at(source, artist, settings)
        co = source.config_overrides or {}
        return SourceRecord(
            id=source.id,
            artist_id=source.artist_id,
            artist_name=artist.name,
            artist_slug=artist.slug,
            platform=source.platform,
            url=source.url,
            enabled=source.enabled,
            config_overrides=source.config_overrides,
            last_checked_at=source.last_checked_at.isoformat() if source.last_checked_at else None,
            last_error=source.last_error,
            error_type=source.error_type,
            check_interval_override=source.check_interval_override,
            consecutive_failures=source.consecutive_failures or 0,
            next_check_at=nxt.isoformat() if nxt else None,
            backfill_runs_remaining=source.backfill_runs_remaining or 0,
            backfill_state=co.get("_backfill_state"),
            backfill_chunks=int(co.get("_backfill_chunks", 0)),
            backfill_bypass_seen=bool(co.get("_backfill_bypass_seen")),
            backfill_recapture=bool(co.get("_backfill_recapture")),
            backfill_posts=int(co.get("_backfill_posts", 0)),
            tier_gated_count=(gated_counts or {}).get(source.id),
        )

    async def _row_to_record(self, source: Source) -> SourceRecord:
        artist = (await self.session.execute(
            select(Artist).where(Artist.id == source.artist_id)
        )).scalar_one()
        settings = await self._load_settings()
        return self._build_record(source, artist, settings)

    async def list(
        self, artist_id: int | None = None, failing: bool = False,
        include_synthetic: bool = False,
    ) -> list[SourceRecord]:
        stmt = select(Source, Artist).join(Artist, Artist.id == Source.artist_id)
        if artist_id is not None:
            stmt = stmt.where(Source.artist_id == artist_id)
        if not include_synthetic:
            # Pre-alembic-0030 sidecar synthetic anchors
            # have url='sidecar:<platform>:<slug>' and exist only to give
            # imported Posts a NOT-NULL Source FK. They aren't pollable
            # feeds; the Subscriptions UI used to render them as phantom
            # subscriptions. Hide by default.
            stmt = stmt.where(~Source.url.like("sidecar:%"))
        if failing:
            # Worst-first so the rollup card surfaces the loudest failures.
            # Shared clause: the front-door ribbon counts with the same one, so
            # it can never report a number this list then contradicts.
            stmt = stmt.where(failing_sources_clause()).order_by(
                Source.consecutive_failures.desc(), Artist.name.asc(),
            )
        else:
            stmt = stmt.order_by(Artist.name.asc(), Source.id.asc())
        rows = (await self.session.execute(stmt)).all()
        settings = await self._load_settings()
        # Only tier-gated rows need the join — on a healthy library that is an
        # empty list and _tier_gated_counts short-circuits without a query.
        gated_counts = await self._tier_gated_counts(
            [s.id for s, _a in rows if s.error_type == ErrorType.TIER_LIMITED]
        )
        return [self._build_record(s, a, settings, gated_counts) for s, a in rows]

    async def get(self, source_id: int) -> SourceRecord | None:
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            return None
        return await self._row_to_record(source)

    async def create(
        self,
        *,
        artist_id: int,
        platform: str,
        url: str,
        enabled: bool = True,
        config_overrides: dict | None = None,
        check_interval_override: int | None = None,
    ) -> SourceRecord:
        await self._artist_or_raise(artist_id)
        platform = self._validate_platform(platform)
        url = self._validate_url(url)
        config_overrides = self._validate_config(config_overrides)

        prior_count = (await self.session.execute(
            select(func.count(Source.id)).where(Source.artist_id == artist_id)
        )).scalar_one()

        # Plan #693: a freshly added subscription has no archive yet, so it
        # should walk its full post history once. Arm run-until-done backfill
        # (state="running" + the chunk cap); the time-boxed chunks march to the
        # bottom across ticks, then flip to "complete" and tick mode takes over.
        # Disabled sources (incl. sidecar synthetics, url='sidecar:...') are
        # never polled, so leave them idle.
        if enabled:
            config_overrides = {**(config_overrides or {}), "_backfill_state": "running"}
            backfill_runs = BACKFILL_MAX_CHUNKS
        else:
            backfill_runs = 0
        source = Source(
            artist_id=artist_id, platform=platform, url=url,
            enabled=enabled, config_overrides=config_overrides,
            check_interval_override=check_interval_override,
            backfill_runs_remaining=backfill_runs,
        )
        self.session.add(source)
        try:
            await self.session.flush()
        except IntegrityError as exc:
            await self.session.rollback()
            existing = (await self.session.execute(
                select(Source.id).where(
                    Source.artist_id == artist_id,
                    Source.platform == platform,
                    Source.url == url,
                )
            )).scalar_one()
            raise DuplicateSourceError(existing_id=existing) from exc

        if prior_count == 0:
            await self.session.execute(
                Artist.__table__.update()
                .where(Artist.id == artist_id)
                .values(is_subscription=True)
            )
        await self.session.commit()
        return await self._row_to_record(source)

    async def update(self, source_id: int, **fields) -> SourceRecord:
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")

        for key in list(fields.keys()):
            if key not in _EDITABLE:
                fields.pop(key)
        if "platform" in fields:
            fields["platform"] = self._validate_platform(fields["platform"])
        if "url" in fields:
            fields["url"] = self._validate_url(fields["url"])
        if "config_overrides" in fields:
            fields["config_overrides"] = self._merged_config(
                source, self._validate_config(fields["config_overrides"])
            )

        # Computed BEFORE the setattr loop, while `source.url` is still the old
        # one. See the invalidation below.
        url_changed = "url" in fields and fields["url"] != source.url

        for key, value in fields.items():
            setattr(source, key, value)

        if url_changed:
            # Repointing a source at a different creator makes a cached campaign
            # id WRONG, not merely stale, and `patreon_resolver` consults that
            # cache BEFORE attempting any lookup — so a kept id would resolve the
            # old creator forever, and the membership join (#387 C4) would report
            # a confident wrong match.
            #
            # This needs saying explicitly only because of the merge above: until
            # then the wholesale overwrite wiped the id as an accident of the
            # bug, which masked this. Preserving the id makes the invalidation
            # this service's job.
            #
            # The backfill cursor is deliberately NOT cleared. It is opaque
            # platform state the walk already validates with its own stall
            # guard, and dropping it would restart a long backfill over a
            # cosmetic URL edit (http->https, adding `/c/`).
            co = dict(source.config_overrides or {})
            for key in [k for k in co if k.endswith(_CAMPAIGN_ID_SUFFIX)]:
                co.pop(key)
            source.config_overrides = co

        # Disabling a source clears its failure state (operator: disable the subs
        # you're not paying for without them lingering as "failing"). Re-enabling
        # then starts clean; the next real run re-derives health. Only on the
        # explicit disable — an unrelated edit to an already-disabled source
        # leaves its (already-clear) state alone.
        if fields.get("enabled") is False:
            source.last_error = None
            source.error_type = None
            source.consecutive_failures = 0
        try:
            await self.session.flush()
        except IntegrityError as exc:
            await self.session.rollback()
            existing = (await self.session.execute(
                select(Source.id).where(
                    Source.artist_id == source.artist_id,
                    Source.platform == source.platform,
                    Source.url == source.url,
                    Source.id != source.id,
                )
            )).scalar_one()
            raise DuplicateSourceError(existing_id=existing) from exc
        await self.session.commit()
        return await self._row_to_record(source)

    async def start_backfill(self, source_id: int) -> SourceRecord:
        """Plan #693: arm a run-until-done backfill. Sets state="running" and
        the chunk cap; download runs then walk the full post history in
        time-boxed chunks (skip:True + BACKFILL_CHUNK_SECONDS), resuming from
        the cursor each chunk, until gallery-dl reaches the bottom (→ state
        "complete") or the cap/stall-guard pauses it (→ "stalled"). Clears any
        prior cursor/chunk/stall state so a re-start walks fresh from the top."""
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")
        co = dict(source.config_overrides or {})
        co["_backfill_state"] = "running"
        for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks",
                  "_backfill_posts"):
            co.pop(k, None)
        source.config_overrides = co
        source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
        await self.session.commit()
        return await self._row_to_record(source)

    async def start_recovery(self, source_id: int) -> SourceRecord:
        """Plan #697: arm a RECOVERY walk — a backfill that bypasses the Patreon
        seen-ledger so deliberately-dropped-and-deleted near-dups get re-fetched
        and re-evaluated under the CURRENT pHash threshold (tier-2 disk still
        spares files we kept). Reuses the entire #693 backfill state machine
        (time-boxed chunks, cursor checkpoint, complete/stall lifecycle) plus the
        `_backfill_bypass_seen` flag that flips download mode to recovery. Clears
        any prior cursor/chunk/stall state so it walks fresh from the top. The
        flag is cleared on completion (download_service) and on stop.

        Recovery is Patreon-only (the seen-ledger is Patreon's); for other
        platforms the flag is inert (download_service ignores it) and the walk
        runs as a plain backfill. The UI gates the action to Patreon sources."""
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")
        co = dict(source.config_overrides or {})
        co["_backfill_state"] = "running"
        co["_backfill_bypass_seen"] = True
        for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks",
                  "_backfill_posts"):
            co.pop(k, None)
        source.config_overrides = co
        source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
        await self.session.commit()
        return await self._row_to_record(source)

    async def start_recapture(self, source_id: int) -> SourceRecord:
        """#830: arm a RECAPTURE walk — a backfill that re-grabs EVERY post's body
        + external links (detail-fetching empty bodies) and localizes already-on-
        disk inline images, WITHOUT re-downloading media. Reuses the entire #693
        backfill state machine plus a `_backfill_recapture` flag that flips
        download mode to recapture. Distinct from recovery (which re-downloads the
        whole source); the two flags are mutually exclusive, so arming recapture
        clears bypass_seen. Clears prior cursor/chunk/stall state so it walks
        fresh from the top. The flag is cleared on completion (download_service)
        and on stop. Recapture is Patreon-only (the native ingester's post-record
        capture); inert elsewhere. The UI gates the action to Patreon sources."""
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")
        co = dict(source.config_overrides or {})
        co["_backfill_state"] = "running"
        co["_backfill_recapture"] = True
        co.pop("_backfill_bypass_seen", None)  # mutually exclusive with recovery
        for k in ("_backfill_cursor", "_backfill_cursor_stalls", "_backfill_chunks",
                  "_backfill_posts"):
            co.pop(k, None)
        source.config_overrides = co
        source.backfill_runs_remaining = BACKFILL_MAX_CHUNKS
        await self.session.commit()
        return await self._row_to_record(source)

    async def stop_backfill(self, source_id: int) -> SourceRecord:
        """Plan #693: cancel an in-progress backfill — back to idle/tick mode.
        Clears the running state + cursor/chunk/stall bookkeeping."""
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")
        co = dict(source.config_overrides or {})
        for k in ("_backfill_state", "_backfill_cursor", "_backfill_cursor_stalls",
                  "_backfill_chunks", "_backfill_bypass_seen", "_backfill_recapture",
                  "_backfill_posts"):
            co.pop(k, None)
        source.config_overrides = co
        source.backfill_runs_remaining = 0
        await self.session.commit()
        return await self._row_to_record(source)

    async def reassign(self, source_id: int, target_artist_id: int) -> SourceRecord:
        """Move a Source — and the content it brought in — to another artist
        (#130). The slug/storage path is IMMUTABLE, so NO files move: only the
        artist attribution changes (reads use ImageRecord.path). Re-points the
        source's Posts and the ImageRecords it contributed (those still
        attributed to the old artist — images shared with another artist are
        left alone). If the old artist is left fully empty (no sources, images,
        or posts) it's deleted (ArtistVisit cascades); if it just lost its last
        source, its is_subscription flag clears. No-op when already on target."""
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")
        target = await self._artist_or_raise(target_artist_id)
        old_artist_id = source.artist_id
        if old_artist_id == target_artist_id:
            return await self._row_to_record(source)

        source.artist_id = target_artist_id
        target.is_subscription = True
        # Re-attribute this source's posts (Post.artist_id is denormalized).
        await self.session.execute(
            update(Post).where(Post.source_id == source_id)
            .values(artist_id=target_artist_id)
        )
        # Re-attribute the images this source contributed that are still on the
        # OLD artist. Scoping to artist_id == old avoids stealing an image that
        # a different artist's source also contributed.
        contributed = select(ImageProvenance.image_record_id).where(
            ImageProvenance.source_id == source_id
        )
        await self.session.execute(
            update(ImageRecord)
            .where(
                ImageRecord.artist_id == old_artist_id,
                ImageRecord.id.in_(contributed),
            )
            .values(artist_id=target_artist_id)
        )
        await self.session.flush()

        old = (await self.session.execute(
            select(Artist).where(Artist.id == old_artist_id)
        )).scalar_one_or_none()
        if old is not None:
            n_src = (await self.session.execute(
                select(func.count(Source.id)).where(Source.artist_id == old_artist_id)
            )).scalar_one()
            n_img = (await self.session.execute(
                select(func.count(ImageRecord.id)).where(
                    ImageRecord.artist_id == old_artist_id
                )
            )).scalar_one()
            n_post = (await self.session.execute(
                select(func.count(Post.id)).where(Post.artist_id == old_artist_id)
            )).scalar_one()
            if n_src == 0 and n_img == 0 and n_post == 0:
                await self.session.delete(old)      # ArtistVisit cascades
            elif n_src == 0:
                old.is_subscription = False
        await self.session.commit()
        return await self._row_to_record(source)

    async def delete(self, source_id: int) -> None:
        source = (await self.session.execute(
            select(Source).where(Source.id == source_id)
        )).scalar_one_or_none()
        if source is None:
            raise LookupError(f"source id={source_id} not found")
        artist_id = source.artist_id
        await self.session.delete(source)
        await self.session.flush()

        remaining = (await self.session.execute(
            select(func.count(Source.id)).where(Source.artist_id == artist_id)
        )).scalar_one()
        if remaining == 0:
            await self.session.execute(
                Artist.__table__.update()
                .where(Artist.id == artist_id)
                .values(is_subscription=False)
            )
        await self.session.commit()