[Feature] Add workspace doctor command for health diagnostics and autofix

extralit-server/src/extralit_server/api/handlers/v1/workspaces.py

@@ -26,14 +26,17 @@
 )
 from extralit_server.api.schemas.v1.workspaces import (
     WorkspaceCreate,
+    WorkspaceDoctorCheckResult,
+    WorkspaceDoctorResponse,
     Workspaces,
     WorkspaceUserCreate,
 )
 from extralit_server.contexts import accounts, files
 from extralit_server.database import get_async_db
 from extralit_server.errors import GenericServerError
 from extralit_server.errors.future import NotFoundError, NotUniqueError, UnprocessableEntityError
-from extralit_server.models import User, Workspace, WorkspaceUser
+from extralit_server.models import Dataset, User, Workspace, WorkspaceUser
+from extralit_server.search_engine import get_search_engine
 from extralit_server.security import auth
 
 router = APIRouter(tags=["workspaces"])
@@ -178,3 +181,201 @@ async def delete_workspace_user(
     await accounts.delete_workspace_user(db, workspace_user)
 
     return await workspace_user.awaitable_attrs.user
+
+
+@router.post("/workspaces/{workspace_id}/doctor", response_model=WorkspaceDoctorResponse)
+async def workspace_doctor(
+    *,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    workspace_id: UUID,
+    current_user: Annotated[User, Security(auth.get_current_user)],
+    s3_client=Depends(files.get_s3_client),
+    autofix: bool = True,
+):
+    """
+    Run diagnostics on a workspace and optionally auto-fix issues.
+
+    Checks:
+    - S3 bucket exists (can auto-fix)
+    - Bucket has proper versioning policy (informational)
+    - RQ worker pool connectivity (informational)
+    """
+    await authorize(current_user, WorkspacePolicy.get(workspace_id))
+
+    workspace = await Workspace.get_or_raise(db, workspace_id)
+    checks = []
+
+    # Check 1: S3 bucket exists
+    bucket_exists = await files.bucket_exists(s3_client, workspace.name)
+    if bucket_exists:
+        checks.append(
+            WorkspaceDoctorCheckResult(
+                check_name="s3_bucket",
+                status="ok",
+                message=f"S3 bucket '{workspace.name}' exists",
+                fixed=False,
+            )
+        )
+    else:
+        if autofix:
+            try:
+                await files.create_bucket(s3_client, workspace.name)
+                checks.append(
+                    WorkspaceDoctorCheckResult(
+                        check_name="s3_bucket",
+                        status="ok",
+                        message=f"S3 bucket '{workspace.name}' was missing and has been created",
+                        fixed=True,
+                    )
+                )
+            except Exception as e:
+                checks.append(
+                    WorkspaceDoctorCheckResult(
+                        check_name="s3_bucket",
+                        status="error",
+                        message=f"S3 bucket '{workspace.name}' does not exist and failed to create: {e!s}",
+                        fixed=False,
+                    )
+                )
+        else:
+            checks.append(
+                WorkspaceDoctorCheckResult(
+                    check_name="s3_bucket",
+                    status="error",
+                    message=f"S3 bucket '{workspace.name}' does not exist (autofix disabled)",
+                    fixed=False,
+                )
+            )
+
+    # Check 2: Bucket versioning policy
+    if bucket_exists or any(check.check_name == "s3_bucket" and check.fixed for check in checks):
+        versioning = await files.get_bucket_versioning(s3_client, workspace.name)
+        if versioning:
+            if versioning["status"] == "Enabled":
+                checks.append(
+                    WorkspaceDoctorCheckResult(
+                        check_name="bucket_versioning",
+                        status="ok",
+                        message=f"Bucket versioning is enabled (Status: {versioning['status']})",
+                        fixed=False,
+                    )
+                )
+            else:
+                checks.append(
+                    WorkspaceDoctorCheckResult(
+                        check_name="bucket_versioning",
+                        status="warning",
+                        message=f"Bucket versioning is not enabled (Status: {versioning['status']})",
+                        fixed=False,
+                    )
+                )
+        else:
+            checks.append(
+                WorkspaceDoctorCheckResult(
+                    check_name="bucket_versioning",
+                    status="warning",
+                    message="Could not retrieve bucket versioning configuration",
+                    fixed=False,
+                )
+            )
+
+    # Check 3: RQ worker pool connectivity
+    try:
+        from extralit_server.jobs.queues import DEFAULT_QUEUE
+
+        # Try to ping Redis through the queue connection
+        connection = DEFAULT_QUEUE.connection
+        connection.ping()
+
+        checks.append(
+            WorkspaceDoctorCheckResult(
+                check_name="rq_worker_pool",
+                status="ok",
+                message="Redis Queue worker pool is reachable",
+                fixed=False,
+            )
+        )
+    except Exception as e:
+        checks.append(
+            WorkspaceDoctorCheckResult(
+                check_name="rq_worker_pool",
+                status="warning",
+                message=f"Could not connect to RQ worker pool: {e!s}",
+                fixed=False,
+            )
+        )
+
+    # Check 4: Elasticsearch indexes for datasets (informational only)
+    try:
+        # Get datasets for this workspace
+        from sqlalchemy import select
+
+        result = await db.execute(select(Dataset).where(Dataset.workspace_id == workspace.id))
+        datasets = result.scalars().all()
+
+        if datasets:
+            async with get_search_engine() as search_engine:
+                missing_indexes = []
+                for dataset in datasets:
+                    index_name = f"ex.{dataset.id}"
+                    index_exists = await search_engine._index_exists_request(index_name)
+                    if not index_exists:
+                        missing_indexes.append(dataset.name)
+
+                if missing_indexes:
+                    checks.append(
+                        WorkspaceDoctorCheckResult(
+                            check_name="elasticsearch_indexes",
+                            status="warning",
+                            message=f"Missing Elasticsearch indexes for {len(missing_indexes)} dataset(s): {', '.join(missing_indexes[:3])}{'...' if len(missing_indexes) > 3 else ''}",
+                            fixed=False,
+                        )
+                    )
+                else:
+                    checks.append(
+                        WorkspaceDoctorCheckResult(
+                            check_name="elasticsearch_indexes",
+                            status="ok",
+                            message=f"All {len(datasets)} dataset(s) have Elasticsearch indexes",
+                            fixed=False,
+                        )
+                    )
+        else:
+            checks.append(
+                WorkspaceDoctorCheckResult(
+                    check_name="elasticsearch_indexes",
+                    status="ok",
+                    message="No datasets found for this workspace",
+                    fixed=False,
+                )
+            )
+    except Exception as e:
+        checks.append(
+            WorkspaceDoctorCheckResult(
+                check_name="elasticsearch_indexes",
+                status="warning",
+                message=f"Could not check Elasticsearch indexes: {e!s}",
+                fixed=False,
+            )
+        )
+
+    # Determine overall status
+    has_errors = any(check.status == "error" for check in checks)
+    has_fixed = any(check.fixed for check in checks)
+    has_warnings = any(check.status == "warning" for check in checks)
+
+    if has_errors:
+        overall_status = "issues_found"
+    elif has_fixed:
+        overall_status = "issues_fixed"
+    elif has_warnings:
+        overall_status = "issues_found"
+    else:
+        overall_status = "healthy"
+
+    return WorkspaceDoctorResponse(
+        workspace_id=workspace.id,
+        workspace_name=workspace.name,
+        checks=checks,
+        overall_status=overall_status,
+    )

extralit-server/src/extralit_server/api/schemas/v1/workspaces.py

@@ -38,3 +38,17 @@ class Workspaces(BaseModel):
 
 class WorkspaceUserCreate(BaseModel):
     user_id: UUID
+
+
+class WorkspaceDoctorCheckResult(BaseModel):
+    check_name: str
+    status: str  # "ok", "warning", "error"
+    message: str
+    fixed: bool = False
+
+
+class WorkspaceDoctorResponse(BaseModel):
+    workspace_id: UUID
+    workspace_name: str
+    checks: list[WorkspaceDoctorCheckResult]
+    overall_status: str  # "healthy", "issues_found", "issues_fixed"

extralit-server/src/extralit_server/contexts/files.py

@@ -357,6 +357,38 @@ async def delete_object(s3_client, bucket: str, object: str, version_id: str | N
         raise HTTPException(status_code=500, detail=f"Internal server error: {e!s}")
 
 
+async def bucket_exists(s3_client: "S3Client", bucket_name: str) -> bool:
+    """Check if S3 bucket exists."""
+    try:
+        await s3_client.head_bucket(Bucket=bucket_name)
+        return True
+    except ClientError as e:
+        if e.response["Error"]["Code"] in ["404", "NoSuchBucket"]:
+            return False
+        # For other errors (like permissions), log and return False
+        _LOGGER.warning(f"Error checking bucket {bucket_name}: {e}")
+        return False
+    except Exception as e:
+        _LOGGER.warning(f"Unexpected error checking bucket {bucket_name}: {e}")
+        return False
+
+
+async def get_bucket_versioning(s3_client: "S3Client", bucket_name: str) -> dict[str, str] | None:
+    """Get bucket versioning configuration."""
+    try:
+        response = await s3_client.get_bucket_versioning(Bucket=bucket_name)
+        return {
+            "status": response.get("Status", "Disabled"),
+            "mfa_delete": response.get("MFADelete", "Disabled"),
+        }
+    except ClientError as e:
+        _LOGGER.error(f"Error getting bucket versioning for {bucket_name}: {e}")
+        return None
+    except Exception as e:
+        _LOGGER.error(f"Unexpected error getting bucket versioning for {bucket_name}: {e}")
+        return None
+
+
 async def create_bucket(
     s3_client: "S3Client",
     workspace_name: str,