-- Raw data records: original uploaded files stored as-is (provenance),
-- org + project scoped. Files live in the dedicated `raw-data` bucket at
--   {organization_id}/{raw_data_id}/{filename}
-- Many-to-many provenance to cell_measurements via measurement_raw_data_sources.

CREATE TABLE IF NOT EXISTS "public"."raw_data" (
    "id"               "uuid"                   DEFAULT "gen_random_uuid"() NOT NULL,
    "organization_id"  "uuid"                   NOT NULL,
    "project_id"       "uuid"                   NOT NULL,
    "name"             "text"                   NOT NULL,
    "filename"         "text"                   NOT NULL,
    "content_type"     "text",
    "size_bytes"       bigint,
    "storage_path"     "text"                   NOT NULL,
    "metadata"         "jsonb"                  NOT NULL DEFAULT '{}'::"jsonb",
    "created_by"       "uuid",
    "created_at"       timestamp with time zone DEFAULT "now"() NOT NULL,
    "updated_at"       timestamp with time zone DEFAULT "now"() NOT NULL
);
ALTER TABLE "public"."raw_data" OWNER TO "postgres";

COMMENT ON TABLE "public"."raw_data" IS
    'Original uploaded raw files stored as-is (provenance), org + project '
    'scoped. Files live in the raw-data bucket at '
    '{organization_id}/{raw_data_id}/{filename}. Linked many-to-many to '
    'cell_measurements via measurement_raw_data_sources.';

-- PK (guarded)
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint" c
        JOIN "pg_class" t ON c."conrelid" = t."oid"
        JOIN "pg_namespace" n ON t."relnamespace" = n."oid"
        WHERE c."conname" = 'raw_data_pkey' AND n."nspname" = 'public'
          AND t."relname" = 'raw_data'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_pkey" PRIMARY KEY ("id");
    END IF;
END $$;

-- FK → organizations
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint" WHERE "conname" = 'raw_data_organization_id_fkey'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_organization_id_fkey"
            FOREIGN KEY ("organization_id")
            REFERENCES "public"."organizations"("id") ON DELETE CASCADE;
    END IF;
END $$;

-- FK → projects (composite: organization_id cannot drift from owning project).
-- Split ADD ... NOT VALID + VALIDATE, matching the sanctioned pattern in
-- 20260504100000_add_material_property_datasets.sql:100-123. Targets the
-- projects_id_organization_id_key UNIQUE(id, organization_id) added in
-- 20260501000005_add_composite_project_org_fkeys.sql.
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint"
        WHERE "conname" = 'raw_data_project_id_organization_id_fkey'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_project_id_organization_id_fkey"
            FOREIGN KEY ("project_id", "organization_id")
            REFERENCES "public"."projects"("id", "organization_id")
            ON DELETE CASCADE NOT VALID;
    END IF;
END $$;
DO $$ BEGIN
    IF EXISTS (
        SELECT 1 FROM "pg_constraint"
        WHERE "conname" = 'raw_data_project_id_organization_id_fkey'
          AND NOT "convalidated"
    ) THEN
        ALTER TABLE "public"."raw_data"
            VALIDATE CONSTRAINT "raw_data_project_id_organization_id_fkey";
    END IF;
END $$;

-- FK → users (nullable)
DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM "pg_constraint" WHERE "conname" = 'raw_data_created_by_fkey'
    ) THEN
        ALTER TABLE ONLY "public"."raw_data"
            ADD CONSTRAINT "raw_data_created_by_fkey"
            FOREIGN KEY ("created_by")
            REFERENCES "public"."users"("id") ON DELETE SET NULL;
    END IF;
END $$;

CREATE INDEX IF NOT EXISTS "idx_raw_data_project_id_created_at"
    ON "public"."raw_data" USING "btree" ("project_id", "created_at" DESC);
CREATE INDEX IF NOT EXISTS "idx_raw_data_organization_id"
    ON "public"."raw_data" USING "btree" ("organization_id");

CREATE OR REPLACE TRIGGER "handle_raw_data_updated_at"
    BEFORE UPDATE ON "public"."raw_data"
    FOR EACH ROW EXECUTE FUNCTION "extensions"."moddatetime"('updated_at');

ALTER TABLE "public"."raw_data" ENABLE ROW LEVEL SECURITY;

-- Table RLS (reuse cell_measurement:* verb set, like analyses)
DROP POLICY IF EXISTS "Users can create raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can create raw_data in their org" ON "public"."raw_data"
    FOR INSERT TO "authenticated"
    WITH CHECK ("public"."has_permission_in_org"("organization_id", 'cell_measurement:create'::"text"));
DROP POLICY IF EXISTS "Users can read raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can read raw_data in their org" ON "public"."raw_data"
    FOR SELECT TO "authenticated"
    USING ("public"."has_permission_in_org"("organization_id", 'cell_measurement:read'::"text"));
DROP POLICY IF EXISTS "Users can update raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can update raw_data in their org" ON "public"."raw_data"
    FOR UPDATE TO "authenticated"
    USING ("public"."has_permission_in_org"("organization_id", 'cell_measurement:update'::"text"))
    WITH CHECK ("public"."has_permission_in_org"("organization_id", 'cell_measurement:update'::"text"));
DROP POLICY IF EXISTS "Users can delete raw_data in their org" ON "public"."raw_data";
CREATE POLICY "Users can delete raw_data in their org" ON "public"."raw_data"
    FOR DELETE TO "authenticated"
    USING ("public"."has_permission_in_org"("organization_id", 'cell_measurement:delete'::"text"));

GRANT ALL ON TABLE "public"."raw_data" TO "anon";
GRANT ALL ON TABLE "public"."raw_data" TO "authenticated";
GRANT ALL ON TABLE "public"."raw_data" TO "service_role";

-- ---------------------------------------------------------------------------
-- Join table: measurement_raw_data_sources (many-to-many provenance)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS "public"."measurement_raw_data_sources" (
    "id"                   "uuid"                   DEFAULT "gen_random_uuid"() NOT NULL,
    "cell_measurement_id"  "uuid"                   NOT NULL,
    "raw_data_id"          "uuid"                   NOT NULL,
    "created_at"           timestamp with time zone DEFAULT "now"() NOT NULL
);
ALTER TABLE "public"."measurement_raw_data_sources" OWNER TO "postgres";

DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_pkey') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_pkey" PRIMARY KEY ("id");
    END IF;
END $$;
DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_measurement_fkey') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_measurement_fkey"
            FOREIGN KEY ("cell_measurement_id")
            REFERENCES "public"."cell_measurements"("id") ON DELETE CASCADE;
    END IF;
END $$;
DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_raw_data_fkey') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_raw_data_fkey"
            FOREIGN KEY ("raw_data_id")
            REFERENCES "public"."raw_data"("id") ON DELETE CASCADE;
    END IF;
END $$;
DO $$ BEGIN
    IF NOT EXISTS (SELECT 1 FROM "pg_constraint" WHERE "conname" = 'measurement_raw_data_sources_unique_pair') THEN
        ALTER TABLE ONLY "public"."measurement_raw_data_sources"
            ADD CONSTRAINT "measurement_raw_data_sources_unique_pair"
            UNIQUE ("cell_measurement_id", "raw_data_id");
    END IF;
END $$;

CREATE INDEX IF NOT EXISTS "idx_mrds_measurement_id"
    ON "public"."measurement_raw_data_sources" USING "btree" ("cell_measurement_id");
CREATE INDEX IF NOT EXISTS "idx_mrds_raw_data_id"
    ON "public"."measurement_raw_data_sources" USING "btree" ("raw_data_id");

ALTER TABLE "public"."measurement_raw_data_sources" ENABLE ROW LEVEL SECURITY;

-- Join-table RLS: gate via the linked raw_data row's org (which is what a user
-- must already be able to read). A row is visible/insertable/deletable iff the
-- caller has the matching cell_measurement:* permission on the raw_data's org.
DROP POLICY IF EXISTS "Users can read mrds in their org" ON "public"."measurement_raw_data_sources";
CREATE POLICY "Users can read mrds in their org" ON "public"."measurement_raw_data_sources"
    FOR SELECT TO "authenticated"
    USING (EXISTS (
        SELECT 1 FROM "public"."raw_data" rd
        WHERE rd."id" = "raw_data_id"
          AND "public"."has_permission_in_org"(rd."organization_id", 'cell_measurement:read'::"text")
    ));
DROP POLICY IF EXISTS "Users can create mrds in their org" ON "public"."measurement_raw_data_sources";
CREATE POLICY "Users can create mrds in their org" ON "public"."measurement_raw_data_sources"
    FOR INSERT TO "authenticated"
    WITH CHECK (EXISTS (
        SELECT 1 FROM "public"."raw_data" rd
        WHERE rd."id" = "raw_data_id"
          AND "public"."has_permission_in_org"(rd."organization_id", 'cell_measurement:update'::"text")
    ));
DROP POLICY IF EXISTS "Users can delete mrds in their org" ON "public"."measurement_raw_data_sources";
CREATE POLICY "Users can delete mrds in their org" ON "public"."measurement_raw_data_sources"
    FOR DELETE TO "authenticated"
    USING (EXISTS (
        SELECT 1 FROM "public"."raw_data" rd
        WHERE rd."id" = "raw_data_id"
          AND "public"."has_permission_in_org"(rd."organization_id", 'cell_measurement:update'::"text")
    ));

GRANT ALL ON TABLE "public"."measurement_raw_data_sources" TO "anon";
GRANT ALL ON TABLE "public"."measurement_raw_data_sources" TO "authenticated";
GRANT ALL ON TABLE "public"."measurement_raw_data_sources" TO "service_role";

INSERT INTO storage.buckets (id, name, public, file_size_limit, allowed_mime_types)
VALUES ('raw-data', 'raw-data', false, null, null)
ON CONFLICT (id) DO NOTHING;

-- Storage bucket RLS for raw-data. Path: {organization_id}/{raw_data_id}/{filename}
DROP POLICY IF EXISTS "Allow read storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow read storage for raw_data with permission" ON "storage"."objects"
    FOR SELECT USING (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:read'));
DROP POLICY IF EXISTS "Allow insert storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow insert storage for raw_data with permission" ON "storage"."objects"
    FOR INSERT WITH CHECK (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:create'));
DROP POLICY IF EXISTS "Allow update storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow update storage for raw_data with permission" ON "storage"."objects"
    FOR UPDATE USING (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:update'));
DROP POLICY IF EXISTS "Allow delete storage for raw_data with permission" ON "storage"."objects";
CREATE POLICY "Allow delete storage for raw_data with permission" ON "storage"."objects"
    FOR DELETE USING (
        "bucket_id" = 'raw-data' AND "public"."has_permission_in_org"(
            "public"."resolve_org_id_from_path_token"(("storage"."foldername"("name"))[1]), 'storage:delete'));

git add supabase/migrations/*_add_raw_data.sql supabase/migrations/*_create_raw_data_bucket.sql supabase/seeds/buckets.sql
git commit -m "feat(raw-data): add raw_data table, join table, bucket + RLS"

"""Pydantic models for the raw_data entity.

A raw_data record is an original uploaded file stored as-is (provenance),
owned by an organization and a project. Files live in the ``raw-data`` bucket
at ``{organization_id}/{raw_data_id}/{filename}``. Raw-data records link
many-to-many to cell measurements via ``measurement_raw_data_sources``.
"""

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class RawDataBase(BaseModel):
    """Fields shared by create/read representations of a raw-data record."""

    project_id: str = Field(..., description="Owning project ID (required).")
    name: str = Field(..., description="User-provided label for the raw file.")
    metadata: dict = Field(
        default_factory=dict, description="Loose free-form metadata."
    )


class RawData(RawDataBase):
    """A raw-data record as returned by the API."""

    id: str = Field(..., description="Unique identifier.")
    organization_id: str = Field(..., description="Owning organization.")
    filename: str = Field(..., description="Original uploaded filename.")
    content_type: str | None = Field(None, description="MIME type, if known.")
    size_bytes: int | None = Field(None, description="File size in bytes.")
    storage_path: str = Field(..., description="Path within the raw-data bucket.")
    created_by: str | None = Field(None, description="User ID of the uploader.")
    created_at: datetime = Field(..., description="Creation timestamp.")
    updated_at: datetime = Field(..., description="Last update timestamp.")

    model_config = ConfigDict(from_attributes=True)


class UpdateRawData(BaseModel):
    """Partial update for a raw-data record. Does not replace the file."""

    name: str | None = Field(None, description="New label.")
    metadata: dict | None = Field(None, description="Replacement metadata dict.")


class AttachRawData(BaseModel):
    """Body for bulk-attaching raw-data records to a measurement."""

    raw_data_ids: list[str] = Field(
        ..., min_length=1, description="Raw-data IDs to attach to the measurement."
    )

git add backend/src/pydantic_models/raw_data.py
git commit -m "feat(raw-data): add pydantic models"

from typing import Annotated

from fastapi import Depends
from supabase import AsyncClient

from src.deps import get_supabase_user_client
from src.exceptions import AppError
from src.pydantic_models.pagination import PaginatedList
from src.pydantic_models.raw_data import RawData
from src.repositories.base import BaseRepository

_JOIN_TABLE = "measurement_raw_data_sources"


class RawDataRepository(BaseRepository[RawData]):
    """Repository for raw-data records and their measurement links."""

    def __init__(self, supabase: AsyncClient):
        super().__init__(supabase, "raw_data", RawData)

    async def list_by_project(
        self,
        project_id: str,
        organization_id: str,
        limit: int = 100,
        offset: int = 0,
    ) -> PaginatedList[RawData]:
        """List raw-data records for a project, newest first.

        Parameters
        ----------
        project_id : str
            Project whose raw-data records to list.
        organization_id : str
            Organization ID used as an additional RLS guard.
        limit : int, optional
            Maximum records to return. Defaults to 100.
        offset : int, optional
            Records to skip. Defaults to 0.

        Returns
        -------
        PaginatedList[RawData]
            Paginated list with count and total.
        """
        response = await (
            self.supabase.table(self.table_name)
            .select("*", count="exact")
            .eq("project_id", project_id)
            .eq("organization_id", organization_id)
            .order("created_at", desc=True)
            .range(offset, offset + limit - 1)
            .execute()
        )
        items = [RawData(**row) for row in (response.data or [])]
        return self._wrap_paginated(items, response)

    async def get_raw_data(self, raw_data_id: str) -> RawData | None:
        """Fetch a single raw-data record by ID."""
        return await self.get({"id": raw_data_id})

    async def create_raw_data(
        self, organization_id: str, created_by: str | None, data: dict
    ) -> RawData:
        """Insert a raw-data row.

        Parameters
        ----------
        organization_id : str
            Owning organization.
        created_by : str | None
            Uploader user ID, may be None for service-account creation.
        data : dict
            Fields: project_id, name, filename, content_type, size_bytes,
            storage_path, metadata.

        Returns
        -------
        RawData
            The created record.

        Raises
        ------
        AppError
            If the insert fails.
        """
        payload = {
            "organization_id": organization_id,
            "created_by": created_by,
            **data,
        }
        result = await self.create(payload)
        if not result:
            raise AppError("Failed to create raw_data record")
        return result

    async def delete_raw_data(self, raw_data_id: str) -> bool:
        """Delete a raw-data row by ID."""
        return await self.delete(raw_data_id)

    # --- join table (measurement_raw_data_sources) ---

    async def attach_measurement_sources(
        self, cell_measurement_id: str, raw_data_ids: list[str]
    ) -> None:
        """Idempotently link a measurement to raw-data records (bulk).

        Uses an upsert that ignores duplicate ``(cell_measurement_id,
        raw_data_id)`` pairs, so re-attaching already-linked ids is a no-op.

        Parameters
        ----------
        cell_measurement_id : str
            The measurement to link.
        raw_data_ids : list[str]
            Raw-data IDs to link. A no-op when empty.
        """
        if not raw_data_ids:
            return
        rows = [
            {"cell_measurement_id": cell_measurement_id, "raw_data_id": rid}
            for rid in raw_data_ids
        ]
        await (
            self.supabase.table(_JOIN_TABLE)
            .upsert(rows, ignore_duplicates=True, on_conflict="cell_measurement_id,raw_data_id")
            .execute()
        )

    async def detach_measurement_source(
        self, cell_measurement_id: str, raw_data_id: str
    ) -> None:
        """Remove one measurement↔raw-data link."""
        await (
            self.supabase.table(_JOIN_TABLE)
            .delete()
            .eq("cell_measurement_id", cell_measurement_id)
            .eq("raw_data_id", raw_data_id)
            .execute()
        )

    async def list_sources_for_measurement(
        self, cell_measurement_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[RawData]:
        """List raw-data records linked to a measurement, via the join table."""
        response = await (
            self.supabase.table(_JOIN_TABLE)
            .select("raw_data!inner(*)", count="exact")
            .eq("cell_measurement_id", cell_measurement_id)
            .range(offset, offset + limit - 1)
            .execute()
        )
        items = [RawData(**row["raw_data"]) for row in (response.data or [])]
        return self._wrap_paginated(items, response)

    async def list_measurement_ids_for_raw_data(
        self, raw_data_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[str]:
        """List measurement IDs linked to a raw-data record."""
        response = await (
            self.supabase.table(_JOIN_TABLE)
            .select("cell_measurement_id", count="exact")
            .eq("raw_data_id", raw_data_id)
            .range(offset, offset + limit - 1)
            .execute()
        )
        items = [row["cell_measurement_id"] for row in (response.data or [])]
        return self._wrap_paginated(items, response)


def get_raw_data_repository(
    supabase: Annotated[AsyncClient, Depends(get_supabase_user_client)],
) -> RawDataRepository:
    """FastAPI dependency that returns a RawDataRepository."""
    return RawDataRepository(supabase)

git add backend/src/repositories/raw_data.py backend/tests/test_repositories/test_raw_data_repository.py
git commit -m "feat(raw-data): add DB repository with join-table methods"

import logging
from typing import Annotated
from urllib.parse import urlparse, urlunparse

from fastapi import Depends
from supabase import AsyncClient

from src.deps import get_supabase_user_client
from src.exceptions import ExternalServiceError
from src.settings import settings
from src.utils.storage_utils import async_upload_file_to_storage

logger = logging.getLogger(__name__)

BUCKET_NAME = "raw-data"


class RawDataBucketRepository:
    """Storage repository for raw-data files.

    Each record stores a single original file at
    ``{organization_id}/{raw_data_id}/{filename}`` in the ``raw-data`` bucket.
    """

    bucket_name = BUCKET_NAME

    def __init__(self, supabase: AsyncClient) -> None:
        self.supabase = supabase
        self.bucket = supabase.storage.from_(self.bucket_name)

    def build_path(self, organization_id: str, raw_data_id: str, filename: str) -> str:
        """Return the storage path for a raw-data file.

        Parameters
        ----------
        organization_id : str
            Organization UUID (first path token, used by RLS).
        raw_data_id : str
            Raw-data record UUID.
        filename : str
            Original uploaded filename.

        Returns
        -------
        str
            Path relative to the bucket root.
        """
        return f"{organization_id}/{raw_data_id}/{filename}"

    async def upload(
        self,
        path: str,
        file_content: bytes,
        content_type: str | None,
    ) -> dict:
        """Upload a raw-data file (no overwrite).

        Raises
        ------
        ExternalServiceError
            If the upload fails.
        """
        options = {"upsert": "false"}
        if content_type:
            options["content_type"] = content_type
        try:
            return await async_upload_file_to_storage(
                supabase=self.supabase,
                filepath=path,
                file_content=file_content,
                bucket_name=self.bucket_name,
                file_options=options,
            )
        except Exception as e:
            raise ExternalServiceError(
                self.bucket_name, f"Failed to upload raw-data file at {path}: {e}"
            ) from e

    async def create_signed_download_url(
        self,
        path: str,
        expires_in: int = 300,
        download_filename: str | None = None,
    ) -> str:
        """Create a short-lived signed download URL. Copied from analysis_bucket."""
        options = {"download": download_filename} if download_filename else {}
        try:
            result = await self.bucket.create_signed_url(
                path=path, expires_in=expires_in, options=options or None
            )
            url = result.get("signedURL") or result.get("signed_url")
            if not url:
                raise ExternalServiceError(
                    self.bucket_name, "Signed URL not returned by storage API"
                )
            if settings.STORAGE_PUBLIC_URL:
                parsed = urlparse(url)
                public = urlparse(settings.STORAGE_PUBLIC_URL)
                url = urlunparse(
                    parsed._replace(scheme=public.scheme, netloc=public.netloc)
                )
            return url
        except ExternalServiceError:
            raise
        except Exception as e:
            raise ExternalServiceError(
                self.bucket_name, f"Failed to create signed URL for {path}: {e}"
            ) from e

    async def delete_at_path(self, path: str) -> None:
        """Delete a single file at the given bucket path (best-effort)."""
        try:
            await self.bucket.remove(paths=[path])
        except Exception as e:
            logger.warning(
                "Failed to delete raw-data file at %s; may be orphaned: %s", path, e
            )


def get_raw_data_bucket_repository(
    supabase: Annotated[AsyncClient, Depends(get_supabase_user_client)],
) -> RawDataBucketRepository:
    """FastAPI dependency that returns a RawDataBucketRepository."""
    return RawDataBucketRepository(supabase)

git add backend/src/repositories/raw_data_bucket.py
git commit -m "feat(raw-data): add storage bucket repository"

import logging
from typing import Annotated

from fastapi import Depends

from src.exceptions import BadRequestError, NotFoundError
from src.pydantic_models.pagination import PaginatedList
from src.pydantic_models.raw_data import RawData
from src.repositories.cell_measurements import (
    CellMeasurementRepository,
    get_cell_measurement_repository,
)
from src.repositories.raw_data import RawDataRepository, get_raw_data_repository
from src.repositories.raw_data_bucket import (
    RawDataBucketRepository,
    get_raw_data_bucket_repository,
)

logger = logging.getLogger(__name__)


class RawDataService:
    """Business logic for raw-data records and measurement provenance links.

    Ordering invariants (same as AnalysisService):

    - **create**: DB row first, then file upload. On upload failure the row is
      deleted and the error re-raised — no orphan row without its file.
    - **delete**: DB row removed first, then best-effort file delete. A storage
      failure orphans a file (recoverable) rather than leaving a visible record
      with inaccessible data.

    Deleting a *measurement* cascades away join rows but never touches raw_data
    rows or files (a raw file may back other measurements).
    """

    resource_type = "raw_data"

    def __init__(
        self,
        raw_data_repo: RawDataRepository,
        bucket_repo: RawDataBucketRepository,
        measurement_repo: CellMeasurementRepository,
    ) -> None:
        self.repo = raw_data_repo
        self.bucket_repo = bucket_repo
        self.measurement_repo = measurement_repo

    async def create(
        self,
        *,
        organization_id: str,
        user_id: str,
        project_id: str,
        name: str,
        filename: str,
        content_type: str | None,
        size_bytes: int | None,
        metadata: dict,
        file_content: bytes,
    ) -> RawData:
        """Create a raw-data record and upload its file (row first, rollback on
        upload failure)."""
        # Insert row without storage_path, then patch it in once we know the id.
        record = await self.repo.create_raw_data(
            organization_id=organization_id,
            created_by=user_id,
            data={
                "project_id": project_id,
                "name": name,
                "filename": filename,
                "content_type": content_type,
                "size_bytes": size_bytes,
                "metadata": metadata,
                "storage_path": "",  # set below
            },
        )
        path = self.bucket_repo.build_path(organization_id, record.id, filename)
        try:
            await self.bucket_repo.upload(path, file_content, content_type)
            updated = await self.repo.update({"id": record.id}, {"storage_path": path})
        except Exception:
            logger.exception("Rolling back raw_data %s after upload failure", record.id)
            try:
                await self.repo.delete_raw_data(record.id)
            except Exception:
                logger.exception("Failed to roll back raw_data row %s", record.id)
            raise
        return updated or record

    async def get(self, raw_data_id: str, organization_id: str) -> RawData:
        """Fetch one record, guarding org ownership."""
        record = await self.repo.get_raw_data(raw_data_id)
        if not record or record.organization_id != organization_id:
            raise NotFoundError(self.resource_type, raw_data_id)
        return record

    async def get_download_url(
        self, raw_data_id: str, organization_id: str, expires_in: int = 300
    ) -> str:
        """Return a signed download URL for a record's file."""
        record = await self.get(raw_data_id, organization_id)
        return await self.bucket_repo.create_signed_download_url(
            record.storage_path,
            expires_in=expires_in,
            download_filename=record.filename,
        )

    async def list_by_project(
        self, project_id: str, organization_id: str, limit: int, offset: int
    ) -> PaginatedList[RawData]:
        """List a project's raw-data records."""
        return await self.repo.list_by_project(
            project_id=project_id,
            organization_id=organization_id,
            limit=limit,
            offset=offset,
        )

    async def update(
        self,
        raw_data_id: str,
        organization_id: str,
        name: str | None,
        metadata: dict | None,
    ) -> RawData:
        """Partial-update name/metadata. Does not replace the file."""
        record = await self.get(raw_data_id, organization_id)
        new_data: dict = {}
        if name is not None:
            new_data["name"] = name
        if metadata is not None:
            new_data["metadata"] = metadata
        if not new_data:
            return record
        updated = await self.repo.update({"id": raw_data_id}, new_data)
        if updated is None:
            raise NotFoundError(self.resource_type, raw_data_id)
        return updated

    async def delete(self, raw_data_id: str, organization_id: str) -> None:
        """Delete a record then best-effort its file."""
        record = await self.get(raw_data_id, organization_id)
        await self.repo.delete_raw_data(raw_data_id)
        try:
            await self.bucket_repo.delete_at_path(record.storage_path)
        except Exception:
            logger.warning(
                "Storage delete failed for raw_data %s; file orphaned", raw_data_id
            )

    # --- provenance (measurement-centric) ---

    async def attach_to_measurement(
        self,
        cell_measurement_id: str,
        raw_data_ids: list[str],
        organization_id: str,
    ) -> PaginatedList[RawData]:
        """Bulk-attach raw-data records to a measurement (idempotent).

        Raises
        ------
        NotFoundError
            If the measurement is missing or not in the caller's org.
        BadRequestError
            If any raw_data_id is missing or in another org.
        """
        measurement = await self.measurement_repo.get_by_id(cell_measurement_id)
        if measurement is None or measurement.organization_id != organization_id:
            raise NotFoundError("measurement", cell_measurement_id)
        for rid in raw_data_ids:
            rec = await self.repo.get_raw_data(rid)
            if rec is None or rec.organization_id != organization_id:
                raise BadRequestError(
                    f"raw_data '{rid}' does not exist in this organization."
                )
        await self.repo.attach_measurement_sources(cell_measurement_id, raw_data_ids)
        return await self.repo.list_sources_for_measurement(cell_measurement_id)

    async def detach_from_measurement(
        self, cell_measurement_id: str, raw_data_id: str, organization_id: str
    ) -> None:
        """Remove one measurement↔raw-data link (org-guarded via measurement)."""
        measurement = await self.measurement_repo.get_by_id(cell_measurement_id)
        if measurement is None or measurement.organization_id != organization_id:
            raise NotFoundError("measurement", cell_measurement_id)
        await self.repo.detach_measurement_source(cell_measurement_id, raw_data_id)

    async def list_sources_for_measurement(
        self, cell_measurement_id: str, organization_id: str, limit: int, offset: int
    ) -> PaginatedList[RawData]:
        """List a measurement's raw-data sources (org-guarded via measurement)."""
        measurement = await self.measurement_repo.get_by_id(cell_measurement_id)
        if measurement is None or measurement.organization_id != organization_id:
            raise NotFoundError("measurement", cell_measurement_id)
        return await self.repo.list_sources_for_measurement(
            cell_measurement_id, limit=limit, offset=offset
        )


def get_raw_data_service(
    raw_data_repo: Annotated[RawDataRepository, Depends(get_raw_data_repository)],
    bucket_repo: Annotated[
        RawDataBucketRepository, Depends(get_raw_data_bucket_repository)
    ],
    measurement_repo: Annotated[
        CellMeasurementRepository, Depends(get_cell_measurement_repository)
    ],
) -> RawDataService:
    """FastAPI dependency constructing a RawDataService."""
    return RawDataService(
        raw_data_repo=raw_data_repo,
        bucket_repo=bucket_repo,
        measurement_repo=measurement_repo,
    )

git add backend/src/services/raw_data_service.py backend/tests/test_services/test_raw_data_service.py
git commit -m "feat(raw-data): add service with create/delete invariants and provenance attach"

import json
import logging
from typing import Annotated

from fastapi import APIRouter, Depends, Form, Query, UploadFile, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

from src.deps import get_current_organization_id, get_current_user_id
from src.exceptions import BadRequestError
from src.pydantic_models.pagination import PaginatedResponse
from src.pydantic_models.raw_data import RawData, UpdateRawData
from src.services.raw_data_service import RawDataService, get_raw_data_service

logger = logging.getLogger(__name__)

router = APIRouter(tags=["Raw Data"])


def _parse_metadata(metadata_json: str) -> dict:
    """Parse a JSON-encoded metadata object (raises BadRequestError on bad input)."""
    try:
        metadata = json.loads(metadata_json)
    except (json.JSONDecodeError, TypeError, ValueError) as exc:
        raise BadRequestError(f"metadata must be a valid JSON object: {exc}") from exc
    if not isinstance(metadata, dict):
        raise BadRequestError("metadata must be a JSON object.")
    return metadata


@router.post("", response_model=RawData, status_code=status.HTTP_201_CREATED,
             summary="Upload a raw-data file")
async def create_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    user_id: Annotated[str, Depends(get_current_user_id)],
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    file: UploadFile,
    project_id: str = Form(...),
    name: str = Form(...),
    metadata: str = Form("{}"),
) -> JSONResponse:
    """Upload a raw file, stored as-is in the raw-data bucket under the project."""
    metadata_dict = _parse_metadata(metadata)
    raw = await file.read()
    record = await service.create(
        organization_id=organization_id,
        user_id=user_id,
        project_id=project_id,
        name=name,
        filename=file.filename or "upload.bin",
        content_type=file.content_type,
        size_bytes=len(raw),
        metadata=metadata_dict,
        file_content=raw,
    )
    return JSONResponse(
        content=record.model_dump(mode="json"),
        status_code=status.HTTP_201_CREATED,
    )


@router.get("", response_model=PaginatedResponse[RawData],
            summary="List raw-data records for a project")
async def list_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    project_id: str = Query(..., description="Filter by owning project ID"),
    limit: int = Query(100, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> JSONResponse:
    """List a project's raw-data records, newest first."""
    result = await service.list_by_project(
        project_id=project_id, organization_id=organization_id,
        limit=limit, offset=offset,
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))


@router.get("/{raw_data_id}", response_model=RawData, summary="Get a raw-data record")
async def get_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Retrieve a single raw-data record by ID."""
    record = await service.get(raw_data_id, organization_id)
    return JSONResponse(content=record.model_dump(mode="json"))


@router.get("/{raw_data_id}/download-url",
            summary="Get a signed download URL for a raw-data file")
async def get_raw_data_download_url(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Return ``{"url": "..."}`` — a signed URL valid for 5 minutes."""
    url = await service.get_download_url(raw_data_id, organization_id, expires_in=300)
    return JSONResponse(content={"url": url})


@router.patch("/{raw_data_id}", response_model=RawData,
              summary="Update a raw-data record's metadata")
async def update_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    body: UpdateRawData,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Update name/metadata. Does not replace the file."""
    updated = await service.update(
        raw_data_id=raw_data_id, organization_id=organization_id,
        name=body.name, metadata=body.metadata,
    )
    return JSONResponse(content=updated.model_dump(mode="json"))


@router.delete("/{raw_data_id}", status_code=status.HTTP_204_NO_CONTENT,
               summary="Delete a raw-data record")
async def delete_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> None:
    """Delete a raw-data record and its file."""
    await service.delete(raw_data_id, organization_id)


@router.get("/{raw_data_id}/cell_measurements",
            response_model=PaginatedResponse[str],
            summary="List measurement IDs produced from a raw-data record")
async def list_raw_data_measurements(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    limit: int = Query(100, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> JSONResponse:
    """Reverse provenance read: measurements linked to this raw-data record."""
    await service.get(raw_data_id, organization_id)  # org guard / 404
    result = await service.repo.list_measurement_ids_for_raw_data(
        raw_data_id, limit=limit, offset=offset
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))

from src.routes.raw_data import router as raw_data_router
...
app.include_router(raw_data_router, prefix="/raw_data")

git add backend/src/routes/raw_data.py backend/src/main.py backend/tests/test_routes/test_raw_data_routes.py
git commit -m "feat(raw-data): add /raw_data CRUD routes and mount router"

from src.pydantic_models.pagination import PaginatedResponse
from src.pydantic_models.raw_data import AttachRawData, RawData
from src.services.raw_data_service import RawDataService, get_raw_data_service

from src.pydantic_models.raw_data import AttachRawData, RawData
from src.services.raw_data_service import RawDataService, get_raw_data_service

@direct_router.post("/{measurement_id}/raw_data",
                    response_model=PaginatedResponse[RawData],
                    summary="Attach raw-data records to a measurement")
async def attach_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    measurement_id: str,
    body: AttachRawData,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> JSONResponse:
    """Bulk-attach raw-data records to a measurement (idempotent)."""
    result = await service.attach_to_measurement(
        cell_measurement_id=measurement_id,
        raw_data_ids=body.raw_data_ids,
        organization_id=organization_id,
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))

@direct_router.delete("/{measurement_id}/raw_data/{raw_data_id}",
                      status_code=status.HTTP_204_NO_CONTENT,
                      summary="Detach a raw-data record from a measurement")
async def detach_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    measurement_id: str,
    raw_data_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
) -> None:
    """Remove one measurement↔raw-data link."""
    await service.detach_from_measurement(measurement_id, raw_data_id, organization_id)

@direct_router.get("/{measurement_id}/raw_data",
                   response_model=PaginatedResponse[RawData],
                   summary="List a measurement's raw-data sources")
async def list_measurement_raw_data(
    organization_id: Annotated[str, Depends(get_current_organization_id)],
    measurement_id: str,
    service: Annotated[RawDataService, Depends(get_raw_data_service)],
    limit: int = Query(100, ge=1, le=100),
    offset: int = Query(0, ge=0),
) -> JSONResponse:
    """List raw-data records linked to a measurement."""
    result = await service.list_sources_for_measurement(
        measurement_id, organization_id, limit=limit, offset=offset
    )
    return JSONResponse(content=jsonable_encoder(result.to_response_dict()))

git add backend/src/routes/cell_measurements.py backend/tests/test_routes/test_measurement_raw_data_routes.py
git commit -m "feat(raw-data): measurement-centric attach/detach/list provenance routes"

class RawData(BaseModel):
    """A raw-data record. The API defines the full schema; extra fields allowed."""

    model_config = ConfigDict(extra="allow")
    id: str
    project_id: str
    name: str
    filename: str

"""Client for raw-data records (original uploaded files, org + project scoped)."""

import json
import os
from typing import IO, Any

from .models import PaginatedList, RawData, _build_endpoint, _parse_list_response


class RawDataClient:
    """Upload and manage raw-data files and their measurement links."""

    _BASE = "/raw_data"

    def __init__(self, client: Any) -> None:
        self.client = client

    def upload(
        self,
        project_id: str,
        file: str | os.PathLike | IO[bytes],
        name: str | None = None,
        metadata: dict | None = None,
    ) -> RawData:
        """Upload a raw file, stored as-is under the given project.

        Parameters
        ----------
        project_id : str
            Owning project ID.
        file : str | os.PathLike | IO[bytes]
            Path to a file, or an open binary file-like object.
        name : str | None, optional
            Label for the record. Defaults to the file's basename.
        metadata : dict | None, optional
            Free-form metadata.

        Returns
        -------
        RawData
            The created record.
        """
        handle, filename, opened = _resolve_file(file)
        try:
            data = {
                "project_id": project_id,
                "name": name or filename,
                "metadata": json.dumps(metadata or {}),
            }
            files = {"file": (filename, handle, "application/octet-stream")}
            response = self.client.upload_multipart(self._BASE, data=data, files=files)
        finally:
            if opened:
                handle.close()
        if not isinstance(response, dict):
            raise TypeError("Unexpected upload response")
        return RawData(**response)

    def list(
        self, project_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[RawData]:
        """List raw-data records for a project."""
        endpoint = _build_endpoint(
            self._BASE, {"project_id": project_id, "limit": limit, "offset": offset}
        )
        return _parse_list_response(self.client.get(endpoint), RawData)

    def get(self, raw_data_id: str) -> RawData:
        """Fetch one raw-data record."""
        return RawData(**self.client.get(f"{self._BASE}/{raw_data_id}"))

    def download_url(self, raw_data_id: str) -> str:
        """Get a short-lived signed download URL."""
        return self.client.get(f"{self._BASE}/{raw_data_id}/download-url")["url"]

    def update(
        self, raw_data_id: str, name: str | None = None, metadata: dict | None = None
    ) -> RawData:
        """Partial-update a record's name/metadata."""
        payload: dict = {}
        if name is not None:
            payload["name"] = name
        if metadata is not None:
            payload["metadata"] = metadata
        return RawData(**self.client.patch(f"{self._BASE}/{raw_data_id}", payload))

    def delete(self, raw_data_id: str) -> None:
        """Delete a raw-data record and its file."""
        self.client.delete(f"{self._BASE}/{raw_data_id}")

    def list_measurements(
        self, raw_data_id: str, limit: int = 100, offset: int = 0
    ) -> PaginatedList[str]:
        """List measurement IDs produced from this raw-data record.

        The endpoint returns plain string ids, so build the PaginatedList
        directly — do NOT use ``_parse_list_response``, which calls
        ``model_class(**item)`` and would crash on ``str(**"uuid")``.
        """
        endpoint = _build_endpoint(
            f"{self._BASE}/{raw_data_id}/cell_measurements",
            {"limit": limit, "offset": offset},
        )
        resp = self.client.get(endpoint)
        return PaginatedList(
            items=list(resp.get("items", [])),
            total=resp.get("total", 0),
            limit=limit,
            offset=offset,
        )


def _resolve_file(
    file: str | os.PathLike | IO[bytes],
) -> tuple[IO[bytes], str, bool]:
    """Return an open binary handle, a filename, and whether we opened it.

    The third element tells the caller to close the handle in a ``finally`` only
    when this function opened it (a path was passed) — never close a caller-owned
    file-like object.
    """
    if isinstance(file, (str, os.PathLike)):
        return open(file, "rb"), os.path.basename(str(file)), True
    filename = os.path.basename(getattr(file, "name", "upload.bin"))
    return file, filename, False

from .raw_data import RawDataClient
...
self.raw_data = RawDataClient(self)

git add packages/ionworks-api/ionworks/raw_data.py packages/ionworks-api/ionworks/models.py packages/ionworks-api/ionworks/client.py packages/ionworks-api/ionworks/__init__.py packages/ionworks-api/tests/test_raw_data.py
git commit -m "feat(sdk): add RawDataClient for raw-data upload/list/get/update/delete"

from .models import PaginatedList, RawData, _build_endpoint, _parse_list_response

def attach_raw_data(self, cell_measurement_id: str, raw_data_ids: list[str]) -> None:
    """Attach raw-data records to this measurement (bulk, idempotent)."""
    self.client.post(
        f"/cell_measurements/{cell_measurement_id}/raw_data",
        {"raw_data_ids": raw_data_ids},
    )

def detach_raw_data(self, cell_measurement_id: str, raw_data_id: str) -> None:
    """Remove one raw-data link from this measurement."""
    self.client.delete(
        f"/cell_measurements/{cell_measurement_id}/raw_data/{raw_data_id}"
    )

def list_raw_data(
    self, cell_measurement_id: str, limit: int = 100, offset: int = 0
) -> PaginatedList[RawData]:
    """List raw-data records linked to this measurement."""
    endpoint = _build_endpoint(
        f"/cell_measurements/{cell_measurement_id}/raw_data",
        {"limit": limit, "offset": offset},
    )
    return _parse_list_response(self.client.get(endpoint), RawData)

git add packages/ionworks-api/ionworks/cell_measurement.py packages/ionworks-api/tests/
git commit -m "feat(sdk): measurement attach/detach/list raw-data provenance methods"

git add packages/skills/skills/discover-api/SKILL.md packages/skills/skills/upload-data/SKILL.md packages/ionworks-api/.claude/skills/ionworks-dev-sync/SKILL.md
git commit -m "docs(skills): document raw-data SDK client and measurement attach"