"""Play history repository — adapter over ``AsyncSession``.""" import datetime as dt import uuid from sqlalchemy import func, select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app.domain.entities.history import PlayHistoryEntry from app.infrastructure.db.models.play_history import PlayHistoryModel def _to_entity(row: PlayHistoryModel) -> PlayHistoryEntry: return PlayHistoryEntry( id=row.id, user_id=row.user_id, track_id=row.track_id, played_at=row.played_at, play_duration_seconds=row.play_duration_seconds, completed=row.completed, ) class SqlAlchemyHistoryRepository: def __init__(self, session: AsyncSession) -> None: self._session = session async def add( self, *, user_id: uuid.UUID, track_id: uuid.UUID, played_at: dt.datetime, play_duration_seconds: int | None, completed: bool, ) -> PlayHistoryEntry: row = PlayHistoryModel( user_id=user_id, track_id=track_id, played_at=played_at, play_duration_seconds=play_duration_seconds, completed=completed, ) self._session.add(row) await self._session.flush() await self._session.refresh(row) return _to_entity(row) async def add_event( self, *, id: uuid.UUID, user_id: uuid.UUID, track_id: uuid.UUID, played_at: dt.datetime, play_duration_seconds: int | None, completed: bool, ) -> bool: """Idempotent append for sync push: insert a client-generated play event, skipping it if the ``id`` already exists (a replay). Returns whether a new row was stored. Defined before ``list`` (name-shadowing).""" stmt = ( pg_insert(PlayHistoryModel) .values( id=id, user_id=user_id, track_id=track_id, played_at=played_at, play_duration_seconds=play_duration_seconds, completed=completed, ) .on_conflict_do_nothing(index_elements=["id"]) .returning(PlayHistoryModel.id) ) inserted = (await self._session.execute(stmt)).scalar_one_or_none() return inserted is not None async def list_since( self, user_id: uuid.UUID, *, since: dt.datetime | None, until: dt.datetime ) -> list[PlayHistoryEntry]: """Play events for a user ingested in the half-open window ``(since, until]`` (by ``synced_at``, the server-side sync key — so events pushed with an older ``played_at`` still surface), oldest first. ``since=None`` returns everything up to ``until``.""" stmt = select(PlayHistoryModel).where( PlayHistoryModel.user_id == user_id, PlayHistoryModel.synced_at <= until ) if since is not None: stmt = stmt.where(PlayHistoryModel.synced_at > since) stmt = stmt.order_by(PlayHistoryModel.synced_at) rows = (await self._session.execute(stmt)).scalars().all() return [_to_entity(r) for r in rows] async def list(self, *, user_id: uuid.UUID, limit: int, offset: int) -> list[PlayHistoryEntry]: rows = ( ( await self._session.execute( select(PlayHistoryModel) .where(PlayHistoryModel.user_id == user_id) .order_by(PlayHistoryModel.played_at.desc()) .limit(limit) .offset(offset) ) ) .scalars() .all() ) return [_to_entity(r) for r in rows] async def count(self, *, user_id: uuid.UUID) -> int: return ( await self._session.execute( select(func.count()) .select_from(PlayHistoryModel) .where(PlayHistoryModel.user_id == user_id) ) ).scalar_one()