feat: m2m api

app/api/server.py

@@ -96,6 +96,16 @@ async def lifespan(app: FastAPI):
 
     asyncio.create_task(_self_ping())
 
+    if _settings.supabase_upload_enabled:
+        try:
+            from app.services.media_storage_service import get_storage_service
+
+            storage = await get_storage_service()
+            bucket = await storage.ensure_bucket(_settings.supabase_storage_bucket)
+            _logger.info("Supabase Storage bucket ensured: %s", bucket)
+        except Exception as exc:
+            _logger.error("Failed to ensure Supabase Storage bucket at startup: %s", exc)
+
     await _scheduler_service.start()
     _logger.info("Scheduler service started")
 
@@ -111,6 +121,8 @@ async def lifespan(app: FastAPI):
     await close_maps_service()
     from app.api.v1.gcs import close_gcs_service
     await close_gcs_service()
+    from app.services.media_storage_service import close_storage_service
+    await close_storage_service()
     from app.services.supabase import get_supabase_client
     client = get_supabase_client()
     if client:
@@ -133,6 +145,7 @@ def create_application() -> FastAPI:
             {"name": "Verify", "description": "Phone number and identity verification"},
             {"name": "Vector Stores", "description": "Create, manage, and search vector stores for RAG"},
             {"name": "URL Shortener", "description": "Create and manage short URLs with analytics"},
+            {"name": "Media-to-Media Conversion", "description": "PDF-to-image and image-to-image conversion with local or Supabase Storage output"},
         ],
         lifespan=lifespan,
     )

app/api/v1/media_convert.py

@@ -0,0 +1,310 @@
+from __future__ import annotations
+
+import asyncio
+import time
+from pathlib import Path
+from typing import Annotated, Any, Dict, Optional, Tuple
+from urllib.parse import urlparse
+
+import httpx
+from fastapi import (
+    APIRouter,
+    File,
+    Form,
+    HTTPException,
+    UploadFile,
+)
+from fastapi.responses import FileResponse, JSONResponse
+from pydantic import ValidationError
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.models.schemas import (
+    ImageConversionParams,
+    MediaConversionData,
+    MediaConversionResponse,
+    MediaFormatsResponse,
+    MediaImageFormat,
+    MediaImageUrlRequest,
+    MediaPDFUrlRequest,
+    PDFConversionParams,
+)
+from app.services.media_conversion_service import (
+    MediaConversionError,
+    _IMAGE_OUTPUT_FORMATS,
+    _PDF_OUTPUT_FORMATS,
+    media_conversion_service,
+)
+from app.services.media_storage_service import MediaStorageError
+
+router = APIRouter()
+_logger = get_logger(__name__)
+_settings = get_settings()
+_MAX_UPLOAD_BYTES = _settings.max_upload_bytes
+
+_MIME_BY_EXT = {
+    ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
+    ".png": "image/png", ".webp": "image/webp",
+    ".bmp": "image/bmp", ".gif": "image/gif", ".tiff": "image/tiff",
+}
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _ok_response(start: float, data: MediaConversionData) -> MediaConversionResponse:
+    return MediaConversionResponse(
+        success=True,
+        time_ms=round((time.perf_counter() - start) * 1000, 3),
+        data=data,
+    )
+
+
+def _error_response(start: float, exc: Exception) -> JSONResponse:
+    if isinstance(exc, MediaConversionError):
+        code, message = exc.status_code, exc.message
+    elif isinstance(exc, MediaStorageError):
+        code, message = exc.status_code, exc.message
+    else:
+        code, message = 500, f"Media conversion failed: {exc}"
+    _logger.error("media_conversion_error status=%s error=%s", code, message)
+    return JSONResponse(
+        status_code=code,
+        content={
+            "success": False,
+            "message": message,
+            "time_ms": round((time.perf_counter() - start) * 1000, 3),
+        },
+    )
+
+
+async def _convert_with_timeout(coro) -> Any:
+    try:
+        return await asyncio.wait_for(
+            coro, timeout=_settings.media_conversion_timeout_seconds
+        )
+    except asyncio.TimeoutError as exc:
+        raise MediaConversionError(
+            f"Conversion timed out after {_settings.media_conversion_timeout_seconds}s.",
+            status_code=504,
+        ) from exc
+
+
+def _build_model(model_cls, values: Dict[str, Any]) -> Any:
+    try:
+        return model_cls.model_validate(values)
+    except ValidationError as exc:
+        first = exc.errors()[0]
+        raise MediaConversionError(
+            f"Invalid conversion parameters: {first.get('msg', str(exc))}",
+            status_code=422,
+        ) from exc
+
+
+def _read_upload(file: UploadFile) -> bytes:
+    if file is None or not file.filename:
+        raise HTTPException(status_code=400, detail={"success": False, "message": "No file provided."})
+    raw = file.file.read()
+    if not raw:
+        raise HTTPException(status_code=422, detail={"success": False, "message": "Uploaded file is empty."})
+    if len(raw) > _MAX_UPLOAD_BYTES:
+        raise HTTPException(
+            status_code=413,
+            detail={"success": False, "message": f"File exceeds {_settings.max_upload_mb} MB limit."},
+        )
+    return raw
+
+
+async def _read_stream_limited(resp, limit: int) -> bytes:
+    """Read a streaming response, aborting with a 413 once it exceeds ``limit`` bytes."""
+    chunks: list[bytes] = []
+    total = 0
+    async for chunk in resp.aiter_bytes():
+        total += len(chunk)
+        if total > limit:
+            raise MediaConversionError(
+                f"Remote file exceeds {_settings.max_upload_mb} MB limit.", status_code=413
+            )
+        chunks.append(chunk)
+    return b"".join(chunks)
+
+
+async def _fetch_url(url: str) -> Tuple[bytes, str]:
+    """Verify the URL is downloadable, then download it within the upload size cap."""
+    parsed = urlparse(url)
+    filename = parsed.path.split("/")[-1] or "media"
+    async with httpx.AsyncClient(
+        timeout=30.0,
+        follow_redirects=True,
+        headers={"User-Agent": "agentdeck-media-convert/1.0"},
+    ) as client:
+        # 1) Pre-flight: confirm the URL responds before pulling the body.
+        declared_length: Optional[int] = None
+        try:
+            head_resp = await client.head(url)
+            if head_resp.status_code in (405, 501):  # HEAD not supported -> validate via GET
+                declared_length = None
+            else:
+                head_resp.raise_for_status()
+                content_length = head_resp.headers.get("content-length")
+                if content_length and content_length.isdigit():
+                    declared_length = int(content_length)
+        except httpx.HTTPError as exc:
+            status = exc.response.status_code if exc.response else "network error"
+            raise MediaConversionError(
+                f"URL is not downloadable (HTTP {status}).",
+                status_code=400,
+            ) from exc
+
+        # 2) Reject oversized files up front using the declared Content-Length.
+        if declared_length is not None and declared_length > _MAX_UPLOAD_BYTES:
+            raise MediaConversionError(
+                f"Remote file is {declared_length} bytes, exceeding the "
+                f"{_settings.max_upload_mb} MB limit.",
+                status_code=413,
+            )
+
+        # 3) Stream the download with an enforced byte cap (guards against
+        #    servers that omit or misreport Content-Length).
+        try:
+            async with client.stream("GET", url) as resp:
+                resp.raise_for_status()
+                data = await _read_stream_limited(resp, _MAX_UPLOAD_BYTES)
+        except httpx.HTTPError as exc:
+            raise MediaConversionError(
+                f"Failed to download URL: {exc}", status_code=400
+            ) from exc
+
+    if not data:
+        raise MediaConversionError("Remote file is empty.", status_code=422)
+    return data, filename
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+@router.post(
+    "/media-convert/pdf",
+    response_model=MediaConversionResponse,
+    summary="Convert an uploaded PDF to images",
+)
+async def convert_pdf_file(
+    file: Annotated[UploadFile, File(description="PDF file to convert")],
+    format: Annotated[MediaImageFormat, Form(description="Output format (PNG/JPEG/WEBP)")] = MediaImageFormat.PNG,
+    dpi: int = Form(150, description="Render DPI (72-300)"),
+    quality: int = Form(85, description="Output quality (1-100)"),
+    pages: Optional[str] = Form(None, description="Page spec: '1', '1-3', '1,3,5-7'"),
+    grayscale: bool = Form(False),
+    transparent_bg: bool = Form(False),
+    split_page: bool = Form(False),
+):
+    start = time.perf_counter()
+    raw = _read_upload(file)
+    try:
+        params = _build_model(PDFConversionParams, {
+            "format": format, "dpi": dpi, "quality": quality, "pages": pages,
+            "grayscale": grayscale, "transparent_bg": transparent_bg, "split_page": split_page,
+        })
+        data = await _convert_with_timeout(
+            media_conversion_service.convert_pdf(raw, params, file.filename or "upload.pdf")
+        )
+    except Exception as exc:
+        return _error_response(start, exc)
+    return _ok_response(start, data)
+
+
+@router.post(
+    "/media-convert/pdf/url",
+    response_model=MediaConversionResponse,
+    summary="Convert a PDF from a URL to images",
+)
+async def convert_pdf_url(body: MediaPDFUrlRequest):
+    start = time.perf_counter()
+    try:
+        raw, filename = await _fetch_url(body.url)
+        data = await _convert_with_timeout(
+            media_conversion_service.convert_pdf(raw, body.params, body.url)
+        )
+    except Exception as exc:
+        return _error_response(start, exc)
+    return _ok_response(start, data)
+
+
+@router.post(
+    "/media-convert/image",
+    response_model=MediaConversionResponse,
+    summary="Convert an uploaded image to another image format",
+)
+async def convert_image_file(
+    file: Annotated[UploadFile, File(description="Image file to convert")],
+    format: Annotated[MediaImageFormat, Form(description="Target format")] = MediaImageFormat.JPEG,
+    quality: int = Form(85, description="Output quality (1-100)"),
+    grayscale: bool = Form(False),
+    width: Optional[int] = Form(None, description="Resize width"),
+    height: Optional[int] = Form(None, description="Resize height"),
+    rotate: float = Form(0.0, description="Rotate clockwise in degrees"),
+    flip: Optional[str] = Form(None, description="horizontal or vertical"),
+):
+    start = time.perf_counter()
+    raw = _read_upload(file)
+    try:
+        params = _build_model(ImageConversionParams, {
+            "format": format, "quality": quality, "grayscale": grayscale,
+            "width": width, "height": height, "rotate": rotate, "flip": flip,
+        })
+        data = await _convert_with_timeout(
+            media_conversion_service.convert_image(raw, file.filename or "image", params)
+        )
+    except Exception as exc:
+        return _error_response(start, exc)
+    return _ok_response(start, data)
+
+
+@router.post(
+    "/media-convert/image/url",
+    response_model=MediaConversionResponse,
+    summary="Convert an image from a URL to another image format",
+)
+async def convert_image_url(body: MediaImageUrlRequest):
+    start = time.perf_counter()
+    try:
+        raw, filename = await _fetch_url(body.url)
+        data = await _convert_with_timeout(
+            media_conversion_service.convert_image(raw, body.url, body.params)
+        )
+    except Exception as exc:
+        return _error_response(start, exc)
+    return _ok_response(start, data)
+
+
+@router.get(
+    "/media-convert/files/{job_id}/{filename}",
+    summary="Download a converted file (local mode)",
+    responses={404: {"description": "File not found"}},
+)
+async def download_file(job_id: str, filename: str):
+    root = Path(_settings.media_output_dir).resolve()
+    if job_id != Path(job_id).name or filename != Path(filename).name:
+        raise HTTPException(status_code=404, detail={"success": False, "message": "File not found."})
+    path = (root / job_id / "out" / filename).resolve()
+    if not str(path).startswith(str(root)) or not path.is_file():
+        raise HTTPException(status_code=404, detail={"success": False, "message": "File not found."})
+    media_type = _MIME_BY_EXT.get(path.suffix.lower(), "application/octet-stream")
+    return FileResponse(path, media_type=media_type, filename=filename)
+
+
+@router.get(
+    "/media-convert/formats",
+    response_model=MediaFormatsResponse,
+    summary="List supported media conversions",
+)
+async def list_formats():
+    return MediaFormatsResponse(
+        success=True,
+        pdf_to_image=sorted(_PDF_OUTPUT_FORMATS),
+        image_to_image=sorted(_IMAGE_OUTPUT_FORMATS),
+        storage_enabled=_settings.supabase_upload_enabled,
+        signed_url_ttl_seconds=_settings.supabase_signed_url_ttl_seconds,
+    )

app/api/v1/qr_decoder.py

@@ -6,6 +6,7 @@ from typing import List, Optional
 from fastapi import APIRouter, File, HTTPException, UploadFile
 from pydantic import BaseModel, Field
 
+from app.config import get_settings
 from app.core.logger import get_logger
 from app.services.qr_decoder_service import QRDecoderService
 
@@ -13,9 +14,10 @@ logger = get_logger(__name__)
 
 router = APIRouter()
 
-MAX_UPLOAD_BYTES = 20 * 1024 * 1024
+_settings = get_settings()
+MAX_UPLOAD_BYTES = _settings.max_upload_bytes
 
-_service = QRDecoderService()
+_service = QRDecoderService(max_file_size_mb=_settings.max_upload_mb)
 
 
 class QRDecodeUrlRequest(BaseModel):

app/api/v1/router.py

@@ -16,6 +16,7 @@ from app.api.v1 import (
     google_oauth,
     json_extract,
     keys_extract,
+    media_convert,
     qr_decoder,
     qr_generator,
     reconcile,
@@ -57,6 +58,7 @@ 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(media_convert.router, tags=["Media-to-Media Conversion"])
 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

@@ -124,6 +124,24 @@ class Settings(BaseSettings):
     google_oauth_state_ttl_minutes: int = 10
     google_oauth_jwks_ttl_seconds: int = 3600
 
+    # Media-to-Media conversion settings
+    media_output_dir: str = "./data/media-convert"
+    media_max_workers: int = 4
+    media_default_dpi: int = 150
+    media_max_dpi: int = 300
+    media_default_quality: int = 85
+    media_max_pages: int = 200
+    media_max_memory_mb: int = 512
+    media_max_image_pixels: int = 50_000_000
+    media_upload_concurrency: int = 4
+    media_signed_url_ttl_seconds: int = 86400
+    media_conversion_timeout_seconds: int = 300
+
+    # Supabase Storage upload for converted media
+    supabase_upload_enabled: bool = Field(default=False, alias="SUPABASE_UPLOAD_ENABLED")
+    supabase_storage_bucket: str = Field(default="media-convert", alias="SUPABASE_STORAGE_BUCKET")
+    supabase_signed_url_ttl_seconds: int = Field(default=86400, alias="SUPABASE_SIGNED_URL_TTL_SECONDS")
+
     # Scheduler settings
     max_http_timeout: float = 300.0
     default_scheduler_timezone: str = "UTC"

app/models/schemas.py

@@ -1449,3 +1449,188 @@ class GCSURLResponse(BaseModel):
     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
+
+
+# ---------------------------------------------------------------------------
+# Media-to-Media conversion
+# ---------------------------------------------------------------------------
+
+class MediaImageFormat(str, Enum):
+    """Output image formats supported by the media converter."""
+
+    JPEG = "JPEG"
+    JPG = "JPEG"
+    PNG = "PNG"
+    WEBP = "WEBP"
+    BMP = "BMP"
+    GIF = "GIF"
+    TIFF = "TIFF"
+
+
+class PDFConversionParams(BaseModel):
+    """Options for PDF -> image conversion."""
+
+    format: MediaImageFormat = Field(MediaImageFormat.PNG, description="Output image format")
+    dpi: int = Field(150, ge=72, le=300, description="Render DPI (72-300)")
+    quality: int = Field(85, ge=1, le=100, description="Output quality (1-100, JPEG/WEBP/TIFF)")
+    pages: Optional[str] = Field(None, description="Page spec: '1', '1-3', '1,3,5-7'. Empty = all pages.")
+    grayscale: bool = Field(False, description="Convert output to grayscale")
+    transparent_bg: bool = Field(False, description="Preserve transparency (PNG only)")
+    split_page: bool = Field(False, description="True: each page is a separate image. False: stitch into one tall image.")
+
+    @field_validator("format", mode="before")
+    @classmethod
+    def normalise_format(cls, v):
+        if isinstance(v, str):
+            upper = v.upper()
+            if upper == "JPG":
+                return "JPEG"
+            return upper
+        return v
+
+    @field_validator("pages")
+    @classmethod
+    def validate_page_spec(cls, v):
+        if v is None or str(v).strip() == "":
+            return None
+        spec = str(v)
+        if not re.match(r"^\s*(\d+(-\d+)?)(\s*,\s*(\d+(-\d+)?))*\s*$", spec):
+            raise ValueError(
+                f"Invalid page spec '{v}'. Use comma-separated numbers or ranges, "
+                "e.g. '1', '1-3', '1,3,5-7'."
+            )
+        for token in spec.split(","):
+            token = token.strip()
+            if "-" in token:
+                start, end = (int(x) for x in token.split("-", 1))
+                if start < 1 or end < 1:
+                    raise ValueError(f"Page numbers are 1-indexed; got '{token}'")
+                if start > end:
+                    raise ValueError(f"Invalid range '{token}': start must be <= end")
+            elif int(token) < 1:
+                raise ValueError(f"Page numbers are 1-indexed; got '{token}'")
+        return spec
+
+    @model_validator(mode="after")
+    def validate_transparent_bg(self):
+        if self.transparent_bg and self.format not in (MediaImageFormat.PNG, MediaImageFormat.WEBP):
+            raise ValueError("transparent_bg is only supported for PNG or WEBP output")
+        return self
+
+
+class ImageConversionParams(BaseModel):
+    """Options for image -> image conversion."""
+
+    format: MediaImageFormat = Field(MediaImageFormat.JPEG, description="Target output format")
+    quality: int = Field(85, ge=1, le=100, description="Output quality (1-100, JPEG/WEBP/TIFF)")
+    grayscale: bool = Field(False, description="Convert output to grayscale")
+    width: Optional[int] = Field(None, ge=1, le=100000, description="Resize width (keeps aspect ratio if height omitted)")
+    height: Optional[int] = Field(None, ge=1, le=100000, description="Resize height (keeps aspect ratio if width omitted)")
+    rotate: float = Field(0.0, ge=0, le=360, description="Rotate clockwise in degrees")
+    flip: Optional[Literal["horizontal", "vertical"]] = Field(None, description="Flip the image")
+
+    @field_validator("format", mode="before")
+    @classmethod
+    def normalise_format(cls, v):
+        if isinstance(v, str):
+            upper = v.upper()
+            if upper == "JPG":
+                return "JPEG"
+            return upper
+        return v
+
+    @model_validator(mode="after")
+    def validate_resize(self):
+        if self.width is None and self.height is None:
+            return self
+        if self.width is not None and self.width < 1:
+            raise ValueError("width must be a positive integer")
+        if self.height is not None and self.height < 1:
+            raise ValueError("height must be a positive integer")
+        return self
+
+
+class MediaOutputFile(BaseModel):
+    """A single converted media file produced by a conversion job."""
+
+    filename: str = Field(..., description="Output file name")
+    page_number: Optional[int] = Field(None, description="Source page number for PDF conversions (1-indexed)")
+    width: int = Field(0, description="Pixel width")
+    height: int = Field(0, description="Pixel height")
+    size_bytes: int = Field(0, description="File size in bytes")
+    format: str = Field(..., description="Output format (e.g. JPEG, PNG)")
+    content_type: str = Field("application/octet-stream", description="MIME type of the output")
+    url: str = Field(..., description="Download URL (Supabase signed URL or local endpoint)")
+    base64: Optional[str] = Field(None, description="Base64-encoded file bytes (local mode only)")
+
+
+class MediaUploadSummary(BaseModel):
+    """How the converted files are exposed to the caller."""
+
+    mode: Literal["storage", "local"] = Field("local", description="'storage' = Supabase Storage, 'local' = served from this service")
+    bucket: Optional[str] = Field(None, description="Supabase Storage bucket used (storage mode)")
+    total_files: int = Field(0, description="Number of files uploaded / linked")
+    failed_uploads: int = Field(0, description="Number of uploads that failed (storage mode)")
+    url_ttl_seconds: Optional[int] = Field(None, description="Signed URL lifetime in seconds (storage mode)")
+    expires_at: Optional[str] = Field(None, description="ISO-8601 timestamp when signed URLs expire (storage mode)")
+
+
+class MediaConversionData(BaseModel):
+    """Structured result of a media-to-media conversion."""
+
+    job_id: str = Field(..., description="Unique job identifier")
+    source: str = Field(..., description="Source file name or URL")
+    input_format: str = Field(..., description="Detected input media format")
+    output_format: str = Field(..., description="Primary output format")
+    total_pages: int = Field(0, description="Source page count (PDF) or 1 (images)")
+    converted_files: int = Field(0, description="Number of output files produced")
+    outputs: List[MediaOutputFile] = Field(default_factory=list, description="Converted files with URLs")
+    upload: MediaUploadSummary = Field(default_factory=MediaUploadSummary, description="Upload / exposure details")
+    warning: Optional[str] = Field(None, description="Warnings, e.g. signed URL expiry notice")
+
+
+class MediaConversionResponse(BaseModel):
+    """Envelope for synchronous media conversion responses."""
+
+    success: bool
+    time_ms: float = 0.0
+    data: Optional[MediaConversionData] = None
+    error: Optional[str] = None
+
+
+class MediaFormatsResponse(BaseModel):
+    """Lists the supported media-to-media conversions."""
+
+    success: bool
+    pdf_to_image: List[str] = Field(default_factory=list, description="Output formats from PDF input")
+    image_to_image: List[str] = Field(default_factory=list, description="Output formats from image input")
+    storage_enabled: bool = Field(False, description="Whether Supabase Storage upload is enabled")
+    signed_url_ttl_seconds: int = Field(86400, description="Signed URL lifetime when storage is enabled")
+
+
+class MediaPDFUrlRequest(BaseModel):
+    """Body for converting a PDF fetched from a URL."""
+
+    url: str = Field(..., description="Publicly reachable PDF URL")
+    params: PDFConversionParams = Field(default_factory=PDFConversionParams)
+
+    @field_validator("url")
+    @classmethod
+    def validate_scheme(cls, v: str) -> str:
+        if not v.startswith(("http://", "https://")):
+            raise ValueError("Only http/https URLs are supported.")
+        return v
+
+
+class MediaImageUrlRequest(BaseModel):
+    """Body for converting an image fetched from a URL."""
+
+    url: str = Field(..., description="Publicly reachable image URL")
+    params: ImageConversionParams = Field(default_factory=ImageConversionParams)
+
+    @field_validator("url")
+    @classmethod
+    def validate_scheme(cls, v: str) -> str:
+        if not v.startswith(("http://", "https://")):
+            raise ValueError("Only http/https URLs are supported.")
+        return v

app/services/media_conversion_service.py

@@ -0,0 +1,596 @@
+"""Media-to-Media conversion service.
+
+Converts between media formats without blocking the FastAPI event loop:
+
+  * PDF -> images (JPEG / PNG / WEBP) using pypdfium2 + Pillow.
+  * Image -> image (JPEG / PNG / WEBP / BMP / GIF / TIFF) using Pillow.
+
+All CPU-bound pixel work runs in the shared thread pool
+(:mod:`app.core.thread_pool`), mirroring the architecture of the document
+converter and the reference PDF-conversion service.
+
+Output files are written to a per-job directory and then either:
+
+  * exposed through local download endpoints (``SUPABASE_UPLOAD_ENABLED=false``),
+    or
+  * uploaded to Supabase Storage with 24-hour signed URLs
+    (``SUPABASE_UPLOAD_ENABLED=true``) and a warning stating the expiry.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import io
+import re
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+from app.config import get_settings
+from app.core.logger import get_logger
+from app.core.thread_pool import thread_pool as _thread_pool
+from app.models.schemas import (
+    ImageConversionParams,
+    MediaConversionData,
+    MediaOutputFile,
+    MediaUploadSummary,
+    PDFConversionParams,
+)
+
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+_PAGE_SPEC_RE = re.compile(r"^\s*(\d+(-\d+)?)(\s*,\s*(\d+(-\d+)?))*\s*$")
+
+_PDF_OUTPUT_FORMATS = frozenset({"JPEG", "PNG", "WEBP"})
+_IMAGE_OUTPUT_FORMATS = frozenset({"JPEG", "PNG", "WEBP", "BMP", "GIF", "TIFF"})
+
+_EXT_BY_FORMAT: Dict[str, str] = {
+    "JPEG": "jpg", "PNG": "png", "WEBP": "webp",
+    "BMP": "bmp", "GIF": "gif", "TIFF": "tiff",
+}
+_MIME_BY_FORMAT: Dict[str, str] = {
+    "JPEG": "image/jpeg", "PNG": "image/png", "WEBP": "image/webp",
+    "BMP": "image/bmp", "GIF": "image/gif", "TIFF": "image/tiff",
+}
+
+
+class MediaConversionError(Exception):
+    """Raised for invalid input or failed conversions, mapped to HTTP errors."""
+
+    def __init__(self, message: str, status_code: int = 400) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+def _fmt_str(fmt) -> str:
+    return fmt.value if hasattr(fmt, "value") else str(fmt)
+
+
+def _ext_for(fmt: str) -> str:
+    return _EXT_BY_FORMAT.get(fmt, "bin")
+
+
+def _save_kwargs(fmt: str, quality: int) -> Dict[str, Any]:
+    if fmt == "JPEG":
+        return {"quality": quality, "optimize": True}
+    if fmt == "PNG":
+        return {"optimize": True}
+    if fmt == "WEBP":
+        return {"quality": quality, "method": 4}
+    if fmt == "TIFF":
+        return {"compression": "tiff_lzw"}
+    return {}
+
+
+def _normalise_for_jpeg(img):
+    """Return an RGB image suitable for JPEG, flattening alpha onto white."""
+    from PIL import Image
+
+    mode = img.mode
+    if mode == "RGB":
+        return img
+    if mode in ("RGBA", "LA", "P"):
+        if mode == "P":
+            img = img.convert("RGBA")
+        bg = Image.new("RGB", img.size, (255, 255, 255))
+        mask = img.split()[-1] if mode in ("RGBA", "LA") else None
+        bg.paste(img, mask=mask)
+        return bg
+    return img.convert("RGB")
+
+
+def _normalise_for_png(img):
+    """Return an image whose mode PIL accepts for PNG / WEBP."""
+    if img.mode in ("RGB", "RGBA", "L", "LA"):
+        return img
+    if img.mode == "P":
+        return img.convert("RGBA")
+    return img.convert("RGB")
+
+
+def _parse_pages(spec: Optional[str], total_pages: int) -> List[int]:
+    """Expand a page spec ('1', '1-3', '1,3,5-7') into 0-indexed page indices."""
+    if not spec or not spec.strip():
+        return list(range(total_pages))
+    indices: set[int] = set()
+    for token in str(spec).split(","):
+        token = token.strip()
+        if "-" in token:
+            start, end = (int(x) for x in token.split("-", 1))
+            indices.update(range(start - 1, end))
+        else:
+            indices.add(int(token) - 1)
+    valid = sorted(i for i in indices if 0 <= i < total_pages)
+    if not valid:
+        raise MediaConversionError(
+            f"Page spec '{spec}' contains no pages within the document's {total_pages} page(s).",
+            status_code=422,
+        )
+    return valid
+
+
+def _guard_memory(page_count: int, dpi: int) -> None:
+    bytes_per_page = (dpi * 8.5) * (dpi * 11) * 3
+    estimated_mb = (bytes_per_page * page_count) / (1024 * 1024)
+    if estimated_mb > _settings.media_max_memory_mb:
+        raise MediaConversionError(
+            f"Requested conversion would need ~{estimated_mb:.0f} MB of memory "
+            f"({page_count} pages at {dpi} DPI). Reduce DPI or select fewer pages.",
+            status_code=422,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Thread-pool worker functions (must stay top-level for pickling)
+# ---------------------------------------------------------------------------
+
+def _pdf_total_pages(data: bytes) -> int:
+    import pypdfium2 as pdfium
+
+    try:
+        doc = pdfium.PdfDocument(data)
+    except Exception as exc:
+        raise MediaConversionError(
+            f"Input is not a valid PDF: {exc}", status_code=422
+        ) from exc
+    try:
+        total = len(doc)
+    finally:
+        doc.close()
+    if total == 0:
+        raise MediaConversionError("PDF contains no pages.", status_code=422)
+    return total
+
+
+def _split_pdf_pages(data: bytes, page_indices: List[int]) -> Dict[int, bytes]:
+    """Extract each requested page into its own single-page PDF blob."""
+    import pypdfium2 as pdfium
+
+    blobs: Dict[int, bytes] = {}
+    try:
+        src = pdfium.PdfDocument(data)
+        for page_idx in page_indices:
+            dst = pdfium.PdfDocument.new()
+            try:
+                dst.import_pages(src, pages=[page_idx])
+                with io.BytesIO() as buf:
+                    dst.save(buf)
+                    blobs[page_idx] = buf.getvalue()
+            finally:
+                dst.close()
+    finally:
+        src.close()
+    return blobs
+
+
+def _render_pdf_page(blob: bytes, params: PDFConversionParams, out_path: Path) -> Tuple[int, int, int]:
+    """Render a single-page PDF blob to an image file. Returns (width, height, size_bytes)."""
+    from PIL import Image
+    import pypdfium2 as pdfium
+
+    fmt = _fmt_str(params.format)
+    doc = None
+    bitmap = None
+    try:
+        doc = pdfium.PdfDocument(blob)
+        scale = params.dpi / 72.0
+        bitmap = doc[0].render(scale=scale)
+        img = bitmap.to_pil()
+    except Exception as exc:
+        raise MediaConversionError(f"Failed to render PDF page: {exc}", status_code=500) from exc
+    finally:
+        if bitmap is not None:
+            try:
+                bitmap.close()
+            except Exception:
+                pass
+        if doc is not None:
+            try:
+                doc.close()
+            except Exception:
+                pass
+
+    if fmt == "JPEG":
+        img = _normalise_for_jpeg(img)
+    else:
+        if params.transparent_bg:
+            img = _normalise_for_png(img)
+        else:
+            img = _normalise_for_png(img)
+            if img.mode in ("RGBA", "LA"):
+                bg = Image.new("RGB", img.size, (255, 255, 255))
+                bg.paste(img, mask=img.split()[-1] if img.mode == "RGBA" else None)
+                img = bg
+    if params.grayscale:
+        img = img.convert("L")
+
+    try:
+        img.save(str(out_path), format=fmt, **_save_kwargs(fmt, params.quality))
+    except OSError as exc:
+        raise MediaConversionError(f"Failed to write output file '{out_path.name}': {exc}", status_code=500) from exc
+
+    stat = out_path.stat()
+    return img.width, img.height, stat.st_size
+
+
+def _stitch_images(page_files: List[Tuple[int, Path]], fmt: str, out_path: Path, quality: int) -> Tuple[int, int, int]:
+    """Stitch rendered page images vertically into one tall image."""
+    from PIL import Image
+
+    images: List[Image.Image] = []
+    for _, path in sorted(page_files, key=lambda t: t[0]):
+        img = Image.open(path)
+        if fmt == "JPEG":
+            img = _normalise_for_jpeg(img)
+        else:
+            img = _normalise_for_png(img)
+        images.append(img)
+
+    total_width = max(im.width for im in images)
+    total_height = sum(im.height for im in images)
+    mode = images[0].mode
+    fill = (255, 255, 255) if mode == "RGB" else 255
+    stitched = Image.new(mode, (total_width, total_height), color=fill)
+    y_offset = 0
+    for img in images:
+        stitched.paste(img, (0, y_offset))
+        y_offset += img.height
+
+    stitched.save(str(out_path), format=fmt, **_save_kwargs(fmt, quality))
+    stat = out_path.stat()
+    return stitched.width, stitched.height, stat.st_size
+
+
+def _convert_image_bytes(data: bytes, params: ImageConversionParams, out_path: Path) -> Tuple[int, int, int, str]:
+    """Convert raw image bytes to the requested format. Returns (width, height, size_bytes, detected_format)."""
+    from PIL import Image, UnidentifiedImageError
+
+    try:
+        img = Image.open(io.BytesIO(data))
+        detected = (img.format or "UNKNOWN").upper()
+        img.load()
+    except (UnidentifiedImageError, OSError, ValueError) as exc:
+        raise MediaConversionError(
+            f"Input is not a valid or supported image: {exc}", status_code=422
+        ) from exc
+
+    if params.rotate:
+        img = img.rotate(params.rotate % 360, expand=True)
+    if params.flip == "horizontal":
+        img = img.transpose(Image.FLIP_LEFT_RIGHT)
+    elif params.flip == "vertical":
+        img = img.transpose(Image.FLIP_TOP_BOTTOM)
+    if params.grayscale:
+        img = img.convert("L")
+
+    if params.width is not None or params.height is not None:
+        orig_w, orig_h = img.size
+        if params.width is not None and params.height is not None:
+            target = (params.width, params.height)
+        elif params.width is not None:
+            ratio = params.width / orig_w
+            target = (params.width, max(1, round(orig_h * ratio)))
+        else:
+            ratio = params.height / orig_h
+            target = (max(1, round(orig_w * ratio)), params.height)
+        if target[0] * target[1] > _settings.media_max_image_pixels:
+            raise MediaConversionError(
+                f"Resized image ({target[0]}x{target[1]}) exceeds the "
+                f"{_settings.media_max_image_pixels} pixel limit.",
+                status_code=422,
+            )
+        img = img.resize(target, Image.LANCZOS)
+
+    fmt = _fmt_str(params.format)
+    if fmt == "JPEG":
+        img = _normalise_for_jpeg(img)
+    elif fmt in ("PNG", "WEBP") and img.mode not in ("RGB", "RGBA", "L", "LA"):
+        img = _normalise_for_png(img)
+
+    try:
+        img.save(str(out_path), format=fmt, **_save_kwargs(fmt, params.quality))
+    except OSError as exc:
+        raise MediaConversionError(f"Failed to write output file '{out_path.name}': {exc}", status_code=500) from exc
+
+    stat = out_path.stat()
+    return img.width, img.height, stat.st_size, detected
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+class MediaConversionService:
+    """Orchestrates media conversion plus local/storage exposure of results."""
+
+    def __init__(self, storage=None) -> None:
+        self._storage = storage
+
+    async def _resolve_storage(self):
+        if self._storage is not None:
+            return self._storage
+        from app.services.media_storage_service import get_storage_service
+
+        return await get_storage_service()
+
+    async def convert_pdf(
+        self,
+        data: bytes,
+        params: PDFConversionParams,
+        source: str,
+        job_id: Optional[str] = None,
+    ) -> MediaConversionData:
+        """Convert a PDF to images. Runs validation + rendering in the thread pool."""
+        job_id = job_id or str(uuid.uuid4())
+        fmt = _fmt_str(params.format)
+        if fmt not in _PDF_OUTPUT_FORMATS:
+            raise MediaConversionError(
+                f"Unsupported output format '{fmt}'. PDF can only be converted to JPEG, PNG or WEBP.",
+                status_code=422,
+            )
+
+        loop = asyncio.get_running_loop()
+        total_pages = await loop.run_in_executor(_thread_pool, _pdf_total_pages, data)
+        if total_pages > _settings.media_max_pages:
+            raise MediaConversionError(
+                f"PDF has {total_pages} pages, exceeding the {_settings.media_max_pages} page limit.",
+                status_code=422,
+            )
+        page_indices = _parse_pages(params.pages, total_pages)
+        _guard_memory(len(page_indices), params.dpi)
+
+        start = time.perf_counter()
+        job_dir = self._job_dir(job_id)
+        out_dir = job_dir / "out"
+        out_dir.mkdir(parents=True, exist_ok=True)
+
+        page_blobs = await loop.run_in_executor(
+            _thread_pool, _split_pdf_pages, data, page_indices
+        )
+
+        if params.split_page:
+            render_coros = [
+                loop.run_in_executor(
+                    _thread_pool,
+                    _render_pdf_page,
+                    page_blobs[idx],
+                    params,
+                    out_dir / f"page_{idx + 1:04d}.{_ext_for(fmt)}",
+                )
+                for idx in sorted(page_blobs)
+            ]
+            outcomes = await asyncio.gather(*render_coros, return_exceptions=True)
+            files: List[MediaOutputFile] = []
+            for idx, outcome in zip(sorted(page_blobs), outcomes):
+                if isinstance(outcome, Exception):
+                    _logger.error("page_render_failed job=%s page=%d error=%s", job_id, idx + 1, outcome)
+                    raise MediaConversionError(
+                        f"Failed to render page {idx + 1}: {outcome}", status_code=500
+                    ) from outcome
+                width, height, size = outcome
+                files.append(self._output_file(
+                    f"page_{idx + 1:04d}.{_ext_for(fmt)}", idx + 1,
+                    width, height, size, fmt,
+                ))
+        else:
+            if len(page_blobs) == 1:
+                width, height, size = await loop.run_in_executor(
+                    _thread_pool,
+                    _render_pdf_page,
+                    page_blobs[sorted(page_blobs)[0]],
+                    params,
+                    out_dir / f"stitched.{_ext_for(fmt)}",
+                )
+            else:
+                render_coros = [
+                    loop.run_in_executor(
+                        _thread_pool,
+                        _render_pdf_page,
+                        page_blobs[idx],
+                        params,
+                        out_dir / f"_page_{idx + 1:04d}.{_ext_for(fmt)}",
+                    )
+                    for idx in sorted(page_blobs)
+                ]
+                outcomes = await asyncio.gather(*render_coros, return_exceptions=True)
+                for idx, outcome in zip(sorted(page_blobs), outcomes):
+                    if isinstance(outcome, Exception):
+                        raise MediaConversionError(
+                            f"Failed to render page {idx + 1}: {outcome}", status_code=500
+                        ) from outcome
+                page_files = [(idx, out_dir / f"_page_{idx + 1:04d}.{_ext_for(fmt)}") for idx in sorted(page_blobs)]
+                width, height, size = await loop.run_in_executor(
+                    _thread_pool,
+                    _stitch_images,
+                    page_files,
+                    fmt,
+                    out_dir / f"stitched.{_ext_for(fmt)}",
+                    params.quality,
+                )
+                for _, tmp_path in page_files:
+                    tmp_path.unlink(missing_ok=True)
+            files = [self._output_file(
+                f"stitched.{_ext_for(fmt)}", None, width, height, size, fmt,
+            )]
+
+        upload, warning = await self._expose(files, job_id, out_dir, job_dir)
+        duration_ms = round((time.perf_counter() - start) * 1000, 2)
+        _logger.info(
+            "pdf_conversion_complete job=%s pages=%d duration_ms=%s mode=%s",
+            job_id, total_pages, duration_ms, upload.mode,
+        )
+        return MediaConversionData(
+            job_id=job_id,
+            source=source,
+            input_format="PDF",
+            output_format=fmt,
+            total_pages=total_pages,
+            converted_files=len(files),
+            outputs=files,
+            upload=upload,
+            warning=warning,
+        )
+
+    async def convert_image(
+        self,
+        data: bytes,
+        filename: str,
+        params: ImageConversionParams,
+        job_id: Optional[str] = None,
+    ) -> MediaConversionData:
+        """Convert an image to a target image format."""
+        job_id = job_id or str(uuid.uuid4())
+        fmt = _fmt_str(params.format)
+        if fmt not in _IMAGE_OUTPUT_FORMATS:
+            raise MediaConversionError(
+                f"Unsupported output format '{fmt}'.",
+                status_code=422,
+            )
+
+        start = time.perf_counter()
+        job_dir = self._job_dir(job_id)
+        out_dir = job_dir / "out"
+        out_dir.mkdir(parents=True, exist_ok=True)
+        stem = Path(filename or "image").stem or "image"
+        out_path = out_dir / f"{stem}.{_ext_for(fmt)}"
+
+        loop = asyncio.get_running_loop()
+        width, height, size, detected = await loop.run_in_executor(
+            _thread_pool, _convert_image_bytes, data, params, out_path
+        )
+
+        file = self._output_file(out_path.name, None, width, height, size, fmt)
+        upload, warning = await self._expose([file], job_id, out_dir, job_dir)
+        duration_ms = round((time.perf_counter() - start) * 1000, 2)
+        _logger.info(
+            "image_conversion_complete job=%s input=%s duration_ms=%s mode=%s",
+            job_id, detected, duration_ms, upload.mode,
+        )
+        return MediaConversionData(
+            job_id=job_id,
+            source=filename,
+            input_format=detected or "IMAGE",
+            output_format=fmt,
+            total_pages=1,
+            converted_files=1,
+            outputs=[file],
+            upload=upload,
+            warning=warning,
+        )
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _job_dir(job_id: str) -> Path:
+        root = Path(_settings.media_output_dir).resolve()
+        root.mkdir(parents=True, exist_ok=True)
+        return root / job_id
+
+    @staticmethod
+    def _output_file(
+        filename: str,
+        page_number: Optional[int],
+        width: int,
+        height: int,
+        size_bytes: int,
+        fmt: str,
+    ) -> MediaOutputFile:
+        return MediaOutputFile(
+            filename=filename,
+            page_number=page_number,
+            width=width,
+            height=height,
+            size_bytes=size_bytes,
+            format=fmt,
+            content_type=_MIME_BY_FORMAT.get(fmt, "application/octet-stream"),
+            url="",  # filled in by _expose
+        )
+
+    async def _expose(
+        self,
+        files: List[MediaOutputFile],
+        job_id: str,
+        out_dir: Path,
+        job_dir: Path,
+    ) -> Tuple[MediaUploadSummary, Optional[str]]:
+        """Expose output files via Supabase signed URLs or local download endpoints."""
+        if _settings.supabase_upload_enabled:
+            storage = await self._resolve_storage()
+            bucket = await storage.ensure_bucket(_settings.supabase_storage_bucket)
+            ttl = _settings.supabase_signed_url_ttl_seconds
+            semaphore = asyncio.Semaphore(max(1, _settings.media_upload_concurrency))
+
+            from app.services.media_storage_service import iso_expiry
+
+            async def _upload_one(f: MediaOutputFile) -> MediaOutputFile:
+                storage_path = f"{job_id}/{f.filename}"
+                async with semaphore:
+                    await storage.upload_file(
+                        bucket, storage_path,
+                        (out_dir / f.filename).read_bytes(), f.content_type,
+                    )
+                    f.url = await storage.create_signed_url(bucket, storage_path, ttl)
+                return f
+
+            outcomes = await asyncio.gather(*[_upload_one(f) for f in files], return_exceptions=True)
+            failed = 0
+            for f, outcome in zip(files, outcomes):
+                if isinstance(outcome, Exception):
+                    failed += 1
+                    _logger.error("storage_upload_failed job=%s file=%s error=%s", job_id, f.filename, outcome)
+                    f.url = f"/api/v1/media-convert/files/{job_id}/{f.filename}"
+
+            expires_at = iso_expiry(ttl)
+            warning = (
+                f"Converted files were uploaded to Supabase Storage bucket "
+                f"'{bucket}'. The returned signed URLs are valid for 24 hours "
+                f"(expire at {expires_at}). Regenerate by re-running the conversion."
+            )
+            return (
+                MediaUploadSummary(
+                    mode="storage",
+                    bucket=bucket,
+                    total_files=len(files),
+                    failed_uploads=failed,
+                    url_ttl_seconds=ttl,
+                    expires_at=expires_at,
+                ),
+                warning,
+            )
+
+        for f in files:
+            f.url = f"/api/v1/media-convert/files/{job_id}/{f.filename}"
+            try:
+                f.base64 = base64.b64encode((out_dir / f.filename).read_bytes()).decode("ascii")
+            except OSError:
+                f.base64 = None
+        return MediaUploadSummary(mode="local", total_files=len(files), failed_uploads=0), None
+
+
+media_conversion_service = MediaConversionService()

app/services/media_storage_service.py

@@ -0,0 +1,181 @@
+"""Supabase Storage integration for the Media-to-Media conversion APIs.
+
+Provides a lazy, connection-light async client that:
+  1. Ensures the destination bucket exists (creating it if needed).
+  2. Uploads converted media files.
+  3. Returns short-lived signed URLs (default 24 hours) for download.
+
+All storage3 calls are already async/HTTP-based, so no thread-pool offload is
+required here. The client is created on first use and closed during shutdown.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+from app.config import get_settings
+from app.core.logger import get_logger
+
+_logger = get_logger(__name__)
+_settings = get_settings()
+
+
+class MediaStorageError(Exception):
+    """Raised when Supabase Storage operations fail."""
+
+    def __init__(self, message: str, status_code: int = 502) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+class MediaStorageService:
+    """Async client for uploading converted media to Supabase Storage."""
+
+    def __init__(self) -> None:
+        self._client: Optional[Any] = None
+        self._client_lock = asyncio.Lock()
+        self._known_buckets: set[str] = set()
+
+    # ------------------------------------------------------------------
+    # Client management
+    # ------------------------------------------------------------------
+
+    async def _get_client(self):
+        """Lazily build the Supabase Storage async client."""
+        if self._client is None:
+            async with self._client_lock:
+                if self._client is None:
+                    url = _settings.supabase_url
+                    key = _settings.supabase_service_role_key
+                    if not url or not key:
+                        raise MediaStorageError(
+                            "Supabase is not configured. Set SUPABASE_URL and "
+                            "SUPABASE_SERVICE_ROLE_KEY to enable storage uploads.",
+                            status_code=503,
+                        )
+                    from storage3 import AsyncStorageClient
+
+                    self._client = AsyncStorageClient(
+                        url,
+                        headers={
+                            "apikey": key,
+                            "Authorization": f"Bearer {key}",
+                        },
+                        timeout=60,
+                    )
+                    _logger.info("Supabase Storage client initialized")
+        return self._client
+
+    async def close(self) -> None:
+        async with self._client_lock:
+            if self._client is not None:
+                try:
+                    await self._client.aclose()
+                except Exception as exc:  # pragma: no cover - defensive
+                    _logger.debug("Error closing Supabase Storage client: %s", exc)
+                self._client = None
+
+    # ------------------------------------------------------------------
+    # Bucket management
+    # ------------------------------------------------------------------
+
+    async def ensure_bucket(self, bucket_id: str) -> str:
+        """Verify the bucket exists, creating it (private) if it does not."""
+        bucket_id = (bucket_id or _settings.supabase_storage_bucket or "media-convert").strip()
+        if not bucket_id:
+            raise MediaStorageError("A non-empty storage bucket name is required.", status_code=400)
+        if bucket_id in self._known_buckets:
+            return bucket_id
+
+        client = await self._get_client()
+        exists = False
+        try:
+            buckets = await client.list_buckets()
+            exists = any(getattr(b, "id", None) == bucket_id for b in buckets)
+        except Exception as exc:
+            _logger.warning("Could not list Supabase buckets: %s", exc)
+
+        if not exists:
+            try:
+                await client.create_bucket(bucket_id, name=bucket_id, options={"public": False})
+                _logger.info("Created Supabase Storage bucket: %s", bucket_id)
+            except Exception as exc:
+                raise MediaStorageError(
+                    f"Failed to create storage bucket '{bucket_id}': {exc}", status_code=502
+                ) from exc
+
+        self._known_buckets.add(bucket_id)
+        return bucket_id
+
+    # ------------------------------------------------------------------
+    # Uploads & signed URLs
+    # ------------------------------------------------------------------
+
+    async def upload_file(self, bucket_id: str, path: str, data: bytes, content_type: str) -> None:
+        """Upload raw bytes to ``bucket_id`` under ``path``."""
+        client = await self._get_client()
+        try:
+            response = await client.from_(bucket_id).upload(
+                path,
+                data,
+                file_options={"content-type": content_type, "cache-control": "3600"},
+            )
+            if response is None or getattr(response, "error", None):
+                _logger.error("Supabase upload returned an error: %s", response)
+                raise MediaStorageError(
+                    "Supabase Storage upload failed (unknown response).", status_code=502
+                )
+        except MediaStorageError:
+            raise
+        except Exception as exc:
+            raise MediaStorageError(
+                f"Supabase Storage upload failed for '{path}': {exc}", status_code=502
+            ) from exc
+
+    async def create_signed_url(self, bucket_id: str, path: str, ttl_seconds: int) -> str:
+        """Create a signed download URL valid for ``ttl_seconds``."""
+        client = await self._get_client()
+        try:
+            result = await client.from_(bucket_id).create_signed_url(path, int(ttl_seconds))
+            signed = getattr(result, "signed_url", None) or (result.get("signedURL") if isinstance(result, dict) else None)
+            if not signed:
+                raise MediaStorageError(
+                    "Supabase returned an empty signed URL.", status_code=502
+                )
+            return signed
+        except MediaStorageError:
+            raise
+        except Exception as exc:
+            raise MediaStorageError(
+                f"Failed to create signed URL for '{path}': {exc}", status_code=502
+            ) from exc
+
+
+_storage_service: Optional[MediaStorageService] = None
+_storage_service_lock = asyncio.Lock()
+
+
+async def get_storage_service() -> MediaStorageService:
+    """Return the shared MediaStorageService singleton."""
+    global _storage_service
+    if _storage_service is None:
+        async with _storage_service_lock:
+            if _storage_service is None:
+                _storage_service = MediaStorageService()
+    return _storage_service
+
+
+async def close_storage_service() -> None:
+    global _storage_service
+    if _storage_service is not None:
+        await _storage_service.close()
+        _storage_service = None
+
+
+def iso_expiry(ttl_seconds: int) -> str:
+    """ISO-8601 UTC timestamp ``ttl_seconds`` from now."""
+    return datetime.fromtimestamp(time.time() + ttl_seconds, tz=timezone.utc).isoformat()