FabledCurator/backend/app/services/membership_roster.py

"""The learned membership roster: what the account actually subscribes to.

Milestone 387, phase C. Sibling of `service_roster` (milestone 365) and built
on the same insight — an absence is only observable against a record of
presence. There, a stopped worker; here, a subscription that lapsed.

## Nothing calls this yet

`touch_membership` is written before its caller because the caller (the sweep,
C3) needs a client seam (C2) that needs Patreon's real response characterised
from a captured sample (C0), and that capture needs the operator's browser
session. The write side does not depend on any of it: an upsert keyed on
(platform, external_campaign_id) is the same regardless of what the payload
turns out to look like, and `details` carries whatever C0 finds.

## Why the whitelist lives here and not in the column

`platform_membership.status` is an unconstrained String holding the PLATFORM's
own word — `active_patron`, not some normalised FC value. The mapping from
those words to FC's meaning is a read-site concern and belongs in code that can
be corrected without a migration, because the vocabulary comes from whatever
each platform says and will be discovered per platform rather than designed up
front. `MEMBERSHIP_STATUS` below is a place for that knowledge to accumulate as
platforms are characterised; it is deliberately empty of guesses today.
"""

from __future__ import annotations

import logging
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime, timedelta

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

from ..models import MembershipSync, PlatformMembership, Source

log = logging.getLogger(__name__)

# Platform word -> whether the account currently has paid access.
#
# Every entry here must come from a CHARACTERISED response, never from API docs
# or a plausible guess — project rule 130, and inventing a status before seeing
# it in a real payload is exactly the failure it names.
#
# patreon: from a live capture of the operator's own session, 2026-09-10
# (Scribe note #3886). Only two values were OBSERVED in `patron_status` and
# only those two are here.
#
# `declined_patron` is deliberately ABSENT even though it looks obviously
# right. It appears in the request's `filter[membership_type]`, and the capture
# proved that filter is NOT the same vocabulary as the attribute — a row
# selected by the filter as `free_member` came back with
# `patron_status: former_patron`, a word the filter does not contain. Reading
# the filter as an enum is the specific mistake the capture caught; adding
# `declined_patron` on the strength of it would be repeating that mistake one
# step later.
#
# Unknown words are NOT an error: an unrecognised status means the roster
# records evidence it cannot yet interpret, which is a better state than
# dropping the row or asserting a meaning for it.
MEMBERSHIP_STATUS: dict[str, dict[str, bool]] = {
    "patreon": {
        "active_patron": True,
        "former_patron": False,
    },
}


def has_paid_access(
    platform: str, status: str | None, *, is_free_member: bool = False,
) -> bool | None:
    """Does this membership mean the account currently PAYS for access?

    Returns None for a status this code has not been taught, which callers must
    treat as "unknown" rather than as False. The difference matters: False says
    the operator has lost access, and asserting that from an unrecognised word
    would tell them to cancel a source they are still paying for.

    `is_free_member` is a second axis, not a status, and that is Patreon's
    design rather than ours: the capture shows a free follow expressed as a
    boolean alongside `patron_status`, so a "current" membership can still be
    one nobody is paying for. Taking status alone would report a free follower
    as a paying patron, and C4 would then never offer to clean it up.

    (Honest limit: the capture contains no ACTIVE free member, so it cannot
    demonstrate the two axes coming apart. The separation is what the payload's
    shape says; the sample only shows it is possible, not that it happens.)
    """
    if status is None:
        return None
    known = MEMBERSHIP_STATUS.get(platform, {}).get(status)
    if known is None:
        return None
    if not known:
        return False
    return not is_free_member


async def touch_membership(
    session: AsyncSession,
    *,
    platform: str,
    external_campaign_id: str,
    display_name: str | None = None,
    url: str | None = None,
    status: str | None = None,
    tier_names: list | None = None,
    amount_cents: int | None = None,
    currency: str | None = None,
    details: dict | None = None,
) -> None:
    """Record that this membership was observed just now.

    Upsert rather than read-modify-write, for the same reason as
    `service_roster.touch_service`: a sweep may overlap its own previous run,
    and the last writer is simply the most recent sighting.

    `first_seen_at` is deliberately NOT in the update set. It is the one field
    that answers "has this ever been true", which is what makes a membership's
    later DISAPPEARANCE readable as a lapse rather than indistinguishable from
    a creator FC never knew about. Every other column is last-writer-wins,
    including status — a membership that goes from active to former must move.
    """
    stmt = pg_insert(PlatformMembership).values(
        platform=platform,
        external_campaign_id=external_campaign_id,
        display_name=display_name,
        url=url,
        status=status,
        tier_names=tier_names,
        amount_cents=amount_cents,
        currency=currency,
        details=details or {},
    )
    stmt = stmt.on_conflict_do_update(
        constraint="uq_platform_membership_platform_campaign",
        set_={
            "display_name": stmt.excluded.display_name,
            "url": stmt.excluded.url,
            "status": stmt.excluded.status,
            "tier_names": stmt.excluded.tier_names,
            "amount_cents": stmt.excluded.amount_cents,
            "currency": stmt.excluded.currency,
            "details": stmt.excluded.details,
            "last_seen_at": func.now(),
        },
    )
    await session.execute(stmt)


# ---------------------------------------------------------------------------
# The sweep, and the state that makes its failures readable (#387 C3)
# ---------------------------------------------------------------------------
#
# How long a successful sync stays trustworthy. Beyond this the roster is
# STALE, and C4 must refuse to draw conclusions from it — "you are tracking 12
# sources you no longer subscribe to", computed from a roster that stopped
# syncing a week ago, is an invitation to cancel things the operator is still
# paying for.
#
# Generous relative to the daily cadence: a few missed runs are a blip, not a
# reason to stop trusting a roster that changes on a billing cycle.
ROSTER_STALE_AFTER = timedelta(days=3)


async def get_sync_state(session: AsyncSession, platform: str) -> MembershipSync | None:
    return (await session.execute(
        select(MembershipSync).where(MembershipSync.platform == platform)
    )).scalar_one_or_none()


def roster_is_fresh(state: MembershipSync | None, *, now: datetime | None = None) -> bool:
    """May a caller draw CONCLUSIONS from this roster?

    False for never-synced and for stale, and those are deliberately the same
    answer here even though the UI must tell them apart: both mean the roster
    is not evidence. The asymmetry that matters is that `False` never means
    "you subscribe to nothing" — it means "we do not know", and a caller that
    cannot represent "we do not know" must not be asking this question.
    """
    if state is None or state.last_success_at is None:
        return False
    now = now or datetime.now(UTC)
    return (now - state.last_success_at) <= ROSTER_STALE_AFTER


async def _record_sync(session: AsyncSession, platform: str, **values) -> None:
    stmt = pg_insert(MembershipSync).values(platform=platform, **values)
    await session.execute(stmt.on_conflict_do_update(
        constraint="uq_membership_sync_platform",
        set_={**values, "updated_at": func.now()},
    ))


async def sync_platform(
    session: AsyncSession,
    *,
    platform: str,
    fetch: Callable[[], Awaitable[list]],
    now: datetime | None = None,
) -> dict:
    """Walk one platform's roster and record what happened.

    `fetch` is injected rather than built here so the error-to-state mapping —
    the part with the consequences — is testable without a credential, and so
    this service needs to know nothing about how any particular client is
    constructed.

    THE FETCH COMPLETES BEFORE ANYTHING IS WRITTEN. That ordering is the whole
    safety property: a walk that dies half way through pagination writes
    nothing, so a failure can never leave a roster that is partly this week's
    and partly last week's. (`touch_membership` never deletes, so a failure
    cannot empty the roster either — but "intact" should mean intact, not
    merely non-empty.)

    Returns a summary dict; never raises for a platform failure, because one
    platform failing must not abort the others.
    """
    now = now or datetime.now(UTC)
    await _record_sync(session, platform, last_attempt_at=now)
    await session.commit()

    try:
        memberships = await fetch()
    except Exception as exc:  # noqa: BLE001 - deliberately broad, see below
        # Broad on purpose: a sweep is a background job, and ANY escape here
        # kills the run for every other platform too. The exception's class
        # name is recorded so the distinction the client drew (auth vs drift
        # vs transport) survives into the UI, which is where it is actionable.
        #
        # EXCEPT the worker asking us to stop. Celery raises its soft time
        # limit as an ordinary Exception subclass, so a broad catch swallows
        # the shutdown request and lets the sweep run on into the HARD limit,
        # where it is SIGKILLed mid-transaction. A sweep that cannot be stopped
        # is worse than one that fails. (KeyboardInterrupt and SystemExit are
        # BaseException and pass through this clause already.)
        from celery.exceptions import SoftTimeLimitExceeded

        if isinstance(exc, SoftTimeLimitExceeded):
            raise
        await session.rollback()
        await _record_sync(
            session, platform,
            last_error_type=type(exc).__name__,
            last_error_message=str(exc)[:2000],
        )
        await session.commit()
        log.warning("membership sync failed for %s: %s", platform, exc)
        return {"platform": platform, "ok": False, "error": type(exc).__name__}

    for m in memberships:
        await touch_membership(
            session,
            platform=platform,
            external_campaign_id=m.campaign_id,
            display_name=m.display_name,
            url=m.url,
            status=m.status,
            tier_names=m.tier_names or None,
            amount_cents=m.amount_cents,
            currency=m.currency,
            details={**(m.details or {}), "is_free_member": m.is_free_member},
        )
    await _record_sync(
        session, platform,
        last_success_at=now,
        last_count=len(memberships),
        # Cleared on success — a stale error beside a fresh success would read
        # as "still broken" forever.
        last_error_type=None,
        last_error_message=None,
    )
    await session.commit()
    log.info("membership sync ok for %s: %d membership(s)", platform, len(memberships))
    return {"platform": platform, "ok": True, "count": len(memberships)}


# ---------------------------------------------------------------------------
# Membership <-> Source identity (#387 C4)
# ---------------------------------------------------------------------------
#
# "Is this membership already tracked?" is asked by TWO features — C4's
# reconciliation buckets and E4's creator suggestions — and it lives here, once,
# on purpose. Built inline in C4 it would have looked finished while leaving E4
# matching on name similarity alone, so the two would answer the same question
# differently and only one of them would be right.
#
# E4 and C4 use it from opposite sides: E4 as the NEGATIVE check (propose only
# where nothing matches) and C4 as the join itself.

# Any platform that caches its creator id does so under this suffix; see
# `download_service._phase3_persist`, which writes `patreon_campaign_id`.
_CAMPAIGN_KEY_SUFFIX = "_campaign_id"


def identity_keys_for_source(source: Source) -> set[str]:
    """Every platform-side creator id cached on this source.

    Reads ANY `<platform>_campaign_id` override rather than naming Patreon's,
    so a second platform participates by caching its id under the same suffix —
    no registry, no `if platform ==` branch (rule 169). A source that has never
    been walked has cached nothing and simply contributes no exact key, which is
    what makes the handle fallback below necessary rather than merely tolerated.
    """
    keys = set()
    for name, value in (source.config_overrides or {}).items():
        if name.endswith(_CAMPAIGN_KEY_SUFFIX) and isinstance(value, str) and value:
            keys.add(value)
    return keys


def url_tail(url: str | None) -> str | None:
    """The creator handle at the end of a source URL, lowercased.

    Deliberately the same derivation as `PlatformMembership.vanity_or_none`'s
    own fallback, so both sides of the comparison reduce a URL to a handle the
    same way. Query strings and fragments are stripped first; Patreon's `/c/`
    and `/cw/` forms both end in the vanity, so they need no special case (the
    missing-`/c/` regex is what broke creator detection in #1485).

    Returns None for the pre-0030 `sidecar:` synthetic anchors, which are not
    feeds and must never match anything.
    """
    if not url or url.startswith("sidecar:"):
        return None
    cleaned = url.split("?", 1)[0].split("#", 1)[0]
    tail = cleaned.rstrip("/").rsplit("/", 1)[-1]
    return tail.lower() or None


def match_kind(source: Source, membership: PlatformMembership) -> str | None:
    """How this source and this membership are known to be the same creator.

    Returns "campaign" for an exact platform-id match, "vanity" for agreeing URL
    handles, or None for no evidence.

    THE ORDER MUST NOT BE INVERTED. The campaign id is exact and the handle is
    not, but the id is only written AFTER a source has been walked at least once
    — so checking the handle first would let a stale or renamed URL outvote the
    authoritative id on every source FC has actually polled.
    """
    if source.platform != membership.platform:
        return None
    if membership.external_campaign_id in identity_keys_for_source(source):
        return "campaign"
    vanity = membership.vanity_or_none()
    tail = url_tail(source.url)
    if vanity and tail and vanity.strip().lower() == tail:
        return "vanity"
    return None


async def source_for_membership(
    session: AsyncSession, membership: PlatformMembership,
) -> Source | None:
    """The source FC already tracks for this membership, if there is one.

    Scoped to the membership's own platform, so a creator tracked on Discord and
    subscribed to on Patreon does not read as already-tracked — that pairing is
    E4's suggestion to make, not an identity.
    """
    rows = (await session.execute(
        select(Source).where(Source.platform == membership.platform)
    )).scalars().all()
    for source in rows:
        if match_kind(source, membership):
            return source
    return None