Skip to main content

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