优化推送订单功能

.env-example

@@ -35,3 +35,16 @@ PLAYX_SESSION_EXPIRE_SECONDS=3600
 # PlayX API（商城调用 PlayX 时使用）
 PLAYX_API_BASE_URL=
 PLAYX_API_SECRET_KEY=
+
+# 推送订单url
+PLAYX_ANGPOW_IMPORT_BASE_URL=https://ss2-staging2.ttwd8.com
+# 推送订单接口
+PLAYX_ANGPOW_IMPORT_PATH=/cashmarket/v3/merchant-api/angpow-imports
+# 商户编码merchant_code
+PLAYX_ANGPOW_MERCHANT_CODE=plx
+# HMAC 密钥（与对端一致）
+PLAYX_ANGPOW_IMPORT_AUTH_KEY=
+# HTTPS：CA 证书包绝对路径（推荐下载 https://curl.se/ca/cacert.pem 后填写，解决 cURL error 60）
+PLAYX_ANGPOW_IMPORT_CACERT=
+# 是否校验 SSL（1=校验；0=不校验，仅本地调试，生产勿用）
+PLAYX_ANGPOW_IMPORT_VERIFY_SSL=1

app/process/AngpowImportJobs.php

@@ -0,0 +1,328 @@
+<?php
+
+namespace app\process;
+
+use app\common\model\MallItem;
+use app\common\model\MallOrder;
+use app\common\model\MallUserAsset;
+use GuzzleHttp\Client;
+use Workerman\Timer;
+use Workerman\Worker;
+
+/**
+ * Angpow 导入推送任务
+ * - 数据源：mall_order（type=BONUS）
+ * - 推送频率：每 30 秒
+ * - 批量：每次最多 100 条（对方文档限制）
+ * - 幂等：merchant_code + report_date 级别签名；每条订单通过 external_transaction_id 在本地控制只推送一次
+ */
+class AngpowImportJobs
+{
+    private const TIMER_SECONDS = 30;
+    private const BATCH_LIMIT = 100;
+    private const MAX_RETRY = 3;
+
+    protected Client $http;
+
+    public function __construct()
+    {
+        // 确保定时任务只在一个 worker 上运行
+        if (!Worker::getAllWorkers()) {
+            return;
+        }
+
+        $this->http = new Client($this->buildGuzzleOptions());
+
+        Timer::add(self::TIMER_SECONDS, [$this, 'pushPendingOrders']);
+    }
+
+    /**
+     * Guzzle 默认校验 HTTPS；Windows 未配置 CA 时会出现 cURL error 60。
+     * 优先使用 PLAYX_ANGPOW_IMPORT_CACERT 指向 cacert.pem；否则可按环境关闭校验（仅开发）。
+     */
+    private function buildGuzzleOptions(): array
+    {
+        $options = [
+            'timeout' => 20,
+            'http_errors' => false,
+        ];
+
+        $conf = config('playx.angpow_import');
+        if (!is_array($conf)) {
+            return $options;
+        }
+
+        $caFile = $conf['ca_file'] ?? '';
+        if (is_string($caFile) && $caFile !== '' && is_file($caFile)) {
+            $options['verify'] = $caFile;
+
+            return $options;
+        }
+
+        $verifySsl = $conf['verify_ssl'] ?? true;
+        if ($verifySsl === false) {
+            $options['verify'] = false;
+        }
+
+        return $options;
+    }
+
+    public function pushPendingOrders(): void
+    {
+        $conf = config('playx.angpow_import');
+        if (!is_array($conf)) {
+            return;
+        }
+
+        $baseUrl = $conf['base_url'] ?? '';
+        $path = $conf['path'] ?? '';
+        $merchantCode = $conf['merchant_code'] ?? '';
+        $authKey = $conf['auth_key'] ?? '';
+
+        if (!is_string($baseUrl) || $baseUrl === '' || !is_string($path) || $path === '' || !is_string($merchantCode) || $merchantCode === '') {
+            return;
+        }
+        if (!is_string($authKey) || $authKey === '') {
+            return;
+        }
+
+        $url = rtrim($baseUrl, '/') . $path;
+
+        $orders = MallOrder::where('type', MallOrder::TYPE_BONUS)
+            ->whereIn('grant_status', [MallOrder::GRANT_NOT_SENT, MallOrder::GRANT_FAILED_RETRYABLE])
+            ->where('status', MallOrder::STATUS_PENDING)
+            ->where('retry_count', '<', self::MAX_RETRY)
+            ->order('id', 'asc')
+            ->limit(self::BATCH_LIMIT)
+            ->select();
+
+        if ($orders->isEmpty()) {
+            return;
+        }
+
+        $reportDate = strval(time());
+        $signatureInput = 'merchant_code=' . $merchantCode . '&report_date=' . $reportDate;
+        $signature = $this->buildSignature($signatureInput, $authKey);
+        if ($signature === null) {
+            return;
+        }
+
+        $payload = [
+            'merchant_code' => $merchantCode,
+            'report_date' => $reportDate,
+            'angpow' => [],
+            'currency_visual' => [
+                [
+                    'currency' => strval($conf['currency'] ?? 'MYR'),
+                    'visual_name' => strval($conf['visual_name'] ?? 'Angpow'),
+                ],
+            ],
+        ];
+
+        $orderIds = [];
+        foreach ($orders as $order) {
+            if (!$order instanceof MallOrder) {
+                continue;
+            }
+            $row = $this->buildAngpowRow($order);
+            if ($row === null) {
+                // 构造失败：直接标为可重试失败
+                $this->markFailedAttempt($order, 'Build payload failed');
+                continue;
+            }
+            $payload['angpow'][] = $row;
+            $orderIds[] = $order->id;
+        }
+
+        if (empty($payload['angpow'])) {
+            return;
+        }
+
+        // 先标记“已尝试发送”，避免并发重复推送；同时只在这里累加 retry_count（一次推送=一次尝试）
+        $now = time();
+        foreach ($orders as $order) {
+            if (!$order instanceof MallOrder) {
+                continue;
+            }
+            if (!in_array($order->id, $orderIds, true)) {
+                continue;
+            }
+            $retry = $order->retry_count ?? 0;
+            if (!is_int($retry)) {
+                $retry = is_numeric($retry) ? intval($retry) : 0;
+            }
+            $order->retry_count = $retry + 1;
+            $order->grant_status = MallOrder::GRANT_SENT_PENDING;
+            $order->update_time = $now;
+            $order->save();
+        }
+
+        $res = null;
+        $body = '';
+        try {
+            $res = $this->http->post($url, [
+                'headers' => [
+                    'Content-Type' => 'application/json',
+                    'X-Request-Signature' => $signature,
+                ],
+                'json' => $payload,
+            ]);
+            $body = strval($res->getBody());
+        } catch (\Throwable $e) {
+            // 网络/异常：对这一批订单记一次失败尝试
+            foreach ($orders as $order) {
+                if (!$order instanceof MallOrder) {
+                    continue;
+                }
+                $this->markFailedAttempt($order, $e->getMessage());
+            }
+            return;
+        }
+
+        $data = json_decode($body, true);
+        if (!is_array($data)) {
+            foreach ($orders as $order) {
+                if (!$order instanceof MallOrder) {
+                    continue;
+                }
+                $this->markFailedAttempt($order, 'Invalid response');
+            }
+            return;
+        }
+
+        $code = $data['code'] ?? null;
+        $message = $data['message'] ?? '';
+        $msg = is_string($message) ? $message : 'Request failed';
+
+        // 成功：code=0
+        if ($code === '0' || $code === 0) {
+            MallOrder::whereIn('id', $orderIds)->update([
+                'grant_status' => MallOrder::GRANT_ACCEPTED,
+                'fail_reason' => null,
+                'update_time' => time(),
+            ]);
+            return;
+        }
+
+        // 失败：整批视为失败（对方未提供逐条返回）
+        foreach ($orders as $order) {
+            if (!$order instanceof MallOrder) {
+                continue;
+            }
+            $this->markFailedAttempt($order, $msg);
+        }
+    }
+
+    private function buildAngpowRow(MallOrder $order): ?array
+    {
+        $asset = MallUserAsset::where('playx_user_id', $order->user_id)->find();
+        if (!$asset) {
+            if (is_string($order->user_id) && ctype_digit($order->user_id)) {
+                $byId = MallUserAsset::where('id', $order->user_id)->find();
+                if ($byId) {
+                    $asset = $byId;
+                }
+            }
+        }
+        if (!$asset || !is_string($asset->playx_user_id ?? null) || strval($asset->playx_user_id) === '') {
+            return null;
+        }
+
+        $item = null;
+        if ($order->mallItem) {
+            $item = $order->mallItem;
+        } else {
+            $item = MallItem::where('id', $order->mall_item_id)->find();
+        }
+        if (!$item) {
+            return null;
+        }
+
+        $createTime = $order->create_time ?? null;
+        if (!is_int($createTime)) {
+            if (is_numeric($createTime)) {
+                $createTime = intval($createTime);
+            } else {
+                $createTime = time();
+            }
+        }
+
+        $start = gmdate('Y-m-d\TH:i:s\Z', $createTime);
+        $end = gmdate('Y-m-d\TH:i:s\Z', $createTime + 86400);
+
+        return [
+            'member_login' => strval($asset->playx_user_id),
+            'start_time' => $start,
+            'end_time' => $end,
+            'amount' => $order->amount,
+            'reward_name' => strval($item->title ?? ''),
+            'description' => strval($item->description ?? ''),
+            'member_inbox_message' => 'Congratulations! You received an angpow.',
+            'category' => strval($item->category ?? ''),
+            'category_title' => strval($item->category_title ?? ''),
+            'one_time_turnover' => 'yes',
+            'multiplier' => $order->multiplier,
+        ];
+    }
+
+    private function markFailedAttempt(MallOrder $order, string $reason): void
+    {
+        // retry_count 在“准备发送”阶段已 +1；此处用当前 retry_count 作为 attempt 编号
+        $retryCount = $order->retry_count ?? 0;
+        $attempt = is_int($retryCount) ? $retryCount : (is_numeric($retryCount) ? intval($retryCount) : 0);
+        if ($attempt <= 0) {
+            $attempt = 1;
+            $order->retry_count = 1;
+        }
+
+        $prev = $order->fail_reason;
+        $prefix = 'attempt ' . $attempt . ': ';
+        $line = $prefix . $reason;
+        $newReason = $line;
+        if (is_string($prev) && $prev !== '') {
+            $newReason = $prev . "\n" . $line;
+        }
+
+        $final = $attempt >= self::MAX_RETRY;
+        $order->grant_status = $final ? MallOrder::GRANT_FAILED_FINAL : MallOrder::GRANT_FAILED_RETRYABLE;
+        $order->fail_reason = $newReason;
+        $order->save();
+    }
+
+    /**
+     * 生成对方要求的 Base64(HMAC-SHA1)
+     * - 文档中示例 python 会 base64_decode(key) 后参与 hmac
+     * - 生产 key 由 BA 提供，可能是 base64 或 hex；这里做兼容处理
+     */
+    private function buildSignature(string $input, string $authKey): ?string
+    {
+        $keyBytes = null;
+
+        $maybeBase64 = base64_decode($authKey, true);
+        if ($maybeBase64 !== false && $maybeBase64 !== '') {
+            $keyBytes = $maybeBase64;
+        }
+
+        if ($keyBytes === null) {
+            $isHex = ctype_xdigit($authKey) && (strlen($authKey) % 2 === 0);
+            if ($isHex) {
+                $hex = hex2bin($authKey);
+                if ($hex !== false && $hex !== '') {
+                    $keyBytes = $hex;
+                }
+            }
+        }
+
+        if ($keyBytes === null) {
+            $keyBytes = $authKey;
+        }
+
+        $raw = hash_hmac('sha1', $input, $keyBytes, true);
+        if (!is_string($raw) || $raw === '') {
+            return null;
+        }
+
+        return base64_encode($raw);
+    }
+}
+

composer.json

@@ -40,7 +40,8 @@
     "nelexa/zip": "^4.0.0",
     "voku/anti-xss": "^4.1",
     "topthink/think-validate": "^3.0",
-    "firebase/php-jwt": "^7.0"
+    "firebase/php-jwt": "^7.0",
+    "guzzlehttp/guzzle": "^7.10"
   },
   "suggest": {
     "ext-event": "For better performance. "

config/playx.php

@@ -33,4 +33,22 @@ return [
         'balance_credit_url' => '/api/v1/balance/credit',
         'transaction_status_url' => '/api/v1/transaction/status',
     ],
+    // Angpow Import API（商城调用对方 cashmarket/merchant-api 时使用）
+    'angpow_import' => [
+        // 对方 base_url，例如 https://ss2-staging2.ttwd8.com
+        'base_url' => strval(env('PLAYX_ANGPOW_IMPORT_BASE_URL', '')),
+        // 路径：文档示例为 /api/v3/merchant/angpow-imports；对方 curl 示例为 /cashmarket/v3/merchant-api/angpow-imports
+        'path' => strval(env('PLAYX_ANGPOW_IMPORT_PATH', '/api/v3/merchant/angpow-imports')),
+        // merchant_code（固定 plx 或按环境配置）
+        'merchant_code' => strval(env('PLAYX_ANGPOW_MERCHANT_CODE', 'plx')),
+        // HMAC-SHA1 的 auth key（生产环境由 BA 提供）
+        'auth_key' => strval(env('PLAYX_ANGPOW_IMPORT_AUTH_KEY', '')),
+        // HTTPS：指定 CA 证书包路径（推荐，下载 https://curl.se/ca/cacert.pem 后填绝对路径，可避免 cURL error 60）
+        'ca_file' => strval(env('PLAYX_ANGPOW_IMPORT_CACERT', '')),
+        // 是否校验 SSL（生产必须为 true；本地无 CA 时可临时 false，勿用于生产）
+        'verify_ssl' => filter_var(env('PLAYX_ANGPOW_IMPORT_VERIFY_SSL', '1'), FILTER_VALIDATE_BOOLEAN),
+        // 固定货币展示映射
+        'currency' => 'MYR',
+        'visual_name' => 'Angpow',
+    ],
 ];

config/process.php

@@ -15,6 +15,7 @@
 use support\Log;
 use support\Request;
 use app\process\Http;
+use app\process\AngpowImportJobs;
 
 global $argv;
 
@@ -65,4 +66,9 @@ return [
         'handler' => app\process\PlayxJobs::class,
         'reloadable' => false,
     ],
+    // Angpow 导入推送任务：订单兑换后推送到对方平台
+    'angpow_import_jobs' => [
+        'handler' => AngpowImportJobs::class,
+        'reloadable' => false,
+    ],
 ];