Merge pull request 'SubscribeStar → native core ingester + native-ingest DRY pass (milestone #71)' (#117) from dev into main

alembic/versions/0054_subscribestar_ledgers.py

@@ -0,0 +1,82 @@
+"""subscribestar_seen_media + subscribestar_failed_media: per-source ledgers
+
+Revision ID: 0054
+Revises: 0053
+Create Date: 2026-06-17
+
+SubscribeStar native ingester (phase 1 of the gallery-dl → native-core
+migration). Mirrors the Patreon ledger tables (0037/0038): a seen-ledger so
+routine walks skip already-ingested media (recovery bypasses it) and a
+dead-letter ledger so persistently-failing media stops re-burning backfill
+chunks. `filehash` is a CDN content hash when present, else a synthesized
+``<post_id>:<filename>`` key — hence String(128). UNIQUE (source_id, filehash)
+is the upsert key on each.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0054"
+down_revision: Union[str, None] = "0053"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "subscribestar_seen_media",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "source_id",
+            sa.Integer,
+            sa.ForeignKey("source.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("filehash", sa.String(128), nullable=False),
+        sa.Column("post_id", sa.String(64), nullable=True),
+        sa.Column(
+            "seen_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.UniqueConstraint(
+            "source_id", "filehash", name="uq_subscribestar_seen_media_source_id"
+        ),
+    )
+    op.create_table(
+        "subscribestar_failed_media",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "source_id",
+            sa.Integer,
+            sa.ForeignKey("source.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("filehash", sa.String(128), nullable=False),
+        sa.Column("attempts", sa.Integer, nullable=False, server_default="1"),
+        sa.Column("last_error", sa.Text, nullable=True),
+        sa.Column(
+            "first_failed_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.Column(
+            "last_failed_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.UniqueConstraint(
+            "source_id", "filehash", name="uq_subscribestar_failed_media_source_id"
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("subscribestar_failed_media")
+    op.drop_table("subscribestar_seen_media")

backend/app/models/__init__.py

@@ -24,6 +24,8 @@ from .series_chapter import SeriesChapter
 from .series_page import SeriesPage
 from .series_suggestion import SeriesSuggestion
 from .source import Source
+from .subscribestar_failed_media import SubscribeStarFailedMedia
+from .subscribestar_seen_media import SubscribeStarSeenMedia
 from .tag import Tag, TagKind, image_tag
 from .tag_alias import TagAlias
 from .tag_allowlist import TagAllowlist
@@ -41,6 +43,8 @@ __all__ = [
     "Credential",
     "PatreonFailedMedia",
     "PatreonSeenMedia",
+    "SubscribeStarFailedMedia",
+    "SubscribeStarSeenMedia",
     "Post",
     "PostAttachment",
     "SeriesChapter",

backend/app/models/subscribestar_failed_media.py

@@ -0,0 +1,44 @@
+"""SubscribeStarFailedMedia — per-source dead-letter ledger of SubscribeStar
+media that keeps failing to download/validate.
+
+Mirror of PatreonFailedMedia. Media that fails every walk (404'd CDN URL,
+deleted post, persistently-corrupt bytes) would otherwise re-error forever and
+re-burn backfill chunks. After ``attempts`` reaches the dead-letter threshold
+the ingester skips it on routine tick/backfill walks (recovery still
+re-attempts). A later clean download clears the row.
+
+`filehash` is the same per-media key the seen-ledger uses (CDN content hash or a
+synthesized ``<post_id>:<filename>`` key) — hence String(128). UNIQUE
+(source_id, filehash) is the upsert key.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.types import DateTime
+
+from .base import Base
+
+
+class SubscribeStarFailedMedia(Base):
+    __tablename__ = "subscribestar_failed_media"
+    __table_args__ = (
+        UniqueConstraint(
+            "source_id", "filehash", name="uq_subscribestar_failed_media_source_id"
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_id: Mapped[int] = mapped_column(
+        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
+    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    first_failed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    last_failed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

backend/app/models/subscribestar_seen_media.py

@@ -0,0 +1,40 @@
+"""SubscribeStarSeenMedia — per-source ledger of SubscribeStar media already
+downloaded+processed.
+
+Mirror of PatreonSeenMedia for the SubscribeStar native ingester (replacing
+gallery-dl). One queryable row per (source, media) so routine walks skip media
+we've already ingested; recovery mode bypasses the ledger to re-walk.
+
+`filehash` is a CDN content hash when the media URL carries one, else a
+synthesized ``<post_id>:<filename>`` key (SubscribeStar URLs aren't always
+content-addressed) — hence String(128) rather than 32.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import ForeignKey, Integer, String, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.types import DateTime
+
+from .base import Base
+
+
+class SubscribeStarSeenMedia(Base):
+    __tablename__ = "subscribestar_seen_media"
+    __table_args__ = (
+        # Dedup key the downloader upserts against: one ledger row per
+        # (source, media). A second sighting of the same media is a no-op.
+        UniqueConstraint(
+            "source_id", "filehash", name="uq_subscribestar_seen_media_source_id"
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    source_id: Mapped[int] = mapped_column(
+        ForeignKey("source.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    filehash: Mapped[str] = mapped_column(String(128), nullable=False)
+    post_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    seen_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

backend/app/services/download_backends.py

@@ -24,19 +24,32 @@ import asyncio
 from pathlib import Path
 
 from .gallery_dl import DownloadResult, ErrorType
+from .native_ingest_common import NativeIngestError
 from .patreon_ingester import PatreonIngester
 from .patreon_resolver import extract_vanity, resolve_campaign_id_for_source
+from .subscribestar_ingester import SubscribeStarIngester
 
 # Platforms whose download + verify go through the native ingester rather than
-# gallery-dl. gallery-dl still serves every other platform (subscribestar,
-# hentaifoundry, discord, pixiv, deviantart) unchanged.
-NATIVE_INGESTER_PLATFORMS = frozenset({"patreon"})
+# gallery-dl. gallery-dl still serves the rest (hentaifoundry, discord, pixiv,
+# deviantart) until they migrate too.
+NATIVE_INGESTER_PLATFORMS = frozenset({"patreon", "subscribestar"})
 
 # Mirrors patreon_resolver._CAMPAIGNS_URL — surfaced in resolution-failure
 # messages so the operator sees the exact lookup endpoint that was hit.
 _CAMPAIGNS_API = "https://www.patreon.com/api/campaigns"
 
 
+def _native_ingester_cls(platform: str):
+    """The native ingester class for `platform` (uniform constructor signature).
+    A call-time lookup (not a module-level dict captured at import) so tests can
+    monkeypatch db_mod.PatreonIngester / SubscribeStarIngester and have the
+    dispatch pick up the replacement."""
+    return {
+        "patreon": PatreonIngester,
+        "subscribestar": SubscribeStarIngester,
+    }[platform]
+
+
 def uses_native_ingester(platform: str) -> bool:
     """True when `platform` is served by the native ingester (not gallery-dl).
     The single predicate the download path and verify both route on."""
@@ -81,22 +94,36 @@ async def run_download(
     return result, None
 
 
+async def _resolve_native_campaign_id(
+    platform: str, url: str, cookies_path: str | None, overrides: dict,
+) -> tuple[str | None, str | None]:
+    """`(campaign_id, resolved_campaign_id)` for a native source. SubscribeStar's
+    feed id IS the creator URL (no lookup → resolved None). Patreon resolves the
+    campaign id from the vanity URL (resolved non-None when a lookup actually ran,
+    so phase 3 caches it)."""
+    if platform == "subscribestar":
+        return url, None
+    return await resolve_campaign_id_for_source(url, cookies_path, overrides)
+
+
 async def _run_native_ingester(
     ctx: dict, source_config, mode: str | None, gdl, sync_session_factory,
 ) -> tuple[DownloadResult, str | None]:
-    """Patreon (today the only native platform): resolve the campaign id, then run
-    the native ingester in a worker thread (it is sync requests/subprocess).
+    """Run the native ingester for a native platform in a worker thread (sync
+    requests/subprocess). Patreon resolves a campaign id from the vanity URL;
+    SubscribeStar's feed id is the creator URL itself. A campaign id we cannot
+    resolve is a loud NOT_FOUND — never a silent empty success.
 
-    `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 the way the old gallery-dl retry
-    did. A campaign id we cannot resolve is a loud NOT_FOUND — never a silent
-    empty success.
+    `resolved_campaign_id` is non-None only when a lookup ran this call, so phase
+    3 caches it the way the old gallery-dl retry did.
     """
+    platform = ctx["platform"]
     overrides = ctx["config_overrides"] or {}
-    campaign_id, resolved_campaign_id = await resolve_campaign_id_for_source(
-        ctx["url"], ctx["cookies_path"], overrides
+    campaign_id, resolved_campaign_id = await _resolve_native_campaign_id(
+        platform, ctx["url"], ctx["cookies_path"], overrides
     )
     if not campaign_id:
+        # Only reachable for Patreon (SubscribeStar's campaign id is the URL).
         url = ctx["url"]
         vanity = extract_vanity(url)
         return (
@@ -104,7 +131,7 @@ async def _run_native_ingester(
                 success=False,
                 url=url,
                 artist_slug=ctx["artist_slug"],
-                platform="patreon",
+                platform=platform,
                 error_type=ErrorType.NOT_FOUND,
                 error_message=(
                     f"Could not resolve Patreon campaign id. source_url={url!r}; "
@@ -127,7 +154,7 @@ async def _run_native_ingester(
         if source_config.sleep_request is not None
         else max(0.5, rate_limit / 4)
     )
-    ingester = PatreonIngester(
+    ingester = _native_ingester_cls(platform)(
         images_root=gdl.images_root,
         cookies_path=ctx["cookies_path"],
         session_factory=sync_session_factory,
@@ -174,10 +201,8 @@ async def preview_source(
     """
     import asyncio
 
-    from .patreon_client import PatreonAPIError
-
-    campaign_id, _ = await resolve_campaign_id_for_source(
-        url, cookies_path, config_overrides or {}
+    campaign_id, _ = await _resolve_native_campaign_id(
+        platform, url, cookies_path, config_overrides or {}
     )
     if not campaign_id:
         vanity = extract_vanity(url)
@@ -188,7 +213,7 @@ async def preview_source(
                 "(cookies expired, or the creator moved/renamed?)."
             )
         }
-    ingester = PatreonIngester(
+    ingester = _native_ingester_cls(platform)(
         images_root=images_root,
         cookies_path=cookies_path,
         session_factory=sync_session_factory,
@@ -199,7 +224,7 @@ async def preview_source(
             None,
             lambda: ingester.preview(source_id, campaign_id, page_limit=page_limit),
         )
-    except PatreonAPIError as exc:
+    except NativeIngestError as exc:
         return {"error": f"Couldn't preview: {exc}"}
     return result
 
@@ -221,7 +246,12 @@ async def verify_source_credential(
     """
     if uses_native_ingester(platform):
         # Native ingester platforms verify via their own lightweight auth probe
-        # (resolve campaign id + one authenticated API page). Patreon today.
+        # (one authenticated feed fetch). SubscribeStar's probe takes the creator
+        # URL directly; Patreon's resolves the campaign id first.
+        if platform == "subscribestar":
+            from .subscribestar_ingester import verify_subscribestar_credential
+
+            return await verify_subscribestar_credential(url, cookies_path, config_overrides)
         from .patreon_ingester import verify_patreon_credential
 
         return await verify_patreon_credential(url, cookies_path, config_overrides)

backend/app/services/gallery_dl.py

@@ -298,11 +298,8 @@ class GalleryDLService:
     # removed at the plan-#697 cutover — it now uses the native ingester
     # (services/patreon_ingester.py), not gallery-dl.
     PLATFORM_DEFAULTS = {
-        "subscribestar": {
-            "content_types": ["all"],
-            "directory": ["{date:%Y-%m-%d}_{id}_{title[:40]}"],
-            "filename": "{num:>02}_{filename}.{extension}",
-        },
+        # subscribestar removed — it's a native-ingester platform now (#71); the
+        # remaining entries are the gallery-dl platforms not yet migrated.
         "hentaifoundry": {
             "content_types": ["all"],
             "directory": [],

backend/app/services/ingest_core.py

@@ -36,6 +36,7 @@ from sqlalchemy import delete, func, select, text
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from .gallery_dl import DownloadResult, ErrorType, make_run_stats
+from .native_ingest_common import NativeAuthError, NativeDriftError
 
 log = logging.getLogger(__name__)
 
@@ -88,6 +89,7 @@ class Ingester:
         ledger_key: Callable[[object], str],
         platform: str,
         error_base: type[Exception],
+        drift_label: str | None = None,
     ):
         self.client = client
         self.downloader = downloader
@@ -99,6 +101,10 @@ class Ingester:
         self._ledger_key = ledger_key
         self._platform = platform
         self._error_base = error_base
+        # Human label for the API_DRIFT message ("<label> changed — ingester needs
+        # update"). Defaults to the platform name; adapters pass a richer phrase
+        # (e.g. "Patreon API", "SubscribeStar markup").
+        self._drift_label = drift_label or platform
 
     # -- public ------------------------------------------------------------
 
@@ -234,6 +240,15 @@ class Ingester:
                 ),
             )
 
+        # #899 L1: emit run milestones through the real logger (not only the
+        # in-memory log_lines → DownloadResult.stdout, which is persisted to the
+        # DownloadEvent ONLY at phase 3). A worker SIGKILL/OOM/hard-time-limit
+        # mid-walk would otherwise leave NO trace; these land in the container log
+        # in real time regardless of whether the event gets finalized.
+        log.info(
+            "%s ingest START (%s): source=%s campaign=%s resume_cursor=%s",
+            self._platform, mode, source_id, campaign_id, resume_cursor,
+        )
         try:
             for post, included, page_cursor in self.client.iter_posts(
                 campaign_id, cursor=resume_cursor
@@ -243,6 +258,15 @@ class Ingester:
                 # after it. Carried as DownloadResult.cursor (plan #704).
                 if page_cursor and page_cursor != emitted_cursor:
                     emitted_cursor = page_cursor
+                    # #899 L1: a per-page breadcrumb in the container log (pages can
+                    # be minutes apart on image-dense backfills) — survives a worker
+                    # kill so the operator sees how far a since-died walk got.
+                    log.info(
+                        "%s ingest progress (%s, source=%s): posts=%d downloaded=%d "
+                        "skipped=%d errors=%d quarantined=%d gated=%d cursor=%s",
+                        self._platform, mode, source_id, posts_processed, downloaded,
+                        skipped_count, errors, quarantined, gated_skipped, emitted_cursor,
+                    )
                     # plan #705 #6: persist the cursor at each page boundary so a
                     # worker SIGKILL mid-chunk resumes near the crash, not the
                     # chunk start. (phase 3 still writes the final cursor — same
@@ -444,6 +468,10 @@ class Ingester:
         # posts), so don't re-write them — return PARTIAL (reads as "ok/progress",
         # the lifecycle no-ops since state is gone) instead of a false "complete".
         if stopped:
+            log.info(
+                "%s ingest STOPPED by operator (%s, source=%s): %d file(s) this chunk",
+                self._platform, mode, source_id, downloaded,
+            )
             return _result(
                 success=False, return_code=-1,
                 error_type=ErrorType.PARTIAL,
@@ -461,7 +489,7 @@ class Ingester:
             log_lines.append(f"{quarantined} media item(s) quarantined (invalid)")
         if dead_lettered:
             log_lines.append(f"{dead_lettered} media item(s) skipped (dead-lettered)")
-        log_lines.append(
+        summary = (
             f"{self._platform} ingest ({mode}): {downloaded} downloaded, "
             f"{skipped_count} skipped, {quarantined} quarantined, "
             f"{dead_lettered} dead-lettered, {errors} error(s), "
@@ -475,6 +503,9 @@ class Ingester:
             + (", reached end" if reached_bottom else "")
             + (", time-boxed" if budget_hit else "")
         )
+        log_lines.append(summary)
+        # #899 L1: also to the container log (survives event-finalization failure).
+        log.info("%s (source=%s)", summary, source_id)
 
         if budget_hit:
             # A chunk that hit its time-box but made forward progress is a
@@ -601,13 +632,41 @@ class Ingester:
     # -- failure mapping (adapter overrides) -------------------------------
 
     def _failure_result(self, exc: Exception, _result) -> DownloadResult:
-        """Map a platform client-error to a typed failed DownloadResult. The base
-        gives a safe default; adapters override with their exception taxonomy."""
-        log.warning("%s ingest failed: %s", self._platform, exc)
-        return _result(
+        """Map a platform client-error to a loud, typed failed DownloadResult —
+        NEVER a silent zero-download "success". The mapping is shared across
+        platforms via the NativeAuthError/NativeDriftError taxonomy (the platform
+        client raises subclasses), so a new platform gets it for free:
+          - NativeAuthError  → AUTH_ERROR   (rotate the credential)
+          - NativeDriftError → API_DRIFT    (the ingester/scraper needs updating)
+          - HTTP 429 / 404   → RATE_LIMITED / NOT_FOUND
+          - other HTTP status→ HTTP_ERROR;  transport failure → NETWORK_ERROR
+        Auth/Drift are matched first (they also carry a status_code in some paths).
+        """
+        message = str(exc)
+        if isinstance(exc, NativeAuthError):
+            error_type = ErrorType.AUTH_ERROR
+        elif isinstance(exc, NativeDriftError):
+            error_type = ErrorType.API_DRIFT
+            message = f"{self._drift_label} changed — ingester needs update: {message}"
+        else:
+            status = getattr(exc, "status_code", None)
+            if status == 429:
+                error_type = ErrorType.RATE_LIMITED
+            elif status == 404:
+                error_type = ErrorType.NOT_FOUND
+            elif status is not None:
+                error_type = ErrorType.HTTP_ERROR
+            else:
+                error_type = ErrorType.NETWORK_ERROR
+        log.warning("%s ingest failed (%s): %s", self._platform, error_type.value, message)
+        result = _result(
             success=False, return_code=1,
-            error_type=ErrorType.UNKNOWN_ERROR, error_message=str(exc),
+            error_type=error_type, error_message=message,
         )
+        # plan #708 B1: carry the server's Retry-After up to the cooldown.
+        if error_type == ErrorType.RATE_LIMITED:
+            result.retry_after_seconds = getattr(exc, "retry_after", None)
+        return result
 
     # -- seen-ledger (short-lived sessions) --------------------------------
 

backend/app/services/native_ingest_common.py

@@ -0,0 +1,345 @@
+"""Shared primitives for the native-ingest platform adapters (Patreon,
+SubscribeStar, …) — the single home for logic the per-platform client/downloader
+modules would otherwise each copy.
+
+DRY pass 2026-06-17 (#899): these used to live in `patreon_*` with the
+SubscribeStar modules importing patreon privates (wrong owner + sibling-coupling).
+They're platform-agnostic, so they live here and both adapters import them. The
+per-platform modules keep only what genuinely differs (feed parsing, the media
+shape, Patreon's Mux/yt-dlp video branch + detail-fetch enrichment).
+
+FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import http.cookiejar
+import json
+import logging
+import os
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from urllib.parse import urlsplit
+
+import requests
+
+from ..utils.paths import filehash_from_url, safe_ext
+from .file_validator import is_validatable, quarantine_file, validate_file
+
+log = logging.getLogger(__name__)
+
+_USER_AGENT = (
+    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
+)
+
+# 429 backoff (plan #703): ride out a transient API rate-limit instead of failing
+# the whole walk. Honor the server's Retry-After; else exponential, capped.
+_MAX_429_RETRIES = 3
+_BACKOFF_BASE_SECONDS = 2.0
+_BACKOFF_CAP_SECONDS = 30.0
+
+# Media-download tuning (shared by every platform downloader).
+_TIMEOUT_SECONDS = 120.0
+_CHUNK = 1 << 16
+_MAX_MEDIA_RETRIES = 3
+_TRANSIENT_TRANSPORT_EXC = (
+    requests.ConnectionError,
+    requests.Timeout,
+    requests.exceptions.ChunkedEncodingError,
+)
+
+_TITLE_MAX = 40
+# Windows/gallery-dl path-restrict forbidden set + path separators.
+_FORBIDDEN = set('<>:"/\\|?*')
+
+
+# -- shared exception taxonomy --------------------------------------------
+# Every native client raises one of these (platform subclasses keep an
+# isinstance-distinct platform name AND the semantic Auth/Drift class), so the
+# base Ingester._failure_result can map them platform-agnostically.
+
+class NativeIngestError(Exception):
+    """Base for a native-ingest client failure. `status_code` carries the HTTP
+    status when the failure was an HTTP response (None for transport/parse);
+    `retry_after` carries the server's 429 Retry-After hint so the cooldown can
+    match it (plan #708 B1)."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: int | None = None,
+        retry_after: float | None = None,
+    ):
+        super().__init__(message)
+        self.status_code = status_code
+        self.retry_after = retry_after
+
+
+class NativeAuthError(NativeIngestError):
+    """Authentication/authorization failure — expired/missing credential or an
+    insufficient tier. The fix is rotating the credential, NOT updating the
+    ingester. Maps to error_type 'auth_error'."""
+
+
+class NativeDriftError(NativeIngestError):
+    """A response did not match the shape the ingester depends on (JSON:API field
+    set, or scraped HTML structure). Fail loud so the import step flags 'the
+    platform changed' instead of silently importing nothing. Maps to API_DRIFT."""
+
+
+# -- HTTP session ----------------------------------------------------------
+
+def make_session(
+    cookies_path: str | Path | None,
+    *,
+    accept: str = "*/*",
+    extra_headers: dict | None = None,
+) -> requests.Session:
+    """Build a requests.Session loaded with the Netscape cookies.txt
+    CredentialService materializes. `accept` sets the Accept header (the JSON:API
+    vs HTML feed differ); `extra_headers` adds platform headers (e.g.
+    X-Requested-With). Missing/unparseable cookies log a warning, never fail."""
+    session = requests.Session()
+    headers = {"User-Agent": _USER_AGENT, "Accept": accept}
+    if extra_headers:
+        headers.update(extra_headers)
+    session.headers.update(headers)
+    if cookies_path and os.path.isfile(str(cookies_path)):
+        try:
+            jar = http.cookiejar.MozillaCookieJar(str(cookies_path))
+            jar.load(ignore_discard=True, ignore_expires=True)
+            session.cookies = jar  # type: ignore[assignment]
+        except (OSError, http.cookiejar.LoadError) as exc:
+            log.warning("Could not load cookies from %s: %s", cookies_path, exc)
+    return session
+
+
+def retry_after_seconds(
+    resp: requests.Response,
+    attempt: int,
+    *,
+    base: float = _BACKOFF_BASE_SECONDS,
+    cap: float = _BACKOFF_CAP_SECONDS,
+) -> float:
+    """Backoff for a 429: the numeric Retry-After header if present, else
+    exponential base·2^(attempt-1), both capped."""
+    header = resp.headers.get("Retry-After")
+    if header:
+        try:
+            return min(float(header), cap)
+        except (TypeError, ValueError):
+            pass
+    return min(base * (2 ** max(0, attempt - 1)), cap)
+
+
+# -- filename / path helpers -----------------------------------------------
+
+def sanitize_segment(name: str) -> str:
+    """Make `name` safe for one filesystem path segment: replace separators, the
+    Windows-forbidden set, and control chars with `_`; strip trailing dots/spaces
+    (gallery-dl path-restrict). Never empty (falls back to `_`)."""
+    out = ["_" if (ch in _FORBIDDEN or ord(ch) < 32) else ch for ch in name]
+    cleaned = "".join(out).rstrip(". ")
+    return cleaned or "_"
+
+
+def basename_from_url(url: str) -> str:
+    """Derive a sane filename from a URL when the media has no name: path basename
+    with a junk-extension guard (safe_ext), bounded stem; falls back to the URL's
+    content hash, then "file"."""
+    path = urlsplit(url).path
+    base = os.path.basename(path)
+    if base:
+        ext = safe_ext(base)
+        stem = base[: -len(Path(base).suffix)] if Path(base).suffix else base
+        stem = stem[:120] or "file"
+        return f"{stem}{ext}"
+    return filehash_from_url(url) or "file"
+
+
+def post_dir_name(post: dict) -> str:
+    """`<YYYY-MM-DD>_<post_id>_<title40>` matching gallery-dl's layout (date prefix
+    omitted when published_at is missing/unparseable; title is empty for platforms
+    with no title field). Accepts both ISO and trailing-`Z` published_at."""
+    post_id = str(post.get("id") or "")
+    attrs = post.get("attributes") or {}
+    title = attrs.get("title")
+    title40 = (title if isinstance(title, str) else "")[:_TITLE_MAX]
+    published = attrs.get("published_at")
+    date_prefix = None
+    if isinstance(published, str) and published:
+        s = published.strip()
+        if s.endswith("Z"):
+            s = s[:-1] + "+00:00"
+        try:
+            date_prefix = f"{datetime.fromisoformat(s):%Y-%m-%d}"
+        except ValueError:
+            date_prefix = None
+    raw = f"{date_prefix}_{post_id}_{title40}" if date_prefix else f"{post_id}_{title40}"
+    return sanitize_segment(raw)
+
+
+# -- per-item download outcomes (shared dataclasses) -----------------------
+
+@dataclass
+class MediaOutcome:
+    """Per-media result of a download_post pass. status ∈ downloaded /
+    skipped_seen / skipped_disk / quarantined / error. `path` is the on-disk file
+    (downloaded / skipped_disk), the quarantine dest (quarantined), or None;
+    `error` is the failure/validation reason (error/quarantined) else None."""
+
+    media: object
+    status: str
+    path: Path | None
+    error: str | None
+
+
+@dataclass
+class PostRecordOutcome:
+    """Result of write_post_record — mirrors the MediaOutcome contract so the core
+    reports per-post handling. `path` is the _post.json sidecar (None when the post
+    had no id); the rest is the captured body's shape for the run log."""
+
+    path: Path | None
+    post_type: str | None
+    title: str | None
+    body_chars: int
+
+
+# -- base downloader (shared fetch/validate plumbing) ----------------------
+
+class BaseNativeDownloader:
+    """Shared download plumbing for native-platform downloaders: the streaming
+    GET (transient-retry + Range-resume) and file validation/quarantine. Platform
+    downloaders subclass this and implement `download_post` / `write_post_record`
+    / the per-media sidecar (and any platform-specific fetch, e.g. Patreon's
+    Mux/yt-dlp video branch). PURE: no DB; the seen-skip is an injected predicate.
+    """
+
+    def __init__(
+        self,
+        images_root: Path,
+        cookies_path: str | None = None,
+        *,
+        platform: str,
+        validate: bool = True,
+        rate_limit: float = 0.0,
+        session: requests.Session | None = None,
+    ):
+        self.images_root = Path(images_root)
+        self.cookies_path = str(cookies_path) if cookies_path else None
+        self.platform = platform
+        self._validate = validate
+        self._rate_limit = rate_limit or 0.0
+        self.session = session if session is not None else make_session(cookies_path)
+
+    # -- download seams ----------------------------------------------------
+
+    def _fetch_get(self, url: str, dest: Path) -> Path:
+        """Stream `url` to a .part then atomic-rename to `dest`."""
+        part = dest.with_name(dest.name + ".part")
+        try:
+            self._fetch_to_file(url, part)
+        except Exception:
+            with contextlib.suppress(OSError):
+                part.unlink()
+            raise
+        os.replace(part, dest)
+        return dest
+
+    def _fetch_to_file(self, url: str, dest: Path) -> None:
+        """Stream a URL to `dest`, retrying TRANSIENT failures (transport blips,
+        429 honoring Retry-After, 5xx) with backoff + resume-from-disk (Range);
+        failing fast on permanent 4xx (404/403). Resume: a retry with bytes on
+        disk asks `Range: bytes=<have>-`; 206 → append, 200 → restart clean, 416 →
+        already complete. The caller stages into a `.part` so a non-range server
+        never corrupts the output."""
+        attempt = 0
+        while True:
+            have = dest.stat().st_size if dest.exists() else 0
+            headers = {"Range": f"bytes={have}-"} if have > 0 else None
+            try:
+                resp = self.session.get(
+                    url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers,
+                )
+                if (resp.status_code == 429 or resp.status_code >= 500) \
+                        and attempt < _MAX_MEDIA_RETRIES:
+                    attempt += 1
+                    delay = retry_after_seconds(resp, attempt)
+                    log.warning(
+                        "%s media transient HTTP %d (%s) — backing off %.1fs "
+                        "(retry %d/%d)",
+                        self.platform, resp.status_code, url, delay, attempt,
+                        _MAX_MEDIA_RETRIES,
+                    )
+                    time.sleep(delay)
+                    continue
+                if have > 0 and resp.status_code == 416:
+                    return
+                resp.raise_for_status()
+                mode = "ab" if (have > 0 and resp.status_code == 206) else "wb"
+                with open(dest, mode) as fh:
+                    for chunk in resp.iter_content(chunk_size=_CHUNK):
+                        if chunk:
+                            fh.write(chunk)
+                return
+            except _TRANSIENT_TRANSPORT_EXC as exc:
+                if attempt >= _MAX_MEDIA_RETRIES:
+                    raise
+                attempt += 1
+                delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS)
+                log.warning(
+                    "%s media transport error (%s) — backing off %.1fs "
+                    "(retry %d/%d): %s",
+                    self.platform, url, delay, attempt, _MAX_MEDIA_RETRIES, exc,
+                )
+                time.sleep(delay)
+
+    # -- validation --------------------------------------------------------
+
+    def _validate_path(
+        self, path: Path, artist_slug: str, source_url: str | None = None
+    ) -> tuple[str | None, Path | None]:
+        """Validate a freshly-written file; quarantine if bad (shared
+        file_validator move + provenance sidecar). Returns (reason,
+        quarantine_dest) when quarantined, else (None, None). Logs the quarantine
+        so a corrupt file is visible in the worker logs, not just counted (#899
+        L2)."""
+        if not self._validate or not is_validatable(path):
+            return None, None
+        try:
+            result = validate_file(path)
+        except Exception as exc:
+            log.warning("Validator raised on %s: %s", path, exc)
+            return None, None
+        if result.ok:
+            return None, None
+        dest = quarantine_file(
+            self.images_root, path, artist_slug, self.platform,
+            url=source_url, result=result,
+        )
+        reason = result.reason or "validation failed"
+        log.warning(
+            "%s quarantined %s (%s) — %s",
+            self.platform, dest or path, artist_slug, reason,
+        )
+        return reason, (dest or path)
+
+    # -- sidecar (per-media, minimal) --------------------------------------
+
+    def _write_minimal_sidecar(
+        self, post: dict, media_path: Path, *, source_url: str | None = None
+    ) -> Path:
+        """Post-first per-media sidecar (#856): image identity ONLY
+        (category/id/source_url). The post body/links live solely in _post.json."""
+        data: dict = {"category": self.platform, "id": str(post.get("id") or "")}
+        if source_url:
+            data["source_url"] = source_url
+        sidecar_path = media_path.with_suffix(".json")
+        sidecar_path.write_text(json.dumps(data, indent=2))
+        return sidecar_path

backend/app/services/patreon_client.py

@@ -26,9 +26,7 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
 
 from __future__ import annotations
 
-import http.cookiejar
 import logging
-import os
 import re
 import time
 from collections.abc import Iterator
@@ -39,46 +37,23 @@ from urllib.parse import parse_qs, urlsplit
 
 import requests
 
-from ..utils.paths import filehash_from_url, safe_ext
+from ..utils.paths import filehash_from_url
 from ..utils.prosemirror import post_body_html
+from .native_ingest_common import (
+    _MAX_429_RETRIES,
+    NativeAuthError,
+    NativeDriftError,
+    NativeIngestError,
+    basename_from_url,
+    make_session,
+    retry_after_seconds,
+)
 
 log = logging.getLogger(__name__)
 
 _POSTS_URL = "https://www.patreon.com/api/posts"
-_USER_AGENT = (
-    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
-    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
-)
 _TIMEOUT_SECONDS = 30.0
 
-# 429 backoff (plan #703): ride out a transient API rate-limit instead of
-# failing the whole walk (which would stamp RATE_LIMITED → platform-wide
-# cooldown → every Patreon source dark). Honor the server's `Retry-After`;
-# otherwise exponential base·2^(n-1), capped. Only after the retries are
-# exhausted does the 429 propagate as terminal RATE_LIMITED.
-_MAX_429_RETRIES = 3
-_BACKOFF_BASE_SECONDS = 2.0
-_BACKOFF_CAP_SECONDS = 30.0
-
-
-def _retry_after_seconds(
-    resp: requests.Response,
-    attempt: int,
-    *,
-    base: float = _BACKOFF_BASE_SECONDS,
-    cap: float = _BACKOFF_CAP_SECONDS,
-) -> float:
-    """Backoff delay for a 429: the `Retry-After` seconds header if present and
-    numeric, else exponential `base·2^(attempt-1)`, both capped. (HTTP-date form
-    of Retry-After is rare here and falls through to exponential.)"""
-    header = resp.headers.get("Retry-After")
-    if header:
-        try:
-            return min(float(header), cap)
-        except (TypeError, ValueError):
-            pass
-    return min(base * (2 ** max(0, attempt - 1)), cap)
-
 # JSON:API request contract (observed from real traffic — see module plan).
 _INCLUDE = (
     "campaign,access_rules,attachments,attachments_media,audio,images,media,"
@@ -101,30 +76,12 @@ _FIELDS_CAMPAIGN = "name,url"
 _CONTENT_IMG_RE = re.compile(r"<img\b[^>]*?\bsrc=[\"']([^\"']+)[\"']", re.IGNORECASE)
 
 
-class PatreonAPIError(Exception):
-    """Base for native Patreon client failures.
-
-    `status_code` carries the HTTP status when the failure was an HTTP response
-    (None for transport-level / parse failures), so the ingester can map it to a
-    DownloadResult.error_type (429 → rate_limited, 404 → not_found, …).
-    `retry_after` carries the server's `Retry-After` seconds on a terminal 429, so
-    the platform cooldown can match the server's hint instead of a flat default
-    (plan #708 B1).
-    """
-
-    def __init__(
-        self,
-        message: str,
-        *,
-        status_code: int | None = None,
-        retry_after: float | None = None,
-    ):
-        super().__init__(message)
-        self.status_code = status_code
-        self.retry_after = retry_after
+class PatreonAPIError(NativeIngestError):
+    """Base for native Patreon client failures. status_code / retry_after are
+    inherited from NativeIngestError (HTTP status; 429 Retry-After hint)."""
 
 
-class PatreonAuthError(PatreonAPIError):
+class PatreonAuthError(PatreonAPIError, NativeAuthError):
     """Authentication / authorization failure — missing or expired session
     cookies, an insufficient pledge tier, or an HTML login/challenge page served
     where JSON was expected. DISTINCT from drift: the fix is rotating the
@@ -132,7 +89,7 @@ class PatreonAuthError(PatreonAPIError):
     """
 
 
-class PatreonDriftError(PatreonAPIError):
+class PatreonDriftError(PatreonAPIError, NativeDriftError):
     """A JSON response did not match the JSON:API shape we depend on.
 
     Raised for: a missing top-level `data` list, `data` not a list, or a media
@@ -168,49 +125,12 @@ class MediaItem:
     post_id: str
 
 
-def _load_session(cookies_path: str | Path | None) -> requests.Session:
-    session = requests.Session()
-    session.headers.update(
-        {
-            "User-Agent": _USER_AGENT,
-            "Accept": "application/vnd.api+json",
-        }
-    )
-    if cookies_path and os.path.isfile(str(cookies_path)):
-        try:
-            jar = http.cookiejar.MozillaCookieJar(str(cookies_path))
-            jar.load(ignore_discard=True, ignore_expires=True)
-            session.cookies = jar  # type: ignore[assignment]
-        except (OSError, http.cookiejar.LoadError) as exc:
-            log.warning("Could not load Patreon cookies from %s: %s", cookies_path, exc)
-    return session
-
-
 def _filehash(url: str) -> str | None:
     # Delegate to the shared extractor (utils.paths) so capture-time persistence
     # and render-time inline-image matching use the EXACT same identity.
     return filehash_from_url(url)
 
 
-def _basename_from_url(url: str) -> str:
-    """Derive a sane filename from a URL when the media has no file_name.
-
-    Strips query/fragment, takes the path basename, and drops a junk
-    extension (the importer._safe_ext gotcha) so we never write base64 noise
-    as a name. Falls back to the filehash, then to "file".
-    """
-    path = urlsplit(url).path
-    base = os.path.basename(path)
-    if base:
-        ext = safe_ext(base)
-        stem = base[: -len(Path(base).suffix)] if Path(base).suffix else base
-        # Keep the stem bounded; URL-encoded stems can be enormous.
-        stem = stem[:120] or "file"
-        return f"{stem}{ext}"
-    fh = _filehash(url)
-    return fh or "file"
-
-
 def parse_cursor_from_url(url: str | None) -> str | None:
     """Extract the `page[cursor]` query param from a links.next URL."""
     if not url:
@@ -238,7 +158,7 @@ class PatreonClient:
         max_retries: int = _MAX_429_RETRIES,
     ):
         self.cookies_path = str(cookies_path) if cookies_path else None
-        self._session = _load_session(cookies_path)
+        self._session = make_session(cookies_path, accept="application/vnd.api+json")
         # Politeness: seconds to sleep before each /api/posts page fetch (paces
         # the rate-limited API endpoint). 0 = no pacing. plan #703.
         self._request_sleep = request_sleep or 0.0
@@ -283,7 +203,7 @@ class PatreonClient:
             # through to the terminal RATE_LIMITED raise below.
             if resp.status_code == 429 and attempt < self._max_retries:
                 attempt += 1
-                delay = _retry_after_seconds(resp, attempt)
+                delay = retry_after_seconds(resp, attempt)
                 log.warning(
                     "Patreon 429 (campaign_id=%s) — backing off %.1fs (retry %d/%d)",
                     campaign_id, delay, attempt, self._max_retries,
@@ -416,7 +336,7 @@ class PatreonClient:
         # uses. A genuine schema change shows up as no URL (above) or a media id
         # absent from `included` (caller), not a missing name.
         file_name = attrs.get("file_name")
-        filename = file_name if isinstance(file_name, str) and file_name else _basename_from_url(url)
+        filename = file_name if isinstance(file_name, str) and file_name else basename_from_url(url)
         return MediaItem(
             url=url,
             filename=filename,
@@ -462,7 +382,7 @@ class PatreonClient:
                 items.append(
                     MediaItem(
                         url=large_url,
-                        filename=_basename_from_url(large_url),
+                        filename=basename_from_url(large_url),
                         kind="image_large",
                         filehash=_filehash(large_url),
                         post_id=post_id,
@@ -481,7 +401,7 @@ class PatreonClient:
                 filename = (
                     pf_name
                     if isinstance(pf_name, str) and pf_name
-                    else _basename_from_url(pf_url)
+                    else basename_from_url(pf_url)
                 )
                 items.append(
                     MediaItem(
@@ -503,7 +423,7 @@ class PatreonClient:
                 items.append(
                     MediaItem(
                         url=src,
-                        filename=_basename_from_url(src),
+                        filename=basename_from_url(src),
                         kind="content",
                         filehash=_filehash(src),
                         post_id=post_id,

backend/app/services/patreon_downloader.py

@@ -27,46 +27,33 @@ FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
 
 from __future__ import annotations
 
-import contextlib
 import json
 import logging
 import os
 import subprocess
 import time
 from collections.abc import Callable
-from dataclasses import dataclass
-from datetime import datetime
 from pathlib import Path
 from urllib.parse import urlsplit
 
 import requests
 
 from ..utils.prosemirror import post_body_html
-from .file_validator import is_validatable, quarantine_file, validate_file
-from .patreon_client import (
+from .native_ingest_common import (
     _BACKOFF_CAP_SECONDS,
-    _load_session,
-    _retry_after_seconds,
+    _MAX_MEDIA_RETRIES,
+    BaseNativeDownloader,
+    MediaOutcome,
+    PostRecordOutcome,
+    post_dir_name,
+    sanitize_segment,
 )
 
 log = logging.getLogger(__name__)
 
-_TITLE_MAX = 40
+# yt-dlp subprocess wall-clock per attempt (video only; the shared HTTP fetch
+# budgets live in BaseNativeDownloader).
 _TIMEOUT_SECONDS = 120.0
-_CHUNK = 1 << 16
-# Retry a media GET that hits a TRANSIENT failure within the same pass (plan
-# #705 #8): a transport blip (connection reset / timeout / truncated stream), a
-# 429, or a 5xx. PERMANENT failures (404 gone, 403 forbidden) fail fast straight
-# to the error/dead-letter path — no point re-fetching them. Keeps a momentary
-# network hiccup from becoming a per-item error that waits for the next walk.
-_MAX_MEDIA_RETRIES = 3
-# requests transport errors worth retrying (vs. an HTTPError, which is a real
-# server response and is classified by status code).
-_TRANSIENT_TRANSPORT_EXC = (
-    requests.ConnectionError,
-    requests.Timeout,
-    requests.exceptions.ChunkedEncodingError,
-)
 
 # Referer/Origin yt-dlp must send for Mux-hosted Patreon video. Mux's JWT
 # playback policy checks Referer/Origin on every request, so yt-dlp must send
@@ -78,26 +65,6 @@ _VIDEO_HEADERS = {
     "Origin": "https://www.patreon.com",
 }
 
-# Characters Windows/gallery-dl path-restrict forbids, plus path separators.
-_FORBIDDEN = set('<>:"/\\|?*')
-
-
-def _sanitize(name: str) -> str:
-    """Make `name` safe for a single filesystem path segment.
-
-    Replaces path separators, the Windows-forbidden set <>:"/\\|?* and control
-    characters with `_`, then strips trailing dots/spaces (gallery-dl
-    path-restrict behavior). Never returns empty (falls back to "_").
-    """
-    out = []
-    for ch in name:
-        if ch in _FORBIDDEN or ord(ch) < 32:
-            out.append("_")
-        else:
-            out.append(ch)
-    cleaned = "".join(out).rstrip(". ")
-    return cleaned or "_"
-
 
 def _is_video_url(url: str) -> bool:
     parts = urlsplit(url)
@@ -106,73 +73,12 @@ def _is_video_url(url: str) -> bool:
     return parts.path.lower().endswith(".m3u8")
 
 
-def _post_dir_name(post: dict) -> str:
-    """Build the post directory name matching gallery-dl's layout."""
-    post_id = str(post.get("id") or "")
-    attrs = post.get("attributes") or {}
-    title = attrs.get("title")
-    title = title if isinstance(title, str) else ""
-    title40 = title[:_TITLE_MAX]
-
-    published = attrs.get("published_at")
-    date_prefix = None
-    if isinstance(published, str) and published:
-        s = published.strip()
-        if s.endswith("Z"):
-            s = s[:-1] + "+00:00"
-        try:
-            dt = datetime.fromisoformat(s)
-        except ValueError:
-            dt = None
-        if dt is not None:
-            date_prefix = f"{dt:%Y-%m-%d}"
-
-    if date_prefix:
-        raw = f"{date_prefix}_{post_id}_{title40}"
-    else:
-        raw = f"{post_id}_{title40}"
-    return _sanitize(raw)
-
-
-@dataclass
-class MediaOutcome:
-    """Per-media result of a download_post pass.
-
-    status is one of: "downloaded", "skipped_seen", "skipped_disk",
-    "quarantined", "error". `path` is the final on-disk path for "downloaded"
-    (the actual yt-dlp output for video), the path that already existed for
-    "skipped_disk", or the _quarantine destination for "quarantined"; None for
-    "skipped_seen" and (usually) "error". `error` carries the failure/validation
-    reason for "error"/"quarantined", else None.
-    """
-
-    media: object  # MediaItem (avoid importing the name for a bare annotation)
-    status: str
-    path: Path | None
-    error: str | None
-
-
-@dataclass
-class PostRecordOutcome:
-    """Result of write_post_record — mirrors the download_post → MediaOutcome
-    contract so the engine reports per-post handling without re-reading the post.
-    `path` is the _post.json sidecar (None when the post had no id); the rest is
-    the captured body's shape (post_type + final char count) for the run log.
-    """
-
-    path: Path | None
-    post_type: str | None
-    title: str | None
-    body_chars: int
-
-
-class PatreonDownloader:
-    """Download resolved Patreon media to gallery-dl's on-disk layout.
-
-    PURE: no DB. The HTTP session and the yt-dlp invocation are injectable seams
-    so tests run without network or a real subprocess:
-      - pass `session=` to stub `session.get`, or monkeypatch `_fetch_to_file`.
-      - monkeypatch `_run_ytdlp` to avoid spawning yt-dlp.
+class PatreonDownloader(BaseNativeDownloader):
+    """Download resolved Patreon media to gallery-dl's on-disk layout. Subclasses
+    BaseNativeDownloader for the shared streaming GET (transient-retry +
+    Range-resume) and validation/quarantine; adds the Mux/HLS yt-dlp video branch
+    and the detail-fetch body enrichment. PURE: no DB. `_run_ytdlp` is
+    monkeypatchable and the HTTP session is the injectable `session=` seam.
     """
 
     def __init__(
@@ -185,23 +91,16 @@ class PatreonDownloader:
         session: requests.Session | None = None,
         content_fetcher: Callable[[str], str | None] | None = None,
     ):
-        self.images_root = Path(images_root)
-        self.cookies_path = str(cookies_path) if cookies_path else None
-        self._validate = validate
+        super().__init__(
+            images_root, cookies_path, platform="patreon",
+            validate=validate, rate_limit=rate_limit, session=session,
+        )
         # Best-effort enrichment seam: (post_id) -> full HTML body, or None. The
         # feed endpoint often omits `content`; the adapter wires this to
         # PatreonClient.fetch_post_detail_content so the sidecar captures the
         # real body (formatting + inline <img> + external <a href> links).
         # None in unit tests / when enrichment isn't wanted.
         self._content_fetcher = content_fetcher
-        # Politeness: seconds to sleep before each actual media download (paces
-        # the CDN; honors ImportSettings.download_rate_limit_seconds, the same
-        # value gallery-dl used as its between-downloads `sleep`). 0 = no pacing.
-        # Applied only to real downloads, not to seen/disk skips. plan #703.
-        self._rate_limit = rate_limit or 0.0
-        # Build a cookie-loaded session the same way patreon_client does, so the
-        # CDN GETs carry the creator's auth.
-        self.session = session if session is not None else _load_session(cookies_path)
 
     # -- public ------------------------------------------------------------
 
@@ -234,7 +133,7 @@ class PatreonDownloader:
         would tier-1 skip it — so the engine can backfill source_filehash for
         inline-image localization. Genuinely-missing seen media is NOT refetched.
         """
-        post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post)
+        post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post)
         outcomes: list[MediaOutcome] = []
 
         for i, media in enumerate(media_items, start=1):
@@ -279,7 +178,7 @@ class PatreonDownloader:
             return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
 
         nn = f"{index:02d}"
-        final_name = _sanitize(f"{nn}_{media.filename}")
+        final_name = sanitize_segment(f"{nn}_{media.filename}")
         media_path = post_dir / final_name
 
         # tier-2: already on disk.
@@ -333,87 +232,9 @@ class PatreonDownloader:
         self._write_sidecar(post, out_path, source_url=media.url)
         return MediaOutcome(media=media, status="downloaded", path=out_path, error=None)
 
-    # -- download seams ----------------------------------------------------
-
-    def _fetch_get(self, url: str, dest: Path) -> Path:
-        """Stream `url` to a .part file then atomic-rename to `dest`.
-
-        Thin wrapper over `_fetch_to_file` so tests can stub either the whole
-        GET path (`_fetch_to_file`) or just `session.get`.
-        """
-        part = dest.with_name(dest.name + ".part")
-        try:
-            self._fetch_to_file(url, part)
-        except Exception:
-            with contextlib.suppress(OSError):
-                part.unlink()
-            raise
-        os.replace(part, dest)
-        return dest
-
-    def _fetch_to_file(self, url: str, dest: Path) -> None:
-        """Stream a non-video URL to `dest` via the (stubbable) session, retrying
-        TRANSIENT failures within the same pass (plan #705 #8) and RESUMING from
-        the bytes already on disk via a Range request when a retry follows a
-        mid-download cut (plan #708 B5).
-
-        Retried (backoff): transport blips (connection reset / timeout /
-        truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After),
-        and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter
-        path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a
-        permanent failure is pointless.
-
-        Resume: on a retry, if bytes already landed in `dest`, ask for the rest
-        with `Range: bytes=<have>-`. A 206 means the server honored it → append; a
-        200 means it ignored it (served the whole file) → start clean. The caller
-        (_fetch_get) stages into a `.part`, so a non-range server never corrupts
-        the output — the worst case is re-downloading from zero, as before.
-        """
-        attempt = 0
-        while True:
-            have = dest.stat().st_size if dest.exists() else 0
-            headers = {"Range": f"bytes={have}-"} if have > 0 else None
-            try:
-                resp = self.session.get(
-                    url, stream=True, timeout=_TIMEOUT_SECONDS, headers=headers,
-                )
-                if (resp.status_code == 429 or resp.status_code >= 500) \
-                        and attempt < _MAX_MEDIA_RETRIES:
-                    attempt += 1
-                    delay = _retry_after_seconds(resp, attempt)
-                    log.warning(
-                        "Patreon media transient HTTP %d (%s) — backing off "
-                        "%.1fs (retry %d/%d)",
-                        resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES,
-                    )
-                    time.sleep(delay)
-                    continue
-                # A Range that starts at/past EOF (we already have the whole file)
-                # comes back 416 — the bytes we kept ARE the file.
-                if have > 0 and resp.status_code == 416:
-                    return
-                # 2xx → ok; 4xx-non-429 (or an exhausted 429/5xx) → HTTPError
-                # (permanent for this pass) → not caught below → per-item error.
-                resp.raise_for_status()
-                # 206 → server honored the Range; append after the kept bytes.
-                # Anything else (200) → it served the whole file → start clean.
-                mode = "ab" if (have > 0 and resp.status_code == 206) else "wb"
-                with open(dest, mode) as fh:
-                    for chunk in resp.iter_content(chunk_size=_CHUNK):
-                        if chunk:
-                            fh.write(chunk)
-                return
-            except _TRANSIENT_TRANSPORT_EXC as exc:
-                if attempt >= _MAX_MEDIA_RETRIES:
-                    raise  # exhausted → terminal error outcome
-                attempt += 1
-                delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS)
-                log.warning(
-                    "Patreon media transport error (%s) — backing off %.1fs "
-                    "(retry %d/%d): %s",
-                    url, delay, attempt, _MAX_MEDIA_RETRIES, exc,
-                )
-                time.sleep(delay)
+    # -- video (Mux/HLS via yt-dlp) ----------------------------------------
+    # The plain-GET streaming path (_fetch_get / _fetch_to_file) and
+    # _validate_path are inherited from BaseNativeDownloader.
 
     def _run_ytdlp(self, url: str, dest: Path, headers: dict) -> Path | None:
         """Invoke yt-dlp to fetch a Mux/HLS stream to (around) `dest`.
@@ -495,35 +316,6 @@ class PatreonDownloader:
                 return cand
         return None
 
-    # -- validation --------------------------------------------------------
-
-    def _validate_path(
-        self, path: Path, artist_slug: str, source_url: str | None = None
-    ) -> tuple[str | None, Path | None]:
-        """Validate a freshly-written file; quarantine if bad.
-
-        Uses the shared `file_validator.quarantine_file` — same move + provenance
-        sidecar gallery-dl writes (the native path used to skip the sidecar; that
-        parity gap is closed here). Returns `(reason, quarantine_dest)` when
-        quarantined (dest is the original path if the move itself failed), else
-        `(None, None)` (ok / not validatable / disabled). plan #704: the dest is
-        surfaced so the run reports a real quarantined-paths list.
-        """
-        if not self._validate or not is_validatable(path):
-            return None, None
-        try:
-            result = validate_file(path)
-        except Exception as exc:
-            log.warning("Validator raised on %s: %s", path, exc)
-            return None, None
-        if result.ok:
-            return None, None
-        dest = quarantine_file(
-            self.images_root, path, artist_slug, "patreon",
-            url=source_url, result=result,
-        )
-        return (result.reason or "validation failed"), (dest or path)
-
     # -- sidecar -----------------------------------------------------------
 
     def _write_sidecar(
@@ -614,7 +406,7 @@ class PatreonDownloader:
             return PostRecordOutcome(
                 path=None, post_type=post_type, title=title, body_chars=0,
             )
-        post_dir = self.images_root / artist_slug / "patreon" / _post_dir_name(post)
+        post_dir = self.images_root / artist_slug / "patreon" / post_dir_name(post)
         post_dir.mkdir(parents=True, exist_ok=True)
         path = self._write_sidecar_data(post, post_dir / "_post.json")
         # _write_sidecar_data has by now memoized any detail-fetched body onto

backend/app/services/patreon_ingester.py

@@ -35,15 +35,8 @@ from collections.abc import Callable
 from pathlib import Path
 
 from ..models import PatreonFailedMedia, PatreonSeenMedia
-from .gallery_dl import DownloadResult, ErrorType
 from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester
-from .patreon_client import (
-    MediaItem,
-    PatreonAPIError,
-    PatreonAuthError,
-    PatreonClient,
-    PatreonDriftError,
-)
+from .patreon_client import MediaItem, PatreonAPIError, PatreonClient
 from .patreon_downloader import PatreonDownloader
 from .patreon_resolver import extract_vanity, resolve_campaign_id_for_source
 
@@ -129,51 +122,11 @@ class PatreonIngester(Ingester):
             ledger_key=_ledger_key,
             platform="patreon",
             error_base=PatreonAPIError,
+            # API_DRIFT message phrasing; the base Ingester._failure_result owns
+            # the auth/drift/HTTP→error_type mapping now (shared across platforms).
+            drift_label="Patreon API",
         )
 
-    # -- failure mapping (Patreon exception taxonomy) ----------------------
-
-    def _failure_result(self, exc: Exception, _result) -> DownloadResult:
-        """Map a client-level exception to a loud, typed failed DownloadResult.
-
-        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. The typed mapping lets FailingSourcesCard render the right chip
-        and tells the operator what to do:
-          - PatreonAuthError  → AUTH_ERROR  (rotate cookies)
-          - PatreonDriftError → API_DRIFT   (ingester field-set/parser needs update)
-          - HTTP 429 / 404    → RATE_LIMITED / NOT_FOUND
-          - other HTTP status → HTTP_ERROR;  transport failure → NETWORK_ERROR
-
-        PatreonAuthError and PatreonDriftError both subclass PatreonAPIError, so
-        they must be matched before the generic HTTP/transport fallthrough.
-        """
-        message = str(exc)
-        if isinstance(exc, PatreonAuthError):
-            error_type = ErrorType.AUTH_ERROR
-        elif isinstance(exc, PatreonDriftError):
-            error_type = ErrorType.API_DRIFT
-            message = f"Patreon API changed — ingester needs update: {message}"
-        else:  # generic PatreonAPIError: HTTP non-2xx (status_code set) or transport
-            status = getattr(exc, "status_code", None)
-            if status == 429:
-                error_type = ErrorType.RATE_LIMITED
-            elif status == 404:
-                error_type = ErrorType.NOT_FOUND
-            elif status is not None:
-                error_type = ErrorType.HTTP_ERROR
-            else:
-                error_type = ErrorType.NETWORK_ERROR
-        log.warning("Patreon ingest failed (%s): %s", error_type.value, message)
-        result = _result(
-            success=False, return_code=1,
-            error_type=error_type, error_message=message,
-        )
-        # plan #708 B1: carry the server's Retry-After up to the cooldown.
-        if error_type == ErrorType.RATE_LIMITED:
-            result.retry_after_seconds = getattr(exc, "retry_after", None)
-        return result
-
 
 async def verify_patreon_credential(
     url: str,

backend/app/services/platform_lock.py

@@ -21,9 +21,10 @@ from ..config import get_config
 log = logging.getLogger(__name__)
 
 # Platforms walked one-at-a-time. gallery-dl platforms are intentionally NOT
-# here: each runs as a self-pacing subprocess and they're lower-volume. Add a
-# platform here to cap it to a single concurrent walk.
-SERIALIZED_PLATFORMS = frozenset({"patreon"})
+# here: each runs as a self-pacing subprocess and they're lower-volume. The
+# native-ingester platforms are serialized (one paced scrape/API walk at a time).
+# Add a platform here to cap it to a single concurrent walk.
+SERIALIZED_PLATFORMS = frozenset({"patreon", "subscribestar"})
 
 _LOCK_PREFIX = "fc:download_lock:"
 

backend/app/services/subscribestar_client.py

@@ -0,0 +1,418 @@
+"""Native SubscribeStar client — the SubscribeStar adapter's read path.
+
+Unlike Patreon (a clean JSON:API), SubscribeStar has NO public API: gallery-dl
+scrapes its HTML, and so do we. The platform-agnostic core (`ingest_core`) only
+calls `client.iter_posts` / `extract_media` / the optional seams, so the
+HTML-scrape divergence is contained entirely to this module.
+
+Feed shape (characterized from a live sample 2026-06-17 — Scribe note
+"SubscribeStar HTML characterization"):
+  - Page 1 is the creator page HTML: GET <base>/<slug>.
+  - Each page embeds `data-role="infinite_scroll-next_page" href="/posts?...&page=N
+    &slug=<slug>&sort_by=newest"`. That path returns JSON {"html": "<...posts...>"};
+    the fragment carries the NEXT page link, until it runs out.
+  - A post is `<div class="post is-shown ..." data-id="<post_id>" ...>`; body lives
+    in `.post-content .trix-content`; date in `.post-date`; media in a `data-gallery`
+    JSON manifest on `.uploads-images`.
+
+`campaign_id` for SubscribeStar is the full creator URL (e.g.
+https://subscribestar.adult/sabu) — the host (.com vs .adult) and slug both come
+from it, so no separate resolver is needed.
+
+Drift is loud on purpose (mirrors patreon_client): an HTML login/age-gate where
+the feed was expected is AUTH (rotate cookies); a feed whose post structure we
+can't parse at all is DRIFT (the scraper needs updating). FC runs on a
+plain-HTTP homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import time
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime
+from html import unescape
+from pathlib import Path
+from urllib.parse import urljoin, urlsplit
+
+import requests
+
+from ..utils.paths import filehash_from_url
+from .native_ingest_common import (
+    _MAX_429_RETRIES,
+    NativeAuthError,
+    NativeDriftError,
+    NativeIngestError,
+    basename_from_url,
+    make_session,
+    retry_after_seconds,
+)
+
+log = logging.getLogger(__name__)
+
+_TIMEOUT_SECONDS = 30.0
+# Accept header for the HTML feed + JSON "load more" + the XHR marker.
+_ACCEPT = "text/html,application/xhtml+xml,application/json,*/*"
+_XHR_HEADERS = {"X-Requested-With": "XMLHttpRequest"}
+
+# A post block opens with this wrapper; we slice the page between consecutive
+# occurrences (regex can't match the balanced close of nested divs, so chunk-
+# per-post is the robust approach gallery-dl also uses).
+_POST_OPEN = '<div class="post is-shown'
+_POST_ID_RE = re.compile(r'data-id="(\d+)"')
+_POST_DATE_RE = re.compile(r'<div class="post-date">([^<]+)</div>')
+# The body: inner of `.post-content .trix-content`. Non-greedy to the LAST
+# </div></div> in the chunk would over-capture; instead capture from trix-content
+# to the close of the post-content wrapper, which the post-body block ends with.
+_BODY_RE = re.compile(
+    r'<div class="post-content"[^>]*>(.*?)</div>\s*</div>\s*</div>', re.DOTALL
+)
+_GALLERY_RE = re.compile(r'data-gallery="([^"]*)"')
+_NEXT_PAGE_RE = re.compile(
+    r'data-role="infinite_scroll-next_page"\s+href="([^"]+)"'
+)
+# Signals an auth/age wall served in place of the feed (cookies expired or the
+# age cookie missing) rather than a real — possibly empty — creator feed.
+_LOGIN_MARKERS = ("/session/new", 'data-role="sign_in"', "age_confirmation_warning")
+
+
+class SubscribeStarAPIError(NativeIngestError):
+    """Base for native SubscribeStar client failures. status_code / retry_after
+    are inherited from NativeIngestError."""
+
+
+class SubscribeStarAuthError(SubscribeStarAPIError, NativeAuthError):
+    """Auth/authorization failure — expired cookies, missing age cookie, or an
+    HTML login/age wall served where the feed was expected. Fix = rotate the
+    credential, not update the scraper. Maps to error_type 'auth_error'."""
+
+
+class SubscribeStarDriftError(SubscribeStarAPIError, NativeDriftError):
+    """The feed HTML did not match the structure we scrape (no recognizable post
+    blocks AND no known empty-feed state). The scrape analog of API drift — fail
+    loud so the import step flags 'SubscribeStar changed its markup' instead of
+    silently importing nothing."""
+
+
+@dataclass
+class MediaItem:
+    """One resolved downloadable item belonging to a SubscribeStar post.
+
+    Fields mirror patreon_client.MediaItem (so the downloader is structurally the
+    same) plus `media_id` — the stable per-upload gallery id. SubscribeStar's
+    full-res URL is an opaque `/post_uploads?payload=...` (not content-addressed),
+    so `filehash` is usually None and the ledger keys on `<post_id>:<media_id>`.
+    """
+
+    url: str
+    filename: str
+    kind: str
+    filehash: str | None
+    post_id: str
+    media_id: str
+
+
+def _split_creator_url(campaign_id: str) -> tuple[str, str]:
+    """`campaign_id` is the creator URL → (base, slug).
+
+    base = scheme://host (preserving .com vs .adult); slug = first path segment.
+    """
+    parts = urlsplit(campaign_id)
+    base = f"{parts.scheme or 'https'}://{parts.netloc}"
+    slug = parts.path.strip("/").split("/")[0] if parts.path else ""
+    return base, slug
+
+
+def _parse_ss_datetime(text: str) -> str | None:
+    """SubscribeStar renders human dates: 'Jun 17, 2026 03:19 am' (and an
+    'Updated on <date>' variant). Return ISO-8601 (UTC-naive) or None."""
+    s = unescape(text or "").strip()
+    if s.lower().startswith("updated on "):
+        s = s[len("updated on "):].strip()
+    for fmt in ("%b %d, %Y %I:%M %p", "%b %d, %Y"):
+        try:
+            return datetime.strptime(s, fmt).isoformat()
+        except ValueError:
+            continue
+    return None
+
+
+class SubscribeStarClient:
+    """Synchronous SubscribeStar HTML-scrape read client. Construct with a path
+    to a Netscape cookies.txt (the same file CredentialService.get_cookies_path
+    materializes, already carrying the age cookie via augment_cookies)."""
+
+    def __init__(
+        self,
+        cookies_path: str | Path | None,
+        *,
+        request_sleep: float = 0.0,
+        max_retries: int = _MAX_429_RETRIES,
+    ):
+        self.cookies_path = str(cookies_path) if cookies_path else None
+        self._session = make_session(
+            cookies_path, accept=_ACCEPT, extra_headers=_XHR_HEADERS
+        )
+        self._request_sleep = request_sleep or 0.0
+        self._max_retries = max_retries
+
+    # -- request -----------------------------------------------------------
+
+    def _get(self, url: str) -> requests.Response:
+        if self._request_sleep > 0:
+            time.sleep(self._request_sleep)
+        attempt = 0
+        while True:
+            try:
+                resp = self._session.get(url, timeout=_TIMEOUT_SECONDS)
+            except requests.RequestException as exc:
+                raise SubscribeStarAPIError(
+                    f"SubscribeStar request failed ({url}): {exc}"
+                ) from exc
+            if resp.status_code == 429 and attempt < self._max_retries:
+                attempt += 1
+                delay = retry_after_seconds(resp, attempt)
+                log.warning(
+                    "SubscribeStar 429 (%s) — backing off %.1fs (retry %d/%d)",
+                    url, delay, attempt, self._max_retries,
+                )
+                time.sleep(delay)
+                continue
+            break
+        if resp.status_code in (401, 403):
+            raise SubscribeStarAuthError(
+                f"SubscribeStar returned HTTP {resp.status_code} — auth rejected "
+                f"(cookies expired or tier insufficient; {url})",
+                status_code=resp.status_code,
+            )
+        if resp.status_code != 200:
+            retry_after = None
+            if resp.status_code == 429:
+                hdr = resp.headers.get("Retry-After")
+                if hdr:
+                    try:
+                        retry_after = float(hdr)
+                    except (TypeError, ValueError):
+                        retry_after = None
+            raise SubscribeStarAPIError(
+                f"SubscribeStar returned HTTP {resp.status_code} ({url})",
+                status_code=resp.status_code,
+                retry_after=retry_after,
+            )
+        return resp
+
+    def _feed_html(self, url: str) -> str:
+        """Page 1: the creator page (full HTML document)."""
+        resp = self._get(url)
+        text = resp.text or ""
+        if any(m in text for m in _LOGIN_MARKERS) and _POST_OPEN not in text:
+            raise SubscribeStarAuthError(
+                f"SubscribeStar served a login/age wall instead of the feed "
+                f"(cookies expired or age cookie missing; {url})"
+            )
+        return text
+
+    def _loadmore_html(self, url: str) -> str:
+        """Subsequent pages: GET returns JSON {"html": "<fragment>"}."""
+        resp = self._get(url)
+        try:
+            payload = resp.json()
+        except ValueError as exc:
+            raise SubscribeStarAuthError(
+                f"SubscribeStar 'load more' returned non-JSON (session expired?; "
+                f"{url}): {exc}"
+            ) from exc
+        html = payload.get("html") if isinstance(payload, dict) else None
+        return html if isinstance(html, str) else ""
+
+    # -- parsing -----------------------------------------------------------
+
+    @staticmethod
+    def _post_chunks(html: str) -> list[str]:
+        """Slice the page into one chunk per post (between consecutive wrapper
+        opens). Regex can't match a post's balanced close, so each chunk runs to
+        the next post's open (or end of fragment) — enough to scope per-post
+        field extraction."""
+        starts = [m.start() for m in re.finditer(re.escape(_POST_OPEN), html)]
+        chunks = []
+        for i, start in enumerate(starts):
+            end = starts[i + 1] if i + 1 < len(starts) else len(html)
+            chunks.append(html[start:end])
+        return chunks
+
+    def _parse_post(self, chunk: str) -> dict | None:
+        m = _POST_ID_RE.search(chunk)
+        if not m:
+            return None
+        post_id = m.group(1)
+        date_m = _POST_DATE_RE.search(chunk)
+        published = _parse_ss_datetime(date_m.group(1)) if date_m else None
+        body_m = _BODY_RE.search(chunk)
+        content = body_m.group(1).strip() if body_m else ""
+        return {
+            "id": post_id,
+            "attributes": {
+                # SubscribeStar has no title field; the importer synthesizes a
+                # display title from the body's first line (sidecar util).
+                "title": "",
+                "content": content,
+                "published_at": published,
+                "post_type": "subscribestar",
+            },
+            # Raw chunk retained so extract_media parses the data-gallery manifest
+            # and post_is_gated can scan for the locked-teaser marker.
+            "_html": chunk,
+        }
+
+    def _parse_posts(self, html: str) -> list[dict]:
+        posts = []
+        for chunk in self._post_chunks(html):
+            post = self._parse_post(chunk)
+            if post is not None:
+                posts.append(post)
+        return posts
+
+    @staticmethod
+    def _next_page_href(html: str) -> str | None:
+        m = _NEXT_PAGE_RE.search(html)
+        return unescape(m.group(1)) if m else None
+
+    def extract_media(self, post: dict, included_index: dict) -> list[MediaItem]:
+        """Resolve downloadable media from the post's `data-gallery` JSON
+        manifest. Each item carries a stable `id`, `original_filename`, `type`,
+        and a full-res `url` (`/post_uploads?payload=...`, relative — joined to the
+        post's base). `included_index` is unused (HTML carries media inline)."""
+        chunk = post.get("_html") or ""
+        base = post.get("_base") or "https://www.subscribestar.com"
+        post_id = str(post.get("id") or "")
+        items: list[MediaItem] = []
+        for gm in _GALLERY_RE.finditer(chunk):
+            try:
+                gallery = json.loads(unescape(gm.group(1)))
+            except (ValueError, TypeError):
+                continue
+            if not isinstance(gallery, list):
+                continue
+            for it in gallery:
+                if not isinstance(it, dict):
+                    continue
+                rel = it.get("url")
+                if not isinstance(rel, str) or not rel:
+                    continue
+                url = urljoin(base + "/", rel)
+                media_id = str(it.get("id") or "")
+                name = it.get("original_filename")
+                filename = name if isinstance(name, str) and name else basename_from_url(url)
+                items.append(
+                    MediaItem(
+                        url=url,
+                        filename=filename,
+                        kind=str(it.get("type") or "image"),
+                        filehash=filehash_from_url(url),
+                        post_id=post_id,
+                        media_id=media_id,
+                    )
+                )
+        return items
+
+    @staticmethod
+    def post_meta(post: dict) -> dict:
+        """Title + date for the preview sample. Title is synthesized from the body
+        (SubscribeStar has no title field)."""
+        attrs = post.get("attributes") or {}
+        return {"title": None, "date": attrs.get("published_at")}
+
+    @staticmethod
+    def post_is_gated(post: dict) -> bool:
+        """True when the subscriber cannot view this post (locked teaser). #874
+        "no stub for gated content": the core skips it ENTIRELY (no media, no
+        post-record), so we don't replicate a hollow teaser in Curator.
+
+        SubscribeStar does NOT expose downloadable preview media for locked posts
+        (no `data-gallery`), so the Patreon "downloaded blurred junk" failure
+        can't happen here — this gate only prevents capturing an empty teaser
+        stub. Detect the locked marker conservatively (default to NOT gated when
+        absent, so we never over-filter an accessible text post). The exact marker
+        is being confirmed against a live locked sample; until then we gate only on
+        the explicit lock-overlay class SubscribeStar renders on a paywalled post.
+        """
+        chunk = post.get("_html") or ""
+        return 'class="post-content-locked"' in chunk or "for-locked_content" in chunk
+
+    @staticmethod
+    def post_record_key(post: dict) -> tuple[str, str] | None:
+        """`(ledger_key, post_id)` for a post's seen-ledger entry (the `post:<id>`
+        synthetic key gates post-record capture through the same ledger as media),
+        or None when the post has no id."""
+        pid = post.get("id")
+        pid = str(pid) if pid is not None else ""
+        if not pid:
+            return None
+        return (f"post:{pid}", pid)
+
+    # -- iteration ---------------------------------------------------------
+
+    def iter_posts(
+        self, campaign_id: str, cursor: str | None = None
+    ) -> Iterator[tuple[dict, dict, str | None]]:
+        """Yield (post, {}, page_cursor) for every post in the feed.
+
+        `campaign_id` is the creator URL. `cursor` is the relative "load more"
+        href that fetches a page (None → page 1, the creator page HTML). The
+        yielded `page_cursor` is the href that FETCHED this post's page, so the
+        core checkpoints a value that re-fetches the same page on resume (matching
+        the Patreon cursor contract).
+        """
+        base, slug = _split_creator_url(campaign_id)
+        if not slug:
+            raise SubscribeStarDriftError(
+                f"Could not extract a creator slug from {campaign_id!r}"
+            )
+        current = cursor
+        first_page = True
+        while True:
+            page_cursor = current
+            if current is None:
+                html = self._feed_html(f"{base}/{slug}")
+            else:
+                html = self._loadmore_html(urljoin(base + "/", current))
+            posts = self._parse_posts(html)
+            if first_page and not posts and _POST_OPEN not in html:
+                # Page 1 with no recognizable post wrappers at all: either a brand
+                # new creator with zero posts, or our scraper is stale. The core's
+                # body canary catches systematic emptiness across a populated feed;
+                # here we only raise if the feed container itself is missing.
+                if 'data-role="posts_container-list"' not in html:
+                    raise SubscribeStarDriftError(
+                        f"SubscribeStar feed for {slug!r} had no posts and no "
+                        "recognizable feed container — markup changed?"
+                    )
+            for post in posts:
+                post["_base"] = base
+                yield post, {}, page_cursor
+            next_href = self._next_page_href(html)
+            if not next_href:
+                return
+            current = next_href
+            first_page = False
+
+    # -- verify ------------------------------------------------------------
+
+    def verify_auth(self, campaign_id: str) -> tuple[bool | None, str]:
+        """Cheap auth probe: fetch the first feed page and report whether the
+        credential authenticated, without downloading anything."""
+        base, slug = _split_creator_url(campaign_id)
+        if not slug:
+            return None, f"Couldn't parse a SubscribeStar creator from {campaign_id!r}"
+        try:
+            html = self._feed_html(f"{base}/{slug}")
+        except SubscribeStarAuthError as exc:
+            return False, f"SubscribeStar rejected the credential — {exc}"
+        except SubscribeStarAPIError as exc:
+            return None, f"Couldn't verify (network/HTTP issue): {exc}"
+        if _POST_OPEN in html or 'data-role="posts_container-list"' in html:
+            return True, "Credentials valid — the SubscribeStar feed loaded."
+        return None, "Couldn't verify — SubscribeStar feed shape unrecognized."

backend/app/services/subscribestar_downloader.py

@@ -0,0 +1,206 @@
+"""Native SubscribeStar media downloader — the SubscribeStar counterpart to
+patreon_downloader.
+
+Given a SubscribeStar post and its resolved `MediaItem`s
+(subscribestar_client.extract_media), download the media to gallery-dl's on-disk
+layout (so existing gallery-dl downloads are recognized on disk and not
+re-fetched at cutover), write the post-first sidecars the importer consumes, and
+report per-media outcomes.
+
+Simpler than the Patreon downloader: SubscribeStar serves every upload as a
+direct file via `/post_uploads?payload=...` (plain GET — no Mux/HLS, no yt-dlp),
+and the full post body is already present in the feed HTML (no detail-endpoint
+enrichment). PURE: no DB; the seen-skip is an injected predicate.
+
+On-disk layout (matches gallery-dl's subscribestar config
+`{date:%Y-%m-%d}_{id}_{title[:40]}` / `{num:>02}_{filename}`): SubscribeStar posts
+have no title, so the directory is `<YYYY-MM-DD>_<post_id>_` — the display title is
+synthesized by the importer from the body, so the bare dir name is on-disk only.
+
+FC runs on a plain-HTTP homelab; nothing here uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from collections.abc import Callable
+from pathlib import Path
+
+import requests
+
+from .native_ingest_common import (
+    BaseNativeDownloader,
+    MediaOutcome,
+    PostRecordOutcome,
+    post_dir_name,
+    sanitize_segment,
+)
+
+log = logging.getLogger(__name__)
+
+
+class SubscribeStarDownloader(BaseNativeDownloader):
+    """Download resolved SubscribeStar media to gallery-dl's on-disk layout.
+    Subclasses BaseNativeDownloader for the shared streaming GET (transient-retry +
+    Range-resume) and validation/quarantine. No video branch (SubscribeStar serves
+    files directly via /post_uploads) and no detail-fetch (the body is already in
+    the feed HTML). PURE: no DB."""
+
+    def __init__(
+        self,
+        images_root: Path,
+        cookies_path: str | None = None,
+        *,
+        validate: bool = True,
+        rate_limit: float = 0.0,
+        session: requests.Session | None = None,
+    ):
+        super().__init__(
+            images_root, cookies_path, platform="subscribestar",
+            validate=validate, rate_limit=rate_limit, session=session,
+        )
+
+    # -- public ------------------------------------------------------------
+
+    def download_post(
+        self,
+        post: dict,
+        media_items: list,
+        artist_slug: str,
+        *,
+        is_seen: Callable[[object], bool] = lambda m: False,
+        should_stop: Callable[[], bool] = lambda: False,
+        recapture: bool = False,
+    ) -> list[MediaOutcome]:
+        """Download every media item of one post; return per-item outcomes.
+        Mirrors PatreonDownloader.download_post (two-tier skip, mid-post time-box,
+        recapture surfacing) minus the video branch."""
+        post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post)
+        outcomes: list[MediaOutcome] = []
+        for i, media in enumerate(media_items, start=1):
+            if should_stop():
+                break
+            try:
+                outcomes.append(
+                    self._download_one(
+                        post, media, post_dir, artist_slug, i, is_seen,
+                        recapture=recapture,
+                    )
+                )
+            except Exception as exc:  # resilient: isolate one item's failure
+                log.warning(
+                    "SubscribeStar media failed (post %s, item %d): %s",
+                    post.get("id"), i, exc,
+                )
+                outcomes.append(
+                    MediaOutcome(media=media, status="error", path=None, error=str(exc))
+                )
+        return outcomes
+
+    # -- per-item ----------------------------------------------------------
+
+    def _download_one(
+        self,
+        post: dict,
+        media,
+        post_dir: Path,
+        artist_slug: str,
+        index: int,
+        is_seen: Callable[[object], bool],
+        *,
+        recapture: bool = False,
+    ) -> MediaOutcome:
+        seen = is_seen(media)
+        if seen and not recapture:
+            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
+
+        nn = f"{index:02d}"
+        media_path = post_dir / sanitize_segment(f"{nn}_{media.filename}")
+
+        if media_path.exists():  # tier-2: already on disk
+            return MediaOutcome(
+                media=media, status="skipped_disk", path=media_path, error=None
+            )
+        # recapture: a seen item not on disk is NOT re-downloaded (recovery's job).
+        if seen:
+            return MediaOutcome(media=media, status="skipped_seen", path=None, error=None)
+
+        post_dir.mkdir(parents=True, exist_ok=True)
+        if self._rate_limit > 0:
+            time.sleep(self._rate_limit)
+
+        out_path = self._fetch_get(media.url, media_path)
+        reason, quarantine_dest = self._validate_path(out_path, artist_slug, media.url)
+        if reason is not None:
+            return MediaOutcome(
+                media=media, status="quarantined", path=quarantine_dest, error=reason,
+            )
+        self._write_sidecar(post, out_path, source_url=media.url)
+        return MediaOutcome(media=media, status="downloaded", path=out_path, error=None)
+
+    # The plain-GET streaming path (_fetch_get / _fetch_to_file) and
+    # _validate_path are inherited from BaseNativeDownloader.
+
+    # -- sidecar -----------------------------------------------------------
+
+    def _write_sidecar(
+        self, post: dict, media_path: Path, *, source_url: str | None = None
+    ) -> Path:
+        """Per-media sidecar — post-first (#856): image identity ONLY
+        (category/id/source_url). The post body/links live solely in _post.json."""
+        return self._write_sidecar_data(
+            post, media_path.with_suffix(".json"), source_url=source_url, minimal=True,
+        )
+
+    def _write_sidecar_data(
+        self, post: dict, sidecar_path: Path, *, source_url: str | None = None,
+        minimal: bool = False,
+    ) -> Path:
+        """Serialize the post's metadata. minimal=True → per-media sidecar (image
+        identity only); else the full post record (body/title/date/url)."""
+        if minimal:
+            data = {"category": "subscribestar", "id": str(post.get("id") or "")}
+            if source_url:
+                data["source_url"] = source_url
+            sidecar_path.write_text(json.dumps(data, indent=2))
+            return sidecar_path
+        attrs = post.get("attributes") or {}
+        content = attrs.get("content")
+        # SubscribeStar synthesizes the post permalink from the id (matches the
+        # platforms/subscribestar.py derive_post_url helper).
+        pid = str(post.get("id") or "")
+        data = {
+            "category": "subscribestar",
+            "id": pid,
+            "post_id": pid,  # ensures derive_post_url has its key on the native path
+            "title": attrs.get("title") if isinstance(attrs.get("title"), str) else "",
+            "content": content if isinstance(content, str) else "",
+            "published_at": attrs.get("published_at"),
+        }
+        if source_url:
+            data["source_url"] = source_url
+        sidecar_path.write_text(json.dumps(data, indent=2))
+        return sidecar_path
+
+    def write_post_record(self, post: dict, artist_slug: str) -> PostRecordOutcome:
+        """Write the post-first `_post.json` (body/links/metadata) — the sole
+        writer of the post record on the native path. SubscribeStar's body is
+        already in the feed HTML, so no detail-fetch is needed."""
+        attrs = post.get("attributes") or {}
+        title = attrs.get("title") if isinstance(attrs.get("title"), str) else None
+        post_type = attrs.get("post_type") if isinstance(attrs.get("post_type"), str) else None
+        pid = str(post.get("id") or "")
+        if not pid:
+            return PostRecordOutcome(
+                path=None, post_type=post_type, title=title, body_chars=0,
+            )
+        post_dir = self.images_root / artist_slug / "subscribestar" / post_dir_name(post)
+        post_dir.mkdir(parents=True, exist_ok=True)
+        path = self._write_sidecar_data(post, post_dir / "_post.json")
+        body = attrs.get("content")
+        body_chars = len(body) if isinstance(body, str) else 0
+        return PostRecordOutcome(
+            path=path, post_type=post_type, title=title, body_chars=body_chars,
+        )

backend/app/services/subscribestar_ingester.py

@@ -0,0 +1,109 @@
+"""Native SubscribeStar ingester — the SubscribeStar ADAPTER over the
+platform-agnostic core (`ingest_core.Ingester`).
+
+Thin counterpart to patreon_ingester: wires the SubscribeStar client/downloader/
+ledger models/constraints/key into the core and supplies the SubscribeStar
+failure mapping. The three modes (tick / backfill / recovery / recapture), the
+seen + dead-letter ledgers, cursor checkpointing, and the post-first capture all
+live in the core — identical to Patreon. `download_service.download_source`
+drives `SubscribeStarIngester.run` exactly as it drives the Patreon one.
+
+`campaign_id` is the creator URL (the client derives host + slug from it), so no
+campaign-id resolver is needed. FC runs on a plain-HTTP homelab; nothing here
+uses a secure-context Web API.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Callable
+from pathlib import Path
+
+from ..models import SubscribeStarFailedMedia, SubscribeStarSeenMedia
+from .ingest_core import DEAD_LETTER_THRESHOLD, Ingester
+from .subscribestar_client import MediaItem, SubscribeStarAPIError, SubscribeStarClient
+from .subscribestar_downloader import SubscribeStarDownloader
+
+__all__ = [
+    "DEAD_LETTER_THRESHOLD",
+    "SubscribeStarIngester",
+    "_ledger_key",
+    "verify_subscribestar_credential",
+]
+
+log = logging.getLogger(__name__)
+
+_LEDGER_KEY_MAX = 128
+
+
+def _ledger_key(media: MediaItem) -> str:
+    """Stable per-media identity for the cross-run seen-ledger. SubscribeStar's
+    full-res URL is an opaque `/post_uploads?payload=...` (no content hash), so
+    `media.filehash` is normally None and the stable proxy is the gallery item id
+    scoped to its post: `<post_id>:<media_id>`. Bounded to the column width."""
+    if media.filehash:
+        return media.filehash
+    return f"{media.post_id}:{media.media_id}"[:_LEDGER_KEY_MAX]
+
+
+class SubscribeStarIngester(Ingester):
+    """Walk a SubscribeStar creator's posts, download unseen media, return a
+    `DownloadResult`. A thin adapter over `ingest_core.Ingester`; `client` /
+    `downloader` are injectable seams so unit tests run without network."""
+
+    def __init__(
+        self,
+        images_root: Path,
+        cookies_path: str | None,
+        session_factory: Callable[[], object],
+        *,
+        validate: bool = True,
+        rate_limit: float = 0.0,
+        request_sleep: float = 0.0,
+        client: SubscribeStarClient | None = None,
+        downloader: SubscribeStarDownloader | None = None,
+    ):
+        self.images_root = Path(images_root)
+        self.cookies_path = str(cookies_path) if cookies_path else None
+        resolved_client = (
+            client
+            if client is not None
+            else SubscribeStarClient(cookies_path, request_sleep=request_sleep)
+        )
+        resolved_downloader = (
+            downloader
+            if downloader is not None
+            else SubscribeStarDownloader(
+                self.images_root, cookies_path, validate=validate, rate_limit=rate_limit,
+            )
+        )
+        super().__init__(
+            client=resolved_client,
+            downloader=resolved_downloader,
+            session_factory=session_factory,
+            seen_model=SubscribeStarSeenMedia,
+            failed_model=SubscribeStarFailedMedia,
+            seen_constraint="uq_subscribestar_seen_media_source_id",
+            failed_constraint="uq_subscribestar_failed_media_source_id",
+            ledger_key=_ledger_key,
+            platform="subscribestar",
+            error_base=SubscribeStarAPIError,
+            # API_DRIFT message phrasing; the base Ingester._failure_result owns
+            # the auth/drift/HTTP→error_type mapping (shared across platforms).
+            drift_label="SubscribeStar markup",
+        )
+
+
+async def verify_subscribestar_credential(
+    url: str,
+    cookies_path: str | None,
+    overrides: dict | None,
+) -> tuple[bool | None, str]:
+    """Native SubscribeStar credential probe — fetches ONE feed page via
+    SubscribeStarClient.verify_auth. `campaign_id` is just the creator URL (no
+    resolver). Returns the uniform `(ok, message)` contract so
+    download_backends.verify_credential treats it like the gallery-dl probe."""
+    client = SubscribeStarClient(cookies_path)
+    loop = asyncio.get_running_loop()
+    return await loop.run_in_executor(None, client.verify_auth, url)

frontend/src/components/subscriptions/SourceActions.vue

@@ -28,7 +28,7 @@
 
     <KebabMenu size="small" :min-width="260" label="More actions">
       <v-list-item
-        v-if="isPatreon && !running"
+        v-if="isNative && !running"
         prepend-icon="mdi-backup-restore"
         @click="emit('recover', source)"
       >
@@ -39,7 +39,7 @@
         </v-list-item-subtitle>
       </v-list-item>
       <v-list-item
-        v-if="isPatreon && !running"
+        v-if="isNative && !running"
         prepend-icon="mdi-text-box-search-outline"
         @click="emit('recapture', source)"
       >
@@ -49,7 +49,7 @@
           already on disk — without re-downloading media
         </v-list-item-subtitle>
       </v-list-item>
-      <v-divider v-if="isPatreon && !running" />
+      <v-divider v-if="isNative && !running" />
       <v-list-item
         base-color="error"
         prepend-icon="mdi-delete-outline"
@@ -74,7 +74,10 @@ const emit = defineEmits(['check', 'backfill', 'recover', 'recapture', 'remove']
 const running = computed(() => props.source.backfill_state === 'running')
 const recovering = computed(() => !!props.source.backfill_bypass_seen)
 const recapturing = computed(() => !!props.source.backfill_recapture)
-const isPatreon = computed(() => props.source.platform === 'patreon')
+// Recover / recapture are native-ingester features (ledger-bypass re-walk and
+// post-text re-grab), available to every native platform — not just Patreon.
+const NATIVE_PLATFORMS = ['patreon', 'subscribestar']
+const isNative = computed(() => NATIVE_PLATFORMS.includes(props.source.platform))
 </script>
 
 <style scoped>

tests/test_api_credentials.py

@@ -165,25 +165,26 @@ async def test_verify_runs_gallery_dl_and_stamps_on_success(client, db, monkeypa
         return (True, "Credentials valid — the feed authenticated.")
     monkeypatch.setattr(gdl_mod.GalleryDLService, "verify", _fake_verify)
 
+    # hentaifoundry is a gallery-dl platform (subscribestar migrated to native).
     await client.post("/api/credentials", json={
-        "platform": "subscribestar", "credential_type": "cookies", "data": _NETSCAPE,
+        "platform": "hentaifoundry", "credential_type": "cookies", "data": _NETSCAPE,
     })
     artist = Artist(name="Maewix", slug="maewix")
     db.add(artist)
     await db.flush()
     db.add(Source(
-        artist_id=artist.id, platform="subscribestar",
-        url="https://www.subscribestar.com/maewix", enabled=True, config_overrides={},
+        artist_id=artist.id, platform="hentaifoundry",
+        url="https://www.hentai-foundry.com/user/Maewix", enabled=True, config_overrides={},
     ))
     await db.commit()
 
-    resp = await client.post("/api/credentials/subscribestar/verify")
+    resp = await client.post("/api/credentials/hentaifoundry/verify")
     body = await resp.get_json()
     assert body["valid"] is True
     assert body["last_verified"] is not None
 
     # The stamp is persisted on the credential record.
-    rec = await (await client.get("/api/credentials/subscribestar")).get_json()
+    rec = await (await client.get("/api/credentials/hentaifoundry")).get_json()
     assert rec["last_verified"] is not None
 
 

tests/test_api_sources.py

@@ -430,8 +430,8 @@ async def test_gallery_dl_platform_arm_skips_pre_flight(client, artist, db, monk
     monkeypatch.setattr(db_mod, "verify_source_credential", _boom)
 
     src = Source(
-        artist_id=artist.id, platform="subscribestar",
-        url="https://www.subscribestar.com/x", enabled=True,
+        artist_id=artist.id, platform="hentaifoundry",
+        url="https://www.hentai-foundry.com/user/x", enabled=True,
     )
     db.add(src)
     await db.commit()

tests/test_download_backends.py

@@ -7,15 +7,16 @@ from backend.app.services.download_backends import (
 )
 
 
-def test_patreon_is_native():
-    assert uses_native_ingester("patreon") is True
-    assert "patreon" in NATIVE_INGESTER_PLATFORMS
+def test_native_platforms():
+    for platform in ("patreon", "subscribestar"):
+        assert uses_native_ingester(platform) is True
+        assert platform in NATIVE_INGESTER_PLATFORMS
 
 
 def test_gallery_dl_platforms_are_not_native():
-    # The five platforms still served by gallery-dl must NOT route to the
-    # native ingester — guards an accidental over-broad migration.
-    for platform in ("subscribestar", "hentaifoundry", "discord", "pixiv", "deviantart"):
+    # The platforms still served by gallery-dl must NOT route to the native
+    # ingester — guards an accidental over-broad migration.
+    for platform in ("hentaifoundry", "discord", "pixiv", "deviantart"):
         assert uses_native_ingester(platform) is False
 
 

tests/test_gallery_dl_service.py

@@ -259,7 +259,7 @@ def test_build_config_emits_tick_skip_value(gdl):
     scans once 20 contiguous archived items are seen."""
     from backend.app.services.gallery_dl import TICK_SKIP_VALUE
     cfg = gdl._build_config_for_source(
-        platform="subscribestar",
+        platform="hentaifoundry",
         source_config=SourceConfig(),
         artist_slug="alice",
         skip_value=TICK_SKIP_VALUE,
@@ -272,7 +272,7 @@ def test_build_config_emits_backfill_skip_value(gdl):
     full post history."""
     from backend.app.services.gallery_dl import BACKFILL_SKIP_VALUE
     cfg = gdl._build_config_for_source(
-        platform="subscribestar",
+        platform="hentaifoundry",
         source_config=SourceConfig(),
         artist_slug="alice",
         skip_value=BACKFILL_SKIP_VALUE,

tests/test_patreon_client.py

@@ -12,13 +12,13 @@ import pytest
 import requests
 
 from backend.app.services import patreon_client as pc_mod
+from backend.app.services.native_ingest_common import retry_after_seconds
 from backend.app.services.patreon_client import (
     MediaItem,
     PatreonAPIError,
     PatreonAuthError,
     PatreonClient,
     PatreonDriftError,
-    _retry_after_seconds,
     parse_cursor_from_url,
 )
 
@@ -288,19 +288,19 @@ def test_verify_auth_inconclusive_on_network(monkeypatch):
 
 
 def test_retry_after_seconds_honors_numeric_header():
-    assert _retry_after_seconds(_FakeResp(429, headers={"Retry-After": "7"}), 1) == 7.0
+    assert retry_after_seconds(_FakeResp(429, headers={"Retry-After": "7"}), 1) == 7.0
 
 
 def test_retry_after_seconds_exponential_without_header():
     resp = _FakeResp(429)
-    assert _retry_after_seconds(resp, 1) == 2.0
-    assert _retry_after_seconds(resp, 2) == 4.0
-    assert _retry_after_seconds(resp, 3) == 8.0
+    assert retry_after_seconds(resp, 1) == 2.0
+    assert retry_after_seconds(resp, 2) == 4.0
+    assert retry_after_seconds(resp, 3) == 8.0
 
 
 def test_retry_after_seconds_caps():
-    assert _retry_after_seconds(_FakeResp(429), 10) == 30.0
-    assert _retry_after_seconds(_FakeResp(429, headers={"Retry-After": "999"}), 1) == 30.0
+    assert retry_after_seconds(_FakeResp(429), 10) == 30.0
+    assert retry_after_seconds(_FakeResp(429, headers={"Retry-After": "999"}), 1) == 30.0
 
 
 def _client_with_sequence(monkeypatch, responses):

tests/test_patreon_downloader.py

@@ -14,13 +14,14 @@ import pytest
 import requests
 
 from backend.app.services import patreon_downloader as pd_mod
-from backend.app.services.patreon_client import MediaItem
-from backend.app.services.patreon_downloader import (
+from backend.app.services.native_ingest_common import (
     MediaOutcome,
-    PatreonDownloader,
     PostRecordOutcome,
-    _sanitize,
+    post_dir_name,
+    sanitize_segment,
 )
+from backend.app.services.patreon_client import MediaItem
+from backend.app.services.patreon_downloader import PatreonDownloader
 from backend.app.utils.sidecar import find_sidecar
 
 # Minimal valid PNG so the file_validator passes on the happy path.
@@ -414,10 +415,10 @@ def test_validation_off_keeps_bytes(tmp_path):
 
 
 def test_sanitize_helper():
-    assert _sanitize('a/b') == "a_b"
-    assert _sanitize('x<>:"|?*y') == "x_______y"
-    assert _sanitize("trail.  ") == "trail"
-    assert _sanitize("") == "_"
+    assert sanitize_segment('a/b') == "a_b"
+    assert sanitize_segment('x<>:"|?*y') == "x_______y"
+    assert sanitize_segment("trail.  ") == "trail"
+    assert sanitize_segment("") == "_"
 
 
 def test_media_outcome_shape():
@@ -613,7 +614,7 @@ def test_media_sidecar_is_minimal_post_first(tmp_path):
     post["attributes"]["content"] = ""  # even an empty feed body isn't fetched here
     dl.download_post(post, [_img("a.png"), _img("b.png")], "artist-x")
 
-    post_dir = tmp_path / "artist-x" / "patreon" / pd_mod._post_dir_name(post)
+    post_dir = tmp_path / "artist-x" / "patreon" / post_dir_name(post)
     sc = find_sidecar(post_dir / "01_a.png")
     assert sc is not None
     data = json.loads(sc.read_text())

tests/test_platform_lock.py

@@ -9,8 +9,14 @@ pytestmark = pytest.mark.integration
 
 def test_non_serialized_platform_has_no_lock():
     # gallery-dl platforms aren't capped — they get no lock at all.
-    assert platform_lock("subscribestar", ttl_seconds=60) is None
     assert platform_lock("deviantart", ttl_seconds=60) is None
+    assert platform_lock("hentaifoundry", ttl_seconds=60) is None
+
+
+def test_subscribestar_is_serialized():
+    # subscribestar is a native-ingester platform now — one paced walk at a time.
+    lock = platform_lock("subscribestar", ttl_seconds=60)
+    assert lock is not None
 
 
 def test_patreon_serializes_to_one_holder():

tests/test_subscribestar_native.py

@@ -0,0 +1,292 @@
+"""Unit tests for the native SubscribeStar client / downloader / ingester.
+
+No network, no DB (PURE modules → fast lane, unmarked). The HTML feed is fed via
+monkeypatched `_feed_html` / `_loadmore_html`; the media HTTP layer via a fake
+`session` seam. Canned HTML mirrors the real markup characterized in the Step-0
+spike (Scribe note "SubscribeStar HTML characterization").
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import requests
+
+from backend.app.services.gallery_dl import DownloadResult, ErrorType
+from backend.app.services.native_ingest_common import post_dir_name
+from backend.app.services.subscribestar_client import (
+    MediaItem,
+    SubscribeStarAPIError,
+    SubscribeStarAuthError,
+    SubscribeStarClient,
+    SubscribeStarDriftError,
+    _parse_ss_datetime,
+)
+from backend.app.services.subscribestar_downloader import SubscribeStarDownloader
+from backend.app.services.subscribestar_ingester import (
+    SubscribeStarIngester,
+    _ledger_key,
+)
+from backend.app.utils.sidecar import find_sidecar
+
+_PNG_HEAD = b"\x89PNG\r\n\x1a\n"
+_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + b"\x00\x00\x00\x00IEND\xaeB`\x82"
+
+
+def _gallery_attr(items: list[dict]) -> str:
+    # data-gallery is HTML-entity-escaped JSON.
+    return json.dumps(items).replace("&", "&").replace('"', """)
+
+
+def _post_html(post_id="111", date="May 01, 2026 12:00 pm", body="<p>hello</p>",
+               media: list[dict] | None = None, locked=False):
+    gallery = ""
+    if media is not None:
+        gallery = (
+            f'<div class="uploads-images" data-gallery="{_gallery_attr(media)}"></div>'
+        )
+    lock = ' for-locked_content' if locked else ""
+    return (
+        f'<div class="post is-shown false false{lock}" data-id="{post_id}" '
+        f'data-infinite-scroll-id="{post_id}">'
+        f'<div class="post-date">{date}</div>'
+        f'<div class="post-body" data-role="post-body">'
+        f'<div class="post-content" data-role="post_content-text">'
+        f'<div class="trix-content">{body}</div></div></div>'
+        f'{gallery}</div>'
+    )
+
+
+def _feed_page(posts_html: str, next_href: str | None = None):
+    nxt = (
+        f'<div data-role="infinite_scroll-next_page" href="{next_href}"></div>'
+        if next_href else ""
+    )
+    return (
+        '<html><body><div data-role="posts_container-list">'
+        f'{posts_html}{nxt}</div></body></html>'
+    )
+
+
+# -- client: dates -----------------------------------------------------------
+
+def test_parse_ss_datetime_variants():
+    assert _parse_ss_datetime("Jun 17, 2026 03:19 am") == "2026-06-17T03:19:00"
+    assert _parse_ss_datetime("Updated on Jul 11, 2025 02:39 pm") == "2025-07-11T14:39:00"
+    assert _parse_ss_datetime("nonsense") is None
+
+
+# -- client: iteration + pagination -----------------------------------------
+
+def test_iter_posts_parses_and_paginates(monkeypatch):
+    client = SubscribeStarClient(None)
+    page1 = _feed_page(
+        _post_html("111") + _post_html("112"),
+        next_href="/posts?page=2&slug=x&sort_by=newest",
+    )
+    page2 = _feed_page(_post_html("113"))  # no next link → stop
+    monkeypatch.setattr(client, "_feed_html", lambda url: page1)
+    monkeypatch.setattr(client, "_loadmore_html", lambda url: page2)
+
+    out = list(client.iter_posts("https://subscribestar.adult/x"))
+    ids = [p["id"] for p, _inc, _cur in out]
+    assert ids == ["111", "112", "113"]
+    # page_cursor is the value that FETCHED each post's page: None for page 1, the
+    # next_page href for page 2.
+    assert out[0][2] is None
+    assert out[2][2] == "/posts?page=2&slug=x&sort_by=newest"
+    # base is stamped for media URL joining.
+    assert out[0][0]["_base"] == "https://subscribestar.adult"
+
+
+def test_iter_posts_drift_when_no_container(monkeypatch):
+    client = SubscribeStarClient(None)
+    monkeypatch.setattr(client, "_feed_html", lambda url: "<html>nothing</html>")
+    try:
+        list(client.iter_posts("https://subscribestar.adult/x"))
+    except SubscribeStarDriftError:
+        return
+    raise AssertionError("expected SubscribeStarDriftError")
+
+
+# -- client: media extraction ------------------------------------------------
+
+def test_extract_media_from_gallery():
+    client = SubscribeStarClient(None)
+    media = [
+        {"id": 9001, "type": "image", "original_filename": "art.jpg",
+         "url": "/post_uploads?payload=ABC"},
+        {"id": 9002, "type": "image", "original_filename": "art2.jpg",
+         "url": "/post_uploads?payload=DEF"},
+    ]
+    [post] = client._parse_posts(_feed_page(_post_html("111", media=media)))
+    post["_base"] = "https://subscribestar.adult"
+    items = client.extract_media(post, {})
+    assert [m.media_id for m in items] == ["9001", "9002"]
+    assert items[0].url == "https://subscribestar.adult/post_uploads?payload=ABC"
+    assert items[0].filename == "art.jpg"
+    assert items[0].post_id == "111"
+
+
+def test_text_post_has_no_media_but_keeps_body():
+    client = SubscribeStarClient(None)
+    [post] = client._parse_posts(_feed_page(_post_html("111", body="<p>text only</p>")))
+    post["_base"] = "https://subscribestar.adult"
+    assert client.extract_media(post, {}) == []
+    assert "text only" in post["attributes"]["content"]
+
+
+# -- client: gating + record key --------------------------------------------
+
+def test_post_is_gated():
+    client = SubscribeStarClient(None)
+    [open_post] = client._parse_posts(_feed_page(_post_html("1", media=[])))
+    [locked] = client._parse_posts(_feed_page(_post_html("2", locked=True)))
+    assert client.post_is_gated(open_post) is False
+    assert client.post_is_gated(locked) is True
+
+
+def test_post_record_key():
+    assert SubscribeStarClient.post_record_key({"id": "55"}) == ("post:55", "55")
+    assert SubscribeStarClient.post_record_key({}) is None
+
+
+# -- downloader --------------------------------------------------------------
+
+class _FakeResponse:
+    def __init__(self, payload=_PNG_BYTES, status_code=200):
+        self._payload = payload
+        self.status_code = status_code
+        self.headers: dict = {}
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise requests.HTTPError(f"HTTP {self.status_code}")
+
+    def iter_content(self, chunk_size=65536):
+        for i in range(0, len(self._payload), chunk_size):
+            yield self._payload[i:i + chunk_size]
+
+
+class _FakeSession:
+    def __init__(self):
+        self.calls: list[str] = []
+
+    def get(self, url, stream=False, timeout=None, headers=None):
+        self.calls.append(url)
+        return _FakeResponse()
+
+
+def _post(post_id="111", date="May 01, 2026 12:00 pm"):
+    return {
+        "id": post_id,
+        "attributes": {
+            "title": "",
+            "content": "<div>body text</div>",
+            "published_at": _parse_ss_datetime(date),
+            "post_type": "subscribestar",
+        },
+    }
+
+
+def _img(name, post_id="111", media_id="9001"):
+    return MediaItem(
+        url=f"https://subscribestar.adult/post_uploads?payload={name}",
+        filename=name, kind="image", filehash=None,
+        post_id=post_id, media_id=media_id,
+    )
+
+
+def _downloader(tmp_path: Path, session=None, validate=True):
+    return SubscribeStarDownloader(
+        images_root=tmp_path, cookies_path=None, validate=validate,
+        session=session or _FakeSession(),
+    )
+
+
+def test_post_dir_name_empty_title_matches_gallery_dl():
+    # SubscribeStar has no title → "<date>_<id>_" (matches gallery-dl's layout so
+    # existing downloads dedup on disk at cutover).
+    assert post_dir_name(_post("111")) == "2026-05-01_111_"
+
+
+def test_download_post_layout_and_outcomes(tmp_path):
+    # .png names so the PNG bytes pass the content/extension validator.
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(_post(), [_img("a.png"), _img("b.png", media_id="9002")], "artist-x")
+    post_dir = tmp_path / "artist-x" / "subscribestar" / "2026-05-01_111_"
+    assert (post_dir / "01_a.png").is_file()
+    assert (post_dir / "02_b.png").is_file()
+    assert [o.status for o in outcomes] == ["downloaded", "downloaded"]
+
+
+def test_per_media_sidecar_is_minimal(tmp_path):
+    dl = _downloader(tmp_path)
+    outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
+    sidecar = find_sidecar(outcomes[0].path)
+    data = json.loads(sidecar.read_text())
+    assert data["category"] == "subscribestar"
+    assert data["id"] == "111"
+    assert data["source_url"].endswith("payload=a.png")
+    # post-first: no body/title in the per-media sidecar.
+    assert "content" not in data and "title" not in data
+
+
+def test_write_post_record(tmp_path):
+    dl = _downloader(tmp_path)
+    rec = dl.write_post_record(_post("111"), "artist-x")
+    assert rec.path.name == "_post.json"
+    data = json.loads(rec.path.read_text())
+    assert data["category"] == "subscribestar"
+    assert data["id"] == "111" and data["post_id"] == "111"
+    assert "body text" in data["content"]
+    assert rec.body_chars > 0
+
+
+def test_skip_seen_does_not_download(tmp_path):
+    session = _FakeSession()
+    dl = _downloader(tmp_path, session=session)
+    outcomes = dl.download_post(
+        _post(), [_img("a.jpg")], "artist-x", is_seen=lambda m: True,
+    )
+    assert [o.status for o in outcomes] == ["skipped_seen"]
+    assert session.calls == []
+
+
+# -- ingester ----------------------------------------------------------------
+
+def test_ledger_key_prefers_filehash_then_post_media():
+    assert _ledger_key(_img("a.jpg")) == "111:9001"  # no filehash → post:media
+    hashed = MediaItem(
+        url="u", filename="a.jpg", kind="image", filehash="deadbeef",
+        post_id="111", media_id="9001",
+    )
+    assert _ledger_key(hashed) == "deadbeef"
+
+
+def _ingester(tmp_path):
+    return SubscribeStarIngester(
+        images_root=tmp_path, cookies_path=None, session_factory=lambda: None,
+    )
+
+
+def _mk_result(**kw) -> DownloadResult:
+    # _failure_result supplies success/return_code/error_type/error_message in kw;
+    # the factory only injects the always-required identity fields.
+    return DownloadResult(url="u", artist_slug="a", platform="subscribestar", **kw)
+
+
+def test_failure_result_maps_exceptions(tmp_path):
+    ing = _ingester(tmp_path)
+    assert ing._failure_result(SubscribeStarAuthError("x"), _mk_result).error_type == ErrorType.AUTH_ERROR
+    assert ing._failure_result(SubscribeStarDriftError("x"), _mk_result).error_type == ErrorType.API_DRIFT
+    assert ing._failure_result(
+        SubscribeStarAPIError("x", status_code=429), _mk_result
+    ).error_type == ErrorType.RATE_LIMITED
+    assert ing._failure_result(
+        SubscribeStarAPIError("x", status_code=404), _mk_result
+    ).error_type == ErrorType.NOT_FOUND
+    assert ing._failure_result(
+        SubscribeStarAPIError("x"), _mk_result
+    ).error_type == ErrorType.NETWORK_ERROR