"""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

from ..models import Source
from .gallery_dl import ErrorType


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


# --- shared Source health predicates ----------------------------------------
#
# The subscriptions rollup, the front-door status ribbon and the list endpoint
# all have to agree on what "failing" and "no access" MEAN, or the ribbon says
# 3 and the card it links to shows 4. Same reasoning as get_or_create above:
# divergent copies of one predicate are how the drift creeps in. Defined here
# rather than in source_service because scheduler_service needs them too, and
# source_service already imports scheduler_service (the other direction would
# be a cycle).


def failing_sources_clause():
    """A source is FAILING when its runs are actually erroring.

    Deliberately not `last_error IS NOT NULL` — a tier-limited source clears
    last_error and keeps a chip, and must never be counted as broken.
    """
    return Source.consecutive_failures > 0


def no_access_sources_clause():
    """A source we can't see the content of: the walk works, the tier doesn't
    grant it (#874 / milestone #387 phase A). Not a failure — kept separate
    from failing_sources_clause on purpose, and the two are disjoint because
    an informational class only ever rides an otherwise-OK run."""
    return Source.error_type == ErrorType.TIER_LIMITED