*/ private array $after_commit_callbacks = []; private function __construct() {} public static function get_instance(): self { if (self::$instance === null) { self::$instance = new self(); } return self::$instance; } public function transaction_active(): bool { return $this->transaction_active; } public function after_commit(callable $callback): bool { if (!$this->transaction_active) { return false; } $this->after_commit_callbacks[] = $callback; return true; } public function balance(int $user_id, AssetType|string $asset): int { global $wpdb; if ($user_id <= 0) { return 0; } $asset = AssetType::from_value($asset); $table = $this->table('asset_account'); return max(0, (int) $wpdb->get_var($wpdb->prepare( "SELECT balance_minor FROM {$table} WHERE user_id = %d AND asset = %s", $user_id, $asset->value ))); } public function balance_decimal(int $user_id): string { return AssetAmount::from_minor(AssetType::BALANCE, $this->balance($user_id, AssetType::BALANCE))->decimal(); } public function format_balance(int $minor): string { return AssetAmount::from_minor(AssetType::BALANCE, max(0, $minor))->formatted(); } public function result_for_operation(string $operation_key): ?LedgerResult { global $wpdb; if ($operation_key === '' || strlen($operation_key) > 191) { return null; } $transaction_id = (int) $wpdb->get_var($wpdb->prepare( "SELECT id FROM {$this->table('asset_transaction')} WHERE operation_key = %s LIMIT 1", $operation_key )); return $transaction_id > 0 ? $this->load_result($transaction_id, true) : null; } /** * @param array $ref * @param array $metadata */ public function credit( int $user_id, AssetType|string $asset, int $amount_minor, string $operation_key, string $reason, array $ref = [], string $remark = '', int $actor_user_id = 0, array $metadata = [], ?callable $business_mutation = null, ?callable $guard = null ) { try { $asset = AssetType::from_value($asset); } catch (\ValueError $e) { return new \WP_Error('aksrt_ledger_asset', __('资产类型无效', 'aksrt'), ['status' => 400]); } if ($amount_minor <= 0) { return new \WP_Error('aksrt_ledger_amount', __('入账金额必须大于零', 'aksrt'), ['status' => 400]); } return $this->post(new LedgerCommand( $operation_key, sanitize_key($reason), [[ 'user_id' => $user_id, 'asset' => $asset->value, 'delta_minor' => $amount_minor, 'entry_type' => 'credit', 'remark' => $remark, ]], $this->ref_type($ref), $this->ref_key($ref), $actor_user_id, $metadata ), $guard, $business_mutation); } /** * @param array $ref * @param array $metadata */ public function debit( int $user_id, AssetType|string $asset, int $amount_minor, string $operation_key, string $reason, array $ref = [], string $remark = '', int $actor_user_id = 0, array $metadata = [], ?callable $business_mutation = null, ?callable $guard = null ) { try { $asset = AssetType::from_value($asset); } catch (\ValueError $e) { return new \WP_Error('aksrt_ledger_asset', __('资产类型无效', 'aksrt'), ['status' => 400]); } if ($amount_minor <= 0) { return new \WP_Error('aksrt_ledger_amount', __('扣减金额必须大于零', 'aksrt'), ['status' => 400]); } return $this->post(new LedgerCommand( $operation_key, sanitize_key($reason), [[ 'user_id' => $user_id, 'asset' => $asset->value, 'delta_minor' => -$amount_minor, 'entry_type' => 'debit', 'remark' => $remark, ]], $this->ref_type($ref), $this->ref_key($ref), $actor_user_id, $metadata ), $guard, $business_mutation); } /** * @param array $ref * @param array $metadata */ public function transfer( int $sender_id, int $recipient_id, AssetType|string $asset, int $gross_minor, int $fee_minor, string $operation_key, string $reason, array $ref = [], string $sender_remark = '', string $recipient_remark = '', array $metadata = [] ) { try { $asset = AssetType::from_value($asset); } catch (\ValueError $e) { return new \WP_Error('aksrt_ledger_asset', __('资产类型无效', 'aksrt'), ['status' => 400]); } if ($sender_id <= 0 || $recipient_id <= 0 || $sender_id === $recipient_id || $gross_minor <= 0 || $fee_minor < 0 || $fee_minor >= $gross_minor) { return new \WP_Error('aksrt_ledger_transfer', __('转账参数无效', 'aksrt'), ['status' => 400]); } return $this->post(LedgerCommand::transfer( $operation_key, sanitize_key($reason), $sender_id, $recipient_id, $asset, $gross_minor, $fee_minor, $this->ref_type($ref), $this->ref_key($ref), $sender_remark, $recipient_remark, $metadata )); } /** * @param array $ref * @param array $metadata */ public function exchange( int $user_id, int $balance_minor, int $points, string $operation_key, string $reason, array $ref = [], string $remark = '', array $metadata = [], ?callable $guard = null, ?callable $business_mutation = null ) { if ($user_id <= 0 || $balance_minor <= 0 || $points <= 0) { return new \WP_Error('aksrt_ledger_exchange', __('兑换参数无效', 'aksrt'), ['status' => 400]); } return $this->post(new LedgerCommand( $operation_key, sanitize_key($reason), [ ['user_id' => $user_id, 'asset' => AssetType::BALANCE->value, 'delta_minor' => -$balance_minor, 'entry_type' => 'exchange_out', 'remark' => $remark], ['user_id' => $user_id, 'asset' => AssetType::POINTS->value, 'delta_minor' => $points, 'entry_type' => 'exchange_in', 'remark' => $remark], ], $this->ref_type($ref), $this->ref_key($ref), $user_id, $metadata ), $guard, $business_mutation); } /** * @param null|callable(int,array>):(true|\WP_Error) $guard * @param null|callable(int,LedgerResult):(true|\WP_Error) $business_mutation * @return LedgerResult|\WP_Error */ public function post(LedgerCommand $command, ?callable $guard = null, ?callable $business_mutation = null) { if ($this->transaction_active) { return new \WP_Error('aksrt_ledger_nested', __('账务事务内禁止再次提交账务', 'aksrt'), ['status' => 500]); } $valid = $command->validate(); if (is_wp_error($valid)) { return $valid; } $payload_hash = $command->payload_hash(); for ($attempt = 1; $attempt <= self::MAX_ATTEMPTS; $attempt++) { $result = $this->attempt_post($command, $payload_hash, $guard, $business_mutation); if (!is_wp_error($result) || $result->get_error_code() !== 'aksrt_ledger_retry') { return $result; } usleep(random_int(10000, 50000)); } return new \WP_Error('aksrt_ledger_db', __('账务处理失败,请稍后重试', 'aksrt'), ['status' => 500]); } private function attempt_post(LedgerCommand $command, string $payload_hash, ?callable $guard, ?callable $business_mutation) { global $wpdb; $account_table = $this->table('asset_account'); $transaction_table = $this->table('asset_transaction'); $entry_table = $this->table('asset_entry'); $created_at = self::utc_now(); if ($wpdb->query('START TRANSACTION') === false) { return new \WP_Error('aksrt_ledger_db', __('无法开启账务事务', 'aksrt'), ['status' => 500]); } $this->transaction_active = true; $this->after_commit_callbacks = []; try { $inserted = $wpdb->insert($transaction_table, [ 'operation_key' => $command->operation_key, 'payload_hash' => $payload_hash, 'reason' => $command->reason, 'ref_type' => $command->ref_type !== '' ? sanitize_key($command->ref_type) : null, 'ref_key' => $command->ref_key !== '' ? substr(sanitize_text_field($command->ref_key), 0, 191) : null, 'actor_user_id' => max(0, $command->actor_user_id), 'metadata' => $command->metadata !== [] ? wp_json_encode($command->metadata, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) : null, 'ip' => self::client_ip(), 'ua' => self::client_ua(), 'created_at' => $created_at, ], ['%s', '%s', '%s', '%s', '%s', '%d', '%s', '%s', '%s', '%s']); if ($inserted !== 1) { $errno = $this->last_db_errno(); $wpdb->query('ROLLBACK'); $this->transaction_active = false; $this->after_commit_callbacks = []; $existing = $this->replay($command->operation_key, $payload_hash); if ($existing !== null) { return $existing; } return $this->db_error($errno); } $transaction_id = (int) $wpdb->insert_id; $account_keys = []; foreach ($command->entries as $entry) { $key = (string) $entry['asset'] . ':' . (int) $entry['user_id']; $account_keys[$key] = ['user_id' => (int) $entry['user_id'], 'asset' => (string) $entry['asset']]; } $account_keys = array_values($account_keys); usort($account_keys, static function (array $left, array $right): int { $asset_order = strcmp((string) $left['asset'], (string) $right['asset']); return $asset_order !== 0 ? $asset_order : ((int) $left['user_id'] <=> (int) $right['user_id']); }); foreach ($account_keys as $account) { $wpdb->query($wpdb->prepare( "INSERT IGNORE INTO {$account_table} (user_id, asset, balance_minor, version, updated_at) VALUES (%d, %s, 0, 0, %s)", $account['user_id'], $account['asset'], $created_at )); } $balances = []; foreach ($account_keys as $account) { $minor = $wpdb->get_var($wpdb->prepare( "SELECT balance_minor FROM {$account_table} WHERE user_id = %d AND asset = %s FOR UPDATE", $account['user_id'], $account['asset'] )); if ($minor === null) { throw new \RuntimeException('asset_account_lock_failed'); } $balances[$account['user_id']][$account['asset']] = (int) $minor; } if ($guard !== null) { $guard_result = $guard($transaction_id, $balances); if (is_wp_error($guard_result)) { if ($this->is_retryable_errno($this->last_db_errno())) { throw new \RuntimeException('ledger_guard_lock_failed'); } throw new LedgerAbort($guard_result); } if ($guard_result !== true) { throw new \RuntimeException('ledger_guard_failed'); } } $projected_balances = $balances; $prepared_entries = []; foreach ($command->entries as $index => $entry) { $user_id = (int) $entry['user_id']; $asset = (string) $entry['asset']; $delta = (int) $entry['delta_minor']; $before = (int) $projected_balances[$user_id][$asset]; if ($delta > 0 && $before > PHP_INT_MAX - $delta) { throw new LedgerAbort(new \WP_Error('aksrt_ledger_overflow', __('资产余额超出可用范围', 'aksrt'), ['status' => 400])); } $after = $before + $delta; if ($after < 0) { $code = $asset === AssetType::POINTS->value ? 'aksrt_points_insufficient' : 'aksrt_balance_insufficient'; throw new LedgerAbort(new \WP_Error($code, $asset === AssetType::POINTS->value ? __('积分不足', 'aksrt') : __('余额不足', 'aksrt'), [ 'status' => 402, 'balance_minor' => $before, 'required_minor' => abs($delta), ])); } $projected_balances[$user_id][$asset] = $after; $prepared_entries[] = [ 'transaction_id' => $transaction_id, 'line_no' => $index + 1, 'user_id' => $user_id, 'asset' => $asset, 'entry_type' => sanitize_key((string) ($entry['entry_type'] ?? 'adjustment')), 'delta_minor' => $delta, 'balance_after_minor' => $after, 'counterparty_user_id' => max(0, (int) ($entry['counterparty_user_id'] ?? 0)) ?: null, 'remark' => !empty($entry['remark']) ? substr(sanitize_text_field((string) $entry['remark']), 0, 255) : null, 'created_at' => $created_at, ]; } foreach ($account_keys as $account) { $user_id = (int) $account['user_id']; $asset = (string) $account['asset']; $updated = $wpdb->query($wpdb->prepare( "UPDATE {$account_table} SET balance_minor = %d, version = version + 1, updated_at = %s WHERE user_id = %d AND asset = %s", (int) $projected_balances[$user_id][$asset], $created_at, $user_id, $asset )); if ($updated !== 1) { throw new \RuntimeException('asset_account_update_failed'); } } $written_entries = []; foreach ($prepared_entries as $entry_row) { if ($wpdb->insert($entry_table, $entry_row, ['%d', '%d', '%d', '%s', '%s', '%d', '%d', '%d', '%s', '%s']) !== 1) { throw new \RuntimeException('asset_entry_insert_failed'); } $written_entries[] = array_merge(['id' => (int) $wpdb->insert_id], $entry_row); } $result = new LedgerResult($transaction_id, false, $written_entries, $projected_balances); if ($business_mutation !== null) { $mutation_result = $business_mutation($transaction_id, $result); if (is_wp_error($mutation_result)) { if ($this->is_retryable_errno($this->last_db_errno())) { throw new \RuntimeException('ledger_business_lock_failed'); } throw new LedgerAbort($mutation_result); } if ($mutation_result !== true) { throw new \RuntimeException('ledger_business_mutation_failed'); } } $flow_number = \AKSRT\Finance\FinancialFlowService::get_instance()->record_ledger( $command, $transaction_id, $written_entries, $created_at ); if (is_wp_error($flow_number)) { throw new LedgerAbort($flow_number); } $result = new LedgerResult($transaction_id, false, $written_entries, $projected_balances, $flow_number); if ($wpdb->query('COMMIT') === false) { throw new \RuntimeException('ledger_commit_failed'); } $after_commit_callbacks = $this->after_commit_callbacks; $this->after_commit_callbacks = []; $this->transaction_active = false; try { do_action('aksrt_asset_transaction_committed', $result, $command); } catch (\Throwable $event_error) { if (defined('WP_DEBUG') && WP_DEBUG) { error_log('[AKSRT Ledger Event] ' . $event_error->getMessage()); } } foreach ($after_commit_callbacks as $callback) { try { $callback(); } catch (\Throwable $event_error) { if (defined('WP_DEBUG') && WP_DEBUG) { error_log('[AKSRT Ledger After Commit] ' . $event_error->getMessage()); } } } return $result; } catch (LedgerAbort $e) { $wpdb->query('ROLLBACK'); $this->clean_rolled_back_user_caches($command); $this->transaction_active = false; $this->after_commit_callbacks = []; return $e->error; } catch (\Throwable $e) { $errno = $this->last_db_errno(); $wpdb->query('ROLLBACK'); $this->clean_rolled_back_user_caches($command); $this->transaction_active = false; $this->after_commit_callbacks = []; return $this->db_error($errno, $e->getMessage()); } } private function replay(string $operation_key, string $payload_hash) { global $wpdb; $transaction_table = $this->table('asset_transaction'); $row = $wpdb->get_row($wpdb->prepare( "SELECT id, payload_hash FROM {$transaction_table} WHERE operation_key = %s", $operation_key ), ARRAY_A); if (!is_array($row)) { return null; } if (!hash_equals((string) $row['payload_hash'], $payload_hash)) { return new \WP_Error('aksrt_ledger_idempotency_conflict', __('相同操作标识已用于其他账务请求', 'aksrt'), ['status' => 409]); } return $this->load_result((int) $row['id'], true); } private function load_result(int $transaction_id, bool $replayed): LedgerResult { global $wpdb; $entry_table = $this->table('asset_entry'); $entries = $wpdb->get_results($wpdb->prepare( "SELECT * FROM {$entry_table} WHERE transaction_id = %d ORDER BY line_no ASC", $transaction_id ), ARRAY_A); $balances = []; foreach ((array) $entries as &$entry) { foreach (['id', 'transaction_id', 'line_no', 'user_id', 'delta_minor', 'balance_after_minor', 'counterparty_user_id'] as $field) { $entry[$field] = (int) ($entry[$field] ?? 0); } $balances[$entry['user_id']][(string) $entry['asset']] = $entry['balance_after_minor']; } unset($entry); $flow_number = \AKSRT\Finance\FinancialFlowService::get_instance()->number_for_ledger_transaction($transaction_id); return new LedgerResult($transaction_id, $replayed, (array) $entries, $balances, $flow_number); } private function db_error(int $errno, string $detail = ''): \WP_Error { if ($this->is_retryable_errno($errno)) { return new \WP_Error('aksrt_ledger_retry', __('账务事务发生锁冲突', 'aksrt'), ['status' => 409]); } if (defined('WP_DEBUG') && WP_DEBUG && $detail !== '') { error_log('[AKSRT Ledger] ' . $detail); } return new \WP_Error('aksrt_ledger_db', __('账务数据库写入失败', 'aksrt'), ['status' => 500]); } private function is_retryable_errno(int $errno): bool { return in_array($errno, [1205, 1213], true); } private function clean_rolled_back_user_caches(LedgerCommand $command): void { if (!function_exists('clean_user_cache')) { return; } $user_ids = []; foreach ($command->entries as $entry) { $user_id = (int) ($entry['user_id'] ?? 0); if ($user_id > 0) { $user_ids[$user_id] = $user_id; } } foreach ($user_ids as $user_id) { clean_user_cache($user_id); } } private function last_db_errno(): int { global $wpdb; if (isset($wpdb->dbh) && $wpdb->dbh instanceof \mysqli) { return mysqli_errno($wpdb->dbh); } if (isset($wpdb->last_errno)) { return (int) $wpdb->last_errno; } if (preg_match('/\\b(1205|1213)\\b/', (string) ($wpdb->last_error ?? ''), $matches)) { return (int) $matches[1]; } return 0; } /** @param array $ref */ private function ref_type(array $ref): string { return sanitize_key((string) ($ref['type'] ?? '')); } /** @param array $ref */ private function ref_key(array $ref): string { foreach (['key', 'order', 'id', 'ref_id'] as $field) { if (isset($ref[$field]) && (string) $ref[$field] !== '') { return (string) $ref[$field]; } } return ''; } private function table(string $name): string { return \AKSRT\Core\DBMigrator::table($name); } private static function utc_now(): string { return (new \DateTimeImmutable('now', new \DateTimeZone('UTC')))->format('Y-m-d H:i:s.u'); } private static function client_ip(): ?string { $ip = function_exists('aksrt_get_client_ip') ? (string) aksrt_get_client_ip() : (string) ($_SERVER['REMOTE_ADDR'] ?? ''); $ip = substr(sanitize_text_field(wp_unslash($ip)), 0, 64); return $ip !== '' ? $ip : null; } private static function client_ua(): ?string { $ua = substr(sanitize_text_field(wp_unslash((string) ($_SERVER['HTTP_USER_AGENT'] ?? ''))), 0, 255); return $ua !== '' ? $ua : null; } }