diff --git a/backend/app/api/tags.py b/backend/app/api/tags.py index b189c1f..4fbe0a0 100644 --- a/backend/app/api/tags.py +++ b/backend/app/api/tags.py @@ -11,6 +11,7 @@ from ..services.bulk_tag_service import BulkTagService from ..services.series_match_service import SeriesMatchService from ..services.series_service import SeriesError, SeriesService from ..services.tag_directory_service import TagDirectoryService +from ..services.tag_query import serialize_tag from ..services.tag_service import ( TagMergeConflict, TagService, @@ -72,17 +73,7 @@ async def autocomplete(): hits = await svc.autocomplete(q, kind=kind, limit=limit) return jsonify( - [ - { - "id": h.id, - "name": h.name, - "kind": h.kind, - "fandom_id": h.fandom_id, - "fandom_name": h.fandom_name, - "image_count": h.image_count, - } - for h in hits - ] + [{**serialize_tag(h), "image_count": h.image_count} for h in hits] ) @@ -165,18 +156,7 @@ async def list_tags_for_image(image_id: int): async with get_session() as session: svc = TagService(session) tags = await svc.list_for_image(image_id) - return jsonify( - [ - { - "id": t.id, - "name": t.name, - "kind": t.kind.value, - "fandom_id": t.fandom_id, - "fandom_name": t.fandom_name, - } - for t in tags - ] - ) + return jsonify([serialize_tag(t) for t in tags]) @tags_bp.route("/images//tags", methods=["POST"]) diff --git a/backend/app/services/artist_service.py b/backend/app/services/artist_service.py index e72e3c0..20acf06 100644 --- a/backend/app/services/artist_service.py +++ b/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,7 +23,9 @@ from ..models import ( ) from ..models.tag import image_tag from ..utils.slug import slugify -from .gallery_service import decode_cursor, encode_cursor, thumbnail_url +from .db_helpers import get_or_create +from .gallery_service import thumbnail_url +from .pagination import decode_cursor, encode_cursor @dataclass(frozen=True) @@ -250,12 +251,10 @@ class ArtistService: ) async def find_or_create(self, name: str) -> tuple[Artist, bool]: - """Return (artist, created). Slug-keyed; idempotent under races. - - Audit 2026-06-02: switched from session.rollback() to a - begin_nested savepoint + IntegrityError recovery so a lost - race doesn't unwind the calling request's surrounding work. - Mirrors importer._get_or_create. + """Return (artist, created). Slug-keyed; idempotent under races via the + shared race-safe db_helpers.get_or_create (savepoint + IntegrityError + recovery). A new artist also seeds an ArtistVisit so the directory's + `+N new` badge starts at 0. """ cleaned = (name or "").strip() if not cleaned: @@ -263,12 +262,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() - try: + async def _create() -> Artist: artist = Artist(name=cleaned, slug=slug) self.session.add(artist) await self.session.flush() @@ -279,13 +274,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() - except IntegrityError: - await sp.rollback() - existing = (await self.session.execute(select_existing)).scalar_one() - return existing, False - await self.session.commit() - return artist, True + return artist + + artist, created = await get_or_create( + self.session, select_existing, _create + ) + if created: + await self.session.commit() + return artist, created async def autocomplete(self, prefix: str, limit: int = 20) -> list[Artist]: cleaned = (prefix or "").strip() diff --git a/backend/app/services/db_helpers.py b/backend/app/services/db_helpers.py new file mode 100644 index 0000000..992a76d --- /dev/null +++ b/backend/app/services/db_helpers.py @@ -0,0 +1,52 @@ +"""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 sqlalchemy import Select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + + +async def get_or_create[T]( + 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 diff --git a/backend/app/services/extension_service.py b/backend/app/services/extension_service.py index 25d02a8..a5a932f 100644 --- a/backend/app/services/extension_service.py +++ b/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) - )).scalar_one_or_none() - if existing is not None: - return existing, False - sp = await self.session.begin_nested() - try: + + async def _create() -> Artist: artist = Artist(name=raw_name, slug=slug, is_subscription=True) self.session.add(artist) await self.session.flush() - await sp.commit() - return artist, True - except IntegrityError: - await sp.rollback() - recovered = (await self.session.execute( - select(Artist).where(Artist.slug == slug) - )).scalar_one() - return recovered, False + return artist + + return await get_or_create( + self.session, select(Artist).where(Artist.slug == slug), _create + ) 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(Source).where( - Source.artist_id == artist_id, - Source.platform == platform, - Source.url == url, - ) - )).scalar_one_or_none() - if existing is not None: - return existing, False - sp = await self.session.begin_nested() - try: + select_existing = select(Source).where( + Source.artist_id == artist_id, + Source.platform == platform, + Source.url == url, + ) + + async def _create() -> Source: # 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() - except IntegrityError: - await sp.rollback() - recovered = (await self.session.execute( - select(Source).where( - Source.artist_id == artist_id, - Source.platform == platform, - Source.url == url, - ) - )).scalar_one() - return recovered, False - await self.session.commit() - return src, True + return src + + src, created = await get_or_create( + self.session, select_existing, _create + ) + if created: + await self.session.commit() + return src, created diff --git a/backend/app/services/gallery_service.py b/backend/app/services/gallery_service.py index 44cbe66..b8a0721 100644 --- a/backend/app/services/gallery_service.py +++ b/backend/app/services/gallery_service.py @@ -14,7 +14,6 @@ Decoding rejects malformed cursors with a ValueError; the API layer translates that to HTTP 400. """ -import base64 from dataclasses import dataclass from datetime import datetime @@ -24,8 +23,8 @@ from sqlalchemy.orm import aliased from ..models import Artist, ImageProvenance, ImageRecord, Post, Source, Tag from ..models.tag import image_tag - -CURSOR_SEPARATOR = "|" +from .pagination import decode_cursor, encode_cursor +from .tag_query import fandom_join_alias, serialize_tag, tag_columns # Reserved `platform` filter value selecting images with NO platformed # provenance (filesystem imports). Returned by facets() as a null-valued @@ -35,20 +34,6 @@ CURSOR_SEPARATOR = "|" UNSOURCED_PLATFORM = "__unsourced__" -def encode_cursor(effective_date: datetime, image_id: int) -> str: - raw = f"{effective_date.isoformat()}{CURSOR_SEPARATOR}{image_id}" - return base64.urlsafe_b64encode(raw.encode()).decode() - - -def decode_cursor(cursor: str) -> tuple[datetime, int]: - try: - raw = base64.urlsafe_b64decode(cursor.encode()).decode() - ts_part, id_part = raw.split(CURSOR_SEPARATOR, 1) - return datetime.fromisoformat(ts_part), int(id_part) - except Exception as exc: - raise ValueError(f"invalid cursor: {cursor!r}") from exc - - def _effective_date_col(): """The materialized gallery sort key: image_record.effective_date (alembic 0035) = COALESCE(primary post's post_date, created_at), @@ -559,16 +544,10 @@ class GalleryService: if record is None: return None # Self-join Tag to resolve a character's fandom NAME (not just id) so the - # modal chip can label it without an N+1 — mirrors list_for_image. - fandom_alias = Tag.__table__.alias("fandom_lookup") + # modal chip can label it without an N+1 (shared tag_query helpers). + fandom_alias = fandom_join_alias() tag_stmt = ( - select( - Tag.id, - Tag.name, - Tag.kind, - Tag.fandom_id, - fandom_alias.c.name.label("fandom_name"), - ) + select(*tag_columns(fandom_alias)) .select_from( Tag.__table__ .join(image_tag, image_tag.c.tag_id == Tag.id) @@ -615,16 +594,7 @@ class GalleryService: {"id": artist.id, "name": artist.name, "slug": artist.slug} if artist is not None else None ), - "tags": [ - { - "id": t.id, - "name": t.name, - "kind": t.kind.value if hasattr(t.kind, "value") else t.kind, - "fandom_id": t.fandom_id, - "fandom_name": t.fandom_name, - } - for t in tags - ], + "tags": [serialize_tag(t) for t in tags], "neighbors": neighbors, } diff --git a/backend/app/services/pagination.py b/backend/app/services/pagination.py new file mode 100644 index 0000000..5f3d01d --- /dev/null +++ b/backend/app/services/pagination.py @@ -0,0 +1,31 @@ +"""Shared opaque keyset-pagination cursor. + +A cursor is base64(``|``). The sort key is whatever +DESC-ordered timestamp a feed paginates on (gallery: effective_date; posts: +COALESCE(post_date, downloaded_at); artists: created_at). `decode_cursor` rejects +a malformed cursor with ValueError, which the API layer maps to HTTP 400. + +This was hand-rolled identically in gallery_service and post_feed_service (with +artist_service importing gallery's copy). Two divergent copies of a cursor format +silently break pagination in whichever feed drifts, so it lives once here now +(DRY pattern sweep 2026-06-10). +""" + +import base64 +from datetime import datetime + +CURSOR_SEPARATOR = "|" + + +def encode_cursor(sort_key: datetime, row_id: int) -> str: + raw = f"{sort_key.isoformat()}{CURSOR_SEPARATOR}{row_id}" + return base64.urlsafe_b64encode(raw.encode()).decode() + + +def decode_cursor(cursor: str) -> tuple[datetime, int]: + try: + raw = base64.urlsafe_b64decode(cursor.encode()).decode() + ts_part, id_part = raw.split(CURSOR_SEPARATOR, 1) + return datetime.fromisoformat(ts_part), int(id_part) + except Exception as exc: + raise ValueError(f"invalid cursor: {cursor!r}") from exc diff --git a/backend/app/services/post_feed_service.py b/backend/app/services/post_feed_service.py index f98debc..a4e1f78 100644 --- a/backend/app/services/post_feed_service.py +++ b/backend/app/services/post_feed_service.py @@ -1,9 +1,8 @@ """FC-3e: cursor-paginated read service for the Posts stream. -Mirrors GalleryService.scroll's cursor encoding so the frontend pattern -is identical: base64 of "|". Sort key is -COALESCE(Post.post_date, Post.downloaded_at) so posts without a -publish date sort by when we captured them. +Uses the shared `pagination` cursor (base64 of "|") so +every feed paginates identically. Sort key here is COALESCE(Post.post_date, +Post.downloaded_at) so posts without a publish date sort by when we captured them. Pure read-surface; no writes. The service composes the post dict (thumbnails from every image linked to the post — its own primary images @@ -12,9 +11,6 @@ attachments from PostAttachment) so the API layer can jsonify directly. """ from __future__ import annotations -import base64 -from datetime import datetime - from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession @@ -28,26 +24,12 @@ from ..models import ( ) from ..utils.text import html_to_plain, truncate_at_word from .gallery_service import thumbnail_url +from .pagination import decode_cursor, encode_cursor -CURSOR_SEPARATOR = "|" DESCRIPTION_LIMIT = 280 THUMBNAIL_LIMIT = 6 -def encode_cursor(sort_key: datetime, post_id: int) -> str: - raw = f"{sort_key.isoformat()}{CURSOR_SEPARATOR}{post_id}" - return base64.urlsafe_b64encode(raw.encode()).decode() - - -def decode_cursor(cursor: str) -> tuple[datetime, int]: - try: - raw = base64.urlsafe_b64decode(cursor.encode()).decode() - ts_part, id_part = raw.split(CURSOR_SEPARATOR, 1) - return datetime.fromisoformat(ts_part), int(id_part) - except Exception as exc: - raise ValueError(f"invalid cursor: {cursor!r}") from exc - - def _sort_key(): """Postgres COALESCE expression used in ORDER BY and WHERE clauses.""" return func.coalesce(Post.post_date, Post.downloaded_at) diff --git a/backend/app/services/tag_query.py b/backend/app/services/tag_query.py new file mode 100644 index 0000000..2dfbfbd --- /dev/null +++ b/backend/app/services/tag_query.py @@ -0,0 +1,44 @@ +"""Shared tag-with-fandom query columns + serialization. + +Resolving a character tag's fandom NAME via a Tag self-join (Tag.fandom_id -> the +fandom Tag) and serializing the canonical +``{id, name, kind, fandom_id, fandom_name}`` dict were hand-written in +TagService.autocomplete / .list_for_image and GalleryService.get_image_with_tags +(the last two added by the fandom-on-chip feature, 978f49a). One source now +(DRY pattern sweep 2026-06-10) so a new tag field is added in exactly one place. + +TagDirectoryService selects the FULL Tag ORM plus an image-count aggregate (a +different select shape), so it keeps its own variant — not folded in here. +""" + +from ..models import Tag + + +def fandom_join_alias(): + """A Tag self-join alias for resolving a tag's fandom name. Outerjoin it on + ``Tag.fandom_id == .c.id`` and select via `tag_columns(alias)`.""" + return Tag.__table__.alias("fandom_lookup") + + +def tag_columns(fandom_alias): + """The canonical (id, name, kind, fandom_id, fandom_name) column set for a + tag select that outerjoins `fandom_alias` (from `fandom_join_alias()`).""" + return [ + Tag.id, + Tag.name, + Tag.kind, + Tag.fandom_id, + fandom_alias.c.name.label("fandom_name"), + ] + + +def serialize_tag(row) -> dict: + """Serialize a row carrying .id/.name/.kind/.fandom_id/.fandom_name to the + canonical tag dict. `kind` may be a TagKind enum or a plain string.""" + return { + "id": row.id, + "name": row.name, + "kind": row.kind.value if hasattr(row.kind, "value") else row.kind, + "fandom_id": row.fandom_id, + "fandom_name": row.fandom_name, + } diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index 335afc3..3de2115 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -7,12 +7,13 @@ 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 +from .tag_query import fandom_join_alias, tag_columns log = logging.getLogger(__name__) @@ -130,20 +131,14 @@ class TagService: .order_by(Tag.id) .limit(1) ) - existing = (await self.session.execute(stmt)).scalar_one_or_none() - if existing: - return existing - - sp = await self.session.begin_nested() - try: + async def _create() -> Tag: 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() - return (await self.session.execute(stmt)).scalar_one() + + tag, _created = await get_or_create(self.session, stmt, _create) + return tag async def autocomplete( self, @@ -165,7 +160,7 @@ class TagService: else_=2, ) - fandom_alias = Tag.__table__.alias("fandom_lookup") + fandom_alias = fandom_join_alias() image_count = ( select(func.count(image_tag.c.image_record_id)) .where(image_tag.c.tag_id == Tag.id) @@ -175,11 +170,7 @@ class TagService: stmt = ( select( - Tag.id, - Tag.name, - Tag.kind, - Tag.fandom_id, - fandom_alias.c.name.label("fandom_name"), + *tag_columns(fandom_alias), image_count.label("image_count"), ) .select_from(Tag.__table__.outerjoin( @@ -230,15 +221,9 @@ class TagService: NAME (not just fandom_id) via a self-join on Tag, so the UI can label a character chip with its fandom without an N+1 (mirrors the autocomplete/directory resolution).""" - fandom_alias = Tag.__table__.alias("fandom_lookup") + fandom_alias = fandom_join_alias() stmt = ( - select( - Tag.id, - Tag.name, - Tag.kind, - Tag.fandom_id, - fandom_alias.c.name.label("fandom_name"), - ) + select(*tag_columns(fandom_alias)) .select_from( Tag.__table__ .join(image_tag, image_tag.c.tag_id == Tag.id) diff --git a/frontend/src/components/artist/ArtistDangerZone.vue b/frontend/src/components/artist/ArtistDangerZone.vue index 478d785..275c4a4 100644 --- a/frontend/src/components/artist/ArtistDangerZone.vue +++ b/frontend/src/components/artist/ArtistDangerZone.vue @@ -1,9 +1,6 @@