DAO Layer — Feature Spec¶

"""
Curaway — Unified Data Store Error Taxonomy.

All data access errors are wrapped in DataStoreError subclasses.
Services catch these instead of store-specific exceptions (SQLAlchemy,
neo4j, botocore, httpx).
"""


class DataStoreError(Exception):
    """Base class for all data store errors."""

    def __init__(
        self,
        message: str,
        *,
        store: str,               # "postgres", "neo4j", "redis", "r2", "qstash"
        operation: str,           # "read", "write", "delete", "connect"
        retry_safe: bool = False,
        original_error: Exception | None = None,
    ):
        super().__init__(message)
        self.store = store
        self.operation = operation
        self.retry_safe = retry_safe
        self.original_error = original_error

    def to_dict(self) -> dict:
        return {
            "error_type": type(self).__name__,
            "store": self.store,
            "operation": self.operation,
            "retry_safe": self.retry_safe,
            "message": str(self),
        }


class StoreUnavailableError(DataStoreError):
    """Store is down, connection refused, or circuit breaker is open."""

    def __init__(self, store: str, operation: str = "connect", **kwargs):
        super().__init__(
            f"{store} is unavailable",
            store=store,
            operation=operation,
            retry_safe=True,
            **kwargs,
        )


class TenantIsolationViolation(DataStoreError):
    """A query was attempted without a tenant_id filter."""

    def __init__(self, store: str = "postgres", operation: str = "read"):
        super().__init__(
            "Query attempted without tenant_id — this is a bug, not a user error",
            store=store,
            operation=operation,
            retry_safe=False,
        )


class RecordNotFoundError(DataStoreError):
    """Expected record does not exist (after tenant-scoped lookup)."""

    def __init__(self, store: str, resource_type: str, resource_id: str):
        super().__init__(
            f"{resource_type} {resource_id} not found",
            store=store,
            operation="read",
            retry_safe=False,
        )


class DuplicateRecordError(DataStoreError):
    """Unique constraint violation."""

    def __init__(self, store: str, detail: str = ""):
        super().__init__(
            f"Duplicate record: {detail}" if detail else "Duplicate record",
            store=store,
            operation="write",
            retry_safe=False,
        )


class QueryError(DataStoreError):
    """Malformed query or constraint violation (not connectivity)."""

    def __init__(self, store: str, detail: str, **kwargs):
        super().__init__(
            f"Query error: {detail}",
            store=store,
            operation="read",
            retry_safe=False,
            **kwargs,
        )

"""
Curaway — Base Repository.

All PostgreSQL repositories inherit from this. Provides:
1. Tenant-scoped query builder (enforced — no unscoped reads)
2. Automatic AuditLog creation on writes
3. Error wrapping (SQLAlchemy exceptions -> DataStoreError)
"""

import uuid
import logging
from datetime import datetime, timezone

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.exc import IntegrityError, OperationalError

from app.errors import (
    DataStoreError,
    StoreUnavailableError,
    DuplicateRecordError,
    QueryError,
)
from app.models.audit import AuditLog

logger = logging.getLogger(__name__)


class BaseRepository:
    """Base class for all PostgreSQL repositories.

    Subclasses set `model` to the SQLAlchemy model class.
    All read methods must use `_scoped_query()` to enforce tenant isolation.
    """

    model = None  # Subclasses override: e.g., Patient, Case

    def __init__(self, db: AsyncSession):
        self.db = db

    # ── Tenant-scoped query builder ──

    def _scoped_query(self, tenant_id: str):
        """Return a SELECT pre-filtered by tenant_id.

        This is the ONLY way to start a query in a repository.
        Direct `select(self.model)` without tenant scoping is forbidden.
        """
        return select(self.model).where(self.model.tenant_id == tenant_id)

    def _scoped_count(self, tenant_id: str):
        """Return a COUNT query pre-filtered by tenant_id."""
        return select(func.count(self.model.id)).where(
            self.model.tenant_id == tenant_id
        )

    # ── Soft-delete filtering ──

    def _active_only(self, query):
        """Add is_deleted=False filter if the model supports soft-delete."""
        if hasattr(self.model, "is_deleted"):
            return query.where(self.model.is_deleted == False)  # noqa: E712
        return query

    # ── Audit logging ──

    def _audit(
        self,
        action: str,
        resource_id: str,
        tenant_id: str,
        actor_id: str = "system",
        actor_type: str = "user",
        changes: dict | None = None,
    ):
        """Create an AuditLog entry. Called automatically by write methods."""
        self.db.add(
            AuditLog(
                id=str(uuid.uuid4()),
                actor_id=actor_id,
                actor_type=actor_type,
                action=action,
                resource_type=self.model.__name__ if self.model else "unknown",
                resource_id=resource_id,
                tenant_id=tenant_id,
                changes=changes,
            )
        )

    # ── Error wrapping ──

    async def _execute(self, query):
        """Execute a query with error wrapping."""
        try:
            return await self.db.execute(query)
        except OperationalError as e:
            raise StoreUnavailableError(
                "postgres", operation="read", original_error=e
            ) from e
        except IntegrityError as e:
            error_msg = str(e.orig) if e.orig else str(e)
            if "unique" in error_msg.lower() or "duplicate" in error_msg.lower():
                raise DuplicateRecordError("postgres", detail=error_msg) from e
            raise QueryError("postgres", detail=error_msg, original_error=e) from e
        except Exception as e:
            raise DataStoreError(
                f"Unexpected database error: {e}",
                store="postgres",
                operation="read",
                original_error=e,
            ) from e

    async def _flush(self):
        """Flush pending changes with error wrapping."""
        try:
            await self.db.flush()
        except OperationalError as e:
            raise StoreUnavailableError(
                "postgres", operation="write", original_error=e
            ) from e
        except IntegrityError as e:
            error_msg = str(e.orig) if e.orig else str(e)
            if "unique" in error_msg.lower() or "duplicate" in error_msg.lower():
                raise DuplicateRecordError("postgres", detail=error_msg) from e
            raise QueryError("postgres", detail=error_msg, original_error=e) from e

    # ── Common read patterns ──

    async def get_by_id(self, record_id: str, tenant_id: str):
        """Fetch a single record by ID, scoped to tenant. Returns None if not found."""
        query = self._scoped_query(tenant_id).where(self.model.id == record_id)
        query = self._active_only(query)
        result = await self._execute(query)
        return result.scalar_one_or_none()

    async def list_paginated(
        self,
        tenant_id: str,
        page: int = 1,
        page_size: int = 20,
        order_by=None,
    ) -> tuple[list, int]:
        """List records with pagination, scoped to tenant.

        Returns (records, total_count).
        """
        base = self._scoped_query(tenant_id)
        base = self._active_only(base)

        count_q = self._scoped_count(tenant_id)
        if hasattr(self.model, "is_deleted"):
            count_q = count_q.where(self.model.is_deleted == False)  # noqa: E712

        total = (await self._execute(count_q)).scalar() or 0

        if order_by is not None:
            base = base.order_by(order_by)
        elif hasattr(self.model, "created_at"):
            base = base.order_by(self.model.created_at.desc())

        base = base.offset((page - 1) * page_size).limit(page_size)
        result = await self._execute(base)

        return list(result.scalars().all()), total

"""
Curaway — Patient Repository.

Typed data access for the Patient model with tenant isolation,
audit logging, and error wrapping.
"""

import uuid
from datetime import datetime, timezone

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.patient import Patient
from app.models.event import Event
from app.repositories.base import BaseRepository


class PatientRepository(BaseRepository):
    model = Patient

    async def get_by_id(self, patient_id: str, tenant_id: str) -> Patient | None:
        """Fetch a patient by ID, scoped to tenant. Excludes soft-deleted."""
        query = (
            self._scoped_query(tenant_id)
            .where(Patient.id == patient_id)
        )
        query = self._active_only(query)
        result = await self._execute(query)
        return result.scalar_one_or_none()

    async def get_by_auth_id(
        self, external_auth_id: str, tenant_id: str
    ) -> Patient | None:
        """Fetch a patient by Clerk auth ID, scoped to tenant."""
        query = (
            self._scoped_query(tenant_id)
            .where(Patient.external_auth_id == external_auth_id)
        )
        query = self._active_only(query)
        result = await self._execute(query)
        return result.scalar_one_or_none()

    async def list_active(
        self,
        tenant_id: str,
        intake_status: str | None = None,
        is_active: bool | None = True,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[Patient], int]:
        """List patients with filtering and pagination, scoped to tenant."""
        query = self._scoped_query(tenant_id)
        query = self._active_only(query)
        count_query = self._scoped_count(tenant_id)
        if hasattr(Patient, "is_deleted"):
            count_query = count_query.where(Patient.is_deleted == False)  # noqa: E712

        if intake_status:
            query = query.where(Patient.intake_status == intake_status)
            count_query = count_query.where(Patient.intake_status == intake_status)
        if is_active is not None:
            query = query.where(Patient.is_active == is_active)
            count_query = count_query.where(Patient.is_active == is_active)

        total = (await self._execute(count_query)).scalar() or 0
        query = query.order_by(Patient.created_at.desc())
        query = query.offset((page - 1) * page_size).limit(page_size)
        result = await self._execute(query)

        return list(result.scalars().all()), total

    async def create(
        self,
        patient: Patient,
        tenant_id: str,
        actor_id: str = "system",
    ) -> Patient:
        """Persist a new Patient. Auto-creates audit log and intake event."""
        self.db.add(patient)

        self._audit(
            action="patient.create",
            resource_id=patient.id,
            tenant_id=tenant_id,
            actor_id=actor_id,
        )

        self.db.add(
            Event(
                id=str(uuid.uuid4()),
                event_type="intake.started",
                tenant_id=tenant_id,
                actor_id=actor_id,
                patient_id=patient.id,
                payload={
                    "locale": patient.preferred_locale,
                    "source": patient.referral_source,
                },
                source_service="intake_api",
            )
        )

        await self._flush()
        return patient

    async def update(
        self,
        patient: Patient,
        changes: dict,
        actor_id: str = "system",
    ) -> Patient:
        """Update a patient and log changes. Caller sets fields before calling."""
        patient.updated_at = datetime.now(timezone.utc)

        if changes:
            self._audit(
                action="patient.update",
                resource_id=patient.id,
                tenant_id=patient.tenant_id,
                actor_id=actor_id,
                changes=changes,
            )

        await self._flush()
        return patient

    async def update_intake_status(
        self,
        patient: Patient,
        new_status: str,
        actor_id: str = "system",
    ) -> Patient:
        """Advance patient intake status with audit + event."""
        old_status = patient.intake_status
        patient.intake_status = new_status
        patient.updated_at = datetime.now(timezone.utc)

        self._audit(
            action="patient.intake_status_change",
            resource_id=patient.id,
            tenant_id=patient.tenant_id,
            actor_id=actor_id,
            changes={"intake_status": {"before": old_status, "after": new_status}},
        )

        self.db.add(
            Event(
                id=str(uuid.uuid4()),
                event_type=f"intake.{new_status}",
                tenant_id=patient.tenant_id,
                actor_id=actor_id,
                patient_id=patient.id,
                payload={"old_status": old_status, "new_status": new_status},
                source_service="intake_api",
            )
        )

        await self._flush()
        return patient

async def generate_case_number(self, tenant_id: str) -> str:
    """Generate CRW-{YEAR}-{5-digit} case number, scoped to tenant."""
    year = datetime.now(timezone.utc).year
    prefix = f"CRW-{year}-"
    result = await self._execute(
        select(func.max(Case.case_number))
        .where(Case.tenant_id == tenant_id)
        .where(Case.case_number.like(f"{prefix}%"))
    )
    max_number = result.scalar()
    if max_number:
        seq = int(max_number.split("-")[-1]) + 1
    else:
        seq = 1
    return f"{prefix}{seq:05d}"

async def get_patient(db: AsyncSession, patient_id: str, tenant_id: str) -> Patient | None:
    result = await db.execute(
        select(Patient).where(
            Patient.id == patient_id,
            Patient.tenant_id == tenant_id,
            Patient.is_deleted == False,
        )
    )
    return result.scalar_one_or_none()

async def get_patient(db: AsyncSession, patient_id: str, tenant_id: str) -> Patient | None:
    repo = PatientRepository(db)
    return await repo.get_by_id(patient_id, tenant_id)

async def _generate_case_number(db: AsyncSession) -> str:
    year = datetime.now(timezone.utc).year
    prefix = f"CRW-{year}-"
    result = await db.execute(
        text("SELECT MAX(case_number) FROM cases WHERE case_number LIKE :prefix"),
        {"prefix": f"{prefix}%"},
    )
    max_number = result.scalar()
    ...

async def create_case(db: AsyncSession, patient_id: str, tenant_id: str) -> dict:
    repo = CaseRepository(db)
    return await repo.create(patient_id, tenant_id)

def _scoped_by_case(self, case_id: str, tenant_id: str):
    """Scope query by case_id, with tenant_id validation via join."""
    return (
        select(self.model)
        .join(Case, self.model.case_id == Case.id)
        .where(self.model.case_id == case_id)
        .where(Case.tenant_id == tenant_id)
    )

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

checkpointer = AsyncPostgresSaver.from_conn_string(DATABASE_URL)
graph = triage_agent_graph.compile(checkpointer=checkpointer)

class BaseUnscopedRepository:
    """For system tables and cross-tenant admin operations.

    WARNING: Only use for:
    - System-wide tables (providers, exchange_rates, treatment_categories)
    - Super-admin operations with explicit authorization check
    - GDPR erasure cascade (which must delete across all tenants)

    All uses MUST be logged in audit_log with actor_type='system_admin'.
    """

    model = None

    def __init__(self, db: AsyncSession):
        self.db = db

    def _query(self):
        """Unscoped query. No tenant filter."""
        return select(self.model)

    async def _execute(self, query):
        """Same error wrapping as BaseRepository."""
        # ... same try/except pattern