bot-28/app/Services/CollectService.php

<?php


namespace App\Services;

use App\Services\BaseService;
use App\Models\Collect;
use App\Models\Recharge;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Log;
use App\Helpers\TronHelper;
use App\Services\WalletService;

/**
 * 归集记录
*/
class CollectService extends BaseService 
{
    public static string $MODEL = Collect::class;
    public static $THRESHOLD = 10; // 最小归集金额（USDT）

    public static function getRequiredTrxBalance(): float
    {
        $value = (float) config('app.collect_required_trx_balance', 15);
        return $value > 0 ? $value : 15.0;
    }
    /**
     * @description: 模型
     * @return {string}
     */    
    public static function model() :string
    {
        return Collect::class;
    }

    /**
     * @description: 枚举
     * @return {*}
     */    
    public static function enum() :string
    {
        return '';
    }

    /**
     * @description: 获取查询条件
     * @param {array} $search 查询内容
     * @return {array}
     */    
    public static function getWhere(array $search = []) :array
    {
        $where = [];
        if(isset($search['coin']) && !empty($search['coin'])){
            $where[] = ['coin', '=', $search['coin']];
        }
        if(isset($search['net']) && !empty($search['net'])){
            $where[] = ['net', '=', $search['net']];
        }
        if(isset($search['to_address']) && !empty($search['to_address'])){
            $where[] = ['to_address', '=', $search['to_address']];
        }
        if(isset($search['from_address']) && !empty($search['from_address'])){
            $where[] = ['from_address', '=', $search['from_address']];
        }
        if(isset($search['id']) && !empty($search['id'])){
            $where[] = ['id', '=', $search['id']];
        }

        if(isset($search['txid']) && !empty($search['txid'])){
            $where[] = ['txid', '=', $search['txid']];
        }
        if(isset($search['status']) && $search['status'] != ''){
            $where[] = ['status', '=', $search['status']];
        }
        if (isset($search['amount']) && is_numeric($search['amount'])) {
            $where[] = ['amount', '>=', $search['amount']];
        }
        return $where;
    }

     /**
     * @description: 查询单条数据
     * @param array $search
     * @return \App\Models\Coin|null
     */
    public static function findOne(array $search): ?Collect
    {
        return self::model()::where(self::getWhere($search))->first();
    }

    /**
     * @description: 查询所有数据
     * @param array $search
     * @return \Illuminate\Database\Eloquent\Collection
     */
    public static function findAll(array $search = [])
    {
        return self::model()::where(self::getWhere($search))->get();
    }

    /**
     * @description: 分页查询
     * @param array $search
     * @return \Illuminate\Contracts\Pagination\LengthAwarePaginator
     */
    public static function paginate(array $search = [])
    {
        $limit = isset($search['limit'])?$search['limit']:15;
        $paginator = self::model()::where(self::getWhere($search))->paginate($limit);
        return ['total' => $paginator->total(), 'data' => $paginator->items()];
    }
    

    /**
     * @description: 生成归集记录
     * @param {*} $address
     * @param {*} $coin
     * @param {*} $net
     * @return {*}
     */    
    public static function createCollect($address ,$coin ,$net)
    {
        $amount = TronHelper::getTrc20Balance($address);    // 获取地址的余额

        $info = self::findOne(['from_address' => $address ,'status' => self::model()::STATUS_STAY]);
        
        if($amount >= 0 ){
            if(empty($info)){
                $data = [];
                $data['from_address'] = $address;
                $data['amount'] = $amount;
                $data['coin'] = $coin;
                $data['net'] = $net;
                self::model()::create($data);

            }else{
                $info->amount = $amount;
                $info->save();
            }
        }

    }

    /**
     * @description: 按会员补建或刷新待归集记录，不执行链上归集
     * @param {*} $memberId
     * @return array
     */
    public static function syncMemberCollectRecords($memberId)
    {
        $recharges = Recharge::where('member_id', $memberId)
            ->where('status', Recharge::STATUS_SUCCESS)
            ->where('type', Recharge::TYPE_AUTO)
            ->orderByDesc('id')
            ->get();

        $result = [
            'success' => true,
            'member_id' => $memberId,
            'recharge_count' => $recharges->count(),
            'processed_count' => 0,
            'created_count' => 0,
            'updated_count' => 0,
            'skipped_count' => 0,
            'failed_count' => 0,
            'items' => [],
        ];

        if ($recharges->isEmpty()) {
            $result['success'] = false;
            $result['message'] = '未找到已确认的自动充值记录';
            return $result;
        }

        $seen = [];
        foreach ($recharges as $recharge) {
            $address = $recharge->to_address;
            if (empty($address) || isset($seen[$address])) {
                continue;
            }

            $seen[$address] = true;
            self::resetInvalidStartedCollectsByAddress($address);
            $item = [
                'from_address' => $address,
                'recharge_txid' => $recharge->txid,
                'action' => 'pending',
                'collect_id' => null,
                'amount' => null,
                'error' => null,
            ];

            $openCollect = self::model()::where('from_address', $address)
                ->whereIn('status', [self::model()::STATUS_STAY, self::model()::STATUS_START])
                ->orderByDesc('id')
                ->first();

            if ($openCollect && intval($openCollect->status) === self::model()::STATUS_START) {
                $item['action'] = 'skipped_started';
                $item['collect_id'] = $openCollect->id;
                $item['amount'] = $openCollect->amount;
                $result['skipped_count']++;
                $result['processed_count']++;
                $result['items'][] = $item;
                continue;
            }

            $hadPending = $openCollect && intval($openCollect->status) === self::model()::STATUS_STAY;

            try {
                self::createCollect($address, $recharge->coin, $recharge->net);
                $collect = self::model()::where('from_address', $address)
                    ->where('status', self::model()::STATUS_STAY)
                    ->orderByDesc('id')
                    ->first();

                $item['action'] = $hadPending ? 'updated' : 'created';
                $item['collect_id'] = $collect->id ?? null;
                $item['amount'] = $collect->amount ?? null;

                if ($hadPending) {
                    $result['updated_count']++;
                } else {
                    $result['created_count']++;
                }
            } catch (\Throwable $e) {
                $item['action'] = 'failed';
                $item['error'] = $e->getMessage();
                $result['failed_count']++;
            }

            $result['processed_count']++;
            $result['items'][] = $item;
        }

        if ($result['processed_count'] === 0) {
            $result['success'] = false;
            $result['message'] = '未找到可处理的充值地址';
        }

        return $result;
    }

    private static function resetInvalidStartedCollectsByAddress($address)
    {
        $updated = self::model()::where('from_address', $address)
            ->where('status', self::model()::STATUS_START)
            ->where(function ($query) {
                $query->whereNull('txid')
                    ->orWhere('txid', '');
            })
            ->update([
                'status' => self::model()::STATUS_STAY,
                'to_address' => null,
                'remark' => 'reset invalid started collect',
                'updated_at' => now(),
            ]);

        if ($updated > 0) {
            Log::warning('reset invalid started collects', [
                'from_address' => $address,
                'count' => $updated,
            ]);
        }

        return $updated;
    }

    /**
     * @description: 处理指定会员待归集记录，会执行链上归集
     * @param {*} $memberId
     * @return array
     */
    public static function syncCollectStayByMember($memberId)
    {
        $result = [
            'member_id' => $memberId,
            'threshold' => self::$THRESHOLD,
            'from_address' => null,
            'to_address' => null,
            'pending_count' => 0,
            'handled_count' => 0,
            'success_count' => 0,
            'fail_count' => 0,
            'items' => [],
        ];

        $walletInfo = WalletService::findOne(['member_id' => $memberId, 'coin' => 'USDT']);
        if (empty($walletInfo) || empty($walletInfo->address)) {
            $result['message'] = '未找到该用户的USDT钱包地址';
            Log::warning('syncCollectStayByMember skipped: wallet missing', [
                'member_id' => $memberId,
            ]);
            return $result;
        }

        $result['from_address'] = $walletInfo->address;
        self::resetInvalidStartedCollectsByAddress($walletInfo->address);

        $to_address = self::getUsdtAddress();
        $trx_private_key = self::getTrxPrivateKey();
        $result['to_address'] = $to_address;

        Log::info('syncCollectStayByMember start', [
            'member_id' => $memberId,
            'from_address' => $walletInfo->address,
            'threshold' => self::$THRESHOLD,
            'required_trx_balance' => self::getRequiredTrxBalance(),
            'to_address' => $to_address,
            'has_trx_private_key' => !empty($trx_private_key),
        ]);

        if (!$to_address || !$trx_private_key) {
            $result['message'] = '归集配置不完整';
            Log::warning('syncCollectStayByMember skipped: missing config', [
                'member_id' => $memberId,
                'to_address' => $to_address,
                'has_trx_private_key' => !empty($trx_private_key),
            ]);
            return $result;
        }

        $list = self::findAll([
            'status' => self::model()::STATUS_STAY,
            'amount' => self::$THRESHOLD,
            'from_address' => $walletInfo->address,
        ]);
        $result['pending_count'] = $list->count();

        if ($list->isEmpty()) {
            $result['message'] = '该用户没有待归集记录';
            Log::info('syncCollectStayByMember finished: no pending collects', $result);
            return $result;
        }

        foreach ($list as $v) {
            $item = [
                'id' => $v['id'],
                'from_address' => $v['from_address'],
                'amount' => $v['amount'],
                'status' => 'pending',
            ];
            $data = [];
            $wallets = WalletService::findOne(['address' => $v['from_address']]);

            if (empty($wallets) || empty($wallets['private_key'])) {
                $item['status'] = 'wallet_not_found';
                $item['error'] = '未找到归集钱包私钥';
                $data['remark'] = $item['error'];
                $data['updated_at'] = now();
                self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
                $result['fail_count']++;
                $result['handled_count']++;
                $result['items'][] = $item;
                Log::warning('syncCollectStayByMember wallet missing', $item + ['member_id' => $memberId]);
                continue;
            }

            $privateKey = $wallets['private_key'];
            $trxBalance = TronHelper::getTrxBalance($v['from_address']);
            $item['trx_balance'] = $trxBalance;
            $requiredTrxBalance = self::getRequiredTrxBalance();
            $item['required_trx_balance'] = $requiredTrxBalance;

            if ($trxBalance < $requiredTrxBalance) {
                $topupAmount = round($requiredTrxBalance - $trxBalance, 6);
                $item['trx_topup_amount'] = $topupAmount;
                $trxResult = TronHelper::sendTrx($trx_private_key, $v['from_address'], $topupAmount);
                $item['trx_topup_result'] = $trxResult;
                Log::info('syncCollectStayByMember topup trx', [
                    'member_id' => $memberId,
                    'from_address' => $v['from_address'],
                    'trx_balance' => $trxBalance,
                    'required_trx_balance' => $requiredTrxBalance,
                    'topup_amount' => $topupAmount,
                    'result' => $trxResult,
                ]);

                if ($trxResult === false || is_string($trxResult)) {
                    $error = is_string($trxResult) ? $trxResult : 'TRX能量补充失败';
                    $data['status'] = self::model()::STATUS_STAY;
                    $data['to_address'] = null;
                    $data['txid'] = null;
                    $data['remark'] = $error;
                    $data['updated_at'] = now();
                    $item['status'] = 'trx_topup_failed';
                    $item['error'] = $error;
                    self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
                    $result['fail_count']++;
                    $result['handled_count']++;
                    $result['items'][] = $item;
                    Log::warning('syncCollectStayByMember topup trx failed', $item + ['member_id' => $memberId]);
                    continue;
                }

                $trxBalance = TronHelper::getTrxBalance($v['from_address']);
                $item['trx_balance_after_topup'] = $trxBalance;
                if ($trxBalance < $requiredTrxBalance) {
                    $error = 'TRX余额仍不足，停止归集';
                    $data['status'] = self::model()::STATUS_STAY;
                    $data['to_address'] = null;
                    $data['txid'] = null;
                    $data['remark'] = $error;
                    $data['updated_at'] = now();
                    $item['status'] = 'trx_balance_insufficient';
                    $item['error'] = $error;
                    self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
                    $result['fail_count']++;
                    $result['handled_count']++;
                    $result['items'][] = $item;
                    Log::warning('syncCollectStayByMember trx balance still insufficient', $item + ['member_id' => $memberId]);
                    continue;
                }
            }

            $transferResult = TronHelper::transferUSDT($privateKey, $to_address, $v['amount']);
            $item['transfer_result'] = $transferResult;

            if (is_array($transferResult) && !empty($transferResult['success'])) {
                $data['to_address'] = $to_address;
                $data['txid'] = $transferResult['txid'] ?? '';
                $data['remark'] = 'success';
                $data['status'] = self::model()::STATUS_START;
                $item['status'] = 'success';
                $item['txid'] = $data['txid'];
                $result['success_count']++;
                Log::info('syncCollectStayByMember transfer success', $item + ['member_id' => $memberId]);
            } else {
                $error = is_array($transferResult)
                    ? ($transferResult['error'] ?? 'USDT归集失败')
                    : (is_string($transferResult) ? $transferResult : 'USDT归集失败');
                $data['status'] = self::model()::STATUS_STAY;
                $data['to_address'] = null;
                $data['txid'] = null;
                $data['remark'] = $error;
                $item['status'] = 'failed';
                $item['error'] = $error;
                $result['fail_count']++;
                Log::warning('syncCollectStayByMember transfer failed', $item + ['member_id' => $memberId]);
            }

            $data['updated_at'] = now();
            self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
            $result['handled_count']++;
            $result['items'][] = $item;
        }

        Log::info('syncCollectStayByMember finished', $result);
        return $result;
    }

    /**
     * @description: 处理待归集的
     * @return {*}
     */    
    public static function syncCollectStay()
    {
        $result = [
            'threshold' => self::$THRESHOLD,
            'to_address' => null,
            'pending_count' => 0,
            'handled_count' => 0,
            'success_count' => 0,
            'fail_count' => 0,
            'items' => [],
        ];

        $to_address = self::getUsdtAddress();   // 转账的接收地址
        $trx_private_key = self::getTrxPrivateKey(); // 获取TRX能量的秘钥
        $result['to_address'] = $to_address;

        Log::info('syncCollectStay start', [
            'threshold' => self::$THRESHOLD,
            'required_trx_balance' => self::getRequiredTrxBalance(),
            'to_address' => $to_address,
            'has_trx_private_key' => !empty($trx_private_key),
        ]);

        if (!$to_address || !$trx_private_key) {
            $result['message'] = '归集配置不完整';
            Log::warning('syncCollectStay skipped: missing config', [
                'to_address' => $to_address,
                'has_trx_private_key' => !empty($trx_private_key),
            ]);
            return $result;
        }

        $list = self::findAll(['status' => self::model()::STATUS_STAY ,'amount' => self::$THRESHOLD]);
        $result['pending_count'] = $list->count();

        if ($list->isEmpty()) {
            $result['message'] = '没有待归集记录';
            Log::info('syncCollectStay finished: no pending collects', $result);
            return $result;
        }

        foreach($list as $k => $v){
            $item = [
                'id' => $v['id'],
                'from_address' => $v['from_address'],
                'amount' => $v['amount'],
                'status' => 'pending',
            ];
            $data = [];
            $wallets = WalletService::findOne(['address' => $v['from_address']]);

            if (empty($wallets) || empty($wallets['private_key'])) {
                $item['status'] = 'wallet_not_found';
                $item['error'] = '未找到归集钱包私钥';
                $data['remark'] = $item['error'];
                $data['updated_at'] = now();
                self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
                $result['fail_count']++;
                $result['handled_count']++;
                $result['items'][] = $item;
                Log::warning('syncCollectStay wallet missing', $item);
                continue;
            }

            $privateKey = $wallets['private_key'];
            $trxBalance = TronHelper::getTrxBalance($v['from_address']);
            $item['trx_balance'] = $trxBalance;
            $requiredTrxBalance = self::getRequiredTrxBalance();
            $item['required_trx_balance'] = $requiredTrxBalance;

            if($trxBalance < $requiredTrxBalance){
                $topupAmount = round($requiredTrxBalance - $trxBalance, 6);
                $item['trx_topup_amount'] = $topupAmount;
                $trxResult = TronHelper::sendTrx($trx_private_key,$v['from_address'],$topupAmount);
                $item['trx_topup_result'] = $trxResult;
                Log::info('syncCollectStay topup trx', [
                    'from_address' => $v['from_address'],
                    'trx_balance' => $trxBalance,
                    'required_trx_balance' => $requiredTrxBalance,
                    'topup_amount' => $topupAmount,
                    'result' => $trxResult,
                ]);

                if ($trxResult === false || is_string($trxResult)) {
                    $error = is_string($trxResult) ? $trxResult : 'TRX能量补充失败';
                    $data['status'] = self::model()::STATUS_STAY;
                    $data['to_address'] = null;
                    $data['txid'] = null;
                    $data['remark'] = $error;
                    $data['updated_at'] = now();
                    $item['status'] = 'trx_topup_failed';
                    $item['error'] = $error;
                    self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
                    $result['fail_count']++;
                    $result['handled_count']++;
                    $result['items'][] = $item;
                    Log::warning('syncCollectStay topup trx failed', $item);
                    continue;
                }

                $trxBalance = TronHelper::getTrxBalance($v['from_address']);
                $item['trx_balance_after_topup'] = $trxBalance;
                if ($trxBalance < $requiredTrxBalance) {
                    $error = 'TRX余额仍不足，停止归集';
                    $data['status'] = self::model()::STATUS_STAY;
                    $data['to_address'] = null;
                    $data['txid'] = null;
                    $data['remark'] = $error;
                    $data['updated_at'] = now();
                    $item['status'] = 'trx_balance_insufficient';
                    $item['error'] = $error;
                    self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
                    $result['fail_count']++;
                    $result['handled_count']++;
                    $result['items'][] = $item;
                    Log::warning('syncCollectStay trx balance still insufficient', $item);
                    continue;
                }
            }

            $transferResult = TronHelper::transferUSDT($privateKey,$to_address,$v['amount']);
            $item['transfer_result'] = $transferResult;

            if(is_array($transferResult) && !empty($transferResult['success'])){
                $data['to_address'] = $to_address;
                $data['txid'] = $transferResult['txid'] ?? '';
                $data['remark'] = 'success';
                $data['status'] = self::model()::STATUS_START;
                $item['status'] = 'success';
                $item['txid'] = $data['txid'];
                $result['success_count']++;
                Log::info('syncCollectStay transfer success', $item);
            }else{
                $error = is_array($transferResult)
                    ? ($transferResult['error'] ?? 'USDT归集失败')
                    : (is_string($transferResult) ? $transferResult : 'USDT归集失败');
                $data['status'] = self::model()::STATUS_STAY;
                $data['to_address'] = null;
                $data['txid'] = null;
                $data['remark'] = $error;
                $item['status'] = 'failed';
                $item['error'] = $error;
                $result['fail_count']++;
                Log::warning('syncCollectStay transfer failed', $item);
            }
            $data['updated_at'] = now();
            self::model()::where(self::getWhere(['id' => $v['id']]))->update($data);
            $result['handled_count']++;
            $result['items'][] = $item;
        }

        Log::info('syncCollectStay finished', $result);
        return $result;

    }

    /**
     * @description: 获取归集平台的接收地址
     * @return {*}
     */    
    public static function getUsdtAddress()
    {
        $usdt_address = config('app.usdt_address');
        return $usdt_address;
    }

    /**
     * @description: 获取TRX能量账号秘钥
     * @return {*}
     */    
    public static function getTrxPrivateKey()
    {
        $str = config('app.trx_private_key');
        return $str;
    }
   
}