feat: gcs impl

app/api/server.py

@@ -109,6 +109,8 @@ async def lifespan(app: FastAPI):
     await close_oauth_service()
     from app.api.v1.google_maps import close_maps_service
     await close_maps_service()
+    from app.api.v1.gcs import close_gcs_service
+    await close_gcs_service()
     from app.services.supabase import get_supabase_client
     client = get_supabase_client()
     if client:

app/api/v1/gcs.py

@@ -0,0 +1,1024 @@
+from __future__ import annotations
+
+import base64
+import json
+import re
+import time
+from typing import Any, Dict, Optional
+
+from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, Query, Response, UploadFile
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.models.schemas import (
+    GCSACLRequest,
+    GCSBaseCredentialsRequest,
+    GCSBucketIAMRequest,
+    GCSBucketRequest,
+    GCSBucketRestoreRequest,
+    GCSBucketUpdateRequest,
+    GCSComposeRequest,
+    GCSDownloadResponse,
+    GCSGenericResponse,
+    GCSListResponse,
+    GCSObjectTransferRequest,
+    GCSObjectUpdateRequest,
+    GCSResumableUploadRequest,
+    GCSURLRequest,
+    GCSURLResponse,
+    GCSWatchRequest,
+)
+from app.services.gcs_service import GCSCredentials, GCSCredentialsError, GCSError, GCSService
+
+router = APIRouter(prefix="/google/gcs", tags=["Google Cloud Storage"])
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+_MAX_UPLOAD_BYTES = _settings.max_upload_bytes
+
+_gcs_service = GCSService()
+
+
+def get_gcs_service() -> GCSService:
+    return _gcs_service
+
+
+async def close_gcs_service() -> None:
+    await _gcs_service.close()
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers
+# ---------------------------------------------------------------------------
+
+def _elapsed_ms(start: float) -> float:
+    return round((time.perf_counter() - start) * 1000, 2)
+
+
+async def _resolve(
+    service: GCSService,
+    *,
+    payload: Optional[Any] = None,
+    url: Optional[str] = None,
+    file_bytes: Optional[bytes] = None,
+    file_name: Optional[str] = None,
+) -> GCSCredentials:
+    try:
+        return await service.resolve_credentials(
+            payload=payload, url=url, file_bytes=file_bytes, file_name=file_name
+        )
+    except GCSCredentialsError as exc:
+        raise HTTPException(status_code=400, detail=exc.message)
+
+
+def _creds_fields(body: Any) -> Dict[str, Any]:
+    """Extract credential kwargs from a GCS request model."""
+    payload = body.credentials if isinstance(body.credentials, dict) else body.credentials_json
+    return {"payload": payload, "url": body.credentials_url}
+
+
+def _error_response(cls, start: float, exc: GCSError):
+    return cls(success=False, time_ms=_elapsed_ms(start), error=exc.message)
+
+
+async def _read_file_limited(f: UploadFile, *, what: str) -> bytes:
+    data = await f.read()
+    if len(data) > _MAX_UPLOAD_BYTES:
+        raise HTTPException(
+            status_code=413,
+            detail=f"{what} exceeds the {_MAX_UPLOAD_BYTES} byte upload limit",
+        )
+    return data
+
+
+# ---------------------------------------------------------------------------
+# Buckets — CRUD
+# ---------------------------------------------------------------------------
+
+@router.post("/buckets", response_model=GCSGenericResponse,
+             summary="Create a Cloud Storage bucket")
+async def create_bucket(
+    body: GCSBucketRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    bucket_body: Dict[str, Any] = dict(body.bucket or {})
+    if body.location:
+        bucket_body["location"] = body.location
+    if body.storage_class:
+        bucket_body["storageClass"] = body.storage_class
+    try:
+        result = await service.create_bucket(creds, body.name, project=body.project_id, bucket_body=bucket_body)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}", response_model=GCSGenericResponse,
+            summary="Get bucket metadata")
+async def get_bucket(
+    bucket: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials", description="Service account JSON (or JSON string)"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url", description="URL to a service account JSON file"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.get_bucket(creds, bucket)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets", response_model=GCSListResponse,
+            summary="List buckets in a project")
+async def list_buckets(
+    project_id: Optional[str] = Query(None, description="GCP project id (defaults to the service account project)"),
+    prefix: Optional[str] = Query(None, description="Filter buckets whose names start with this prefix"),
+    max_results: Optional[int] = Query(None, ge=1, le=1000, description="Maximum number of results to return"),
+    page_token: Optional[str] = Query(None, description="Pagination token from a previous response"),
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials", description="Service account JSON (or JSON string)"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url", description="URL to a service account JSON file"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.list_buckets(creds, project=project_id, prefix=prefix,
+                                            max_results=max_results, page_token=page_token)
+        data = result.get("data") or {}
+        items = data.get("items", [])
+        return GCSListResponse(
+            success=True, time_ms=_elapsed_ms(start), count=len(items),
+            next_page_token=data.get("nextPageToken"),
+            items=items, prefixes=[],
+        )
+    except GCSError as exc:
+        return _error_response(GCSListResponse, start, exc)
+
+
+@router.patch("/buckets/{bucket}", response_model=GCSGenericResponse,
+              summary="Patch bucket metadata (partial update)")
+async def patch_bucket(
+    bucket: str,
+    body: GCSBucketUpdateRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.patch_bucket(creds, bucket, body.bucket)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.put("/buckets/{bucket}", response_model=GCSGenericResponse,
+            summary="Replace bucket metadata (full update)")
+async def update_bucket(
+    bucket: str,
+    body: GCSBucketUpdateRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.update_bucket(creds, bucket, body.bucket)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.delete("/buckets/{bucket}", response_model=GCSGenericResponse,
+               summary="Delete a bucket")
+async def delete_bucket(
+    bucket: str,
+    body: Optional[Dict[str, Any]] = None,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(
+        service,
+        payload=(body or {}).get("credentials_json") or (body or {}).get("credentials"),
+        url=(body or {}).get("credentials_url"),
+    )
+    try:
+        await service.delete_bucket(creds, bucket)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data={"deleted": bucket})
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+# ---------------------------------------------------------------------------
+# Buckets — IAM & permissions
+# ---------------------------------------------------------------------------
+
+@router.post("/buckets/{bucket}/restore", response_model=GCSGenericResponse,
+             summary="Restore a soft-deleted bucket")
+async def restore_bucket(
+    bucket: str,
+    body: GCSBucketRestoreRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.restore_bucket(creds, bucket, if_metageneration_match=body.if_metageneration_match)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}/iam", response_model=GCSGenericResponse,
+            summary="Get bucket IAM policy")
+async def get_bucket_iam(
+    bucket: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.get_bucket_iam(creds, bucket)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.put("/buckets/{bucket}/iam", response_model=GCSGenericResponse,
+            summary="Set bucket IAM policy")
+async def set_bucket_iam(
+    bucket: str,
+    body: GCSBucketIAMRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.set_bucket_iam(creds, bucket, body.policy)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}/iam/testPermissions", response_model=GCSGenericResponse,
+            summary="Test permissions on a bucket")
+async def test_bucket_permissions(
+    bucket: str,
+    permissions: str = Query(..., description="Comma-separated permissions to test, e.g. 'storage.buckets.get,storage.objects.list'"),
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    permission_list = [p.strip() for p in permissions.split(",") if p.strip()]
+    if not permission_list:
+        raise HTTPException(status_code=422, detail="At least one permission is required.")
+    try:
+        result = await service.test_bucket_permissions(creds, bucket, permission_list)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+# ---------------------------------------------------------------------------
+# Buckets — default object ACLs
+# ---------------------------------------------------------------------------
+
+@router.get("/buckets/{bucket}/defaultAcl", response_model=GCSGenericResponse,
+            summary="List default object ACLs")
+async def list_default_object_acl(
+    bucket: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.list_default_object_acl(creds, bucket)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/defaultAcl", response_model=GCSGenericResponse,
+             summary="Add a default object ACL entry")
+async def insert_default_object_acl(
+    bucket: str,
+    body: GCSACLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    acl = {"entity": body.entity, "role": body.role}
+    try:
+        result = await service.insert_default_object_acl(creds, bucket, acl)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}/defaultAcl/{entity}", response_model=GCSGenericResponse,
+            summary="Get a default object ACL entry")
+async def get_default_object_acl(
+    bucket: str,
+    entity: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.get_default_object_acl(creds, bucket, entity)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.patch("/buckets/{bucket}/defaultAcl/{entity}", response_model=GCSGenericResponse,
+              summary="Patch a default object ACL entry")
+async def patch_default_object_acl(
+    bucket: str,
+    entity: str,
+    body: GCSACLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    acl = {"entity": body.entity, "role": body.role}
+    try:
+        result = await service.patch_default_object_acl(creds, bucket, entity, acl)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.put("/buckets/{bucket}/defaultAcl/{entity}", response_model=GCSGenericResponse,
+            summary="Update a default object ACL entry")
+async def update_default_object_acl(
+    bucket: str,
+    entity: str,
+    body: GCSACLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    acl = {"entity": body.entity, "role": body.role}
+    try:
+        result = await service.update_default_object_acl(creds, bucket, entity, acl)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.delete("/buckets/{bucket}/defaultAcl/{entity}", response_model=GCSGenericResponse,
+               summary="Delete a default object ACL entry")
+async def delete_default_object_acl(
+    bucket: str,
+    entity: str,
+    body: Optional[Dict[str, Any]] = None,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(
+        service,
+        payload=(body or {}).get("credentials_json") or (body or {}).get("credentials"),
+        url=(body or {}).get("credentials_url"),
+    )
+    try:
+        await service.delete_default_object_acl(creds, bucket, entity)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data={"deleted": entity})
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+# ---------------------------------------------------------------------------
+# Objects — list / upload / download
+# ---------------------------------------------------------------------------
+
+@router.get("/buckets/{bucket}/objects", response_model=GCSListResponse,
+            summary="List objects in a bucket")
+async def list_objects(
+    bucket: str,
+    prefix: Optional[str] = Query(None, description="Filter objects whose names start with this prefix"),
+    delimiter: Optional[str] = Query(None, description="Fold object names at this delimiter (e.g. '/')"),
+    max_results: Optional[int] = Query(None, ge=1, le=1000, description="Maximum number of results"),
+    page_token: Optional[str] = Query(None, description="Pagination token from a previous response"),
+    versions: Optional[bool] = Query(None, description="Include object versions"),
+    match_glob: Optional[str] = Query(None, description="Glob pattern to filter object names"),
+    start_offset: Optional[str] = Query(None, description="Lexicographically first object to return"),
+    end_offset: Optional[str] = Query(None, description="Lexicographically last object to return"),
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.list_objects(
+            creds, bucket, prefix=prefix, delimiter=delimiter, max_results=max_results,
+            page_token=page_token, versions=versions, match_glob=match_glob,
+            start_offset=start_offset, end_offset=end_offset,
+        )
+        data = result.get("data") or {}
+        items = data.get("items", [])
+        return GCSListResponse(
+            success=True, time_ms=_elapsed_ms(start), count=len(items),
+            next_page_token=data.get("nextPageToken"),
+            items=items, prefixes=data.get("prefixes", []),
+        )
+    except GCSError as exc:
+        return _error_response(GCSListResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/upload", response_model=GCSGenericResponse,
+             summary="Upload an object from a file or a URL (up to the server upload limit)")
+async def upload_object(
+    bucket: str,
+    name: str = Form(..., min_length=1, max_length=1024, description="Destination object name"),
+    file: Optional[UploadFile] = File(None, description="Object content to upload"),
+    file_url: Optional[str] = Form(None, description="Publicly reachable URL whose content becomes the object"),
+    content_type: str = Form("application/octet-stream", description="Content-Type of the uploaded object"),
+    metadata_json: str = Form(None, description="Optional JSON object of object metadata to set on upload"),
+    if_generation_match: Optional[int] = Form(None, description="Generation match condition"),
+    credentials_file: Optional[UploadFile] = File(None, description="Service account JSON file (alternative to inline credentials)"),
+    credentials_json: Optional[str] = Form(None, description="Service account JSON as a string"),
+    credentials_url: Optional[str] = Form(None, description="URL to a service account JSON file"),
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+
+    if file is None and not file_url:
+        raise HTTPException(status_code=422, detail="Either 'file' or 'file_url' must be provided to upload an object.")
+    if file is not None and file_url:
+        raise HTTPException(status_code=422, detail="Provide either 'file' or 'file_url', not both.")
+    if not re.fullmatch(r"[\w./+-]{1,255}", content_type):
+        raise HTTPException(status_code=422, detail="content_type must be a valid media type.")
+    cleaned_name = name.strip()
+    if not cleaned_name or cleaned_name.startswith("/"):
+        raise HTTPException(status_code=422, detail="Object name must be non-empty and must not start with '/'.")
+
+    if credentials_file is not None:
+        cred_bytes = await _read_file_limited(credentials_file, what="credentials file")
+        creds = await _resolve(service, file_bytes=cred_bytes, file_name=credentials_file.filename)
+    else:
+        creds = await _resolve(service, payload=credentials_json, url=credentials_url)
+
+    if file_url:
+        try:
+            content, fetched_type = await service.fetch_url_content(file_url)
+        except GCSError as exc:
+            raise HTTPException(status_code=exc.status_code, detail=exc.message)
+        if len(content) > _MAX_UPLOAD_BYTES:
+            raise HTTPException(status_code=413, detail=f"Content from file_url exceeds the {_MAX_UPLOAD_BYTES} byte upload limit")
+        if content_type == "application/octet-stream":
+            content_type = fetched_type or content_type
+    else:
+        content = await _read_file_limited(file, what="file")
+
+    metadata: Dict[str, Any] = {}
+    if metadata_json:
+        try:
+            metadata = json.loads(metadata_json)
+            if not isinstance(metadata, dict):
+                raise ValueError("metadata_json must be a JSON object")
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise HTTPException(status_code=400, detail=f"Invalid metadata_json: {exc}")
+
+    try:
+        result = await service.upload_object(
+            creds, bucket, cleaned_name, content,
+            content_type=content_type, metadata=metadata,
+            if_generation_match=if_generation_match,
+        )
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/upload/resumable", response_model=GCSGenericResponse,
+             summary="Start a resumable upload session and return its upload URI")
+async def initiate_resumable_upload(
+    bucket: str,
+    body: GCSResumableUploadRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.initiate_resumable_upload(
+            creds, bucket, body.name,
+            content_type=body.content_type or "application/octet-stream",
+            object_metadata=body.object_metadata,
+        )
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data={"session_uri": result.get("session_uri")})
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/watch", response_model=GCSGenericResponse,
+             summary="Register an object-change notification channel for a bucket")
+async def watch_all_objects(
+    bucket: str,
+    body: GCSWatchRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.watch_all_objects(creds, bucket, body.channel)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}/objects/{object:path}/download", include_in_schema=False)
+async def download_object_raw(
+    bucket: str,
+    object: str,
+    generation: Optional[int] = Query(None, description="Specific object generation to download"),
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    """Download an object's raw bytes (returns the file directly)."""
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.download_object(creds, bucket, object, generation=generation)
+    except GCSError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.message)
+    content = result.get("data")
+    if not isinstance(content, bytes):
+        content = content or b""
+    content_type = result.get("content_type") or "application/octet-stream"
+    return Response(
+        content=content,
+        media_type=content_type,
+        headers={"Content-Disposition": f'attachment; filename="{object.split("/")[-1]}"'},
+    )
+
+
+@router.get("/buckets/{bucket}/objects/{object:path}/download/base64", response_model=GCSDownloadResponse,
+            summary="Download an object as base64 (JSON response)")
+async def download_object_base64(
+    bucket: str,
+    object: str,
+    generation: Optional[int] = Query(None, description="Specific object generation to download"),
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.download_object(creds, bucket, object, generation=generation)
+    except GCSError as exc:
+        return _error_response(GCSDownloadResponse, start, exc)
+    content = result.get("data")
+    content_bytes = content if isinstance(content, bytes) else (content or b"")
+    return GCSDownloadResponse(
+        success=True, time_ms=_elapsed_ms(start),
+        content_base64=base64.b64encode(content_bytes).decode("ascii"),
+        content_type=result.get("content_type"),
+        size_bytes=len(content_bytes),
+    )
+
+
+@router.get("/buckets/{bucket}/objects/{object:path}/public-url", response_model=GCSURLResponse,
+            summary="Generate the public URL for an object")
+async def public_url(
+    bucket: str,
+    object: str,
+):
+    start = time.perf_counter()
+    url = GCSService.public_url(bucket, object)
+    return GCSURLResponse(success=True, time_ms=_elapsed_ms(start), url=url, method="GET")
+
+
+# ---------------------------------------------------------------------------
+# Objects — IAM
+# ---------------------------------------------------------------------------
+
+@router.get("/buckets/{bucket}/objects/{object:path}/iam", response_model=GCSGenericResponse,
+            summary="Get object IAM policy")
+async def get_object_iam(
+    bucket: str,
+    object: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.get_object_iam(creds, bucket, object)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.put("/buckets/{bucket}/objects/{object:path}/iam", response_model=GCSGenericResponse,
+            summary="Set object IAM policy")
+async def set_object_iam(
+    bucket: str,
+    object: str,
+    body: GCSBucketIAMRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.set_object_iam(creds, bucket, object, body.policy)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+# ---------------------------------------------------------------------------
+# Objects — ACLs
+# ---------------------------------------------------------------------------
+
+@router.get("/buckets/{bucket}/objects/{object:path}/acl", response_model=GCSGenericResponse,
+            summary="List object ACLs")
+async def list_object_acl(
+    bucket: str,
+    object: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.list_object_acl(creds, bucket, object)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/acl", response_model=GCSGenericResponse,
+             summary="Add an object ACL entry")
+async def insert_object_acl(
+    bucket: str,
+    object: str,
+    body: GCSACLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    acl = {"entity": body.entity, "role": body.role}
+    try:
+        result = await service.insert_object_acl(creds, bucket, object, acl)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}/objects/{object:path}/acl/{entity}", response_model=GCSGenericResponse,
+            summary="Get an object ACL entry")
+async def get_object_acl(
+    bucket: str,
+    object: str,
+    entity: str,
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.get_object_acl(creds, bucket, object, entity)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.patch("/buckets/{bucket}/objects/{object:path}/acl/{entity}", response_model=GCSGenericResponse,
+              summary="Patch an object ACL entry")
+async def patch_object_acl(
+    bucket: str,
+    object: str,
+    entity: str,
+    body: GCSACLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    acl = {"entity": body.entity, "role": body.role}
+    try:
+        result = await service.patch_object_acl(creds, bucket, object, entity, acl)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.put("/buckets/{bucket}/objects/{object:path}/acl/{entity}", response_model=GCSGenericResponse,
+            summary="Update an object ACL entry")
+async def update_object_acl(
+    bucket: str,
+    object: str,
+    entity: str,
+    body: GCSACLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    acl = {"entity": body.entity, "role": body.role}
+    try:
+        result = await service.update_object_acl(creds, bucket, object, entity, acl)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.delete("/buckets/{bucket}/objects/{object:path}/acl/{entity}", response_model=GCSGenericResponse,
+               summary="Delete an object ACL entry")
+async def delete_object_acl(
+    bucket: str,
+    object: str,
+    entity: str,
+    body: Optional[Dict[str, Any]] = None,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    payload = (body or {}).get("credentials_json") or (body or {}).get("credentials")
+    url = (body or {}).get("credentials_url")
+    creds = await _resolve(service, payload=payload, url=url)
+    try:
+        await service.delete_object_acl(creds, bucket, object, entity)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data={"deleted": entity})
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.get("/buckets/{bucket}/objects/{object:path}", response_model=GCSGenericResponse,
+            summary="Get object metadata")
+async def get_object(
+    bucket: str,
+    object: str,
+    generation: Optional[int] = Query(None, description="Specific object generation"),
+    service: GCSService = Depends(get_gcs_service),
+    x_credentials: Optional[str] = Header(None, alias="X-Goog-Credentials"),
+    x_credentials_url: Optional[str] = Header(None, alias="X-Goog-Credentials-Url"),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, payload=x_credentials, url=x_credentials_url)
+    try:
+        result = await service.get_object(creds, bucket, object, generation=generation)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.patch("/buckets/{bucket}/objects/{object:path}", response_model=GCSGenericResponse,
+              summary="Patch object metadata (partial update)")
+async def patch_object(
+    bucket: str,
+    object: str,
+    body: GCSObjectUpdateRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.patch_object(creds, bucket, object, body.object,
+                                            if_generation_match=body.if_generation_match)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.put("/buckets/{bucket}/objects/{object:path}", response_model=GCSGenericResponse,
+            summary="Replace object metadata (full update)")
+async def update_object(
+    bucket: str,
+    object: str,
+    body: GCSObjectUpdateRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.update_object(creds, bucket, object, body.object,
+                                             if_generation_match=body.if_generation_match)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.delete("/buckets/{bucket}/objects/{object:path}", response_model=GCSGenericResponse,
+               summary="Delete an object")
+async def delete_object(
+    bucket: str,
+    object: str,
+    body: Optional[Dict[str, Any]] = None,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    payload = (body or {}).get("credentials_json") or (body or {}).get("credentials")
+    url = (body or {}).get("credentials_url")
+    creds = await _resolve(service, payload=payload, url=url)
+    generation = (body or {}).get("generation")
+    if_generation_match = (body or {}).get("if_generation_match")
+    try:
+        await service.delete_object(creds, bucket, object,
+                                    generation=generation, if_generation_match=if_generation_match)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data={"deleted": object})
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+# ---------------------------------------------------------------------------
+# Objects — copy / move / rewrite / compose
+# ---------------------------------------------------------------------------
+
+@router.post("/buckets/{bucket}/objects/{object:path}/copy", response_model=GCSGenericResponse,
+             summary="Copy an object to a destination")
+async def copy_object(
+    bucket: str,
+    object: str,
+    body: GCSObjectTransferRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.copy_object(
+            creds, bucket, object, body.destination_bucket, body.destination_name,
+            source_generation=body.source_generation, body=body.object,
+        )
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/move", response_model=GCSGenericResponse,
+             summary="Move an object to a destination")
+async def move_object(
+    bucket: str,
+    object: str,
+    body: GCSObjectTransferRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.move_object(
+            creds, bucket, object, body.destination_bucket, body.destination_name,
+            source_generation=body.source_generation, if_generation_match=body.if_generation_match,
+            body=body.object,
+        )
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/rewrite", response_model=GCSGenericResponse,
+             summary="Rewrite (async copy) an object to a destination")
+async def rewrite_object(
+    bucket: str,
+    object: str,
+    body: GCSObjectTransferRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.rewrite_object(
+            creds, bucket, object, body.destination_bucket, body.destination_name,
+            rewrite_token=body.rewrite_token,
+            max_bytes_rewritten_per_call=body.max_bytes_rewritten_per_call,
+            body=body.object,
+        )
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/hold", response_model=GCSGenericResponse,
+             summary="Set an event-based hold on an object")
+async def set_object_hold(
+    bucket: str,
+    object: str,
+    body: GCSBaseCredentialsRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.set_object_hold(creds, bucket, object)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/releaseHold", response_model=GCSGenericResponse,
+             summary="Release an event-based hold on an object")
+async def release_object_hold(
+    bucket: str,
+    object: str,
+    body: GCSBaseCredentialsRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.release_object_hold(creds, bucket, object)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/lockRetentionPolicy", response_model=GCSGenericResponse,
+             summary="Lock a retention policy on an object (permanent)")
+async def lock_object_retention(
+    bucket: str,
+    object: str,
+    body: GCSBucketRestoreRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.lock_object_retention(
+            creds, bucket, object, if_metageneration_match=body.if_metageneration_match
+        )
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/compose", response_model=GCSGenericResponse,
+             summary="Compose multiple source objects into this destination object")
+async def compose_object(
+    bucket: str,
+    object: str,
+    body: GCSComposeRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        result = await service.compose_object(creds, bucket, object, body.source_objects, body=body.destination)
+        return GCSGenericResponse(success=True, time_ms=_elapsed_ms(start), data=result.get("data"))
+    except GCSError as exc:
+        return _error_response(GCSGenericResponse, start, exc)
+
+
+# ---------------------------------------------------------------------------
+# Public & signed URLs
+# ---------------------------------------------------------------------------
+
+@router.post("/buckets/{bucket}/objects/{object:path}/signed-url", response_model=GCSURLResponse,
+             summary="Generate a V4 signed download URL for an object")
+async def signed_download_url(
+    bucket: str,
+    object: str,
+    body: GCSURLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        url, expires = await service.download_signed_url(
+            creds, bucket, object,
+            expires_in_seconds=body.expires_in_seconds,
+            response_content_type=body.response_content_type,
+            response_disposition=body.response_disposition,
+        )
+        return GCSURLResponse(
+            success=True, time_ms=_elapsed_ms(start), url=url,
+            method="GET", expires_in_seconds=expires,
+        )
+    except GCSError as exc:
+        return _error_response(GCSURLResponse, start, exc)
+
+
+@router.post("/buckets/{bucket}/objects/{object:path}/signed-upload-url", response_model=GCSURLResponse,
+             summary="Generate a V4 signed upload URL for an object (PUT)")
+async def signed_upload_url(
+    bucket: str,
+    object: str,
+    body: GCSURLRequest,
+    service: GCSService = Depends(get_gcs_service),
+):
+    start = time.perf_counter()
+    creds = await _resolve(service, **_creds_fields(body))
+    try:
+        url, expires = await service.upload_signed_url(
+            creds, bucket, object,
+            expires_in_seconds=body.expires_in_seconds,
+            content_type=body.content_type,
+        )
+        return GCSURLResponse(
+            success=True, time_ms=_elapsed_ms(start), url=url,
+            method="PUT", expires_in_seconds=expires,
+        )
+    except GCSError as exc:
+        return _error_response(GCSURLResponse, start, exc)

app/api/v1/router.py

@@ -11,6 +11,7 @@ from app.api.v1 import (
     csv_analysis,
     database,
     embeddings,
+    gcs,
     google_maps,
     google_oauth,
     json_extract,
@@ -55,6 +56,7 @@ api_v1_router.include_router(webhook_socket.router, tags=["Webhook / Socket"])
 api_v1_router.include_router(csv_analysis.router, tags=["CSV Analysis"])
 api_v1_router.include_router(google_maps.router, tags=["Google Maps"])
 api_v1_router.include_router(google_oauth.router, tags=["Google OAuth"])
+api_v1_router.include_router(gcs.router, tags=["Google Cloud Storage"])
 api_v1_router.include_router(json_extract.router, tags=["JSON Extractor"])
 api_v1_router.include_router(keys_extract.router, prefix="/json", tags=["Keys Extractor"])
 api_v1_router.include_router(qr_decoder.router, tags=["QR Decoder"])

app/config.py

@@ -102,6 +102,17 @@ class Settings(BaseSettings):
     google_maps_timeout: int = 15
     google_maps_max_retries: int = 2
 
+    # Google Cloud Storage (GCS) settings
+    gcs_bucket_name: str = Field(default="", alias="GCS_BUCKET_NAME")
+    gcs_service_account_key_path: str = Field(default="", alias="GCS_SERVICE_ACCOUNT_KEY_PATH")
+    gcs_api_base_url: str = "https://storage.googleapis.com/storage/v1"
+    gcs_upload_base_url: str = "https://storage.googleapis.com/upload/storage/v1"
+    gcs_token_uri: str = "https://oauth2.googleapis.com/token"
+    gcs_timeout: float = 30.0
+    gcs_max_retries: int = 2
+    gcs_default_expires_seconds: int = 3600
+    gcs_max_expires_seconds: int = 604800
+
     # Google OAuth 2.0 (OpenID Connect) Sign-In settings
     google_oauth_auth_url: str = "https://accounts.google.com/o/oauth2/v2/auth"
     google_oauth_token_url: str = "https://oauth2.googleapis.com/token"

app/models/schemas.py

@@ -1187,3 +1187,265 @@ class GoogleStaticMapResponse(BaseModel):
     format: str = Field("png", description="Image format")
     warning: Optional[str] = Field(None, description="Warning returned by the Maps Static API, if any")
     error: Optional[str] = None
+
+
+# ---------------------------------------------------------------------------
+# Google Cloud Storage (GCS) — Service Account JSON API
+# ---------------------------------------------------------------------------
+
+GCS_BUCKET_NAME_REGEX = r"^[a-z0-9][a-z0-9._-]{1,61}[a-z0-9]$"
+
+
+class GCSBaseCredentialsRequest(BaseModel):
+    """Reusable service-account credential fields accepted by every GCS endpoint.
+
+    Credentials may be supplied inline as a JSON object (``credentials``), as a
+    raw JSON string (``credentials_json``), or via a URL (``credentials_url``).
+    When none are supplied, the server falls back to the environment-configured
+    service account (``GCS_SERVICE_ACCOUNT_KEY_PATH``).
+    """
+
+    credentials: Optional[Dict[str, Any]] = Field(
+        None, description="Service account key as an inline JSON object"
+    )
+    credentials_json: Optional[str] = Field(
+        None, description="Service account key as a raw JSON string"
+    )
+    credentials_url: Optional[str] = Field(
+        None, description="Publicly reachable URL that returns the service account JSON"
+    )
+
+
+class GCSBucketRequest(GCSBaseCredentialsRequest):
+    """Create or fully replace a bucket resource."""
+
+    name: str = Field(..., pattern=GCS_BUCKET_NAME_REGEX, description="Globally unique bucket name")
+    project_id: Optional[str] = Field(
+        None, description="GCP project id used for the operation (defaults to the service account project)"
+    )
+    location: Optional[str] = Field(
+        None, pattern=r"^[A-Za-z0-9._-]+$", description="Bucket location (e.g. 'US-EAST1')"
+    )
+    storage_class: Optional[Literal["STANDARD", "NEARLINE", "COLDLINE", "ARCHIVE", "STANDARD_IA", "MULTI_REGIONAL", "REGIONAL", "DURABLE_REDUCED_AVAILABILITY"]] = Field(
+        None, description="Storage class for the bucket"
+    )
+    bucket: Optional[Dict[str, Any]] = Field(
+        None, description="Full GCS Bucket resource; merged over the convenience fields above"
+    )
+
+
+class GCSBucketUpdateRequest(GCSBaseCredentialsRequest):
+    """Patch (partial) or update (full) a bucket's metadata."""
+
+    bucket: Dict[str, Any] = Field(..., description="GCS Bucket resource fields to apply")
+
+
+class GCSBucketDeleteRequest(GCSBaseCredentialsRequest):
+    """Delete a bucket, optionally guarded by a metageneration match."""
+
+    if_metageneration_match: Optional[int] = Field(
+        None, description="Only delete if the bucket metageneration matches"
+    )
+
+
+class GCSBucketIAMRequest(GCSBaseCredentialsRequest):
+    policy: Dict[str, Any] = Field(..., description="GCS IAM Policy resource")
+
+
+class GCSBucketPermissionsRequest(GCSBaseCredentialsRequest):
+    permissions: List[str] = Field(
+        ..., min_length=1, description="Permissions to test (e.g. 'storage.buckets.get', 'storage.objects.list')"
+    )
+
+
+class GCSACLRequest(GCSBaseCredentialsRequest):
+    """Insert or update an ACL entry."""
+
+    entity: str = Field(
+        ..., min_length=1, max_length=1024,
+        description="ACL entity (e.g. 'allUsers', 'user-email@example.com', 'project-owners-1234')",
+    )
+    role: Literal["OWNER", "READER", "WRITER"] = Field(..., description="ACL role")
+
+    @field_validator("entity")
+    @classmethod
+    def _clean_entity(cls, v: str) -> str:
+        cleaned = v.strip()
+        if not cleaned:
+            raise ValueError("entity must not be empty or whitespace")
+        return cleaned
+
+
+class GCSObjectUpdateRequest(GCSBaseCredentialsRequest):
+    """Patch or update object metadata."""
+
+    object: Dict[str, Any] = Field(..., description="GCS Object resource fields to apply")
+    if_generation_match: Optional[int] = Field(
+        None, description="Only apply if the object generation matches"
+    )
+
+
+class GCSObjectTransferRequest(GCSBaseCredentialsRequest):
+    """Copy, move, or rewrite an object to a destination."""
+
+    destination_bucket: str = Field(..., pattern=GCS_BUCKET_NAME_REGEX, description="Destination bucket")
+    destination_name: str = Field(..., min_length=1, description="Destination object name")
+    source_generation: Optional[int] = Field(
+        None, description="Generation of the source object to operate on"
+    )
+    if_generation_match: Optional[int] = Field(
+        None, description="Generation condition for the destination object"
+    )
+    rewrite_token: Optional[str] = Field(
+        None, description="Rewrite token from a previous (paged) rewrite response"
+    )
+    max_bytes_rewritten_per_call: Optional[int] = Field(
+        None, ge=1, description="Maximum bytes to rewrite per call for large object rewrites"
+    )
+    object: Optional[Dict[str, Any]] = Field(
+        None, description="Destination object metadata to apply during the transfer"
+    )
+
+
+class GCSComposeRequest(GCSBaseCredentialsRequest):
+    """Compose multiple source objects into a single destination object."""
+
+    source_objects: List[str] = Field(..., min_length=1, description="Names of source objects to compose")
+    destination: Optional[Dict[str, Any]] = Field(
+        None, description="Destination object metadata"
+    )
+
+    @field_validator("source_objects")
+    @classmethod
+    def _clean_sources(cls, v: List[str]) -> List[str]:
+        cleaned = [name.strip() for name in v if name and name.strip()]
+        if not cleaned:
+            raise ValueError("source_objects must contain at least one non-empty object name")
+        if any(name.startswith("/") or len(name) > 1024 for name in cleaned):
+            raise ValueError("source object names must be 1-1024 characters and not start with '/'")
+        return cleaned
+
+
+class GCSObjectDownloadRequest(GCSBaseCredentialsRequest):
+    """Request an object's bytes as base64 (JSON-friendly alternative to the raw GET)."""
+
+    generation: Optional[int] = Field(None, description="Specific object generation to download")
+
+
+class GCSObjectDeleteRequest(GCSBaseCredentialsRequest):
+    """Delete an object, optionally pinned to a generation or guarded by a match."""
+
+    generation: Optional[int] = Field(None, description="Specific object generation to delete")
+    if_generation_match: Optional[int] = Field(
+        None, description="Only delete if the object generation matches"
+    )
+
+
+class GCSResumableUploadRequest(GCSBaseCredentialsRequest):
+    """Start a resumable upload session for an object."""
+
+    name: str = Field(
+        ..., min_length=1, max_length=1024, description="Destination object name"
+    )
+    content_type: Optional[str] = Field(
+        None, description="Content-Type of the object (defaults to application/octet-stream)"
+    )
+    object_metadata: Optional[Dict[str, Any]] = Field(
+        None, description="Object metadata to apply to the created object"
+    )
+
+    @field_validator("name")
+    @classmethod
+    def _validate_object_name(cls, v: str) -> str:
+        cleaned = v.strip()
+        if cleaned.startswith("/"):
+            raise ValueError("object name must not start with '/'")
+        return cleaned
+
+
+class GCSBucketRestoreRequest(GCSBaseCredentialsRequest):
+    """Restore a soft-deleted bucket."""
+
+    if_metageneration_match: Optional[int] = Field(
+        None, description="Only restore if the bucket metageneration matches"
+    )
+
+
+class GCSWatchRequest(GCSBaseCredentialsRequest):
+    """Register an object-change notification channel for a bucket."""
+
+    channel: Dict[str, Any] = Field(..., description="GCS Channel resource for object change notifications")
+
+
+class GCSURLRequest(GCSBaseCredentialsRequest):
+    """Generate a public or signed URL for an object."""
+
+    expires_in_seconds: Optional[int] = Field(
+        None, ge=1, le=604800, description="Signed URL lifetime in seconds (max 604800)"
+    )
+    content_type: Optional[str] = Field(
+        None, description="Content-Type to sign for PUT (upload) signed URLs"
+    )
+    response_content_type: Optional[str] = Field(
+        None, description="For downloads: override the served Content-Type"
+    )
+    response_disposition: Optional[str] = Field(
+        None, description="For downloads: set the Content-Disposition (e.g. 'attachment; filename=...')"
+    )
+
+    @field_validator("content_type", "response_content_type")
+    @classmethod
+    def _validate_media_type(cls, v: Optional[str]) -> Optional[str]:
+        if v is not None and not re.fullmatch(r"[\w./+-]{1,255}", v):
+            raise ValueError("content_type must be a valid media type (e.g. 'application/pdf')")
+        return v
+
+    @field_validator("response_disposition")
+    @classmethod
+    def _validate_disposition(cls, v: Optional[str]) -> Optional[str]:
+        if v is not None and len(v) > 500:
+            raise ValueError("response_disposition must be at most 500 characters")
+        return v
+
+
+class GCSGenericResponse(BaseModel):
+    """Envelope for most GCS operations."""
+
+    success: bool
+    time_ms: float = Field(0.0, description="Server-side processing time in milliseconds")
+    data: Optional[Dict[str, Any]] = Field(None, description="GCS API response payload")
+    error: Optional[str] = None
+
+
+class GCSListResponse(BaseModel):
+    """Envelope for paginated list operations."""
+
+    success: bool
+    time_ms: float = 0.0
+    count: int = 0
+    next_page_token: Optional[str] = None
+    items: List[Dict[str, Any]] = Field(default_factory=list, description="Listed items")
+    prefixes: List[str] = Field(default_factory=list, description="Common prefixes when a delimiter was used")
+    error: Optional[str] = None
+
+
+class GCSDownloadResponse(BaseModel):
+    """Envelope for object downloads returned as base64."""
+
+    success: bool
+    time_ms: float = 0.0
+    content_base64: Optional[str] = Field(None, description="Base64-encoded object bytes")
+    content_type: Optional[str] = Field(None, description="Content-Type of the object")
+    size_bytes: int = 0
+    error: Optional[str] = None
+
+
+class GCSURLResponse(BaseModel):
+    """Envelope for public / signed URL generation."""
+
+    success: bool
+    time_ms: float = 0.0
+    url: Optional[str] = Field(None, description="Generated URL")
+    method: Optional[str] = Field(None, description="HTTP method the signed URL authorizes")
+    expires_in_seconds: Optional[int] = Field(None, description="Signed URL lifetime in seconds")
+    error: Optional[str] = None

app/services/gcs_service.py

@@ -0,0 +1,1077 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import json
+import re
+import time
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import quote, urlencode
+
+import httpx
+from pydantic import BaseModel, ValidationError
+
+from app.config import get_settings
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+# Scopes needed to perform full object/bucket management via the JSON API.
+_GCS_SCOPES = (
+    "https://www.googleapis.com/auth/devstorage.full_control "
+    "https://www.googleapis.com/auth/cloud-platform"
+)
+
+# Transient HTTP status codes that are safe to retry against Google.
+_RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504})
+
+# Cloud Storage bucket naming rules (3-63 chars, lowercase letters/digits/_/.-).
+BUCKET_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{1,61}[a-z0-9]$")
+
+_SIGNING_ALGORITHM = "GOOG4-RSA-SHA256"
+_SIGNING_REGION = "auto"
+_SIGNING_SERVICE = "storage"
+
+
+class GCSCredentialsError(Exception):
+    """Raised when service-account credentials are missing or malformed."""
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.message = message
+
+
+class GCSError(Exception):
+    """Raised for upstream Cloud Storage failures mapped to client-facing errors."""
+
+    def __init__(self, message: str, status_code: int = 400) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+class GCSCredentials(BaseModel):
+    """Validated Google service-account JSON key material."""
+
+    type: str = "service_account"
+    project_id: str = ""
+    private_key_id: str = ""
+    private_key: str
+    client_email: str
+    client_id: str = ""
+    auth_uri: str = "https://accounts.google.com/o/oauth2/auth"
+    token_uri: str = "https://oauth2.googleapis.com/token"
+    auth_provider_x509_cert_url: str = "https://www.googleapis.com/oauth2/v1/certs"
+    client_x509_cert_url: str = ""
+    universe_domain: str = "googleapis.com"
+
+    @property
+    def scope_key(self) -> str:
+        """Unique key used to share cached access tokens across requests."""
+        return f"{self.client_email}|{self.private_key_id}"
+
+
+class _TokenCache:
+    """In-memory cache of short-lived Google access tokens, keyed per service account."""
+
+    def __init__(self) -> None:
+        self._entries: Dict[str, Dict[str, Any]] = {}
+        self._lock = asyncio.Lock()
+
+    def _key(self, creds: GCSCredentials) -> str:
+        return f"{creds.client_email}|{creds.private_key_id}"
+
+    def get(self, creds: GCSCredentials) -> Optional[str]:
+        entry = self._entries.get(self._key(creds))
+        if entry and entry["expires_at"] > time.time():
+            return entry["token"]
+        return None
+
+    async def set(self, creds: GCSCredentials, token: str, expires_in: int) -> None:
+        self._entries[self._key(creds)] = {
+            "token": token,
+            "expires_at": time.time() + max(expires_in - 60, 60),
+        }
+
+
+class GCSService:
+    """Asynchronous, connection-pooled client for the Google Cloud Storage JSON API."""
+
+    def __init__(self) -> None:
+        self._client: Optional[httpx.AsyncClient] = None
+        self._client_lock = asyncio.Lock()
+        self._token_cache = _TokenCache()
+
+    # ------------------------------------------------------------------
+    # HTTP client management
+    # ------------------------------------------------------------------
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            async with self._client_lock:
+                if self._client is None or self._client.is_closed:
+                    self._client = httpx.AsyncClient(
+                        timeout=_settings.gcs_timeout,
+                        follow_redirects=True,
+                        limits=httpx.Limits(
+                            max_connections=100,
+                            max_keepalive_connections=20,
+                            keepalive_expiry=30,
+                        ),
+                    )
+        return self._client
+
+    async def close(self) -> None:
+        async with self._client_lock:
+            if self._client is not None and not self._client.is_closed:
+                await self._client.aclose()
+            self._client = None
+
+    # ------------------------------------------------------------------
+    # Credential resolution (JSON body / JSON string / URL / file / env)
+    # ------------------------------------------------------------------
+
+    async def resolve_credentials(
+        self,
+        *,
+        payload: Optional[Any] = None,
+        url: Optional[str] = None,
+        file_bytes: Optional[bytes] = None,
+        file_name: Optional[str] = None,
+    ) -> GCSCredentials:
+        """Resolve a validated service-account credential from any supported source.
+
+        Resolution precedence: inline ``payload`` (dict or JSON string), then a
+        URL to fetch, then uploaded ``file_bytes``, then the env-configured path.
+        """
+        raw: Optional[str] = None
+        source: str = ""
+
+        if file_bytes is not None:
+            source = f"uploaded file '{file_name or 'credentials'}'"
+            raw = file_bytes.decode("utf-8")
+        elif payload is not None:
+            if isinstance(payload, dict):
+                return self._validate_credentials(payload, "request body")
+            if isinstance(payload, str) and payload.strip():
+                source = "request body"
+                raw = payload
+            else:
+                raise GCSCredentialsError(
+                    "credentials must be a JSON object or a JSON string."
+                )
+        elif url:
+            source = f"url '{url}'"
+            raw = await self._fetch_credentials_url(url)
+        elif _settings.gcs_service_account_key_path:
+            source = f"file '{_settings.gcs_service_account_key_path}'"
+            try:
+                with open(_settings.gcs_service_account_key_path, "r", encoding="utf-8") as f:
+                    raw = f.read()
+            except OSError as exc:
+                raise GCSCredentialsError(
+                    f"Unable to read service account file at "
+                    f"'{_settings.gcs_service_account_key_path}': {exc}"
+                )
+        else:
+            raise GCSCredentialsError(
+                "No GCS service account credentials provided. Pass 'credentials' "
+                "(JSON object) or 'credentials_json' in the body, 'credentials_url', "
+                "upload a credentials file, or set GCS_SERVICE_ACCOUNT_KEY_PATH."
+            )
+
+        try:
+            data = json.loads(raw)
+        except (json.JSONDecodeError, TypeError) as exc:
+            raise GCSCredentialsError(
+                f"Service account data from {source} is not valid JSON: {exc}"
+            )
+        return self._validate_credentials(data, source)
+
+    @staticmethod
+    def _validate_credentials(data: Any, source: str) -> GCSCredentials:
+        if not isinstance(data, dict):
+            raise GCSCredentialsError(f"Service account data from {source} must be a JSON object.")
+        try:
+            return GCSCredentials(**data)
+        except ValidationError as exc:
+            raise GCSCredentialsError(
+                f"Service account data from {source} is missing required fields: "
+                f"{', '.join(e['loc'][0] for e in exc.errors())}"
+            )
+
+    @staticmethod
+    def _validate_http_url(url: str) -> None:
+        if not isinstance(url, str) or not url.strip():
+            raise GCSCredentialsError("A non-empty URL string is required.")
+        if len(url) > 2048:
+            raise GCSCredentialsError("URL must be at most 2048 characters.")
+        parts = url.split("://", 1)
+        if len(parts) != 2 or parts[0].lower() not in ("http", "https"):
+            raise GCSCredentialsError("URL must use the http or https scheme.")
+
+    async def _fetch_credentials_url(self, url: str) -> str:
+        self._validate_http_url(url)
+        client = await self._get_client()
+        try:
+            response = await client.get(url)
+            response.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise GCSCredentialsError(f"Failed to fetch service account JSON from '{url}': {exc}")
+        return response.text
+
+    async def fetch_url_content(self, url: str) -> Tuple[bytes, str]:
+        """Fetch arbitrary content from a URL for use as an object body."""
+        self._validate_http_url(url)
+        client = await self._get_client()
+        try:
+            response = await client.get(url)
+            response.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise GCSError(f"Failed to fetch content from '{url}': {exc}", status_code=400)
+        return response.content, response.headers.get("content-type", "application/octet-stream")
+
+    # ------------------------------------------------------------------
+    # Access token acquisition (service account JWT -> bearer token)
+    # ------------------------------------------------------------------
+
+    async def get_access_token(self, creds: GCSCredentials) -> str:
+        cached = self._token_cache.get(creds)
+        if cached:
+            return cached
+        async with self._token_cache._lock:
+            cached = self._token_cache.get(creds)
+            if cached:
+                return cached
+            token, expires_in = await self._fetch_access_token(creds)
+            await self._token_cache.set(creds, token, expires_in)
+            return token
+
+    async def _fetch_access_token(self, creds: GCSCredentials) -> Tuple[str, int]:
+        assertion = self._build_signed_jwt(creds)
+        body = urlencode({
+            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+            "assertion": assertion,
+        })
+        client = await self._get_client()
+        last_error: Optional[str] = None
+        for attempt in range(1 + _settings.gcs_max_retries):
+            try:
+                response = await client.request(
+                    "POST",
+                    creds.token_uri or _settings.gcs_token_uri,
+                    content=body,
+                    headers={"Content-Type": "application/x-www-form-urlencoded"},
+                )
+                if response.status_code == 200:
+                    data = response.json()
+                    token: str = data.get("access_token", "")
+                    if not token:
+                        raise GCSCredentialsError(
+                            "Google did not return an access token for the service account."
+                        )
+                    return token, int(data.get("expires_in", 3600))
+
+                last_error = self._format_token_error(response)
+                _logger.warning(
+                    "GCS token exchange failed (attempt %d/%d): %s",
+                    attempt + 1, 1 + _settings.gcs_max_retries, last_error,
+                )
+                if response.status_code in _RETRYABLE_STATUS and attempt < _settings.gcs_max_retries:
+                    await asyncio.sleep(2.0 ** attempt)
+                    continue
+                raise GCSCredentialsError(last_error)
+
+            except httpx.TimeoutException:
+                last_error = "Google token endpoint timed out."
+            except httpx.RequestError as exc:
+                last_error = f"Google token endpoint request failed: {exc}"
+            except GCSCredentialsError:
+                raise
+            except Exception as exc:
+                last_error = f"Unexpected error during token exchange: {exc}"
+
+            if attempt < _settings.gcs_max_retries:
+                await asyncio.sleep(2.0 ** attempt)
+
+        raise GCSCredentialsError(last_error or "Unknown token exchange failure.")
+
+    @staticmethod
+    def _format_token_error(response: httpx.Response) -> str:
+        try:
+            body = response.json()
+            error_desc = body.get("error_description", "") or body.get("error", "")
+            if error_desc:
+                return error_desc if isinstance(error_desc, str) else str(error_desc)
+        except Exception:
+            pass
+        return f"Google OAuth token exchange error (HTTP {response.status_code})."
+
+    def _build_signed_jwt(self, creds: GCSCredentials) -> str:
+        now = int(time.time())
+        header = {"alg": "RS256", "typ": "JWT"}
+        claims = {
+            "iss": creds.client_email,
+            "scope": _GCS_SCOPES,
+            "aud": creds.token_uri or _settings.gcs_token_uri,
+            "iat": now,
+            "exp": now + 3600,
+        }
+        from cryptography.hazmat.primitives import hashes
+        from cryptography.hazmat.primitives.asymmetric import padding
+        from cryptography.hazmat.primitives.serialization import load_pem_private_key
+
+        signing_input = (
+            base64.urlsafe_b64encode(json.dumps(header, separators=(",", ":")).encode()).rstrip(b"=")
+            + b"."
+            + base64.urlsafe_b64encode(json.dumps(claims, separators=(",", ":")).encode()).rstrip(b"=")
+        )
+        private_key = load_pem_private_key(creds.private_key.encode("utf-8"), password=None)
+        signature = private_key.sign(signing_input, padding.PKCS1v15(), hashes.SHA256())
+        return (
+            signing_input
+            + b"."
+            + base64.urlsafe_b64encode(signature).rstrip(b"=")
+        ).decode("ascii")
+
+    # ------------------------------------------------------------------
+    # Generic authenticated JSON API request
+    # ------------------------------------------------------------------
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        creds: GCSCredentials,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        json_body: Optional[Any] = None,
+        extra_headers: Optional[Dict[str, str]] = None,
+        content: Optional[bytes] = None,
+        upload: bool = False,
+    ) -> Dict[str, Any]:
+        token = await self.get_access_token(creds)
+        base_url = _settings.gcs_upload_base_url if upload else _settings.gcs_api_base_url
+        url = f"{base_url}{path}"
+        headers = {"Authorization": f"Bearer {token}"}
+        if extra_headers:
+            headers.update(extra_headers)
+
+        client = await self._get_client()
+        last_error: Optional[str] = None
+        last_status: int = 0
+        for attempt in range(1 + _settings.gcs_max_retries):
+            try:
+                response = await client.request(
+                    method,
+                    url,
+                    params=params,
+                    json=json_body,
+                    headers=headers,
+                    content=content,
+                )
+                if response.status_code < 400:
+                    return self._success_result(response)
+
+                last_status = response.status_code
+                last_error = self._format_gcs_error(response)
+                _logger.warning(
+                    "GCS API error on %s %s: HTTP %s -> %s",
+                    method, path, response.status_code, last_error,
+                )
+
+                if response.status_code in _RETRYABLE_STATUS:
+                    if attempt < _settings.gcs_max_retries:
+                        await asyncio.sleep(2.0 ** attempt)
+                    continue
+                if 400 <= response.status_code < 500:
+                    raise GCSError(last_error, status_code=response.status_code)
+
+            except GCSError:
+                raise
+            except httpx.TimeoutException:
+                last_error = "Request timed out"
+                _logger.warning("GCS API timeout on %s %s (attempt %d/%d)", method, path, attempt + 1, 1 + _settings.gcs_max_retries)
+            except httpx.RequestError as exc:
+                last_error = f"Request failed: {exc}"
+                _logger.warning("GCS API request error on %s %s: %s (attempt %d/%d)", method, path, last_error, attempt + 1, 1 + _settings.gcs_max_retries)
+            except Exception as exc:
+                last_error = f"Unexpected error: {exc}"
+                _logger.error("GCS API unexpected error on %s %s: %s", method, path, last_error)
+                break
+
+            if attempt < _settings.gcs_max_retries:
+                await asyncio.sleep(1.0 * (attempt + 1))
+
+        raise GCSError(last_error or f"GCS API error (HTTP {last_status}).", status_code=502)
+
+    @staticmethod
+    def _success_result(response: httpx.Response) -> Dict[str, Any]:
+        content_type = response.headers.get("content-type", "")
+        result: Dict[str, Any] = {
+            "success": True,
+            "status_code": response.status_code,
+            "content_type": content_type,
+            "headers": dict(response.headers),
+            "error": None,
+        }
+        if response.content and content_type.startswith("application/json"):
+            result["data"] = response.json()
+        else:
+            result["data"] = response.content
+        return result
+
+    @staticmethod
+    def _format_gcs_error(response: httpx.Response) -> str:
+        status = response.status_code
+        message = ""
+        reason = ""
+        try:
+            body = response.json()
+            error = body.get("error", {}) if isinstance(body, dict) else {}
+            message = error.get("message", "") or ""
+            errors = error.get("errors", [])
+            if errors and isinstance(errors[0], dict):
+                reason = errors[0].get("reason", "") or ""
+        except Exception:
+            pass
+
+        if status == 401:
+            return "Invalid or expired Google service account credentials."
+        if status == 403:
+            if "permission" in message.lower() or reason in ("forbidden", "required", "storagePermissionDenied"):
+                return "Permission denied. The service account lacks the required IAM role for this operation."
+            return f"Access forbidden: {message}".rstrip(".")
+        if status == 404:
+            return "The requested bucket or object was not found."
+        if status == 409:
+            return "Conflict. A bucket or object with the same name already exists, or a generation condition failed."
+        if status == 429:
+            return "API rate limit exceeded. Please wait and retry."
+        if message:
+            return message.rstrip(".") + "."
+        return f"Google Cloud Storage error (HTTP {status})."
+
+    # ------------------------------------------------------------------
+    # Buckets
+    # ------------------------------------------------------------------
+
+    async def create_bucket(
+        self,
+        creds: GCSCredentials,
+        name: str,
+        *,
+        project: Optional[str] = None,
+        bucket_body: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        project_id = project or creds.project_id
+        body: Dict[str, Any] = dict(bucket_body or {})
+        body["name"] = name
+        params = {"project": project_id}
+        return await self._request("POST", "/b", creds, params=params, json_body=body)
+
+    async def get_bucket(self, creds: GCSCredentials, bucket: str) -> Dict[str, Any]:
+        return await self._request("GET", f"/b/{quote(bucket, safe='')}", creds)
+
+    async def list_buckets(
+        self,
+        creds: GCSCredentials,
+        *,
+        project: Optional[str] = None,
+        prefix: Optional[str] = None,
+        max_results: Optional[int] = None,
+        page_token: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"project": project or creds.project_id}
+        if prefix:
+            params["prefix"] = prefix
+        if max_results is not None:
+            params["maxResults"] = max_results
+        if page_token:
+            params["pageToken"] = page_token
+        return await self._request("GET", "/b", creds, params=params)
+
+    async def patch_bucket(
+        self, creds: GCSCredentials, bucket: str, body: Dict[str, Any],
+        *, if_metageneration_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
+        return await self._request("PATCH", f"/b/{quote(bucket, safe='')}", creds, params=params, json_body=body)
+
+    async def update_bucket(
+        self, creds: GCSCredentials, bucket: str, body: Dict[str, Any],
+        *, if_metageneration_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
+        return await self._request("PUT", f"/b/{quote(bucket, safe='')}", creds, params=params, json_body=body)
+
+    async def delete_bucket(
+        self, creds: GCSCredentials, bucket: str,
+        *, if_metageneration_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
+        return await self._request("DELETE", f"/b/{quote(bucket, safe='')}", creds, params=params)
+
+    # ------------------------------------------------------------------
+    # Bucket IAM & permissions
+    # ------------------------------------------------------------------
+
+    async def get_bucket_iam(self, creds: GCSCredentials, bucket: str) -> Dict[str, Any]:
+        return await self._request("GET", f"/b/{quote(bucket, safe='')}/iam", creds)
+
+    async def set_bucket_iam(
+        self, creds: GCSCredentials, bucket: str, policy: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request("PUT", f"/b/{quote(bucket, safe='')}/iam", creds, json_body=policy)
+
+    async def test_bucket_permissions(
+        self, creds: GCSCredentials, bucket: str, permissions: List[str],
+    ) -> Dict[str, Any]:
+        params = {"permissions": ",".join(permissions)}
+        return await self._request("GET", f"/b/{quote(bucket, safe='')}/iam/testPermissions", creds, params=params)
+
+    # ------------------------------------------------------------------
+    # Default object ACLs
+    # ------------------------------------------------------------------
+
+    async def list_default_object_acl(self, creds: GCSCredentials, bucket: str) -> Dict[str, Any]:
+        return await self._request("GET", f"/b/{quote(bucket, safe='')}/defaultObjectAcl", creds)
+
+    async def insert_default_object_acl(
+        self, creds: GCSCredentials, bucket: str, body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request("POST", f"/b/{quote(bucket, safe='')}/defaultObjectAcl", creds, json_body=body)
+
+    async def get_default_object_acl(
+        self, creds: GCSCredentials, bucket: str, entity: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "GET", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds
+        )
+
+    async def patch_default_object_acl(
+        self, creds: GCSCredentials, bucket: str, entity: str, body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "PATCH", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds, json_body=body
+        )
+
+    async def update_default_object_acl(
+        self, creds: GCSCredentials, bucket: str, entity: str, body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "PUT", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds, json_body=body
+        )
+
+    async def delete_default_object_acl(
+        self, creds: GCSCredentials, bucket: str, entity: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "DELETE", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds
+        )
+
+    # ------------------------------------------------------------------
+    # Objects
+    # ------------------------------------------------------------------
+
+    async def list_objects(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        *,
+        prefix: Optional[str] = None,
+        delimiter: Optional[str] = None,
+        max_results: Optional[int] = None,
+        page_token: Optional[str] = None,
+        versions: Optional[bool] = None,
+        match_glob: Optional[str] = None,
+        start_offset: Optional[str] = None,
+        end_offset: Optional[str] = None,
+        include_trailing_delimiter: Optional[bool] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {}
+        if prefix is not None:
+            params["prefix"] = prefix
+        if delimiter is not None:
+            params["delimiter"] = delimiter
+        if max_results is not None:
+            params["maxResults"] = max_results
+        if page_token:
+            params["pageToken"] = page_token
+        if versions is not None:
+            params["versions"] = versions
+        if match_glob:
+            params["matchGlob"] = match_glob
+        if start_offset is not None:
+            params["startOffset"] = start_offset
+        if end_offset is not None:
+            params["endOffset"] = end_offset
+        if include_trailing_delimiter is not None:
+            params["includeTrailingDelimiter"] = include_trailing_delimiter
+        return await self._request("GET", f"/b/{quote(bucket, safe='')}/o", creds, params=params)
+
+    async def upload_object(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        content: bytes,
+        *,
+        content_type: str = "application/octet-stream",
+        metadata: Optional[Dict[str, Any]] = None,
+        if_generation_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"uploadType": "media", "name": name}
+        if if_generation_match is not None:
+            params["ifGenerationMatch"] = if_generation_match
+        if metadata:
+            for key, value in metadata.items():
+                params[key] = value
+        headers = {"Content-Type": content_type}
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o", creds,
+            params=params, content=content, extra_headers=headers, upload=True,
+        )
+
+    async def download_object(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        generation: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"alt": "media"}
+        if generation is not None:
+            params["generation"] = generation
+        return await self._request(
+            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params
+        )
+
+    async def get_object(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        generation: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"generation": generation} if generation is not None else None
+        return await self._request("GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params)
+
+    async def patch_object(
+        self, creds: GCSCredentials, bucket: str, name: str, body: Dict[str, Any],
+        *, if_generation_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifGenerationMatch": if_generation_match} if if_generation_match is not None else None
+        return await self._request(
+            "PATCH", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params, json_body=body
+        )
+
+    async def update_object(
+        self, creds: GCSCredentials, bucket: str, name: str, body: Dict[str, Any],
+        *, if_generation_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifGenerationMatch": if_generation_match} if if_generation_match is not None else None
+        return await self._request(
+            "PUT", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params, json_body=body
+        )
+
+    async def delete_object(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        generation: Optional[int] = None,
+        if_generation_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {}
+        if generation is not None:
+            params["generation"] = generation
+        if if_generation_match is not None:
+            params["ifGenerationMatch"] = if_generation_match
+        params = params or None
+        return await self._request("DELETE", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params)
+
+    async def copy_object(
+        self,
+        creds: GCSCredentials,
+        source_bucket: str,
+        source_name: str,
+        destination_bucket: str,
+        destination_name: str,
+        *,
+        source_generation: Optional[int] = None,
+        body: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        path = (
+            f"/b/{quote(source_bucket, safe='')}/o/{quote(source_name, safe='')}"
+            f"/copyTo/b/{quote(destination_bucket, safe='')}/o/{quote(destination_name, safe='')}"
+        )
+        params = {"sourceGeneration": source_generation} if source_generation is not None else None
+        return await self._request("POST", path, creds, params=params, json_body=body)
+
+    async def move_object(
+        self,
+        creds: GCSCredentials,
+        source_bucket: str,
+        source_name: str,
+        destination_bucket: str,
+        destination_name: str,
+        *,
+        source_generation: Optional[int] = None,
+        if_generation_match: Optional[int] = None,
+        body: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        path = (
+            f"/b/{quote(source_bucket, safe='')}/o/{quote(source_name, safe='')}"
+            f"/moveTo/b/{quote(destination_bucket, safe='')}/o/{quote(destination_name, safe='')}"
+        )
+        params: Dict[str, Any] = {}
+        if source_generation is not None:
+            params["sourceGeneration"] = source_generation
+        if if_generation_match is not None:
+            params["ifGenerationMatch"] = if_generation_match
+        return await self._request("POST", path, creds, params=params or None, json_body=body)
+
+    async def rewrite_object(
+        self,
+        creds: GCSCredentials,
+        source_bucket: str,
+        source_name: str,
+        destination_bucket: str,
+        destination_name: str,
+        *,
+        rewrite_token: Optional[str] = None,
+        max_bytes_rewritten_per_call: Optional[int] = None,
+        body: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        path = (
+            f"/b/{quote(source_bucket, safe='')}/o/{quote(source_name, safe='')}"
+            f"/rewriteTo/b/{quote(destination_bucket, safe='')}/o/{quote(destination_name, safe='')}"
+        )
+        params: Dict[str, Any] = {}
+        if rewrite_token:
+            params["rewriteToken"] = rewrite_token
+        if max_bytes_rewritten_per_call is not None:
+            params["maxBytesRewrittenPerCall"] = max_bytes_rewritten_per_call
+        return await self._request("POST", path, creds, params=params or None, json_body=body)
+
+    async def compose_object(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        destination_name: str,
+        source_objects: List[str],
+        *,
+        body: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        compose_body: Dict[str, Any] = {
+            "sourceObjects": [{"name": name} for name in source_objects]
+        }
+        if body:
+            compose_body["destination"] = body
+        path = f"/b/{quote(bucket, safe='')}/o/{quote(destination_name, safe='')}/compose"
+        return await self._request("POST", path, creds, json_body=compose_body)
+
+    async def initiate_resumable_upload(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        content_type: str = "application/octet-stream",
+        object_metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Create a resumable upload session and return its URI.
+
+        The client can then ``PUT`` the object bytes to ``session_uri`` (optionally
+        in chunks with ``Content-Range``) to complete the upload.
+        """
+        params: Dict[str, Any] = {"uploadType": "resumable", "name": name}
+        body: Dict[str, Any] = dict(object_metadata or {})
+        body.setdefault("name", name)
+        body.setdefault("contentType", content_type)
+        result = await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o", creds,
+            params=params, json_body=body,
+            extra_headers={"Content-Type": "application/json; charset=UTF-8"},
+            upload=True,
+        )
+        session_uri = (result.get("headers") or {}).get("location")
+        return {
+            "success": True,
+            "session_uri": session_uri,
+            "data": result.get("data"),
+            "error": None,
+        }
+
+    # ------------------------------------------------------------------
+    # Object holds & retention policy
+    # ------------------------------------------------------------------
+
+    async def set_object_hold(
+        self, creds: GCSCredentials, bucket: str, name: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/hold", creds
+        )
+
+    async def release_object_hold(
+        self, creds: GCSCredentials, bucket: str, name: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/releaseHold", creds
+        )
+
+    async def lock_object_retention(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        if_metageneration_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/lockRetentionPolicy",
+            creds, params=params,
+        )
+
+    # ------------------------------------------------------------------
+    # Notifications (object change watch)
+    # ------------------------------------------------------------------
+
+    async def watch_all_objects(
+        self, creds: GCSCredentials, bucket: str, channel_body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o/watch", creds, json_body=channel_body
+        )
+
+    # ------------------------------------------------------------------
+    # Bucket restore (soft-deleted buckets)
+    # ------------------------------------------------------------------
+
+    async def restore_bucket(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        *,
+        if_metageneration_match: Optional[int] = None,
+    ) -> Dict[str, Any]:
+        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/restore", creds, params=params
+        )
+
+    # ------------------------------------------------------------------
+    # Object IAM
+    # ------------------------------------------------------------------
+
+    async def get_object_iam(
+        self, creds: GCSCredentials, bucket: str, name: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/iam", creds
+        )
+
+    async def set_object_iam(
+        self, creds: GCSCredentials, bucket: str, name: str, policy: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "PUT", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/iam", creds, json_body=policy
+        )
+
+    # ------------------------------------------------------------------
+    # Object ACLs
+    # ------------------------------------------------------------------
+
+    async def list_object_acl(
+        self, creds: GCSCredentials, bucket: str, name: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl", creds
+        )
+
+    async def insert_object_acl(
+        self, creds: GCSCredentials, bucket: str, name: str, body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl", creds, json_body=body
+        )
+
+    async def get_object_acl(
+        self, creds: GCSCredentials, bucket: str, name: str, entity: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}", creds
+        )
+
+    async def patch_object_acl(
+        self, creds: GCSCredentials, bucket: str, name: str, entity: str, body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "PATCH", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}",
+            creds, json_body=body,
+        )
+
+    async def update_object_acl(
+        self, creds: GCSCredentials, bucket: str, name: str, entity: str, body: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "PUT", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}",
+            creds, json_body=body,
+        )
+
+    async def delete_object_acl(
+        self, creds: GCSCredentials, bucket: str, name: str, entity: str,
+    ) -> Dict[str, Any]:
+        return await self._request(
+            "DELETE", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}", creds
+        )
+
+    # ------------------------------------------------------------------
+    # Public URLs
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def public_url(bucket: str, name: str) -> str:
+        return f"https://storage.googleapis.com/{bucket}/{name}"
+
+    # ------------------------------------------------------------------
+    # Signed URLs (V4, RSA-SHA256)
+    # ------------------------------------------------------------------
+
+    def sign_url_v4(
+        self,
+        creds: GCSCredentials,
+        method: str,
+        bucket: str,
+        name: str,
+        *,
+        expires_in_seconds: Optional[int] = None,
+        content_type: Optional[str] = None,
+        query_params: Optional[Dict[str, str]] = None,
+        response_content_type: Optional[str] = None,
+        response_disposition: Optional[str] = None,
+    ) -> Tuple[str, int]:
+        """Generate a V4 signed URL for a single GCS operation.
+
+        Returns ``(signed_url, effective_expires_seconds)``.
+        """
+        expires = (
+            expires_in_seconds
+            if expires_in_seconds is not None
+            else _settings.gcs_default_expires_seconds
+        )
+        if expires <= 0:
+            raise GCSError("expires_in_seconds must be greater than 0.", status_code=400)
+        if expires > _settings.gcs_max_expires_seconds:
+            raise GCSError(
+                f"expires_in_seconds cannot exceed {_settings.gcs_max_expires_seconds} seconds (7 days).",
+                status_code=400,
+            )
+
+        now = datetime.now(timezone.utc)
+        request_timestamp = now.strftime("%Y%m%dT%H%M%SZ")
+        datestamp = now.strftime("%Y%m%d")
+        credential_scope = f"{datestamp}/{_SIGNING_REGION}/{_SIGNING_SERVICE}/goog4_request"
+        credential = f"{creds.client_email}/{credential_scope}"
+
+        host = "storage.googleapis.com"
+        canonical_uri = f"/{bucket}/{name}"
+
+        headers: Dict[str, str] = {"host": host}
+        if content_type:
+            headers["content-type"] = content_type
+
+        ordered_headers = {k.lower(): str(v).lower() for k, v in sorted(headers.items())}
+        canonical_headers = "".join(f"{k}:{v}\n" for k, v in ordered_headers.items())
+        signed_headers = ";".join(ordered_headers.keys())
+
+        query: Dict[str, Any] = {
+            "X-Goog-Algorithm": _SIGNING_ALGORITHM,
+            "X-Goog-Credential": credential,
+            "X-Goog-Date": request_timestamp,
+            "X-Goog-Expires": str(expires),
+            "X-Goog-SignedHeaders": signed_headers,
+        }
+        if response_content_type:
+            query["response-content-type"] = response_content_type
+        if response_disposition:
+            query["response-content-disposition"] = response_disposition
+        if query_params:
+            query.update(query_params)
+
+        canonical_query_string = "&".join(
+            f"{quote(str(k), safe='')}={quote(str(v), safe='')}"
+            for k, v in sorted(query.items())
+        )
+
+        canonical_request = "\n".join([
+            method.upper(),
+            canonical_uri,
+            canonical_query_string,
+            canonical_headers,
+            signed_headers,
+            "UNSIGNED-PAYLOAD",
+        ])
+
+        canonical_request_hash = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
+        string_to_sign = "\n".join([
+            _SIGNING_ALGORITHM,
+            request_timestamp,
+            credential_scope,
+            canonical_request_hash,
+        ])
+
+        from cryptography.hazmat.primitives import hashes
+        from cryptography.hazmat.primitives.asymmetric import padding
+        from cryptography.hazmat.primitives.serialization import load_pem_private_key
+
+        private_key = load_pem_private_key(creds.private_key.encode("utf-8"), password=None)
+        signature = private_key.sign(string_to_sign.encode("utf-8"), padding.PKCS1v15(), hashes.SHA256())
+        signature_hex = signature.hex()
+
+        signed_url = (
+            f"https://{host}{canonical_uri}?{canonical_query_string}"
+            f"&X-Goog-Signature={signature_hex}"
+        )
+        return signed_url, expires
+
+    async def download_signed_url(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        expires_in_seconds: Optional[int] = None,
+        response_content_type: Optional[str] = None,
+        response_disposition: Optional[str] = None,
+    ) -> Tuple[str, int]:
+        return self.sign_url_v4(
+            creds, "GET", bucket, name,
+            expires_in_seconds=expires_in_seconds,
+            response_content_type=response_content_type,
+            response_disposition=response_disposition,
+        )
+
+    async def upload_signed_url(
+        self,
+        creds: GCSCredentials,
+        bucket: str,
+        name: str,
+        *,
+        expires_in_seconds: Optional[int] = None,
+        content_type: Optional[str] = None,
+    ) -> Tuple[str, int]:
+        return self.sign_url_v4(
+            creds, "PUT", bucket, name,
+            expires_in_seconds=expires_in_seconds,
+            content_type=content_type,
+        )