"""Like repository — adapter over ``AsyncSession``. Likes are an append-only event log. Current state = latest event per (user, track). """ import datetime as dt import uuid from sqlalchemy import Subquery, func, select from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app.domain.entities.like import Like from app.domain.entities.track import Track from app.infrastructure.db.models.like import LikeModel from app.infrastructure.db.models.track import TrackModel def _to_entity(row: LikeModel) -> Like: return Like( id=row.id, user_id=row.user_id, track_id=row.track_id, value=row.value, created_at=row.created_at, ) def _track_to_entity(row: TrackModel) -> Track: return Track( id=row.id, title=row.title, artist_id=row.artist_id, album_id=row.album_id, storage_uri=row.storage_uri, file_format=row.file_format, file_size=row.file_size, source=row.source, source_id=row.source_id, duration_seconds=row.duration_seconds, genre=row.genre, year=row.year, track_number=row.track_number, metadata_status=row.metadata_status, metadata_error=row.metadata_error, enriched_at=row.enriched_at, availability=row.availability, created_at=row.created_at, updated_at=row.updated_at, ) class SqlAlchemyLikeRepository: def __init__(self, session: AsyncSession) -> None: self._session = session async def add(self, *, user_id: uuid.UUID, track_id: uuid.UUID, value: str) -> Like: row = LikeModel(user_id=user_id, track_id=track_id, value=value) 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, value: str, created_at: dt.datetime, ) -> bool: """Idempotent append for sync push: insert a client-generated like event, skipping it if the ``id`` already exists (a replay). Preserves the client's ``created_at`` (the event happened offline earlier). Returns whether a new row was stored.""" stmt = ( pg_insert(LikeModel) .values( id=id, user_id=user_id, track_id=track_id, value=value, created_at=created_at, ) .on_conflict_do_nothing(index_elements=["id"]) .returning(LikeModel.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[Like]: """Like 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 ``created_at`` still surface), oldest first. ``since=None`` returns everything up to ``until``.""" stmt = select(LikeModel).where( LikeModel.user_id == user_id, LikeModel.synced_at <= until ) if since is not None: stmt = stmt.where(LikeModel.synced_at > since) stmt = stmt.order_by(LikeModel.synced_at) rows = (await self._session.execute(stmt)).scalars().all() return [_to_entity(r) for r in rows] def _latest_events_sq( self, user_id: uuid.UUID, track_ids: list[uuid.UUID] | None ) -> Subquery: """The latest like event per ``track_id`` for a user, as a subquery. ``DISTINCT ON (track_id)`` with a deterministic tiebreaker (``created_at`` then ``id``) picks exactly one row per track even when two events share an identical ``created_at`` — likes carry a client-supplied timestamp from offline sync, so ties are realistic and a plain ``max()``+equality-join would return both rows (double-counting the track).""" stmt = select( LikeModel.track_id, LikeModel.value.label("value"), LikeModel.created_at.label("created_at"), ).where(LikeModel.user_id == user_id) if track_ids is not None: stmt = stmt.where(LikeModel.track_id.in_(track_ids)) return ( stmt.distinct(LikeModel.track_id) .order_by( LikeModel.track_id, LikeModel.created_at.desc(), LikeModel.id.desc(), ) .subquery() ) async def get_latest_state( self, *, user_id: uuid.UUID, track_ids: list[uuid.UUID] ) -> list[Like]: if not track_ids: return [] rows = ( ( await self._session.execute( select(LikeModel) .where( LikeModel.user_id == user_id, LikeModel.track_id.in_(track_ids), ) .distinct(LikeModel.track_id) .order_by( LikeModel.track_id, LikeModel.created_at.desc(), LikeModel.id.desc(), ) ) ) .scalars() .all() ) return [_to_entity(r) for r in rows] async def list_liked_tracks( self, *, user_id: uuid.UUID, limit: int, offset: int ) -> list[Track]: # Tracks where the latest like event has value='like', ordered by like time desc latest_sq = self._latest_events_sq(user_id, None) rows = ( ( await self._session.execute( select(TrackModel) .join(latest_sq, TrackModel.id == latest_sq.c.track_id) .where(latest_sq.c.value == "like") .order_by(latest_sq.c.created_at.desc()) .limit(limit) .offset(offset) ) ) .scalars() .all() ) return [_track_to_entity(r) for r in rows] async def count_liked_tracks(self, *, user_id: uuid.UUID) -> int: latest_sq = self._latest_events_sq(user_id, None) return ( await self._session.execute( select(func.count()) .select_from(latest_sq) .where(latest_sq.c.value == "like") ) ).scalar_one()