gh-christianlouis-docuelevate/app/api/integrations.py

"""API endpoints for managing per-user integrations (sources and destinations).

Provides CRUD operations for :class:`~app.models.UserIntegration` records.
Each record represents one ingestion source (e.g. IMAP, Watch Folder) or
storage destination (e.g. S3, Dropbox, Google Drive) configured by a user.

Sensitive credentials are encrypted at rest using Fernet symmetric encryption
(keyed from ``SESSION_SECRET``) via :mod:`app.utils.encryption`.  Credential
values are **never** returned in API responses.
"""

import json
import logging
from typing import Annotated, Any

from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.database import get_db
from app.models import IntegrationDirection, IntegrationType, UserIntegration
from app.utils.encryption import decrypt_value, encrypt_value
from app.utils.user_scope import get_current_owner_id

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/integrations", tags=["integrations"])

DbSession = Annotated[Session, Depends(get_db)]

# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------


def _get_owner_id(request: Request) -> str:
    """Return the current user's owner ID, raising 401 if unauthenticated."""
    owner_id = get_current_owner_id(request)
    if owner_id is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    return owner_id


CurrentOwner = Annotated[str, Depends(_get_owner_id)]

# ---------------------------------------------------------------------------
# Pydantic schemas
# ---------------------------------------------------------------------------

_VALID_DIRECTIONS = IntegrationDirection.ALL
_VALID_TYPES = IntegrationType.ALL


class IntegrationCreate(BaseModel):
    """Schema for creating a new integration."""

    direction: str = Field(..., description="'SOURCE' or 'DESTINATION'")
    integration_type: str = Field(..., description="Integration type (e.g. 'IMAP', 'S3', 'DROPBOX')")
    name: str = Field(..., min_length=1, max_length=255, description="Human-readable label")
    config: dict[str, Any] | None = Field(default=None, description="Non-sensitive configuration (JSON object)")
    credentials: dict[str, Any] | None = Field(
        default=None, description="Sensitive credentials (JSON object, encrypted at rest)"
    )
    is_active: bool = Field(default=True, description="Whether the integration is active")


class IntegrationUpdate(BaseModel):
    """Schema for updating an existing integration (all fields optional)."""

    name: str | None = Field(default=None, min_length=1, max_length=255)
    config: dict[str, Any] | None = None
    credentials: dict[str, Any] | None = None
    is_active: bool | None = None


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------


def _validate_direction(direction: str) -> None:
    """Raise 400 if *direction* is not a known value."""
    if direction not in _VALID_DIRECTIONS:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Invalid direction '{direction}'. Must be one of: {sorted(_VALID_DIRECTIONS)}",
        )


def _validate_integration_type(integration_type: str) -> None:
    """Raise 400 if *integration_type* is not a known value."""
    if integration_type not in _VALID_TYPES:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Invalid integration_type '{integration_type}'. Must be one of: {sorted(_VALID_TYPES)}",
        )


# ---------------------------------------------------------------------------
# Serialisation helpers
# ---------------------------------------------------------------------------


def _to_response(integration: UserIntegration) -> dict[str, Any]:
    """Serialise a :class:`UserIntegration` row to a response dict.

    Credentials are **never** included; only a boolean flag indicating
    whether credentials have been configured is returned.
    """
    config_data: dict[str, Any] | None = None
    if integration.config:
        try:
            config_data = json.loads(integration.config)
        except (json.JSONDecodeError, TypeError):
            config_data = None

    return {
        "id": integration.id,
        "owner_id": integration.owner_id,
        "direction": integration.direction,
        "integration_type": integration.integration_type,
        "name": integration.name,
        "config": config_data,
        "has_credentials": bool(integration.credentials),
        "is_active": integration.is_active,
        "last_used_at": integration.last_used_at.isoformat() if integration.last_used_at else None,
        "last_error": integration.last_error,
        "created_at": integration.created_at.isoformat() if integration.created_at else None,
        "updated_at": integration.updated_at.isoformat() if integration.updated_at else None,
    }


def _encode_credentials(credentials: dict[str, Any] | None) -> str | None:
    """Serialise *credentials* dict to an encrypted JSON string for storage."""
    if not credentials:
        return None
    plaintext = json.dumps(credentials)
    return encrypt_value(plaintext)


def _decode_credentials(stored: str | None) -> dict[str, Any] | None:
    """Decrypt and deserialise stored credentials back to a dict.

    Returns ``None`` when *stored* is empty or cannot be decoded.
    """
    if not stored:
        return None
    plaintext = decrypt_value(stored)
    if not plaintext:
        return None
    try:
        return json.loads(plaintext)
    except (json.JSONDecodeError, TypeError):
        logger.error("Failed to decode credentials JSON after decryption")
        return None


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


@router.get("/", summary="List integrations for the current user")
def list_integrations(
    request: Request,
    db: DbSession,
    owner_id: CurrentOwner,
    direction: str | None = None,
    integration_type: str | None = None,
) -> list[dict[str, Any]]:
    """Return all integrations belonging to the authenticated user.

    Optional query-string filters:

    - ``direction`` — ``SOURCE`` or ``DESTINATION``
    - ``integration_type`` — e.g. ``IMAP``, ``S3``, ``DROPBOX``
    """
    query = db.query(UserIntegration).filter(UserIntegration.owner_id == owner_id)

    if direction is not None:
        _validate_direction(direction)
        query = query.filter(UserIntegration.direction == direction)

    if integration_type is not None:
        _validate_integration_type(integration_type)
        query = query.filter(UserIntegration.integration_type == integration_type)

    integrations = query.order_by(UserIntegration.id).all()
    return [_to_response(i) for i in integrations]


@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a new integration")
def create_integration(
    request: Request,
    body: IntegrationCreate,
    db: DbSession,
    owner_id: CurrentOwner,
) -> dict[str, Any]:
    """Create a new source or destination integration for the current user.

    ``credentials`` are encrypted at rest using Fernet symmetric encryption
    before being persisted and are **never** returned in API responses.
    """
    _validate_direction(body.direction)
    _validate_integration_type(body.integration_type)

    integration = UserIntegration(
        owner_id=owner_id,
        direction=body.direction,
        integration_type=body.integration_type,
        name=body.name,
        config=json.dumps(body.config) if body.config is not None else None,
        credentials=_encode_credentials(body.credentials),
        is_active=body.is_active,
    )

    try:
        db.add(integration)
        db.commit()
        db.refresh(integration)
    except Exception:
        db.rollback()
        raise

    logger.info(
        "User %s created %s integration %d (%s)",
        owner_id,
        body.direction,
        integration.id,
        body.integration_type,
    )
    return _to_response(integration)


@router.get("/{integration_id}", summary="Get a single integration")
def get_integration(
    integration_id: int,
    request: Request,
    db: DbSession,
    owner_id: CurrentOwner,
) -> dict[str, Any]:
    """Return a single integration by ID (must belong to the current user)."""
    integration = (
        db.query(UserIntegration)
        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
        .first()
    )
    if not integration:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
    return _to_response(integration)


@router.put("/{integration_id}", summary="Update an integration")
def update_integration(
    integration_id: int,
    request: Request,
    body: IntegrationUpdate,
    db: DbSession,
    owner_id: CurrentOwner,
) -> dict[str, Any]:
    """Update an existing integration.  Only provided fields are changed.

    When ``credentials`` is supplied the stored value is replaced in full
    with the freshly encrypted version of the new credentials dict.
    """
    integration = (
        db.query(UserIntegration)
        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
        .first()
    )
    if not integration:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")

    if body.name is not None:
        integration.name = body.name
    if body.config is not None:
        integration.config = json.dumps(body.config)
    if body.credentials is not None:
        integration.credentials = _encode_credentials(body.credentials)
    if body.is_active is not None:
        integration.is_active = body.is_active

    # Reset last_error so the next operation gives a fresh result
    integration.last_error = None

    try:
        db.commit()
        db.refresh(integration)
    except Exception:
        db.rollback()
        raise

    logger.info("User %s updated integration %d", owner_id, integration_id)
    return _to_response(integration)


@router.delete("/{integration_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete an integration")
def delete_integration(
    integration_id: int,
    request: Request,
    db: DbSession,
    owner_id: CurrentOwner,
) -> None:
    """Delete an integration permanently."""
    integration = (
        db.query(UserIntegration)
        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
        .first()
    )
    if not integration:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")

    try:
        db.delete(integration)
        db.commit()
    except Exception:
        db.rollback()
        raise

    logger.info("User %s deleted integration %d", owner_id, integration_id)


@router.get("/{integration_id}/credentials", summary="Retrieve decrypted credentials for an integration")
def get_integration_credentials(
    integration_id: int,
    request: Request,
    db: DbSession,
    owner_id: CurrentOwner,
) -> dict[str, Any]:
    """Return the decrypted credentials dict for a saved integration.

    This endpoint is intended for internal use by background tasks that need
    to authenticate with a third-party service.  Treat the response as
    sensitive — it contains plaintext secrets.
    """
    integration = (
        db.query(UserIntegration)
        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
        .first()
    )
    if not integration:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")

    credentials = _decode_credentials(integration.credentials)
    return {"credentials": credentials or {}}