diff options
Diffstat (limited to 'src/FundLab.Api/Persistence.fs')
| -rw-r--r-- | src/FundLab.Api/Persistence.fs | 331 |
1 files changed, 327 insertions, 4 deletions
diff --git a/src/FundLab.Api/Persistence.fs b/src/FundLab.Api/Persistence.fs index e48d9b1..88c0350 100644 --- a/src/FundLab.Api/Persistence.fs +++ b/src/FundLab.Api/Persistence.fs @@ -28,6 +28,7 @@ type FundRecord = InitialUnitNav: decimal IsSynthetic: bool AvailableCash: decimal + ReservedCash: decimal Status: string } @@ -37,6 +38,36 @@ type FundWriteResult = | IdempotencyConflict | Invalid of string +[<CLIMutable>] +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 @@ -92,6 +123,28 @@ type FundRepository(connectionString: string) = 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 = @@ -109,6 +162,7 @@ type FundRepository(connectionString: string) = InitialUnitNav = fund.InitialUnitNav IsSynthetic = fund.IsSynthetic AvailableCash = fund.AvailableCash + ReservedCash = fund.FrozenCash Status = statusText fund.Status } @@ -121,7 +175,8 @@ type FundRepository(connectionString: string) = InitialUnitNav = reader.GetDecimal(4) IsSynthetic = reader.GetBoolean(5) AvailableCash = reader.GetDecimal(6) - Status = reader.GetString(7) + ReservedCash = reader.GetDecimal(7) + Status = reader.GetString(8) } let dateTimeOffsetFromReader (reader: DbDataReader) index = @@ -189,7 +244,7 @@ type FundRepository(connectionString: string) = transaction """ SELECT id, name, currency, initial_cash, initial_unit_nav, - is_synthetic, available_cash, status + is_synthetic, available_cash, reserved_cash, status FROM funds WHERE id = @fund_id """ @@ -219,10 +274,10 @@ type FundRepository(connectionString: string) = """ INSERT INTO funds (id, name, currency, initial_cash, initial_unit_nav, - is_synthetic, available_cash, status) + is_synthetic, available_cash, reserved_cash, status) VALUES (@id, @name, @currency, @initial_cash, @initial_unit_nav, - @is_synthetic, @available_cash, @status) + @is_synthetic, @available_cash, @reserved_cash, @status) """ addParameter command "id" NpgsqlDbType.Uuid (box fund.Id) |> ignore @@ -232,6 +287,7 @@ type FundRepository(connectionString: string) = 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 @@ -250,6 +306,121 @@ type FundRepository(connectionString: string) = 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<DateTimeOffset>(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<DateTimeOffset>(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 @@ -371,6 +542,43 @@ type FundRepository(connectionString: string) = 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 @@ -552,3 +760,118 @@ type FundRepository(connectionString: string) = () 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<SubscriptionOrderRecord>() + + while reader.Read() do + records.Add(orderRecordFromReader reader) + + records |> Seq.toList |