namespace FundLab.Api open System open System.Data open System.Data.Common open System.Globalization open System.Security.Cryptography open System.Text open FundLab.Domain open Npgsql open NpgsqlTypes [] type FundCreateCommand = { Name: string InitialCash: decimal InitialUnitNav: decimal IsSynthetic: bool } type FundRecord = { Id: Guid Name: string Currency: string InitialCash: decimal InitialUnitNav: decimal IsSynthetic: bool AvailableCash: decimal ReservedCash: decimal Status: string } type FundWriteResult = | Created of FundRecord | Replayed of FundRecord | IdempotencyConflict | Invalid of string [] type SubscriptionOrderCommand = { FundCode: string Amount: decimal FeeAmount: decimal } type SubscriptionOrderRecord = { Id: Guid FundId: Guid FundCode: string Amount: decimal FeeAmount: decimal ReservedTotal: decimal Status: string SubmittedAt: DateTimeOffset IsSynthetic: bool } type SubscriptionOrderWriteResult = | OrderCreated of SubscriptionOrderRecord | OrderReplayed of SubscriptionOrderRecord | OrderIdempotencyConflict | OrderInvalid of string | OrderFundNotFound | OrderInstrumentNotFound | OrderInsufficientFunds type FundRepository(connectionString: string) = let cashMaximum = 999999999999999999.99m let unitNavMaximum = 99999999999999999999.99999999m let schema = """ CREATE TABLE IF NOT EXISTS funds ( id uuid PRIMARY KEY, name text NOT NULL, currency text NOT NULL, initial_cash numeric(20, 2) NOT NULL, initial_unit_nav numeric(28, 8) NOT NULL, is_synthetic boolean NOT NULL, available_cash numeric(20, 2) NOT NULL, status text NOT NULL, created_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS fund_idempotencies ( idempotency_key text PRIMARY KEY, request_hash text NOT NULL, fund_id uuid NOT NULL REFERENCES funds(id), created_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS instruments ( code text PRIMARY KEY CHECK (code ~ '^[0-9]{6}$'), name text NOT NULL, fund_type text NULL, source text NOT NULL, source_revision text NOT NULL, source_collected_at timestamptz NOT NULL, source_payload_hash text NOT NULL, first_seen_at timestamptz NOT NULL DEFAULT now(), last_seen_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS fund_nav_observations ( instrument_code text NOT NULL REFERENCES instruments(code), nav_date date NOT NULL, published_at timestamptz NULL, nav numeric(28, 8) NOT NULL, accumulated_nav numeric(28, 8) NULL, daily_return numeric(20, 8) NULL, source text NOT NULL, source_revision text NOT NULL, source_collected_at timestamptz NOT NULL, source_payload_hash text NOT NULL, first_seen_at timestamptz NOT NULL DEFAULT now(), last_seen_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (instrument_code, nav_date) ); CREATE INDEX IF NOT EXISTS fund_nav_observations_date_idx ON fund_nav_observations (instrument_code, nav_date); ALTER TABLE funds ADD COLUMN IF NOT EXISTS reserved_cash numeric(20, 2) NOT NULL DEFAULT 0; CREATE TABLE IF NOT EXISTS subscription_orders ( id uuid PRIMARY KEY, fund_id uuid NOT NULL REFERENCES funds(id), fund_code text NOT NULL, amount numeric(20, 2) NOT NULL CHECK (amount > 0), fee_amount numeric(20, 2) NOT NULL CHECK (fee_amount > 0), reserved_total numeric(20, 2) NOT NULL CHECK (reserved_total > 0), status text NOT NULL, is_synthetic boolean NOT NULL, submitted_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS subscription_order_idempotencies ( idempotency_key text PRIMARY KEY, request_hash text NOT NULL, order_id uuid NOT NULL REFERENCES subscription_orders(id), fund_id uuid NOT NULL REFERENCES funds(id), created_at timestamptz NOT NULL DEFAULT now() ); """ let statusText status = match status with | FundStatus.Empty -> "empty" | FundStatus.Active -> "active" | FundStatus.ZeroUnits -> "zero_units" let recordFromLedger (fund: LedgerFund) = { Id = fund.Id Name = fund.Name Currency = fund.Currency InitialCash = fund.InitialCash InitialUnitNav = fund.InitialUnitNav IsSynthetic = fund.IsSynthetic AvailableCash = fund.AvailableCash ReservedCash = fund.FrozenCash Status = statusText fund.Status } let recordFromReader (reader: DbDataReader) = { Id = reader.GetGuid(0) Name = reader.GetString(1) Currency = reader.GetString(2) InitialCash = reader.GetDecimal(3) InitialUnitNav = reader.GetDecimal(4) IsSynthetic = reader.GetBoolean(5) AvailableCash = reader.GetDecimal(6) ReservedCash = reader.GetDecimal(7) Status = reader.GetString(8) } let dateTimeOffsetFromReader (reader: DbDataReader) index = reader.GetFieldValue(index) let optionalDateTimeOffsetFromReader (reader: DbDataReader) index = if reader.IsDBNull(index) then None else Some(dateTimeOffsetFromReader reader index) let optionalDecimalFromReader (reader: DbDataReader) index = if reader.IsDBNull(index) then None else Some(reader.GetDecimal(index)) let instrumentRecordFromReader (reader: DbDataReader) : MarketDataInstrumentRecord = { Code = reader.GetString(0) Name = reader.GetString(1) FundType = if reader.IsDBNull(2) then None else Some(reader.GetString(2)) Source = reader.GetString(3) SourceRevision = reader.GetString(4) SourceCollectedAt = dateTimeOffsetFromReader reader 5 SourcePayloadHash = reader.GetString(6) FirstSeenAt = dateTimeOffsetFromReader reader 7 LastSeenAt = dateTimeOffsetFromReader reader 8 } let navRecordFromReader (reader: DbDataReader) : MarketDataNavRecord = { Code = reader.GetString(0) NavDate = reader.GetFieldValue(1) PublishedAt = optionalDateTimeOffsetFromReader reader 2 Nav = reader.GetDecimal(3) AccumulatedNav = optionalDecimalFromReader reader 4 DailyReturn = optionalDecimalFromReader reader 5 Source = reader.GetString(6) SourceRevision = reader.GetString(7) SourceCollectedAt = dateTimeOffsetFromReader reader 8 SourcePayloadHash = reader.GetString(9) FirstSeenAt = dateTimeOffsetFromReader reader 10 LastSeenAt = dateTimeOffsetFromReader reader 11 } let commandWithTransaction (connection: NpgsqlConnection) (transaction: NpgsqlTransaction option) sql = let command = connection.CreateCommand() command.CommandText <- sql match transaction with | Some value -> command.Transaction <- value | None -> () command let addParameter (command: NpgsqlCommand) name dbType (value: obj) = let parameter = command.Parameters.Add(name, dbType) parameter.Value <- value parameter let optionalParameterValue value = match value with | Some actual -> box actual | None -> box DBNull.Value let findFund connection transaction fundId = use command = commandWithTransaction connection transaction """ SELECT id, name, currency, initial_cash, initial_unit_nav, is_synthetic, available_cash, reserved_cash, status FROM funds WHERE id = @fund_id """ addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore use reader = command.ExecuteReader() if reader.Read() then Some(recordFromReader reader) else None let findIdempotency connection transaction key = use command = commandWithTransaction connection transaction "SELECT request_hash, fund_id FROM fund_idempotencies WHERE idempotency_key = @idempotency_key" addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore use reader = command.ExecuteReader() if reader.Read() then Some(reader.GetString(0), reader.GetGuid(1)) else None let insertFund connection transaction (fund: LedgerFund) = use command = commandWithTransaction connection transaction """ INSERT INTO funds (id, name, currency, initial_cash, initial_unit_nav, is_synthetic, available_cash, reserved_cash, status) VALUES (@id, @name, @currency, @initial_cash, @initial_unit_nav, @is_synthetic, @available_cash, @reserved_cash, @status) """ addParameter command "id" NpgsqlDbType.Uuid (box fund.Id) |> ignore addParameter command "name" NpgsqlDbType.Text (box fund.Name) |> ignore addParameter command "currency" NpgsqlDbType.Text (box fund.Currency) |> ignore addParameter command "initial_cash" NpgsqlDbType.Numeric (box fund.InitialCash) |> ignore addParameter command "initial_unit_nav" NpgsqlDbType.Numeric (box fund.InitialUnitNav) |> ignore addParameter command "is_synthetic" NpgsqlDbType.Boolean (box fund.IsSynthetic) |> ignore addParameter command "available_cash" NpgsqlDbType.Numeric (box fund.AvailableCash) |> ignore addParameter command "reserved_cash" NpgsqlDbType.Numeric (box fund.FrozenCash) |> ignore addParameter command "status" NpgsqlDbType.Text (box (statusText fund.Status)) |> ignore command.ExecuteNonQuery() |> ignore let insertIdempotency connection transaction key requestHash fundId = use command = commandWithTransaction connection transaction """ INSERT INTO fund_idempotencies (idempotency_key, request_hash, fund_id) VALUES (@idempotency_key, @request_hash, @fund_id) """ addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore addParameter command "request_hash" NpgsqlDbType.Text (box requestHash) |> ignore addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore command.ExecuteNonQuery() |> ignore let orderStatusText = "submitted" let orderRecordFromReader (reader: DbDataReader) : SubscriptionOrderRecord = { Id = reader.GetGuid(0) FundId = reader.GetGuid(1) FundCode = reader.GetString(2) Amount = reader.GetDecimal(3) FeeAmount = reader.GetDecimal(4) ReservedTotal = reader.GetDecimal(5) Status = reader.GetString(6) IsSynthetic = reader.GetBoolean(7) SubmittedAt = reader.GetFieldValue(8) } let findOrder connection transaction orderId = use command = commandWithTransaction connection transaction """ SELECT id, fund_id, fund_code, amount, fee_amount, reserved_total, status, is_synthetic, submitted_at FROM subscription_orders WHERE id = @order_id """ addParameter command "order_id" NpgsqlDbType.Uuid (box orderId) |> ignore use reader = command.ExecuteReader() if reader.Read() then Some(orderRecordFromReader reader) else None let findOrderIdempotency connection transaction key = use command = commandWithTransaction connection transaction "SELECT request_hash, fund_id, order_id FROM subscription_order_idempotencies WHERE idempotency_key = @idempotency_key" addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore use reader = command.ExecuteReader() if reader.Read() then Some(reader.GetString(0), reader.GetGuid(1), reader.GetGuid(2)) else None let instrumentExists connection transaction code = use command = commandWithTransaction connection transaction "SELECT 1 FROM instruments WHERE code = @code" addParameter command "code" NpgsqlDbType.Text (box code) |> ignore use reader = command.ExecuteReader() reader.Read() let lockFundForOrder connection transaction fundId = use command = commandWithTransaction connection transaction "SELECT is_synthetic FROM funds WHERE id = @fund_id FOR UPDATE" addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore use reader = command.ExecuteReader() if reader.Read() then Some(reader.GetBoolean(0)) else None let insertSubscriptionOrder connection transaction (order: SubscriptionOrderRecord) = use command = commandWithTransaction connection transaction """ INSERT INTO subscription_orders (id, fund_id, fund_code, amount, fee_amount, reserved_total, status, is_synthetic) VALUES (@id, @fund_id, @fund_code, @amount, @fee_amount, @reserved_total, @status, @is_synthetic) RETURNING submitted_at """ addParameter command "id" NpgsqlDbType.Uuid (box order.Id) |> ignore addParameter command "fund_id" NpgsqlDbType.Uuid (box order.FundId) |> ignore addParameter command "fund_code" NpgsqlDbType.Text (box order.FundCode) |> ignore addParameter command "amount" NpgsqlDbType.Numeric (box order.Amount) |> ignore addParameter command "fee_amount" NpgsqlDbType.Numeric (box order.FeeAmount) |> ignore addParameter command "reserved_total" NpgsqlDbType.Numeric (box order.ReservedTotal) |> ignore addParameter command "status" NpgsqlDbType.Text (box order.Status) |> ignore addParameter command "is_synthetic" NpgsqlDbType.Boolean (box order.IsSynthetic) |> ignore use reader = command.ExecuteReader() reader.Read() |> ignore reader.GetFieldValue(0) let insertOrderIdempotency connection transaction key requestHash orderId fundId = use command = commandWithTransaction connection transaction """ INSERT INTO subscription_order_idempotencies (idempotency_key, request_hash, order_id, fund_id) VALUES (@idempotency_key, @request_hash, @order_id, @fund_id) """ addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore addParameter command "request_hash" NpgsqlDbType.Text (box requestHash) |> ignore addParameter command "order_id" NpgsqlDbType.Uuid (box orderId) |> ignore addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore command.ExecuteNonQuery() |> ignore let upsertInstrument connection transaction (payload: MarketDataSearchPayload) payloadHash (instrument: MarketDataInstrument) = use command = commandWithTransaction connection transaction """ INSERT INTO instruments (code, name, fund_type, source, source_revision, source_collected_at, source_payload_hash) VALUES (@code, @name, @fund_type, @source, @source_revision, @source_collected_at, @source_payload_hash) ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name, fund_type = EXCLUDED.fund_type, source = EXCLUDED.source, source_revision = EXCLUDED.source_revision, source_collected_at = EXCLUDED.source_collected_at, source_payload_hash = EXCLUDED.source_payload_hash, last_seen_at = now() """ addParameter command "code" NpgsqlDbType.Text (box instrument.Code) |> ignore addParameter command "name" NpgsqlDbType.Text (box instrument.Name) |> ignore addParameter command "fund_type" NpgsqlDbType.Text (optionalParameterValue instrument.FundType) |> ignore addParameter command "source" NpgsqlDbType.Text (box payload.Source) |> ignore addParameter command "source_revision" NpgsqlDbType.Text (box payload.SourceRevision) |> ignore addParameter command "source_collected_at" NpgsqlDbType.TimestampTz (box payload.CollectedAt) |> ignore addParameter command "source_payload_hash" NpgsqlDbType.Text (box payloadHash) |> ignore command.ExecuteNonQuery() |> ignore let upsertNavObservation connection transaction (payload: MarketDataNavPayload) payloadHash (observation: MarketDataObservation) = use command = commandWithTransaction connection transaction """ INSERT INTO fund_nav_observations (instrument_code, nav_date, published_at, nav, accumulated_nav, daily_return, source, source_revision, source_collected_at, source_payload_hash) VALUES (@instrument_code, @nav_date, @published_at, @nav, @accumulated_nav, @daily_return, @source, @source_revision, @source_collected_at, @source_payload_hash) ON CONFLICT (instrument_code, nav_date) DO UPDATE SET published_at = EXCLUDED.published_at, nav = EXCLUDED.nav, accumulated_nav = EXCLUDED.accumulated_nav, daily_return = EXCLUDED.daily_return, source = EXCLUDED.source, source_revision = EXCLUDED.source_revision, source_collected_at = EXCLUDED.source_collected_at, source_payload_hash = EXCLUDED.source_payload_hash, last_seen_at = now() """ addParameter command "instrument_code" NpgsqlDbType.Text (box payload.Code) |> ignore addParameter command "nav_date" NpgsqlDbType.Date (box observation.NavDate) |> ignore addParameter command "published_at" NpgsqlDbType.TimestampTz (optionalParameterValue observation.PublishedAt) |> ignore addParameter command "nav" NpgsqlDbType.Numeric (box observation.Nav) |> ignore addParameter command "accumulated_nav" NpgsqlDbType.Numeric (optionalParameterValue observation.AccumulatedNav) |> ignore addParameter command "daily_return" NpgsqlDbType.Numeric (optionalParameterValue observation.DailyReturn) |> ignore addParameter command "source" NpgsqlDbType.Text (box payload.Source) |> ignore addParameter command "source_revision" NpgsqlDbType.Text (box payload.SourceRevision) |> ignore addParameter command "source_collected_at" NpgsqlDbType.TimestampTz (box payload.CollectedAt) |> ignore addParameter command "source_payload_hash" NpgsqlDbType.Text (box payloadHash) |> ignore command.ExecuteNonQuery() |> ignore let requestHash (command: FundCreateCommand) = let invariant = CultureInfo.InvariantCulture let encoded (value: string) = sprintf "%d:%s" value.Length value let name = if isNull command.Name then "" else command.Name let payload = String.concat "|" [ "fund-create" encoded name (encoded (command.InitialCash.ToString("G29", invariant))) (encoded (command.InitialUnitNav.ToString("G29", invariant))) (encoded (command.IsSynthetic.ToString())) ] Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payload))) let validateStorageRange (command: FundCreateCommand) = if command.InitialCash > cashMaximum then Error "initial cash exceeds database precision" elif command.InitialUnitNav > unitNavMaximum then Error "initial unit NAV exceeds database precision" else Ok() let orderRequestHash (fundId: Guid) (command: SubscriptionOrderCommand) = let invariant = CultureInfo.InvariantCulture let encoded (value: string) = sprintf "%d:%s" value.Length value let fundCode = if isNull command.FundCode then "" else command.FundCode let payload = String.concat "|" [ "subscription-order" encoded (fundId.ToString("D")) encoded fundCode (encoded (command.Amount.ToString("G29", invariant))) (encoded (command.FeeAmount.ToString("G29", invariant))) ] Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payload))) let validateOrderCommand (command: SubscriptionOrderCommand) = if String.IsNullOrWhiteSpace command.FundCode then Error "fund code cannot be empty" elif command.Amount <= 0m then Error "amount must be positive" elif command.FeeAmount <= 0m then Error "fee amount must be positive" elif Decimal.Round(command.Amount, 2) <> command.Amount then Error "amount exceeds cash precision" elif Decimal.Round(command.FeeAmount, 2) <> command.FeeAmount then Error "fee amount exceeds cash precision" elif command.Amount > cashMaximum then Error "amount exceeds database precision" elif command.FeeAmount > cashMaximum then Error "fee amount exceeds database precision" elif command.Amount + command.FeeAmount > cashMaximum then Error "reserved total exceeds database precision" else Ok() let ledgerErrorMessage error = match error with | InvalidIdentifier label -> sprintf "%s is invalid" label | InvalidAmount label -> sprintf "%s is invalid" label | InvalidPrecision label -> sprintf "%s has invalid precision" label | InvalidState message -> message | FundAlreadyExists fundId -> sprintf "fund %O already exists" fundId | other -> sprintf "%A" other member _.EnsureSchema() = use connection = new NpgsqlConnection(connectionString) connection.Open() use command = connection.CreateCommand() command.CommandText <- schema command.ExecuteNonQuery() |> ignore member _.GetFund(fundId: Guid) = use connection = new NpgsqlConnection(connectionString) connection.Open() findFund connection None fundId member _.UpsertInstruments(payload: MarketDataSearchPayload, payloadHash: string) = use connection = new NpgsqlConnection(connectionString) connection.Open() use transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted) try for instrument in payload.Instruments do upsertInstrument connection (Some transaction) payload payloadHash instrument transaction.Commit() with error -> try transaction.Rollback() with _ -> () raise error member _.GetInstrument(code: string) = use connection = new NpgsqlConnection(connectionString) connection.Open() use command = commandWithTransaction connection None """ SELECT code, name, fund_type, source, source_revision, source_collected_at, source_payload_hash, first_seen_at, last_seen_at FROM instruments WHERE code = @code """ addParameter command "code" NpgsqlDbType.Text (box code) |> ignore use reader = command.ExecuteReader() if reader.Read() then Some(instrumentRecordFromReader reader) else None member _.UpsertNavObservations(payload: MarketDataNavPayload, payloadHash: string) = use connection = new NpgsqlConnection(connectionString) connection.Open() use transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted) try for observation in payload.Observations do upsertNavObservation connection (Some transaction) payload payloadHash observation transaction.Commit() with error -> try transaction.Rollback() with _ -> () raise error member _.GetNav(code: string, fromDate: DateOnly option, toDate: DateOnly option) = use connection = new NpgsqlConnection(connectionString) connection.Open() let conditions = ResizeArray() conditions.Add("instrument_code = @instrument_code") if fromDate.IsSome then conditions.Add("nav_date >= @from_date") if toDate.IsSome then conditions.Add("nav_date <= @to_date") use command = commandWithTransaction connection None (sprintf """ SELECT instrument_code, nav_date, published_at, nav, accumulated_nav, daily_return, source, source_revision, source_collected_at, source_payload_hash, first_seen_at, last_seen_at FROM fund_nav_observations WHERE %s ORDER BY nav_date ASC """ (String.concat " AND " conditions)) addParameter command "instrument_code" NpgsqlDbType.Text (box code) |> ignore match fromDate with | Some value -> addParameter command "from_date" NpgsqlDbType.Date (box value) |> ignore | None -> () match toDate with | Some value -> addParameter command "to_date" NpgsqlDbType.Date (box value) |> ignore | None -> () use reader = command.ExecuteReader() let records = ResizeArray() while reader.Read() do records.Add(navRecordFromReader reader) records |> Seq.toList member _.CreateFund(idempotencyKey: string, command: FundCreateCommand) = if String.IsNullOrWhiteSpace idempotencyKey then FundWriteResult.Invalid "idempotency key cannot be empty" else match validateStorageRange command with | Error message -> FundWriteResult.Invalid message | Ok() -> let fingerprint = requestHash command use connection = new NpgsqlConnection(connectionString) connection.Open() use transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted) try use lockCommand = commandWithTransaction connection (Some transaction) "SELECT pg_advisory_xact_lock(hashtext(@lock_key))" addParameter lockCommand "lock_key" NpgsqlDbType.Text (box idempotencyKey) |> ignore lockCommand.ExecuteNonQuery() |> ignore match findIdempotency connection (Some transaction) idempotencyKey with | Some(existingHash, fundId) when existingHash = fingerprint -> match findFund connection (Some transaction) fundId with | Some fund -> transaction.Commit() FundWriteResult.Replayed fund | None -> transaction.Rollback() FundWriteResult.Invalid "idempotency record references a missing fund" | Some _ -> transaction.Rollback() FundWriteResult.IdempotencyConflict | None -> let fundId = Guid.NewGuid() match Ledger.initializeFund fundId command.Name command.InitialCash command.InitialUnitNav command.IsSynthetic Ledger.empty with | Error error -> transaction.Rollback() FundWriteResult.Invalid(ledgerErrorMessage error) | Ok state -> match Ledger.getFund fundId state with | Error error -> transaction.Rollback() FundWriteResult.Invalid(ledgerErrorMessage error) | Ok fund -> insertFund connection (Some transaction) fund insertIdempotency connection (Some transaction) idempotencyKey fingerprint fundId transaction.Commit() FundWriteResult.Created(recordFromLedger fund) with error -> try transaction.Rollback() with _ -> () raise error member _.CreateSubscriptionOrder(idempotencyKey: string, fundId: Guid, command: SubscriptionOrderCommand) = if String.IsNullOrWhiteSpace idempotencyKey then OrderInvalid "idempotency key cannot be empty" else match validateOrderCommand command with | Error message -> OrderInvalid message | Ok() -> let fingerprint = orderRequestHash fundId command let reservedTotal = command.Amount + command.FeeAmount use connection = new NpgsqlConnection(connectionString) connection.Open() use transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted) try use lockCommand = commandWithTransaction connection (Some transaction) "SELECT pg_advisory_xact_lock(hashtext(@lock_key))" addParameter lockCommand "lock_key" NpgsqlDbType.Text (box idempotencyKey) |> ignore lockCommand.ExecuteNonQuery() |> ignore match findOrderIdempotency connection (Some transaction) idempotencyKey with | Some(existingHash, existingFundId, orderId) when existingHash = fingerprint && existingFundId = fundId -> match findOrder connection (Some transaction) orderId with | Some order -> transaction.Commit() OrderReplayed order | None -> transaction.Rollback() OrderInvalid "idempotency record references a missing order" | Some _ -> transaction.Rollback() OrderIdempotencyConflict | None -> match lockFundForOrder connection (Some transaction) fundId with | None -> transaction.Rollback() OrderFundNotFound | Some isSynthetic -> if instrumentExists connection (Some transaction) command.FundCode then use cashCommand = commandWithTransaction connection (Some transaction) """ UPDATE funds SET available_cash = available_cash - @reserved_total, reserved_cash = reserved_cash + @reserved_total WHERE id = @fund_id AND available_cash >= @reserved_total """ addParameter cashCommand "reserved_total" NpgsqlDbType.Numeric (box reservedTotal) |> ignore addParameter cashCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore if cashCommand.ExecuteNonQuery() = 0 then transaction.Rollback() OrderInsufficientFunds else let order: SubscriptionOrderRecord = { Id = Guid.NewGuid() FundId = fundId FundCode = command.FundCode Amount = command.Amount FeeAmount = command.FeeAmount ReservedTotal = reservedTotal Status = orderStatusText SubmittedAt = DateTimeOffset.UnixEpoch IsSynthetic = isSynthetic } let submittedAt = insertSubscriptionOrder connection (Some transaction) order insertOrderIdempotency connection (Some transaction) idempotencyKey fingerprint order.Id fundId transaction.Commit() OrderCreated { order with SubmittedAt = submittedAt } else transaction.Rollback() OrderInstrumentNotFound with error -> try transaction.Rollback() with _ -> () raise error member _.GetSubscriptionOrders(fundId: Guid) = use connection = new NpgsqlConnection(connectionString) connection.Open() use command = commandWithTransaction connection None """ SELECT id, fund_id, fund_code, amount, fee_amount, reserved_total, status, is_synthetic, submitted_at FROM subscription_orders WHERE fund_id = @fund_id ORDER BY submitted_at DESC, id """ addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore use reader = command.ExecuteReader() let records = ResizeArray() while reader.Read() do records.Add(orderRecordFromReader reader) records |> Seq.toList