nautilus_infrastructure/sql/models/
orders.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 std::str::FromStr;
17
18use indexmap::IndexMap;
19use nautilus_core::{UUID4, UnixNanos};
20use nautilus_model::{
21    enums::{
22        ContingencyType, LiquiditySide, OrderSide, OrderStatus, OrderType, TimeInForce, TriggerType,
23    },
24    events::{
25        OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied, OrderEmulated,
26        OrderEventAny, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
27        OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
28        OrderSnapshot, OrderSubmitted, OrderTriggered, OrderUpdated,
29    },
30    identifiers::{
31        AccountId, ClientOrderId, ExecAlgorithmId, InstrumentId, OrderListId, PositionId,
32        StrategyId, TradeId, TraderId, VenueOrderId,
33    },
34    types::{Currency, Price, Quantity},
35};
36use rust_decimal::Decimal;
37use sqlx::{FromRow, Row, postgres::PgRow};
38use ustr::Ustr;
39
40use crate::sql::models::{decode_money, enums::TrailingOffsetTypePg};
41
42#[derive(Debug)]
43pub struct OrderEventAnyRow(pub OrderEventAny);
44
45#[derive(Debug)]
46pub struct OrderAcceptedRow(pub OrderAccepted);
47
48#[derive(Debug)]
49pub struct OrderCancelRejectedRow(pub OrderCancelRejected);
50
51#[derive(Debug)]
52pub struct OrderCanceledRow(pub OrderCanceled);
53
54#[derive(Debug)]
55pub struct OrderDeniedRow(pub OrderDenied);
56
57#[derive(Debug)]
58pub struct OrderEmulatedRow(pub OrderEmulated);
59
60#[derive(Debug)]
61pub struct OrderExpiredRow(pub OrderExpired);
62
63#[derive(Debug)]
64pub struct OrderFilledRow(pub OrderFilled);
65
66#[derive(Debug)]
67pub struct OrderFillVoidedRow(pub OrderFillVoided);
68
69#[derive(Debug)]
70pub struct OrderInitializedRow(pub OrderInitialized);
71
72#[derive(Debug)]
73pub struct OrderModifyRejectedRow(pub OrderModifyRejected);
74
75#[derive(Debug)]
76pub struct OrderPendingCancelRow(pub OrderPendingCancel);
77
78#[derive(Debug)]
79pub struct OrderPendingUpdateRow(pub OrderPendingUpdate);
80
81#[derive(Debug)]
82pub struct OrderRejectedRow(pub OrderRejected);
83
84#[derive(Debug)]
85pub struct OrderReleasedRow(pub OrderReleased);
86
87#[derive(Debug)]
88pub struct OrderSubmittedRow(pub OrderSubmitted);
89
90#[derive(Debug)]
91pub struct OrderTriggeredRow(pub OrderTriggered);
92
93#[derive(Debug)]
94pub struct OrderUpdatedRow(pub OrderUpdated);
95
96#[derive(Debug)]
97pub struct OrderSnapshotRow(pub OrderSnapshot);
98
99impl<'r> FromRow<'r, PgRow> for OrderEventAnyRow {
100    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
101        let kind = row.get::<String, _>("kind");
102        if kind == "OrderAccepted" {
103            let row = OrderAcceptedRow::from_row(row)?;
104            Ok(Self(OrderEventAny::Accepted(row.0)))
105        } else if kind == "OrderCancelRejected" {
106            let row = OrderCancelRejectedRow::from_row(row)?;
107            Ok(Self(OrderEventAny::CancelRejected(row.0)))
108        } else if kind == "OrderCanceled" {
109            let row = OrderCanceledRow::from_row(row)?;
110            Ok(Self(OrderEventAny::Canceled(row.0)))
111        } else if kind == "OrderDenied" {
112            let row = OrderDeniedRow::from_row(row)?;
113            Ok(Self(OrderEventAny::Denied(row.0)))
114        } else if kind == "OrderEmulated" {
115            let row = OrderEmulatedRow::from_row(row)?;
116            Ok(Self(OrderEventAny::Emulated(row.0)))
117        } else if kind == "OrderExpired" {
118            let row = OrderExpiredRow::from_row(row)?;
119            Ok(Self(OrderEventAny::Expired(row.0)))
120        } else if kind == "OrderFillVoided" {
121            let row = OrderFillVoidedRow::from_row(row)?;
122            Ok(Self(OrderEventAny::FillVoided(row.0)))
123        } else if kind == "OrderFilled" {
124            let row = OrderFilledRow::from_row(row)?;
125            Ok(Self(OrderEventAny::Filled(row.0)))
126        } else if kind == "OrderInitialized" {
127            let row = OrderInitializedRow::from_row(row)?;
128            Ok(Self(OrderEventAny::Initialized(row.0)))
129        } else if kind == "OrderModifyRejected" {
130            let row = OrderModifyRejectedRow::from_row(row)?;
131            Ok(Self(OrderEventAny::ModifyRejected(row.0)))
132        } else if kind == "OrderPendingCancel" {
133            let row = OrderPendingCancelRow::from_row(row)?;
134            Ok(Self(OrderEventAny::PendingCancel(row.0)))
135        } else if kind == "OrderPendingUpdate" {
136            let row = OrderPendingUpdateRow::from_row(row)?;
137            Ok(Self(OrderEventAny::PendingUpdate(row.0)))
138        } else if kind == "OrderRejected" {
139            let row = OrderRejectedRow::from_row(row)?;
140            Ok(Self(OrderEventAny::Rejected(row.0)))
141        } else if kind == "OrderReleased" {
142            let row = OrderReleasedRow::from_row(row)?;
143            Ok(Self(OrderEventAny::Released(row.0)))
144        } else if kind == "OrderSubmitted" {
145            let row = OrderSubmittedRow::from_row(row)?;
146            Ok(Self(OrderEventAny::Submitted(row.0)))
147        } else if kind == "OrderTriggered" {
148            let row = OrderTriggeredRow::from_row(row)?;
149            Ok(Self(OrderEventAny::Triggered(row.0)))
150        } else if kind == "OrderUpdated" {
151            let row = OrderUpdatedRow::from_row(row)?;
152            Ok(Self(OrderEventAny::Updated(row.0)))
153        } else {
154            Err(sqlx::Error::Decode(
155                format!("Unknown order event kind: {kind} in Postgres transformation").into(),
156            ))
157        }
158    }
159}
160
161impl<'r> FromRow<'r, PgRow> for OrderInitializedRow {
162    #[expect(
163        clippy::too_many_lines,
164        reason = "SQL row mapping mirrors the full order initialized event constructor"
165    )]
166    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
167        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
168        let client_order_id = row
169            .try_get::<&str, _>("client_order_id")
170            .map(ClientOrderId::from)?;
171        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
172        let strategy_id = row
173            .try_get::<&str, _>("strategy_id")
174            .map(StrategyId::from)?;
175        let instrument_id = row
176            .try_get::<&str, _>("instrument_id")
177            .map(InstrumentId::from)?;
178        let order_type = row
179            .try_get::<&str, _>("order_type")
180            .map(|x| OrderType::from_str(x).unwrap())?;
181        let order_side = row
182            .try_get::<&str, _>("order_side")
183            .map(|x| OrderSide::from_str(x).unwrap())?;
184        let quantity = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
185        let time_in_force = row
186            .try_get::<&str, _>("time_in_force")
187            .map(|x| TimeInForce::from_str(x).unwrap())?;
188        let post_only = row.try_get::<bool, _>("post_only")?;
189        let reduce_only = row.try_get::<bool, _>("reduce_only")?;
190        let quote_quantity = row.try_get::<bool, _>("quote_quantity")?;
191        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
192        let ts_event = row.try_get::<String, _>("ts_event").map(UnixNanos::from)?;
193        let ts_init = row.try_get::<String, _>("ts_init").map(UnixNanos::from)?;
194        let price = row
195            .try_get::<Option<&str>, _>("price")
196            .ok()
197            .and_then(|x| x.map(Price::from));
198        let activation_price = row
199            .try_get::<Option<&str>, _>("activation_price")
200            .ok()
201            .and_then(|x| x.map(Price::from));
202        let trigger_price = row
203            .try_get::<Option<&str>, _>("trigger_price")
204            .ok()
205            .and_then(|x| x.map(Price::from));
206        let trigger_type = row
207            .try_get::<Option<&str>, _>("trigger_type")
208            .ok()
209            .and_then(parse_trigger_type);
210        let limit_offset = row
211            .try_get::<Option<&str>, _>("limit_offset")
212            .ok()
213            .and_then(|x| x.and_then(|s| Decimal::from_str(s).ok()));
214        let trailing_offset = row
215            .try_get::<Option<&str>, _>("trailing_offset")
216            .ok()
217            .and_then(|x| x.and_then(|s| Decimal::from_str(s).ok()));
218        let trailing_offset_type = row
219            .try_get::<Option<TrailingOffsetTypePg>, _>("trailing_offset_type")
220            .ok()
221            .flatten()
222            .and_then(|value| value.0);
223        let expire_time = row
224            .try_get::<Option<&str>, _>("expire_time")
225            .ok()
226            .and_then(|x| x.map(UnixNanos::from));
227        let display_qty = row
228            .try_get::<Option<&str>, _>("display_qty")
229            .ok()
230            .and_then(|x| x.map(Quantity::from));
231        let emulation_trigger = row
232            .try_get::<Option<&str>, _>("emulation_trigger")
233            .ok()
234            .and_then(parse_trigger_type);
235        let trigger_instrument_id = row
236            .try_get::<Option<&str>, _>("trigger_instrument_id")
237            .ok()
238            .and_then(|x| x.map(InstrumentId::from));
239        let contingency_type = row
240            .try_get::<Option<&str>, _>("contingency_type")
241            .ok()
242            .and_then(parse_contingency_type);
243        let order_list_id = row
244            .try_get::<Option<&str>, _>("order_list_id")
245            .ok()
246            .and_then(|x| x.map(OrderListId::from));
247        let linked_order_ids = row
248            .try_get::<Vec<String>, _>("linked_order_ids")
249            .ok()
250            .map(|x| x.iter().map(|x| ClientOrderId::from(x.as_str())).collect());
251        let parent_order_id = row
252            .try_get::<Option<&str>, _>("parent_order_id")
253            .ok()
254            .and_then(|x| x.map(ClientOrderId::from));
255        let exec_algorithm_id = row
256            .try_get::<Option<&str>, _>("exec_algorithm_id")
257            .ok()
258            .and_then(|x| x.map(ExecAlgorithmId::from));
259        let exec_algorithm_params: Option<IndexMap<Ustr, Ustr>> = row
260            .try_get::<Option<serde_json::Value>, _>("exec_algorithm_params")
261            .ok()
262            .and_then(|x| x.map(|x| serde_json::from_value::<IndexMap<String, String>>(x).unwrap()))
263            .map(|x| {
264                x.into_iter()
265                    .map(|(k, v)| (Ustr::from(k.as_str()), Ustr::from(v.as_str())))
266                    .collect()
267            });
268        let exec_spawn_id = row
269            .try_get::<Option<&str>, _>("exec_spawn_id")
270            .ok()
271            .and_then(|x| x.map(ClientOrderId::from));
272        let tags = tags_from_row(row);
273        let mut order_event = OrderInitialized::new_checked(
274            trader_id,
275            strategy_id,
276            instrument_id,
277            client_order_id,
278            order_side,
279            order_type,
280            quantity,
281            time_in_force,
282            post_only,
283            reduce_only,
284            quote_quantity,
285            reconciliation,
286            event_id,
287            ts_event,
288            ts_init,
289            price,
290            activation_price,
291            trigger_price,
292            trigger_type,
293            limit_offset,
294            trailing_offset,
295            trailing_offset_type,
296            expire_time,
297            display_qty,
298            emulation_trigger,
299            trigger_instrument_id,
300            contingency_type,
301            order_list_id,
302            linked_order_ids,
303            parent_order_id,
304            exec_algorithm_id,
305            exec_algorithm_params,
306            exec_spawn_id,
307            tags,
308        )
309        .map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
310        order_event.causation_id = causation_id_from_row(row)?;
311        Ok(Self(order_event))
312    }
313}
314
315impl<'r> FromRow<'r, PgRow> for OrderAcceptedRow {
316    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
317        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
318        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
319        let strategy_id = row
320            .try_get::<&str, _>("strategy_id")
321            .map(StrategyId::from)?;
322        let instrument_id = row
323            .try_get::<&str, _>("instrument_id")
324            .map(InstrumentId::from)?;
325        let client_order_id = row
326            .try_get::<&str, _>("client_order_id")
327            .map(ClientOrderId::from)?;
328        let venue_order_id = row
329            .try_get::<&str, _>("venue_order_id")
330            .map(VenueOrderId::from)?;
331        let account_id = row.try_get::<&str, _>("account_id").map(AccountId::from)?;
332        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
333        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
334        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
335        let causation_id = causation_id_from_row(row)?;
336        let mut order_event = OrderAccepted::new(
337            trader_id,
338            strategy_id,
339            instrument_id,
340            client_order_id,
341            venue_order_id,
342            account_id,
343            event_id,
344            ts_event,
345            ts_init,
346            reconciliation,
347        );
348        order_event.causation_id = causation_id;
349        Ok(Self(order_event))
350    }
351}
352
353impl<'r> FromRow<'r, PgRow> for OrderCancelRejectedRow {
354    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
355        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
356        let strategy_id = row
357            .try_get::<&str, _>("strategy_id")
358            .map(StrategyId::from)?;
359        let instrument_id = row
360            .try_get::<&str, _>("instrument_id")
361            .map(InstrumentId::from)?;
362        let client_order_id = row
363            .try_get::<&str, _>("client_order_id")
364            .map(ClientOrderId::from)?;
365        let reason = row.try_get::<&str, _>("reason").map(Ustr::from)?;
366        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
367        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
368        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
369        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
370        let venue_order_id = row
371            .try_get::<Option<&str>, _>("venue_order_id")?
372            .map(Into::into);
373        let account_id = row
374            .try_get::<Option<&str>, _>("account_id")?
375            .map(Into::into);
376        let causation_id = causation_id_from_row(row)?;
377        let mut order_event = OrderCancelRejected::new(
378            trader_id,
379            strategy_id,
380            instrument_id,
381            client_order_id,
382            reason,
383            event_id,
384            ts_event,
385            ts_init,
386            reconciliation,
387            venue_order_id,
388            account_id,
389        );
390        order_event.causation_id = causation_id;
391        Ok(Self(order_event))
392    }
393}
394
395impl<'r> FromRow<'r, PgRow> for OrderCanceledRow {
396    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
397        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
398        let strategy_id = row
399            .try_get::<&str, _>("strategy_id")
400            .map(StrategyId::from)?;
401        let instrument_id = row
402            .try_get::<&str, _>("instrument_id")
403            .map(InstrumentId::from)?;
404        let client_order_id = row
405            .try_get::<&str, _>("client_order_id")
406            .map(ClientOrderId::from)?;
407        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
408        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
409        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
410        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
411        let venue_order_id = row
412            .try_get::<Option<&str>, _>("venue_order_id")?
413            .map(Into::into);
414        let account_id = row
415            .try_get::<Option<&str>, _>("account_id")?
416            .map(Into::into);
417        let reason = row.try_get::<Option<&str>, _>("reason")?.map(Ustr::from);
418        let causation_id = causation_id_from_row(row)?;
419        let mut order_event = OrderCanceled::new(
420            trader_id,
421            strategy_id,
422            instrument_id,
423            client_order_id,
424            event_id,
425            ts_event,
426            ts_init,
427            reconciliation,
428            venue_order_id,
429            account_id,
430            reason,
431        );
432        order_event.causation_id = causation_id;
433        Ok(Self(order_event))
434    }
435}
436
437impl<'r> FromRow<'r, PgRow> for OrderDeniedRow {
438    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
439        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
440        let strategy_id = row
441            .try_get::<&str, _>("strategy_id")
442            .map(StrategyId::from)?;
443        let instrument_id = row
444            .try_get::<&str, _>("instrument_id")
445            .map(InstrumentId::from)?;
446        let client_order_id = row
447            .try_get::<&str, _>("client_order_id")
448            .map(ClientOrderId::from)?;
449        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
450        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
451        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
452        let reason = row.try_get::<&str, _>("reason").map(Ustr::from)?;
453        let causation_id = causation_id_from_row(row)?;
454        let mut order_event = OrderDenied::new(
455            trader_id,
456            strategy_id,
457            instrument_id,
458            client_order_id,
459            reason,
460            event_id,
461            ts_event,
462            ts_init,
463        );
464        order_event.causation_id = causation_id;
465        Ok(Self(order_event))
466    }
467}
468
469impl<'r> FromRow<'r, PgRow> for OrderEmulatedRow {
470    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
471        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
472        let strategy_id = row
473            .try_get::<&str, _>("strategy_id")
474            .map(StrategyId::from)?;
475        let instrument_id = row
476            .try_get::<&str, _>("instrument_id")
477            .map(InstrumentId::from)?;
478        let client_order_id = row
479            .try_get::<&str, _>("client_order_id")
480            .map(ClientOrderId::from)?;
481        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
482        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
483        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
484        let causation_id = causation_id_from_row(row)?;
485        let mut order_event = OrderEmulated::new(
486            trader_id,
487            strategy_id,
488            instrument_id,
489            client_order_id,
490            event_id,
491            ts_event,
492            ts_init,
493        );
494        order_event.causation_id = causation_id;
495        Ok(Self(order_event))
496    }
497}
498
499impl<'r> FromRow<'r, PgRow> for OrderExpiredRow {
500    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
501        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
502        let strategy_id = row
503            .try_get::<&str, _>("strategy_id")
504            .map(StrategyId::from)?;
505        let instrument_id = row
506            .try_get::<&str, _>("instrument_id")
507            .map(InstrumentId::from)?;
508        let client_order_id = row
509            .try_get::<&str, _>("client_order_id")
510            .map(ClientOrderId::from)?;
511        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
512        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
513        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
514        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
515        let venue_order_id = row
516            .try_get::<Option<&str>, _>("venue_order_id")?
517            .map(Into::into);
518        let account_id = row
519            .try_get::<Option<&str>, _>("account_id")?
520            .map(Into::into);
521        let causation_id = causation_id_from_row(row)?;
522        let mut order_event = OrderExpired::new(
523            trader_id,
524            strategy_id,
525            instrument_id,
526            client_order_id,
527            event_id,
528            ts_event,
529            ts_init,
530            reconciliation,
531            venue_order_id,
532            account_id,
533        );
534        order_event.causation_id = causation_id;
535        Ok(Self(order_event))
536    }
537}
538
539impl<'r> FromRow<'r, PgRow> for OrderFilledRow {
540    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
541        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
542        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
543        let strategy_id = row
544            .try_get::<&str, _>("strategy_id")
545            .map(StrategyId::from)?;
546        let instrument_id = row
547            .try_get::<&str, _>("instrument_id")
548            .map(InstrumentId::from)?;
549        let client_order_id = row
550            .try_get::<&str, _>("client_order_id")
551            .map(ClientOrderId::from)?;
552        let venue_order_id = row
553            .try_get::<&str, _>("venue_order_id")
554            .map(VenueOrderId::from)?;
555        let account_id = row.try_get::<&str, _>("account_id").map(AccountId::from)?;
556        let trade_id = row.try_get::<&str, _>("trade_id").map(TradeId::from)?;
557        let order_side = row
558            .try_get::<&str, _>("order_side")
559            .map(|x| OrderSide::from_str(x).unwrap())?;
560        let order_type = row
561            .try_get::<&str, _>("order_type")
562            .map(|x| OrderType::from_str(x).unwrap())?;
563        let last_px = row.try_get::<&str, _>("last_px").map(Price::from)?;
564        let last_qty = row.try_get::<&str, _>("last_qty").map(Quantity::from)?;
565        let currency = row.try_get::<&str, _>("currency").map(Currency::from)?;
566        let liquidity_side = row
567            .try_get::<&str, _>("liquidity_side")
568            .map(|x| LiquiditySide::from_str(x).unwrap())?;
569        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
570        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
571        let position_id = row
572            .try_get::<Option<&str>, _>("position_id")
573            .map(|x| x.map(PositionId::from))?;
574        let commission = row
575            .try_get::<Option<&str>, _>("commission")?
576            .map(|value| decode_money(value, "commission"))
577            .transpose()?;
578        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
579        let info = decode_info(row)?;
580        let causation_id = causation_id_from_row(row)?;
581        let mut order_event = OrderFilled::new(
582            trader_id,
583            strategy_id,
584            instrument_id,
585            client_order_id,
586            venue_order_id,
587            account_id,
588            trade_id,
589            order_side,
590            order_type,
591            last_qty,
592            last_px,
593            currency,
594            liquidity_side,
595            event_id,
596            ts_event,
597            ts_init,
598            reconciliation,
599            position_id,
600            commission,
601            info,
602        );
603        order_event.causation_id = causation_id;
604        Ok(Self(order_event))
605    }
606}
607
608impl<'r> FromRow<'r, PgRow> for OrderFillVoidedRow {
609    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
610        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
611        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
612        let strategy_id = row
613            .try_get::<&str, _>("strategy_id")
614            .map(StrategyId::from)?;
615        let instrument_id = row
616            .try_get::<&str, _>("instrument_id")
617            .map(InstrumentId::from)?;
618        let client_order_id = row
619            .try_get::<&str, _>("client_order_id")
620            .map(ClientOrderId::from)?;
621        let venue_order_id = row
622            .try_get::<&str, _>("venue_order_id")
623            .map(VenueOrderId::from)?;
624        let account_id = row.try_get::<&str, _>("account_id").map(AccountId::from)?;
625        let correction_id = row
626            .try_get::<Option<&str>, _>("correction_id")?
627            .map(Ustr::from)
628            .ok_or_else(|| {
629                sqlx::Error::Decode(
630                    "OrderFillVoided row has no correction_id; it predates the column and \
631                     the value cannot be recovered"
632                        .into(),
633                )
634            })?;
635        let trade_id = row.try_get::<&str, _>("trade_id").map(TradeId::from)?;
636        let voided_qty = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
637        let commission_voided = row
638            .try_get::<Option<&str>, _>("commission")?
639            .map(|value| decode_money(value, "commission"))
640            .transpose()?;
641        let order_side = OrderSide::from_str(row.try_get::<&str, _>("order_side")?)
642            .map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
643        let order_type = OrderType::from_str(row.try_get::<&str, _>("order_type")?)
644            .map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
645        let last_px = row.try_get::<&str, _>("last_px").map(Price::from)?;
646        let currency = row.try_get::<&str, _>("currency").map(Currency::from)?;
647        let liquidity_side = LiquiditySide::from_str(row.try_get::<&str, _>("liquidity_side")?)
648            .map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
649        let position_id = row
650            .try_get::<Option<&str>, _>("position_id")?
651            .map(PositionId::from);
652        let reason = row.try_get::<Option<&str>, _>("reason")?.map(Ustr::from);
653        let info = decode_info(row)?;
654        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
655        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
656        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
657        let is_reopened = row
658            .try_get::<Option<bool>, _>("is_reopened")?
659            .unwrap_or(false);
660        let causation_id = causation_id_from_row(row)?;
661        let mut order_event = OrderFillVoided::new(
662            trader_id,
663            strategy_id,
664            instrument_id,
665            client_order_id,
666            venue_order_id,
667            account_id,
668            correction_id,
669            trade_id,
670            voided_qty,
671            commission_voided,
672            order_side,
673            order_type,
674            last_px,
675            currency,
676            liquidity_side,
677            position_id,
678            reason,
679            info,
680            event_id,
681            ts_event,
682            ts_init,
683            reconciliation,
684            is_reopened,
685        );
686        order_event.causation_id = causation_id;
687        Ok(Self(order_event))
688    }
689}
690
691impl<'r> FromRow<'r, PgRow> for OrderModifyRejectedRow {
692    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
693        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
694        let strategy_id = row
695            .try_get::<&str, _>("strategy_id")
696            .map(StrategyId::from)?;
697        let instrument_id = row
698            .try_get::<&str, _>("instrument_id")
699            .map(InstrumentId::from)?;
700        let client_order_id = row
701            .try_get::<&str, _>("client_order_id")
702            .map(ClientOrderId::from)?;
703        let reason = row.try_get::<&str, _>("reason").map(Ustr::from)?;
704        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
705        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
706        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
707        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
708        let venue_order_id = row
709            .try_get::<Option<&str>, _>("venue_order_id")?
710            .map(Into::into);
711        let account_id = row
712            .try_get::<Option<&str>, _>("account_id")?
713            .map(Into::into);
714        let causation_id = causation_id_from_row(row)?;
715        let mut order_event = OrderModifyRejected::new(
716            trader_id,
717            strategy_id,
718            instrument_id,
719            client_order_id,
720            reason,
721            event_id,
722            ts_event,
723            ts_init,
724            reconciliation,
725            venue_order_id,
726            account_id,
727        );
728        order_event.causation_id = causation_id;
729        Ok(Self(order_event))
730    }
731}
732
733impl<'r> FromRow<'r, PgRow> for OrderPendingCancelRow {
734    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
735        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
736        let strategy_id = row
737            .try_get::<&str, _>("strategy_id")
738            .map(StrategyId::from)?;
739        let instrument_id = row
740            .try_get::<&str, _>("instrument_id")
741            .map(InstrumentId::from)?;
742        let client_order_id = row
743            .try_get::<&str, _>("client_order_id")
744            .map(ClientOrderId::from)?;
745        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
746        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
747        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
748        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
749        let venue_order_id = row
750            .try_get::<Option<&str>, _>("venue_order_id")?
751            .map(Into::into);
752        let account_id = row
753            .try_get::<Option<&str>, _>("account_id")?
754            .map(Into::into);
755        let causation_id = causation_id_from_row(row)?;
756        let mut order_event = OrderPendingCancel::new(
757            trader_id,
758            strategy_id,
759            instrument_id,
760            client_order_id,
761            account_id,
762            event_id,
763            ts_event,
764            ts_init,
765            reconciliation,
766            venue_order_id,
767        );
768        order_event.causation_id = causation_id;
769        Ok(Self(order_event))
770    }
771}
772
773impl<'r> FromRow<'r, PgRow> for OrderPendingUpdateRow {
774    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
775        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
776        let strategy_id = row
777            .try_get::<&str, _>("strategy_id")
778            .map(StrategyId::from)?;
779        let instrument_id = row
780            .try_get::<&str, _>("instrument_id")
781            .map(InstrumentId::from)?;
782        let client_order_id = row
783            .try_get::<&str, _>("client_order_id")
784            .map(ClientOrderId::from)?;
785        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
786        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
787        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
788        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
789        let venue_order_id = row
790            .try_get::<Option<&str>, _>("venue_order_id")?
791            .map(Into::into);
792        let account_id = row
793            .try_get::<Option<&str>, _>("account_id")?
794            .map(Into::into);
795        let causation_id = causation_id_from_row(row)?;
796        let mut order_event = OrderPendingUpdate::new(
797            trader_id,
798            strategy_id,
799            instrument_id,
800            client_order_id,
801            account_id,
802            event_id,
803            ts_event,
804            ts_init,
805            reconciliation,
806            venue_order_id,
807        );
808        order_event.causation_id = causation_id;
809        Ok(Self(order_event))
810    }
811}
812
813impl<'r> FromRow<'r, PgRow> for OrderRejectedRow {
814    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
815        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
816        let strategy_id = row
817            .try_get::<&str, _>("strategy_id")
818            .map(StrategyId::from)?;
819        let instrument_id = row
820            .try_get::<&str, _>("instrument_id")
821            .map(InstrumentId::from)?;
822        let client_order_id = row
823            .try_get::<&str, _>("client_order_id")
824            .map(ClientOrderId::from)?;
825        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
826        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
827        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
828        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
829        let account_id = row.try_get::<&str, _>("account_id").map(AccountId::from)?;
830        let reason = row.try_get::<&str, _>("reason").map(Ustr::from)?;
831        // Rows written before this column existed decode as NULL; false is its meaning
832        let due_post_only = row
833            .try_get::<Option<bool>, _>("due_post_only")?
834            .unwrap_or(false);
835        let causation_id = causation_id_from_row(row)?;
836        let mut order_event = OrderRejected::new(
837            trader_id,
838            strategy_id,
839            instrument_id,
840            client_order_id,
841            account_id,
842            reason,
843            event_id,
844            ts_event,
845            ts_init,
846            reconciliation,
847            due_post_only,
848        );
849        order_event.causation_id = causation_id;
850        Ok(Self(order_event))
851    }
852}
853
854impl<'r> FromRow<'r, PgRow> for OrderReleasedRow {
855    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
856        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
857        let strategy_id = row
858            .try_get::<&str, _>("strategy_id")
859            .map(StrategyId::from)?;
860        let instrument_id = row
861            .try_get::<&str, _>("instrument_id")
862            .map(InstrumentId::from)?;
863        let client_order_id = row
864            .try_get::<&str, _>("client_order_id")
865            .map(ClientOrderId::from)?;
866        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
867        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
868        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
869        let released_price = row
870            .try_get::<Option<&str>, _>("released_price")?
871            .map(Price::from)
872            .ok_or_else(|| {
873                sqlx::Error::Decode(
874                    "OrderReleased row has no released_price; it predates the column and \
875                     the value cannot be recovered"
876                        .into(),
877                )
878            })?;
879        let causation_id = causation_id_from_row(row)?;
880        let mut order_event = OrderReleased::new(
881            trader_id,
882            strategy_id,
883            instrument_id,
884            client_order_id,
885            released_price,
886            event_id,
887            ts_event,
888            ts_init,
889        );
890        order_event.causation_id = causation_id;
891        Ok(Self(order_event))
892    }
893}
894
895impl<'r> FromRow<'r, PgRow> for OrderSubmittedRow {
896    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
897        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
898        let strategy_id = row
899            .try_get::<&str, _>("strategy_id")
900            .map(StrategyId::from)?;
901        let instrument_id = row
902            .try_get::<&str, _>("instrument_id")
903            .map(InstrumentId::from)?;
904        let client_order_id = row
905            .try_get::<&str, _>("client_order_id")
906            .map(ClientOrderId::from)?;
907        let account_id = row.try_get::<&str, _>("account_id").map(AccountId::from)?;
908        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
909        let ts_event = row
910            .try_get::<String, _>("ts_event")
911            .map(|res| UnixNanos::from(res.as_str()))?;
912        let ts_init = row
913            .try_get::<String, _>("ts_init")
914            .map(|res| UnixNanos::from(res.as_str()))?;
915        let causation_id = causation_id_from_row(row)?;
916        let mut order_event = OrderSubmitted::new(
917            trader_id,
918            strategy_id,
919            instrument_id,
920            client_order_id,
921            account_id,
922            event_id,
923            ts_event,
924            ts_init,
925        );
926        order_event.causation_id = causation_id;
927        Ok(Self(order_event))
928    }
929}
930
931impl<'r> FromRow<'r, PgRow> for OrderTriggeredRow {
932    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
933        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
934        let strategy_id = row
935            .try_get::<&str, _>("strategy_id")
936            .map(StrategyId::from)?;
937        let instrument_id = row
938            .try_get::<&str, _>("instrument_id")
939            .map(InstrumentId::from)?;
940        let client_order_id = row
941            .try_get::<&str, _>("client_order_id")
942            .map(ClientOrderId::from)?;
943        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
944        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
945        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
946        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
947        let venue_order_id = row
948            .try_get::<Option<&str>, _>("venue_order_id")?
949            .map(Into::into);
950        let account_id = row
951            .try_get::<Option<&str>, _>("account_id")?
952            .map(Into::into);
953        let causation_id = causation_id_from_row(row)?;
954        let mut order_event = OrderTriggered::new(
955            trader_id,
956            strategy_id,
957            instrument_id,
958            client_order_id,
959            event_id,
960            ts_event,
961            ts_init,
962            reconciliation,
963            venue_order_id,
964            account_id,
965        );
966        order_event.causation_id = causation_id;
967        Ok(Self(order_event))
968    }
969}
970
971impl<'r> FromRow<'r, PgRow> for OrderUpdatedRow {
972    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
973        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
974        let strategy_id = row
975            .try_get::<&str, _>("strategy_id")
976            .map(StrategyId::from)?;
977        let instrument_id = row
978            .try_get::<&str, _>("instrument_id")
979            .map(InstrumentId::from)?;
980        let client_order_id = row
981            .try_get::<&str, _>("client_order_id")
982            .map(ClientOrderId::from)?;
983        let event_id = row.try_get::<&str, _>("id").map(UUID4::from)?;
984        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
985        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
986        let reconciliation = row.try_get::<bool, _>("reconciliation")?;
987        let venue_order_id = row
988            .try_get::<Option<&str>, _>("venue_order_id")?
989            .map(Into::into);
990        let account_id = row
991            .try_get::<Option<&str>, _>("account_id")?
992            .map(Into::into);
993        let quantity = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
994        let price = row.try_get::<Option<&str>, _>("price")?.map(Price::from);
995        let trigger_price = row
996            .try_get::<Option<&str>, _>("trigger_price")?
997            .map(Price::from);
998        let protection_price = row
999            .try_get::<Option<&str>, _>("protection_price")?
1000            .map(Price::from);
1001        let is_quote_quantity = row.try_get::<bool, _>("quote_quantity")?;
1002        let causation_id = causation_id_from_row(row)?;
1003        let mut order_event = OrderUpdated::new(
1004            trader_id,
1005            strategy_id,
1006            instrument_id,
1007            client_order_id,
1008            quantity,
1009            event_id,
1010            ts_event,
1011            ts_init,
1012            reconciliation,
1013            venue_order_id,
1014            account_id,
1015            price,
1016            trigger_price,
1017            protection_price,
1018            is_quote_quantity,
1019        );
1020        order_event.causation_id = causation_id;
1021        Ok(Self(order_event))
1022    }
1023}
1024
1025impl<'r> FromRow<'r, PgRow> for OrderSnapshotRow {
1026    #[expect(
1027        clippy::too_many_lines,
1028        reason = "SQL row mapping mirrors the full order snapshot constructor"
1029    )]
1030    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
1031        let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
1032        let strategy_id = row
1033            .try_get::<&str, _>("strategy_id")
1034            .map(StrategyId::from)?;
1035        let instrument_id = row
1036            .try_get::<&str, _>("instrument_id")
1037            .map(InstrumentId::from)?;
1038        let client_order_id = row
1039            .try_get::<&str, _>("client_order_id")
1040            .map(ClientOrderId::from)?;
1041        let venue_order_id = row
1042            .try_get::<Option<&str>, _>("venue_order_id")
1043            .ok()
1044            .and_then(|x| x.map(VenueOrderId::from));
1045        let position_id = row
1046            .try_get::<Option<&str>, _>("position_id")
1047            .ok()
1048            .and_then(|x| x.map(PositionId::from));
1049        let account_id = row
1050            .try_get::<Option<&str>, _>("account_id")
1051            .ok()
1052            .and_then(|x| x.map(AccountId::from));
1053        let last_trade_id = row
1054            .try_get::<Option<&str>, _>("last_trade_id")
1055            .ok()
1056            .and_then(|x| x.map(TradeId::from));
1057        let order_type = row
1058            .try_get::<&str, _>("order_type")
1059            .map(|x| OrderType::from_str(x).expect("Invalid `OrderType`"))?;
1060        let order_side = row
1061            .try_get::<&str, _>("order_side")
1062            .map(|x| OrderSide::from_str(x).expect("Invalid `OrderSide`"))?;
1063        let quantity = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
1064        let price = row
1065            .try_get::<Option<&str>, _>("price")
1066            .ok()
1067            .and_then(|x| x.map(Price::from));
1068        let activation_price = row
1069            .try_get::<Option<&str>, _>("activation_price")
1070            .ok()
1071            .and_then(|x| x.map(Price::from));
1072        let trigger_price = row
1073            .try_get::<Option<&str>, _>("trigger_price")
1074            .ok()
1075            .and_then(|x| x.map(Price::from));
1076        let trigger_type = row
1077            .try_get::<Option<&str>, _>("trigger_type")
1078            .ok()
1079            .and_then(parse_trigger_type);
1080        let limit_offset = row
1081            .try_get::<Option<&str>, _>("limit_offset")
1082            .ok()
1083            .and_then(|x| x.and_then(|s| Decimal::from_str(s).ok()));
1084        let trailing_offset = row
1085            .try_get::<Option<&str>, _>("trailing_offset")
1086            .ok()
1087            .and_then(|x| x.and_then(|s| Decimal::from_str(s).ok()));
1088        let trailing_offset_type = row
1089            .try_get::<Option<TrailingOffsetTypePg>, _>("trailing_offset_type")
1090            .ok()
1091            .flatten()
1092            .and_then(|value| value.0);
1093        let time_in_force = row
1094            .try_get::<&str, _>("time_in_force")
1095            .map(|x| TimeInForce::from_str(x).expect("Invalid `TimeInForce`"))?;
1096        let expire_time = row
1097            .try_get::<Option<&str>, _>("expire_time")
1098            .ok()
1099            .and_then(|x| x.map(UnixNanos::from));
1100        let filled_qty = row.try_get::<&str, _>("filled_qty").map(Quantity::from)?;
1101        let liquidity_side = row
1102            .try_get::<Option<&str>, _>("liquidity_side")
1103            .ok()
1104            .and_then(|x| x.map(|x| LiquiditySide::from_str(x).expect("Invalid `LiquiditySide`")));
1105        let avg_px = row.try_get::<Option<Decimal>, _>("avg_px").ok().flatten();
1106        let slippage = row.try_get::<Option<Decimal>, _>("slippage").ok().flatten();
1107        let commissions = row
1108            .try_get::<Option<Vec<String>>, _>("commissions")?
1109            .unwrap_or_default()
1110            .into_iter()
1111            .map(|value| decode_money(&value, "commissions"))
1112            .collect::<Result<Vec<_>, _>>()?;
1113        let status = row
1114            .try_get::<&str, _>("status")
1115            .map(|x| OrderStatus::from_str(x).expect("Invalid `OrderStatus`"))?;
1116        let is_post_only = row.try_get::<bool, _>("is_post_only")?;
1117        let is_reduce_only = row.try_get::<bool, _>("is_reduce_only")?;
1118        let is_quote_quantity = row.try_get::<bool, _>("is_quote_quantity")?;
1119        let display_qty = row
1120            .try_get::<Option<&str>, _>("display_qty")
1121            .ok()
1122            .and_then(|x| x.map(Quantity::from));
1123        let emulation_trigger = row
1124            .try_get::<Option<&str>, _>("emulation_trigger")
1125            .ok()
1126            .and_then(parse_trigger_type);
1127        let trigger_instrument_id = row
1128            .try_get::<Option<&str>, _>("trigger_instrument_id")
1129            .ok()
1130            .and_then(|x| x.map(InstrumentId::from));
1131        let contingency_type = row
1132            .try_get::<Option<&str>, _>("contingency_type")
1133            .ok()
1134            .and_then(parse_contingency_type);
1135        let order_list_id = row
1136            .try_get::<Option<&str>, _>("order_list_id")
1137            .ok()
1138            .and_then(|x| x.map(OrderListId::from));
1139        let linked_order_ids = row
1140            .try_get::<Option<Vec<String>>, _>("linked_order_ids")
1141            .ok()
1142            .and_then(|ids| ids.map(|ids| ids.into_iter().map(ClientOrderId::from).collect()));
1143        let parent_order_id = row
1144            .try_get::<Option<&str>, _>("parent_order_id")
1145            .ok()
1146            .and_then(|x| x.map(ClientOrderId::from));
1147        let exec_algorithm_id = row
1148            .try_get::<Option<&str>, _>("exec_algorithm_id")
1149            .ok()
1150            .and_then(|x| x.map(ExecAlgorithmId::from));
1151        let exec_algorithm_params: Option<IndexMap<Ustr, Ustr>> = row
1152            .try_get::<Option<serde_json::Value>, _>("exec_algorithm_params")
1153            .ok()
1154            .and_then(|x| {
1155                x.map(|x| {
1156                    serde_json::from_value::<IndexMap<String, String>>(x)
1157                        .expect("Invalid exec algorithm params")
1158                })
1159            })
1160            .map(|x| {
1161                x.into_iter()
1162                    .map(|(k, v)| (Ustr::from(k.as_str()), Ustr::from(v.as_str())))
1163                    .collect()
1164            });
1165        let exec_spawn_id = row
1166            .try_get::<Option<&str>, _>("exec_spawn_id")
1167            .ok()
1168            .and_then(|x| x.map(ClientOrderId::from));
1169        let tags = tags_from_row(row);
1170        let init_id = row.try_get::<&str, _>("init_id").map(UUID4::from)?;
1171        let ts_init = row.try_get::<String, _>("ts_init").map(UnixNanos::from)?;
1172        let ts_last = row.try_get::<String, _>("ts_last").map(UnixNanos::from)?;
1173
1174        let snapshot = OrderSnapshot {
1175            trader_id,
1176            strategy_id,
1177            instrument_id,
1178            client_order_id,
1179            venue_order_id,
1180            position_id,
1181            account_id,
1182            last_trade_id,
1183            order_type,
1184            order_side,
1185            quantity,
1186            price,
1187            activation_price,
1188            trigger_price,
1189            trigger_type,
1190            limit_offset,
1191            trailing_offset,
1192            trailing_offset_type,
1193            time_in_force,
1194            expire_time,
1195            filled_qty,
1196            liquidity_side,
1197            avg_px,
1198            slippage,
1199            commissions,
1200            status,
1201            is_post_only,
1202            is_reduce_only,
1203            is_quote_quantity,
1204            display_qty,
1205            emulation_trigger,
1206            trigger_instrument_id,
1207            contingency_type,
1208            order_list_id,
1209            linked_order_ids,
1210            parent_order_id,
1211            exec_algorithm_id,
1212            exec_algorithm_params,
1213            exec_spawn_id,
1214            tags,
1215            init_id,
1216            ts_init,
1217            ts_last,
1218            causation_id: None,
1219        };
1220
1221        Ok(Self(snapshot))
1222    }
1223}
1224
1225fn causation_id_from_row(row: &PgRow) -> Result<Option<UUID4>, sqlx::Error> {
1226    row.try_get::<Option<&str>, _>("causation_id")?
1227        .map(|value| UUID4::from_str(value).map_err(|e| sqlx::Error::Decode(e.into())))
1228        .transpose()
1229}
1230
1231fn decode_info(row: &PgRow) -> Result<Option<IndexMap<Ustr, Ustr>>, sqlx::Error> {
1232    let value = row.try_get::<Option<serde_json::Value>, _>("info")?;
1233    let Some(value) = value else {
1234        return Ok(None);
1235    };
1236
1237    let decoded: IndexMap<String, String> =
1238        serde_json::from_value(value).map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
1239
1240    Ok(Some(
1241        decoded
1242            .into_iter()
1243            .map(|(k, v)| (Ustr::from(k.as_str()), Ustr::from(v.as_str())))
1244            .collect(),
1245    ))
1246}
1247
1248fn tags_from_row(row: &PgRow) -> Option<Vec<Ustr>> {
1249    row.try_get::<Vec<String>, _>("tags")
1250        .ok()
1251        .map(|tags| tags.iter().map(|tag| Ustr::from(tag.as_str())).collect())
1252}
1253
1254fn parse_trigger_type(value: Option<&str>) -> Option<TriggerType> {
1255    value.and_then(|value| {
1256        if value.eq_ignore_ascii_case("NO_TRIGGER") {
1257            None
1258        } else {
1259            Some(TriggerType::from_str(value).expect("Invalid `TriggerType`"))
1260        }
1261    })
1262}
1263
1264fn parse_contingency_type(value: Option<&str>) -> Option<ContingencyType> {
1265    value.and_then(|value| {
1266        if value.eq_ignore_ascii_case("NO_CONTINGENCY") {
1267            None
1268        } else {
1269            Some(ContingencyType::from_str(value).expect("Invalid `ContingencyType`"))
1270        }
1271    })
1272}
1273
1274#[cfg(test)]
1275mod tests {
1276    use rstest::rstest;
1277
1278    use super::*;
1279
1280    #[rstest]
1281    #[case(None, None)]
1282    #[case(Some("NO_TRIGGER"), None)]
1283    #[case(Some("LAST_PRICE"), Some(TriggerType::LastPrice))]
1284    fn test_parse_trigger_type_accepts_legacy_absence(
1285        #[case] value: Option<&str>,
1286        #[case] expected: Option<TriggerType>,
1287    ) {
1288        assert_eq!(parse_trigger_type(value), expected);
1289    }
1290
1291    #[rstest]
1292    #[case(None, None)]
1293    #[case(Some("NO_CONTINGENCY"), None)]
1294    #[case(Some("OCO"), Some(ContingencyType::Oco))]
1295    fn test_parse_contingency_type_accepts_legacy_absence(
1296        #[case] value: Option<&str>,
1297        #[case] expected: Option<ContingencyType>,
1298    ) {
1299        assert_eq!(parse_contingency_type(value), expected);
1300    }
1301}
nautilus_infrastructure/sql/models/orders.rs

nautilus_infrastructure/sql/models/
orders.rs
