oxen-server 0.65.1

use std::path::{Path, PathBuf};

use crate::errors::OxenHttpError;
use crate::helpers::get_repo_async;
use crate::params::{DFOptsQuery, PageNumQuery, app_data, df_opts_query, path_param, query_param};
use crate::tasks;

use actix_web::{HttpRequest, HttpResponse, web};

use futures_util::StreamExt;
use liboxen::constants::{self, TABLE_NAME};
use liboxen::core::db::data_frames::df_db::with_df_db_manager;
use liboxen::core::db::data_frames::workspace_df_db::schema_without_oxen_cols;
use liboxen::core::repo_locks;
use liboxen::error::OxenError;
use liboxen::model::{ParsedResource, Schema, Workspace};
use liboxen::opts::{DFOpts, SliceRange};
use liboxen::repositories;
use liboxen::repositories::data_frames::schemas::{
    get_by_path_async, get_staged_schema_with_staged_db_manager_async,
};
use liboxen::repositories::workspaces::data_frames::is_indexed_async;
use liboxen::util::paginate;
use liboxen::view::data_frames::DataFramePayload;
use liboxen::view::entries::ResourceVersion;
use liboxen::view::entries::{PaginatedMetadataEntries, PaginatedMetadataEntriesResponse};
use liboxen::view::json_data_frame_view::WorkspaceJsonDataFrameViewResponse;
use liboxen::view::workspaces::RenameRequest;
use liboxen::view::{JsonDataFrameViews, StatusMessage, StatusMessageDescription};
use tempfile::TempPath;
use tokio::fs::File;
use tokio_util::io::ReaderStream;

pub mod columns;
pub mod embeddings;
pub mod rows;

/// Bytes read from an exported data frame per response chunk.
const EXPORT_READ_BUFFER_SIZE: usize = 2 * 1024 * 1024;

pub async fn get(
    req: HttpRequest,
    query: web::Query<DFOptsQuery>,
) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };
    let file_path = PathBuf::from(path_param(&req, "path")?);

    let mut opts = DFOpts::empty();
    opts = df_opts_query::parse_opts(&query, &mut opts)?;
    opts.path = Some(file_path.clone());

    // Clamp pagination to the valid 1-based domain once, when read: downstream a 0 page derives an
    // empty "0..0" slice that panics and a 0 page_size breaks the slice, the total_pages division,
    // and the indexed branch's SQL LIMIT.
    opts.page = Some(query.page.unwrap_or(constants::DEFAULT_PAGE_NUM).max(1));
    opts.page_size = Some(
        query
            .page_size
            .unwrap_or(constants::DEFAULT_PAGE_SIZE)
            .max(1),
    );

    let is_indexed = is_indexed_async(&workspace, &file_path).await?;

    if !is_indexed {
        let commit = workspace.commit.clone();
        let resource: ParsedResource = ParsedResource {
            path: file_path.clone(),
            version: PathBuf::from(commit.id.to_string()),
            resource: file_path.clone(),
            workspace: None,
            commit: Some(commit.clone()),
            branch: None,
        };

        // The read path paginates only via opts.slice, not page/page_size, so convert the requested
        // page into a slice before reading (as controllers::data_frames::get does). Without it the
        // read returns the whole frame regardless of the page. Skip if slice/row is set.
        if opts.slice_indices()?.is_none() {
            let (page, page_size) = opts.page_bounds();
            opts.slice = Some(SliceRange::for_page(page, page_size));
        }

        let data_frame_slice =
            repositories::data_frames::get_slice(&repo, &resource.clone(), &resource.path, &opts)
                .await?;

        let df = data_frame_slice.slice;
        let count = if opts.has_filter_transform() {
            data_frame_slice.total_entries
        } else {
            data_frame_slice.schemas.slice.size.height
        };

        let df_schema = if let Some(schema) = get_by_path_async(&repo, &commit, &file_path).await? {
            schema
        } else {
            Schema::from_polars(df.schema())
        };

        let mut df_views =
            JsonDataFrameViews::from_df_and_opts_unpaginated(df, df_schema, count, &opts).await?;

        // Metadata can be staged before the frame is ever indexed; surface it.
        let staged_schema =
            get_staged_schema_with_staged_db_manager_async(&workspace.workspace_repo, &file_path)
                .await?;
        repositories::workspaces::data_frames::columns::update_column_schemas(
            staged_schema,
            &mut df_views,
        );

        let response = WorkspaceJsonDataFrameViewResponse {
            status: StatusMessage::resource_found(),
            data_frame: Some(df_views),
            resource: None,
            commit: None, // Not at a committed state
            derived_resource: None,
            is_indexed,
        };

        return Ok(HttpResponse::Ok().json(response));
    }

    log::debug!("querying data frame {file_path:?}");
    log::debug!("opts: {opts:?}");
    // Each read is DuckDB work, and each gets its own offload from the request
    // thread rather than sharing one: per docs/async_policy.md the granularity is
    // one offload per operation, so the two stay independently convertible.
    let count = {
        let workspace = workspace.clone();
        let file_path = file_path.clone();
        let opts = opts.clone();
        tasks::spawn_blocking(move || {
            repositories::workspaces::data_frames::count_for_query(&workspace, &file_path, &opts)
        })
        .await
        .map_err(OxenError::from)??
    };
    let df = {
        let workspace = workspace.clone();
        let file_path = file_path.clone();
        let opts = opts.clone();
        tasks::spawn_blocking(move || {
            repositories::workspaces::data_frames::query(&workspace, &file_path, &opts)
        })
        .await
        .map_err(OxenError::from)??
    };

    let Some(df_schema) = get_by_path_async(&repo, &workspace.commit, &file_path).await? else {
        log::warn!("Failed to get schema for data frame {file_path:?}");
        return Err(OxenHttpError::NotFound);
    };

    let resource = ResourceVersion {
        path: file_path.to_string_lossy().to_string(),
        version: workspace.commit.id.to_string(),
    };

    let mut df_views =
        JsonDataFrameViews::from_df_and_opts_unpaginated(df, df_schema, count, &opts).await?;

    let new_schema =
        get_staged_schema_with_staged_db_manager_async(&workspace.workspace_repo, &file_path)
            .await?;
    repositories::workspaces::data_frames::columns::update_column_schemas(
        new_schema,
        &mut df_views,
    );

    let response = WorkspaceJsonDataFrameViewResponse {
        status: StatusMessage::resource_found(),
        data_frame: Some(df_views),
        resource: Some(resource),
        commit: None, // Not at a committed state
        derived_resource: None,
        is_indexed,
    };

    Ok(HttpResponse::Ok().json(response))
}

pub async fn get_schema(req: HttpRequest) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    // Serving a schema opens the workspace's staged RocksDB read-write, creating it when absent, so
    // this GET counts as a write against a stop-the-world op (workspace clear / migration / prune).
    let _write = repo_locks::begin_write(&repo)?;
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };
    let file_path = PathBuf::from(path_param(&req, "path")?);

    let staged_schema =
        get_staged_schema_with_staged_db_manager_async(&workspace.workspace_repo, &file_path)
            .await?;

    // Never index on a read: rebuilding the staged table would discard any
    // staged row edits (e.g. behind a stale index marker). Serve the committed
    // schema until something else indexes the frame.
    let schema = if is_indexed_async(&workspace, &file_path).await? {
        let db_path = repositories::workspaces::data_frames::duckdb_path(&workspace, &file_path);
        let mut schema = tasks::spawn_blocking(move || {
            with_df_db_manager(&db_path, |manager| {
                manager.with_conn(|conn| schema_without_oxen_cols(conn, TABLE_NAME))
            })
        })
        .await
        .map_err(OxenError::from)??;
        // The staged table carries the workspace's columns but none of its metadata.
        if let Some(staged_schema) = staged_schema {
            schema.update_metadata_from_schema(&staged_schema);
        }
        schema
    } else if let Some(staged_schema) = staged_schema {
        // A data frame the workspace added is absent from the commit, so its staged node holds
        // the only copy of the schema, metadata included.
        staged_schema
    } else {
        get_by_path_async(&repo, &workspace.commit, &file_path)
            .await?
            .ok_or(OxenHttpError::NotFound)?
    };

    Ok(HttpResponse::Ok().json(schema))
}

/// Set the metadata on a data frame's schema itself — the file-level
/// counterpart to the per-column metadata endpoint. Staged into the
/// workspace, so no commit is required.
#[utoipa::path(
    put,
    path = "/api/repos/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/schema/{path}",
    description = "Set the metadata on a data frame's schema itself, the HTTP equivalent of `oxen schemas add <path> -m '{...}'`. Staged into the workspace, so no commit is required. Replaces any existing schema metadata wholesale.",
    tag = "Workspace Data Frames",
    params(
        ("namespace" = String, Path, description = "Namespace of the repository", example = "ox"),
        ("repo_name" = String, Path, description = "Name of the repository", example = "ImageNet-1k"),
        ("workspace_id" = String, Path, description = "ID or name of the workspace", example = "b3f27f05-0955-4076-805f-39575853b27b"),
        ("path" = String, Path, description = "Path to the data frame within the repository", example = "annotations/train.csv"),
    ),
    request_body(
        content = String,
        description = "The metadata to store on the schema, under a `metadata` key. The `_oxen` key is reserved for metadata oxen itself interprets.",
        example = json!({
            "metadata": {
                "task": "bounding_box",
                "description": "Extracting bounding boxes from images"
            }
        })
    ),
    responses(
        (status = 200, description = "Schema metadata updated", body = StatusMessage),
        (status = 400, description = "Missing or invalid `metadata` in the request body, or the path is not a tabular data frame"),
        (status = 404, description = "Repository, workspace, or data frame not found"),
        (status = 409, description = "The data frame is staged for removal in this workspace")
    )
)]
pub async fn put_schema_metadata(
    req: HttpRequest,
    body: String,
) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let _write = repo_locks::begin_write(&repo)?;
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };
    let file_path = PathBuf::from(path_param(&req, "path")?);

    let parsed_json: serde_json::Value = serde_json::from_str(&body)?;
    let Some(metadata) = parsed_json.get("metadata").cloned() else {
        return Err(OxenHttpError::BasicError("metadata is required".into()));
    };

    tasks::spawn_blocking(move || {
        repositories::workspaces::data_frames::schemas::add_schema_metadata(
            &repo, &workspace, &file_path, &metadata,
        )
    })
    .await
    .map_err(OxenError::from)??;

    Ok(HttpResponse::Ok().json(StatusMessage::resource_updated()))
}

fn determine_extension<'a>(opts: &'a DFOpts, file_path: &'a Path) -> &'a str {
    match &opts.output {
        // If the user specified a format, we'll export to that format
        Some(output) => output
            .extension()
            .unwrap_or_default()
            .to_str()
            .unwrap_or_default(),
        None => file_path
            .extension()
            .unwrap_or_default()
            .to_str()
            .unwrap_or_default(),
    }
}

// fn determine_content_type(extension: &str) -> &str {
//   match extension {
//       "parquet" => "application/octet-stream",
//       "json" | "jsonl" | "ndjson" => "application/json",
//       _ => "text/csv",
//   }
// }

pub async fn download(
    req: HttpRequest,
    query: web::Query<DFOptsQuery>,
) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };
    let file_path = PathBuf::from(path_param(&req, "path")?);
    let Some(filename) = file_path.file_name().and_then(|n| n.to_str()) else {
        log::warn!(
            "Unable to get filename from request param path: {}",
            file_path.display()
        );
        return Err(OxenHttpError::BadRequest(
            "Unable to parse filename from 'path' parameter".into(),
        ));
    };
    let content_disposition = format!("attachment; filename=\"{filename}\"");

    let opts = df_opts_from_query(&query, file_path.clone())?;

    if !is_indexed_async(&workspace, &file_path).await? {
        let file_exists = tasks::spawn_blocking(move || {
            file_exists_in_workspace_or_commit(&workspace, &file_path)
        })
        .await
        .map_err(OxenError::from)??;
        if !file_exists {
            return Err(OxenHttpError::NotFound);
        }

        let response = df_not_indexed_response();

        return Ok(HttpResponse::Ok().json(response));
    }

    log::debug!("exporting data frame {file_path:?}");
    log::debug!("opts: {opts:?}");

    let extension = determine_extension(&opts, &file_path);
    let export_path = std::env::temp_dir().join(format!("{}.{}", uuid::Uuid::new_v4(), extension));
    // Removes the export when dropped.
    let export_path = TempPath::try_from_path(export_path)?;

    // Export the data frame. An `Ok(Err(..))` is a failed export, which answers 400.
    let export = tasks::spawn_blocking(move || -> Result<_, OxenError> {
        if let Err(e) = repositories::workspaces::data_frames::export(
            &workspace,
            &file_path,
            &opts,
            &export_path,
        ) {
            log::warn!("Error exporting data frame {file_path:?}: {e:?}");
            return Ok(Err(e));
        }
        let file = std::fs::File::open(&export_path)?;
        let len = file.metadata()?.len();
        Ok(Ok((export_path, file, len)))
    })
    .await
    .map_err(OxenError::from)??;
    let (export_path, file, len) = match export {
        Ok(export) => export,
        Err(e) => {
            let response = StatusMessageDescription::bad_request(format!("{e:?}"));
            return Ok(HttpResponse::BadRequest().json(response));
        }
    };

    // The body owns the export, so the file lasts until the body is sent or abandoned.
    let body = ReaderStream::with_capacity(File::from_std(file), EXPORT_READ_BUFFER_SIZE).map(
        move |chunk| {
            let _export_path = &export_path;
            chunk
        },
    );

    Ok(HttpResponse::Ok()
        .no_chunking(len)
        // .append_header(("Content-Type", determine_content_type(extension)))
        .append_header(("Content-Type", "text/csv"))
        .append_header(("Content-Disposition", content_disposition))
        .streaming(body))
}

fn df_opts_from_query(
    query: &web::Query<DFOptsQuery>,
    file_path: PathBuf,
) -> Result<DFOpts, OxenError> {
    let mut opts = DFOpts::empty();
    opts = df_opts_query::parse_opts(query, &mut opts)?;
    opts.path = Some(file_path);

    opts.page = Some(query.page.unwrap_or(constants::DEFAULT_PAGE_NUM));
    opts.page_size = Some(query.page_size.unwrap_or(constants::DEFAULT_PAGE_SIZE));
    Ok(opts)
}

/// Check if the file exists in the repository or the workspace.
/// If not, then it should be a genuine 404.
fn file_exists_in_workspace_or_commit(
    workspace: &Workspace,
    file_path: &Path,
) -> Result<bool, OxenError> {
    let file_exists = {
        (
            // does the file exist in the base repository?
            repositories::tree::get_file_by_path(
                &workspace.base_repo,
                &workspace.commit,
                file_path,
            )?
            .is_some()
        ) || (
            // if not, does it exist in the workspace
            repositories::workspaces::files::exists(workspace, file_path)?
        )
    };
    Ok(file_exists)
}

fn df_not_indexed_response() -> WorkspaceJsonDataFrameViewResponse {
    WorkspaceJsonDataFrameViewResponse {
        status: StatusMessage::resource_found(),
        data_frame: None,
        resource: None,
        commit: None, // Not at a committed state
        derived_resource: None,
        is_indexed: false,
    }
}

pub async fn get_by_branch(
    req: HttpRequest,
    query: web::Query<PageNumQuery>,
) -> actix_web::Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let branch_name = query_param(&req, "branch").to_string();
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };

    let page = query.page.unwrap_or(constants::DEFAULT_PAGE_NUM);
    let page_size = query.page_size.unwrap_or(constants::DEFAULT_PAGE_SIZE);

    let editable_entries = tasks::spawn_blocking(move || -> Result<_, OxenError> {
        // Staged dataframes must be on a branch.
        let branch = repositories::branches::get_by_name(&repo, &branch_name)?;

        let commit = repositories::commits::get_by_id(&repo, &branch.commit_id)?
            .ok_or_else(|| OxenError::resource_not_found(&branch.commit_id))?;

        let entries = repositories::entries::list_tabular_files_in_repo(&repo, &commit)?;
        log::debug!("got {} tabular entries", entries.len());

        let mut editable_entries = vec![];
        for entry in entries {
            log::debug!("considering entry {entry:?}");
            let path = PathBuf::from(&entry.filename);
            if repositories::workspaces::data_frames::is_indexed(&workspace, &path)? {
                editable_entries.push(entry);
            } else {
                log::debug!("not indexed {path:?}");
            }
        }
        Ok(editable_entries)
    })
    .await
    .map_err(OxenError::from)??;

    let (paginated_entries, pagination) = paginate(editable_entries, page, page_size);
    Ok(HttpResponse::Ok().json(PaginatedMetadataEntriesResponse {
        status: StatusMessage::resource_found(),
        entries: PaginatedMetadataEntries {
            entries: paginated_entries,
            pagination,
        },
    }))
}

/// Index a data frame into DuckDB for querying
pub async fn put(req: HttpRequest, body: String) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let _write = repo_locks::begin_write(&repo)?;
    let file_path = PathBuf::from(path_param(&req, "path")?);

    log::debug!("workspace {workspace_id} data frame put {file_path:?}");
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };
    let data: DataFramePayload = serde_json::from_str(&body)?;
    log::debug!("workspace {workspace_id} data frame put {data:?}");

    let to_index = data.is_indexed;
    let is_indexed = is_indexed_async(&workspace, &file_path).await?;

    if !is_indexed && to_index {
        repositories::workspaces::data_frames::index(&repo, &workspace, &file_path).await?;
    } else if is_indexed && !to_index {
        tasks::spawn_blocking(move || {
            repositories::workspaces::data_frames::unindex(&workspace, &file_path)
        })
        .await
        .map_err(OxenError::from)??;
    }

    Ok(HttpResponse::Ok().json(StatusMessage::resource_updated()))
}

pub async fn delete(req: HttpRequest) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;

    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let _write = repo_locks::begin_write(&repo)?;
    let file_path = PathBuf::from(path_param(&req, "path")?);
    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };

    repositories::workspaces::data_frames::restore(&repo, &workspace, file_path).await?;

    Ok(HttpResponse::Ok().json(StatusMessage::resource_deleted()))
}

pub async fn rename(req: HttpRequest, body: String) -> Result<HttpResponse, OxenHttpError> {
    let app_data = app_data(&req)?;
    let namespace = path_param(&req, "namespace")?.to_string();
    let repo_name = path_param(&req, "repo_name")?.to_string();
    let workspace_id = path_param(&req, "workspace_id")?.to_string();
    let repo = get_repo_async(app_data, &namespace, &repo_name).await?;
    let _write = repo_locks::begin_write(&repo)?;
    let path = PathBuf::from(path_param(&req, "path")?);
    // Attempt to parse the body
    let body: RenameRequest = serde_json::from_str(&body)?; // Use the Json wrapper to get the inner value

    // Check if new_path is valid
    if body.new_path.is_empty() {
        return Err(OxenHttpError::BadRequest("new_path cannot be empty".into()));
    }

    let new_path = PathBuf::from(body.new_path);

    let Some(workspace) = repositories::workspaces::get_async(&repo, &workspace_id).await? else {
        return Ok(HttpResponse::NotFound()
            .json(StatusMessageDescription::workspace_not_found(workspace_id)));
    };

    let new_path_exists = {
        let workspace = workspace.clone();
        let new_path = new_path.clone();
        tasks::spawn_blocking(move || {
            repositories::entries::get_file(&repo, &workspace.commit, &new_path)
        })
        .await
        .map_err(OxenError::from)??
        .is_some()
    };
    if new_path_exists {
        return Err(OxenHttpError::BadRequest("new_path already exists".into()));
    }

    repositories::workspaces::data_frames::rename(&workspace, &path, &new_path).await?;

    Ok(HttpResponse::Ok().json(StatusMessage::resource_updated()))
}

#[cfg(test)]
mod tests {
    use crate::app_data::OxenAppData;
    use crate::controllers;
    use crate::test;
    use actix_web::body::{BodySize, MessageBody};
    use actix_web::{App, web};
    use liboxen::core::db::data_frames::df_db::with_df_db_manager;
    use liboxen::error::OxenError;
    use liboxen::model::Schema;
    use liboxen::opts::DFOpts;
    use liboxen::repositories;
    use liboxen::util;
    use liboxen::view::json_data_frame_view::WorkspaceJsonDataFrameViewResponse;
    use serde_json::json;

    #[actix_web::test]
    async fn test_workspace_data_frames() -> Result<(), OxenError> {
        liboxen::test::init_test_env();
        let sync_dir = test::get_sync_dir()?;
        let namespace = "Testing-Namespace";
        let repo_name = "Testing-Name";
        let repo = test::create_local_repo(&sync_dir, namespace, repo_name)?;

        let csv_dir = repo.path.join("data");
        util::fs::create_dir_all(&csv_dir)?;
        for name in ["test.csv", "stale.csv"] {
            util::fs::write_to_path(csv_dir.join(name), "col_a,col_b\n1,2\n3,4\n")?;
        }
        // 25 rows, larger than the page_size of 10 the pagination checks request, so their pages
        // are real pages.
        let mut history = String::from("id,label\n");
        for i in 0..25 {
            history.push_str(&format!("{i},label_{i}\n"));
        }
        util::fs::write_to_path(csv_dir.join("history.csv"), &history)?;
        util::fs::write_to_path(repo.path.join("README.md"), "# Not a data frame\n")?;
        repositories::add(&repo, &repo.path).await?;
        let commit = repositories::commit(&repo, "Add data frames")?;

        // A commit can only have one non-editable workspace, so every case below works on its own
        // file in this one.
        let workspace_id = uuid::Uuid::new_v4().to_string();
        let workspace = repositories::workspaces::create(&repo, &commit, &workspace_id, false)?;

        let workspace_uri = format!("/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}");
        let schema_route =
            "/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/schema/{path:.*}";
        let app = actix_web::test::init_service(
            App::new()
                .app_data(OxenAppData::new(sync_dir.clone()))
                .route(
                    schema_route,
                    web::put().to(controllers::workspaces::data_frames::put_schema_metadata),
                )
                .route(
                    schema_route,
                    web::get().to(controllers::workspaces::data_frames::get_schema),
                )
                .route(
                    "/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/download/{path:.*}",
                    web::get().to(controllers::workspaces::data_frames::download),
                ),
        )
        .await;

        // Schema metadata round-trips on a committed data frame without indexing it.
        let file_path = std::path::Path::new("data/test.csv");
        let uri = format!("{workspace_uri}/data_frames/schema/{}", file_path.display());
        let metadata = json!({"task": "classification"});
        let req = actix_web::test::TestRequest::put()
            .uri(&uri)
            .set_json(json!({"metadata": metadata}))
            .to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        assert!(
            !repositories::workspaces::data_frames::duckdb_path(&workspace, file_path).exists(),
            "writing schema metadata should not index an unindexed data frame"
        );

        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        let bytes = actix_http::body::to_bytes(resp.into_body()).await.unwrap();
        let schema: Schema = serde_json::from_slice(&bytes)?;
        assert_eq!(schema.metadata, Some(metadata.clone()));
        assert!(
            !repositories::workspaces::data_frames::has_staged_table(&workspace, file_path)?,
            "reading a schema should not build a staged table for an unindexed data frame"
        );

        // A committed data frame that is not indexed downloads as JSON reporting it unindexed.
        let uri = format!("{workspace_uri}/data_frames/download/data/test.csv");
        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        let bytes = actix_http::body::to_bytes(resp.into_body()).await.unwrap();
        let response: WorkspaceJsonDataFrameViewResponse = serde_json::from_slice(&bytes)?;
        assert!(!response.is_indexed);
        assert!(response.data_frame.is_none());

        // A path in neither the base commit nor the workspace is a 404.
        let uri = format!("{workspace_uri}/data_frames/download/data/this_does_not_exist.csv");
        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::NOT_FOUND);

        // A non-tabular path is a client mistake, not a server fault: it has to come back 4xx so
        // callers can distinguish it from a real failure.
        let req = actix_web::test::TestRequest::put()
            .uri(&format!("{workspace_uri}/data_frames/schema/README.md"))
            .set_json(json!({"metadata": {"task": "classification"}}))
            .to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::BAD_REQUEST);
        let bytes = actix_http::body::to_bytes(resp.into_body()).await.unwrap();
        let body: serde_json::Value = serde_json::from_slice(&bytes)?;
        assert_eq!(
            body["error"]["type"], "not_a_data_frame",
            "the response must name the cause, not just fail: {body}"
        );

        // A data frame the workspace added is editable before it is committed, so its schema has
        // to be readable there too: the PUT and the GET have to agree on which paths exist.
        let file_path = std::path::Path::new("data/uncommitted.csv");
        let workspace_csv = workspace.workspace_repo.path.join(file_path);
        util::fs::create_dir_all(workspace_csv.parent().unwrap())?;
        util::fs::write_to_path(&workspace_csv, "col_a,col_b\n1,2\n")?;
        repositories::workspaces::files::add(&workspace, &workspace_csv).await?;

        let uri = format!("{workspace_uri}/data_frames/schema/{}", file_path.display());
        let req = actix_web::test::TestRequest::put()
            .uri(&uri)
            .set_json(json!({"metadata": metadata}))
            .to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);

        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        let bytes = actix_http::body::to_bytes(resp.into_body()).await.unwrap();
        let schema: Schema = serde_json::from_slice(&bytes)?;
        assert_eq!(schema.metadata, Some(metadata));
        let names: Vec<&str> = schema.fields.iter().map(|f| f.name.as_str()).collect();
        assert_eq!(names, vec!["col_a", "col_b"]);

        // The staged node's schema is served as stored, so its hash has to already agree with
        // its fields.
        let mut recomputed = schema.clone();
        recomputed.recompute_hash();
        assert_eq!(schema.hash, recomputed.hash);

        // A data frame only staged in the workspace, never indexed, downloads as JSON reporting
        // it unindexed.
        let workspace_csv_path = workspace.dir().join("data/new.csv");
        util::fs::create_dir_all(workspace_csv_path.parent().unwrap())?;
        util::fs::write_to_path(&workspace_csv_path, "x,y\n10,20\n")?;
        repositories::workspaces::files::add(&workspace, &workspace_csv_path).await?;
        let uri = format!("{workspace_uri}/data_frames/download/data/new.csv");
        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        let bytes = actix_http::body::to_bytes(resp.into_body()).await.unwrap();
        let response: WorkspaceJsonDataFrameViewResponse = serde_json::from_slice(&bytes)?;
        assert!(!response.is_indexed);
        assert!(response.data_frame.is_none());

        // An indexed data frame downloads as CSV.
        let file_path = std::path::Path::new("data/test.csv");
        repositories::workspaces::data_frames::index(&repo, &workspace, file_path).await?;
        let uri = format!(
            "{workspace_uri}/data_frames/download/{}",
            file_path.display()
        );
        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        let content_type = resp
            .headers()
            .get("Content-Type")
            .unwrap()
            .to_str()
            .unwrap();
        assert_eq!(content_type, "text/csv");
        let declared_size = resp.response().body().size();
        let bytes = actix_http::body::to_bytes(resp.into_body()).await.unwrap();
        assert_eq!(
            declared_size,
            BodySize::Sized(bytes.len() as u64),
            "the streamed export declares its length up front"
        );
        let body = String::from_utf8(bytes.to_vec()).unwrap();
        assert!(body.contains("col_a"));
        assert!(body.contains("col_b"));

        // Writing or reading schema metadata leaves a stale staged table's rows alone.
        let file_path = std::path::Path::new("data/stale.csv");
        repositories::workspaces::data_frames::index(&repo, &workspace, file_path).await?;
        repositories::workspaces::data_frames::rows::add(
            &repo,
            &workspace,
            file_path,
            &json!({"col_a": 5, "col_b": 6}),
        )?;
        assert_eq!(
            repositories::workspaces::data_frames::count(&workspace, file_path)?,
            3
        );
        let db_path = repositories::workspaces::data_frames::duckdb_path(&workspace, file_path);
        with_df_db_manager(&db_path, |manager| {
            manager.with_conn(|conn| {
                conn.execute(
                    &format!("DROP TABLE \"{}\"", liboxen::constants::INDEX_META_TABLE),
                    [],
                )?;
                Ok(())
            })
        })?;
        assert!(!repositories::workspaces::data_frames::is_indexed(
            &workspace, file_path
        )?);

        let uri = format!("{workspace_uri}/data_frames/schema/{}", file_path.display());
        let req = actix_web::test::TestRequest::put()
            .uri(&uri)
            .set_json(json!({"metadata": {"task": "classification"}}))
            .to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        assert_eq!(
            repositories::workspaces::data_frames::count(&workspace, file_path)?,
            3,
            "writing metadata must not discard rows from a stale staged table"
        );
        assert!(!repositories::workspaces::data_frames::is_indexed(
            &workspace, file_path
        )?);

        // Reading the schema takes the same unindexed branch, so it must leave the stale table
        // alone too.
        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);
        assert_eq!(
            repositories::workspaces::data_frames::count(&workspace, file_path)?,
            3,
            "reading a schema must not discard rows from a stale staged table"
        );
        assert!(!repositories::workspaces::data_frames::is_indexed(
            &workspace, file_path
        )?);

        // An unindexed data frame larger than the requested `page_size` must paginate: each page
        // returns at most `page_size` rows and an out-of-range page returns zero rows so
        // pagination terminates. Indexing it afterwards checks the indexed branch paginates
        // identically, so the two paths stay in lockstep.
        let file_path = "data/history.csv";
        let page = |query: DFOpts| {
            get_data_frame_page(
                &sync_dir,
                namespace,
                repo_name,
                &workspace_id,
                file_path,
                query,
            )
        };

        let page_1 = page(page_query(1, 10, None)).await;
        assert!(!page_1.is_indexed);
        assert_eq!(page_row_count(&page_1), 10);
        assert_eq!(page_ids(&page_1), (0..10).collect::<Vec<i64>>());
        let pagination = &page_1.data_frame.as_ref().unwrap().view.pagination;
        assert_eq!(pagination.total_pages, 3);
        assert_eq!(pagination.total_entries, 25);

        // Page 3 → the remaining 5 rows.
        let page_3 = page(page_query(3, 10, None)).await;
        assert_eq!(page_row_count(&page_3), 5);
        assert_eq!(page_ids(&page_3), (20..25).collect::<Vec<i64>>());

        // Page 4 is past the end → 0 rows. This is the case a pager relies on to terminate.
        let page_4 = page(page_query(4, 10, None)).await;
        assert_eq!(page_row_count(&page_4), 0);
        let pagination = &page_4.data_frame.as_ref().unwrap().view.pagination;
        assert_eq!(pagination.total_pages, 3);
        assert_eq!(pagination.total_entries, 25);

        // Out-of-domain pagination inputs are clamped to the 1-based domain rather than panicking
        // or returning the whole frame: page 0 is read as page 1 (10 rows)...
        let page_zero = page(page_query(0, 10, None)).await;
        assert_eq!(page_row_count(&page_zero), 10);
        // ...and page_size 0 is read as page_size 1 (a single row), not the full 25-row frame.
        let zero_page_size = page(page_query(1, 0, None)).await;
        assert_eq!(page_row_count(&zero_page_size), 1);
        assert_eq!(
            zero_page_size
                .data_frame
                .as_ref()
                .unwrap()
                .view
                .pagination
                .total_entries,
            25
        );

        // A client that wants a whole result asks for it as one maximal page. The widest page a
        // request can name has to mean "all of them" on either branch, rather than overflowing
        // the slice or exceeding what a SQL `LIMIT` accepts.
        let widest = || page_query(1, usize::MAX, Some("SELECT * FROM df"));
        let response = page(widest()).await;
        assert!(
            !response.is_indexed,
            "sql is ignored until the frame is indexed"
        );
        assert_eq!(page_row_count(&response), 25);

        repositories::workspaces::data_frames::index(
            &repo,
            &workspace,
            std::path::Path::new(file_path),
        )
        .await?;

        let indexed_expectations: [(usize, Vec<i64>); 3] = [
            (1, (0..10).collect()),
            (3, (20..25).collect()),
            (4, Vec::new()),
        ];
        for (page_number, expected_ids) in indexed_expectations {
            let indexed_page = page(page_query(page_number, 10, None)).await;
            assert!(indexed_page.is_indexed);
            assert_eq!(page_row_count(&indexed_page), expected_ids.len());
            assert_eq!(page_ids(&indexed_page), expected_ids);
            let pagination = &indexed_page.data_frame.as_ref().unwrap().view.pagination;
            assert_eq!(pagination.total_pages, 3);
            assert_eq!(pagination.total_entries, 25);
        }

        let response = page(widest()).await;
        assert!(response.is_indexed);
        assert_eq!(page_row_count(&response), 25);

        // A `sql` query is paginated like any other read, and the pagination the response reports
        // describes the rows the query selects rather than the whole frame, or a client paging by
        // `total_pages` walks past the end of the result. 15 of the 25 rows match, so 2 pages of
        // 10 and nothing after them.
        let sql = "SELECT * FROM df WHERE id >= 10";
        let expectations: [(usize, Vec<i64>); 3] = [
            (1, (10..20).collect()),
            (2, (20..25).collect()),
            (3, Vec::new()),
        ];
        for (page_number, expected_ids) in expectations {
            let response = page(page_query(page_number, 10, Some(sql))).await;
            assert!(response.is_indexed);
            assert_eq!(
                page_row_count(&response),
                expected_ids.len(),
                "page {page_number} must hold only its own rows"
            );
            assert_eq!(page_ids(&response), expected_ids);
            let pagination = &response.data_frame.as_ref().unwrap().view.pagination;
            assert_eq!(pagination.total_entries, 15);
            assert_eq!(pagination.total_pages, 2);
        }

        drop((app, workspace));
        test::cleanup_repo_and_sync_dir(repo, &sync_dir)?;
        Ok(())
    }

    /// The query string a page request carries, built the way the client builds
    /// it so the handler parses what it parses in production.
    fn page_query(page: usize, page_size: usize, sql: Option<&str>) -> DFOpts {
        let mut opts = DFOpts::empty();
        opts.page = Some(page);
        opts.page_size = Some(page_size);
        opts.sql = sql.map(str::to_string);
        opts
    }

    /// GET a page of a workspace data frame through the `get` handler.
    async fn get_data_frame_page(
        sync_dir: &std::path::Path,
        namespace: &str,
        repo_name: &str,
        workspace_id: &str,
        file_path: &str,
        query: DFOpts,
    ) -> WorkspaceJsonDataFrameViewResponse {
        let app = actix_web::test::init_service(
            App::new()
                .app_data(OxenAppData::new(sync_dir.to_path_buf()))
                .route(
                    "/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/resource/{path:.*}",
                    web::get().to(controllers::workspaces::data_frames::get),
                ),
        )
        .await;

        let uri = format!(
            "/oxen/{namespace}/{repo_name}/workspaces/{workspace_id}/data_frames/resource/{file_path}?{}",
            query.to_http_query_params()
        );
        let req = actix_web::test::TestRequest::get().uri(&uri).to_request();
        let resp = actix_web::test::call_service(&app, req).await;
        assert_eq!(resp.status(), actix_web::http::StatusCode::OK);

        let bytes = actix_http::body::to_bytes(resp.into_body())
            .await
            .expect("could not read response body");
        serde_json::from_slice(&bytes).expect("could not deserialize data frame response")
    }

    /// Number of rows the response carries in its view (the size of the returned page).
    fn page_row_count(response: &WorkspaceJsonDataFrameViewResponse) -> usize {
        let data_frame = response
            .data_frame
            .as_ref()
            .expect("expected a data frame in the response");
        let array_len = data_frame
            .view
            .data
            .as_array()
            .expect("expected the view data to be a JSON array")
            .len();
        // The reported view height and the serialized row array must agree.
        assert_eq!(array_len, data_frame.view.size.height);
        array_len
    }

    /// The `id` column of every row on a page, in the order the page returned them.
    fn page_ids(response: &WorkspaceJsonDataFrameViewResponse) -> Vec<i64> {
        response
            .data_frame
            .as_ref()
            .expect("expected a data frame in the response")
            .view
            .data
            .as_array()
            .expect("expected the view data to be a JSON array")
            .iter()
            .map(|row| {
                row.get("id")
                    .and_then(serde_json::Value::as_i64)
                    .expect("expected an integer id on every row")
            })
            .collect()
    }
}