diff --git a/alembic/versions/0095_membership_sync.py b/alembic/versions/0095_membership_sync.py new file mode 100644 index 0000000..f180b01 --- /dev/null +++ b/alembic/versions/0095_membership_sync.py @@ -0,0 +1,63 @@ +"""membership_sync — whether the roster actually synced, and when. + +Milestone 387, step C3. + +`platform_membership` (0091) records what was SEEN. This records whether +looking happened at all — a different fact, and the one that makes an empty +roster readable. + +Without it, three situations collapse into one: the account subscribes to +nothing, the sweep never ran, or the sweep failed. All three leave zero rows +in `platform_membership`. "You are tracking 12 sources you no longer subscribe +to" is correct in the first case and an invitation to cancel things the +operator is actively paying for in the other two, which is why C4 gates its +CONCLUSIONS on `last_success_at` rather than merely displaying it. + +Two timestamps rather than one, deliberately: `last_attempt_at` moves every +run, `last_success_at` only on a clean walk, and the gap between them is what +lets the UI say "last synced 3 days ago, tried 20 minutes ago, failing". + +Revision ID: 0095 +Revises: 0094 +Create Date: 2026-09-11 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0095" +down_revision: Union[str, None] = "0094" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "membership_sync", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("platform", sa.String(length=64), nullable=False), + sa.Column("last_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_success_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_count", sa.Integer(), nullable=True), + # No CHECK: this carries an exception class name, and the vocabulary is + # whatever the client raises — same call as source.error_type. + sa.Column("last_error_type", sa.String(length=64), nullable=True), + sa.Column("last_error_message", sa.Text(), nullable=True), + sa.Column( + "updated_at", sa.DateTime(timezone=True), + server_default=sa.text("now()"), nullable=False, + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_membership_sync")), + # The upsert's conflict target, named explicitly because the service + # references it by name in ON CONFLICT. + sa.UniqueConstraint("platform", name="uq_membership_sync_platform"), + ) + # No secondary indexes: one row per platform, so every read is a short scan + # and an index would be write cost buying nothing (#3301 removed seven of + # that shape). Same reasoning as platform_membership in 0091. + + +def downgrade() -> None: + op.drop_table("membership_sync") diff --git a/backend/app/api/sources.py b/backend/app/api/sources.py index 2463ea2..2832095 100644 --- a/backend/app/api/sources.py +++ b/backend/app/api/sources.py @@ -1,10 +1,11 @@ """FC-3a: CRUD over Source rows. FC-3c adds POST //check.""" from quart import Blueprint, jsonify, request -from sqlalchemy import select +from sqlalchemy import func, select from ..extensions import get_session -from ..models import DownloadEvent, Source +from ..models import DownloadEvent, MembershipSync, PlatformMembership, Source +from ..services.membership_roster import roster_is_fresh from ..services.scheduler_service import active_platform_cooldowns, scheduler_status from ..services.source_service import ( KNOWN_PLATFORMS, @@ -288,3 +289,57 @@ async def check_source(source_id: int): download_source.delay(source_id) return jsonify({"download_event_id": event_id, "status": "pending"}), 202 + + +# --- #387 C3: the membership roster's sync state -------------------------- +# +# Rule 164's visibility requirement lives here. A roster that failed to sync, +# or never has, must be DISTINGUISHABLE from an account that subscribes to +# nothing — otherwise the reconciliation this unlocks would tell the operator +# to cancel sources they are actively paying for. + + +@sources_bp.route("/membership-sync", methods=["GET"]) +async def membership_sync_status(): + async with get_session() as session: + rows = (await session.execute(select(MembershipSync))).scalars().all() + counts = dict( + (await session.execute( + select(PlatformMembership.platform, func.count()) + .group_by(PlatformMembership.platform) + )).all() + ) + return jsonify({"platforms": [ + { + "platform": r.platform, + "last_attempt_at": r.last_attempt_at.isoformat() if r.last_attempt_at else None, + # NULL here means NEVER, and the UI must say so in words. Rendering + # it as 0 or as "-" is the exact conflation this endpoint exists to + # prevent. + "last_success_at": r.last_success_at.isoformat() if r.last_success_at else None, + "last_count": r.last_count, + "last_error_type": r.last_error_type, + "last_error_message": r.last_error_message, + # Whether a CONCLUSION may be drawn from this roster — not merely + # whether it looks recent. C4 gates on this, and it is computed + # server-side so no caller can forget to. + "fresh": roster_is_fresh(r), + "known_memberships": counts.get(r.platform, 0), + } + for r in sorted(rows, key=lambda r: r.platform) + ]}) + + +@sources_bp.route("/membership-sync", methods=["POST"]) +async def trigger_membership_sync(): + """Run the roster sweep now. + + The beat schedule runs daily, which is right for a billing-cycle fact but + far too slow when the operator has just connected a credential and wants to + see whether it works. Queued rather than run inline: it crosses the network + to an external service and the request path is not where that belongs. + """ + from ..tasks.maintenance import sync_memberships + + sync_memberships.delay() + return jsonify({"queued": True}) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index a2c5e8b..b702fc6 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -214,6 +214,12 @@ def make_celery() -> Celery: # points at exists as a grouping (#388 E5). No-op unless # discord_link_enabled. }, + "sync-memberships-daily": { + "task": "backend.app.tasks.maintenance.sync_memberships", + "schedule": 86400.0, # daily — memberships change on a BILLING + # cycle, not a download cadence (#387 C3). No-op per platform + # when the client lacks the seam or no credential exists. + }, "integrity-verify-weekly": { "task": "backend.app.tasks.maintenance.verify_integrity", "schedule": 604800.0, # weekly diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 1d7dfa2..ae31627 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -21,6 +21,7 @@ from .import_batch import ImportBatch from .import_settings import ImportSettings from .import_task import ImportTask from .library_audit_run import LibraryAuditRun +from .membership_sync import MembershipSync from .ml_settings import MLSettings from .patreon_failed_media import PatreonFailedMedia from .patreon_seen_media import PatreonSeenMedia @@ -82,6 +83,7 @@ __all__ = [ "ImportTask", "ImportSettings", "LibraryAuditRun", + "MembershipSync", "MLSettings", "HeadAutoApplyRun", "HeadMetric", diff --git a/backend/app/models/membership_sync.py b/backend/app/models/membership_sync.py new file mode 100644 index 0000000..d624418 --- /dev/null +++ b/backend/app/models/membership_sync.py @@ -0,0 +1,77 @@ +"""membership_sync — did the roster actually sync, and when. + +Milestone 387, step C3. + +`platform_membership` records what was SEEN. This records whether looking +happened at all, and that is a different fact — the one that makes an empty +roster readable. + +## Why this table has to exist + +Without it, three very different situations are one indistinguishable state: + +* the account genuinely subscribes to nothing, +* the sweep has never run, +* the sweep ran and failed. + +All three produce zero rows in `platform_membership`. Telling the operator +"you are tracking 12 sources you do not subscribe to" is correct in the first +case and catastrophic in the other two — it is an invitation to cancel things +they are actively paying for. C4 must therefore gate its CONCLUSIONS on +`last_success_at`, not merely display it. + +`MAX(platform_membership.last_seen_at)` was the tempting shortcut and does not +work: it cannot distinguish "synced fine, found nothing" from "never synced". +`task_run` was the other candidate and is worse — its retention prunes ok rows +after 24h, so a sweep that last succeeded three days ago would leave no trace +at all. + +## Separate attempt and success timestamps, deliberately + +`last_attempt_at` moves every run; `last_success_at` moves only on a clean +walk. The GAP between them is the staleness signal, and keeping them apart is +what lets the UI say "last synced 3 days ago, last tried 20 minutes ago, +failing" — which is a different message from either half alone. +""" + +from datetime import datetime + +from sqlalchemy import DateTime, Integer, String, Text, UniqueConstraint, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class MembershipSync(Base): + __tablename__ = "membership_sync" + __table_args__ = ( + UniqueConstraint("platform", name="uq_membership_sync_platform"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + platform: Mapped[str] = mapped_column(String(64), nullable=False) + + # Moves on EVERY run, success or not — so "we are trying" is visible even + # while "we are succeeding" is not. + last_attempt_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + # Moves only on a COMPLETE walk. This is the freshness signal C4 gates its + # conclusions on; NULL means never — which must never be rendered as zero. + last_success_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + # How many memberships the last SUCCESSFUL walk saw. Paired with + # last_success_at so "0" is only ever readable as a real zero. + last_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + + # Cleared on success. Plain String, no CHECK — this carries an exception + # class name (PatreonAuthError, PatreonDriftError, ...) and the vocabulary + # is whatever the client raises, exactly as source.error_type works. + last_error_type: Mapped[str | None] = mapped_column(String(64), nullable=True) + last_error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, + server_default=func.now(), onupdate=func.now(), + ) diff --git a/backend/app/services/membership_roster.py b/backend/app/services/membership_roster.py index 53dce25..417aa36 100644 --- a/backend/app/services/membership_roster.py +++ b/backend/app/services/membership_roster.py @@ -28,11 +28,14 @@ from __future__ import annotations import logging -from sqlalchemy import func +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 PlatformMembership +from ..models import MembershipSync, PlatformMembership log = logging.getLogger(__name__) @@ -146,3 +149,130 @@ async def touch_membership( }, ) 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)} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 18ea9c4..721cc92 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1204,3 +1204,111 @@ def match_post_associations() -> str: if not res["enabled"]: return "disabled" return f"scanned={res['scanned']} proposed={res['proposed']}" + + +# The wall-clock budget for ONE platform's roster walk. Rule 156 distinguishes +# this from the per-REQUEST timeout the client already has: a paginated roster +# behind an endpoint that answers every page slowly-but-within-timeout would +# never trip that one, and would sit on a worker indefinitely. This is the wait +# that bounds the whole walk. +MEMBERSHIP_SYNC_BUDGET_SECONDS = 240.0 + + +@celery.task( + name="backend.app.tasks.maintenance.sync_memberships", + soft_time_limit=900, time_limit=1200, +) +def sync_memberships() -> str: + """Milestone 387 C3: walk each platform's membership roster into the DB. + + Daily, because memberships change on a BILLING cycle rather than a download + cadence — polling an account-scoped endpoint more often would be both + pointless and less polite than the browser. + + Rule 89's four, and where each actually lives: + * recovery — `touch_membership` is an upsert and never deletes, so + recovery is simply the next run; a dead run leaves the previous roster + intact rather than a half-written one (`sync_platform` completes the + fetch before writing anything). + * retention — per C1, rows age out and are never deleted on + disappearance, because disappearing IS the signal C4 reads. + * wall-clock timeout — MEMBERSHIP_SYNC_BUDGET_SECONDS per platform, + plus the task's own soft/hard limits. + * duration tracking — the TaskRun celery-signal plumbing, same as every + other sweep here. + + Rule 164: this is the rule's own named exception — a genuinely external + feature that may call out. It never gates startup, and a failure leaves a + VISIBLE stale state (membership_sync) rather than an empty roster that + reads as "you subscribe to nothing". + """ + import asyncio + + from ..services.credential_crypto import CredentialCrypto + from ..services.credential_service import CredentialService + from ..services.membership_roster import sync_platform + from ..services.patreon_client import PatreonClient + from ._async_session import async_session_factory + + key_path = IMAGES_ROOT / "secrets" / "credential_key.b64" + + # platform -> how to build a client from a cookies path. A platform is in + # the sweep only if it is here AND its client exposes `iter_memberships` + # AND a credential exists — three independent gates, each silent, so + # adding SubscribeStar (D1) is one line here and nothing else. + builders = {"patreon": PatreonClient} + + async def _run() -> dict: + async_factory, engine = async_session_factory() + results = [] + try: + for platform, build in builders.items(): + async with async_factory() as session: + cred = CredentialService(session, CredentialCrypto(key_path)) + cookies = await cred.get_cookies_path(platform) + if cookies is None: + # No credential is not an error — the operator simply has + # not connected this platform. Recording a failure here + # would light up the UI for a feature they never enabled. + results.append({"platform": platform, "skipped": "no credential"}) + continue + + client = build(str(cookies)) + if getattr(client, "iter_memberships", None) is None: + # The seam, probed not required (#387 C2). A client without + # it makes the feature invisible for that platform — no + # flag, no config row, no "unsupported" branch. + results.append({"platform": platform, "skipped": "no seam"}) + continue + + async def fetch(_client=client): + # The client is sync (`requests`); run it off the loop so a + # slow roster does not block the event loop, and bound the + # whole walk rather than only its individual requests. + def _walk(): + user_id = _client.current_user_id() + return list(_client.iter_memberships(user_id)) + + return await asyncio.wait_for( + asyncio.to_thread(_walk), + timeout=MEMBERSHIP_SYNC_BUDGET_SECONDS, + ) + + async with async_factory() as session: + results.append( + await sync_platform(session, platform=platform, fetch=fetch) + ) + return {"results": results} + finally: + await engine.dispose() + + res = asyncio.run(_run()) + parts = [] + for r in res["results"]: + if "skipped" in r: + parts.append(f"{r['platform']}=skipped({r['skipped']})") + elif r.get("ok"): + parts.append(f"{r['platform']}={r['count']}") + else: + parts.append(f"{r['platform']}=FAILED({r['error']})") + return " ".join(parts) or "no platforms" diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 3e10a34..05c5827 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -16,6 +16,7 @@ + @@ -83,6 +84,7 @@ import VideoEmbeddingCard from './VideoEmbeddingCard.vue' import CropProposersCard from './CropProposersCard.vue' import HeadsCard from './HeadsCard.vue' import DiscordGroupingCard from './DiscordGroupingCard.vue' +import MembershipRosterCard from './MembershipRosterCard.vue' import PostAssociationsCard from './PostAssociationsCard.vue' import GpuAgentCard from './GpuAgentCard.vue' import AliasTable from './AliasTable.vue' diff --git a/frontend/src/components/settings/MembershipRosterCard.vue b/frontend/src/components/settings/MembershipRosterCard.vue new file mode 100644 index 0000000..96b8816 --- /dev/null +++ b/frontend/src/components/settings/MembershipRosterCard.vue @@ -0,0 +1,103 @@ + + + + + diff --git a/frontend/src/stores/membershipSync.js b/frontend/src/stores/membershipSync.js new file mode 100644 index 0000000..de1a0fc --- /dev/null +++ b/frontend/src/stores/membershipSync.js @@ -0,0 +1,38 @@ +import { defineStore } from 'pinia' +import { ref } from 'vue' + +import { useApi } from '../composables/useApi.js' +import { useAsyncAction } from '../composables/useAsyncAction.js' +import { toast } from '../utils/toast.js' + +// Backs the subscription-roster card (#387 C3). The whole reason this surface +// exists: a roster that never synced, or failed to, must be DISTINGUISHABLE +// from an account that subscribes to nothing. All three states look like zero +// rows, and only one of them means "you are tracking things you do not pay +// for" — so the UI must never render "never synced" as a number. +export const useMembershipSyncStore = defineStore('membershipSync', () => { + const api = useApi() + const platforms = ref([]) + const { loading, error, run } = useAsyncAction({ errorAs: 'message' }) + + async function load () { + await run(async () => { + const body = await api.get('/api/sources/membership-sync') + platforms.value = body.platforms || [] + }) + } + + async function syncNow () { + try { + await api.post('/api/sources/membership-sync', {}) + toast({ + text: 'Roster sync queued — runs in the background; reload to see the result.', + type: 'info' + }) + } catch (e) { + toast({ text: `Sync failed to queue: ${e.message}`, type: 'error' }) + } + } + + return { platforms, loading, error, load, syncNow } +}) diff --git a/frontend/test/components/membershipRosterCard.spec.js b/frontend/test/components/membershipRosterCard.spec.js new file mode 100644 index 0000000..bfb7b84 --- /dev/null +++ b/frontend/test/components/membershipRosterCard.spec.js @@ -0,0 +1,77 @@ +// @vitest-environment happy-dom +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' + +import MembershipRosterCard from '../../src/components/settings/MembershipRosterCard.vue' +import { useMembershipSyncStore } from '../../src/stores/membershipSync.js' +import { mountWithStore } from '../support/mountComponent.js' + +// #387 C3. The one thing this card must never do is render "never synced" as a +// number. Three situations produce zero memberships — the account subscribes to +// nothing, the sweep never ran, the sweep failed — and only the first means +// "you are tracking sources you do not pay for". Conflating them is how C4 ends +// up telling the operator to cancel things they are actively paying for. + +const mountWith = (platforms) => mountWithStore(MembershipRosterCard, () => { + useMembershipSyncStore().platforms = platforms +}) + +describe('MembershipRosterCard', () => { + beforeEach(() => { + // The card loads on mount; stub it so assertions are about seeded state. + globalThis.fetch = vi.fn(async () => { throw new Error('offline') }) + }) + afterEach(() => vi.restoreAllMocks()) + + it('says "never synced" in words, and states no count', () => { + const w = mountWith([ + { platform: 'patreon', last_success_at: null, last_attempt_at: null, + last_count: null, fresh: false }, + ]) + expect(w.text()).toContain('never synced') + // The count line must be absent entirely — not "0 memberships found". + expect(w.text()).not.toContain('memberships found') + }) + + it('does not claim a count when a sync has been tried but never succeeded', () => { + const w = mountWith([ + { platform: 'patreon', last_success_at: null, + last_attempt_at: new Date(Date.now() - 3600e3).toISOString(), + last_count: null, fresh: false, last_error_type: 'PatreonAuthError', + last_error_message: 'cookies expired' }, + ]) + expect(w.text()).toContain('no successful sync yet') + expect(w.text()).not.toContain('memberships found') + // The error is surfaced, because "rotate your credential" is actionable. + expect(w.text()).toContain('PatreonAuthError') + }) + + it('marks a stale roster as too old to rely on', () => { + const w = mountWith([ + { platform: 'patreon', + last_success_at: new Date(Date.now() - 9 * 86400e3).toISOString(), + last_count: 12, fresh: false }, + ]) + expect(w.text()).toContain('too old to rely on') + }) + + it('states the count only when a successful sync stands behind it', () => { + const w = mountWith([ + { platform: 'patreon', + last_success_at: new Date(Date.now() - 3600e3).toISOString(), + last_count: 12, fresh: true }, + ]) + expect(w.text()).toContain('12 memberships found') + expect(w.text()).not.toContain('never synced') + expect(w.text()).not.toContain('too old') + }) + + it('a real zero is reported as zero, because that one IS an answer', () => { + const w = mountWith([ + { platform: 'patreon', + last_success_at: new Date(Date.now() - 3600e3).toISOString(), + last_count: 0, fresh: true }, + ]) + expect(w.text()).toContain('0 memberships found') + expect(w.text()).not.toContain('never synced') + }) +}) diff --git a/tests/test_membership_sync.py b/tests/test_membership_sync.py new file mode 100644 index 0000000..7d3d340 --- /dev/null +++ b/tests/test_membership_sync.py @@ -0,0 +1,281 @@ +"""Milestone 387 C3: the roster sweep, and the state that makes it readable. + +The failure this step must not have is an empty roster that looks like an +answer. Three situations produce zero rows in `platform_membership` — the +account subscribes to nothing, the sweep never ran, the sweep failed — and +telling the operator "you are tracking 12 sources you no longer subscribe to" +is correct in the first and catastrophic in the other two. So most of what +follows pins the difference between "we know" and "we do not know". +""" +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy import select + +from backend.app.models import MembershipSync, PlatformMembership +from backend.app.services.membership_roster import ( + ROSTER_STALE_AFTER, + get_sync_state, + roster_is_fresh, + sync_platform, +) + +pytestmark = pytest.mark.integration + + +@dataclass +class FakeMembership: + """Stands in for patreon_client.Membership — the sweep only needs the + attribute names, and depending on the real class here would couple this + test to a client it is not testing.""" + + campaign_id: str + display_name: str = "Creator" + url: str = "https://www.patreon.com/creator" + status: str = "active_patron" + is_free_member: bool = False + tier_names: list = None + amount_cents: int | None = 500 + currency: str = "USD" + details: dict = None + + +def _fetch(items): + async def fetch(): + return items + return fetch + + +def _raises(exc): + async def fetch(): + raise exc + return fetch + + +# --- freshness: the question C4 actually asks ------------------------------ + + +def test_a_never_synced_roster_is_not_fresh(): + """NULL last_success_at means NEVER, and never is not zero.""" + assert roster_is_fresh(None) is False + assert roster_is_fresh(MembershipSync(platform="patreon")) is False + + +def test_a_stale_roster_is_not_fresh(): + old = MembershipSync( + platform="patreon", + last_success_at=datetime.now(UTC) - ROSTER_STALE_AFTER - timedelta(hours=1), + ) + assert roster_is_fresh(old) is False + + +def test_a_recent_roster_is_fresh(): + recent = MembershipSync( + platform="patreon", last_success_at=datetime.now(UTC) - timedelta(hours=1), + ) + assert roster_is_fresh(recent) is True + + +def test_a_failing_sync_goes_stale_even_though_it_keeps_trying(): + """The gap between attempt and success IS the signal. A sweep hammering a + broken credential every day must not look healthy because it ran recently. + """ + state = MembershipSync( + platform="patreon", + last_attempt_at=datetime.now(UTC), + last_success_at=datetime.now(UTC) - ROSTER_STALE_AFTER - timedelta(days=1), + last_error_type="PatreonAuthError", + ) + assert roster_is_fresh(state) is False + + +# --- the sweep ------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_a_successful_sweep_writes_the_roster_and_records_success(db): + result = await sync_platform( + db, platform="patreon", + fetch=_fetch([FakeMembership("111"), FakeMembership("222")]), + ) + assert result == {"platform": "patreon", "ok": True, "count": 2} + + rows = (await db.execute(select(PlatformMembership))).scalars().all() + assert {r.external_campaign_id for r in rows} == {"111", "222"} + + state = await get_sync_state(db, "patreon") + assert state.last_success_at is not None + assert state.last_attempt_at is not None + assert state.last_count == 2 + assert state.last_error_type is None + + +@pytest.mark.asyncio +async def test_a_genuinely_empty_roster_is_a_success_not_a_silence(db): + """Zero memberships with a RECENT success is the one case where zero is an + answer — and it has to be distinguishable from the other two.""" + result = await sync_platform(db, platform="patreon", fetch=_fetch([])) + assert result["ok"] is True + state = await get_sync_state(db, "patreon") + assert state.last_success_at is not None + assert state.last_count == 0 + assert roster_is_fresh(state) is True + + +@pytest.mark.asyncio +async def test_a_failed_sweep_leaves_the_previous_roster_intact(db): + """THE property. An empty roster written over a good one is the worst + outcome available here — it reads downstream as 'cancel everything'.""" + await sync_platform( + db, platform="patreon", fetch=_fetch([FakeMembership("111")]), + ) + before = (await db.execute(select(PlatformMembership))).scalars().all() + assert len(before) == 1 + + result = await sync_platform( + db, platform="patreon", fetch=_raises(RuntimeError("boom")), + ) + assert result["ok"] is False + + after = (await db.execute(select(PlatformMembership))).scalars().all() + assert len(after) == 1, "a failure must not remove anything" + assert after[0].external_campaign_id == "111" + + +@pytest.mark.asyncio +async def test_a_failure_records_the_error_and_does_not_advance_success(db): + await sync_platform(db, platform="patreon", fetch=_fetch([FakeMembership("111")])) + db.expunge_all() + first = await get_sync_state(db, "patreon") + success_at = first.last_success_at + + await sync_platform( + db, platform="patreon", fetch=_raises(ValueError("drifted")), + ) + db.expunge_all() + state = await get_sync_state(db, "patreon") + assert state.last_error_type == "ValueError" + assert "drifted" in state.last_error_message + assert state.last_success_at == success_at, "a failure must not look like a sync" + assert state.last_attempt_at >= success_at, "but the ATTEMPT must be recorded" + + +@pytest.mark.asyncio +async def test_a_later_success_clears_the_previous_error(db): + """A stale error beside a fresh success would read as 'still broken'.""" + await sync_platform(db, platform="patreon", fetch=_raises(RuntimeError("x"))) + db.expunge_all() + assert (await get_sync_state(db, "patreon")).last_error_type == "RuntimeError" + + await sync_platform(db, platform="patreon", fetch=_fetch([FakeMembership("1")])) + db.expunge_all() + state = await get_sync_state(db, "patreon") + assert state.last_error_type is None + assert state.last_error_message is None + + +@pytest.mark.asyncio +async def test_the_sweep_never_raises_so_one_platform_cannot_abort_the_others(db): + """A sweep is a background job: any escape kills the run for every OTHER + platform too, so an ordinary failure is caught however exotic.""" + class WeirdError(Exception): + pass + + result = await sync_platform( + db, platform="patreon", fetch=_raises(WeirdError("unexpected")), + ) + assert result["ok"] is False + assert result["error"] == "WeirdError" + + +@pytest.mark.asyncio +async def test_a_base_exception_is_not_swallowed(db): + """KeyboardInterrupt/SystemExit are BaseException, so `except Exception` + lets them through — which is correct: a sweep that cannot be stopped is + worse than one that fails.""" + with pytest.raises(KeyboardInterrupt): + await sync_platform( + db, platform="patreon", fetch=_raises(KeyboardInterrupt("stop")), + ) + + +@pytest.mark.asyncio +async def test_celerys_soft_time_limit_is_not_swallowed_either(db): + """The one that actually needed code. SoftTimeLimitExceeded is an ORDINARY + Exception subclass, so the broad catch would have swallowed the worker's + request to stop and let the sweep run on into the HARD limit, where it is + SIGKILLed mid-transaction. Caught in review, not in production.""" + from celery.exceptions import SoftTimeLimitExceeded + + assert issubclass(SoftTimeLimitExceeded, Exception), ( + "if this ever becomes a BaseException the explicit re-raise is dead code" + ) + with pytest.raises(SoftTimeLimitExceeded): + await sync_platform( + db, platform="patreon", fetch=_raises(SoftTimeLimitExceeded()), + ) + + +@pytest.mark.asyncio +async def test_re_running_preserves_first_seen_and_updates_the_rest(db): + """Recovery is 'run it again' — which only works because the write is an + upsert that never moves first_seen_at (C1).""" + await sync_platform( + db, platform="patreon", + fetch=_fetch([FakeMembership("111", display_name="Old", amount_cents=500)]), + ) + db.expunge_all() + original = (await db.execute(select(PlatformMembership))).scalar_one() + first_seen = original.first_seen_at + db.expunge_all() + + await sync_platform( + db, platform="patreon", + fetch=_fetch([FakeMembership("111", display_name="New", amount_cents=1500)]), + ) + db.expunge_all() + row = (await db.execute(select(PlatformMembership))).scalar_one() + assert row.first_seen_at == first_seen + assert row.display_name == "New" + assert row.amount_cents == 1500 + + +@pytest.mark.asyncio +async def test_a_membership_that_disappears_is_kept_not_deleted(db): + """Disappearance is the signal C4 reads. Deleting the row would destroy it + at exactly the moment it became interesting.""" + await sync_platform( + db, platform="patreon", + fetch=_fetch([FakeMembership("111"), FakeMembership("222")]), + ) + await sync_platform( + db, platform="patreon", fetch=_fetch([FakeMembership("111")]), + ) + rows = (await db.execute(select(PlatformMembership))).scalars().all() + assert {r.external_campaign_id for r in rows} == {"111", "222"} + + +@pytest.mark.asyncio +async def test_the_free_member_flag_survives_into_details(db): + """`is_free_member` is a second axis the roster's columns do not model + (C0 correction 2), so it rides in details rather than being dropped.""" + await sync_platform( + db, platform="patreon", + fetch=_fetch([FakeMembership("111", is_free_member=True, status="former_patron")]), + ) + row = (await db.execute(select(PlatformMembership))).scalar_one() + assert row.details["is_free_member"] is True + assert row.status == "former_patron" + + +@pytest.mark.asyncio +async def test_platforms_keep_separate_sync_state(db): + await sync_platform(db, platform="patreon", fetch=_fetch([FakeMembership("1")])) + await sync_platform(db, platform="subscribestar", fetch=_raises(RuntimeError("x"))) + db.expunge_all() + + assert (await get_sync_state(db, "patreon")).last_error_type is None + assert (await get_sync_state(db, "subscribestar")).last_error_type == "RuntimeError" + assert roster_is_fresh(await get_sync_state(db, "patreon")) is True + assert roster_is_fresh(await get_sync_state(db, "subscribestar")) is False