import ipaddress
import re
import uuid
from typing import Annotated, Literal

from fastapi import APIRouter, Cookie, Header, Request
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from sqlalchemy import func, select

from .auth import AuthDep, SessionDep, _audit, _require_csrf
from .authorization import authorize
from .compute_provider import ComputeProviderDep
from .computer_provisioning import ComputerProvisioningService
from .errors import AppError
from .models import (
    Agent,
    Computer,
    ComputerMetric,
    ComputerNetworkPolicy,
    ComputerOperation,
    ComputerSnapshot,
    ComputerTemplate,
    DeviceSession,
    WorkspaceComputerLimit,
)
from .security import canonical_payload_hash

router = APIRouter(prefix="/api/v1/computers", tags=["computers"])
IDEMPOTENCY_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{8,180}$")
ACTIVE_COMPUTER_STATES = {
    "provisioning", "starting", "running", "idle", "suspending", "stopping",
    "restarting", "recovering", "destroying",
}
OPERATION_RISK = {
    "create": "medium",
    "start": "low",
    "stop": "medium",
    "restart": "medium",
    "pause": "low",
    "resume": "low",
    "destroy": "high",
    "snapshot": "medium",
    "restore": "high",
    "resize": "high",
    "refresh": "low",
}
LIFECYCLE_ALLOWED_STATES = {
    "start": {"stopped"},
    "stop": {"running", "idle", "suspended"},
    "restart": {"running", "idle"},
    "pause": {"running", "idle"},
    "resume": {"suspended"},
    "destroy": {"unconfigured", "stopped", "suspended", "failed"},
}


class LifecycleInput(BaseModel):
    confirm_computer_name: str | None = Field(default=None, max_length=160)


class SnapshotInput(BaseModel):
    name: str = Field(min_length=1, max_length=160)
    includes_memory: bool = False

    @field_validator("name")
    @classmethod
    def strip_name(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("Snapshot name cannot be blank")
        return value


class RestoreInput(BaseModel):
    confirm: bool = False


class ResizeInput(BaseModel):
    cpu_cores: int = Field(ge=1, le=64)
    memory_mb: int = Field(ge=512, le=524288)
    disk_gb: int = Field(ge=10, le=16384)
    confirm: bool = False


class NetworkPolicyInput(BaseModel):
    model_config = ConfigDict(extra="forbid")
    internet_access: bool = True
    lan_access: bool = False
    platform_api_access: bool = True
    other_agent_networks: bool = False
    allowed_domains: list[str] = Field(default_factory=list, max_length=200)
    blocked_domains: list[str] = Field(default_factory=list, max_length=200)

    @field_validator("allowed_domains", "blocked_domains")
    @classmethod
    def validate_domains(cls, values: list[str]) -> list[str]:
        normalized: list[str] = []
        for raw in values:
            value = raw.strip().lower().rstrip(".")
            if (
                not value
                or len(value) > 253
                or "://" in value
                or "/" in value
                or "@" in value
            ):
                raise ValueError("Network policy entries must be DNS hostnames")
            try:
                address = ipaddress.ip_address(value)
            except ValueError:
                address = None
            if address is not None and not address.is_global:
                raise ValueError("Private and special-purpose addresses are not allowed")
            if value not in normalized:
                normalized.append(value)
        return normalized

    @model_validator(mode="after")
    def disjoint_lists(self):
        if set(self.allowed_domains) & set(self.blocked_domains):
            raise ValueError("A domain cannot be both allowed and blocked")
        return self


class TemplateConfigurationInput(BaseModel):
    provider_template_ref: str = Field(min_length=1, max_length=255)
    image_digest: str = Field(min_length=8, max_length=160)
    windows_license_confirmed: bool = False

    @field_validator("provider_template_ref", "image_digest")
    @classmethod
    def strip_values(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("Template configuration cannot be blank")
        return value


class LimitInput(BaseModel):
    maximum_computers: int = Field(ge=1, le=1000)
    maximum_running_computers: int = Field(ge=1, le=1000)
    maximum_cpu_per_computer: int = Field(ge=1, le=64)
    maximum_memory_mb: int = Field(ge=512, le=524288)
    maximum_disk_gb: int = Field(ge=10, le=16384)

    @model_validator(mode="after")
    def running_not_above_total(self):
        if self.maximum_running_computers > self.maximum_computers:
            raise ValueError("Running computer limit cannot exceed the computer limit")
        return self


async def _csrf(
    session: SessionDep, auth: AuthDep, cookie: str | None, header: str | None
) -> None:
    _require_csrf(await session.get(DeviceSession, auth.session_id), cookie, header)


def _require_idempotency(value: str | None) -> str:
    if not value or not IDEMPOTENCY_PATTERN.fullmatch(value):
        raise AppError(
            "IDEMPOTENCY_KEY_REQUIRED",
            "A stable Idempotency-Key header is required for computer operations.",
            400,
        )
    return value


async def _computer(
    session: SessionDep, workspace_id: uuid.UUID, computer_id: uuid.UUID, *, lock: bool = False
) -> Computer:
    statement = select(Computer).where(
        Computer.workspace_id == workspace_id, Computer.id == computer_id
    )
    if lock:
        statement = statement.with_for_update()
    record = await session.scalar(statement)
    if not record:
        raise AppError("COMPUTER_NOT_FOUND", "The computer is unavailable.", 404)
    return record


async def _policy(
    session: SessionDep, workspace_id: uuid.UUID, computer_id: uuid.UUID, *, lock: bool = False
) -> ComputerNetworkPolicy:
    statement = select(ComputerNetworkPolicy).where(
        ComputerNetworkPolicy.workspace_id == workspace_id,
        ComputerNetworkPolicy.computer_id == computer_id,
    )
    if lock:
        statement = statement.with_for_update()
    policy = await session.scalar(statement)
    if not policy:
        raise AppError("COMPUTER_POLICY_MISSING", "The computer network policy is missing.", 409)
    return policy


async def _limits(
    session: SessionDep, workspace_id: uuid.UUID, *, lock: bool = False
) -> WorkspaceComputerLimit:
    statement = select(WorkspaceComputerLimit).where(
        WorkspaceComputerLimit.workspace_id == workspace_id
    )
    if lock:
        statement = statement.with_for_update()
    limits = await session.scalar(statement)
    if not limits:
        raise AppError("COMPUTER_LIMITS_MISSING", "Computer resource limits are missing.", 409)
    return limits


async def _agent_for_computer(
    session: SessionDep, workspace_id: uuid.UUID, computer_id: uuid.UUID
) -> Agent | None:
    return await session.scalar(select(Agent).where(
        Agent.workspace_id == workspace_id, Agent.computer_id == computer_id
    ))


def _template_data(template: ComputerTemplate) -> dict:
    return {
        "id": str(template.id),
        "template_key": template.template_key,
        "name": template.name,
        "os_family": template.os_family,
        "os_distribution": template.os_distribution,
        "os_version": template.os_version,
        "browser": template.browser,
        "minimum_cpu": template.minimum_cpu,
        "minimum_memory_mb": template.minimum_memory_mb,
        "minimum_disk_gb": template.minimum_disk_gb,
        "windows_license_required": template.windows_license_required,
        "status": template.status,
        "provider_configured": bool(template.provider_template_ref),
        "image_digest": template.image_digest,
    }


def _policy_data(policy: ComputerNetworkPolicy) -> dict:
    return {
        "internet_access": policy.internet_access,
        "lan_access": policy.lan_access,
        "platform_api_access": policy.platform_api_access,
        "other_agent_networks": policy.other_agent_networks,
        "host_management_access": False,
        "allowed_domains": policy.allowed_domains,
        "blocked_domains": policy.blocked_domains,
        "version": policy.version,
    }


def _computer_data(record: Computer, agent: Agent | None = None) -> dict:
    return {
        "id": str(record.id),
        "agent": ({"id": str(agent.id), "name": agent.name} if agent else None),
        "name": record.name,
        "os_family": record.os_family,
        "os_distribution": record.os_distribution,
        "os_version": record.os_version,
        "compute_provider": record.compute_provider,
        "provider_configured": bool(record.provider_resource_id),
        "cpu_cores": record.cpu_cores,
        "memory_mb": record.memory_mb,
        "disk_gb": record.disk_gb,
        "gpu": record.gpu,
        "browser": record.browser,
        "persistent_disk": record.persistent_disk,
        "system_privilege": record.system_privilege,
        "start_policy": record.start_policy,
        "status": record.status,
        "desired_state": record.desired_state,
        "ip_address": record.ip_address,
        "browser_status": record.browser_status,
        "current_application": record.current_application,
        "current_task": record.current_task,
        "provider_error_code": record.provider_error_code,
        "provider_error_message": record.provider_error_message,
        "last_seen_at": record.last_seen_at,
        "started_at": record.started_at,
        "stopped_at": record.stopped_at,
        "created_at": record.created_at,
        "updated_at": record.updated_at,
        "version": record.version,
    }


def _operation_data(operation: ComputerOperation) -> dict:
    return {
        "id": str(operation.id),
        "computer_id": str(operation.computer_id),
        "operation": operation.operation,
        "risk": operation.risk,
        "status": operation.status,
        "verification": operation.verification,
        "error_code": operation.error_code,
        "error_message": operation.error_message,
        "started_at": operation.started_at,
        "completed_at": operation.completed_at,
        "created_at": operation.created_at,
    }


async def _new_operation(
    session: SessionDep,
    *,
    auth: AuthDep,
    record: Computer,
    operation_name: str,
    idempotency_key: str,
    payload: dict,
) -> tuple[ComputerOperation, bool]:
    request_hash = canonical_payload_hash({
        "computer_id": str(record.id), "operation": operation_name, "payload": payload
    })
    existing = await session.scalar(select(ComputerOperation).where(
        ComputerOperation.workspace_id == auth.workspace_id,
        ComputerOperation.idempotency_key == idempotency_key,
    ))
    if existing:
        if existing.request_hash != request_hash:
            raise AppError(
                "IDEMPOTENCY_KEY_REUSED",
                "The idempotency key was already used for a different computer operation.",
                409,
            )
        return existing, False
    operation = ComputerOperation(
        id=uuid.uuid4(),
        workspace_id=auth.workspace_id,
        computer_id=record.id,
        requested_by_user_id=auth.user_id,
        operation=operation_name,
        risk=OPERATION_RISK[operation_name],
        idempotency_key=idempotency_key,
        request_hash=request_hash,
        request_redacted=payload,
        status="pending",
    )
    session.add(operation)
    return operation, True


async def _dispatch(
    *,
    session: SessionDep,
    request: Request,
    auth: AuthDep,
    provider: ComputeProviderDep,
    record: Computer,
    operation: ComputerOperation,
    dispatch,
) -> dict:
    await _audit(
        session,
        workspace_id=auth.workspace_id,
        actor_id=auth.user_id,
        event_type="computer.operation_requested",
        request_id=request.state.request_id,
        data={
            "computer_id": str(record.id),
            "operation_id": str(operation.id),
            "operation": operation.operation,
            "risk": operation.risk,
        },
    )
    await session.commit()
    service = ComputerProvisioningService(provider)
    await dispatch(service)
    await _audit(
        session,
        workspace_id=auth.workspace_id,
        actor_id=auth.user_id,
        event_type=f"computer.operation_{operation.status}",
        request_id=request.state.request_id,
        data={
            "computer_id": str(record.id),
            "operation_id": str(operation.id),
            "operation": operation.operation,
            "status": operation.status,
            "error_code": operation.error_code,
        },
    )
    await session.commit()
    await session.refresh(record)
    await session.refresh(operation)
    return {
        "success": True,
        "computer": _computer_data(record, await _agent_for_computer(
            session, auth.workspace_id, record.id
        )),
        "operation": _operation_data(operation),
    }


@router.get("")
async def list_computers(auth: AuthDep, session: SessionDep, provider: ComputeProviderDep):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.observe",
    )
    records = list((await session.scalars(select(Computer).where(
        Computer.workspace_id == auth.workspace_id,
        Computer.status != "destroyed",
    ).order_by(Computer.created_at, Computer.name))).all())
    agents = {
        agent.computer_id: agent
        for agent in (await session.scalars(select(Agent).where(
            Agent.workspace_id == auth.workspace_id, Agent.computer_id.is_not(None)
        ))).all()
    }
    return {
        "success": True,
        "provider_state": "configured" if provider.configured else "unconfigured",
        "computers": [_computer_data(record, agents.get(record.id)) for record in records],
    }


@router.get("/templates")
async def list_templates(auth: AuthDep, session: SessionDep, provider: ComputeProviderDep):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.observe",
    )
    templates = list((await session.scalars(select(ComputerTemplate).where(
        ComputerTemplate.workspace_id == auth.workspace_id,
        ComputerTemplate.status != "disabled",
    ).order_by(ComputerTemplate.os_family, ComputerTemplate.name))).all())
    return {
        "success": True,
        "provider_state": "configured" if provider.configured else "unconfigured",
        "templates": [_template_data(item) for item in templates],
    }


@router.put("/templates/{template_id}")
async def configure_template(
    template_id: uuid.UUID,
    payload: TemplateConfigurationInput,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="settings.modify",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    template = await session.scalar(select(ComputerTemplate).where(
        ComputerTemplate.workspace_id == auth.workspace_id,
        ComputerTemplate.id == template_id,
    ).with_for_update())
    if not template:
        raise AppError("COMPUTER_TEMPLATE_NOT_FOUND", "The OS template is unavailable.", 404)
    if template.windows_license_required and not payload.windows_license_confirmed:
        raise AppError(
            "WINDOWS_LICENSE_CONFIRMATION_REQUIRED",
            "Confirm that the configured Windows image is properly licensed.",
            409,
        )
    template.provider_template_ref = payload.provider_template_ref
    template.image_digest = payload.image_digest
    template.status = "available"
    await _audit(
        session, workspace_id=auth.workspace_id, actor_id=auth.user_id,
        event_type="computer.template_configured", request_id=request.state.request_id,
        data={"template_id": str(template.id), "template_key": template.template_key},
    )
    await session.commit()
    return {"success": True, "template": _template_data(template)}


@router.get("/limits")
async def get_limits(auth: AuthDep, session: SessionDep):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="settings.read",
    )
    limits = await _limits(session, auth.workspace_id)
    return {"success": True, "limits": {
        "maximum_computers": limits.maximum_computers,
        "maximum_running_computers": limits.maximum_running_computers,
        "maximum_cpu_per_computer": limits.maximum_cpu_per_computer,
        "maximum_memory_mb": limits.maximum_memory_mb,
        "maximum_disk_gb": limits.maximum_disk_gb,
    }}


@router.put("/limits")
async def update_limits(
    payload: LimitInput, request: Request, auth: AuthDep, session: SessionDep,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="settings.modify",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    limits = await _limits(session, auth.workspace_id, lock=True)
    current_count = await session.scalar(select(func.count(Computer.id)).where(
        Computer.workspace_id == auth.workspace_id, Computer.status != "destroyed"
    ))
    if (current_count or 0) > payload.maximum_computers:
        raise AppError(
            "COMPUTER_LIMIT_BELOW_USAGE",
            "The new maximum is below the number of existing computers.",
            409,
        )
    for key, value in payload.model_dump().items():
        setattr(limits, key, value)
    await _audit(
        session, workspace_id=auth.workspace_id, actor_id=auth.user_id,
        event_type="computer.limits_updated", request_id=request.state.request_id,
        data=payload.model_dump(),
    )
    await session.commit()
    return {"success": True, "limits": payload.model_dump()}


@router.get("/{computer_id}")
async def get_computer(computer_id: uuid.UUID, auth: AuthDep, session: SessionDep):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.observe",
    )
    record = await _computer(session, auth.workspace_id, computer_id)
    policy = await _policy(session, auth.workspace_id, computer_id)
    agent = await _agent_for_computer(session, auth.workspace_id, computer_id)
    snapshots = list((await session.scalars(select(ComputerSnapshot).where(
        ComputerSnapshot.workspace_id == auth.workspace_id,
        ComputerSnapshot.computer_id == computer_id,
        ComputerSnapshot.status != "deleted",
    ).order_by(ComputerSnapshot.created_at.desc()).limit(100))).all())
    operations = list((await session.scalars(select(ComputerOperation).where(
        ComputerOperation.workspace_id == auth.workspace_id,
        ComputerOperation.computer_id == computer_id,
    ).order_by(ComputerOperation.created_at.desc()).limit(100))).all())
    metric = await session.scalar(select(ComputerMetric).where(
        ComputerMetric.workspace_id == auth.workspace_id,
        ComputerMetric.computer_id == computer_id,
    ).order_by(ComputerMetric.recorded_at.desc()).limit(1))
    return {
        "success": True,
        "computer": _computer_data(record, agent),
        "network_policy": _policy_data(policy),
        "snapshots": [{
            "id": str(item.id), "name": item.name, "status": item.status,
            "includes_memory": item.includes_memory, "automatic": item.automatic,
            "created_at": item.created_at,
        } for item in snapshots],
        "operations": [_operation_data(item) for item in operations],
        "metrics": ({
            "cpu_percent": metric.cpu_percent,
            "memory_used_mb": metric.memory_used_mb,
            "disk_used_gb": metric.disk_used_gb,
            "network_rx_bytes": metric.network_rx_bytes,
            "network_tx_bytes": metric.network_tx_bytes,
            "top_processes": metric.top_processes,
            "recorded_at": metric.recorded_at,
        } if metric else None),
    }


@router.put("/{computer_id}/network-policy")
async def update_network_policy(
    computer_id: uuid.UUID,
    payload: NetworkPolicyInput,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.control",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    record = await _computer(session, auth.workspace_id, computer_id, lock=True)
    if record.status not in {"unconfigured", "stopped", "suspended", "failed"}:
        raise AppError(
            "COMPUTER_POLICY_CHANGE_UNSAFE",
            "Stop or suspend the computer before changing its network policy.",
            409,
        )
    policy = await _policy(session, auth.workspace_id, computer_id, lock=True)
    for key, value in payload.model_dump().items():
        setattr(policy, key, value)
    policy.host_management_access = False
    policy.version += 1
    await _audit(
        session, workspace_id=auth.workspace_id, actor_id=auth.user_id,
        event_type="computer.network_policy_updated", request_id=request.state.request_id,
        data={
            "computer_id": str(computer_id),
            "internet_access": policy.internet_access,
            "lan_access": policy.lan_access,
            "other_agent_networks": policy.other_agent_networks,
        },
    )
    await session.commit()
    return {"success": True, "network_policy": _policy_data(policy)}


@router.post("/{computer_id}/provision")
async def provision_computer(
    computer_id: uuid.UUID,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    provider: ComputeProviderDep,
    idempotency_header: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.control",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    key = _require_idempotency(idempotency_header)
    record = await _computer(session, auth.workspace_id, computer_id, lock=True)
    operation, created = await _new_operation(
        session, auth=auth, record=record, operation_name="create",
        idempotency_key=key, payload={},
    )
    if not created:
        return {"success": True, "computer": _computer_data(record),
                "operation": _operation_data(operation), "replayed": True}
    if record.provider_resource_id or record.status not in {"unconfigured", "pending", "failed"}:
        raise AppError("COMPUTER_ALREADY_PROVISIONED", "The computer is already provisioned.", 409)
    template = await session.scalar(select(ComputerTemplate).where(
        ComputerTemplate.workspace_id == auth.workspace_id,
        ComputerTemplate.id == record.template_id,
    ))
    policy = await _policy(session, auth.workspace_id, computer_id)
    return await _dispatch(
        session=session, request=request, auth=auth, provider=provider, record=record,
        operation=operation,
        dispatch=lambda service: service.create_computer(record, template, policy, operation),
    )


@router.post("/{computer_id}/actions/{action}")
async def computer_lifecycle(
    computer_id: uuid.UUID,
    action: Literal["start", "stop", "restart", "pause", "resume", "destroy", "refresh"],
    payload: LifecycleInput,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    provider: ComputeProviderDep,
    idempotency_header: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.control",
    )
    if action == "destroy":
        await authorize(
            session, workspace_id=auth.workspace_id, user_id=auth.user_id,
            permission="approvals.approve",
        )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    key = _require_idempotency(idempotency_header)
    record = await _computer(session, auth.workspace_id, computer_id, lock=True)
    if action == "destroy" and payload.confirm_computer_name != record.name:
        raise AppError(
            "COMPUTER_DESTROY_CONFIRMATION_REQUIRED",
            "Enter the exact computer name to approve permanent destruction.",
            409,
        )
    operation, created = await _new_operation(
        session, auth=auth, record=record, operation_name=action,
        idempotency_key=key,
        payload={"confirmed": action != "destroy" or bool(payload.confirm_computer_name)},
    )
    if not created:
        return {"success": True, "computer": _computer_data(record),
                "operation": _operation_data(operation), "replayed": True}
    allowed_states = LIFECYCLE_ALLOWED_STATES.get(action)
    if allowed_states is not None and record.status not in allowed_states:
        raise AppError(
            "COMPUTER_LIFECYCLE_STATE_INVALID",
            f"The computer cannot {action} while it is {record.status}.",
            409,
        )
    if action in {"start", "resume", "restart"}:
        limits = await _limits(session, auth.workspace_id, lock=True)
        running = await session.scalar(select(func.count(Computer.id)).where(
            Computer.workspace_id == auth.workspace_id,
            Computer.id != record.id,
            Computer.status.in_(ACTIVE_COMPUTER_STATES),
        ))
        if (running or 0) >= limits.maximum_running_computers:
            raise AppError(
                "COMPUTER_RUNNING_LIMIT_REACHED",
                "The workspace running-computer limit has been reached.",
                409,
            )
    if action == "refresh":
        dispatch = lambda service: service.refresh(record, operation)
    else:
        dispatch = lambda service: service.lifecycle(record, operation)
    return await _dispatch(
        session=session, request=request, auth=auth, provider=provider, record=record,
        operation=operation, dispatch=dispatch,
    )


@router.post("/{computer_id}/snapshots", status_code=201)
async def create_snapshot(
    computer_id: uuid.UUID,
    payload: SnapshotInput,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    provider: ComputeProviderDep,
    idempotency_header: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.control",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    key = _require_idempotency(idempotency_header)
    record = await _computer(session, auth.workspace_id, computer_id, lock=True)
    operation, created = await _new_operation(
        session, auth=auth, record=record, operation_name="snapshot",
        idempotency_key=key, payload=payload.model_dump(),
    )
    if not created:
        snapshot = await session.scalar(select(ComputerSnapshot).where(
            ComputerSnapshot.workspace_id == auth.workspace_id,
            ComputerSnapshot.metadata_redacted["operation_id"].as_string() == str(operation.id),
        ))
        return {"success": True, "operation": _operation_data(operation),
                "snapshot_id": str(snapshot.id) if snapshot else None, "replayed": True}
    snapshot = ComputerSnapshot(
        id=uuid.uuid4(), workspace_id=auth.workspace_id, computer_id=record.id,
        created_by_user_id=auth.user_id, name=payload.name,
        includes_memory=payload.includes_memory, automatic=False,
        status="creating", metadata_redacted={"operation_id": str(operation.id)},
    )
    session.add(snapshot)
    result = await _dispatch(
        session=session, request=request, auth=auth, provider=provider, record=record,
        operation=operation,
        dispatch=lambda service: service.create_snapshot(record, snapshot, operation),
    )
    result["snapshot_id"] = str(snapshot.id)
    return result


@router.post("/{computer_id}/snapshots/{snapshot_id}/restore")
async def restore_snapshot(
    computer_id: uuid.UUID,
    snapshot_id: uuid.UUID,
    payload: RestoreInput,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    provider: ComputeProviderDep,
    idempotency_header: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.control",
    )
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="approvals.approve",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    if not payload.confirm:
        raise AppError(
            "COMPUTER_RESTORE_CONFIRMATION_REQUIRED",
            "Explicitly confirm the destructive snapshot restore.",
            409,
        )
    key = _require_idempotency(idempotency_header)
    record = await _computer(session, auth.workspace_id, computer_id, lock=True)
    snapshot = await session.scalar(select(ComputerSnapshot).where(
        ComputerSnapshot.workspace_id == auth.workspace_id,
        ComputerSnapshot.computer_id == computer_id,
        ComputerSnapshot.id == snapshot_id,
        ComputerSnapshot.status == "ready",
    ).with_for_update())
    if not snapshot:
        raise AppError("COMPUTER_SNAPSHOT_NOT_FOUND", "The ready snapshot is unavailable.", 404)
    operation, created = await _new_operation(
        session, auth=auth, record=record, operation_name="restore",
        idempotency_key=key, payload={"snapshot_id": str(snapshot.id), "confirmed": True},
    )
    if not created:
        return {"success": True, "computer": _computer_data(record),
                "operation": _operation_data(operation), "replayed": True}
    return await _dispatch(
        session=session, request=request, auth=auth, provider=provider, record=record,
        operation=operation,
        dispatch=lambda service: service.restore_snapshot(record, snapshot, operation),
    )


@router.post("/{computer_id}/resize")
async def resize_computer(
    computer_id: uuid.UUID,
    payload: ResizeInput,
    request: Request,
    auth: AuthDep,
    session: SessionDep,
    provider: ComputeProviderDep,
    idempotency_header: Annotated[str | None, Header(alias="Idempotency-Key")] = None,
    hayva_csrf: Annotated[str | None, Cookie()] = None,
    x_csrf_token: Annotated[str | None, Header()] = None,
):
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="computer.control",
    )
    await authorize(
        session, workspace_id=auth.workspace_id, user_id=auth.user_id,
        permission="approvals.approve",
    )
    await _csrf(session, auth, hayva_csrf, x_csrf_token)
    if not payload.confirm:
        raise AppError(
            "COMPUTER_RESIZE_CONFIRMATION_REQUIRED",
            "Explicitly confirm the resource resize.",
            409,
        )
    key = _require_idempotency(idempotency_header)
    record = await _computer(session, auth.workspace_id, computer_id, lock=True)
    if record.status not in {"stopped", "suspended"}:
        raise AppError(
            "COMPUTER_RESIZE_STATE_INVALID",
            "Stop or suspend the computer before resizing it.",
            409,
        )
    limits = await _limits(session, auth.workspace_id, lock=True)
    if (
        payload.cpu_cores > limits.maximum_cpu_per_computer
        or payload.memory_mb > limits.maximum_memory_mb
        or payload.disk_gb > limits.maximum_disk_gb
    ):
        raise AppError(
            "COMPUTER_RESOURCE_LIMIT_EXCEEDED",
            "The requested resources exceed workspace limits.",
            409,
        )
    if payload.disk_gb < record.disk_gb:
        raise AppError("COMPUTER_DISK_SHRINK_UNSAFE", "Computer disks cannot be shrunk.", 409)
    spec = payload.model_dump(exclude={"confirm"})
    operation, created = await _new_operation(
        session, auth=auth, record=record, operation_name="resize",
        idempotency_key=key, payload={**spec, "confirmed": True},
    )
    if not created:
        return {"success": True, "computer": _computer_data(record),
                "operation": _operation_data(operation), "replayed": True}
    return await _dispatch(
        session=session, request=request, auth=auth, provider=provider, record=record,
        operation=operation,
        dispatch=lambda service: service.resize_computer(record, operation, spec),
    )