Struct DatasetRepository 

pub struct DatasetRepository { /* private fields */ }

Repository for dataset persistence in PostgreSQL with pgvector.

Examples

use sqlx::postgres::PgPoolOptions;
use ceres_db::DatasetRepository;

let pool = PgPoolOptions::new()
    .max_connections(5)
    .connect("postgresql://localhost/ceres")
    .await?;

let repo = DatasetRepository::new(pool);

Implementations

impl DatasetRepository

pub fn new(pool: PgPool) -> Self

pub async fn upsert(&self, new_data: &NewDataset) -> Result<Uuid, AppError>

Inserts or updates a dataset. Returns the UUID of the affected row.

TODO(robustness): Return UpsertOutcome to distinguish insert vs update Currently returns only UUID without indicating operation type. Consider: pub enum UpsertOutcome { Created(Uuid), Updated(Uuid) } This enables accurate progress reporting in sync statistics.

pub async fn get_hashes_for_portal( &self, portal_url: &str, ) -> Result<HashMap<String, Option<String>>, AppError>

Returns a map of original_id → content_hash for all datasets from a portal.

TODO(performance): Optimize for large portals (100k+ datasets) Currently loads entire HashMap into memory. Consider: (1) Streaming hash comparison during sync, or (2) Database-side hash check with WHERE clause, or (3) Bloom filter for approximate membership testing

pub async fn update_timestamp_only( &self, portal_url: &str, original_id: &str, ) -> Result<bool, AppError>

Updates only the timestamp for unchanged datasets. Returns true if a row was updated.

pub async fn batch_update_timestamps( &self, portal_url: &str, original_ids: &[String], ) -> Result<u64, AppError>

Batch updates timestamps for multiple unchanged datasets.

Uses a single UPDATE with ANY() for efficiency instead of N individual updates. Returns the number of rows actually updated.

pub async fn get(&self, id: Uuid) -> Result<Option<Dataset>, AppError>

Retrieves a dataset by UUID.

pub async fn search( &self, query_vector: Vector, limit: usize, ) -> Result<Vec<SearchResult>, AppError>

Semantic search using cosine similarity. Returns results ordered by similarity.

pub async fn list_all( &self, portal_filter: Option<&str>, limit: Option<usize>, ) -> Result<Vec<Dataset>, AppError>

Lists datasets with optional portal filter and limit.

For memory-efficient exports of large datasets, use [list_all_stream] instead.

TODO(config): Make default limit configurable via DEFAULT_EXPORT_LIMIT env var Currently hardcoded to 10000.

pub fn list_all_stream<'a>( &'a self, portal_filter: Option<&'a str>, limit: Option<usize>, ) -> BoxStream<'a, Result<Dataset, AppError>>

Lists datasets as a stream with optional portal filter.

Unlike [list_all], this method streams results directly from the database without loading everything into memory. Suitable for large exports.

Arguments

• portal_filter - Optional portal URL to filter by
• limit - Optional maximum number of records (no default limit for streaming)

pub async fn get_sync_status( &self, portal_url: &str, ) -> Result<Option<PortalSyncStatus>, AppError>

Retrieves the sync status for a portal. Returns None if this portal has never been synced.

pub async fn upsert_sync_status( &self, portal_url: &str, last_sync: DateTime<Utc>, sync_mode: &str, sync_status: &str, datasets_synced: i32, ) -> Result<(), AppError>

Updates or inserts the sync status for a portal.

The sync_status parameter indicates the outcome: “completed” or “cancelled”. Only updates last_successful_sync when status is “completed”, preserving the last successful sync time for incremental harvesting after cancellations.

pub async fn health_check(&self) -> Result<(), AppError>

Checks database connectivity by executing a simple query.

pub async fn get_stats(&self) -> Result<DatabaseStats, AppError>

Returns aggregated database statistics.

Trait Implementations

impl Clone for DatasetRepository

fn clone(&self) -> DatasetRepository

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl DatasetStore for DatasetRepository

async fn get_by_id(&self, id: Uuid) -> Result<Option<Dataset>, AppError>

Retrieves a dataset by its unique ID.

async fn get_hashes_for_portal( &self, portal_url: &str, ) -> Result<HashMap<String, Option<String>>, AppError>

Retrieves content hashes for all datasets from a specific portal.

async fn update_timestamp_only( &self, portal_url: &str, original_id: &str, ) -> Result<(), AppError>

Updates only the timestamp for an unchanged dataset.

async fn batch_update_timestamps( &self, portal_url: &str, original_ids: &[String], ) -> Result<u64, AppError>

Batch updates timestamps for multiple unchanged datasets.

async fn upsert(&self, dataset: &NewDataset) -> Result<Uuid, AppError>

Inserts or updates a dataset.

async fn search( &self, query_vector: Vector, limit: usize, ) -> Result<Vec<SearchResult>, AppError>

Performs vector similarity search.

fn list_stream<'a>( &'a self, portal_filter: Option<&'a str>, limit: Option<usize>, ) -> BoxStream<'a, Result<Dataset, AppError>>

Lists datasets as a stream with optional filtering.

async fn get_last_sync_time( &self, portal_url: &str, ) -> Result<Option<DateTime<Utc>>, AppError>

Retrieves the last successful sync timestamp for a portal.

async fn record_sync_status( &self, portal_url: &str, sync_time: DateTime<Utc>, sync_mode: &str, sync_status: &str, datasets_synced: i32, ) -> Result<(), AppError>

Records a sync status for a portal.

async fn health_check(&self) -> Result<(), AppError>

Checks database connectivity.