refactor(dry-B3): extract _get_or_create race-safe find-or-create in Importer

_find_or_create_source, _source_for_sidecar, and _find_or_create_post each
repeated the SELECT → savepoint-INSERT → on-IntegrityError rollback+re-SELECT
pattern. Extracted _get_or_create(stmt, factory): the statement is reused for
the scalar_one_or_none lookup and the scalar_one post-conflict re-fetch, so
all three are reproduced exactly. Centralizing the race-safe pattern in one
place also reduces the risk of the copies drifting (the bug class banked
2026-05-26).

Left _upsert_artist (no savepoint by design) and the ImageProvenance void
ensure-exists block (no return / no re-select) alone — they don't fit.

The rest of the ingest pipeline was already DRY: sidecar parsing lives in
utils/sidecar.py, per-platform quirks in the platforms package, and
_safe_ext/_categorize_error/_build_config are each single-instance.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/services/importer.py

@@ -204,6 +204,32 @@ class Importer:
                 (phash, width or 0, height or 0, image_id)
             )
 
+    def _get_or_create(self, stmt, factory):
+        """Race-safe find-or-create. Run `stmt` (scalar_one_or_none); if a
+        row exists, return it. Otherwise open a savepoint and INSERT
+        ``factory()``; on IntegrityError (a concurrent worker inserted the
+        same row first) roll the savepoint back — NOT the outer transaction,
+        which would lose the surrounding scan's progress — and re-run `stmt`
+        (scalar_one) to return the row the other worker created.
+
+        Centralizes the pattern shared by _find_or_create_source,
+        _source_for_sidecar, and _find_or_create_post. The plain
+        SELECT-then-INSERT version lost races under the 5-min recovery sweep
+        (operator-flagged 2026-05-26)."""
+        existing = self.session.execute(stmt).scalar_one_or_none()
+        if existing is not None:
+            return existing
+        sp = self.session.begin_nested()
+        try:
+            row = factory()
+            self.session.add(row)
+            self.session.flush()
+            sp.commit()
+            return row
+        except IntegrityError:
+            sp.rollback()
+            return self.session.execute(stmt).scalar_one()
+
     def _find_or_create_source(
         self, *, artist_id: int, platform: str, url: str,
     ) -> Source:
@@ -222,31 +248,15 @@ class Importer:
         and re-select — the concurrent op just created the row we
         wanted, so the second select will find it.
         """
-        existing = self.session.execute(
-            select(Source).where(
-                Source.artist_id == artist_id,
-                Source.platform == platform,
-                Source.url == url,
-            )
-        ).scalar_one_or_none()
-        if existing is not None:
-            return existing
-        sp = self.session.begin_nested()
-        try:
-            row = Source(artist_id=artist_id, platform=platform, url=url)
-            self.session.add(row)
-            self.session.flush()
-            sp.commit()
-            return row
-        except IntegrityError:
-            sp.rollback()
-            return self.session.execute(
-                select(Source).where(
-                    Source.artist_id == artist_id,
-                    Source.platform == platform,
-                    Source.url == url,
-                )
-            ).scalar_one()
+        stmt = select(Source).where(
+            Source.artist_id == artist_id,
+            Source.platform == platform,
+            Source.url == url,
+        )
+        return self._get_or_create(
+            stmt,
+            lambda: Source(artist_id=artist_id, platform=platform, url=url),
+        )
 
     def _source_for_sidecar(
         self, *, artist_id: int, platform: str, artist_slug: str,
@@ -268,7 +278,7 @@ class Importer:
         ONE synthetic anchor with url='sidecar:<platform>:<artist_slug>' and
         enabled=False (so the subscription checker doesn't poll it).
         """
-        existing = self.session.execute(
+        stmt = (
             select(Source)
             .where(
                 Source.artist_id == artist_id,
@@ -276,33 +286,16 @@ class Importer:
             )
             .order_by(Source.id.asc())
             .limit(1)
-        ).scalar_one_or_none()
-        if existing is not None:
-            return existing
-        synthetic_url = f"sidecar:{platform}:{artist_slug}"
-        sp = self.session.begin_nested()
-        try:
-            row = Source(
+        )
+        return self._get_or_create(
+            stmt,
+            lambda: Source(
                 artist_id=artist_id,
                 platform=platform,
-                url=synthetic_url,
+                url=f"sidecar:{platform}:{artist_slug}",
                 enabled=False,
-            )
-            self.session.add(row)
-            self.session.flush()
-            sp.commit()
-            return row
-        except IntegrityError:
-            sp.rollback()
-            return self.session.execute(
-                select(Source)
-                .where(
-                    Source.artist_id == artist_id,
-                    Source.platform == platform,
-                )
-                .order_by(Source.id.asc())
-                .limit(1)
-            ).scalar_one()
+            ),
+        )
 
     def _find_or_create_post(
         self, *, source_id: int, external_post_id: str,
@@ -310,29 +303,14 @@ class Importer:
         """Race-safe find-or-create on `post` keyed by
         (source_id, external_post_id). Mirrors `_find_or_create_source`
         — same savepoint + IntegrityError-recovery pattern."""
-        existing = self.session.execute(
-            select(Post).where(
-                Post.source_id == source_id,
-                Post.external_post_id == external_post_id,
-            )
-        ).scalar_one_or_none()
-        if existing is not None:
-            return existing
-        sp = self.session.begin_nested()
-        try:
-            row = Post(source_id=source_id, external_post_id=external_post_id)
-            self.session.add(row)
-            self.session.flush()
-            sp.commit()
-            return row
-        except IntegrityError:
-            sp.rollback()
-            return self.session.execute(
-                select(Post).where(
-                    Post.source_id == source_id,
-                    Post.external_post_id == external_post_id,
-                )
-            ).scalar_one()
+        stmt = select(Post).where(
+            Post.source_id == source_id,
+            Post.external_post_id == external_post_id,
+        )
+        return self._get_or_create(
+            stmt,
+            lambda: Post(source_id=source_id, external_post_id=external_post_id),
+        )
 
     def import_one(self, source: Path) -> ImportResult:
         """Dispatch by kind. Media → normal pipeline. Archive → extract