im/app/model/BalanceLog.php

<?php
namespace app\model;

use Symfony\Component\Console\Input\Input;
use think\Model;
use think\facade\Db;

class BalanceLog extends Base
{

    // 表结构定义（使用时间戳）
    const TABLE_SCHEMA = [
        'id' => 'int(11) NOT NULL AUTO_INCREMENT',
        'user_id' => 'int(11) NOT NULL',
        'currency' => 'varchar(20) NOT NULL',
        'amount' => 'decimal(15,2) NOT NULL',
        'before' => 'decimal(15,2) NOT NULL',
        'after' => 'decimal(15,2) NOT NULL',
        'type' => 'varchar(50) NOT NULL',
        'created_at' => 'int(11) NOT NULL COMMENT \'UNIX timestamp\'', // 改为整型时间戳
        'memo' => 'varchar(255) DEFAULT NULL',
        'PRIMARY KEY (`id`)'
    ];
    function getCreatedAtAttr($v){
        return $v ? explode('.',$v)[0] : '';
    }
    protected function getOptions(): array{
        return array_merge(parent::getOptions(),[
            'connection'  =>  'mongodb',
            // 'append'    =>    [
            //     'from_user',
            //     'to_user'
            // ],
        ]);
    }
    public static function create(array|object $data, array $allowField = [], bool $replace = false):\think\model\contract\Modelable
    {
        $model = new static();
        if(isset($data['currency'])){
            if(in_array($data['currency'],Config('site.allow_balance_log'))){
                $data['status']=isset($data['status']) ? $data['status']:1;
                $data['user_id'] = intval($data['user_id']);
                $data['amount'] = floatval($data['amount']);
                $data['before'] = floatval($data['before']);
                $data['after'] = floatval($data['after']);
                $data['type'] =  $data['type'] instanceof \app\enum\BalanceType ? $data['type']->value : floatval($data['type']);
                $model->setSuffix('_'.strtolower($data['currency']))->allowField($allowField)
                    ->replace($replace)
                    ->save($data, true);
            }
        }
        return $model->fetchModel($model);
    }

    // 创建所有需要的表索引
    public static function createAllIndexes(): array
    {
        $results = [];
        $allow_balance_log = Config('site.allow_balance_log');
        foreach ($allow_balance_log as $currency) {
            $results[$currency] = self::createTableIndexes($currency);
        }
        return $results;
    }
    
    // 创建索引（适配时间戳查询）
    public static function createTableIndexes(string $currency): array
    {
        $table = self::getTableName($currency);
        $results = [];

        try {
            // 确保归档表存在
            if (!self::tableExists($table)) {
                self::createTableStructure($table);
            }
            // 主复合索引（使用时间戳）
            if (!self::indexExists($table, 'idx_user_currency_type_created')) {
                Db::execute("ALTER TABLE `{$table}` ADD INDEX `idx_user_currency_type_created` (`user_id`, `currency`, `type`, `created_at`)");
                $results[] = "Created idx_user_currency_type_created on {$table}";
            }

            // 时间索引（降序优化）
            if (!self::indexExists($table, 'idx_created_at')) {
                Db::execute("ALTER TABLE `{$table}` ADD INDEX `idx_created_at` (`created_at` DESC)");
                $results[] = "Created idx_created_at on {$table}";
            }

        } catch (\Throwable $e) {
            $results['error'] = "Error on {$table}: " . $e->getMessage();
        }

        return $results;
    }

    // 检查索引是否存在
    protected static function indexExists(string $table, string $indexName): bool
    {
        $indexes = Db::query("SHOW INDEX FROM `{$table}` WHERE Key_name = ?", [$indexName]);
        return !empty($indexes);
    }

    // 检查表是否存在
    protected static function tableExists(string $table): bool
    {
        try {
            Db::query("SELECT 1 FROM `{$table}` LIMIT 1");
            return true;
        } catch (\Throwable $e) {
            return false;
        }
    }

    // 数据归档方法（可在定时任务中调用）
    public static function archiveData(int $days = 3): array
    {
        $results = [];
        $allow_balance_log = Config('site.allow_balance_log');
        foreach ($allow_balance_log as $currency) {
            $results[$currency] = self::archiveCurrencyData($currency, $days);
        }
        return $results;
    }

    // 归档指定货币的数据
    protected static function archiveCurrencyData(string $currency, int $days): array
    {
        $table = self::getTableName($currency);
        $archiveTable = $table . '_archive';
        $cutoffTimestamp = time() - ($days * 86400); // 转为时间戳计算
        $result = [
            'table' => $table,
            'archived' => 0,
            'messages' => []
        ];

        try {
            // 确保归档表存在
            if (!self::tableExists($archiveTable)) {
                self::createTableStructure($archiveTable);
                $result['messages'][] = "Created archive table: {$archiveTable}";
            }

            // 分批归档数据
            $totalArchived = 0;
            Db::table($table)
                ->where('created_at', '<=', $cutoffTimestamp)
                ->chunk(1000, function($logs) use ($archiveTable, $table, &$totalArchived) {
                    Db::table($archiveTable)->insertAll($logs);
                    $count = count($logs);
                    Db::table($table)->whereIn('id', array_column($logs, 'id'))->delete();
                    $totalArchived += $count;
                });

            $result['archived'] = $totalArchived;
            $result['messages'][] = "Archived {$totalArchived} records from {$table}";

            // 优化表
            Db::execute("OPTIMIZE TABLE `{$table}`");
            $result['messages'][] = "Optimized table: {$table}";

        } catch (\Throwable $e) {
            $result['error'] = $e->getMessage();
        }

        return $result;
    }

    // 查询方法（示例）
    public static function queryLogs($userId, $currency, $type = null, $startTime = null, $endTime = null)
    {
        $model = new static;
        $query = $model->setSuffix('_'.strtolower($currency))->where('currency', $currency)
            ->where('user_id', intval($userId))
            ->order('created_at', 'desc');
            
        if ($type) {
            if($type == '99999'){
                $query->whereIn('type', [
                    \app\enum\BalanceType::OUTPUT_REWARD->value,
                    \app\enum\BalanceType::WITHDRAW_REWARD->value,
                    \app\enum\BalanceType::PRODUCT_INCOME->value,
                    \app\enum\BalanceType::AGENT_COMMISSION->value,
                    \app\enum\BalanceType::DIFFERENTIAL_COMMISSION->value
            ]);
            }else{
                $query->where('type', intval($type));
            }
        }
        
        if ($startTime) {
            // 支持传入时间戳或日期字符串
            //$startTimestamp = is_numeric($startTime) ? intval($startTime) : strtotime($startTime);
            $query->where('created_at', '>=', $startTime);
        }
        
        if ($endTime) {
            // 支持传入时间戳或日期字符串
            //$endTimestamp = is_numeric($endTime) ? intval($endTime) : strtotime($endTime);
            $query->where('created_at', '<=', $endTime);
        }
        $limit = 10;
        if(request()){
            $limit = input('limit',10);
        }
        return $query->paginate($limit);
    }
    
    // 创建表结构
    protected static function createTableStructure(string $table): bool
    {
        if (self::tableExists($table)) {
            return false;
        }

        $columns = [];
        foreach (self::TABLE_SCHEMA as $column => $definition) {
            if (strpos($definition, 'PRIMARY KEY') === false) {
                $columns[] = "`{$column}` {$definition}";
            }
        }

        $primaryKey = self::TABLE_SCHEMA['PRIMARY KEY'] ?? 'PRIMARY KEY (`id`)';

        $sql = "CREATE TABLE `{$table}` (" .
            implode(', ', $columns) . ", " .
            $primaryKey .
            ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci";

        Db::execute($sql);
        return true;
    }
}