path: root/src/FundLab.Api/Persistence.fs

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
open System.Text.Json

module ConfirmationPolicy =
    let cutoffTimeOfDay = TimeSpan(15, 0, 0)
    let unitsMaximum = 99999999999999999999.99999999m
    let amountMaximum = 999999999999999999.99m

    let private shanghaiZone =
        lazy (
            try
                TimeZoneInfo.FindSystemTimeZoneById("Asia/Shanghai")
            with _ ->
                TimeZoneInfo.CreateCustomTimeZone(
                    "Asia/Shanghai (UTC+8)",
                    TimeSpan(8, 0, 0),
                    "Asia/Shanghai (UTC+8)",
                    "Asia/Shanghai (UTC+8)"
                )
        )

    let shanghaiDate (moment: DateTimeOffset) : DateOnly =
        DateOnly.FromDateTime(TimeZoneInfo.ConvertTime(moment, shanghaiZone.Value).Date)

    let private isWeekend (date: DateOnly) =
        date.DayOfWeek = DayOfWeek.Saturday || date.DayOfWeek = DayOfWeek.Sunday

    let rec private rollToWeekday (date: DateOnly) =
        if isWeekend date then rollToWeekday (date.AddDays 1) else date

    let isModeledTradingDay (date: DateOnly) : bool = not (isWeekend date)

    let tradeDateFor (submittedAt: DateTimeOffset) : DateOnly =
        let local = TimeZoneInfo.ConvertTime(submittedAt, shanghaiZone.Value)
        let date = DateOnly.FromDateTime(local.Date)
        let candidate = if local.TimeOfDay >= cutoffTimeOfDay then date.AddDays 1 else date
        rollToWeekday candidate

    type NavQuote =
        { NavDate: DateOnly
          Nav: decimal
          CollectedAt: DateTimeOffset
          PublishedAt: DateTimeOffset option }

    type ConfirmationComputation =
        {
            Units: decimal
            InvestedCash: decimal
            ResidualCash: decimal
        }

    let navDeferralReason
        (quote: NavQuote)
        (tradeDate: DateOnly)
        (today: DateOnly)
        (now: DateTimeOffset)
        : string option =
        let tradeDateText = tradeDate.ToString("yyyy-MM-dd")

        if quote.Nav <= 0m then
            Some(sprintf "nav for trade date %s is not positive" tradeDateText)
        elif quote.NavDate > today then
            Some(sprintf "nav for trade date %s is dated in the future" tradeDateText)
        elif quote.CollectedAt > now then
            Some(sprintf "nav for trade date %s is not yet collected" tradeDateText)
        elif quote.PublishedAt |> Option.exists (fun publishedAt -> publishedAt > now) then
            Some(sprintf "nav for trade date %s is not yet published" tradeDateText)
        elif quote.NavDate <> tradeDate then
            Some(sprintf "nav for trade date %s is not available yet" tradeDateText)
        else
            None

    let compute (amount: decimal) (nav: decimal) : Result<ConfirmationComputation, string> =
        if amount <= 0m then
            Error "amount must be positive"
        elif amount > amountMaximum then
            Error "amount exceeds supported precision"
        elif nav <= 0m then
            Error "unit nav must be positive"
        elif nav < 0.00000001m then
            Error "unit nav is below database precision"
        else
            let rawUnits = amount / nav

            if rawUnits > unitsMaximum then
                Error "unit amount exceeds database precision"
            else
                let units = Decimal.Truncate(rawUnits * 100000000m) / 100000000m

                if units <= 0m then
                    Error "amount converts to zero units at this unit nav"
                else
                    let invested = Decimal.Truncate(units * nav * 100m) / 100m

                    Ok
                        {
                            Units = units
                            InvestedCash = invested
                            ResidualCash = amount - invested
                        }

[<CLIMutable>]
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

[<CLIMutable>]
type SubscriptionOrderCommand =
    {
        FundCode: string
        Amount: decimal
        FeeAmount: decimal
    }

type ConfirmedQuoteEvidence =
    {
        Nav: decimal
        NavDate: DateOnly
        Source: string
        Revision: string
        CollectedAt: DateTimeOffset
        PublishedAt: DateTimeOffset option
        PayloadHash: string
        FirstSeenAt: DateTimeOffset
    }

type SubscriptionOrderRecord =
    {
        Id: Guid
        FundId: Guid
        FundCode: string
        Amount: decimal
        FeeAmount: decimal
        ReservedTotal: decimal
        Status: string
        SubmittedAt: DateTimeOffset
        IsSynthetic: bool
        TradeDate: DateOnly
        ConfirmIdempotencyKey: string option
        PendingReason: string option
        ConfirmedAt: DateTimeOffset option
        ConfirmedQuote: ConfirmedQuoteEvidence option
        ConfirmedUnits: decimal option
        ConfirmedInvestedCash: decimal option
        ConfirmedResidualCash: decimal option
    }

type SubscriptionOrderWriteResult =
    | OrderCreated of SubscriptionOrderRecord
    | OrderReplayed of SubscriptionOrderRecord
    | OrderIdempotencyConflict
    | OrderInvalid of string
    | OrderFundNotFound
    | OrderInstrumentNotFound
    | OrderInsufficientFunds

type SubscriptionConfirmResult =
    | OrderConfirmed of SubscriptionOrderRecord
    | ConfirmReplayed of SubscriptionOrderRecord
    | ConfirmPendingNav of SubscriptionOrderRecord
    | ConfirmIdempotencyConflict
    | ConfirmAlreadyConfirmed
    | ConfirmOrderNotFound
    | ConfirmInvalidStatus
    | ConfirmInvalid of string

type FundPositionRecord =
    {
        FundId: Guid
        InstrumentCode: string
        Units: decimal
        CostCash: decimal
        LastConfirmedAt: DateTimeOffset
        ValuationNav: decimal option
        ValuationNavDate: DateOnly option
        ValuationCollectedAt: DateTimeOffset option
    }

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);

        CREATE TABLE IF NOT EXISTS fund_nav_observation_evidence (
            instrument_code text NOT NULL REFERENCES instruments(code),
            nav_date date NOT NULL,
            source_payload_hash text NOT NULL,
            source_revision text NOT NULL,
            source text NOT NULL,
            source_collected_at timestamptz NOT NULL,
            published_at timestamptz NULL,
            nav numeric(28, 8) NOT NULL,
            first_seen_at timestamptz NOT NULL DEFAULT now(),
            PRIMARY KEY (instrument_code, nav_date, source_payload_hash)
        );

        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()
        );

        ALTER TABLE subscription_orders DROP CONSTRAINT IF EXISTS subscription_orders_fee_amount_check;
        ALTER TABLE subscription_orders ADD CONSTRAINT subscription_orders_fee_amount_check CHECK (fee_amount >= 0);

        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()
        );

        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirm_idempotency_key text NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS pending_reason text NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_at timestamptz NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav numeric(28, 8) NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_date date NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_source text NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_revision text NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_collected_at timestamptz NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_published_at timestamptz NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_units numeric(28, 8) NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_invested_cash numeric(20, 2) NULL;
        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_residual_cash numeric(20, 2) NULL;

        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_payload_hash text NULL;

        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS confirmed_nav_first_seen_at timestamptz NULL;

        ALTER TABLE subscription_orders ADD COLUMN IF NOT EXISTS trade_date date NULL;

        UPDATE subscription_orders SET trade_date = (submitted_at AT TIME ZONE 'Asia/Shanghai')::date WHERE trade_date IS NULL;

        CREATE TABLE IF NOT EXISTS fund_positions (
            fund_id uuid NOT NULL REFERENCES funds(id),
            instrument_code text NOT NULL REFERENCES instruments(code),
            units numeric(28, 8) NOT NULL CHECK (units > 0),
            cost_cash numeric(20, 2) NOT NULL CHECK (cost_cash >= 0),
            first_confirmed_at timestamptz NOT NULL,
            last_confirmed_at timestamptz NOT NULL,
            PRIMARY KEY (fund_id, instrument_code)
        );

        CREATE TABLE IF NOT EXISTS subscription_order_events (
            id bigserial PRIMARY KEY,
            order_id uuid NOT NULL REFERENCES subscription_orders(id),
            event_type text NOT NULL,
            detail jsonb NOT NULL,
            created_at timestamptz NOT NULL DEFAULT now()
        );

        CREATE TABLE IF NOT EXISTS subscription_confirm_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<DateTimeOffset>(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<DateOnly>(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 readStringOption (reader: DbDataReader) ordinal =
        if reader.IsDBNull ordinal then None else Some(reader.GetString ordinal)

    let readDecimalOption (reader: DbDataReader) ordinal =
        if reader.IsDBNull ordinal then None else Some(reader.GetDecimal ordinal)

    let orderRecordFromReader (reader: DbDataReader) : SubscriptionOrderRecord =
        let confirmedQuote =
            if reader.IsDBNull(13) then
                None
            else
                Some
                    {
                        Nav = reader.GetDecimal(13)
                        NavDate = reader.GetFieldValue<DateOnly>(14)
                        Source = reader.GetString(15)
                        Revision = reader.GetString(16)
                        CollectedAt = reader.GetFieldValue<DateTimeOffset>(17)
                        PublishedAt =
                            if reader.IsDBNull(18) then
                                None
                            else
                                Some(reader.GetFieldValue<DateTimeOffset>(18))
                        PayloadHash =
                            if reader.IsDBNull(19) then
                                ""
                            else
                                reader.GetString(19)
                        FirstSeenAt =
                            if reader.IsDBNull(20) then
                                DateTimeOffset.UnixEpoch
                            else
                                reader.GetFieldValue<DateTimeOffset>(20)
                    }

        {
            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)
            TradeDate = reader.GetFieldValue<DateOnly>(9)
            ConfirmIdempotencyKey = readStringOption reader 10
            PendingReason = readStringOption reader 11
            ConfirmedAt =
                if reader.IsDBNull(12) then
                    None
                else
                    Some(reader.GetFieldValue<DateTimeOffset>(12))
            ConfirmedUnits = readDecimalOption reader 21
            ConfirmedQuote = confirmedQuote
            ConfirmedInvestedCash = readDecimalOption reader 22
            ConfirmedResidualCash = readDecimalOption reader 23
        }

    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,
                       trade_date,
                       confirm_idempotency_key, pending_reason, confirmed_at,
                       confirmed_nav, confirmed_nav_date, confirmed_nav_source,
                       confirmed_nav_revision, confirmed_nav_collected_at, confirmed_nav_published_at,
                       confirmed_nav_payload_hash, confirmed_nav_first_seen_at,
                       confirmed_units, confirmed_invested_cash, confirmed_residual_cash
                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 findConfirmIdempotency connection transaction key =
        use command =
            commandWithTransaction
                connection
                transaction
                "SELECT request_hash, fund_id, order_id FROM subscription_confirm_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 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, trade_date)
                VALUES
                    (@id, @fund_id, @fund_code, @amount, @fee_amount,
                     @reserved_total, @status, @is_synthetic, @trade_date)
                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
        addParameter command "trade_date" NpgsqlDbType.Date (box order.TradeDate) |> 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
                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(),
                    first_seen_at =
                        CASE
                            WHEN fund_nav_observations.source_payload_hash = EXCLUDED.source_payload_hash
                                THEN fund_nav_observations.first_seen_at
                            ELSE now()
                        END
                """

        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

        use evidenceCommand =
            commandWithTransaction
                connection
                transaction
                """
                INSERT INTO fund_nav_observation_evidence
                    (instrument_code, nav_date, source_payload_hash, source_revision, source,
                     source_collected_at, published_at, nav)
                VALUES
                    (@instrument_code, @nav_date, @source_payload_hash, @source_revision, @source,
                     @source_collected_at, @published_at, @nav)
                ON CONFLICT (instrument_code, nav_date, source_payload_hash) DO NOTHING
                """

        addParameter evidenceCommand "instrument_code" NpgsqlDbType.Text (box payload.Code) |> ignore
        addParameter evidenceCommand "nav_date" NpgsqlDbType.Date (box observation.NavDate) |> ignore
        addParameter evidenceCommand "source_payload_hash" NpgsqlDbType.Text (box payloadHash) |> ignore
        addParameter evidenceCommand "source_revision" NpgsqlDbType.Text (box payload.SourceRevision) |> ignore
        addParameter evidenceCommand "source" NpgsqlDbType.Text (box payload.Source) |> ignore
        addParameter evidenceCommand "source_collected_at" NpgsqlDbType.TimestampTz (box payload.CollectedAt) |> ignore

        addParameter
            evidenceCommand
            "published_at"
            NpgsqlDbType.TimestampTz
            (optionalParameterValue observation.PublishedAt)
        |> ignore

        addParameter evidenceCommand "nav" NpgsqlDbType.Numeric (box observation.Nav) |> ignore
        evidenceCommand.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 cannot be negative"
        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<string>()
        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<MarketDataNavRecord>()

        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 submittedAt = DateTimeOffset.UtcNow
                                    let order: SubscriptionOrderRecord =
                                        {
                                            Id = Guid.NewGuid()
                                            FundId = fundId
                                            FundCode = command.FundCode
                                            Amount = command.Amount
                                            FeeAmount = command.FeeAmount
                                            ReservedTotal = reservedTotal
                                            Status = orderStatusText
                                            SubmittedAt = submittedAt
                                            IsSynthetic = isSynthetic
                                            TradeDate = ConfirmationPolicy.tradeDateFor submittedAt
                                            ConfirmIdempotencyKey = None
                                            PendingReason = None
                                            ConfirmedAt = None
                                            ConfirmedQuote = None
                                            ConfirmedUnits = None
                                            ConfirmedInvestedCash = None
                                            ConfirmedResidualCash = None
                                        }

                                    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,
                       trade_date,
                       confirm_idempotency_key, pending_reason, confirmed_at,
                       confirmed_nav, confirmed_nav_date, confirmed_nav_source,
                       confirmed_nav_revision, confirmed_nav_collected_at, confirmed_nav_published_at,
                       confirmed_nav_payload_hash, confirmed_nav_first_seen_at,
                       confirmed_units, confirmed_invested_cash, confirmed_residual_cash
                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

    member _.ConfirmSubscriptionOrder(idempotencyKey: string, fundId: Guid, orderId: Guid) =
        if String.IsNullOrWhiteSpace idempotencyKey then
            ConfirmInvalid "idempotency key cannot be empty"
        else
            use connection = new NpgsqlConnection(connectionString)
            connection.Open()
            use transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted)

            try
                let confirmedAt = DateTimeOffset.UtcNow
                let today = ConfirmationPolicy.shanghaiDate confirmedAt

                let requestHash =
                    Convert
                        .ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(sprintf "confirm:%O:%O" fundId orderId)))
                        .ToLowerInvariant()

                use lockCommand =
                    commandWithTransaction
                        connection
                        (Some transaction)
                        "SELECT pg_advisory_xact_lock(hashtext(@lock_key))"

                addParameter lockCommand "lock_key" NpgsqlDbType.Text (box (sprintf "confirm-order:%O" orderId)) |> ignore
                lockCommand.ExecuteNonQuery() |> ignore

                match findOrder connection (Some transaction) orderId with
                | None ->
                    transaction.Rollback()
                    ConfirmOrderNotFound
                | Some order when order.FundId <> fundId ->
                    transaction.Rollback()
                    ConfirmOrderNotFound
                | Some order ->
                    if order.Status = "confirmed" then
                        match order.ConfirmIdempotencyKey with
                        | Some storedKey when storedKey = idempotencyKey ->
                            transaction.Commit()
                            ConfirmReplayed order
                        | _ ->
                            match findConfirmIdempotency connection (Some transaction) idempotencyKey with
                            | Some(_, _, storedOrderId) when storedOrderId = orderId ->
                                transaction.Commit()
                                ConfirmReplayed order
                            | Some _ ->
                                transaction.Rollback()
                                ConfirmIdempotencyConflict
                            | None ->
                                transaction.Rollback()
                                ConfirmAlreadyConfirmed
                    elif order.Status <> "submitted" && order.Status <> "pending_nav" then
                        transaction.Rollback()
                        ConfirmInvalidStatus
                    else
                        match findConfirmIdempotency connection (Some transaction) idempotencyKey with
                        | Some _ ->
                            transaction.Rollback()
                            ConfirmIdempotencyConflict
                        | None ->
                            let tradeDate = order.TradeDate
                            let tradeDateText = tradeDate.ToString("yyyy-MM-dd")

                            use quoteCommand =
                                commandWithTransaction
                                    connection
                                    (Some transaction)
                                    """
                                    SELECT o.nav, o.nav_date, o.source, o.source_revision, o.source_collected_at, o.published_at,
                                           o.source_payload_hash, o.first_seen_at, e.first_seen_at
                                    FROM fund_nav_observations o
                                    LEFT JOIN fund_nav_observation_evidence e
                                        ON e.instrument_code = o.instrument_code
                                       AND e.nav_date = o.nav_date
                                       AND e.source_payload_hash = o.source_payload_hash
                                    WHERE o.instrument_code = @code AND o.nav_date = @trade_date AND o.nav > 0
                                    ORDER BY o.source_collected_at DESC, o.published_at DESC NULLS LAST, o.source_revision DESC
                                    LIMIT 1
                                    """

                            addParameter quoteCommand "code" NpgsqlDbType.Text (box order.FundCode) |> ignore
                            addParameter quoteCommand "trade_date" NpgsqlDbType.Date (box tradeDate) |> ignore

                            use quoteReader = quoteCommand.ExecuteReader()
                            let quoteFound = quoteReader.Read()

                            let selectedQuote, evidenceFirstSeen =
                                if quoteFound then
                                    let evidenceFirstSeen =
                                        if quoteReader.IsDBNull(8) then
                                            None
                                        else
                                            Some(quoteReader.GetFieldValue<DateTimeOffset>(8))

                                    Some
                                        {
                                            Nav = quoteReader.GetDecimal(0)
                                            NavDate = quoteReader.GetFieldValue<DateOnly>(1)
                                            Source = quoteReader.GetString(2)
                                            Revision = quoteReader.GetString(3)
                                            CollectedAt = quoteReader.GetFieldValue<DateTimeOffset>(4)
                                            PublishedAt =
                                                if quoteReader.IsDBNull(5) then
                                                    None
                                                else
                                                    Some(quoteReader.GetFieldValue<DateTimeOffset>(5))
                                            PayloadHash = quoteReader.GetString(6)
                                            FirstSeenAt = quoteReader.GetFieldValue<DateTimeOffset>(7)
                                        },
                                    evidenceFirstSeen
                                else
                                    None, None

                            quoteReader.Close()

                            let boundQuote =
                                match selectedQuote, evidenceFirstSeen with
                                | Some quote, Some firstSeen -> Some { quote with FirstSeenAt = firstSeen }
                                | _ -> None

                            let deferralReason =
                                if selectedQuote.IsSome && evidenceFirstSeen.IsNone then
                                    Some(
                                        sprintf
                                            "nav revision for trade date %s has no observation evidence recorded"
                                            tradeDateText
                                    )
                                else
                                    match boundQuote with
                                    | None -> Some(sprintf "nav for trade date %s is not available yet" tradeDateText)
                                    | Some quote when quote.FirstSeenAt > confirmedAt ->
                                        Some(
                                            sprintf
                                                "nav revision for trade date %s was first observed after the confirmation attempt"
                                                tradeDateText
                                        )
                                    | Some quote ->
                                        ConfirmationPolicy.navDeferralReason
                                            {
                                                Nav = quote.Nav
                                                NavDate = quote.NavDate
                                                CollectedAt = quote.CollectedAt
                                                PublishedAt = quote.PublishedAt
                                            }
                                            tradeDate
                                            today
                                            confirmedAt

                            let markPending reason =
                                use pendingCommand =
                                    commandWithTransaction
                                        connection
                                        (Some transaction)
                                        """
                                        UPDATE subscription_orders
                                        SET status = 'pending_nav',
                                            pending_reason = @reason,
                                            confirm_idempotency_key = NULL,
                                            confirmed_at = NULL,
                                            confirmed_nav = NULL,
                                            confirmed_nav_date = NULL,
                                            confirmed_nav_source = NULL,
                                            confirmed_nav_revision = NULL,
                                            confirmed_nav_collected_at = NULL,
                                            confirmed_nav_published_at = NULL,
                                            confirmed_nav_payload_hash = NULL,
                                            confirmed_nav_first_seen_at = NULL,
                                            confirmed_units = NULL,
                                            confirmed_invested_cash = NULL,
                                            confirmed_residual_cash = NULL
                                        WHERE id = @order_id
                                        """

                                addParameter pendingCommand "reason" NpgsqlDbType.Text (box reason) |> ignore
                                addParameter pendingCommand "order_id" NpgsqlDbType.Uuid (box orderId) |> ignore
                                pendingCommand.ExecuteNonQuery() |> ignore

                            let readPendingAfterCommit () =
                                match findOrder connection None orderId with
                                | Some pendingOrder -> ConfirmPendingNav pendingOrder
                                | None -> failwith "pending order disappeared after confirmation deferral"

                            match deferralReason with
                            | Some reason ->
                                markPending reason
                                transaction.Commit()
                                readPendingAfterCommit ()
                            | None ->
                                let quote = boundQuote |> Option.get
                                let source = quote.Source
                                let revision = quote.Revision

                                match ConfirmationPolicy.compute order.Amount quote.Nav with
                                | Error message ->
                                    markPending message
                                    transaction.Commit()
                                    readPendingAfterCommit ()
                                | Ok computation ->
                                    use cashCommand =
                                        commandWithTransaction
                                            connection
                                            (Some transaction)
                                            """
                                            UPDATE funds
                                            SET available_cash = available_cash + @residual,
                                                reserved_cash = reserved_cash - @reserved_total
                                            WHERE id = @fund_id AND reserved_cash >= @reserved_total
                                            """

                                    addParameter cashCommand "residual" NpgsqlDbType.Numeric (box computation.ResidualCash) |> ignore
                                    addParameter cashCommand "reserved_total" NpgsqlDbType.Numeric (box order.ReservedTotal) |> ignore
                                    addParameter cashCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore

                                    if cashCommand.ExecuteNonQuery() = 0 then
                                        failwith "insufficient reserved cash for order confirmation"

                                    let publishedParameter =
                                        match quote.PublishedAt with
                                        | Some published -> box published
                                        | None -> box DBNull.Value

                                    use confirmCommand =
                                        commandWithTransaction
                                            connection
                                            (Some transaction)
                                            """
                                            UPDATE subscription_orders
                                            SET status = 'confirmed',
                                                pending_reason = NULL,
                                                confirm_idempotency_key = @idempotency_key,
                                                confirmed_at = @confirmed_at,
                                                confirmed_nav = @nav,
                                                confirmed_nav_date = @nav_date,
                                                confirmed_nav_source = @source,
                                                confirmed_nav_revision = @revision,
                                                confirmed_nav_collected_at = @collected_at,
                                                confirmed_nav_published_at = @published_at,
                                                confirmed_nav_payload_hash = @payload_hash,
                                                confirmed_nav_first_seen_at = @first_seen_at,
                                                confirmed_units = @units,
                                                confirmed_invested_cash = @invested,
                                                confirmed_residual_cash = @residual
                                            WHERE id = @order_id
                                            """

                                    addParameter confirmCommand "idempotency_key" NpgsqlDbType.Text (box idempotencyKey) |> ignore
                                    addParameter confirmCommand "confirmed_at" NpgsqlDbType.TimestampTz (box confirmedAt) |> ignore
                                    addParameter confirmCommand "nav" NpgsqlDbType.Numeric (box quote.Nav) |> ignore
                                    addParameter confirmCommand "nav_date" NpgsqlDbType.Date (box quote.NavDate) |> ignore
                                    addParameter confirmCommand "source" NpgsqlDbType.Text (box source) |> ignore
                                    addParameter confirmCommand "revision" NpgsqlDbType.Text (box revision) |> ignore
                                    addParameter confirmCommand "collected_at" NpgsqlDbType.TimestampTz (box quote.CollectedAt) |> ignore
                                    addParameter confirmCommand "published_at" NpgsqlDbType.TimestampTz publishedParameter |> ignore
                                    addParameter confirmCommand "payload_hash" NpgsqlDbType.Text (box quote.PayloadHash) |> ignore
                                    addParameter confirmCommand "first_seen_at" NpgsqlDbType.TimestampTz (box quote.FirstSeenAt) |> ignore
                                    addParameter confirmCommand "units" NpgsqlDbType.Numeric (box computation.Units) |> ignore
                                    addParameter confirmCommand "invested" NpgsqlDbType.Numeric (box computation.InvestedCash) |> ignore
                                    addParameter confirmCommand "residual" NpgsqlDbType.Numeric (box computation.ResidualCash) |> ignore
                                    addParameter confirmCommand "order_id" NpgsqlDbType.Uuid (box orderId) |> ignore
                                    confirmCommand.ExecuteNonQuery() |> ignore

                                    use positionCommand =
                                        commandWithTransaction
                                            connection
                                            (Some transaction)
                                            """
                                            INSERT INTO fund_positions
                                                (fund_id, instrument_code, units, cost_cash, first_confirmed_at, last_confirmed_at)
                                            VALUES
                                                (@fund_id, @code, @units, @invested, @confirmed_at, @confirmed_at)
                                            ON CONFLICT (fund_id, instrument_code) DO UPDATE
                                            SET units = fund_positions.units + EXCLUDED.units,
                                                cost_cash = fund_positions.cost_cash + EXCLUDED.cost_cash,
                                                last_confirmed_at = EXCLUDED.last_confirmed_at
                                            """

                                    addParameter positionCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
                                    addParameter positionCommand "code" NpgsqlDbType.Text (box order.FundCode) |> ignore
                                    addParameter positionCommand "units" NpgsqlDbType.Numeric (box computation.Units) |> ignore
                                    addParameter positionCommand "invested" NpgsqlDbType.Numeric (box computation.InvestedCash) |> ignore
                                    addParameter positionCommand "confirmed_at" NpgsqlDbType.TimestampTz (box confirmedAt) |> ignore
                                    positionCommand.ExecuteNonQuery() |> ignore

                                    let publishedJson =
                                        match quote.PublishedAt with
                                        | Some published -> sprintf "\"%O\"" published
                                        | None -> "null"

                                    let detail =
                                        sprintf
                                            "{\"fund_id\":\"%O\",\"order_id\":\"%O\",\"units\":\"%M\",\"invested_cash\":\"%M\",\"residual_cash\":\"%M\",\"quote\":{\"nav_date\":\"%s\",\"nav\":\"%M\",\"source\":%s,\"source_revision\":%s,\"source_collected_at\":\"%O\",\"published_at\":%s,\"source_payload_hash\":%s,\"first_seen_at\":\"%O\"}}"
                                            fundId
                                            orderId
                                            computation.Units
                                            computation.InvestedCash
                                            computation.ResidualCash
                                            (quote.NavDate.ToString("yyyy-MM-dd"))
                                            quote.Nav
                                            (JsonSerializer.Serialize source)
                                            (JsonSerializer.Serialize revision)
                                            quote.CollectedAt
                                            publishedJson
                                            (JsonSerializer.Serialize quote.PayloadHash)
                                            quote.FirstSeenAt

                                    use eventCommand =
                                        commandWithTransaction
                                            connection
                                            (Some transaction)
                                            """
                                            INSERT INTO subscription_order_events (order_id, event_type, detail)
                                            VALUES (@order_id, 'confirmed', @detail)
                                            """

                                    addParameter eventCommand "order_id" NpgsqlDbType.Uuid (box orderId) |> ignore
                                    addParameter eventCommand "detail" NpgsqlDbType.Jsonb (box detail) |> ignore
                                    eventCommand.ExecuteNonQuery() |> ignore

                                    use idempotencyCommand =
                                        commandWithTransaction
                                            connection
                                            (Some transaction)
                                            """
                                            INSERT INTO subscription_confirm_idempotencies
                                                (idempotency_key, request_hash, order_id, fund_id)
                                            VALUES
                                                (@idempotency_key, @request_hash, @order_id, @fund_id)
                                            """

                                    addParameter idempotencyCommand "idempotency_key" NpgsqlDbType.Text (box idempotencyKey) |> ignore
                                    addParameter idempotencyCommand "request_hash" NpgsqlDbType.Text (box requestHash) |> ignore
                                    addParameter idempotencyCommand "order_id" NpgsqlDbType.Uuid (box orderId) |> ignore
                                    addParameter idempotencyCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore
                                    idempotencyCommand.ExecuteNonQuery() |> ignore

                                    transaction.Commit()

                                    match findOrder connection None orderId with
                                    | Some confirmedOrder -> OrderConfirmed confirmedOrder
                                    | None -> failwith "confirmed order disappeared after commit"
            with error ->
                try
                    transaction.Rollback()
                with _ ->
                    ()

                raise error

    member _.GetFundPositions(fundId: Guid) =
        use connection = new NpgsqlConnection(connectionString)
        connection.Open()

        use command =
            commandWithTransaction
                connection
                None
                """
                SELECT p.instrument_code, p.units, p.cost_cash, p.last_confirmed_at,
                       q.nav, q.nav_date, q.source_collected_at
                FROM fund_positions p
                LEFT JOIN LATERAL (
                    SELECT nav, nav_date, source_collected_at
                    FROM fund_nav_observations
                    WHERE instrument_code = p.instrument_code
                      AND nav_date <= (now() AT TIME ZONE 'Asia/Shanghai')::date
                      AND nav > 0
                      AND source_collected_at <= now()
                      AND (published_at IS NULL OR published_at <= now())
                    ORDER BY nav_date DESC, source_collected_at DESC
                    LIMIT 1
                ) q ON true
                WHERE p.fund_id = @fund_id
                ORDER BY p.instrument_code
                """

        addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore

        use reader = command.ExecuteReader()
        let records = ResizeArray<FundPositionRecord>()

        while reader.Read() do
            let valuationNav =
                if reader.IsDBNull(4) then None else Some(reader.GetDecimal(4))

            let valuationNavDate =
                if reader.IsDBNull(5) then None else Some(reader.GetFieldValue<DateOnly>(5))

            let valuationCollectedAt =
                if reader.IsDBNull(6) then None else Some(reader.GetFieldValue<DateTimeOffset>(6))

            records.Add(
                {
                    FundId = fundId
                    InstrumentCode = reader.GetString(0)
                    Units = reader.GetDecimal(1)
                    CostCash = reader.GetDecimal(2)
                    LastConfirmedAt = reader.GetFieldValue<DateTimeOffset>(3)
                    ValuationNav = valuationNav
                    ValuationNavDate = valuationNavDate
                    ValuationCollectedAt = valuationCollectedAt
                }
            )

        records |> Seq.toList