feat: add database engine setup and login_sessions table

from typing import Literal

from sqlalchemy.ext.asyncio import (

    AsyncEngine,

    AsyncSession,

    async_sessionmaker,

    create_async_engine,

)

from portal_service.adapters.database.sql import get_metadata

async def build_engine_and_session_maker(

    url: str, echo: bool | Literal["debug"]

) -> tuple[AsyncEngine, async_sessionmaker[AsyncSession]]:

    engine: AsyncEngine = create_async_engine(url, echo=echo)

    session_maker: async_sessionmaker[AsyncSession] = async_sessionmaker(

        bind=engine, expire_on_commit=False

    )

    return engine, session_maker

async def schema_initialization(engine: AsyncEngine) -> None:

    async with engine.begin() as conn:

        await conn.run_sync(get_metadata().create_all)

"""

Database tables.

With POSTGRESQL_SETTINGS__CREATE_SCHEMA_ON_STARTUP=true the app creates missing tables

from this metadata at start-up, as OEG and SRM do. create_all never alters a table that

already exists.

"""

from datetime import datetime

from sqlalchemy import DateTime, Index, LargeBinary, MetaData, String, func

from sqlalchemy.ext.asyncio import AsyncAttrs

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

def get_metadata() -> MetaData:

    return Base.metadata

class Base(AsyncAttrs, DeclarativeBase):

    pass

class LoginSessionRow(Base):

    """

    A browser login session. The cookie holds a session id and a per-session key; the

    tokens Keycloak issued stay here, encrypted with that key. The table alone cannot

    be turned back into cookies or tokens.

    """

    __tablename__ = "login_sessions"

    __table_args__ = (

        Index("idx_login_sessions_subject", "subject"),

        Index("idx_login_sessions_expires_at", "expires_at"),

    )

    # SHA-256 hex of the session id from the cookie.

    id: Mapped[str] = mapped_column(String(64), primary_key=True)

    # Keycloak user id (the token's `sub` claim).

    subject: Mapped[str] = mapped_column(String(255))

    # Kept in clear so a refresh can be scheduled without decrypting.

    access_token_expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))

    # Access, refresh and ID tokens, AES-GCM encrypted with the key held in the cookie.

    tokens: Mapped[bytes] = mapped_column(LargeBinary)

    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    # Hard end of the session, whatever happens to the tokens; expired rows get purged.

    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))