Merge remote-tracking branch 'origin/develop' into feat/qod-v1

# OBSERVABILITY_SETTINGS__LOG_LEVEL="INFO"

# CALLBACK_SETTINGS__TIMEOUT=10.0

# Encrypts the callback tokens clients send (EAM sinkCredential, QoD

# notificationAuthToken) before they are stored. Must be base64url of exactly

# 32 bytes; generate one with:

#   python -c "import os,base64;print(base64.urlsafe_b64encode(os.urandom(32)).decode())"

# Unset (the default) means OEG rejects any request carrying such a token with 501.

# Changing it makes every token already stored unreadable, so keep it stable.

# CALLBACK_SETTINGS__SECRET_KEY=""

# QOD_SETTINGS__SERVICE_SPECIFICATION_ID="7608e902-b927-559f-b448-e7e9061dfa5c"

# LOCATION_RETRIEVAL_SETTINGS__SERVICE_SPECIFICATION_ID="0129e7ce-8e02-5dfd-bae1-0303bcc7676e"

requires-python = "==3.12.*"

dependencies = [

    "asyncpg>=0.31.0",

    "cryptography>=50.0.1",

    "fastapi[standard]>=0.135.1",

    "nats-py>=2.10.0",

    "pyjwt[crypto]>=2.10.1",

    AppDeploymentRow,

    AppInstanceRow,

    AppRegistrationRow,

    CallbackCredentialRow,

    CallbackDeliveryRow,

    CallbackRegistrationRow,

    OperationRow,

    AppDeployment,

    AppInstance,

    AppRegistration,

    CallbackCredential,

    CallbackDelivery,

    CallbackRegistration,

    Operation,

            api_family=row.api_family,

            sink=row.sink,

            event_types=row.event_types,

            sink_credential_ref=row.sink_credential_ref,

            expires_at=row.expires_at,

            is_active=row.is_active,

            created_at=row.created_at,

            api_family=domain.api_family,

            sink=domain.sink,

            event_types=domain.event_types,

            sink_credential_ref=domain.sink_credential_ref,

            expires_at=domain.expires_at,

            is_active=domain.is_active,

        )

class CallbackCredentialMapper:

    @staticmethod

    def to_domain(row: CallbackCredentialRow) -> CallbackCredential:

        return CallbackCredential(

            callback_registration_id=row.callback_registration_id,

            encrypted_token=row.encrypted_token,

            expires_at=row.expires_at,

            created_at=row.created_at,

        )

    @staticmethod

    def to_row(domain: CallbackCredential) -> CallbackCredentialRow:

        return CallbackCredentialRow(

            callback_registration_id=domain.callback_registration_id,

            encrypted_token=domain.encrypted_token,

            expires_at=domain.expires_at,

        )

class CallbackDeliveryMapper:

    @staticmethod

    def to_domain(row: CallbackDeliveryRow) -> CallbackDelivery:

from datetime import datetime

from typing import Any, cast

from uuid import UUID

from sqlalchemy import CursorResult, select, update

from sqlalchemy.ext.asyncio import AsyncSession

from open_exposure_gateway.adapters.database.mappers import CallbackCredentialMapper

from open_exposure_gateway.adapters.database.sql import CallbackCredentialRow

from open_exposure_gateway.domain.models import CallbackCredential

from open_exposure_gateway.ports.database.callbacks import CallbackCredentialRepository

class SqlCallbackCredentialRepository(CallbackCredentialRepository):

    def __init__(self, session: AsyncSession) -> None:

        self._session = session

    async def get_by_registration_id(

        self, callback_registration_id: UUID

    ) -> CallbackCredential | None:

        stmt = select(CallbackCredentialRow).where(

            CallbackCredentialRow.callback_registration_id == callback_registration_id

        )

        row = await self._session.scalar(stmt)

        return CallbackCredentialMapper.to_domain(row) if row is not None else None

    async def save(self, callback_credential: CallbackCredential) -> CallbackCredential:

        merged = await self._session.merge(CallbackCredentialMapper.to_row(callback_credential))

        await self._session.flush()

        stmt = select(CallbackCredentialRow).where(

            CallbackCredentialRow.callback_registration_id == merged.callback_registration_id

        )

        row = await self._session.scalar(stmt)

        if row is None:

            raise RuntimeError("Saved callback credential could not be reloaded")

        return CallbackCredentialMapper.to_domain(row)

    async def revoke_by_registration_id(self, callback_registration_id: UUID) -> None:

        stmt = (

            update(CallbackCredentialRow)

            .where(CallbackCredentialRow.callback_registration_id == callback_registration_id)

            .values(encrypted_token=None)

        )

        await self._session.execute(stmt)

    async def revoke_expired(self, now: datetime) -> int:

        stmt = (

            update(CallbackCredentialRow)

            .where(

                CallbackCredentialRow.encrypted_token.is_not(None),

                CallbackCredentialRow.expires_at.is_not(None),

                CallbackCredentialRow.expires_at <= now,

            )

            .values(encrypted_token=None)

        )

        result = cast(CursorResult[Any], await self._session.execute(stmt))

        return result.rowcount

    ForeignKey,

    Index,

    Integer,

    LargeBinary,

    MetaData,

    String,

    Text,

    tenant_id: Mapped[str] = mapped_column(String(255), nullable=False)

    api_family: Mapped[str] = mapped_column(String(64), nullable=False)

    sink: Mapped[str] = mapped_column(Text, nullable=False)

    sink_credential_ref: Mapped[str | None] = mapped_column(String(255))

    event_types: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)

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

    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)

    )

class CallbackCredentialRow(Base):

    # Separate table rather than a column on callback_registrations: create_all never

    # alters an existing table.

    __tablename__ = "callback_credentials"

    __table_args__ = (Index("idx_callback_credentials_expires_at", "expires_at"),)

    callback_registration_id: Mapped[UUID] = mapped_column(

        ForeignKey("callback_registrations.id"), primary_key=True

    )

    # nonce + AES-GCM ciphertext; NULL once revoked. See application/callback_secrets.py.

    encrypted_token: Mapped[bytes | None] = mapped_column(LargeBinary)

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

    created_at: Mapped[datetime] = mapped_column(

        DateTime(timezone=True),

        nullable=False,

        server_default=func.now(),

    )

class CallbackDeliveryRow(Base):

    __tablename__ = "callback_deliveries"