src/services/ingest_svc.py

"""
TradeFlow AI — Storage & Ingestion Service

PRD §1.6 — Handles file uploads, hashing, and storing to either Supabase or MinIO.
"""

import hashlib
import io
import mimetypes

import boto3
import structlog
from botocore.client import Config

from ..config import settings
from ..dependencies import get_supabase

log = structlog.get_logger()

class StorageService:
    def __init__(self):
        self.backend = settings.STORAGE_BACKEND
        self.bucket = settings.STORAGE_BUCKET_NAME
        self._bucket_initialized = False

        if self.backend == "minio":
            self.s3_client = boto3.client(
                "s3",
                endpoint_url=f"http://{settings.MINIO_ENDPOINT}",
                aws_access_key_id=settings.MINIO_ACCESS_KEY,
                aws_secret_access_key=settings.MINIO_SECRET_KEY.get_secret_value() if hasattr(settings.MINIO_SECRET_KEY, 'get_secret_value') else settings.MINIO_SECRET_KEY,
                config=Config(signature_version="s3v4", s3={'addressing_style': 'path'}),
                region_name="us-east-1",
            )
            self._ensure_minio_bucket()
        else:
            # For supabase storage, we'll use the async supabase client provided by dependencies
            self.s3_client = None

    def _ensure_minio_bucket(self) -> None:
        try:
            self.s3_client.head_bucket(Bucket=self.bucket)
        except Exception:
            try:
                self.s3_client.create_bucket(Bucket=self.bucket)
                # MinIO buckets are private by default. Avoid overly restrictive IP-based
                # policies because the API container does not originate from localhost.
                log.info("Created MinIO bucket", bucket=self.bucket)
            except Exception as e:
                log.error("Failed to create bucket", error=str(e))

    async def upload_document(
        self, batch_id: str, doc_id: str, filename: str, file_bytes: bytes
    ) -> str:
        """Upload a document to the configured storage backend and return the path."""
        content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
        object_path = f"documents/{batch_id}/{doc_id}/{filename}"

        if self.backend == "minio":
            self.s3_client.upload_fileobj(
                io.BytesIO(file_bytes),
                self.bucket,
                object_path,
                ExtraArgs={"ContentType": content_type}
            )
            log.info("Uploaded to MinIO", path=object_path)
            return object_path

        elif self.backend == "supabase":
            supabase = get_supabase()
            
            # Skip bucket creation check because the bucket already exists
            # and the storage3 client may have buggy create_bucket payload formatting.

            await supabase.storage.from_(self.bucket).upload(
                object_path,
                file_bytes,
                {"content-type": content_type}
            )

            log.info("Uploaded to Supabase Storage", path=object_path)
            return object_path

        raise ValueError(f"Unknown storage backend: {self.backend}")

    async def download_document(self, object_path: str) -> bytes:
        """Download a document from the configured storage backend."""
        if self.backend == "minio":
            import asyncio

            def _download() -> bytes:
                response = self.s3_client.get_object(Bucket=self.bucket, Key=object_path)
                return response["Body"].read()

            return await asyncio.to_thread(_download)

        if self.backend == "supabase":
            supabase = get_supabase()
            data = await supabase.storage.from_(self.bucket).download(object_path)
            return bytes(data)

        raise ValueError(f"Unknown storage backend: {self.backend}")

    def compute_hash(self, file_bytes: bytes) -> str:
        """Compute SHA-256 hash of the file."""
        return hashlib.sha256(file_bytes).hexdigest()

    async def get_presigned_url(self, object_path: str, expires_in: int = 3600) -> str:
        """Generate a presigned URL for preview/download."""
        if self.backend == "minio":
            return self.s3_client.generate_presigned_url(
                "get_object",
                Params={"Bucket": self.bucket, "Key": object_path},
                ExpiresIn=expires_in,
            )
        elif self.backend == "supabase":
            # For Supabase, we can use create_signed_url
            supabase = get_supabase()
            res = await supabase.storage.from_(self.bucket).create_signed_url(object_path, expires_in)
            return res.get("signedURL", "")
        return ""

storage_service: StorageService | None = None


def get_storage_service() -> StorageService:
    global storage_service
    if storage_service is None:
        storage_service = StorageService()
    return storage_service