FabledCurator/backend/app/services/membership_reconcile.py

"""Reconciling the learned roster against the sources FC actually tracks.

Milestone 387, step C4. The step the operator asked for; C0-C3 are what make it
trustworthy enough to act on.

## The buckets

1. `subscribed_not_tracked` — you pay for this and FC does not follow it. The
   adoption win, and the only bucket carrying an action.
2. `tracked_not_subscribed` — FC follows this and the roster does not show you
   paying for it. REPORT ONLY, by the operator's decision (2026-09-11): it says
   what it sees and links to the existing Subscriptions row, and offers no
   one-click disable.
3. `matched` — the healthy set. Counted, not listed loudly.
4. `unidentified` — sources this join cannot speak to at all. Reported as
   exactly that, because the alternative is filing them under a verdict.

## Why absence is the dangerous direction

Bucket 1 is safe to be wrong about: the cost of offering a source the operator
does not want is one ignored row. Bucket 2 is not. It is computed from an
ABSENCE — no membership matched — and three different things produce that
absence: the subscription genuinely lapsed, the sweep failed, or the creator
renamed and this source has never been walked so no exact id was ever cached.

Two guards follow from that, and they are the substance of this module:

* the whole bucket is gated on `roster_is_fresh`, so a failed or never-run sweep
  yields an empty list rather than a confident accusation (C3 built the state
  this reads);
* every row carries the BASIS for its claim, so "your membership says former
  patron" and "we know this creator's id and it is not in your roster" and "we
  only have a URL handle to go on" are three different sentences rather than one
  overconfident one.

`has_paid_access` returning None is honoured throughout: unknown is never
rendered as lapsed. That is the whole reason it returns a tri-state.
"""

from __future__ import annotations

from datetime import datetime

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import Artist, MembershipSync, PlatformMembership, Source
from .membership_roster import (
    get_sync_state,
    identity_keys_for_source,
    pair_sources_with_memberships,
    roster_is_fresh,
    url_tail,
)
from .native_ingest_common import has_paid_access

# Why a source appears in `tracked_not_subscribed`. Ordered strongest first —
# the UI renders a different sentence per basis, because collapsing them into
# one would make the weakest claim sound like the strongest.
BASIS_LAPSED = "lapsed"           # a matched membership says access ended
BASIS_ABSENT_EXACT = "absent_exact"    # exact id known, not in a fresh roster
BASIS_ABSENT_HANDLE = "absent_handle"  # only a URL handle to go on


def _membership_row(m: PlatformMembership) -> dict:
    return {
        "id": m.id,
        "platform": m.platform,
        "external_campaign_id": m.external_campaign_id,
        "display_name": m.display_name or m.vanity_or_none(),
        "url": m.url,
        "vanity": m.vanity_or_none(),
        "status": m.status,
        "tier_names": m.tier_names,
        "amount_cents": m.amount_cents,
        "currency": m.currency,
        "paid_access": has_paid_access(
            m.platform, m.status,
            is_free_member=bool((m.details or {}).get("is_free_member")),
        ),
    }


def _source_row(source: Source, artist: Artist) -> dict:
    return {
        "id": source.id,
        "platform": source.platform,
        "url": source.url,
        "enabled": source.enabled,
        "artist": {"id": artist.id, "name": artist.name, "slug": artist.slug},
    }


async def reconcile(
    session: AsyncSession, *, platform: str, now: datetime | None = None,
) -> dict:
    """Sort one platform's memberships and sources into the four buckets.

    Always returns the COMPLETE shape, including when the roster is not fresh —
    a caller reading `len(result["tracked_not_subscribed"])` must not have to
    check which keys exist first. `fresh` is what says whether the emptiness
    means anything.
    """
    state = await get_sync_state(session, platform)
    fresh = roster_is_fresh(state, now=now)

    memberships = (await session.execute(
        select(PlatformMembership).where(PlatformMembership.platform == platform)
    )).scalars().all()
    rows = (await session.execute(
        select(Source, Artist)
        .join(Artist, Artist.id == Source.artist_id)
        .where(Source.platform == platform)
    )).all()

    # The join itself lives in `membership_roster` beside `match_kind`, so C5's
    # gated-reason annotation pairs sources with memberships by exactly the same
    # rule this card sorts them by. Two copies would let the Subscriptions row
    # and this card disagree about which creator a source IS.
    pairs = pair_sources_with_memberships([s for s, _a in rows], memberships)
    matched_membership_ids = {m.id for m, _kind in pairs.values()}

    subscribed_not_tracked = []
    for m in memberships:
        if m.id in matched_membership_ids:
            continue
        paid = has_paid_access(
            m.platform, m.status,
            is_free_member=bool((m.details or {}).get("is_free_member")),
        )
        # A membership FC knows has ENDED is not an adoption opportunity —
        # adding it would start a walk that can only fetch what is already
        # public. Unknown (None) is still offered: the operator can judge it,
        # and refusing to show it would hide a real subscription behind a word
        # this code has not been taught.
        if paid is False:
            continue
        subscribed_not_tracked.append(_membership_row(m))

    tracked_not_subscribed = []
    matched = []
    unidentified = []
    for source, artist in rows:
        pair = pairs.get(source.id)
        if pair is not None:
            m, kind = pair
            paid = has_paid_access(
                m.platform, m.status,
                is_free_member=bool((m.details or {}).get("is_free_member")),
            )
            if paid is False:
                if not source.enabled:
                    # Already off. Reporting a source the operator has already
                    # stopped following is noise, not a finding.
                    continue
                row = _source_row(source, artist)
                row["basis"] = BASIS_LAPSED
                row["matched_by"] = kind
                row["membership"] = _membership_row(m)
                tracked_not_subscribed.append(row)
            else:
                row = _source_row(source, artist)
                row["matched_by"] = kind
                row["membership"] = _membership_row(m)
                matched.append(row)
            continue

        # No membership matched. Whether that MEANS anything depends entirely on
        # how well this source can be identified at all.
        has_exact = bool(identity_keys_for_source(source))
        if not has_exact and url_tail(source.url) is None:
            # Nothing to match on — a sidecar anchor or a URL with no handle.
            # Reported as unidentified rather than silently dropped, so the
            # counts add up to the source list the operator can see.
            unidentified.append(_source_row(source, artist))
            continue
        if not source.enabled:
            # Already off. Telling the operator to stop following something they
            # have stopped following is noise, not a finding.
            continue
        row = _source_row(source, artist)
        row["basis"] = BASIS_ABSENT_EXACT if has_exact else BASIS_ABSENT_HANDLE
        row["matched_by"] = None
        row["membership"] = None
        tracked_not_subscribed.append(row)

    # THE GATE. Everything above computed the bucket; this decides whether it may
    # be shown. A stale or never-run roster makes every absence meaningless, and
    # an absence rendered as a verdict is how this feature would tell the
    # operator to cancel something they are still paying for.
    if not fresh:
        tracked_not_subscribed = []

    return {
        "platform": platform,
        "fresh": fresh,
        # How many sources exist on this platform at all. The UI needs it to
        # decide whether an untrustworthy roster is worth mentioning: with no
        # sources here there is nothing to reconcile, and a stale-roster warning
        # would be noise on an install that simply has not started yet (that
        # empty-install case is C6's, not this card's).
        "tracked_total": len(rows),
        "last_success_at": (
            state.last_success_at.isoformat()
            if state is not None and state.last_success_at else None
        ),
        "subscribed_not_tracked": subscribed_not_tracked,
        "tracked_not_subscribed": tracked_not_subscribed,
        "matched": matched,
        "unidentified": unidentified,
    }


async def reconcile_all(session: AsyncSession, now: datetime | None = None) -> dict:
    """Every platform the roster knows about, in one payload for the UI.

    The platform list is the UNION of platforms with memberships and platforms
    with sync state, not just the former. A sweep that has never succeeded has
    recorded zero memberships, and deriving the list from memberships alone
    would drop exactly that platform from the payload — making a broken
    credential indistinguishable from a platform FC was never asked about. That
    distinction is the whole reason C3 records sync state.
    """
    with_memberships = (await session.execute(
        select(PlatformMembership.platform).distinct()
    )).scalars().all()
    with_state = (await session.execute(
        select(MembershipSync.platform)
    )).scalars().all()
    platforms = set(with_memberships) | set(with_state)
    return {
        "platforms": [
            await reconcile(session, platform=p, now=now) for p in sorted(platforms)
        ]
    }