chitai/backend/src/chitai/services/dependencies.py

# src/chitai/services/dependencies.py

# Standard library
from __future__ import annotations
from typing import Any, AsyncGenerator, Callable, NotRequired, Optional

# Third-party libraries
from advanced_alchemy.extensions.litestar.providers import (
    create_service_provider,
    FilterConfig,
    DEPENDENCY_DEFAULTS,
    DependencyDefaults,
)
from advanced_alchemy.exceptions import NotFoundError
from advanced_alchemy.filters import CollectionFilter, StatementFilter
from advanced_alchemy.service import FilterTypeT
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession
from litestar import Request
from litestar.params import Dependency, Parameter
from litestar.security.jwt import Token
from litestar.exceptions import HTTPException
from litestar.di import Provide

from advanced_alchemy.extensions.litestar.providers import create_filter_dependencies

# Local imports
from chitai import schemas as s
from chitai.database import models as m
from chitai.services import (
    UserService,
    BookService,
    LibraryService,
    BookProgressService,
    ShelfService,
    TagService,
    AuthorService,
    PublisherService,
    KosyncDeviceService,
    KosyncProgressService,
)
from chitai.config import settings
from chitai.services.filters.book import (
    AuthorFilter,
    BookshelfFilter,
    CustomOrderBy,
    ProgressFilter,
    TagFilter,
    TrigramSearchFilter,
)


async def provide_book_service(
    db_session: AsyncSession, current_user: m.User = Dependency(skip_validation=True)
) -> AsyncGenerator[BookService, None]:
    """
    Provide a BookService with per-user data scoped to the caller.

    `current_user` is a required dependency, not an optional argument. It used to
    default to None with the scoping below wrapped in `if current_user:` — and
    when it was not injected, that block was silently skipped, so
    `Book.progress_records` and `Book.list_links` loaded *every* user's rows.
    `Book.progress` returns `progress_records[0]`, so one user could see another
    user's reading position; the shelf checkboxes leaked the same way. Failing
    loudly on a missing user is the point of the change.
    """
    load = [
        selectinload(m.Book.author_links).selectinload(m.BookAuthorLink.author),
        selectinload(m.Book.tag_links).selectinload(m.BookTagLink.tag),
        m.Book.publisher,
        m.Book.files,
        m.Book.identifiers,
        m.Book.series,
        # Reading progress, restricted to the caller.
        #
        # The restriction lives on the relationship via .and_() rather than in a
        # separate with_loader_criteria(). advanced_alchemy's
        # get_abstract_loader_options() keeps only _AbstractLoad,
        # InstrumentedAttribute, RelationshipProperty and "*" entries and drops
        # everything else — and with_loader_criteria() is none of those, so the
        # previous criteria were discarded before reaching a query. A
        # selectinload() carrying its own .and_() survives that filter.
        selectinload(
            m.Book.progress_records.and_(m.BookProgress.user_id == current_user.id)
        ),
        # Bookshelf membership, restricted to the caller
        selectinload(
            m.Book.list_links.and_(
                m.BookListLink.book_list.has(m.BookList.user_id == current_user.id)
            )
        ).selectinload(m.BookListLink.book_list),
    ]

    provider_func = create_service_provider(
        BookService,
        load=load,
        uniquify=True,
        error_messages={
            "integrity": "Book operation failed.",
            "not_found": "The book does not exist.",
        },
        config=settings.alchemy_config,
    )

    async for service in provider_func(db_session=db_session):
        yield service


def create_book_filter_dependencies(
    config: FilterConfig,
    dep_defaults: DependencyDefaults = DEPENDENCY_DEFAULTS,
) -> dict[str, Provide]:
    """Create filter dependencies for books, including custom progress filters.

    Overrides:
    - SearchFilter: Uses trigram search for better fuzzy matching
    - OrderBy: Adds "random" sort order option

    Args:
        config: FilterConfig instance with desired settings.
        dep_defaults: Dependency defaults to use for the filter dependencies

    Returns:
        Dictionary of filter provider functions including base and custom filters.
    """

    # Get base filters first
    filters = create_filter_dependencies(config, dep_defaults)

    # OVERRIDE: id filter typed by the configured id type, not always `str`
    #
    # advanced_alchemy's `provide_id_filter` annotates `ids` as `list[str]` and
    # ignores `config["id_filter"]` entirely, so `?ids=12` reaches the database as
    # the string "12" and Postgres refuses `bigint = character varying`. Nothing
    # called `?ids=` until the duplicates screen needed to fetch a handful of books
    # by id, which is why it went unnoticed.
    if id_type := config.get("id_filter"):
        id_field = config.get("id_field", "id")

        def provide_typed_id_filter(
            ids=Parameter(query="ids", default=None, required=False),
        ) -> CollectionFilter:
            return CollectionFilter(field_name=id_field, values=ids)

        # Attached as a type object rather than written as an annotation: this module
        # has `from __future__ import annotations`, so a written one is stored as the
        # string "Optional[list[id_type]]" and resolved against module globals, where
        # a local named `id_type` does not exist.
        provide_typed_id_filter.__annotations__["ids"] = Optional[list[id_type]]

        filters[dep_defaults.ID_FILTER_DEPENDENCY_KEY] = Provide(
            provide_typed_id_filter, sync_to_thread=False
        )

    # OVERRIDE: Custom search filter with trigram search
    if config.get("search"):
        search_fields = config.get("search")

        def provide_trigram_search_filter(
            search_string: str | None = Parameter(
                title="Field to search",
                query="searchString",
                default=None,
                required=False,
            ),
            ignore_case: bool | None = Parameter(
                title="Search should be case sensitive",
                query="searchIgnoreCase",
                default=config.get("search_ignore_case", False),
                required=False,
            ),
        ) -> TrigramSearchFilter | None:
            if not search_string:
                return None

            # field_names = set(search_fields.split(",")) if isinstance(search_fields, str) else set(search_fields)
            return TrigramSearchFilter(
                field_name="book.title",
                value=search_string,
                ignore_case=ignore_case or False,
            )

        filters[dep_defaults.SEARCH_FILTER_DEPENDENCY_KEY] = Provide(
            provide_trigram_search_filter, sync_to_thread=False
        )

    # OVERRIDE: Custom order by with "random" option
    if config.get("sort_field"):
        sort_field = config.get("sort_field")

        def provide_custom_order_by(
            field_name: str | None = Parameter(
                title="Order by field",
                query="orderBy",
                default=sort_field,
                required=False,
            ),
            sort_order: str | None = Parameter(
                title="Sort order (asc, desc, or random)",
                query="sortOrder",
                default=config.get("sort_order", "desc"),
                required=False,
            ),
            current_user: m.User | None = Dependency(default=None),
        ) -> CustomOrderBy | None:
            if not field_name:
                return None

            # Validate sort_order
            valid_orders = {"asc", "desc", "random"}
            if sort_order not in valid_orders:
                raise ValueError(f"sort_order must be one of {valid_orders}")

            return CustomOrderBy(
                field_name=field_name, sort_order=sort_order, user=current_user
            )

        filters[dep_defaults.ORDER_BY_FILTER_DEPENDENCY_KEY] = Provide(
            provide_custom_order_by, sync_to_thread=False
        )

    return filters


def provide_libraries_filter(
    library_ids: Optional[list[int]] = Parameter(
        title="Filter by libraries", query="libraries", default=None, required=None
    ),
) -> CollectionFilter | None:
    if not library_ids:
        return None

    return CollectionFilter(field_name="library_id", values=library_ids)


def provide_authors_filter(
    author_ids: Optional[list[int]] = Parameter(
        title="Filter by authors", query="authors", default=None, required=False
    ),
) -> AuthorFilter | None:
    if not author_ids:
        return None
    return AuthorFilter(authors=author_ids)


def provide_publishers_filter(
    publisher_ids: Optional[list[int]] = Parameter(
        title="Filter by publishers", query="publishers", default=None, required=False
    ),
) -> CollectionFilter | None:
    if not publisher_ids:
        return None
    return CollectionFilter(field_name="publisher_id", values=publisher_ids)


def provide_tags_filter(
    tag_ids: Optional[list[int]] = Parameter(
        title="Filter by tags", query="tags", default=None, required=False
    ),
) -> TagFilter | None:
    if not tag_ids:
        return None
    return TagFilter(tags=tag_ids)


def provide_bookshelves_filter(
    shelf_ids: Optional[list[int]] = Parameter(
        title="Filter by bookshelves", query="shelves", default=None, required=False
    ),
    current_user: m.User = Dependency(skip_validation=True),
) -> BookshelfFilter | None:
    if not shelf_ids:
        return None
    return BookshelfFilter(lists=shelf_ids, user_id=current_user.id)


def provide_progress_filter(
    progress_statuses: Optional[list[str]] = Parameter(
        title="Filter by progress status",
        query="progress",
        default=None,
        required=False,
    ),
    current_user: m.User = Dependency(skip_validation=True),
) -> ProgressFilter | None:
    if not progress_statuses:
        return None
    return ProgressFilter(user_id=current_user.id, statuses=set(progress_statuses))


def provide_book_filters(
    libraries_filter: CollectionFilter | None = Dependency(skip_validation=True),
    authors_filter: AuthorFilter | None = Dependency(skip_validation=True),
    publishers_filter: CollectionFilter | None = Dependency(skip_validation=True),
    tags_filter: TagFilter | None = Dependency(skip_validation=True),
    bookshelves_filter: BookshelfFilter | None = Dependency(skip_validation=True),
    progress_filter: ProgressFilter | None = Dependency(skip_validation=True),
) -> list[StatementFilter]:
    """Combine all optional filters into a single list."""
    return [
        f
        for f in [
            libraries_filter,
            authors_filter,
            publishers_filter,
            tags_filter,
            bookshelves_filter,
            progress_filter,
        ]
        if f is not None
    ]


provide_library_service = create_service_provider(
    LibraryService,
)

provide_user_service = create_service_provider(
    UserService,
    error_messages={
        "duplicate_key": "Verification token already exists.",
        "integrity": "User operation failed.",
    },
)

provide_shelf_service = create_service_provider(
    ShelfService,
)

provide_progress_service = create_service_provider(
    BookProgressService,
)

provide_tag_service = create_service_provider(TagService)

provide_author_service = create_service_provider(AuthorService)

provide_publisher_service = create_service_provider(PublisherService)


async def get_library_by_id(
    library_service: LibraryService,
    books_service: BookService,
    library_id: int | None = None,
    book_id: int | None = Dependency(),
) -> m.Library:
    """Retrieves the library matching the id."""

    if not library_id:
        try:
            book = await books_service.get(book_id)
            library_id = book.library_id
        except NotFoundError:
            raise HTTPException(status_code=404, detail="The given book does not exist")

    try:
        return await library_service.get(library_id)

    except NotFoundError:
        raise HTTPException(status_code=404, detail="The given library does not exist")


def provide_user(request: Request[m.User, Token, Any]) -> m.User:
    return request.user


def provide_optional_user(request: Request[m.User, Token, Any]) -> m.User | None:
    if request.user:
        return request.user

    return None

async def provide_user_via_basic_auth(request: Request[m.User, None, Any]) -> m.User:
    return request.user


async def provide_user_via_kosync_auth(request: Request[m.User, None, Any]) -> m.User:
    return request.user


provide_kosync_device_service = create_service_provider(KosyncDeviceService)

provide_kosync_progress_service = create_service_provider(KosyncProgressService)