cunkebao_v3/Moncter/app/service/DataSource/Strategy/MongoDBConsumptionStrategy.php

<?php

namespace app\service\DataSource\Strategy;

use app\service\DataSource\DataSourceAdapterInterface;
use app\service\DataSource\PollingStrategyInterface;
use app\utils\LoggerHelper;

/**
 * MongoDB 消费记录轮询策略
 * 
 * 职责：
 * - 提供 MongoDB 专用的轮询策略实现
 * - 展示如何实现自定义业务逻辑
 */
class MongoDBConsumptionStrategy implements PollingStrategyInterface
{
    /**
     * 执行轮询查询
     * 
     * @param DataSourceAdapterInterface $adapter 数据源适配器
     * @param array<string, mixed> $config 数据源配置
     * @param array<string, mixed> $lastSyncInfo 上次同步信息
     * @return array<array<string, mixed>> 查询结果数组
     */
    public function poll(
        DataSourceAdapterInterface $adapter,
        array $config,
        array $lastSyncInfo = []
    ): array {
        // 从配置中获取集合名和查询条件
        $collectionName = $config['collection'] ?? 'consumption_records';
        $lastSyncTime = $lastSyncInfo['last_sync_time'] ?? null;
        $lastSyncId = $lastSyncInfo['last_sync_id'] ?? null;

        // 构建 MongoDB 查询过滤器
        $filter = [];

        // 如果有上次同步时间，只查询新增或更新的记录
        if ($lastSyncTime !== null) {
            $lastSyncTimestamp = is_numeric($lastSyncTime) ? (int)$lastSyncTime : strtotime($lastSyncTime);
            $lastSyncDate = new \MongoDB\BSON\UTCDateTime($lastSyncTimestamp * 1000);
            
            $filter['$or'] = [
                ['created_at' => ['$gt' => $lastSyncDate]],
                ['updated_at' => ['$gt' => $lastSyncDate]],
            ];
        }

        // 如果有上次同步ID，用于去重（可选）
        if ($lastSyncId !== null) {
            $filter['_id'] = ['$gt' => $lastSyncId];
        }

        // 查询选项
        $options = [
            'sort' => ['created_at' => 1, '_id' => 1], // 按创建时间和ID排序
        ];

        // 执行查询（批量查询，每次最多1000条）
        $limit = $config['batch_size'] ?? 1000;
        $offset = 0;
        $allResults = [];

        do {
            // MongoDB 适配器的 queryBatch 方法签名：queryBatch(string $sql, array $params = [], int $offset = 0, int $limit = 1000)
            // 对于 MongoDB，$sql 是集合名，$params 包含 'filter' 和 'options'
            $results = $adapter->queryBatch($collectionName, [
                'filter' => $filter,
                'options' => $options,
            ], $offset, $limit);
            
            if (empty($results)) {
                break;
            }

            $allResults = array_merge($allResults, $results);
            $offset += $limit;

            // 防止无限循环（最多查询10万条）
            if (count($allResults) >= 100000) {
                LoggerHelper::logBusiness('polling_batch_limit_reached', [
                    'collection' => $collectionName,
                    'count' => count($allResults),
                ]);
                break;
            }
        } while (count($results) === $limit);

        LoggerHelper::logBusiness('polling_query_completed', [
            'collection' => $collectionName,
            'result_count' => count($allResults),
            'last_sync_time' => $lastSyncTime,
        ]);

        return $allResults;
    }

    /**
     * 数据转换
     * 
     * @param array<array<string, mixed>> $rawData 原始数据
     * @param array<string, mixed> $config 数据源配置
     * @return array<array<string, mixed>> 转换后的数据
     */
    public function transform(array $rawData, array $config): array
    {
        // 字段映射配置（从外部数据库字段映射到标准字段）
        $fieldMapping = $config['field_mapping'] ?? [
            // 默认映射（MongoDB 使用 _id，需要转换为 id）
            '_id' => 'id',
            'user_id' => 'user_id',
            'amount' => 'amount',
            'store_id' => 'store_id',
            'product_id' => 'product_id',
            'consume_time' => 'consume_time',
            'created_at' => 'created_at',
        ];

        $transformedData = [];

        foreach ($rawData as $record) {
            $transformed = [];

            // 处理 MongoDB 的 _id 字段（转换为字符串）
            if (isset($record['_id'])) {
                if (is_object($record['_id']) && method_exists($record['_id'], '__toString')) {
                    $record['id'] = (string)$record['_id'];
                } else {
                    $record['id'] = (string)$record['_id'];
                }
            }

            // 处理 MongoDB 的日期字段（UTCDateTime 转换为字符串）
            foreach (['created_at', 'updated_at', 'consume_time'] as $dateField) {
                if (isset($record[$dateField])) {
                    if (is_object($record[$dateField]) && method_exists($record[$dateField], 'toDateTime')) {
                        $record[$dateField] = $record[$dateField]->toDateTime()->format('Y-m-d H:i:s');
                    } elseif (is_object($record[$dateField]) && method_exists($record[$dateField], '__toString')) {
                        $record[$dateField] = (string)$record[$dateField];
                    }
                }
            }

            // 应用字段映射
            foreach ($fieldMapping as $standardField => $sourceField) {
                if (isset($record[$sourceField])) {
                    $transformed[$standardField] = $record[$sourceField];
                }
            }

            // 确保必要字段存在
            if (!empty($transformed)) {
                $transformedData[] = $transformed;
            }
        }

        LoggerHelper::logBusiness('polling_transform_completed', [
            'input_count' => count($rawData),
            'output_count' => count($transformedData),
        ]);

        return $transformedData;
    }

    /**
     * 数据验证
     * 
     * @param array<string, mixed> $record 单条记录
     * @param array<string, mixed> $config 数据源配置
     * @return bool 是否通过验证
     */
    public function validate(array $record, array $config): bool
    {
        // 必填字段验证
        $requiredFields = $config['required_fields'] ?? ['user_id', 'amount', 'consume_time'];

        foreach ($requiredFields as $field) {
            if (!isset($record[$field]) || $record[$field] === null || $record[$field] === '') {
                LoggerHelper::logBusiness('polling_validation_failed', [
                    'reason' => "缺少必填字段: {$field}",
                    'record' => $record,
                ]);
                return false;
            }
        }

        // 金额验证（必须为正数）
        if (isset($record['amount'])) {
            $amount = (float)$record['amount'];
            if ($amount <= 0) {
                LoggerHelper::logBusiness('polling_validation_failed', [
                    'reason' => '金额必须大于0',
                    'amount' => $amount,
                ]);
                return false;
            }
        }

        // 时间格式验证（可选）
        if (isset($record['consume_time'])) {
            $time = strtotime($record['consume_time']);
            if ($time === false) {
                LoggerHelper::logBusiness('polling_validation_failed', [
                    'reason' => '时间格式无效',
                    'consume_time' => $record['consume_time'],
                ]);
                return false;
            }
        }

        return true;
    }

    /**
     * 获取策略名称
     * 
     * @return string 策略名称
     */
    public function getName(): string
    {
        return 'mongodb_consumption';
    }
}