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

nautilus_infrastructure/sql/
queries.rs
