Struct Db

pub struct Db {
    pub reader: Pool<Sqlite>,
    pub writer: Pool<Sqlite>,
}

SQLite database implementation for flow trading repositories.

This struct provides separate reader and writer connection pools to a SQLite database, implementing all the repository traits defined in fts-core. The separation of read and write connections allows for better concurrency control and follows SQLite best practices for Write-Ahead Logging (WAL) mode.

Connection Management

• reader: A connection pool for read operations, allowing concurrent reads
• writer: A single-connection pool for write operations, ensuring serialized writes

Example

let config = SqliteConfig::default();
let now = DateTime::from(time::OffsetDateTime::now_utc());
let db = Db::open(&config, now).await?;

Fields

reader: Pool<Sqlite>

Connection pool for read operations

writer: Pool<Sqlite>

Connection pool for write operations (limited to 1 connection)

Implementations

impl Db

pub async fn open(config: &SqliteConfig, as_of: DateTime) -> Result<Self, Error>

Open a connection to the specified SQLite database.

Creates a new database if one doesn’t exist (when create_if_missing is true), applies all pending migrations, and ensures the batch table is initialized.

Arguments

• config - Configuration specifying database path and creation options
• as_of - Initial timestamp for the batch table if creating a new database

Database Configuration

The database is configured with the following settings for optimal performance:

• WAL mode for better concurrency
• Foreign keys enabled for referential integrity
• Optimized cache and memory settings for flow trading workloads

Errors

Returns sqlx::Error if:

• Database connection fails
• Migrations fail to apply
• Initial batch row creation fails

Trait Implementations

impl<T> BatchRepository<T> for Db

where T: Send + Solver<DemandId, PortfolioId, ProductId>, T::Error: Send, T::State: Send, T::PortfolioOutcome: Unpin + Send + Serialize + DeserializeOwned, T::ProductOutcome: Unpin + Send + Serialize + DeserializeOwned,

async fn get_portfolio_outcomes( &self, portfolio_id: Self::PortfolioId, query: DateTimeRangeQuery<Self::DateTime>, limit: usize, ) -> Result<DateTimeRangeResponse<OutcomeRecord<Self::DateTime, T::PortfolioOutcome>, Self::DateTime>, Self::Error>

Get the portfolio’s outcomes

This returns a list of outcomes, each corresponding to a specific point in time. The records are ordered by valid_from in descending order and are grouped by valid_from.

async fn get_product_outcomes( &self, product_id: Self::ProductId, query: DateTimeRangeQuery<Self::DateTime>, limit: usize, ) -> Result<DateTimeRangeResponse<OutcomeRecord<Self::DateTime, T::ProductOutcome>, Self::DateTime>, Self::Error>

Get the product’s outcomes

This returns a list of outcomes, each corresponding to a specific point in time. The records are ordered by valid_from in descending order and are grouped by valid_from.

async fn run_batch( &self, timestamp: Self::DateTime, solver: T, state: T::State, ) -> Result<Result<(), T::Error>, Self::Error>

Execute a batch auction for a specific timestamp.

impl Clone for Db

fn clone(&self) -> Db

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<DemandData: Send + Unpin + Serialize + DeserializeOwned> DemandRepository<DemandData> for Db

async fn get_demand_bidder_id( &self, demand_id: Self::DemandId, ) -> Result<Option<Self::BidderId>, Self::Error>

Get the bidder id associated to the demand

async fn query_demand( &self, bidder_ids: &[Self::BidderId], as_of: Self::DateTime, ) -> Result<Vec<Self::DemandId>, Self::Error>

Query all the demand curves with non-null data associated to any of bidder_ids as-of the specified time.

async fn create_demand( &self, demand_id: Self::DemandId, bidder_id: Self::BidderId, app_data: DemandData, curve_data: Option<DemandCurve>, as_of: Self::DateTime, ) -> Result<(), Self::Error>

Create a new demand with an optional initial curve.

async fn update_demand( &self, demand_id: Self::DemandId, curve_data: Option<DemandCurve>, as_of: Self::DateTime, ) -> Result<bool, Self::Error>

Update the curve data for an existing demand.

async fn get_demand( &self, demand_id: Self::DemandId, as_of: Self::DateTime, ) -> Result<Option<DemandRecord<Self::DateTime, Self::BidderId, Self::DemandId, Self::PortfolioId, DemandData>>, Self::Error>

Retrieve a demand at a specific point in time.

async fn get_demand_history( &self, demand_id: Self::DemandId, query: DateTimeRangeQuery<Self::DateTime>, limit: usize, ) -> Result<DateTimeRangeResponse<ValueRecord<Self::DateTime, DemandCurve>, Self::DateTime>, Self::Error>

Retrieve the history of curve changes for a demand.

impl<PortfolioData: Send + Unpin + Serialize + DeserializeOwned> PortfolioRepository<PortfolioData> for Db

async fn get_portfolio_demand_history( &self, portfolio_id: Self::PortfolioId, query: DateTimeRangeQuery<Self::DateTime>, limit: usize, ) -> Result<DateTimeRangeResponse<ValueRecord<Self::DateTime, DemandGroup<Self::DemandId>>, Self::DateTime>, Self::Error>

Get the history of this portfolio’s demands

This returns a list of records, each containing the state of the portfolio’s demand group at a specific point in time. The records are ordered by valid_from in descending order and are grouped by valid_from. This is important for a more pointer to work correctly, so the demand_group is actually a map of demand_id to weight at that point in time.

async fn get_portfolio_product_history( &self, portfolio_id: Self::PortfolioId, query: DateTimeRangeQuery<Self::DateTime>, limit: usize, ) -> Result<DateTimeRangeResponse<ValueRecord<Self::DateTime, ProductGroup<Self::ProductId>>, Self::DateTime>, Self::Error>

Get the history of this portfolio’s products

This returns a list of records, each containing the state of the portfolio’s product group at a specific point in time. The records are ordered by valid_from in descending order and are grouped by valid_from. This is important for a more pointer to work correctly, so the product_group is actually a map of product_id to weight at that point in time.

async fn get_portfolio_bidder_id( &self, portfolio_id: Self::PortfolioId, ) -> Result<Option<Self::BidderId>, Self::Error>

Get the bidder id associated to the portfolio

async fn query_portfolio( &self, bidder_ids: &[Self::BidderId], as_of: Self::DateTime, ) -> Result<Vec<Self::PortfolioId>, Self::Error>

Query all the portfolios with non-empty groups associated to bidder_id as-of the specified time.

async fn create_portfolio( &self, portfolio_id: Self::PortfolioId, bidder_id: Self::BidderId, app_data: PortfolioData, demand_group: DemandGroup<Self::DemandId>, product_group: ProductGroup<Self::ProductId>, as_of: Self::DateTime, ) -> Result<(), Self::Error>

Create a new portfolio with initial demand and product associations.

async fn update_portfolio( &self, portfolio_id: Self::PortfolioId, demand_group: Option<DemandGroup<Self::DemandId>>, product_group: Option<ProductGroup<Self::ProductId>>, as_of: Self::DateTime, ) -> Result<bool, Self::Error>

Update a portfolio’s demand and/or product associations.

async fn get_portfolio( &self, portfolio_id: Self::PortfolioId, as_of: Self::DateTime, ) -> Result<Option<PortfolioRecord<Self::DateTime, Self::BidderId, Self::PortfolioId, Self::DemandId, Self::ProductId, PortfolioData>>, Self::Error>

Retrieve a portfolio at a specific point in time.

impl<ProductData: Send + Unpin + 'static + Serialize + DeserializeOwned> ProductRepository<ProductData> for Db

async fn create_product( &self, product_id: Self::ProductId, app_data: ProductData, as_of: Self::DateTime, ) -> Result<(), Self::Error>

Define a new root product with no parent.

async fn partition_product<T: Send + IntoIterator<Item = (Self::ProductId, ProductData, f64)>>( &self, product_id: Self::ProductId, children: T, as_of: Self::DateTime, ) -> Result<usize, Self::Error>

where T::IntoIter: Send + ExactSizeIterator,

Partition an existing product into new weighted children.

async fn get_product( &self, product_id: Self::ProductId, _as_of: Self::DateTime, ) -> Result<Option<ProductData>, Self::Error>

Get the data associated with a product at a specific time.

impl Repository for Db

type Error = Error

The error type for underlying operations

type DateTime = DateTime

A type suitable for expressing a timestamp

type BidderId = BidderId

A type representing a bidder id

type ProductId = ProductId

A type representing a product id

type DemandId = DemandId

A type representing a demand id

type PortfolioId = PortfolioId

A type representing a portfolio id