"""All broker persistence paths are scoped and serialize token changes."""

from sqlalchemy import select

from src.domain.provider_auth.models import AuthorizationSession, ProviderConnection, ProviderGrant


class ProviderAuthRepository:
    def __init__(self, db):
        self.db = db

    async def add(self, row):
        self.db.add(row)
        await self.db.flush()
        return row

    async def session_by_state(self, state_hash):
        return await self.db.scalar(
            select(AuthorizationSession)
            .where(AuthorizationSession.state_hash == state_hash)
            .with_for_update()
        )

    async def session_by_completion(self, completion_hash, project_id):
        return await self.db.scalar(
            select(AuthorizationSession)
            .where(
                AuthorizationSession.completion_hash == completion_hash,
                AuthorizationSession.project_id == project_id,
            )
            .with_for_update()
        )

    async def connection(self, workspace_id, project_id, connection_id):
        return await self.db.scalar(
            select(ProviderConnection)
            .where(
                ProviderConnection.id == connection_id,
                ProviderConnection.workspace_id == workspace_id,
                ProviderConnection.project_id == project_id,
            )
            .with_for_update()
        )

    async def by_subject(self, session, subject):
        return await self.db.scalar(
            select(ProviderConnection)
            .where(
                ProviderConnection.workspace_id == session.workspace_id,
                ProviderConnection.project_id == session.project_id,
                ProviderConnection.provider == "google",
                ProviderConnection.actor_type == session.actor_type,
                ProviderConnection.actor_id == session.actor_id,
                ProviderConnection.subject == subject,
            )
            .with_for_update()
        )

    async def grants(self, connection_id):
        return list(
            (
                await self.db.scalars(
                    select(ProviderGrant).where(ProviderGrant.connection_id == connection_id)
                )
            ).all()
        )

    async def list_connections(self, workspace_id, project_id):
        return list(
            (
                await self.db.scalars(
                    select(ProviderConnection).where(
                        ProviderConnection.workspace_id == workspace_id,
                        ProviderConnection.project_id == project_id,
                        ProviderConnection.actor_type == "project",
                    )
                )
            ).all()
        )

    async def resume(self, workspace_id, project_id, session_id, initiated_by):
        return await self.db.scalar(
            select(AuthorizationSession).where(
                AuthorizationSession.id == session_id,
                AuthorizationSession.workspace_id == workspace_id,
                AuthorizationSession.project_id == project_id,
                AuthorizationSession.initiated_by == initiated_by,
            )
        )