from __future__ import annotations

import json
from typing import Literal
from fastapi import APIRouter, Depends, Query, status
from sqlalchemy.orm import Session

from govoplan_core.auth import ApiPrincipal, require_any_scope
from govoplan_files.backend.change_tracking import (
    FILES_CONNECTOR_CREDENTIALS_COLLECTION,
    FILES_CONNECTOR_POLICIES_COLLECTION,
    FILES_CONNECTOR_PROFILES_COLLECTION,
)
from govoplan_files.backend.schemas import (
    FileConnectorCredentialCreateRequest,
    FileConnectorCredentialResponse,
    FileConnectorCredentialsResponse,
    FileConnectorCredentialUpdateRequest,
    FileConnectorDiscoveryRequest,
    FileConnectorDiscoveryResponse,
    FileConnectorSettingsDeltaResponse,
    FileConnectorPolicyEvaluateRequest,
    FileConnectorPolicyEvaluateResponse,
    FileConnectorPolicyResponse,
    FileConnectorPolicyUpdateRequest,
    FileConnectorProfileCreateRequest,
    FileConnectorProfileResponse,
    FileConnectorProfilesResponse,
    FileConnectorProviderResponse,
    FileConnectorProvidersResponse,
)
from govoplan_core.db.session import get_session
from govoplan_files.backend.storage.common import FileStorageError
from govoplan_files.backend.storage.connector_credential_store import (
    create_connector_credential_row,
    get_connector_credential_row,
    list_database_connector_credentials,
    update_connector_credential_row,
)
from govoplan_files.backend.storage.connector_credential_deletion import (
    delete_connector_credential_row,
)
from govoplan_files.backend.storage.connector_browse import (
    ConnectorBrowseError,
    ConnectorBrowseUnsupported,
    browse_connector_profile,
)
from govoplan_files.backend.storage.connector_deployment import (
    connector_effective_endpoint_url,
    reject_api_controlled_deployment_references,
)
from govoplan_files.backend.storage.connector_profile_store import (
    connector_profile_from_row,
    create_connector_profile_row,
)
from govoplan_files.backend.storage.connector_providers import (
    connector_provider_descriptors,
)
from govoplan_files.backend.storage.connector_policy import (
    ConnectorAccessRequest,
    ConnectorPolicyDenied,
    connector_policy_decision,
    connector_policy_sources_from_payload,
)
from govoplan_files.backend.storage.connector_policy_store import (
    connector_policy_response,
    set_connector_policy,
)


from govoplan_files.backend.route_support import (
    FILES_CONNECTOR_CREDENTIAL_RESOURCE,
    FILES_CONNECTOR_POLICY_RESOURCE,
    FILES_CONNECTOR_PROFILE_RESOURCE,
    _audit_connector_discovery_attempt,
    _can_read_disabled_connector_profiles,
    _connector_credential_response,
    _connector_policy_error,
    _credential_row_for_profile,
    _discovery_profile_from_payload,
    _ensure_campaign_file_access,
    _ensure_connector_configuration_allowed,
    _ensure_connector_credential_configuration_allowed,
    _ensure_connector_local_policy_allowed,
    _file_connector_policy_resource_id,
    _file_connector_settings_entries,
    _http_error,
    _record_connector_settings_change,
    _require_connector_credential_write,
    _require_connector_policy_read,
    _require_connector_profile_write,
    _same_endpoint,
    _visible_connector_profiles,
    _webdav_discovery_candidates,
)
from govoplan_files.backend.services.connector_settings_delta import (
    _full_file_connector_settings_delta_response,
    _incremental_file_connector_settings_delta_response,
)

router = APIRouter(prefix="/files", tags=["files"])


@router.post(
    "/connector-policy/evaluate", response_model=FileConnectorPolicyEvaluateResponse
)
def evaluate_connector_policy(
    payload: FileConnectorPolicyEvaluateRequest,
    principal: ApiPrincipal = Depends(
        require_any_scope("files:file:read", "files:file:upload", "files:file:download")
    ),
):
    del principal
    sources = connector_policy_sources_from_payload(
        [item.model_dump(mode="json") for item in payload.policy_sources]
    )
    request = ConnectorAccessRequest.from_provenance(
        payload.source_provenance.model_dump(mode="json", exclude_none=True),
        operation=payload.operation,
    )
    return FileConnectorPolicyEvaluateResponse(
        decision=connector_policy_decision(request, sources).to_dict()
    )


@router.get(
    "/connectors/settings/delta", response_model=FileConnectorSettingsDeltaResponse
)
def connector_settings_delta(
    scope_type: str = Query(default="tenant"),
    scope_id: str | None = Query(default=None),
    provider: str | None = None,
    campaign_id: str | None = None,
    include_disabled: bool = False,
    include_inactive: bool = False,
    owner_type: Literal["user", "group"] | None = None,
    owner_id: str | None = None,
    since: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:read", "admin:settings:read"
        )
    ),
):
    scope_type = scope_type.strip().casefold()
    _require_connector_policy_read(principal, scope_type)
    try:
        if since is None:
            return _full_file_connector_settings_delta_response(
                session,
                principal,
                scope_type=scope_type,
                scope_id=scope_id,
                provider=provider,
                campaign_id=campaign_id,
                include_disabled=include_disabled,
                include_inactive=include_inactive,
                owner_type=owner_type,
                owner_id=owner_id,
            )
        entries, has_more = _file_connector_settings_entries(
            session, tenant_id=principal.tenant_id, since=since, limit=limit
        )
        if entries is None:
            return _full_file_connector_settings_delta_response(
                session,
                principal,
                scope_type=scope_type,
                scope_id=scope_id,
                provider=provider,
                campaign_id=campaign_id,
                include_disabled=include_disabled,
                include_inactive=include_inactive,
                owner_type=owner_type,
                owner_id=owner_id,
            )
        return _incremental_file_connector_settings_delta_response(
            session,
            principal,
            entries=entries,
            has_more=has_more,
            scope_type=scope_type,
            scope_id=scope_id,
            provider=provider,
            campaign_id=campaign_id,
            include_disabled=include_disabled,
            include_inactive=include_inactive,
            owner_type=owner_type,
            owner_id=owner_id,
        )
    except (FileStorageError, ValueError, json.JSONDecodeError) as exc:
        raise _http_error(exc) from exc


@router.get("/connectors/providers", response_model=FileConnectorProvidersResponse)
def list_connector_providers(
    principal: ApiPrincipal = Depends(
        require_any_scope("files:file:read", "files:file:upload", "files:file:admin")
    ),
):
    del principal
    return FileConnectorProvidersResponse(
        providers=[
            FileConnectorProviderResponse(**item.to_response())
            for item in connector_provider_descriptors()
        ]
    )


@router.post("/connectors/discover", response_model=FileConnectorDiscoveryResponse)
def discover_connector_endpoint(
    payload: FileConnectorDiscoveryRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:write", "admin:settings:write"
        )
    ),
):
    try:
        reject_api_controlled_deployment_references(
            password_env=payload.credentials.password_env,
            token_env=payload.credentials.token_env,
            secret_ref=payload.credentials.secret_ref,
            metadata=payload.metadata,
        )
    except ValueError as exc:
        raise _http_error(exc) from exc
    if payload.provider not in {"webdav", "nextcloud"}:
        return FileConnectorDiscoveryResponse(
            provider=payload.provider,
            endpoint_url=None,
            base_path=payload.base_path,
            status="unsupported",
            message=f"Discovery is not implemented for {payload.provider} connectors yet",
        )
    candidates: list[dict[str, str]] = []
    for endpoint_url in _webdav_discovery_candidates(payload):
        profile = _discovery_profile_from_payload(
            payload, endpoint_url, principal=principal
        )
        try:
            _ensure_connector_configuration_allowed(
                session,
                principal,
                connector_id=None,
                credential_id=None,
                provider=profile.provider,
                endpoint_url=endpoint_url,
                base_path=profile.base_path,
                scope_type="tenant",
                scope_id=principal.tenant_id,
                operation="discover",
            )
            _audit_connector_discovery_attempt(
                session,
                principal,
                provider=profile.provider,
                endpoint_url=endpoint_url,
                base_path=profile.base_path,
            )
            browse_connector_profile(profile, path=payload.base_path or "")
        except ConnectorPolicyDenied as exc:
            raise _connector_policy_error(exc) from exc
        except (
            ConnectorBrowseError,
            ConnectorBrowseUnsupported,
            OSError,
            ValueError,
            json.JSONDecodeError,
        ) as exc:
            message = str(exc)
            if "credentials were rejected" in message.casefold():
                if payload.require_valid_credentials:
                    candidates.append(
                        {
                            "endpoint_url": endpoint_url,
                            "status": "credentials_rejected",
                            "message": "The endpoint exists, but the credentials were rejected.",
                        }
                    )
                    return FileConnectorDiscoveryResponse(
                        provider=payload.provider,
                        endpoint_url=endpoint_url,
                        base_path=payload.base_path,
                        status="credentials_rejected",
                        message="The endpoint was found, but login failed with these credentials.",
                        candidates=candidates,
                        metadata={"discovered_by": "webdav-auth-challenge"},
                    )
                candidates.append(
                    {
                        "endpoint_url": endpoint_url,
                        "status": "found",
                        "message": "The endpoint exists, but credentials are required or were rejected.",
                    }
                )
                return FileConnectorDiscoveryResponse(
                    provider=payload.provider,
                    endpoint_url=endpoint_url,
                    base_path=payload.base_path,
                    status="found",
                    message="The endpoint was found. Add working credentials before saving or testing the connection.",
                    candidates=candidates,
                    metadata={"discovered_by": "webdav-auth-challenge"},
                )
            candidates.append(
                {"endpoint_url": endpoint_url, "status": "failed", "message": message}
            )
            continue
        status_value = (
            "usable" if _same_endpoint(endpoint_url, payload.endpoint_url) else "found"
        )
        message = (
            "The supplied URL is directly usable."
            if status_value == "usable"
            else "A usable connector endpoint was discovered."
        )
        candidates.append(
            {"endpoint_url": endpoint_url, "status": status_value, "message": message}
        )
        return FileConnectorDiscoveryResponse(
            provider=payload.provider,
            endpoint_url=endpoint_url,
            base_path=payload.base_path,
            status=status_value,
            message=message,
            candidates=candidates,
            metadata={"discovered_by": "webdav-propfind"},
        )
    return FileConnectorDiscoveryResponse(
        provider=payload.provider,
        endpoint_url=None,
        base_path=payload.base_path,
        status="not_found",
        message="No usable WebDAV endpoint was found for this server URL.",
        candidates=candidates,
    )


@router.get("/connectors/credentials", response_model=FileConnectorCredentialsResponse)
def list_connector_credentials(
    provider: str | None = None,
    include_disabled: bool = False,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:read", "admin:settings:read"
        )
    ),
):
    provider_norm = provider.strip().casefold() if provider else None
    credentials = list_database_connector_credentials(
        session,
        tenant_id=principal.tenant_id,
        include_disabled=include_disabled
        and _can_read_disabled_connector_profiles(principal),
    )
    return FileConnectorCredentialsResponse(
        credentials=[
            FileConnectorCredentialResponse(**credential.to_response())
            for credential in credentials
            if provider_norm is None or credential.provider in {None, provider_norm}
        ]
    )


@router.get(
    "/connectors/policies/{scope_type}", response_model=FileConnectorPolicyResponse
)
def read_connector_policy(
    scope_type: str,
    scope_id: str | None = Query(default=None),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:read", "admin:settings:read"
        )
    ),
):
    _require_connector_policy_read(principal, scope_type)
    try:
        return FileConnectorPolicyResponse(
            **connector_policy_response(
                session,
                tenant_id=principal.tenant_id,
                scope_type=scope_type,
                scope_id=scope_id,
            )
        )
    except (FileStorageError, ValueError, json.JSONDecodeError) as exc:
        raise _http_error(exc, not_found=True) from exc


@router.put(
    "/connectors/policies/{scope_type}", response_model=FileConnectorPolicyResponse
)
def write_connector_policy(
    scope_type: str,
    payload: FileConnectorPolicyUpdateRequest,
    scope_id: str | None = Query(default=None),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:write", "admin:settings:write"
        )
    ),
):
    _require_connector_profile_write(principal, scope_type)
    try:
        set_connector_policy(
            session,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            scope_type=scope_type,
            scope_id=scope_id,
            policy=payload.policy,
        )
        _record_connector_settings_change(
            session,
            collection=FILES_CONNECTOR_POLICIES_COLLECTION,
            resource_type=FILES_CONNECTOR_POLICY_RESOURCE,
            resource_id=_file_connector_policy_resource_id(scope_type, scope_id),
            operation="updated",
            principal=principal,
            tenant_id=None
            if scope_type.strip().casefold() == "system"
            else principal.tenant_id,
            payload={"scope_type": scope_type.strip().casefold(), "scope_id": scope_id},
        )
        session.commit()
        return FileConnectorPolicyResponse(
            **connector_policy_response(
                session,
                tenant_id=principal.tenant_id,
                scope_type=scope_type,
                scope_id=scope_id,
            )
        )
    except (FileStorageError, ValueError, json.JSONDecodeError) as exc:
        session.rollback()
        raise _http_error(exc) from exc


@router.post(
    "/connectors/credentials",
    response_model=FileConnectorCredentialResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_connector_credential(
    payload: FileConnectorCredentialCreateRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:write", "admin:settings:write"
        )
    ),
):
    _require_connector_credential_write(principal, payload.scope_type)
    credentials = payload.credentials
    try:
        _ensure_connector_credential_configuration_allowed(
            session,
            principal,
            credential_id=payload.id,
            provider=payload.provider,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
            operation="configure_credentials",
        )
        _ensure_connector_local_policy_allowed(
            session,
            principal,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
            policy=payload.policy,
        )
        row = create_connector_credential_row(
            session,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            credential_id=payload.id,
            label=payload.label,
            provider=payload.provider,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
            enabled=payload.enabled,
            credential_mode=payload.credential_mode,
            username=credentials.username,
            password=credentials.password,
            token=credentials.token,
            password_env=credentials.password_env,
            token_env=credentials.token_env,
            secret_ref=credentials.secret_ref,
            policy=payload.policy,
            metadata=payload.metadata,
        )
        _record_connector_settings_change(
            session,
            collection=FILES_CONNECTOR_CREDENTIALS_COLLECTION,
            resource_type=FILES_CONNECTOR_CREDENTIAL_RESOURCE,
            resource_id=row.id,
            operation="created",
            principal=principal,
            tenant_id=row.tenant_id,
            payload={
                "scope_type": row.scope_type,
                "scope_id": row.scope_id,
                "provider": row.provider,
            },
        )
        session.commit()
        session.refresh(row)
        return _connector_credential_response(row)
    except ConnectorPolicyDenied as exc:
        session.rollback()
        raise _connector_policy_error(exc) from exc
    except (FileStorageError, ValueError, json.JSONDecodeError) as exc:
        session.rollback()
        raise _http_error(exc) from exc


@router.get(
    "/connectors/credentials/{credential_id}",
    response_model=FileConnectorCredentialResponse,
)
def get_connector_credential(
    credential_id: str,
    include_disabled: bool = False,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:read", "admin:settings:read"
        )
    ),
):
    try:
        row = get_connector_credential_row(
            session,
            tenant_id=principal.tenant_id,
            credential_id=credential_id,
            include_disabled=include_disabled
            and _can_read_disabled_connector_profiles(principal),
        )
        return _connector_credential_response(row)
    except FileStorageError as exc:
        raise _http_error(exc, not_found=True) from exc


@router.patch(
    "/connectors/credentials/{credential_id}",
    response_model=FileConnectorCredentialResponse,
)
def update_connector_credential(
    credential_id: str,
    payload: FileConnectorCredentialUpdateRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:write", "admin:settings:write"
        )
    ),
):
    try:
        row = get_connector_credential_row(
            session,
            tenant_id=principal.tenant_id,
            credential_id=credential_id,
            include_disabled=True,
        )
    except FileStorageError as exc:
        raise _http_error(exc, not_found=True) from exc
    _require_connector_credential_write(principal, row.scope_type)
    credentials = payload.credentials
    try:
        provider = payload.provider if payload.provider is not None else row.provider
        _ensure_connector_credential_configuration_allowed(
            session,
            principal,
            credential_id=row.id,
            provider=provider,
            scope_type=row.scope_type,
            scope_id=row.scope_id,
            operation="configure_credentials",
        )
        if payload.policy is not None:
            _ensure_connector_local_policy_allowed(
                session,
                principal,
                scope_type=row.scope_type,
                scope_id=row.scope_id,
                policy=payload.policy,
            )
        update_connector_credential_row(
            session,
            row,
            user_id=principal.user.id,
            label=payload.label,
            provider=payload.provider,
            enabled=payload.enabled,
            credential_mode=payload.credential_mode,
            username=credentials.username if credentials else None,
            password=credentials.password if credentials else None,
            token=credentials.token if credentials else None,
            password_env=credentials.password_env if credentials else None,
            token_env=credentials.token_env if credentials else None,
            secret_ref=credentials.secret_ref if credentials else None,
            clear_password=payload.clear_password,
            clear_token=payload.clear_token,
            policy=payload.policy,
            metadata=payload.metadata,
        )
        _record_connector_settings_change(
            session,
            collection=FILES_CONNECTOR_CREDENTIALS_COLLECTION,
            resource_type=FILES_CONNECTOR_CREDENTIAL_RESOURCE,
            resource_id=row.id,
            operation="updated",
            principal=principal,
            tenant_id=row.tenant_id,
            payload={
                "scope_type": row.scope_type,
                "scope_id": row.scope_id,
                "provider": row.provider,
            },
        )
        session.commit()
        session.refresh(row)
        return _connector_credential_response(row)
    except ConnectorPolicyDenied as exc:
        session.rollback()
        raise _connector_policy_error(exc) from exc
    except (FileStorageError, ValueError, json.JSONDecodeError) as exc:
        session.rollback()
        raise _http_error(exc) from exc


@router.delete(
    "/connectors/credentials/{credential_id}",
    response_model=FileConnectorCredentialResponse,
)
def deactivate_connector_credential(
    credential_id: str,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:write", "admin:settings:write"
        )
    ),
):
    try:
        row = get_connector_credential_row(
            session,
            tenant_id=principal.tenant_id,
            credential_id=credential_id,
            include_disabled=True,
        )
    except FileStorageError as exc:
        raise _http_error(exc, not_found=True) from exc
    _require_connector_credential_write(principal, row.scope_type)
    try:
        deletion = delete_connector_credential_row(
            session,
            row,
            deletion_reason="api_delete",
            user_id=principal.user.id,
            api_key_id=principal.api_key.id if principal.api_key else None,
        )
        if deletion.changed:
            _record_connector_settings_change(
                session,
                collection=FILES_CONNECTOR_CREDENTIALS_COLLECTION,
                resource_type=FILES_CONNECTOR_CREDENTIAL_RESOURCE,
                resource_id=row.id,
                operation="deleted",
                principal=principal,
                tenant_id=row.tenant_id,
                payload={
                    "scope_type": row.scope_type,
                    "scope_id": row.scope_id,
                    "provider": row.provider,
                },
            )
            for profile in deletion.affected_profiles:
                _record_connector_settings_change(
                    session,
                    collection=FILES_CONNECTOR_PROFILES_COLLECTION,
                    resource_type=FILES_CONNECTOR_PROFILE_RESOURCE,
                    resource_id=profile.id,
                    operation="updated",
                    principal=principal,
                    tenant_id=profile.tenant_id,
                    payload={
                        "scope_type": profile.scope_type,
                        "scope_id": profile.scope_id,
                        "provider": profile.provider,
                        "reason": "credential_deleted",
                    },
                )
        session.commit()
        session.refresh(row)
        return _connector_credential_response(row)
    except FileStorageError as exc:
        session.rollback()
        raise _http_error(exc) from exc
    except Exception:
        session.rollback()
        raise


@router.get("/connectors/profiles", response_model=FileConnectorProfilesResponse)
def list_connector_profiles(
    provider: str | None = None,
    campaign_id: str | None = None,
    include_disabled: bool = False,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:read",
            "files:file:upload",
            "files:file:download",
            "files:file:admin",
            "system:settings:read",
            "admin:settings:read",
        )
    ),
):
    try:
        if campaign_id:
            _ensure_campaign_file_access(session, principal, campaign_id)
        profiles = _visible_connector_profiles(
            session,
            principal,
            provider=provider,
            campaign_id=campaign_id,
            include_disabled=include_disabled
            and _can_read_disabled_connector_profiles(principal),
            include_admin_scopes=_can_read_disabled_connector_profiles(principal),
            include_effective_policy=False,
        )
    except (OSError, ValueError, json.JSONDecodeError) as exc:
        raise _http_error(exc) from exc
    return FileConnectorProfilesResponse(
        profiles=[
            FileConnectorProfileResponse(**profile.to_response())
            for profile in profiles
        ]
    )


@router.post(
    "/connectors/profiles",
    response_model=FileConnectorProfileResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_connector_profile(
    payload: FileConnectorProfileCreateRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(
        require_any_scope(
            "files:file:admin", "system:settings:write", "admin:settings:write"
        )
    ),
):
    _require_connector_profile_write(principal, payload.scope_type)
    credentials = payload.credentials
    try:
        credential_row = _credential_row_for_profile(
            session,
            principal,
            credential_profile_id=payload.credential_profile_id,
            provider=payload.provider,
            profile_id=payload.id,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
        )
        _ensure_connector_configuration_allowed(
            session,
            principal,
            connector_id=payload.id,
            credential_id=payload.credential_profile_id,
            provider=payload.provider,
            endpoint_url=connector_effective_endpoint_url(
                provider=payload.provider,
                endpoint_url=payload.endpoint_url,
                metadata=payload.metadata,
            ),
            base_path=payload.base_path,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
            operation="configure",
        )
        _ensure_connector_local_policy_allowed(
            session,
            principal,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
            policy=payload.policy,
        )
        row = create_connector_profile_row(
            session,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            profile_id=payload.id,
            label=payload.label,
            provider=payload.provider,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
            endpoint_url=payload.endpoint_url,
            base_path=payload.base_path,
            enabled=payload.enabled,
            credential_profile_id=payload.credential_profile_id,
            credential_mode=payload.credential_mode,
            username=credentials.username,
            password=credentials.password,
            token=credentials.token,
            password_env=credentials.password_env,
            token_env=credentials.token_env,
            secret_ref=credentials.secret_ref,
            capabilities=payload.capabilities,
            policy=payload.policy,
            metadata=payload.metadata,
        )
        _record_connector_settings_change(
            session,
            collection=FILES_CONNECTOR_PROFILES_COLLECTION,
            resource_type=FILES_CONNECTOR_PROFILE_RESOURCE,
            resource_id=row.id,
            operation="created",
            principal=principal,
            tenant_id=row.tenant_id,
            payload={
                "scope_type": row.scope_type,
                "scope_id": row.scope_id,
                "provider": row.provider,
            },
        )
        session.commit()
        session.refresh(row)
        return FileConnectorProfileResponse(
            **connector_profile_from_row(
                row, credential_row=credential_row
            ).to_response()
        )
    except ConnectorPolicyDenied as exc:
        session.rollback()
        raise _connector_policy_error(exc) from exc
    except (FileStorageError, ValueError, json.JSONDecodeError) as exc:
        session.rollback()
        raise _http_error(exc) from exc