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 sqlalchemy import func
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import PlatformMembership

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)