refactor(services): shared race-safe get_or_create helper (DRY backend sweep)

The find-or-create dance — SELECT, then a SAVEPOINT INSERT that recovers (not a
full rollback) on IntegrityError when a concurrent worker inserted first — was
hand-rolled identically in 4 async sites: ArtistService.find_or_create,
TagService.find_or_create, ExtensionService._find_or_create_artist and
._find_or_create_source. Divergent copies of exactly this pattern are how the
duplicate-row/race bugs in reference_scalar_one_or_none_duplicates crept in, so
it now lives once in services/db_helpers.get_or_create (returns (row, created);
factory adds+flushes+returns the row; caller owns the outer commit).

Over-DRY guard: SourceService's IntegrityError sites RAISE DuplicateSourceError
(reject-on-conflict, a different concept) — left alone. Importer._get_or_create
is the lone SYNC consumer (already shared by 2 callers) — stays separate, can't
cross the sync/async boundary. §8b: no hand-rolled async find-or-create remains.
Test: get_or_create creates then returns existing without re-invoking the factory.

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

backend/app/services/artist_service.py

@@ -10,7 +10,6 @@ from dataclasses import dataclass
 
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import (
@@ -24,6 +23,7 @@ from ..models import (
 )
 from ..models.tag import image_tag
 from ..utils.slug import slugify
+from .db_helpers import get_or_create
 from .gallery_service import decode_cursor, encode_cursor, thumbnail_url
 
 
@@ -250,12 +250,10 @@ class ArtistService:
         )
 
     async def find_or_create(self, name: str) -> tuple[Artist, bool]:
-        """Return (artist, created). Slug-keyed; idempotent under races.
+        """Return (artist, created). Slug-keyed; idempotent under races via the
-
+        shared race-safe db_helpers.get_or_create (savepoint + IntegrityError
-        Audit 2026-06-02: switched from session.rollback() to a
+        recovery). A new artist also seeds an ArtistVisit so the directory's
-        begin_nested savepoint + IntegrityError recovery so a lost
+        `+N new` badge starts at 0.
-        race doesn't unwind the calling request's surrounding work.
-        Mirrors importer._get_or_create.
         """
         cleaned = (name or "").strip()
         if not cleaned:
@@ -263,12 +261,8 @@ class ArtistService:
         slug = slugify(cleaned)
 
         select_existing = select(Artist).where(Artist.slug == slug)
-        existing = (await self.session.execute(select_existing)).scalar_one_or_none()
-        if existing is not None:
-            return existing, False
 
-        sp = await self.session.begin_nested()
+        async def _create() -> Artist:
-        try:
             artist = Artist(name=cleaned, slug=slug)
             self.session.add(artist)
             await self.session.flush()
@@ -279,13 +273,14 @@ class ArtistService:
             # count every image imported in the same session.
             self.session.add(ArtistVisit(artist_id=artist.id))
             await self.session.flush()
-            await sp.commit()
+            return artist
-        except IntegrityError:
+
-            await sp.rollback()
+        artist, created = await get_or_create(
-            existing = (await self.session.execute(select_existing)).scalar_one()
+            self.session, select_existing, _create
-            return existing, False
+        )
-        await self.session.commit()
+        if created:
-        return artist, True
+            await self.session.commit()
+        return artist, created
 
     async def autocomplete(self, prefix: str, limit: int = 20) -> list[Artist]:
         cleaned = (prefix or "").strip()

backend/app/services/db_helpers.py

@@ -0,0 +1,55 @@
+"""Shared DB-access helpers for the async services.
+
+`get_or_create` centralizes the race-safe find-or-create dance — SELECT, then on
+a miss a savepoint INSERT that recovers (NOT a full rollback) when a concurrent
+worker inserted the same row first. It was hand-rolled identically in
+ArtistService, TagService and ExtensionService; divergent copies are exactly how
+the duplicate-row / race bugs in [[reference_scalar_one_or_none_duplicates]] crept
+in, so it lives in one place now (DRY pattern sweep 2026-06-09).
+
+Note: this is the ASYNC sibling of `Importer._get_or_create` (sync, used by the
+filesystem-import path). The two can't share an implementation across the
+sync/async boundary; the importer one stays as the lone sync consumer.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from typing import TypeVar
+
+from sqlalchemy import Select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+T = TypeVar("T")
+
+
+async def get_or_create(
+    session: AsyncSession,
+    select_stmt: Select,
+    factory: Callable[[], Awaitable[T]],
+) -> tuple[T, bool]:
+    """Race-safe find-or-create. Returns ``(row, created)``.
+
+    Run ``select_stmt`` (scalar_one_or_none); if a row exists, return it with
+    ``created=False``. Otherwise open a SAVEPOINT and ``await factory()`` — which
+    must add its row(s), flush, and return the primary row. On ``IntegrityError``
+    (a concurrent worker inserted the same row first) roll back the SAVEPOINT —
+    NOT the outer transaction, which would lose the caller's surrounding work —
+    and re-run ``select_stmt`` (scalar_one) to return the row the other worker
+    created. The caller owns the outer commit.
+
+    A UNIQUE/partial-unique constraint matching ``select_stmt``'s predicate is
+    required for the recovery to trip; without it a duplicate slips through.
+    """
+    existing = (await session.execute(select_stmt)).scalar_one_or_none()
+    if existing is not None:
+        return existing, False
+    sp = await session.begin_nested()
+    try:
+        row = await factory()
+        await sp.commit()
+        return row, True
+    except IntegrityError:
+        await sp.rollback()
+        return (await session.execute(select_stmt)).scalar_one(), False

backend/app/services/extension_service.py

@@ -11,11 +11,11 @@ from __future__ import annotations
 import re
 
 from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import Artist, Source
 from ..utils.slug import slugify
+from .db_helpers import get_or_create
 from .source_service import BACKFILL_MAX_CHUNKS
 
 
@@ -169,24 +169,16 @@ class ExtensionService:
         500 against uq_artist_slug.
         """
         slug = slugify(raw_name)
-        existing = (await self.session.execute(
+
-            select(Artist).where(Artist.slug == slug)
+        async def _create() -> Artist:
-        )).scalar_one_or_none()
-        if existing is not None:
-            return existing, False
-        sp = await self.session.begin_nested()
-        try:
             artist = Artist(name=raw_name, slug=slug, is_subscription=True)
             self.session.add(artist)
             await self.session.flush()
-            await sp.commit()
+            return artist
-            return artist, True
+
-        except IntegrityError:
+        return await get_or_create(
-            await sp.rollback()
+            self.session, select(Artist).where(Artist.slug == slug), _create
-            recovered = (await self.session.execute(
+        )
-                select(Artist).where(Artist.slug == slug)
-            )).scalar_one()
-            return recovered, False
 
     async def _find_or_create_source(
         self, *, artist_id: int, platform: str, url: str,
@@ -194,17 +186,13 @@ class ExtensionService:
         """Race-safe — same pattern as _find_or_create_artist above. The
         uq_source_artist_platform_url constraint catches the duplicate
         insert; we roll the savepoint back and re-select."""
-        existing = (await self.session.execute(
+        select_existing = select(Source).where(
-            select(Source).where(
+            Source.artist_id == artist_id,
-                Source.artist_id == artist_id,
+            Source.platform == platform,
-                Source.platform == platform,
+            Source.url == url,
-                Source.url == url,
+        )
-            )
+
-        )).scalar_one_or_none()
+        async def _create() -> Source:
-        if existing is not None:
-            return existing, False
-        sp = await self.session.begin_nested()
-        try:
             # New subscription sources arm run-until-done backfill (plan #693)
             # so the first ticks walk the full history (otherwise gallery-dl's
             # exit:20 short-circuits before the archive is built). Mirrors
@@ -219,16 +207,11 @@ class ExtensionService:
             )
             self.session.add(src)
             await self.session.flush()
-            await sp.commit()
+            return src
-        except IntegrityError:
+
-            await sp.rollback()
+        src, created = await get_or_create(
-            recovered = (await self.session.execute(
+            self.session, select_existing, _create
-                select(Source).where(
+        )
-                    Source.artist_id == artist_id,
+        if created:
-                    Source.platform == platform,
+            await self.session.commit()
-                    Source.url == url,
+        return src, created
-                )
-            )).scalar_one()
-            return recovered, False
-        await self.session.commit()
-        return src, True

backend/app/services/tag_service.py

@@ -7,12 +7,12 @@ from dataclasses import dataclass
 
 from sqlalchemy import and_, case, exists, func, select, text, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import Tag, TagKind, image_tag
 from ..models.tag_allowlist import TagAllowlist
 from ..models.tag_reference_embedding import TagReferenceEmbedding
+from .db_helpers import get_or_create
 
 log = logging.getLogger(__name__)
 
@@ -130,20 +130,14 @@ class TagService:
             .order_by(Tag.id)
             .limit(1)
         )
-        existing = (await self.session.execute(stmt)).scalar_one_or_none()
+        async def _create() -> Tag:
-        if existing:
-            return existing
-
-        sp = await self.session.begin_nested()
-        try:
             new_tag = Tag(name=name, kind=kind, fandom_id=fandom_id)
             self.session.add(new_tag)
             await self.session.flush()
-            await sp.commit()
             return new_tag
-        except IntegrityError:
+
-            await sp.rollback()
+        tag, _created = await get_or_create(self.session, stmt, _create)
-            return (await self.session.execute(stmt)).scalar_one()
+        return tag
 
     async def autocomplete(
         self,

tests/test_db_helpers.py

@@ -0,0 +1,34 @@
+"""Race-safe get_or_create helper (shared by the async services)."""
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import Artist
+from backend.app.services.db_helpers import get_or_create
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.mark.asyncio
+async def test_get_or_create_creates_then_returns_existing(db):
+    stmt = select(Artist).where(Artist.slug == "goc-test")
+
+    factory_calls = {"n": 0}
+
+    async def _make():
+        factory_calls["n"] += 1
+        a = Artist(name="GOC Test", slug="goc-test")
+        db.add(a)
+        await db.flush()
+        return a
+
+    row1, created1 = await get_or_create(db, stmt, _make)
+    assert created1 is True
+    assert row1.slug == "goc-test"
+    assert factory_calls["n"] == 1
+
+    # Second call finds the existing row and does NOT invoke the factory.
+    row2, created2 = await get_or_create(db, stmt, _make)
+    assert created2 is False
+    assert row2.id == row1.id
+    assert factory_calls["n"] == 1  # factory not called when the row exists