nautilus_infrastructure/sql/
queries.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use ahash::AHashMap;
17use nautilus_common::signal::Signal;
18use nautilus_model::{
19    accounts::AccountAny,
20    data::{Bar, CustomData, DataType, HasTsInit, QuoteTick, TradeTick},
21    events::{
22        AccountState, OrderEvent, OrderEventAny, OrderFilled, OrderInitialized, OrderSnapshot,
23        position::snapshot::PositionSnapshot,
24    },
25    identifiers::{AccountId, ClientId, ClientOrderId, InstrumentId, PositionId},
26    instruments::{Instrument, InstrumentAny},
27    orders::OrderAny,
28    position::Position,
29    types::Currency,
30};
31use sqlx::{PgPool, Postgres, Row, Transaction};
32
33use super::models::{orders::OrderSnapshotRow, positions::PositionSnapshotRow, types::SignalRow};
34use crate::sql::models::{
35    accounts::AccountEventRow,
36    data::{BarRow, QuoteTickRow, TradeTickRow},
37    enums::{
38        AggregationSourcePg, AggressorSidePg, AssetClassPg, BarAggregationPg, CurrencyTypePg,
39        PriceTypePg, TrailingOffsetTypePg,
40    },
41    general::{GeneralRow, OrderEventOrderClientIdCombination, OrderPositionIndexRow},
42    instruments::InstrumentAnyRow,
43    orders::{OrderEventAnyRow, OrderFilledRow},
44    types::CurrencyRow,
45};
46
47#[derive(Debug)]
48pub struct DatabaseQueries;
49
50impl DatabaseQueries {
51    /// Truncates all tables in the cache database via the provided Postgres `pool`.
52    ///
53    /// # Errors
54    ///
55    /// Returns an error if the TRUNCATE operation fails.
56    pub async fn truncate(pool: &PgPool) -> anyhow::Result<()> {
57        sqlx::query("SELECT truncate_all_tables()")
58            .execute(pool)
59            .await
60            .map(|_| ())
61            .map_err(|e| anyhow::anyhow!("Failed to truncate tables: {e}"))
62    }
63
64    /// Inserts a raw key-value entry into the `general` table via the provided `pool`.
65    ///
66    /// # Errors
67    ///
68    /// Returns an error if the INSERT operation fails.
69    pub async fn add(pool: &PgPool, key: String, value: Vec<u8>) -> anyhow::Result<()> {
70        sqlx::query("INSERT INTO general (id, value) VALUES ($1, $2)")
71            .bind(key)
72            .bind(value)
73            .execute(pool)
74            .await
75            .map(|_| ())
76            .map_err(|e| anyhow::anyhow!("Failed to insert into general table: {e}"))
77    }
78
79    /// Loads all entries from the `general` table via the provided `pool`.
80    ///
81    /// # Errors
82    ///
83    /// Returns an error if the SELECT operation fails.
84    pub async fn load(pool: &PgPool) -> anyhow::Result<AHashMap<String, Vec<u8>>> {
85        sqlx::query_as::<_, GeneralRow>("SELECT * FROM general")
86            .fetch_all(pool)
87            .await
88            .map(|rows| {
89                let mut cache: AHashMap<String, Vec<u8>> = AHashMap::new();
90                for row in rows {
91                    cache.insert(row.id, row.value);
92                }
93                cache
94            })
95            .map_err(|e| anyhow::anyhow!("Failed to load general table: {e}"))
96    }
97
98    /// Inserts or ignores a `Currency` row via the provided `pool`.
99    ///
100    /// # Errors
101    ///
102    /// Returns an error if the INSERT operation fails.
103    pub async fn add_currency(pool: &PgPool, currency: Currency) -> anyhow::Result<()> {
104        sqlx::query(
105            "INSERT INTO currency (id, precision, iso4217, name, currency_type) VALUES ($1, $2, $3, $4, $5::currency_type) ON CONFLICT (id) DO NOTHING"
106        )
107            .bind(currency.code.as_str())
108            .bind(i32::from(currency.precision))
109            .bind(i32::from(currency.iso4217))
110            .bind(currency.name.as_str())
111            .bind(CurrencyTypePg(currency.currency_type))
112            .execute(pool)
113            .await
114            .map(|_| ())
115            .map_err(|e| anyhow::anyhow!("Failed to insert into currency table: {e}"))
116    }
117
118    /// Loads all `Currency` entries via the provided `pool`.
119    ///
120    /// # Errors
121    ///
122    /// Returns an error if the SELECT operation fails.
123    pub async fn load_currencies(pool: &PgPool) -> anyhow::Result<Vec<Currency>> {
124        sqlx::query_as::<_, CurrencyRow>("SELECT * FROM currency ORDER BY id ASC")
125            .fetch_all(pool)
126            .await
127            .map(|rows| rows.into_iter().map(|row| row.0).collect())
128            .map_err(|e| anyhow::anyhow!("Failed to load currencies: {e}"))
129    }
130
131    /// Loads a single `Currency` entry by `code` via the provided `pool`.
132    ///
133    /// # Errors
134    ///
135    /// Returns an error if the SELECT operation fails.
136    pub async fn load_currency(pool: &PgPool, code: &str) -> anyhow::Result<Option<Currency>> {
137        sqlx::query_as::<_, CurrencyRow>("SELECT * FROM currency WHERE id = $1")
138            .bind(code)
139            .fetch_optional(pool)
140            .await
141            .map(|currency| currency.map(|row| row.0))
142            .map_err(|e| anyhow::anyhow!("Failed to load currency: {e}"))
143    }
144
145    /// Inserts or updates an `InstrumentAny` entry via the provided `pool`.
146    ///
147    /// # Errors
148    ///
149    /// Returns an error if the INSERT or UPDATE operation fails.
150    pub async fn add_instrument(
151        pool: &PgPool,
152        kind: &str,
153        instrument: Box<dyn Instrument>,
154    ) -> anyhow::Result<()> {
155        sqlx::query(r#"
156            INSERT INTO "instrument" (
157                id, kind, raw_symbol, base_currency, underlying, quote_currency, settlement_currency, isin, asset_class, exchange,
158                strategy_type, multiplier, option_kind, is_inverse, strike_price, activation_ns, expiration_ns, price_precision, size_precision,
159                price_increment, size_increment, maker_fee, taker_fee, margin_init, margin_maint, lot_size, max_quantity, min_quantity, max_notional,
160                min_notional, max_price, min_price, ts_init, ts_event, created_at, updated_at
161            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::asset_class, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
162            ON CONFLICT (id)
163            DO UPDATE
164            SET
165                kind = $2, raw_symbol = $3, base_currency= $4, underlying = $5, quote_currency = $6, settlement_currency = $7, isin = $8, asset_class = $9, exchange = $10,
166                 strategy_type = $11, multiplier = $12, option_kind = $13, is_inverse = $14, strike_price = $15, activation_ns = $16, expiration_ns = $17 , price_precision = $18, size_precision = $19,
167                 price_increment = $20, size_increment = $21, maker_fee = $22, taker_fee = $23, margin_init = $24, margin_maint = $25, lot_size = $26, max_quantity = $27,
168                 min_quantity = $28, max_notional = $29, min_notional = $30, max_price = $31, min_price = $32, ts_init = $33,  ts_event = $34, updated_at = CURRENT_TIMESTAMP
169            "#)
170            .bind(instrument.id().to_string())
171            .bind(kind)
172            .bind(instrument.raw_symbol().to_string())
173            .bind(instrument.base_currency().map(|x| x.code.as_str()))
174            .bind(instrument.underlying().map(|x| x.to_string()))
175            .bind(instrument.quote_currency().code.as_str())
176            .bind(instrument.settlement_currency().code.as_str())
177            .bind(instrument.isin().map(|x| x.to_string()))
178            .bind(AssetClassPg(instrument.asset_class()))
179            .bind(instrument.exchange().map(|x| x.to_string()))
180            .bind(instrument.strategy_type().map(|x| x.to_string()))
181            .bind(instrument.multiplier().to_string())
182            .bind(instrument.option_kind().map(|x| x.to_string()))
183            .bind(instrument.is_inverse())
184            .bind(instrument.strike_price().map(|x| x.to_string()))
185            .bind(instrument.activation_ns().map(|x| x.to_string()))
186            .bind(instrument.expiration_ns().map(|x| x.to_string()))
187            .bind(i32::from(instrument.price_precision()))
188            .bind(i32::from(instrument.size_precision()))
189            .bind(instrument.price_increment().to_string())
190            .bind(instrument.size_increment().to_string())
191            .bind(instrument.maker_fee().to_string())
192            .bind(instrument.taker_fee().to_string())
193            .bind(instrument.margin_init().to_string())
194            .bind(instrument.margin_maint().to_string())
195            .bind(instrument.lot_size().map(|x| x.to_string()))
196            .bind(instrument.max_quantity().map(|x| x.to_string()))
197            .bind(instrument.min_quantity().map(|x| x.to_string()))
198            .bind(instrument.max_notional().map(|x| x.to_string()))
199            .bind(instrument.min_notional().map(|x| x.to_string()))
200            .bind(instrument.max_price().map(|x| x.to_string()))
201            .bind(instrument.min_price().map(|x| x.to_string()))
202            .bind(instrument.ts_init().to_string())
203            .bind(instrument.ts_event().to_string())
204            .execute(pool)
205            .await
206            .map(|_| ())
207            .map_err(|e| anyhow::anyhow!("Failed to insert item {} into instrument table: {:?}", instrument.id(), e))
208    }
209
210    /// Loads a single `InstrumentAny` entry by `instrument_id` via the provided `pool`.
211    ///
212    /// # Errors
213    ///
214    /// Returns an error if the SELECT operation fails.
215    pub async fn load_instrument(
216        pool: &PgPool,
217        instrument_id: &InstrumentId,
218    ) -> anyhow::Result<Option<InstrumentAny>> {
219        sqlx::query_as::<_, InstrumentAnyRow>("SELECT * FROM instrument WHERE id = $1")
220            .bind(instrument_id.to_string())
221            .fetch_optional(pool)
222            .await
223            .map(|instrument| instrument.map(|row| row.0))
224            .map_err(|e| {
225                anyhow::anyhow!("Failed to load instrument with id {instrument_id},error is: {e}")
226            })
227    }
228
229    /// Loads all `InstrumentAny` entries via the provided `pool`.
230    ///
231    /// # Errors
232    ///
233    /// Returns an error if the SELECT operation fails.
234    pub async fn load_instruments(pool: &PgPool) -> anyhow::Result<Vec<InstrumentAny>> {
235        sqlx::query_as::<_, InstrumentAnyRow>("SELECT * FROM instrument")
236            .fetch_all(pool)
237            .await
238            .map(|rows| rows.into_iter().map(|row| row.0).collect())
239            .map_err(|e| anyhow::anyhow!("Failed to load instruments: {e}"))
240    }
241
242    /// Inserts an `OrderInitialized` event via the provided `pool`.
243    ///
244    /// # Errors
245    ///
246    /// Returns an error if the SQL INSERT or UPDATE operation fails.
247    pub async fn add_order(
248        pool: &PgPool,
249        event: OrderInitialized,
250        client_id: Option<ClientId>,
251    ) -> anyhow::Result<()> {
252        Self::add_order_event(pool, Box::new(event), client_id).await
253    }
254
255    /// Inserts an `OrderSnapshot` entry via the provided `pool`.
256    ///
257    /// # Errors
258    ///
259    /// Returns an error if the SQL INSERT operation fails.
260    ///
261    /// # Panics
262    ///
263    /// Panics if serialization of `snapshot.exec_algorithm_params` fails.
264    #[expect(
265        clippy::too_many_lines,
266        reason = "order snapshot persistence maps the full database schema in one transaction"
267    )]
268    pub async fn add_order_snapshot(pool: &PgPool, snapshot: OrderSnapshot) -> anyhow::Result<()> {
269        let mut transaction = pool.begin().await?;
270
271        // Insert trader if it does not exist
272        // TODO remove this when node and trader initialization is implemented
273        sqlx::query(
274            r#"
275            INSERT INTO "trader" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
276            "#,
277        )
278        .bind(snapshot.trader_id.to_string())
279        .execute(&mut *transaction)
280        .await
281        .map(|_| ())
282        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
283
284        sqlx::query(
285            r#"
286            INSERT INTO "order" (
287                id, trader_id, strategy_id, instrument_id, client_order_id, venue_order_id, position_id,
288                account_id, last_trade_id, order_type, order_side, quantity, price, trigger_price,
289                trigger_type, limit_offset, trailing_offset, trailing_offset_type, time_in_force,
290                expire_time, filled_qty, liquidity_side, avg_px, slippage, commissions, status,
291                is_post_only, is_reduce_only, is_quote_quantity, display_qty, emulation_trigger,
292                trigger_instrument_id, contingency_type, order_list_id, linked_order_ids,
293                parent_order_id, exec_algorithm_id, exec_algorithm_params, exec_spawn_id, tags, init_id, ts_init, ts_last,
294                activation_price, created_at, updated_at
295            ) VALUES (
296                $1, $2, $3, $4, $1, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16,
297                $17::TRAILING_OFFSET_TYPE, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28,
298                $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43,
299                CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
300            )
301            ON CONFLICT (id)
302            DO UPDATE SET
303                trader_id = $2,
304                strategy_id = $3,
305                instrument_id = $4,
306                venue_order_id = $5,
307                position_id = $6,
308                account_id = $7,
309                last_trade_id = $8,
310                order_type = $9,
311                order_side = $10,
312                quantity = $11,
313                price = $12,
314                trigger_price = $13,
315                trigger_type = $14,
316                limit_offset = $15,
317                trailing_offset = $16,
318                trailing_offset_type = $17::TRAILING_OFFSET_TYPE,
319                time_in_force = $18,
320                expire_time = $19,
321                filled_qty = $20,
322                liquidity_side = $21,
323                avg_px = $22,
324                slippage = $23,
325                commissions = $24,
326                status = $25,
327                is_post_only = $26,
328                is_reduce_only = $27,
329                is_quote_quantity = $28,
330                display_qty = $29,
331                emulation_trigger = $30,
332                trigger_instrument_id = $31,
333                contingency_type = $32,
334                order_list_id = $33,
335                linked_order_ids = $34,
336                parent_order_id = $35,
337                exec_algorithm_id = $36,
338                exec_algorithm_params = $37,
339                exec_spawn_id = $38,
340                tags = $39,
341                init_id = $40,
342                ts_init = $41,
343                ts_last = $42,
344                activation_price = $43,
345                updated_at = CURRENT_TIMESTAMP
346        "#)
347            .bind(snapshot.client_order_id.to_string())  // Used for both id and client_order_id
348            .bind(snapshot.trader_id.to_string())
349            .bind(snapshot.strategy_id.to_string())
350            .bind(snapshot.instrument_id.to_string())
351            .bind(snapshot.venue_order_id.map(|x| x.to_string()))
352            .bind(snapshot.position_id.map(|x| x.to_string()))
353            .bind(snapshot.account_id.map(|x| x.to_string()))
354            .bind(snapshot.last_trade_id.map(|x| x.to_string()))
355            .bind(snapshot.order_type.to_string())
356            .bind(snapshot.order_side.to_string())
357            .bind(snapshot.quantity.to_string())
358            .bind(snapshot.price.map(|x| x.to_string()))
359            .bind(snapshot.trigger_price.map(|x| x.to_string()))
360            .bind(snapshot.trigger_type.map(|x| x.to_string()))
361            .bind(snapshot.limit_offset.map(|x| x.to_string()))
362            .bind(snapshot.trailing_offset.map(|x| x.to_string()))
363            .bind(
364                snapshot
365                    .trailing_offset_type
366                    .map(|value| TrailingOffsetTypePg(Some(value))),
367            )
368            .bind(snapshot.time_in_force.to_string())
369            .bind(snapshot.expire_time.map(|x| x.to_string()))
370            .bind(snapshot.filled_qty.to_string())
371            .bind(snapshot.liquidity_side.map(|x| x.to_string()))
372            .bind(snapshot.avg_px)
373            .bind(snapshot.slippage)
374            .bind(snapshot.commissions.iter().map(ToString::to_string).collect::<Vec<String>>())
375            .bind(snapshot.status.to_string())
376            .bind(snapshot.is_post_only)
377            .bind(snapshot.is_reduce_only)
378            .bind(snapshot.is_quote_quantity)
379            .bind(snapshot.display_qty.map(|x| x.to_string()))
380            .bind(
381                snapshot
382                    .emulation_trigger
383                    .map_or_else(|| "NO_TRIGGER".to_string(), |value| value.to_string()),
384            )
385            .bind(snapshot.trigger_instrument_id.map(|x| x.to_string()))
386            .bind(snapshot.contingency_type.map_or_else(
387                || "NO_CONTINGENCY".to_string(),
388                |value| value.to_string(),
389            ))
390            .bind(snapshot.order_list_id.map(|x| x.to_string()))
391            .bind(snapshot.linked_order_ids.map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
392            .bind(snapshot.parent_order_id.map(|x| x.to_string()))
393            .bind(snapshot.exec_algorithm_id.map(|x| x.to_string()))
394            .bind(snapshot.exec_algorithm_params.map(|x| serde_json::to_value(x).unwrap()))
395            .bind(snapshot.exec_spawn_id.map(|x| x.to_string()))
396            .bind(snapshot.tags.map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
397            .bind(snapshot.init_id.to_string())
398            .bind(snapshot.ts_init.to_string())
399            .bind(snapshot.ts_last.to_string())
400            .bind(snapshot.activation_price.map(|x| x.to_string()))
401            .execute(&mut *transaction)
402            .await
403            .map(|_| ())
404            .map_err(|e| anyhow::anyhow!("Failed to insert into order table: {e}"))?;
405
406        transaction
407            .commit()
408            .await
409            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
410    }
411
412    /// Loads an `OrderSnapshot` entry by client order ID via the provided `pool`.
413    ///
414    /// # Errors
415    ///
416    /// Returns an error if the SQL SELECT or deserialization fails.
417    pub async fn load_order_snapshot(
418        pool: &PgPool,
419        client_order_id: &ClientOrderId,
420    ) -> anyhow::Result<Option<OrderSnapshot>> {
421        sqlx::query_as::<_, OrderSnapshotRow>(r#"SELECT * FROM "order" WHERE client_order_id = $1"#)
422            .bind(client_order_id.to_string())
423            .fetch_optional(pool)
424            .await
425            .map(|row| row.map(|row| row.0))
426            .map_err(|e| anyhow::anyhow!("Failed to load order snapshot: {e}"))
427    }
428
429    /// Inserts or updates a `PositionSnapshot` entry via the provided `pool`.
430    ///
431    /// # Errors
432    ///
433    /// Returns an error if the SQL INSERT or UPDATE operation fails, or if beginning the transaction fails.
434    pub async fn add_position_snapshot(
435        pool: &PgPool,
436        snapshot: PositionSnapshot,
437    ) -> anyhow::Result<()> {
438        let mut transaction = pool.begin().await?;
439
440        // Insert trader if it does not exist
441        // TODO remove this when node and trader initialization is implemented
442        sqlx::query(
443            r#"
444            INSERT INTO "trader" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
445        "#,
446        )
447        .bind(snapshot.trader_id.to_string())
448        .execute(&mut *transaction)
449        .await
450        .map(|_| ())
451        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
452
453        sqlx::query(r#"
454            INSERT INTO "position" (
455                id, trader_id, strategy_id, instrument_id, account_id, opening_order_id, closing_order_id, entry, side, signed_qty, quantity, peak_qty,
456                quote_currency, base_currency, settlement_currency, avg_px_open, avg_px_close, realized_return, realized_pnl, unrealized_pnl, commissions,
457                duration_ns, ts_opened, ts_closed, ts_init, ts_last, replay_state, created_at, updated_at
458            ) VALUES (
459                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
460                $21, $22, $23, $24, $25, $26, $27, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
461            )
462            ON CONFLICT (id)
463            DO UPDATE
464            SET
465                trader_id = $2, strategy_id = $3, instrument_id = $4, account_id = $5, opening_order_id = $6, closing_order_id = $7, entry = $8, side = $9, signed_qty = $10, quantity = $11,
466                peak_qty = $12, quote_currency = $13, base_currency = $14, settlement_currency = $15, avg_px_open = $16, avg_px_close = $17, realized_return = $18, realized_pnl = $19, unrealized_pnl = $20,
467                commissions = $21, duration_ns = $22, ts_opened = $23, ts_closed = $24, ts_init = $25, ts_last = $26,
468                replay_state = $27, updated_at = CURRENT_TIMESTAMP
469        "#)
470            .bind(snapshot.position_id.to_string())
471            .bind(snapshot.trader_id.to_string())
472            .bind(snapshot.strategy_id.to_string())
473            .bind(snapshot.instrument_id.to_string())
474            .bind(snapshot.account_id.to_string())
475            .bind(snapshot.opening_order_id.to_string())
476            .bind(snapshot.closing_order_id.map(|x| x.to_string()))
477            .bind(snapshot.entry.to_string())
478            .bind(snapshot.side.to_string())
479            .bind(snapshot.signed_qty)
480            .bind(snapshot.quantity.to_string())
481            .bind(snapshot.peak_qty.to_string())
482            .bind(snapshot.quote_currency.to_string())
483            .bind(snapshot.base_currency.map(|x| x.to_string()))
484            .bind(snapshot.settlement_currency.to_string())
485            .bind(snapshot.avg_px_open)
486            .bind(snapshot.avg_px_close)
487            .bind(snapshot.realized_return)
488            .bind(snapshot.realized_pnl.map(|x| x.to_string()))
489            .bind(snapshot.unrealized_pnl.map(|x| x.to_string()))
490            .bind(snapshot.commissions.iter().map(ToString::to_string).collect::<Vec<String>>())
491            .bind(snapshot.duration_ns.map(|x| x.to_string()))
492            .bind(snapshot.ts_opened.to_string())
493            .bind(snapshot.ts_closed.map(|x| x.to_string()))
494            .bind(snapshot.ts_init.to_string())
495            .bind(snapshot.ts_last.to_string())
496            .bind(snapshot.replay_state)
497            .execute(&mut *transaction)
498            .await
499            .map(|_| ())
500            .map_err(|e| anyhow::anyhow!("Failed to insert into position table: {e}"))?;
501        transaction
502            .commit()
503            .await
504            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
505    }
506
507    /// Loads a `PositionSnapshot` entry by `position_id` via the provided `pool`.
508    ///
509    /// # Errors
510    ///
511    /// Returns an error if the SQL SELECT or deserialization fails.
512    pub async fn load_position_snapshot(
513        pool: &PgPool,
514        position_id: &PositionId,
515    ) -> anyhow::Result<Option<PositionSnapshot>> {
516        sqlx::query_as::<_, PositionSnapshotRow>(r#"SELECT * FROM "position" WHERE id = $1"#)
517            .bind(position_id.to_string())
518            .fetch_optional(pool)
519            .await
520            .map(|row| row.map(|row| row.0))
521            .map_err(|e| anyhow::anyhow!("Failed to load position snapshot: {e}"))
522    }
523
524    /// Checks if an `OrderInitialized` event exists for the given `client_order_id` via the provided `pool`.
525    ///
526    /// # Errors
527    ///
528    /// Returns an error if the SQL SELECT operation fails.
529    pub async fn check_if_order_initialized_exists(
530        pool: &PgPool,
531        client_order_id: ClientOrderId,
532    ) -> anyhow::Result<bool> {
533        sqlx::query(r#"
534            SELECT EXISTS(SELECT 1 FROM "order_event" WHERE client_order_id = $1 AND kind = 'OrderInitialized')
535        "#)
536            .bind(client_order_id.to_string())
537            .fetch_one(pool)
538            .await
539            .map(|row| row.get(0))
540            .map_err(|e| anyhow::anyhow!("Failed to check if order initialized exists: {e}"))
541    }
542
543    /// Checks if any account event exists for the given `account_id` via the provided `pool`.
544    ///
545    /// # Errors
546    ///
547    /// Returns an error if the SQL SELECT operation fails.
548    pub async fn check_if_account_event_exists(
549        pool: &PgPool,
550        account_id: AccountId,
551    ) -> anyhow::Result<bool> {
552        sqlx::query(
553            r#"
554            SELECT EXISTS(SELECT 1 FROM "account_event" WHERE account_id = $1)
555        "#,
556        )
557        .bind(account_id.to_string())
558        .fetch_one(pool)
559        .await
560        .map(|row| row.get(0))
561        .map_err(|e| anyhow::anyhow!("Failed to check if account event exists: {e}"))
562    }
563
564    /// Inserts or updates an order event entry via the provided `pool`.
565    ///
566    /// # Errors
567    ///
568    /// Returns an error if the SQL INSERT or UPDATE operation fails, or if
569    /// serialization of `exec_algorithm_params` fails.
570    #[expect(
571        clippy::too_many_lines,
572        reason = "order event persistence maps the full database schema in one transaction"
573    )]
574    pub async fn add_order_event(
575        pool: &PgPool,
576        order_event: Box<dyn OrderEvent>,
577        client_id: Option<ClientId>,
578    ) -> anyhow::Result<()> {
579        let mut transaction = pool.begin().await?;
580
581        // Insert trader if it does not exist
582        // TODO remove this when node and trader initialization is implemented
583        sqlx::query(
584            r#"
585            INSERT INTO "trader" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
586        "#,
587        )
588        .bind(order_event.trader_id().to_string())
589        .execute(&mut *transaction)
590        .await
591        .map(|_| ())
592        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
593
594        // Insert client if it does not exist
595        // TODO remove this when client initialization is implemented
596        if let Some(client_id) = client_id {
597            sqlx::query(
598                r#"
599                INSERT INTO "client" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
600            "#,
601            )
602            .bind(client_id.to_string())
603            .execute(&mut *transaction)
604            .await
605            .map(|_| ())
606            .map_err(|e| anyhow::anyhow!("Failed to insert into client table: {e}"))?;
607        }
608
609        let exec_algorithm_params = order_event
610            .exec_algorithm_params()
611            .map(serde_json::to_value)
612            .transpose()
613            .map_err(|e| anyhow::anyhow!("Failed to serialize exec algorithm params: {e}"))?;
614
615        sqlx::query(r#"
616            INSERT INTO "order_event" (
617                id, kind, client_order_id, order_type, order_side, trader_id, client_id, reason, strategy_id, instrument_id, trade_id, currency, quantity, time_in_force, liquidity_side,
618                post_only, reduce_only, quote_quantity, reconciliation, price, last_px, last_qty, trigger_price, trigger_type, limit_offset, trailing_offset,
619                trailing_offset_type, expire_time, display_qty, emulation_trigger, trigger_instrument_id, contingency_type,
620                order_list_id, linked_order_ids, parent_order_id,
621                exec_algorithm_id, exec_spawn_id, venue_order_id, account_id, position_id, commission, ts_event, ts_init, activation_price, exec_algorithm_params, tags, created_at, updated_at
622            ) VALUES (
623                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
624                $21, $22, $23, $24, $25, $26::trailing_offset_type, $27, $28, $29, $30, $31, $32, $33, $34,
625                $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
626            )
627            ON CONFLICT (id)
628            DO UPDATE
629            SET
630                kind = $2, client_order_id = $3, order_type = $4, order_side=$5, trader_id = $6, client_id = $7, reason = $8, strategy_id = $9, instrument_id = $10, trade_id = $11, currency = $12,
631                quantity = $13, time_in_force = $14, liquidity_side = $15, post_only = $16, reduce_only = $17, quote_quantity = $18, reconciliation = $19, price = $20, last_px = $21,
632                last_qty = $22, trigger_price = $23, trigger_type = $24, limit_offset = $25, trailing_offset = $26, trailing_offset_type = $27, expire_time = $28, display_qty = $29,
633                emulation_trigger = $30, trigger_instrument_id = $31, contingency_type = $32, order_list_id = $33, linked_order_ids = $34, parent_order_id = $35, exec_algorithm_id = $36,
634                exec_spawn_id = $37, venue_order_id = $38, account_id = $39, position_id = $40, commission = $41, ts_event = $42, ts_init = $43, activation_price = $44,
635                exec_algorithm_params = $45, tags = $46, updated_at = CURRENT_TIMESTAMP
636
637        "#)
638            .bind(order_event.id().to_string())
639            .bind(order_event.type_name())
640            .bind(order_event.client_order_id().to_string())
641            .bind(order_event.order_type().map(|x| x.to_string()))
642            .bind(order_event.order_side().map(|x| x.to_string()))
643            .bind(order_event.trader_id().to_string())
644            .bind(client_id.map(|x| x.to_string()))
645            .bind(order_event.reason().map(|x| x.to_string()))
646            .bind(order_event.strategy_id().to_string())
647            .bind(order_event.instrument_id().to_string())
648            .bind(order_event.trade_id().map(|x| x.to_string()))
649            .bind(order_event.currency().map(|x| x.code.as_str()))
650            .bind(order_event.quantity().map(|x| x.to_string()))
651            .bind(order_event.time_in_force().map(|x| x.to_string()))
652            .bind(order_event.liquidity_side().map(|x| x.to_string()))
653            .bind(order_event.post_only())
654            .bind(order_event.reduce_only())
655            .bind(order_event.quote_quantity())
656            .bind(order_event.reconciliation())
657            .bind(order_event.price().map(|x| x.to_string()))
658            .bind(order_event.last_px().map(|x| x.to_string()))
659            .bind(order_event.last_qty().map(|x| x.to_string()))
660            .bind(order_event.trigger_price().map(|x| x.to_string()))
661            .bind(order_event.trigger_type().map(|x| x.to_string()))
662            .bind(order_event.limit_offset().map(|x| x.to_string()))
663            .bind(order_event.trailing_offset().map(|x| x.to_string()))
664            .bind(
665                order_event
666                    .trailing_offset_type()
667                    .map(|value| TrailingOffsetTypePg(Some(value))),
668            )
669            .bind(order_event.expire_time().map(|x| x.to_string()))
670            .bind(order_event.display_qty().map(|x| x.to_string()))
671            .bind(
672                order_event
673                    .emulation_trigger()
674                    .map_or_else(|| "NO_TRIGGER".to_string(), |value| value.to_string()),
675            )
676            .bind(order_event.trigger_instrument_id().map(|x| x.to_string()))
677            .bind(order_event.contingency_type().map_or_else(
678                || "NO_CONTINGENCY".to_string(),
679                |value| value.to_string(),
680            ))
681            .bind(order_event.order_list_id().map(|x| x.to_string()))
682            .bind(order_event.linked_order_ids().map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
683            .bind(order_event.parent_order_id().map(|x| x.to_string()))
684            .bind(order_event.exec_algorithm_id().map(|x| x.to_string()))
685            .bind(order_event.exec_spawn_id().map(|x| x.to_string()))
686            .bind(order_event.venue_order_id().map(|x| x.to_string()))
687            .bind(order_event.account_id().map(|x| x.to_string()))
688            .bind(order_event.position_id().map(|x| x.to_string()))
689            .bind(order_event.commission().map(|x| x.to_string()))
690            .bind(order_event.ts_event().to_string())
691            .bind(order_event.ts_init().to_string())
692            .bind(order_event.activation_price().map(|x| x.to_string()))
693            .bind(exec_algorithm_params)
694            .bind(order_event.tags().map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
695            .execute(&mut *transaction)
696            .await
697            .map(|_| ())
698            .map_err(|e| anyhow::anyhow!("Failed to insert into order_event table: {e}"))?;
699        transaction
700            .commit()
701            .await
702            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
703    }
704
705    /// Loads all order events for a `client_order_id` via the provided `pool`.
706    ///
707    /// # Errors
708    ///
709    /// Returns an error if the SQL SELECT or deserialization fails.
710    pub async fn load_order_events(
711        pool: &PgPool,
712        client_order_id: &ClientOrderId,
713    ) -> anyhow::Result<Vec<OrderEventAny>> {
714        sqlx::query_as::<_, OrderEventAnyRow>(r#"SELECT * FROM "order_event" event WHERE event.client_order_id = $1 ORDER BY created_at ASC"#)
715        .bind(client_order_id.to_string())
716        .fetch_all(pool)
717        .await
718        .map(|rows| rows.into_iter().map(|row| row.0).collect())
719        .map_err(|e| anyhow::anyhow!("Failed to load order events: {e}"))
720    }
721
722    /// Loads and assembles a complete `OrderAny` for a `client_order_id` via the provided `pool`.
723    ///
724    /// # Errors
725    ///
726    /// Returns an error if assembling events or SQL operations fail.
727    pub async fn load_order(
728        pool: &PgPool,
729        client_order_id: &ClientOrderId,
730    ) -> anyhow::Result<Option<OrderAny>> {
731        let order_events = Self::load_order_events(pool, client_order_id).await;
732
733        match order_events {
734            Ok(order_events) => {
735                if order_events.is_empty() {
736                    return Ok(None);
737                }
738                let order = OrderAny::from_events(order_events).map_err(|e| {
739                    anyhow::anyhow!("Failed to assemble order {client_order_id} from events: {e}")
740                })?;
741                Ok(Some(order))
742            }
743            Err(e) => anyhow::bail!("Failed to load order events: {e}"),
744        }
745    }
746
747    /// Loads and assembles all `OrderAny` entries via the provided `pool`.
748    ///
749    /// # Errors
750    ///
751    /// Returns an error if loading events or SQL operations fail.
752    pub async fn load_orders(pool: &PgPool) -> anyhow::Result<Vec<OrderAny>> {
753        let mut orders: Vec<OrderAny> = Vec::new();
754        let client_order_ids: Vec<ClientOrderId> = sqlx::query(
755            r#"
756            SELECT DISTINCT client_order_id FROM "order_event"
757        "#,
758        )
759        .fetch_all(pool)
760        .await
761        .map(|rows| {
762            rows.into_iter()
763                .map(|row| ClientOrderId::from(row.get::<&str, _>(0)))
764                .collect()
765        })
766        .map_err(|e| anyhow::anyhow!("Failed to load order ids: {e}"))?;
767        for id in client_order_ids {
768            let order = Self::load_order(pool, &id).await?;
769            if let Some(order) = order {
770                orders.push(order);
771            }
772        }
773        Ok(orders)
774    }
775
776    /// Replaces the fill event log for a `position_id` via the provided `pool`.
777    ///
778    /// # Errors
779    ///
780    /// Returns an error if the fill is invalid or if the SQL operations fail.
781    pub async fn add_position(
782        pool: &PgPool,
783        position_id: PositionId,
784        event: &OrderFilled,
785    ) -> anyhow::Result<()> {
786        let event_position_id = Self::event_position_id(event)?;
787        if event_position_id != position_id {
788            anyhow::bail!(
789                "Cannot persist position event {} for mismatched position_id: expected {}, was {}",
790                event.event_id,
791                position_id,
792                event_position_id
793            );
794        }
795
796        let mut transaction = pool.begin().await?;
797
798        sqlx::query(r#"DELETE FROM "position_event" WHERE position_id = $1"#)
799            .bind(position_id.to_string())
800            .execute(&mut *transaction)
801            .await
802            .map(|_| ())
803            .map_err(|e| anyhow::anyhow!("Failed to delete position_event rows: {e}"))?;
804
805        Self::insert_position_event(&mut transaction, event).await?;
806        transaction
807            .commit()
808            .await
809            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
810    }
811
812    /// Appends a fill event for a `Position` via the provided `pool`.
813    ///
814    /// # Errors
815    ///
816    /// Returns an error if the fill is invalid or if the SQL operations fail.
817    pub async fn update_position(pool: &PgPool, event: &OrderFilled) -> anyhow::Result<()> {
818        let mut transaction = pool.begin().await?;
819
820        Self::insert_position_event(&mut transaction, event).await?;
821        transaction
822            .commit()
823            .await
824            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
825    }
826
827    /// Appends an `OrderFilled` event to the position event log via the provided `pool`.
828    ///
829    /// # Errors
830    ///
831    /// Returns an error if the SQL INSERT operation fails.
832    pub async fn add_position_event(pool: &PgPool, event: &OrderFilled) -> anyhow::Result<()> {
833        let mut transaction = pool.begin().await?;
834
835        Self::insert_position_event(&mut transaction, event).await?;
836        transaction
837            .commit()
838            .await
839            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
840    }
841
842    /// Loads all fill events for a `position_id` via the provided `pool`.
843    ///
844    /// # Errors
845    ///
846    /// Returns an error if the SQL SELECT or deserialization fails.
847    pub async fn load_position_events(
848        pool: &PgPool,
849        position_id: &PositionId,
850    ) -> anyhow::Result<Vec<OrderFilled>> {
851        sqlx::query_as::<_, OrderFilledRow>(
852            r#"
853            SELECT *
854            FROM "position_event"
855            WHERE position_id = $1
856            ORDER BY event_sequence ASC
857        "#,
858        )
859        .bind(position_id.to_string())
860        .fetch_all(pool)
861        .await
862        .map(|rows| rows.into_iter().map(|row| row.0).collect())
863        .map_err(|e| anyhow::anyhow!("Failed to load position events: {e}"))
864    }
865
866    /// Loads and replays a complete `Position` for a `position_id` via the provided `pool`.
867    ///
868    /// # Errors
869    ///
870    /// Returns an error if loading events, loading instruments, or replaying fills fails.
871    pub async fn load_position(
872        pool: &PgPool,
873        position_id: &PositionId,
874    ) -> anyhow::Result<Option<Position>> {
875        if let Some(snapshot) = Self::load_position_snapshot(pool, position_id).await?
876            && let Some(replay_state) = snapshot.replay_state
877        {
878            return serde_json::from_value(replay_state)
879                .map(Some)
880                .map_err(|e| anyhow::anyhow!("Failed to decode position replay state: {e}"));
881        }
882
883        let fills = Self::load_position_events(pool, position_id).await?;
884        let Some((first_fill, remaining_fills)) = fills.split_first() else {
885            return Ok(None);
886        };
887        let Some(instrument) = Self::load_instrument(pool, &first_fill.instrument_id).await? else {
888            log::error!(
889                "Instrument not found for position {position_id}: {}",
890                first_fill.instrument_id
891            );
892            return Ok(None);
893        };
894
895        let mut position = Position::new(&instrument, first_fill.clone());
896        for fill in remaining_fills {
897            if position.trade_ids().contains(&fill.trade_id) {
898                anyhow::bail!(
899                    "Duplicate fill event for position {position_id}: {}",
900                    fill.trade_id
901                );
902            }
903            position.apply(fill);
904        }
905
906        Ok(Some(position))
907    }
908
909    /// Loads and replays all `Position` entries via the provided `pool`.
910    ///
911    /// # Errors
912    ///
913    /// Returns an error if loading position IDs or replaying any position fails.
914    pub async fn load_positions(pool: &PgPool) -> anyhow::Result<Vec<Position>> {
915        let position_ids: Vec<PositionId> = sqlx::query(
916            r#"
917            SELECT DISTINCT position_id
918            FROM "position_event"
919            ORDER BY position_id ASC
920        "#,
921        )
922        .fetch_all(pool)
923        .await
924        .map(|rows| {
925            rows.into_iter()
926                .map(|row| PositionId::from(row.get::<&str, _>(0)))
927                .collect()
928        })
929        .map_err(|e| anyhow::anyhow!("Failed to load position ids: {e}"))?;
930
931        let mut positions = Vec::new();
932
933        for id in position_ids {
934            match Self::load_position(pool, &id).await {
935                Ok(Some(position)) => positions.push(position),
936                Ok(None) => log::error!("Position not found: {id}"),
937                Err(e) => log::error!("Failed to load position {id}: {e}"),
938            }
939        }
940
941        Ok(positions)
942    }
943
944    async fn insert_position_event(
945        transaction: &mut Transaction<'_, Postgres>,
946        event: &OrderFilled,
947    ) -> anyhow::Result<()> {
948        let position_id = Self::event_position_id(event)?;
949
950        sqlx::query(
951            r#"
952            INSERT INTO "trader" (id)
953            VALUES ($1)
954            ON CONFLICT (id) DO NOTHING
955        "#,
956        )
957        .bind(event.trader_id.to_string())
958        .execute(&mut **transaction)
959        .await
960        .map(|_| ())
961        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
962
963        sqlx::query(
964            r#"
965            INSERT INTO "position_event" (
966                id, kind, trader_id, strategy_id, instrument_id, client_order_id, venue_order_id,
967                account_id, trade_id, currency, order_type, order_side, last_px, last_qty,
968                liquidity_side, position_id, commission, ts_event, ts_init, created_at, updated_at
969            ) VALUES (
970                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17,
971                $18, $19, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
972            )
973        "#,
974        )
975        .bind(event.event_id.to_string())
976        .bind("OrderFilled")
977        .bind(event.trader_id.to_string())
978        .bind(event.strategy_id.to_string())
979        .bind(event.instrument_id.to_string())
980        .bind(event.client_order_id.to_string())
981        .bind(event.venue_order_id.to_string())
982        .bind(event.account_id.to_string())
983        .bind(event.trade_id.to_string())
984        .bind(event.currency.code.as_str())
985        .bind(event.order_type.to_string())
986        .bind(event.order_side.to_string())
987        .bind(event.last_px.to_string())
988        .bind(event.last_qty.to_string())
989        .bind(event.liquidity_side.to_string())
990        .bind(position_id.to_string())
991        .bind(event.commission.map(|commission| commission.to_string()))
992        .bind(event.ts_event.to_string())
993        .bind(event.ts_init.to_string())
994        .execute(&mut **transaction)
995        .await
996        .map(|_| ())
997        .map_err(|e| anyhow::anyhow!("Failed to insert into position_event table: {e}"))
998    }
999
1000    fn event_position_id(event: &OrderFilled) -> anyhow::Result<PositionId> {
1001        event.position_id.ok_or_else(|| {
1002            anyhow::anyhow!(
1003                "Cannot persist position event with no position_id: {}",
1004                event.event_id
1005            )
1006        })
1007    }
1008
1009    /// Inserts or updates an `AccountState` event via the provided `pool`.
1010    ///
1011    /// # Errors
1012    ///
1013    /// Returns an error if the SQL INSERT or UPDATE operation fails.
1014    pub async fn add_account(
1015        pool: &PgPool,
1016        updated: bool,
1017        account_event: AccountState,
1018    ) -> anyhow::Result<()> {
1019        if updated {
1020            let exists =
1021                Self::check_if_account_event_exists(pool, account_event.account_id).await?;
1022
1023            if !exists {
1024                anyhow::bail!(
1025                    "Account event does not exist for account: {}",
1026                    account_event.account_id
1027                );
1028            }
1029        }
1030
1031        let mut transaction = pool.begin().await?;
1032        let event = serde_json::to_value(&account_event)
1033            .map_err(|e| anyhow::anyhow!("Failed to serialize account event: {e}"))?;
1034        let balances = event
1035            .get("balances")
1036            .cloned()
1037            .ok_or_else(|| anyhow::anyhow!("Serialized account event has no balances"))?;
1038        let margins = event
1039            .get("margins")
1040            .cloned()
1041            .ok_or_else(|| anyhow::anyhow!("Serialized account event has no margins"))?;
1042
1043        sqlx::query(
1044            r#"
1045            INSERT INTO "account" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
1046        "#,
1047        )
1048        .bind(account_event.account_id.to_string())
1049        .execute(&mut *transaction)
1050        .await
1051        .map(|_| ())
1052        .map_err(|e| anyhow::anyhow!("Failed to insert into account table: {e}"))?;
1053
1054        sqlx::query(r#"
1055            INSERT INTO "account_event" (
1056                id, kind, account_id, base_currency, balances, margins, is_reported, ts_event, ts_init, created_at, updated_at
1057            ) VALUES (
1058                $1, $2, $3, $4, $5, $6, $7, $8, $9, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1059            )
1060            ON CONFLICT (id)
1061            DO UPDATE
1062            SET
1063                kind = $2, account_id = $3, base_currency = $4, balances = $5, margins = $6, is_reported = $7,
1064                ts_event = $8, ts_init = $9, updated_at = CURRENT_TIMESTAMP
1065        "#)
1066            .bind(account_event.event_id.to_string())
1067            .bind(account_event.account_type.to_string())
1068            .bind(account_event.account_id.to_string())
1069            .bind(account_event.base_currency.map(|x| x.code.as_str()))
1070            .bind(balances)
1071            .bind(margins)
1072            .bind(account_event.is_reported)
1073            .bind(account_event.ts_event.to_string())
1074            .bind(account_event.ts_init.to_string())
1075            .execute(&mut *transaction)
1076            .await
1077            .map(|_| ())
1078            .map_err(|e| anyhow::anyhow!("Failed to insert into account_event table: {e}"))?;
1079        transaction
1080            .commit()
1081            .await
1082            .map_err(|e| anyhow::anyhow!("Failed to commit add_account transaction: {e}"))
1083    }
1084
1085    /// Loads all account events for `account_id` via the provided `pool`.
1086    ///
1087    /// # Errors
1088    ///
1089    /// Returns an error if the SQL SELECT or deserialization fails.
1090    pub async fn load_account_events(
1091        pool: &PgPool,
1092        account_id: &AccountId,
1093    ) -> anyhow::Result<Vec<AccountState>> {
1094        sqlx::query_as::<_, AccountEventRow>(
1095            r#"SELECT * FROM "account_event" WHERE account_id = $1 ORDER BY created_at ASC"#,
1096        )
1097        .bind(account_id.to_string())
1098        .fetch_all(pool)
1099        .await
1100        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1101        .map_err(|e| anyhow::anyhow!("Failed to load account events: {e}"))
1102    }
1103
1104    /// Loads and assembles a complete `AccountAny` for `account_id` via the provided `pool`.
1105    ///
1106    /// # Errors
1107    ///
1108    /// Returns an error if assembling events or SQL operations fail.
1109    pub async fn load_account(
1110        pool: &PgPool,
1111        account_id: &AccountId,
1112    ) -> anyhow::Result<Option<AccountAny>> {
1113        let account_events = Self::load_account_events(pool, account_id).await;
1114        match account_events {
1115            Ok(account_events) => {
1116                if account_events.is_empty() {
1117                    return Ok(None);
1118                }
1119                let account = AccountAny::from_events(&account_events).map_err(|e| {
1120                    anyhow::anyhow!("Failed to assemble account {account_id} from events: {e}")
1121                })?;
1122                Ok(Some(account))
1123            }
1124            Err(e) => anyhow::bail!("Failed to load account events: {e}"),
1125        }
1126    }
1127
1128    /// Loads and assembles all `AccountAny` entries via the provided `pool`.
1129    ///
1130    /// # Errors
1131    ///
1132    /// Returns an error if loading events or SQL operations fail.
1133    pub async fn load_accounts(pool: &PgPool) -> anyhow::Result<Vec<AccountAny>> {
1134        let mut accounts: Vec<AccountAny> = Vec::new();
1135        let account_ids: Vec<AccountId> = sqlx::query(
1136            r#"
1137            SELECT DISTINCT account_id FROM "account_event"
1138        "#,
1139        )
1140        .fetch_all(pool)
1141        .await
1142        .map(|rows| {
1143            rows.into_iter()
1144                .map(|row| AccountId::from(row.get::<&str, _>(0)))
1145                .collect()
1146        })
1147        .map_err(|e| anyhow::anyhow!("Failed to load account ids: {e}"))?;
1148        for id in account_ids {
1149            let account = Self::load_account(pool, &id).await?;
1150            if let Some(account) = account {
1151                accounts.push(account);
1152            }
1153        }
1154        Ok(accounts)
1155    }
1156
1157    /// Inserts a `TradeTick` entry via the provided `pool`.
1158    ///
1159    /// # Errors
1160    ///
1161    /// Returns an error if the SQL INSERT operation fails.
1162    pub async fn add_trade(pool: &PgPool, trade: &TradeTick) -> anyhow::Result<()> {
1163        sqlx::query(r#"
1164            INSERT INTO "trade" (
1165                instrument_id, price, quantity, aggressor_side, venue_trade_id,
1166                ts_event, ts_init, created_at, updated_at
1167            ) VALUES (
1168                $1, $2, $3, $4::aggressor_side, $5, $6, $7, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1169            )
1170            ON CONFLICT (id)
1171            DO UPDATE
1172            SET
1173                instrument_id = $1, price = $2, quantity = $3, aggressor_side = $4, venue_trade_id = $5,
1174                ts_event = $6, ts_init = $7, updated_at = CURRENT_TIMESTAMP
1175        "#)
1176            .bind(trade.instrument_id.to_string())
1177            .bind(trade.price.to_string())
1178            .bind(trade.size.to_string())
1179            .bind(AggressorSidePg(trade.aggressor_side))
1180            .bind(trade.trade_id.to_string())
1181            .bind(trade.ts_event.to_string())
1182            .bind(trade.ts_init.to_string())
1183            .execute(pool)
1184            .await
1185            .map(|_| ())
1186            .map_err(|e| anyhow::anyhow!("Failed to insert into trade table: {e}"))
1187    }
1188
1189    /// Loads all `TradeTick` entries for `instrument_id` via the provided `pool`.
1190    ///
1191    /// # Errors
1192    ///
1193    /// Returns an error if the SQL SELECT or deserialization fails.
1194    pub async fn load_trades(
1195        pool: &PgPool,
1196        instrument_id: &InstrumentId,
1197    ) -> anyhow::Result<Vec<TradeTick>> {
1198        sqlx::query_as::<_, TradeTickRow>(
1199            r#"SELECT * FROM "trade" WHERE instrument_id = $1 ORDER BY ts_event ASC"#,
1200        )
1201        .bind(instrument_id.to_string())
1202        .fetch_all(pool)
1203        .await
1204        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1205        .map_err(|e| anyhow::anyhow!("Failed to load trades: {e}"))
1206    }
1207
1208    /// Inserts a `QuoteTick` entry via the provided `pool`.
1209    ///
1210    /// # Errors
1211    ///
1212    /// Returns an error if the SQL INSERT operation fails.
1213    pub async fn add_quote(pool: &PgPool, quote: &QuoteTick) -> anyhow::Result<()> {
1214        sqlx::query(r#"
1215            INSERT INTO "quote" (
1216                instrument_id, bid_price, ask_price, bid_size, ask_size, ts_event, ts_init, created_at, updated_at
1217            ) VALUES (
1218                $1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1219            )
1220            ON CONFLICT (id)
1221            DO UPDATE
1222            SET
1223                instrument_id = $1, bid_price = $2, ask_price = $3, bid_size = $4, ask_size = $5,
1224                ts_event = $6, ts_init = $7, updated_at = CURRENT_TIMESTAMP
1225        "#)
1226            .bind(quote.instrument_id.to_string())
1227            .bind(quote.bid_price.to_string())
1228            .bind(quote.ask_price.to_string())
1229            .bind(quote.bid_size.to_string())
1230            .bind(quote.ask_size.to_string())
1231            .bind(quote.ts_event.to_string())
1232            .bind(quote.ts_init.to_string())
1233            .execute(pool)
1234            .await
1235            .map(|_| ())
1236            .map_err(|e| anyhow::anyhow!("Failed to insert into quote table: {e}"))
1237    }
1238
1239    /// Loads all `QuoteTick` entries for `instrument_id` via the provided `pool`.
1240    ///
1241    /// # Errors
1242    ///
1243    /// Returns an error if the SQL SELECT or deserialization fails.
1244    pub async fn load_quotes(
1245        pool: &PgPool,
1246        instrument_id: &InstrumentId,
1247    ) -> anyhow::Result<Vec<QuoteTick>> {
1248        sqlx::query_as::<_, QuoteTickRow>(
1249            r#"SELECT * FROM "quote" WHERE instrument_id = $1 ORDER BY ts_event ASC"#,
1250        )
1251        .bind(instrument_id.to_string())
1252        .fetch_all(pool)
1253        .await
1254        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1255        .map_err(|e| anyhow::anyhow!("Failed to load quotes: {e}"))
1256    }
1257
1258    /// Inserts a `Bar` entry via the provided `pool`.
1259    ///
1260    /// # Errors
1261    ///
1262    /// Returns an error if the SQL INSERT operation fails.
1263    pub async fn add_bar(pool: &PgPool, bar: &Bar) -> anyhow::Result<()> {
1264        if bar.bar_type.is_composite() {
1265            anyhow::bail!(
1266                "Cannot persist bar with composite bar type {}: the bar table stores only \
1267                 the standard form; standardize the bar type before persisting",
1268                bar.bar_type,
1269            );
1270        }
1271
1272        let bar_step = i32::try_from(bar.bar_type.spec().step.get())
1273            .map_err(|e| anyhow::anyhow!("invalid bar step: {e}"))?;
1274
1275        sqlx::query(r#"
1276            INSERT INTO "bar" (
1277                instrument_id, step, bar_aggregation, price_type, aggregation_source, open, high, low, close, volume, ts_event, ts_init, created_at, updated_at
1278            ) VALUES (
1279                $1, $2, $3::bar_aggregation, $4::price_type, $5::aggregation_source, $6, $7, $8, $9, $10, $11, $12, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1280            )
1281            ON CONFLICT (id)
1282            DO UPDATE
1283            SET
1284                instrument_id = $1, step = $2, bar_aggregation = $3::bar_aggregation, price_type = $4::price_type, aggregation_source = $5::aggregation_source,
1285                open = $6, high = $7, low = $8, close = $9, volume = $10, ts_event = $11, ts_init = $12, updated_at = CURRENT_TIMESTAMP
1286        "#)
1287            .bind(bar.bar_type.instrument_id().to_string())
1288            .bind(bar_step)
1289            .bind(BarAggregationPg(bar.bar_type.spec().aggregation))
1290            .bind(PriceTypePg(bar.bar_type.spec().price_type))
1291            .bind(AggregationSourcePg(bar.bar_type.aggregation_source()))
1292            .bind(bar.open.to_string())
1293            .bind(bar.high.to_string())
1294            .bind(bar.low.to_string())
1295            .bind(bar.close.to_string())
1296            .bind(bar.volume.to_string())
1297            .bind(bar.ts_event.to_string())
1298            .bind(bar.ts_init.to_string())
1299            .execute(pool)
1300            .await
1301            .map(|_| ())
1302            .map_err(|e| anyhow::anyhow!("Failed to insert into bar table: {e}"))
1303    }
1304
1305    /// Loads all `Bar` entries for `instrument_id` via the provided `pool`.
1306    ///
1307    /// # Errors
1308    ///
1309    /// Returns an error if the SQL SELECT or deserialization fails.
1310    pub async fn load_bars(
1311        pool: &PgPool,
1312        instrument_id: &InstrumentId,
1313    ) -> anyhow::Result<Vec<Bar>> {
1314        sqlx::query_as::<_, BarRow>(
1315            r#"SELECT * FROM "bar" WHERE instrument_id = $1 ORDER BY ts_event ASC"#,
1316        )
1317        .bind(instrument_id.to_string())
1318        .fetch_all(pool)
1319        .await
1320        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1321        .map_err(|e| anyhow::anyhow!("Failed to load bars: {e}"))
1322    }
1323
1324    /// Loads all distinct client order IDs from order events via the provided `pool`.
1325    ///
1326    /// # Errors
1327    ///
1328    /// Returns an error if the SQL SELECT or iteration fails.
1329    pub async fn load_distinct_order_event_client_ids(
1330        pool: &PgPool,
1331    ) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
1332        let mut map: AHashMap<ClientOrderId, ClientId> = AHashMap::new();
1333        let result = sqlx::query_as::<_, OrderEventOrderClientIdCombination>(
1334            r#"
1335            SELECT DISTINCT ON (client_order_id)
1336                client_order_id AS "client_order_id",
1337                client_id AS "client_id"
1338            FROM "order_event"
1339            WHERE client_id IS NOT NULL
1340            ORDER BY client_order_id, created_at DESC
1341        "#,
1342        )
1343        .fetch_all(pool)
1344        .await
1345        .map_err(|e| anyhow::anyhow!("Failed to load account ids: {e}"))?;
1346
1347        for id in result {
1348            map.insert(id.client_order_id, id.client_id);
1349        }
1350        Ok(map)
1351    }
1352
1353    /// Claims execution-client origins for existing order events in one transaction.
1354    ///
1355    /// # Errors
1356    ///
1357    /// Returns an error if an order has no persisted events, an order is already claimed by a
1358    /// different client, or any SQL operation fails. Any error rolls back the complete batch.
1359    pub async fn index_order_clients(
1360        pool: &PgPool,
1361        claims: &[(ClientOrderId, ClientId)],
1362    ) -> anyhow::Result<()> {
1363        if claims.is_empty() {
1364            return Ok(());
1365        }
1366
1367        let mut transaction = pool.begin().await?;
1368
1369        for (client_order_id, client_id) in claims {
1370            let conflicting_client_id = sqlx::query_scalar::<_, String>(
1371                r#"
1372                SELECT client_id
1373                FROM "order_event"
1374                WHERE client_order_id = $1
1375                  AND client_id IS NOT NULL
1376                  AND client_id <> $2
1377                LIMIT 1
1378            "#,
1379            )
1380            .bind(client_order_id.to_string())
1381            .bind(client_id.to_string())
1382            .fetch_optional(&mut *transaction)
1383            .await
1384            .map_err(|e| anyhow::anyhow!("Failed to validate order client origin: {e}"))?;
1385
1386            if let Some(conflicting_client_id) = conflicting_client_id {
1387                anyhow::bail!(
1388                    "Order {client_order_id} is already claimed by execution client \
1389                     {conflicting_client_id} and cannot be claimed by {client_id}"
1390                );
1391            }
1392
1393            sqlx::query(
1394                r#"
1395                INSERT INTO "client" (id)
1396                VALUES ($1)
1397                ON CONFLICT (id) DO NOTHING
1398            "#,
1399            )
1400            .bind(client_id.to_string())
1401            .execute(&mut *transaction)
1402            .await
1403            .map_err(|e| anyhow::anyhow!("Failed to persist execution client {client_id}: {e}"))?;
1404
1405            let result = sqlx::query(
1406                r#"
1407                UPDATE "order_event"
1408                SET client_id = $2
1409                WHERE client_order_id = $1
1410                  AND (client_id IS NULL OR client_id = $2)
1411            "#,
1412            )
1413            .bind(client_order_id.to_string())
1414            .bind(client_id.to_string())
1415            .execute(&mut *transaction)
1416            .await
1417            .map_err(|e| anyhow::anyhow!("Failed to index order client origin: {e}"))?;
1418
1419            if result.rows_affected() == 0 {
1420                anyhow::bail!("No persisted order events found for {client_order_id}");
1421            }
1422        }
1423
1424        transaction
1425            .commit()
1426            .await
1427            .map_err(|e| anyhow::anyhow!("Failed to commit order client origins: {e}"))
1428    }
1429
1430    /// Inserts or updates an order ID to position ID index entry via the provided `pool`.
1431    ///
1432    /// # Errors
1433    ///
1434    /// Returns an error if the SQL INSERT or UPDATE operation fails.
1435    pub async fn index_order_position(
1436        pool: &PgPool,
1437        client_order_id: ClientOrderId,
1438        position_id: PositionId,
1439    ) -> anyhow::Result<()> {
1440        sqlx::query(
1441            r#"
1442            INSERT INTO "order_position_index" (
1443                client_order_id, position_id, created_at, updated_at
1444            ) VALUES (
1445                $1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1446            )
1447            ON CONFLICT (client_order_id)
1448            DO UPDATE
1449            SET
1450                position_id = $2, updated_at = CURRENT_TIMESTAMP
1451        "#,
1452        )
1453        .bind(client_order_id.to_string())
1454        .bind(position_id.to_string())
1455        .execute(pool)
1456        .await
1457        .map(|_| ())
1458        .map_err(|e| anyhow::anyhow!("Failed to insert into order_position_index table: {e}"))
1459    }
1460
1461    /// Loads the order ID to position ID index via the provided `pool`.
1462    ///
1463    /// # Errors
1464    ///
1465    /// Returns an error if the SQL SELECT or iteration fails.
1466    pub async fn load_index_order_position(
1467        pool: &PgPool,
1468    ) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
1469        let mut map: AHashMap<ClientOrderId, PositionId> = AHashMap::new();
1470        let result = sqlx::query_as::<_, OrderPositionIndexRow>(
1471            r#"
1472            SELECT
1473                client_order_id AS "client_order_id",
1474                position_id AS "position_id"
1475            FROM "order_position_index"
1476        "#,
1477        )
1478        .fetch_all(pool)
1479        .await
1480        .map_err(|e| anyhow::anyhow!("Failed to load order position index: {e}"))?;
1481
1482        for row in result {
1483            map.insert(row.client_order_id, row.position_id);
1484        }
1485        Ok(map)
1486    }
1487
1488    /// Inserts a `Signal` entry via the provided `pool`.
1489    ///
1490    /// # Errors
1491    ///
1492    /// Returns an error if the SQL INSERT operation fails.
1493    pub async fn add_signal(pool: &PgPool, signal: &Signal) -> anyhow::Result<()> {
1494        sqlx::query(
1495            r#"
1496            INSERT INTO "signal" (
1497                name, value, ts_event, ts_init, created_at, updated_at
1498            ) VALUES (
1499                $1, $2, $3, $4, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1500            )
1501            ON CONFLICT (id)
1502            DO UPDATE
1503            SET
1504                name = $1, value = $2, ts_event = $3, ts_init = $4,
1505                updated_at = CURRENT_TIMESTAMP
1506        "#,
1507        )
1508        .bind(signal.name.to_string())
1509        .bind(signal.value.clone())
1510        .bind(signal.ts_event.to_string())
1511        .bind(signal.ts_init.to_string())
1512        .execute(pool)
1513        .await
1514        .map(|_| ())
1515        .map_err(|e| anyhow::anyhow!("Failed to insert into signal table: {e}"))
1516    }
1517
1518    /// Loads all `Signal` entries by `name` via the provided `pool`.
1519    ///
1520    /// # Errors
1521    ///
1522    /// Returns an error if the SQL SELECT or deserialization fails.
1523    pub async fn load_signals(pool: &PgPool, name: &str) -> anyhow::Result<Vec<Signal>> {
1524        sqlx::query_as::<_, SignalRow>(
1525            r#"SELECT * FROM "signal" WHERE name = $1 ORDER BY ts_init ASC"#,
1526        )
1527        .bind(name)
1528        .fetch_all(pool)
1529        .await
1530        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1531        .map_err(|e| anyhow::anyhow!("Failed to load signals: {e}"))
1532    }
1533
1534    /// Inserts a `CustomData` entry via the provided `pool`.
1535    ///
1536    /// Serializes the model `CustomData` to full JSON and stores it in the JSONB `value` column.
1537    ///
1538    /// # Errors
1539    ///
1540    /// Returns an error if the SQL INSERT operation fails.
1541    pub async fn add_custom_data(pool: &PgPool, data: &CustomData) -> anyhow::Result<()> {
1542        let json_bytes = serde_json::to_vec(data)
1543            .map_err(|e| anyhow::anyhow!("CustomData must be valid JSON: {e}"))?;
1544        let value_json: serde_json::Value = serde_json::from_slice(&json_bytes)
1545            .map_err(|e| anyhow::anyhow!("CustomData value must be valid JSON: {e}"))?;
1546        let data_type_obj = value_json
1547            .get("data_type")
1548            .and_then(|v| v.as_object())
1549            .ok_or_else(|| anyhow::anyhow!("CustomData JSON missing data_type"))?;
1550        let data_type_name = data_type_obj
1551            .get("type_name")
1552            .and_then(|v| v.as_str())
1553            .unwrap_or("");
1554        let metadata_json = data_type_obj
1555            .get("metadata")
1556            .cloned()
1557            .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new()));
1558        let identifier = data_type_obj
1559            .get("identifier")
1560            .and_then(|v| v.as_str())
1561            .unwrap_or("");
1562        sqlx::query(
1563            r#"
1564            INSERT INTO "custom" (data_type, metadata, identifier, value, ts_event, ts_init, created_at, updated_at)
1565            VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
1566            ON CONFLICT (id)
1567            DO UPDATE SET
1568                data_type = EXCLUDED.data_type,
1569                metadata = EXCLUDED.metadata,
1570                identifier = EXCLUDED.identifier,
1571                value = EXCLUDED.value,
1572                ts_event = EXCLUDED.ts_event,
1573                ts_init = EXCLUDED.ts_init,
1574                updated_at = CURRENT_TIMESTAMP
1575        "#,
1576        )
1577        .bind(data_type_name)
1578        .bind(&metadata_json)
1579        .bind(identifier)
1580        .bind(&value_json)
1581        .bind(
1582            value_json
1583                .get("ts_event")
1584                .and_then(serde_json::Value::as_u64)
1585                .unwrap_or_else(|| data.ts_init().as_u64())
1586                .to_string(),
1587        )
1588        .bind(data.ts_init().to_string())
1589        .execute(pool)
1590        .await
1591        .map(|_| ())
1592        .map_err(|e| anyhow::anyhow!("Failed to insert into custom table: {e}"))
1593    }
1594
1595    /// Loads all `CustomData` entries of `data_type` via the provided `pool`.
1596    ///
1597    /// Filters by `data_type`, `metadata`, and `identifier` columns to match the requested data type.
1598    ///
1599    /// # Errors
1600    ///
1601    /// Returns an error if the SQL SELECT or deserialization fails.
1602    pub async fn load_custom_data(
1603        pool: &PgPool,
1604        data_type: &DataType,
1605    ) -> anyhow::Result<Vec<CustomData>> {
1606        let metadata_json = data_type.metadata().as_ref().map_or(
1607            Ok(serde_json::Value::Object(serde_json::Map::new())),
1608            serde_json::to_value,
1609        )?;
1610
1611        let type_name = data_type.type_name();
1612        let short_type = type_name.rsplit([':', '.']).next().unwrap_or(type_name);
1613
1614        let rows = match data_type.identifier() {
1615            Some(identifier) => {
1616                sqlx::query(
1617                    r#"SELECT value, ts_event, ts_init FROM "custom"
1618                   WHERE (data_type = $1 OR data_type = $2)
1619                     AND metadata = $3
1620                     AND identifier = $4
1621                   ORDER BY ts_init ASC"#,
1622                )
1623                .bind(type_name)
1624                .bind(short_type)
1625                .bind(&metadata_json)
1626                .bind(identifier)
1627                .fetch_all(pool)
1628                .await
1629            }
1630            None => {
1631                sqlx::query(
1632                    r#"SELECT value, ts_event, ts_init FROM "custom"
1633                   WHERE (data_type = $1 OR data_type = $2)
1634                     AND metadata = $3
1635                     AND identifier = ''
1636                   ORDER BY ts_init ASC"#,
1637                )
1638                .bind(type_name)
1639                .bind(short_type)
1640                .bind(&metadata_json)
1641                .fetch_all(pool)
1642                .await
1643            }
1644        }
1645        .map_err(|e| anyhow::anyhow!("Failed to load custom data: {e}"))?;
1646
1647        let mut results = Vec::with_capacity(rows.len());
1648        for row in rows {
1649            let value_json: serde_json::Value = row.try_get("value")?;
1650            let json_bytes = serde_json::to_vec(&value_json)
1651                .map_err(|e| anyhow::anyhow!("Failed to serialize JSON: {e}"))?;
1652            let custom =
1653                CustomData::from_json_bytes(&json_bytes).map_err(|e| anyhow::anyhow!("{e}"))?;
1654            results.push(custom);
1655        }
1656        Ok(results)
1657    }
1658}
nautilus_infrastructure/sql/queries.rs

nautilus_infrastructure/sql/
queries.rs
