Trait DatasetStore 

pub trait DatasetStore:
    Send
    + Sync
    + Clone {
    // Required methods
    fn get_by_id(
        &self,
        id: Uuid,
    ) -> impl Future<Output = Result<Option<Dataset>, AppError>> + Send;
    fn get_hashes_for_portal(
        &self,
        portal_url: &str,
    ) -> impl Future<Output = Result<HashMap<String, Option<String>>, AppError>> + Send;
    fn update_timestamp_only(
        &self,
        portal_url: &str,
        original_id: &str,
    ) -> impl Future<Output = Result<(), AppError>> + Send;
    fn batch_update_timestamps(
        &self,
        portal_url: &str,
        original_ids: &[String],
    ) -> impl Future<Output = Result<u64, AppError>> + Send;
    fn mark_stale_datasets(
        &self,
        portal_url: &str,
        sync_start: DateTime<Utc>,
    ) -> impl Future<Output = Result<u64, AppError>> + Send;
    fn mark_stale_by_exclusion(
        &self,
        portal_url: &str,
        seen_ids: &[String],
    ) -> impl Future<Output = Result<u64, AppError>> + Send;
    fn upsert(
        &self,
        dataset: &NewDataset,
    ) -> impl Future<Output = Result<Uuid, AppError>> + Send;
    fn batch_upsert(
        &self,
        datasets: &[NewDataset],
    ) -> impl Future<Output = Result<Vec<Uuid>, AppError>> + Send;
    fn search(
        &self,
        query_vector: Vec<f32>,
        limit: usize,
    ) -> impl Future<Output = Result<Vec<SearchResult>, AppError>> + Send;
    fn list_stream<'a>(
        &'a self,
        portal_filter: Option<&'a str>,
        limit: Option<usize>,
    ) -> BoxStream<'a, Result<Dataset, AppError>>;
    fn get_last_sync_time(
        &self,
        portal_url: &str,
    ) -> impl Future<Output = Result<Option<DateTime<Utc>>, AppError>> + Send;
    fn record_sync_status(
        &self,
        portal_url: &str,
        sync_time: DateTime<Utc>,
        sync_mode: &str,
        sync_status: &str,
        datasets_synced: i32,
    ) -> impl Future<Output = Result<(), AppError>> + Send;
    fn get_duplicate_titles(
        &self,
    ) -> impl Future<Output = Result<HashSet<String>, AppError>> + Send;
    fn list_pending_embeddings(
        &self,
        portal_filter: Option<&str>,
        limit: Option<usize>,
    ) -> impl Future<Output = Result<Vec<Dataset>, AppError>> + Send;
    fn count_pending_embeddings(
        &self,
        portal_filter: Option<&str>,
    ) -> impl Future<Output = Result<i64, AppError>> + Send;
    fn health_check(&self) -> impl Future<Output = Result<(), AppError>> + Send;
}

Store for dataset persistence and retrieval.

Implementations handle database operations for datasets.

Required Methods

fn get_by_id( &self, id: Uuid, ) -> impl Future<Output = Result<Option<Dataset>, AppError>> + Send

Retrieves a dataset by its unique ID.

Arguments

• id - The dataset’s UUID

Returns

The dataset if found, or None if not exists.

fn get_hashes_for_portal( &self, portal_url: &str, ) -> impl Future<Output = Result<HashMap<String, Option<String>>, AppError>> + Send

Retrieves content hashes for all datasets from a specific portal.

Used for delta detection to determine which datasets need reprocessing.

Arguments

• portal_url - The source portal URL

Returns

A map from original_id to optional content_hash.

fn update_timestamp_only( &self, portal_url: &str, original_id: &str, ) -> impl Future<Output = Result<(), AppError>> + Send

Updates only the timestamp for an unchanged dataset.

Used when content hash matches but we want to track “last seen” time.

Arguments

• portal_url - The source portal URL
• original_id - The dataset’s original ID from the portal

fn batch_update_timestamps( &self, portal_url: &str, original_ids: &[String], ) -> impl Future<Output = Result<u64, AppError>> + Send

Batch updates timestamps for multiple unchanged datasets.

More efficient than calling update_timestamp_only for each dataset.

Arguments

• portal_url - The source portal URL
• original_ids - Slice of dataset original IDs to update

Returns

The number of rows actually updated.

fn mark_stale_datasets( &self, portal_url: &str, sync_start: DateTime<Utc>, ) -> impl Future<Output = Result<u64, AppError>> + Send

Marks datasets as stale if they were not seen during the latest full sync.

After a successful full sync, any dataset whose last_updated_at is older than sync_start was not present in the portal’s response.

Arguments

• portal_url - The source portal URL
• sync_start - Timestamp recorded at the start of the sync

Returns

The number of datasets newly marked as stale.

fn mark_stale_by_exclusion( &self, portal_url: &str, seen_ids: &[String], ) -> impl Future<Output = Result<u64, AppError>> + Send

Marks datasets as stale if their original_id is NOT in the given set.

This is more efficient than the timestamp-based approach because it avoids updating every unchanged row just to compare timestamps later. Instead, we directly identify stale datasets by exclusion.

Arguments

• portal_url - The source portal URL
• seen_ids - All original_ids seen during the current full sync

Returns

The number of datasets newly marked as stale.

fn upsert( &self, dataset: &NewDataset, ) -> impl Future<Output = Result<Uuid, AppError>> + Send

Inserts or updates a dataset.

Arguments

• dataset - The dataset to upsert

Returns

The UUID of the affected row.

fn batch_upsert( &self, datasets: &[NewDataset], ) -> impl Future<Output = Result<Vec<Uuid>, AppError>> + Send

Batch upserts multiple datasets in a single operation.

Much faster than calling upsert in a loop because it reduces database round-trips and amortizes index update costs.

Arguments

• datasets - Slice of datasets to upsert

Returns

The UUIDs of all affected rows.

fn search( &self, query_vector: Vec<f32>, limit: usize, ) -> impl Future<Output = Result<Vec<SearchResult>, AppError>> + Send

Performs vector similarity search.

Arguments

• query_vector - The embedding vector to search for
• limit - Maximum number of results

Returns

Datasets ranked by similarity score (highest first).

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.

This method returns a stream of datasets for memory-efficient processing of large result sets. Unlike batch methods, it streams results directly from the database without loading everything into memory.

Arguments

• portal_filter - Optional portal URL to filter by
• limit - Optional maximum number of records

fn get_last_sync_time( &self, portal_url: &str, ) -> impl Future<Output = Result<Option<DateTime<Utc>>, AppError>> + Send

Retrieves the last successful sync timestamp for a portal.

Used for incremental harvesting to determine which datasets have been modified since the last sync.

Arguments

• portal_url - The source portal URL

Returns

The timestamp of the last successful sync, or None if never synced.

fn record_sync_status( &self, portal_url: &str, sync_time: DateTime<Utc>, sync_mode: &str, sync_status: &str, datasets_synced: i32, ) -> impl Future<Output = Result<(), AppError>> + Send

Records a sync status for a portal.

Called after a harvest operation to update the sync status. The sync_status parameter indicates the outcome: “completed” or “cancelled”.

Arguments

• portal_url - The source portal URL
• sync_time - The timestamp of this sync
• sync_mode - Either “full” or “incremental” TODO(design): sync_mode/sync_status should be typed enums, not &str
• sync_status - The outcome: “completed” or “cancelled”
• datasets_synced - Number of datasets processed

fn get_duplicate_titles( &self, ) -> impl Future<Output = Result<HashSet<String>, AppError>> + Send

Returns lowercased titles that appear across multiple portals.

Used for cross-portal duplicate detection in Parquet exports. Typically returns ~21k titles (~2MB) for the full dataset.

fn list_pending_embeddings( &self, portal_filter: Option<&str>, limit: Option<usize>, ) -> impl Future<Output = Result<Vec<Dataset>, AppError>> + Send

Lists datasets that have no embedding vector (embedding IS NULL).

Used by crate::EmbeddingService to find datasets needing embedding generation.

Arguments

• portal_filter - Optional portal URL to scope the query
• limit - Maximum number of datasets to return

fn count_pending_embeddings( &self, portal_filter: Option<&str>, ) -> impl Future<Output = Result<i64, AppError>> + Send

Counts datasets with embedding IS NULL.

Used for progress reporting in the embedding service.

Arguments

• portal_filter - Optional portal URL to scope the count

fn health_check(&self) -> impl Future<Output = Result<(), AppError>> + Send

Checks database connectivity.

Performs a simple query to verify the database is reachable and responsive. Used by health check endpoints.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors