From 96c30eba1338d206b7cb74287bf2f68ce03506d2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 5 Jun 2026 21:38:42 -0400 Subject: [PATCH] =?UTF-8?q?feat(patreon):=20phase-2=20ingester=20integrati?= =?UTF-8?q?on=20=E2=80=94=20build=20step=203=20(plan=20#697)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Branch download_service phase 2 by platform: Patreon now routes to the native PatreonIngester (zero per-file HEADs, native cursor/resume, loud drift detection) instead of gallery-dl; the other 5 platforms are unchanged. The ingester returns a DownloadResult-shaped object so phase 1 (DB setup) and phase 3 (import → pHash → thumbs → ML) are untouched. Three modes wired from config_overrides state: - tick: skip seen (tier-1 ledger + tier-2 disk), early-out after N contiguous already-have-it items. - backfill: full-history time-boxed chunk, cursor checkpoint via gallery-dl-style "Cursor: " lines in stdout (reuses the #693 lifecycle + parse_last_cursor verbatim). - recovery: backfill that BYPASSES the tier-1 seen-ledger so dropped-and-deleted near-dups get re-fetched and re-evaluated under the current pHash threshold. Rides the #693 state machine via a _backfill_bypass_seen flag, cleared on completion / stop. The seen-ledger uses short-lived sync sessions (injected sessionmaker), never held across the walk (avoids the connection-reaping trap). Campaign id resolves from override, an id: URL, or a vanity lookup; unresolvable = loud NOT_FOUND, never a silent empty success. Tests: new test_patreon_ingester.py (modes, ledger skip/idempotency, budget→PARTIAL, recovery bypass, tier-2 disk, drift). The patreon-oriented download_service tests now drive the ingester branch via a stub; the gallery-dl campaign-retry test is replaced by resolution/caching coverage. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/services/download_service.py | 151 +++++++--- backend/app/services/patreon_ingester.py | 365 +++++++++++++++++++++++ backend/app/services/source_service.py | 2 +- backend/app/tasks/download.py | 5 + tests/test_download_service.py | 246 ++++++++++----- tests/test_patreon_ingester.py | 324 ++++++++++++++++++++ 6 files changed, 984 insertions(+), 109 deletions(-) create mode 100644 backend/app/services/patreon_ingester.py create mode 100644 tests/test_patreon_ingester.py diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index 8804483..c84d657 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -29,12 +29,14 @@ from .gallery_dl import ( BACKFILL_CHUNK_SECONDS, BACKFILL_SKIP_VALUE, TICK_SKIP_VALUE, + DownloadResult, ErrorType, GalleryDLService, SourceConfig, parse_last_cursor, ) from .importer import Importer +from .patreon_ingester import PatreonIngester from .patreon_resolver import resolve_campaign_id from .platforms import auth_type_for from .scheduler_service import set_platform_cooldown @@ -78,12 +80,18 @@ class DownloadService: gdl: GalleryDLService, importer: Importer, cred_service: CredentialService, + sync_session_factory=None, ): self.async_session = async_session self.sync_session = sync_session self.gdl = gdl self.importer = importer self.cred_service = cred_service + # Sync sessionmaker the native Patreon ingester opens SHORT-LIVED + # sessions from for its seen-ledger reads/writes (never one held across + # the multi-minute walk — see PatreonIngester). Only the patreon branch + # of phase 2 uses it; gallery-dl sources leave it None. + self.sync_session_factory = sync_session_factory async def download_source(self, source_id: int) -> int: """Returns DownloadEvent.id. Idempotent: in-flight events are returned as-is.""" @@ -118,6 +126,14 @@ class DownloadService: # Operator drives this via POST /api/sources/{id}/backfill {action}. overrides = ctx["config_overrides"] or {} in_backfill = overrides.get("_backfill_state") == "running" + # Recovery (plan #697) reuses the entire #693 backfill state machine — + # cursor checkpoint, time-boxed chunks, complete/stall lifecycle — and + # differs only in bypassing the tier-1 seen-ledger so dropped-and-deleted + # near-dups get re-fetched and re-evaluated under the CURRENT pHash + # threshold (tier-2 disk still spares files we kept). The + # `_backfill_bypass_seen` flag rides alongside the running backfill state; + # download mode is "recovery" when both are set. + bypass_seen = bool(overrides.get("_backfill_bypass_seen")) if in_backfill: skip_value: bool | str = BACKFILL_SKIP_VALUE source_config.timeout = BACKFILL_CHUNK_SECONDS @@ -127,46 +143,35 @@ class DownloadService: else: skip_value = TICK_SKIP_VALUE - effective_url = _effective_url( - ctx["platform"], ctx["url"], ctx["config_overrides"] or {} - ) - - dl_result = await self.gdl.download( - url=effective_url, - artist_slug=ctx["artist_slug"], - platform=ctx["platform"], - source_config=source_config, - cookies_path=ctx["cookies_path"], - auth_token=ctx["auth_token"], - skip_value=skip_value, - ) - resolved_campaign_id: str | None = None - if ( - ctx["platform"] == "patreon" - and not dl_result.success - and "patreon_campaign_id" not in (ctx["config_overrides"] or {}) - and _looks_like_campaign_id_failure(dl_result.stdout, dl_result.stderr) - ): - vanity = _extract_patreon_vanity(ctx["url"]) - if vanity: - log.info( - "Attempting campaign-ID resolution for %s (%s)", - ctx["artist_slug"], vanity, - ) - resolved_campaign_id = await resolve_campaign_id( - vanity, ctx["cookies_path"] - ) - if resolved_campaign_id: - dl_result = await self.gdl.download( - url=f"https://www.patreon.com/id:{resolved_campaign_id}", - artist_slug=ctx["artist_slug"], - platform=ctx["platform"], - source_config=source_config, - cookies_path=ctx["cookies_path"], - auth_token=ctx["auth_token"], - skip_value=skip_value, - ) + if ctx["platform"] == "patreon": + # Native ingester (plan #697) fully replaces gallery-dl for Patreon + # in phase 2 — zero per-file HEADs, native cursor/resume, loud drift + # detection. Returns a DownloadResult-shaped object so phase 3 is + # untouched. The gallery-dl Patreon path + its campaign-id retry are + # removed at the step-5 cutover. + if in_backfill and bypass_seen: + mode = "recovery" + elif in_backfill: + mode = "backfill" + else: + mode = "tick" + dl_result, resolved_campaign_id = await self._run_patreon_ingester( + ctx, source_config, mode, + ) + else: + effective_url = _effective_url( + ctx["platform"], ctx["url"], ctx["config_overrides"] or {} + ) + dl_result = await self.gdl.download( + url=effective_url, + artist_slug=ctx["artist_slug"], + platform=ctx["platform"], + source_config=source_config, + cookies_path=ctx["cookies_path"], + auth_token=ctx["auth_token"], + skip_value=skip_value, + ) # A backfill chunk that hit its time-box but made forward progress is # NORMAL, not a failure — reclassify TIMEOUT → PARTIAL so it reads as @@ -190,6 +195,71 @@ class DownloadService: ctx["event_id"], ctx, dl_result, resolved_campaign_id, ) + async def _run_patreon_ingester( + self, ctx: dict, source_config, mode: str, + ) -> tuple[DownloadResult, str | None]: + """Phase-2 Patreon branch: resolve the campaign id, then run the native + ingester in a worker thread (it is sync requests/subprocess). + + Returns (DownloadResult, resolved_campaign_id). `resolved_campaign_id` is + non-None only when we had to look it up from the vanity URL this run, so + phase 3 caches it on the source the same way the old gallery-dl retry did. + A campaign id we cannot resolve is a loud failure (NOT_FOUND) — never a + silent empty success. + """ + overrides = ctx["config_overrides"] or {} + campaign_id = overrides.get("patreon_campaign_id") + resolved_campaign_id: str | None = None + if not campaign_id: + # A `.../id:` URL already carries the campaign id — no lookup + # needed (and the vanity regex deliberately excludes the id: form). + id_match = re.search(r"/id:(\d+)", ctx["url"]) + if id_match: + campaign_id = id_match.group(1) + else: + vanity = _extract_patreon_vanity(ctx["url"]) + if vanity: + campaign_id = await resolve_campaign_id(vanity, ctx["cookies_path"]) + resolved_campaign_id = campaign_id + + if not campaign_id: + return ( + DownloadResult( + success=False, + url=ctx["url"], + artist_slug=ctx["artist_slug"], + platform="patreon", + error_type=ErrorType.NOT_FOUND, + error_message=( + "Could not resolve Patreon campaign id from the source " + "URL (vanity lookup failed — cookies expired or creator " + "moved?)" + ), + ), + None, + ) + + ingester = PatreonIngester( + images_root=self.gdl.images_root, + cookies_path=ctx["cookies_path"], + session_factory=self.sync_session_factory, + validate=self.gdl._validate_files, + ) + loop = asyncio.get_running_loop() + dl_result = await loop.run_in_executor( + None, + lambda: ingester.run( + source_id=ctx["source_id"], + campaign_id=campaign_id, + artist_slug=ctx["artist_slug"], + url=ctx["url"], + mode=mode, + resume_cursor=source_config.resume_cursor, + time_budget_seconds=source_config.timeout, + ), + ) + return dl_result, resolved_campaign_id + async def _phase1_setup(self, source_id: int) -> dict[str, Any]: source = (await self.async_session.execute( select(Source).options(joinedload(Source.artist)).where(Source.id == source_id) @@ -468,6 +538,9 @@ class DownloadService: new_overrides["_backfill_state"] = "complete" new_overrides.pop("_backfill_cursor", None) new_overrides.pop("_backfill_cursor_stalls", None) + # plan #697: a recovery walk shares this lifecycle; clear its bypass + # flag on completion so the next routine tick honors the seen-ledger. + new_overrides.pop("_backfill_bypass_seen", None) src.config_overrides = new_overrides src.backfill_runs_remaining = 0 return diff --git a/backend/app/services/patreon_ingester.py b/backend/app/services/patreon_ingester.py new file mode 100644 index 0000000..6b26f8c --- /dev/null +++ b/backend/app/services/patreon_ingester.py @@ -0,0 +1,365 @@ +"""Native Patreon ingester — phase-2 orchestrator (build step 3). + +Ties build steps 1 (`patreon_client`) and 2 (`patreon_downloader` + +`patreon_seen_media`) together into a single sync walk that REPLACES the +gallery-dl subprocess for Patreon. `download_service.download_source` calls +`PatreonIngester.run(...)` from phase 2 (in a thread, via run_in_executor, since +everything here is sync `requests`/`subprocess`) and gets back a +`DownloadResult` — the exact shape gallery-dl returns — so phase 1 (DB setup) +and phase 3 (import → pHash dedup → thumbnails → ML) are untouched and cannot +tell the difference. + +Three modes (selected by `download_service` from `config_overrides` state): + - tick — newest→oldest, skip seen (tier-1 ledger + tier-2 disk), early-out + after N contiguous already-have-it items (the cheap native + equivalent of gallery-dl's `exit:20`, now free of per-file HEADs). + - backfill — full-history walk in a time-boxed chunk, resuming from the + pagination cursor checkpoint; reaches the bottom → "complete". + - recovery — like backfill but BYPASSES the tier-1 seen-ledger, so + deliberately-dropped-and-deleted near-dups get re-fetched and + re-evaluated under the current pHash threshold (tier-2 disk skip + still spares files we kept). Triggered by the same #693 backfill + state machine plus the `_backfill_bypass_seen` flag, so the whole + cursor/chunk/complete/stall lifecycle is reused verbatim. + +Cursor contract: the ingester emits gallery-dl-style ``Cursor: `` lines +into the returned `stdout`, one per page it walks. That is exactly what +`download_service`'s existing backfill lifecycle reads via +`parse_last_cursor(stdout, stderr)` — so checkpointing, the TIMEOUT→PARTIAL +reclassification, and completion detection all keep working unchanged. + +The seen-ledger lives in Postgres (`patreon_seen_media`). The ingester opens a +SHORT-LIVED sync session per page batch (via an injected sessionmaker) — never +held across a network fetch — so the multi-minute walk can't strand a checked-out +connection for the server to reap ([[db-connection-held-across-subprocess]]). + +FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API. +""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Callable +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert + +from ..models import PatreonSeenMedia +from .gallery_dl import DownloadResult, ErrorType +from .patreon_client import ( + MediaItem, + PatreonAPIError, + PatreonClient, + PatreonDriftError, +) +from .patreon_downloader import PatreonDownloader + +log = logging.getLogger(__name__) + +# gallery-dl's `exit:20` default ported over: stop a tick after this many +# CONTIGUOUS already-have-it media (seen-ledger or on-disk). Native walks have +# zero per-file HEADs, so the only cost of a higher number is a few extra cheap +# ledger lookups — 20 is operator-set headroom against paywalled/undownloadable +# items interleaving with archived ones. +_TICK_SEEN_THRESHOLD = 20 + +# Ledger keys are stored in patreon_seen_media.filehash VARCHAR(128); bound any +# synthesized key so a pathologically long file_name can't overflow the column. +_LEDGER_KEY_MAX = 128 + + +def _ledger_key(media: MediaItem) -> str: + """Stable per-media identity for the cross-run seen-ledger. + + A Patreon CDN URL carries a 32-char MD5 (`media.filehash`) — that is the + natural key. Some media have none: Mux/HLS video (`stream.mux.com`, no + content hash at discovery) and the odd inline-content `` pointing at a + hashless URL. The plan calls the video case the ``video::`` + sentinel; `MediaItem` carries no media_id, so the post-scoped filename is the + stable proxy. Bounded to the column width. + """ + if media.filehash: + return media.filehash + return f"{media.post_id}:{media.filename}"[:_LEDGER_KEY_MAX] + + +class PatreonIngester: + """Walk a Patreon campaign's posts, download unseen media, return a + `DownloadResult`. + + Construct with the per-source `cookies_path` and a sync sessionmaker for the + seen-ledger. `client` / `downloader` are injectable seams so unit tests run + without network, subprocess, or a real CDN. + """ + + def __init__( + self, + images_root: Path, + cookies_path: str | None, + session_factory: Callable[[], object], + *, + validate: bool = True, + client: PatreonClient | None = None, + downloader: PatreonDownloader | None = None, + ): + self.images_root = Path(images_root) + self.cookies_path = str(cookies_path) if cookies_path else None + self.session_factory = session_factory + self.client = client if client is not None else PatreonClient(cookies_path) + self.downloader = ( + downloader + if downloader is not None + else PatreonDownloader( + self.images_root, cookies_path, validate=validate + ) + ) + + # -- public ------------------------------------------------------------ + + def run( + self, + *, + source_id: int, + campaign_id: str, + artist_slug: str, + url: str, + mode: str, + resume_cursor: str | None = None, + time_budget_seconds: float = 870.0, + seen_threshold: int = _TICK_SEEN_THRESHOLD, + ) -> DownloadResult: + """Walk + download for one source, returning a gallery-dl-shaped result. + + `mode` is "tick" | "backfill" | "recovery". Recovery bypasses the tier-1 + seen-ledger (tier-2 disk still skips kept files). The walk stops on: + - budget exhaustion (time_budget_seconds) → TIMEOUT / PARTIAL + - tick early-out (seen_threshold contiguous seen) → success + - reaching the bottom of the feed → success (rc 0) + A client-level failure (drift / auth / network) fails the whole run loud. + """ + bypass_seen = mode == "recovery" + start = time.monotonic() + log_lines: list[str] = [] + written: list[str] = [] + downloaded = 0 + errors = 0 + consecutive_seen = 0 + emitted_cursor: str | None = None + reached_bottom = False + budget_hit = False + early_out = False + + def _result( + *, success: bool, return_code: int, + error_type: ErrorType | None, error_message: str | None, + ) -> DownloadResult: + return DownloadResult( + success=success, + url=url, + artist_slug=artist_slug, + platform="patreon", + files_downloaded=downloaded, + files_quarantined=0, + quarantined_paths=[], + written_paths=written, + stdout="\n".join(log_lines), + stderr="", + return_code=return_code, + error_type=error_type, + error_message=error_message, + duration_seconds=time.monotonic() - start, + ) + + try: + for post, included, page_cursor in self.client.iter_posts( + campaign_id, cursor=resume_cursor + ): + # Checkpoint: emit the cursor that FETCHED this page once, the + # moment we START it — so a chunk cut mid-page resumes the page, + # not the one after it (matches the gallery-dl cursor semantics + # download_service's lifecycle already depends on). + if page_cursor and page_cursor != emitted_cursor: + log_lines.append(f"Cursor: {page_cursor}") + emitted_cursor = page_cursor + + # Time-box check at the post boundary (coarse, like a gallery-dl + # chunk). Backfill/recovery resume from emitted_cursor next chunk. + if time.monotonic() - start >= time_budget_seconds: + budget_hit = True + break + + media = self.client.extract_media(post, included) + if not media: + continue + + keys = [_ledger_key(m) for m in media] + seen = ( + set() + if bypass_seen + else self._seen_keys(source_id, keys) + ) + + def _is_seen(m: MediaItem, _seen=seen) -> bool: + return _ledger_key(m) in _seen + + outcomes = self.downloader.download_post( + post, media, artist_slug, is_seen=_is_seen + ) + + to_mark: list[tuple[str, str]] = [] + for media_item, outcome in zip(media, outcomes, strict=False): + key = _ledger_key(media_item) + if outcome.status == "downloaded": + downloaded += 1 + if outcome.path is not None: + written.append(str(outcome.path)) + to_mark.append((key, media_item.post_id)) + consecutive_seen = 0 + elif outcome.status == "skipped_disk": + # Already on disk (a prior run). Reconcile the ledger so a + # later tick skips it at tier-1 without a disk stat, but + # do NOT re-feed it to phase 3 — attach_in_place would see + # the duplicate sha256 and unlink the on-disk copy. + to_mark.append((key, media_item.post_id)) + consecutive_seen += 1 + elif outcome.status == "skipped_seen": + consecutive_seen += 1 + elif outcome.status == "error": + errors += 1 + # An error neither advances nor resets the run-of-seen. + + if mode == "tick" and consecutive_seen >= seen_threshold: + early_out = True + break + + # Mark seen AFTER the network fetch, on its own short session. + if to_mark: + self._mark_seen(source_id, to_mark) + + if early_out: + break + else: + reached_bottom = True + except (PatreonDriftError, PatreonAPIError) as exc: + return self._failure_result(exc, _result) + + if errors: + log_lines.append(f"{errors} media item(s) failed") + log_lines.append( + f"Patreon ingest ({mode}): {downloaded} downloaded, " + f"{errors} error(s)" + + (", reached end" if reached_bottom else "") + + (", time-boxed" if budget_hit else "") + ) + + if budget_hit: + # A chunk that hit its time-box but made forward progress is a + # NORMAL chunk boundary, not a failure (PARTIAL → status "ok"); the + # next chunk resumes from the emitted cursor. No progress → TIMEOUT, + # which feeds download_service's backfill stall-guard. rc<0 mirrors + # subprocess TimeoutExpired so completion detection stays false. + made_progress = downloaded > 0 or emitted_cursor != resume_cursor + if made_progress: + return _result( + success=False, return_code=-1, + error_type=ErrorType.PARTIAL, + error_message=( + f"Patreon backfill chunk: {downloaded} file(s) — continuing" + ), + ) + return _result( + success=False, return_code=-1, + error_type=ErrorType.TIMEOUT, + error_message="Patreon chunk timed out with no progress", + ) + + # Normal success: reached the bottom, or a tick that early-outed. rc 0 + + # error_type None is REQUIRED for a backfill/recovery walk that reached + # the bottom to be marked COMPLETE by + # download_service._apply_backfill_lifecycle — so we return None even + # when downloaded == 0 (a re-confirming walk that found nothing new still + # completed). NO_NEW_CONTENT would read as "not finished" there and trip + # the stall guard. success=True maps to status "ok" regardless. A tick + # that early-outed also returns here; ticks never set backfill state so + # the lifecycle is a no-op for them. + return _result( + success=True, return_code=0, + error_type=None, error_message=None, + ) + + # -- failure mapping --------------------------------------------------- + + def _failure_result(self, exc: Exception, _result) -> DownloadResult: + """Map a client-level exception to a loud, typed failed DownloadResult. + + Step 3 keeps this deliberately coarse — auth-ish drift vs. everything + else — so the integration is correct end-to-end. Step 4 (drift detection + + error categorization) refines the typed-error mapping and adds the + contract test. Either way we NEVER return a silent zero-download + "success": the whole point of the native ingester is to fail red when + Patreon's API shape or our auth changes. + """ + message = str(exc) + lowered = message.lower() + if isinstance(exc, PatreonDriftError): + if "login" in lowered or "challenge" in lowered or "auth" in lowered: + error_type = ErrorType.AUTH_ERROR + else: + error_type = ErrorType.UNKNOWN_ERROR + message = f"Patreon API changed — ingester needs update: {message}" + else: + error_type = ErrorType.NETWORK_ERROR + log.warning("Patreon ingest failed: %s", message) + return _result( + success=False, return_code=1, + error_type=error_type, error_message=message, + ) + + # -- seen-ledger (short-lived sessions) -------------------------------- + + def _seen_keys(self, source_id: int, keys: list[str]) -> set[str]: + """Which of `keys` are already in the ledger for this source. + + One short SELECT on its own session — opened and closed without any + network in between (the GETs happen after, in download_post). + """ + if not keys: + return set() + with self.session_factory() as session: + rows = session.execute( + select(PatreonSeenMedia.filehash).where( + PatreonSeenMedia.source_id == source_id, + PatreonSeenMedia.filehash.in_(keys), + ) + ).scalars().all() + return set(rows) + + def _mark_seen(self, source_id: int, items: list[tuple[str, str]]) -> None: + """Idempotent upsert of (filehash, post_id) ledger rows for a page. + + ON CONFLICT DO NOTHING against the (source_id, filehash) UNIQUE so a + re-sighting — or a concurrent walk — is a harmless no-op + ([[scalar_one_or_none-duplicates]]: never check-then-insert without the + DB constraint backing it). De-dup the batch locally first so a single + page can't present the same key twice to one INSERT. + """ + seen_local: set[str] = set() + values = [] + for key, post_id in items: + if key in seen_local: + continue + seen_local.add(key) + values.append( + {"source_id": source_id, "filehash": key, "post_id": post_id} + ) + if not values: + return + with self.session_factory() as session: + stmt = pg_insert(PatreonSeenMedia).values(values) + stmt = stmt.on_conflict_do_nothing( + constraint="uq_patreon_seen_media_source_id" + ) + session.execute(stmt) + session.commit() diff --git a/backend/app/services/source_service.py b/backend/app/services/source_service.py index 7ae8f9c..0192a9c 100644 --- a/backend/app/services/source_service.py +++ b/backend/app/services/source_service.py @@ -329,7 +329,7 @@ class SourceService: 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_chunks", "_backfill_bypass_seen"): co.pop(k, None) source.config_overrides = co source.backfill_runs_remaining = 0 diff --git a/backend/app/tasks/download.py b/backend/app/tasks/download.py index d72e78b..42c075d 100644 --- a/backend/app/tasks/download.py +++ b/backend/app/tasks/download.py @@ -143,6 +143,11 @@ def download_source(self, source_id: int) -> int: gdl=gdl, importer=importer, cred_service=cred_service, + # The native Patreon ingester opens its own short-lived + # sync sessions for the seen-ledger (never held across + # the walk). Same factory the importer's sync session + # comes from — a different DB connection per checkout. + sync_session_factory=SyncFactory, ) return await svc.download_source(source_id) finally: diff --git a/tests/test_download_service.py b/tests/test_download_service.py index ee5766f..5e44df9 100644 --- a/tests/test_download_service.py +++ b/tests/test_download_service.py @@ -94,6 +94,23 @@ def _fake_gdl_with_result(result): return fake +def _stub_patreon_ingester(svc, result, resolved_campaign_id=None): + """Route the Patreon phase-2 branch (plan #697 native ingester) to a canned + DownloadResult so these tests exercise download_service's + phase-2-result→phase-3 handling (status mapping, backfill lifecycle, health) + without the real ingester's network/DB. The ingester itself is covered by + test_patreon_ingester.py. Returns a list capturing (ctx, source_config, mode) + per call so a test can assert mode / resume_cursor threading.""" + calls = [] + + async def fake(ctx, source_config, mode): + calls.append({"ctx": ctx, "source_config": source_config, "mode": mode}) + return result, resolved_campaign_id + + svc._run_patreon_ingester = fake + return calls + + @pytest.mark.asyncio async def test_download_source_attaches_written_files( db, db_sync, tmp_path, seed_artist_and_source, @@ -112,12 +129,13 @@ async def test_download_source_attaches_written_files( _make_jpg(f1, split="h") _make_jpg(f2, split="v") - fake_gdl = _fake_gdl_with_result(_make_fake_dl_result( + result = _make_fake_dl_result( success=True, written_paths=[str(f1), str(f2)], files_downloaded=2, stdout=f"{f1}\n{f2}\n", - )) + ) + fake_gdl = _fake_gdl_with_result(result) sync_settings = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) @@ -137,6 +155,7 @@ async def test_download_source_attaches_written_files( async_session=db, sync_session=db_sync, gdl=fake_gdl, importer=importer, cred_service=cred_service, ) + _stub_patreon_ingester(svc, result) event_id = await svc.download_source(source.id) ev = (await db.execute( @@ -214,40 +233,19 @@ async def test_in_flight_idempotency_returns_existing_event( @pytest.mark.asyncio -async def test_patreon_retry_caches_campaign_id( - db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, +async def test_patreon_resolved_campaign_id_is_cached( + db, db_sync, tmp_path, seed_artist_and_source, ): + """plan #697: the native ingester resolves a Patreon vanity → campaign id + on the fly (replacing gallery-dl's reactive campaign-id retry). When phase 2 + reports a freshly-resolved id, phase 3 caches it on the source so later runs + skip the lookup. Stub the ingester to report a resolved id; assert it lands + in config_overrides.""" from backend.app.services.download_service import DownloadService from backend.app.services.importer import Importer _artist, source = seed_artist_and_source - failed = _make_fake_dl_result( - success=False, written_paths=[], stdout="", - stderr="[patreon][error] Failed to extract campaign ID for alice\n", - ) - succeeded = _make_fake_dl_result(success=True, written_paths=[]) - - download_calls = [] - - async def fake_download(url, **k): - download_calls.append(url) - return failed if len(download_calls) == 1 else succeeded - - fake_gdl = MagicMock() - fake_gdl.download = fake_download - fake_gdl._compute_run_stats = lambda *a, **k: { - "exit_code": 0, "downloaded_count": 0, "skipped_count": 0, - "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0, - } - fake_gdl._extract_errors_warnings = lambda *a, **k: "" - fake_gdl._truncate_log = lambda x, **k: x - - monkeypatch.setattr( - "backend.app.services.download_service.resolve_campaign_id", - AsyncMock(return_value="99"), - ) - sync_settings = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() @@ -261,19 +259,102 @@ async def test_patreon_retry_caches_campaign_id( svc = DownloadService( async_session=db, sync_session=db_sync, - gdl=fake_gdl, importer=importer, cred_service=cred_service, + gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)), + importer=importer, cred_service=cred_service, + ) + _stub_patreon_ingester( + svc, _make_fake_dl_result(success=True, written_paths=[]), + resolved_campaign_id="99", ) await svc.download_source(source.id) - assert len(download_calls) == 2 - assert "id:99" in download_calls[1] - overrides = db_sync.execute( select(Source.config_overrides).where(Source.id == source.id) ).scalar_one() assert overrides.get("patreon_campaign_id") == "99" +@pytest.mark.asyncio +async def test_run_patreon_ingester_resolves_vanity_and_runs( + db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, +): + """_run_patreon_ingester: with no cached campaign id, resolve the vanity and + pass the resolved id straight to the ingester; report it back so phase 3 can + cache it.""" + from backend.app.services import download_service as dl_mod + from backend.app.services.download_service import DownloadService + from backend.app.services.gallery_dl import SourceConfig + + _artist, source = seed_artist_and_source + + monkeypatch.setattr( + dl_mod, "resolve_campaign_id", AsyncMock(return_value="4242"), + ) + run_kwargs = {} + + class _FakeIngester: + def __init__(self, **kw): + pass + + def run(self, **kw): + run_kwargs.update(kw) + return _make_fake_dl_result(success=True, written_paths=[]) + + monkeypatch.setattr(dl_mod, "PatreonIngester", _FakeIngester) + + svc = DownloadService( + async_session=db, sync_session=db_sync, + gdl=_fake_gdl_with_result(_make_fake_dl_result(success=True)), + importer=MagicMock(), cred_service=MagicMock(), + sync_session_factory=MagicMock(), + ) + ctx = { + "source_id": source.id, "url": "https://patreon.com/alice", + "artist_slug": "alice", "cookies_path": None, + "config_overrides": {}, + } + result, resolved = await svc._run_patreon_ingester( + ctx, SourceConfig.from_dict({}), "tick", + ) + assert result.success is True + assert resolved == "4242" + assert run_kwargs["campaign_id"] == "4242" + assert run_kwargs["mode"] == "tick" + + +@pytest.mark.asyncio +async def test_run_patreon_ingester_unresolvable_fails_loud( + db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, +): + """A campaign id we can't resolve is a loud NOT_FOUND failure, never a + silent empty success.""" + from backend.app.services import download_service as dl_mod + from backend.app.services.download_service import DownloadService + from backend.app.services.gallery_dl import ErrorType, SourceConfig + + _artist, source = seed_artist_and_source + + monkeypatch.setattr( + dl_mod, "resolve_campaign_id", AsyncMock(return_value=None), + ) + svc = DownloadService( + async_session=db, sync_session=db_sync, + gdl=MagicMock(), importer=MagicMock(), cred_service=MagicMock(), + sync_session_factory=MagicMock(), + ) + ctx = { + "source_id": source.id, "url": "https://patreon.com/alice", + "artist_slug": "alice", "cookies_path": None, + "config_overrides": {}, + } + result, resolved = await svc._run_patreon_ingester( + ctx, SourceConfig.from_dict({}), "tick", + ) + assert result.success is False + assert result.error_type == ErrorType.NOT_FOUND + assert resolved is None + + # --- FC-3d: finalize hook updates Source health columns ------------------- @@ -376,8 +457,12 @@ async def test_finalize_skipped_preserves_failures_clears_error(db): def _backfill_svc(db, db_sync, tmp_path, result): - """DownloadService wired with a fake gdl returning `result` + a real - importer (so an empty written_paths just attaches nothing).""" + """DownloadService wired with a real importer (so empty written_paths just + attaches nothing) whose Patreon phase-2 branch is stubbed to return `result`. + + The seeded source is Patreon, so phase 2 routes to the native ingester + (plan #697); the stub returns the canned result + captures the per-call + (ctx, source_config, mode). Returns (svc, ingester_calls).""" from backend.app.services.download_service import DownloadService from backend.app.services.importer import Importer @@ -395,7 +480,8 @@ def _backfill_svc(db, db_sync, tmp_path, result): async_session=db, sync_session=db_sync, gdl=fake_gdl, importer=importer, cred_service=cred_service, ) - return svc, fake_gdl + calls = _stub_patreon_ingester(svc, result) + return svc, calls @pytest.mark.asyncio @@ -430,28 +516,25 @@ async def test_backfill_chunk_progress_advances_cursor( async def test_backfill_state_running_selects_backfill_mode_and_resumes( db, db_sync, tmp_path, seed_artist_and_source, ): - """state=='running' drives backfill mode (skip:True + chunk budget) and - threads the stored cursor to gallery-dl as the resume point.""" - from backend.app.services.gallery_dl import ( - BACKFILL_CHUNK_SECONDS, - BACKFILL_SKIP_VALUE, - ) + """state=='running' drives backfill mode (chunk budget) and threads the + stored cursor to the native ingester as the resume point.""" + from backend.app.services.gallery_dl import BACKFILL_CHUNK_SECONDS _artist, source = seed_artist_and_source source.config_overrides = {"_backfill_state": "running", "_backfill_cursor": "03:RESUME:here"} source.backfill_runs_remaining = 5 await db.commit() - svc, fake_gdl = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result( + svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result( success=False, written_paths=[], stderr="[patreon][debug] Cursor: 03:RESUME2:next\n", )) await svc.download_source(source.id) - kwargs = fake_gdl.download.call_args.kwargs - assert kwargs["skip_value"] is BACKFILL_SKIP_VALUE - assert kwargs["source_config"].resume_cursor == "03:RESUME:here" - assert kwargs["source_config"].timeout == BACKFILL_CHUNK_SECONDS + assert len(calls) == 1 + assert calls[0]["mode"] == "backfill" + assert calls[0]["source_config"].resume_cursor == "03:RESUME:here" + assert calls[0]["source_config"].timeout == BACKFILL_CHUNK_SECONDS co = (await db.execute( select(Source.config_overrides).where(Source.id == source.id) @@ -605,6 +688,32 @@ async def test_tick_mode_when_not_running_leaves_state_untouched( assert (co or {}).get("_backfill_state") is None +@pytest.mark.asyncio +async def test_recovery_mode_selected_and_flag_cleared_on_complete( + db, db_sync, tmp_path, seed_artist_and_source, +): + """plan #697: `_backfill_bypass_seen` alongside a running backfill selects + recovery mode (ingester bypasses the seen-ledger). A clean rc=0 walk + completes the shared lifecycle AND clears the bypass flag so the next tick + honors the ledger again.""" + _artist, source = seed_artist_and_source + source.config_overrides = {"_backfill_state": "running", "_backfill_bypass_seen": True} + source.backfill_runs_remaining = 5 + await db.commit() + + svc, calls = _backfill_svc(db, db_sync, tmp_path, _make_fake_dl_result( + success=True, written_paths=[], files_downloaded=0, + )) + await svc.download_source(source.id) + + assert calls[0]["mode"] == "recovery" + co = (await db.execute( + select(Source.config_overrides).where(Source.id == source.id) + )).scalar_one() + assert co.get("_backfill_state") == "complete" + assert "_backfill_bypass_seen" not in co + + @pytest.mark.asyncio async def test_partial_error_type_maps_to_ok_status( db, db_sync, tmp_path, seed_artist_and_source, @@ -642,6 +751,7 @@ async def test_partial_error_type_maps_to_ok_status( async_session=db, sync_session=db_sync, gdl=fake_gdl, importer=importer, cred_service=cred_service, ) + _stub_patreon_ingester(svc, fake_result) event_id = await svc.download_source(source.id) ev = (await db.execute( @@ -677,10 +787,11 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image( _make_jpg(f1, split="h") _make_jpg(f2, split="v") - fake_gdl = _fake_gdl_with_result(_make_fake_dl_result( + result = _make_fake_dl_result( success=True, written_paths=[str(f1), str(f2)], files_downloaded=2, stdout=f"{f1}\n{f2}\n", - )) + ) + fake_gdl = _fake_gdl_with_result(result) sync_settings = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) @@ -713,6 +824,7 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image( async_session=db, sync_session=db_sync, gdl=fake_gdl, importer=importer, cred_service=cred_service, ) + _stub_patreon_ingester(svc, result) await svc.download_source(source.id) # Two files attached → two thumbnail enqueues + two ML enqueues, IDs @@ -727,11 +839,11 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image( async def test_releases_db_connections_before_subprocess( db, db_sync, tmp_path, seed_artist_and_source, monkeypatch, ): - """Phase 1's DB connections must be released BEFORE the (up to ~19.5-min - in backfill) gallery-dl subprocess, so they don't idle-die and strand - phase 3 with ConnectionDoesNotExistError (Anduo #40014). Spy on the - async session close and assert it happened before gdl.download runs; - phase 3 must still finalize the event.""" + """Phase 1's DB connections must be released BEFORE the (multi-minute) + phase-2 fetch, so they don't idle-die and strand phase 3 with + ConnectionDoesNotExistError (Anduo #40014). Spy on the async session close + and assert it happened before the phase-2 work (here the native Patreon + ingester) runs; phase 3 must still finalize the event.""" from backend.app.services.download_service import DownloadService from backend.app.services.importer import Importer @@ -747,19 +859,9 @@ async def test_releases_db_connections_before_subprocess( monkeypatch.setattr(db, "close", spy_close) seen = {} - - async def fake_download(url, **k): - seen["closed_before_download"] = closed["async"] - return _make_fake_dl_result(success=True, written_paths=[], stdout="") - - fake_gdl = MagicMock() - fake_gdl.download = fake_download - fake_gdl._compute_run_stats = lambda *a, **k: { - "exit_code": 0, "downloaded_count": 0, "skipped_count": 0, - "per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0, - } - fake_gdl._extract_errors_warnings = lambda *a, **k: "" - fake_gdl._truncate_log = lambda x, **k: x + fake_gdl = _fake_gdl_with_result( + _make_fake_dl_result(success=True, written_paths=[], stdout="") + ) sync_settings = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) @@ -776,9 +878,15 @@ async def test_releases_db_connections_before_subprocess( async_session=db, sync_session=db_sync, gdl=fake_gdl, importer=importer, cred_service=cred_service, ) + + async def fake_ingest(ctx, source_config, mode): + seen["closed_before_phase2"] = closed["async"] + return _make_fake_dl_result(success=True, written_paths=[], stdout=""), None + + svc._run_patreon_ingester = fake_ingest event_id = await svc.download_source(source.id) - assert seen["closed_before_download"] is True + assert seen["closed_before_phase2"] is True ev = (await db.execute( select(DownloadEvent).where(DownloadEvent.id == event_id) )).scalar_one() diff --git a/tests/test_patreon_ingester.py b/tests/test_patreon_ingester.py new file mode 100644 index 0000000..17ac2cf --- /dev/null +++ b/tests/test_patreon_ingester.py @@ -0,0 +1,324 @@ +"""PatreonIngester tests (native ingester build step 3). + +The HTTP client and the media downloader are stubbed (injected seams), so these +exercise the ingester's own logic — mode behavior, the two-tier skip, cursor +emission, the budget cutoff, and the Postgres seen-ledger — without network or a +real CDN. The ledger is real (a sync sessionmaker bound to the test engine), so +the tier-1 skip and the idempotent mark-seen run against actual rows. +""" + +import pytest +from sqlalchemy import func, select +from sqlalchemy.orm import sessionmaker + +from backend.app.models import Artist, PatreonSeenMedia, Source +from backend.app.services.gallery_dl import ErrorType +from backend.app.services.patreon_client import MediaItem, PatreonDriftError +from backend.app.services.patreon_downloader import MediaOutcome +from backend.app.services.patreon_ingester import PatreonIngester, _ledger_key + +pytestmark = pytest.mark.integration + + +# --- fakes ---------------------------------------------------------------- + + +def _media(post_id, n, *, filehash=None, kind="images"): + # Deterministic 32-hex-ish key, unique per (post_id, n). + fh = filehash if filehash is not None else f"{post_id}{n}".encode().hex().ljust(32, "0")[:32] + return MediaItem( + url=f"https://cdn.patreon.com/{fh}/{n}.jpg", + filename=f"{n}.jpg", + kind=kind, + filehash=fh, + post_id=post_id, + ) + + +class _FakeClient: + """Stub PatreonClient. `pages` is a list of (page_cursor, [posts]); each post + is (post_id, [MediaItem]). `raise_on_first` lets a test trip drift.""" + + def __init__(self, pages, raise_exc=None): + self._pages = pages + self._raise_exc = raise_exc + self.consumed_posts = 0 + + def iter_posts(self, campaign_id, cursor=None): + if self._raise_exc is not None: + raise self._raise_exc + for page_cursor, posts in self._pages: + for post_id, media in posts: + self.consumed_posts += 1 + yield {"id": post_id, "_media": media}, {}, page_cursor + + def extract_media(self, post, included_index): + return post["_media"] + + +class _FakeDownloader: + """Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in + `on_disk` report skipped_disk (tier-2); everything else downloads.""" + + def __init__(self, tmp_path, on_disk=None): + self.tmp_path = tmp_path + self.on_disk = set(on_disk or ()) + self.download_calls = 0 + + def download_post(self, post, media_items, artist_slug, *, is_seen): + outcomes = [] + for m in media_items: + if is_seen(m): + outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None)) + elif _ledger_key(m) in self.on_disk: + p = self.tmp_path / f"{m.post_id}_{m.filename}" + outcomes.append(MediaOutcome(media=m, status="skipped_disk", path=p, error=None)) + else: + self.download_calls += 1 + p = self.tmp_path / f"{m.post_id}_{m.filename}" + p.write_bytes(b"x") + outcomes.append(MediaOutcome(media=m, status="downloaded", path=p, error=None)) + return outcomes + + +@pytest.fixture +async def source_id(db): + artist = Artist(name="Ingest", slug="ingest") + db.add(artist) + await db.flush() + source = Source( + artist_id=artist.id, platform="patreon", + url="https://patreon.com/ingest", enabled=True, config_overrides={}, + ) + db.add(source) + await db.commit() + return source.id + + +def _ingester(sync_engine, tmp_path, client, downloader): + factory = sessionmaker(sync_engine, expire_on_commit=False) + return PatreonIngester( + images_root=tmp_path, cookies_path=None, session_factory=factory, + client=client, downloader=downloader, + ) + + +def _count_ledger(sync_engine, source_id): + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + return s.execute( + select(func.count(PatreonSeenMedia.id)).where( + PatreonSeenMedia.source_id == source_id + ) + ).scalar_one() + + +# --- tick ----------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_path): + m1, m2 = _media("p1", 1), _media("p1", 2) + client = _FakeClient([(None, [("p1", [m1, m2])])]) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + + assert result.success is True + assert result.return_code == 0 + assert result.files_downloaded == 2 + assert len(result.written_paths) == 2 + assert _count_ledger(sync_engine, source_id) == 2 + + +@pytest.mark.asyncio +async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path): + m1 = _media("p1", 1) + # Pre-seed the ledger with m1's key. + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1")) + s.commit() + + client = _FakeClient([(None, [("p1", [m1])])]) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + assert result.files_downloaded == 0 + assert downloader.download_calls == 0 + + +@pytest.mark.asyncio +async def test_tick_early_out_after_threshold(source_id, sync_engine, tmp_path): + # Three posts, one already-seen media each; threshold 2 → stop before the 3rd. + seen_media = [_media(f"p{i}", 1) for i in range(1, 4)] + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + for m in seen_media: + s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m), post_id=m.post_id)) + s.commit() + + pages = [(None, [(m.post_id, [m]) for m in seen_media])] + client = _FakeClient(pages) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", seen_threshold=2, + ) + assert result.success is True + # Stopped after the 2nd contiguous seen item — never consumed the 3rd post. + assert client.consumed_posts == 2 + + +# --- backfill ------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine, tmp_path): + pages = [ + (None, [("p1", [_media("p1", 1)])]), + ("CUR2", [("p2", [_media("p2", 1)])]), + ] + client = _FakeClient(pages) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="backfill", + ) + # rc 0 + error_type None is what _apply_backfill_lifecycle reads as "complete". + assert result.return_code == 0 + assert result.error_type is None + assert result.files_downloaded == 2 + # The page-2 cursor was emitted for checkpointing. + assert "Cursor: CUR2" in result.stdout + + +@pytest.mark.asyncio +async def test_backfill_budget_cut_returns_partial_with_progress( + source_id, sync_engine, tmp_path, monkeypatch, +): + # Deterministic clock: start=0, post1 check=10 (ok), post2 check=200 (>budget). + import backend.app.services.patreon_ingester as mod + ticks = iter([0.0, 10.0, 200.0, 250.0]) + last = [0.0] + + def fake_monotonic(): + try: + last[0] = next(ticks) + except StopIteration: + pass + return last[0] + + monkeypatch.setattr(mod.time, "monotonic", fake_monotonic) + + pages = [ + ("CUR1", [("p1", [_media("p1", 1)])]), + ("CUR2", [("p2", [_media("p2", 1)])]), + ] + client = _FakeClient(pages) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="backfill", + resume_cursor=None, time_budget_seconds=100.0, + ) + assert result.error_type == ErrorType.PARTIAL + assert result.return_code == -1 + assert result.files_downloaded == 1 # post1 downloaded before the cut + assert downloader.download_calls == 1 # post2's body never ran + assert "Cursor: CUR1" in result.stdout + + +# --- recovery ------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_recovery_bypasses_seen_ledger(source_id, sync_engine, tmp_path): + m1 = _media("p1", 1) + factory = sessionmaker(sync_engine, expire_on_commit=False) + with factory() as s: + s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1")) + s.commit() + + client = _FakeClient([(None, [("p1", [m1])])]) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="recovery", + ) + # Ledger says seen, but recovery bypasses tier-1 → re-downloaded for a fresh + # pHash evaluation under the current threshold. + assert result.files_downloaded == 1 + assert downloader.download_calls == 1 + + +@pytest.mark.asyncio +async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path): + m1 = _media("p1", 1) + client = _FakeClient([(None, [("p1", [m1])])]) + # File still on disk (a kept image) → tier-2 spares it even under recovery. + downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)}) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="recovery", + ) + assert result.files_downloaded == 0 + assert downloader.download_calls == 0 + # Disk-skip still reconciles the ledger so a later tick skips at tier-1. + assert _count_ledger(sync_engine, source_id) == 1 + + +# --- ledger + drift ------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mark_seen_is_idempotent(source_id, sync_engine, tmp_path): + m1 = _media("p1", 1) + client = _FakeClient([(None, [("p1", [m1])])]) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + # Same media downloaded twice across two runs — the second marks seen again. + ing._mark_seen(source_id, [(_ledger_key(m1), "p1")]) + ing._mark_seen(source_id, [(_ledger_key(m1), "p1")]) + assert _count_ledger(sync_engine, source_id) == 1 + + +@pytest.mark.asyncio +async def test_drift_fails_loud(source_id, sync_engine, tmp_path): + client = _FakeClient([], raise_exc=PatreonDriftError("missing top-level 'data' key")) + downloader = _FakeDownloader(tmp_path) + ing = _ingester(sync_engine, tmp_path, client, downloader) + + result = ing.run( + source_id=source_id, campaign_id="c1", artist_slug="ingest", + url="https://patreon.com/ingest", mode="tick", + ) + assert result.success is False + assert "Patreon API changed" in (result.error_message or "") + + +def test_ledger_key_video_has_no_filehash(): + video = MediaItem( + url="https://stream.mux.com/abc.m3u8", filename="clip.mp4", + kind="postfile", filehash=None, post_id="p9", + ) + assert _ledger_key(video) == "p9:clip.mp4"