diff options
Diffstat (limited to 'src/FundLab.Api/Persistence.fs')
| -rw-r--r-- | src/FundLab.Api/Persistence.fs | 483 |
1 files changed, 479 insertions, 4 deletions
diff --git a/src/FundLab.Api/Persistence.fs b/src/FundLab.Api/Persistence.fs index 6c0374a..e252d0a 100644 --- a/src/FundLab.Api/Persistence.fs +++ b/src/FundLab.Api/Persistence.fs @@ -40,11 +40,24 @@ module ConfirmationPolicy = let isModeledTradingDay (date: DateOnly) : bool = not (isWeekend date) + /// Test-only clock anchor: modules may pin the trading-date clock by exporting + /// FUND_LAB_TEST_TRADE_DATE so suites stay independent of the wall clock. Unset in + /// production, the real Shanghai rule applies unchanged. 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 + match Environment.GetEnvironmentVariable("FUND_LAB_TEST_TRADE_DATE") with + | value when not (String.IsNullOrWhiteSpace value) -> + match DateOnly.TryParseExact(value, "yyyy-MM-dd", CultureInfo.InvariantCulture, DateTimeStyles.None) with + | true, anchored -> anchored + | _ -> + 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 + | _ -> + 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 @@ -353,6 +366,50 @@ type RebalanceExecutionResult = Outcomes: RebalanceOrderOutcome list } +type DividendMode = DividendPolicy.DividendMode + +type DividendCommand = + { + InstrumentCode: string + NavDate: DateOnly + Dps: decimal + Mode: DividendMode + } + +type DividendRecord = + { + Id: Guid + FundId: Guid + InstrumentCode: string + NavDate: DateOnly + Dps: decimal + Mode: DividendMode + Status: string + IsSynthetic: bool + GrossCash: decimal option + CreditedUnits: decimal option + CreditedInvested: decimal option + OrderId: Guid option + PendingReason: string option + CreatedAt: DateTimeOffset + } + +type DividendWriteResult = + | DividendCredited of DividendRecord + | DividendReplayed of DividendRecord + | DividendPendingReinvest of DividendRecord + | DividendIdempotencyConflict + | DividendInvalid of string + | DividendFundNotFound + | DividendInstrumentNotFound + | DividendNoHoldings + +type DividendStageDecision = + | StageReplay of Guid + | StageFail of DividendWriteResult + | StageBooked of DividendRecord + + type CapitalDepositCommand = { Amount: decimal @@ -626,6 +683,31 @@ type FundRepository(connectionString: string) = target_percent numeric(9, 2) NOT NULL CHECK (target_percent > 0 AND target_percent <= 100), PRIMARY KEY (plan_id, instrument_code) ); + + CREATE TABLE IF NOT EXISTS dividend_records ( + id uuid PRIMARY KEY, + fund_id uuid NOT NULL REFERENCES funds(id), + instrument_code text NOT NULL REFERENCES instruments(code), + nav_date date NOT NULL, + dps numeric(28, 8) NOT NULL CHECK (dps > 0), + mode text NOT NULL, + status text NOT NULL, + is_synthetic boolean NOT NULL, + gross_cash numeric(20, 2) NULL, + credited_units numeric(28, 8) NULL, + credited_invested numeric(20, 2) NULL, + order_id uuid NULL, + pending_reason text NULL, + created_at timestamptz NOT NULL DEFAULT now() + ); + + CREATE TABLE IF NOT EXISTS dividend_idempotencies ( + idempotency_key text PRIMARY KEY, + request_hash text NOT NULL, + record_id uuid NOT NULL REFERENCES dividend_records(id), + fund_id uuid NOT NULL REFERENCES funds(id), + created_at timestamptz NOT NULL DEFAULT now() + ); """ let statusText status = @@ -1703,6 +1785,155 @@ type FundRepository(connectionString: string) = match RebalancePolicy.validateTargets command.Targets with | Error message -> Error message | Ok() -> Ok() + let dividendRecordFromReader (reader: DbDataReader) : DividendRecord = + { + Id = reader.GetGuid(0) + FundId = reader.GetGuid(1) + InstrumentCode = reader.GetString(2) + NavDate = reader.GetFieldValue<DateOnly>(3) + Dps = reader.GetDecimal(4) + Mode = + match DividendPolicy.parseMode (reader.GetString(5)) with + | Some mode -> mode + | None -> failwith "dividend mode is invalid" + Status = reader.GetString(6) + IsSynthetic = reader.GetBoolean(7) + GrossCash = readDecimalOption reader 8 + CreditedUnits = readDecimalOption reader 9 + CreditedInvested = readDecimalOption reader 10 + OrderId = if reader.IsDBNull(11) then None else Some(reader.GetGuid(11)) + PendingReason = readStringOption reader 12 + CreatedAt = reader.GetFieldValue<DateTimeOffset>(13) + } + + let findDividendRecord connection transaction recordId = + use command = + commandWithTransaction + connection + transaction + """ + SELECT id, fund_id, instrument_code, nav_date, dps, mode, status, is_synthetic, + gross_cash, credited_units, credited_invested, order_id, pending_reason, created_at + FROM dividend_records WHERE id = @record_id + """ + + addParameter command "record_id" NpgsqlDbType.Uuid (box recordId) |> ignore + + use reader = command.ExecuteReader() + if reader.Read() then Some(dividendRecordFromReader reader) else None + + let findDividendIdempotency connection transaction key = + use command = + commandWithTransaction + connection + transaction + "SELECT request_hash, fund_id, record_id FROM dividend_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 insertDividendRecord connection transaction (record: DividendRecord) = + use command = + commandWithTransaction + connection + transaction + """ + INSERT INTO dividend_records + (id, fund_id, instrument_code, nav_date, dps, mode, status, is_synthetic, gross_cash) + VALUES (@id, @fund_id, @instrument_code, @nav_date, @dps, @mode, @status, @is_synthetic, @gross_cash) + RETURNING created_at + """ + + addParameter command "id" NpgsqlDbType.Uuid (box record.Id) |> ignore + addParameter command "fund_id" NpgsqlDbType.Uuid (box record.FundId) |> ignore + addParameter command "instrument_code" NpgsqlDbType.Text (box record.InstrumentCode) |> ignore + addParameter command "nav_date" NpgsqlDbType.Date (box record.NavDate) |> ignore + addParameter command "dps" NpgsqlDbType.Numeric (box record.Dps) |> ignore + addParameter command "mode" NpgsqlDbType.Text (box (DividendPolicy.modeText record.Mode)) |> ignore + addParameter command "status" NpgsqlDbType.Text (box record.Status) |> ignore + addParameter command "is_synthetic" NpgsqlDbType.Boolean (box record.IsSynthetic) |> ignore + + let grossParameter = + match record.GrossCash with + | Some gross -> box gross + | None -> box DBNull.Value + + addParameter command "gross_cash" NpgsqlDbType.Numeric grossParameter |> ignore + + use reader = command.ExecuteReader() + reader.Read() |> ignore + reader.GetFieldValue<DateTimeOffset>(0) + + let insertDividendIdempotency connection transaction key requestHash recordId fundId = + use command = + commandWithTransaction + connection + transaction + """ + INSERT INTO dividend_idempotencies (idempotency_key, request_hash, record_id, fund_id) + VALUES (@idempotency_key, @request_hash, @record_id, @fund_id) + """ + + addParameter command "idempotency_key" NpgsqlDbType.Text (box key) |> ignore + addParameter command "request_hash" NpgsqlDbType.Text (box requestHash) |> ignore + addParameter command "record_id" NpgsqlDbType.Uuid (box recordId) |> ignore + addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore + command.ExecuteNonQuery() |> ignore + + let finishDividendRecord connection transaction (recordId: Guid) (status: string) (pendingReason: string option) = + use command = + commandWithTransaction + connection + transaction + "UPDATE dividend_records SET status = @status, pending_reason = @reason WHERE id = @record_id" + + addParameter command "status" NpgsqlDbType.Text (box status) |> ignore + + let reasonParameter = + match pendingReason with + | Some reason -> box reason + | None -> box DBNull.Value + + addParameter command "reason" NpgsqlDbType.Text reasonParameter |> ignore + addParameter command "record_id" NpgsqlDbType.Uuid (box recordId) |> ignore + command.ExecuteNonQuery() |> ignore + + let dividendRequestHash (fundId: Guid) (command: DividendCommand) = + let invariant = CultureInfo.InvariantCulture + let encoded (value: string) = sprintf "%d:%s" value.Length value + + let payload = + String.concat + "|" + [ + "dividend" + encoded (fundId.ToString("D")) + encoded command.InstrumentCode + encoded (command.NavDate.ToString("yyyy-MM-dd")) + (encoded (command.Dps.ToString("G29", invariant))) + (encoded (DividendPolicy.modeText command.Mode)) + ] + + Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(payload))) + + let dividendSchemeKey (fundId: Guid) (command: DividendCommand) = + sprintf "dividend-scheme:%O:%s:%s:%s" fundId command.InstrumentCode (command.NavDate.ToString("yyyy-MM-dd")) (DividendPolicy.modeText command.Mode) + + let validateDividendCommand (today: DateOnly) (command: DividendCommand) = + if String.IsNullOrWhiteSpace command.InstrumentCode then + Error "instrument code cannot be empty" + elif command.Dps <= 0m then + Error "dividend per unit must be positive" + else + match DividendPolicy.validateDps command.Dps, DividendPolicy.validateNavDate command.NavDate today with + | Ok(), Ok() -> Ok() + | Error message, _ + | _, Error message -> Error message member _.EnsureSchema() = use connection = new NpgsqlConnection(connectionString) connection.Open() @@ -2643,6 +2874,250 @@ type FundRepository(connectionString: string) = Outcomes = outcomes |> Seq.toList } + member this.RegisterDividend(idempotencyKey: string, fundId: Guid, command: DividendCommand) : DividendWriteResult = + if String.IsNullOrWhiteSpace idempotencyKey then + DividendWriteResult.DividendInvalid "idempotency key cannot be empty" + else + let today = ConfirmationPolicy.tradeDateFor DateTimeOffset.UtcNow + + match validateDividendCommand today command with + | Error message -> DividendWriteResult.DividendInvalid message + | Ok() -> + let schemeKey = dividendSchemeKey fundId command + let fingerprint = dividendRequestHash fundId command + use connection = new NpgsqlConnection(connectionString) + connection.Open() + + // Phase 1: claim the scheme atomically, validate, and book the payout + let claimedStage = + 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 schemeKey) |> ignore + lockCommand.ExecuteNonQuery() |> ignore + + match findDividendIdempotency connection (Some transaction) schemeKey with + | Some(existingHash, existingFundId, recordId) when existingFundId = fundId -> + if existingHash = fingerprint then + transaction.Commit() + StageReplay recordId + else + transaction.Rollback() + StageFail DividendWriteResult.DividendIdempotencyConflict + | Some _ -> + transaction.Rollback() + StageFail DividendWriteResult.DividendIdempotencyConflict + | None -> + match lockFundForOrder connection (Some transaction) fundId with + | None -> + transaction.Rollback() + StageFail DividendWriteResult.DividendFundNotFound + | Some isSynthetic -> + if not (instrumentExists connection (Some transaction) command.InstrumentCode) then + transaction.Rollback() + StageFail DividendWriteResult.DividendInstrumentNotFound + else + use positionCommand = + commandWithTransaction + connection + (Some transaction) + "SELECT units FROM fund_positions WHERE fund_id = @fund_id AND instrument_code = @code FOR UPDATE" + + addParameter positionCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore + addParameter positionCommand "code" NpgsqlDbType.Text (box command.InstrumentCode) |> ignore + + use positionReader = positionCommand.ExecuteReader() + let positionFound = positionReader.Read() + let heldUnits = if positionFound then positionReader.GetDecimal(0) else 0m + positionReader.Close() + + if not positionFound || heldUnits <= 0m then + transaction.Rollback() + StageFail DividendWriteResult.DividendNoHoldings + else + match DividendPolicy.computeCashPayout heldUnits command.Dps with + | Error message -> + transaction.Rollback() + StageFail (DividendWriteResult.DividendInvalid message) + | Ok payout -> + use cashCommand = + commandWithTransaction + connection + (Some transaction) + "UPDATE funds SET available_cash = available_cash + @gross WHERE id = @fund_id" + + addParameter cashCommand "gross" NpgsqlDbType.Numeric (box payout.GrossCash) |> ignore + addParameter cashCommand "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore + + if cashCommand.ExecuteNonQuery() = 0 then + transaction.Rollback() + StageFail DividendWriteResult.DividendFundNotFound + else + let recordId = Guid.NewGuid() + + let record: DividendRecord = + { + Id = recordId + FundId = fundId + InstrumentCode = command.InstrumentCode + NavDate = command.NavDate + Dps = command.Dps + Mode = command.Mode + Status = (if command.Mode = DividendPolicy.Cash then "cash_credited" else "pending_nav") + IsSynthetic = isSynthetic + GrossCash = Some payout.GrossCash + CreditedUnits = None + CreditedInvested = None + OrderId = None + PendingReason = None + CreatedAt = DateTimeOffset.UtcNow + } + + let createdAt = insertDividendRecord connection (Some transaction) record + insertDividendIdempotency connection (Some transaction) schemeKey fingerprint recordId fundId + transaction.Commit() + StageBooked { record with CreatedAt = createdAt } + with error -> + try + transaction.Rollback() + with _ -> + () + + raise error + + // Confirmation of the reinvestment order settles the dividend record: the + // credited units/invested cash are persisted and the row is marked succeeded. + let finalizeDividend (record: DividendRecord) (confirmed: SubscriptionOrderRecord) = + let units = confirmed.ConfirmedUnits |> Option.defaultValue 0m + let invested = confirmed.ConfirmedInvestedCash |> Option.defaultValue 0m + + do + use finalizeConnection = new NpgsqlConnection(connectionString) + finalizeConnection.Open() + + use finalizeTransaction = finalizeConnection.BeginTransaction(IsolationLevel.ReadCommitted) + + use finalizeCommand = + commandWithTransaction + finalizeConnection + (Some finalizeTransaction) + """ + UPDATE dividend_records + SET status = 'succeeded', + credited_units = @units, + credited_invested = @invested, + pending_reason = NULL + WHERE id = @record_id + """ + + addParameter finalizeCommand "units" NpgsqlDbType.Numeric (box units) |> ignore + addParameter finalizeCommand "invested" NpgsqlDbType.Numeric (box invested) |> ignore + addParameter finalizeCommand "record_id" NpgsqlDbType.Uuid (box record.Id) |> ignore + finalizeCommand.ExecuteNonQuery() |> ignore + finalizeTransaction.Commit() + + { record with Status = "succeeded"; CreditedUnits = Some units; CreditedInvested = Some invested } + + // Phase 2: replays never re-book; pending reinvestments retry their confirmation + match claimedStage with + | StageReplay recordId -> + let record = + match findDividendRecord connection None recordId with + | Some record -> record + | None -> failwith "idempotency scheme references a missing dividend" + + match record.Status, record.OrderId with + | "pending_nav", Some orderId -> + let confirmKey = sprintf "dividend-confirm:%O:%s:%s:%s" fundId command.InstrumentCode (command.NavDate.ToString("yyyy-MM-dd")) (command.Dps.ToString("G29", CultureInfo.InvariantCulture)) + + match this.ConfirmSubscriptionOrder(confirmKey, fundId, orderId) with + | SubscriptionConfirmResult.OrderConfirmed confirmed + | SubscriptionConfirmResult.ConfirmReplayed confirmed -> + DividendCredited(finalizeDividend record confirmed) + | SubscriptionConfirmResult.ConfirmPendingNav _ -> + DividendReplayed record + | other -> + failwithf "unexpected dividend confirm replay result: %A" other + | _ -> + DividendReplayed record + | StageFail failure -> failure + | StageBooked record -> + if record.Mode = DividendPolicy.Cash then + DividendCredited record + else + let orderKey = sprintf "dividend:%O:%s:%s:%s" fundId command.InstrumentCode (command.NavDate.ToString("yyyy-MM-dd")) (command.Dps.ToString("G29", CultureInfo.InvariantCulture)) + let confirmKey = sprintf "dividend-confirm:%O:%s:%s:%s" fundId command.InstrumentCode (command.NavDate.ToString("yyyy-MM-dd")) (command.Dps.ToString("G29", CultureInfo.InvariantCulture)) + + match + this.CreateSubscriptionOrder( + orderKey, + fundId, + { FundCode = command.InstrumentCode; Amount = record.GrossCash |> Option.defaultValue 0m; FeeAmount = 0m }, + command.NavDate + ) + with + | SubscriptionOrderWriteResult.OrderCreated order + | SubscriptionOrderWriteResult.OrderReplayed order -> + do + use bindConnection = new NpgsqlConnection(connectionString) + bindConnection.Open() + + use bindTransaction = bindConnection.BeginTransaction(IsolationLevel.ReadCommitted) + + use bindCommand = + commandWithTransaction + bindConnection + (Some bindTransaction) + "UPDATE dividend_records SET order_id = @order_id WHERE id = @record_id" + + addParameter bindCommand "order_id" NpgsqlDbType.Uuid (box order.Id) |> ignore + addParameter bindCommand "record_id" NpgsqlDbType.Uuid (box record.Id) |> ignore + bindCommand.ExecuteNonQuery() |> ignore + bindTransaction.Commit() + + let recordWithOrder = { record with OrderId = Some order.Id } + + match this.ConfirmSubscriptionOrder(confirmKey, fundId, order.Id) with + | SubscriptionConfirmResult.OrderConfirmed confirmed + | SubscriptionConfirmResult.ConfirmReplayed confirmed -> + DividendCredited(finalizeDividend recordWithOrder confirmed) + | SubscriptionConfirmResult.ConfirmPendingNav pending -> + DividendPendingReinvest { recordWithOrder with PendingReason = pending.PendingReason } + | other -> + failwithf "unexpected dividend confirm result: %A" other + | other -> + failwithf "unexpected dividend order result: %A" other + + member _.GetDividendRecords(fundId: Guid) = + use connection = new NpgsqlConnection(connectionString) + connection.Open() + + use command = + commandWithTransaction + connection + None + """ + SELECT id, fund_id, instrument_code, nav_date, dps, mode, status, is_synthetic, + gross_cash, credited_units, credited_invested, order_id, pending_reason, created_at + FROM dividend_records WHERE fund_id = @fund_id ORDER BY created_at DESC, id + """ + + addParameter command "fund_id" NpgsqlDbType.Uuid (box fundId) |> ignore + + use reader = command.ExecuteReader() + let records = ResizeArray<DividendRecord>() + + while reader.Read() do + records.Add(dividendRecordFromReader reader) + + records |> Seq.toList + member _.CreateCapitalDeposit(idempotencyKey: string, fundId: Guid, command: CapitalDepositCommand) : CapitalDepositWriteResult = if String.IsNullOrWhiteSpace idempotencyKey then CapitalDepositWriteResult.CapitalDepositInvalid "idempotency key cannot be empty" |