// 伪代码：大报表自动分片逻辑
class MeetingExportJob implements ShouldQueue
{
    public function handle()
    {
        $total = $this->meeting->attendees()->count();
        $chunkSize = 5000;
        $chunks = ceil($total / $chunkSize);

        if ($chunks <= 1) {
            $this->generateSingleFile();
            return;
        }

        // 创建主任务记录，状态=PROCESSING
        $exportTask = ExportTask::create([
            'meeting_id' => $this->meeting_id,
            'total_chunks' => $chunks,
            'status' => 'processing',
        ]);

        // 分发子任务到队列，携带分片索引
        for ($i = 0; $i < $chunks; $i++) {
            ExportChunkJob::dispatch($exportTask->id, $i, $chunkSize)
                ->onQueue('export_high');
        }
    }
}

// 前端 SSE 监听进度
const evtSource = new EventSource(`/api/export/progress/${taskId}`);
evtSource.onmessage = (e) => {
  const { progress, stage, message } = JSON.parse(e.data);
  updateProgressBar(progress, stage, message);
  if (progress === 100) {
    evtSource.close();
    window.location.href = message.downloadUrl; // 签名下载链接
  }
};

$callback = function () use ($meetingId, $chunkIndex, $chunkSize) {
    $handle = fopen('php://temp', 'w+');
    fputcsv($handle, $this->headers); // 写表头

    Attendee::on('read_replica')
        ->where('meeting_id', $meetingId)
        ->orderBy('id')
        ->skip($chunkIndex * $chunkSize)
        ->take($chunkSize)
        ->chunkById(500, function ($attendees) use ($handle) {
            foreach ($attendees as $a) {
                fputcsv($handle, $this->mapRow($a));
            }
        });

    rewind($handle);
    Storage::disk('oss')->put("exports/{$taskId}/chunk_{$chunkIndex}.csv", $handle);
    fclose($handle);
};

# RabbitMQ 队列声明示例（含死信交换机）
arguments:
  x-dead-letter-exchange: "dlx.export"
  x-dead-letter-routing-key: "export.failed"
  x-message-ttl: 86400000  # 24h TTL 防止堆积

// 生成带有效期、单次使用的下载签名
public function generateSignedUrl(ExportTask $task): string
{
    $payload = [
        'task_id' => $task->id,
        'file_path' => $task->final_file_path,
        'exp' => now()->addMinutes(30)->timestamp,
        'nonce' => Str::random(16),
    ];
    $token = JWT::encode($payload, config('app.export_secret'));
    return route('export.download', ['token' => $token]);
}

-- 每日清理 7 天前的临时分片文件及任务记录
DELETE FROM export_tasks 
WHERE status IN ('completed','failed') 
  AND updated_at < DATE_SUB(NOW(), INTERVAL 7 DAY);

-- 离线任务每小时刷新一次会议宽表（ClickHouse / Doris / MySQL 分区表）
CREATE TABLE meeting_attendee_wide_v2 (
    meeting_id BIGINT,
    attendee_id BIGINT,
    `base_info` JSON,      -- 姓名、手机、公司等基础字段
    `custom_fields` JSON,  -- 动态扩展字段（K-V 结构）
    `checkin_stats` JSON,  -- 签到次数、首签时间、时长等聚合指标
    `tags` ARRAY(STRING),  -- 标签体系
    updated_at DATETIME
) ENGINE=MergeTree()
PARTITION BY toYYYYMMDD(meeting_date)
ORDER BY (meeting_id, attendee_id);

// 根据用户勾选的 columns 动态投影，下推到存储层
$selectedColumns = array_merge(['meeting_id', 'attendee_id'], $userSelectedFields);
$rows = WideTable::on('clickhouse')
    ->where('meeting_id', $meetingId)
    ->select($selectedColumns)  // 列裁剪，IO 极低
    ->orderBy('attendee_id')
    ->cursor(); // 游标流式读取

# docker-compose.renderer.yml
services:
  chrome-renderer:
    image: ghcr.io/puppeteer/puppeteer:latest
    deploy:
      replicas: 4
      resources:
        limits:
          cpus: '2'
          memory: 2G
    environment:
      - CONCURRENCY=2  # 单容器并发渲染数
    volumes:
      - ./fonts:/usr/share/fonts/truetype/custom  # 中文字体解决乱码

sequenceDiagram
    participant Client
    participant ExportAPI
    participant Orchestrator
    participant MQ
    participant ServiceA
    participant ServiceB
    Client->>ExportAPI: POST /export (meeting_id, template_id)
    ExportAPI->>Orchestrator: 创建导出编排实例 (状态=PENDING)
    Orchestrator->>MQ: 发送子任务: fetch_base, fetch_checkin, fetch_survey
    par 并行拉取
        ServiceA->>MQ: 消费 fetch_base -> 写入临时表/Redis -> 回调完成
        ServiceB->>MQ: 消费 fetch_checkin -> 写入临时表 -> 回调完成
    end
    Orchestrator->>Orchestrator: 聚合器监听所有子任务完成事件
    Orchestrator->>MQ: 发送 generate_file 任务

# 消费者启动时动态绑定队列（伪代码）
class TenantAwareConsumer:
    def __init__(self, tenant_id: str):
        self.tenant_id = tenant_id
        # 高优先级队列：VIP 租户、实时导出
        self.high_q = f"export.tenant.{tenant_id}.high"
        # 普通队列
        self.normal_q = f"export.tenant.{tenant_id}.normal"
        # 低优先级队列：定时任务、历史回溯
        self.low_q = f"export.tenant.{tenant_id}.low"
        
        # Redis 令牌桶：每租户并发消费者数上限
        self.rate_limiter = TokenBucket(
            key=f"ratelimit:consumer:{tenant_id}",
            capacity=config.MAX_CONCURRENT_PER_TENANT, # 如 5
            refill_rate=1.0
        )

    def consume_loop(self):
        while True:
            # 优先消费 high，权重 60% / 30% / 10%
            task = self.try_pop_weighted([self.high_q, self.normal_q, self.low_q])
            if not task: 
                time.sleep(1); continue
            
            if not self.rate_limiter.try_acquire():
                self.requeue(task) # 归队等待
                time.sleep(0.5); continue
            
            self.process(task)
            self.rate_limiter.release()

rules:
  - name: "租户导出异常熔断"
    condition: |
      tenant.export_failure_rate_5m > 0.3 
      AND tenant.avg_task_duration_5m > 600
    action:
      - type: "throttle"
        params: { max_concurrent: 1 }  # 降级为单线程
      - type: "alert"
        params: { level: "P2", msg: "租户 {tenant_id} 导出异常，已自动限流" }
      - type: "tag"
        params: { key: "quarantine", value: "true" } # 标记隔离
  - name: "大任务拆分强制生效"
    condition: "task.estimated_rows > 50000 AND task.chunk_size > 10000"
    action:
      - type: "rewrite_task"
        params: { chunk_size: 5000, force_split: true }

graph LR
    A[用户点击导出] --> B[API 网关]
    B --> C[函数计算 FC: CreateTask]
    C --> D[(数据库: 创建任务记录)]
    C --> E[对象存储 OSS: 写入任务元数据 JSON]
    E -- ObjectCreated Event --> F[函数计算 FC: Worker]
    F --> G[读取任务 -> 查询数据 -> 生成 CSV -> 写入 OSS]
    G --> H[函数计算 FC: Merger]
    H -- 所有分片完成触发 --> I[合并文件 -> 生成最终报表]
    I --> J[数据库更新下载链接]
    J --> K[WebSocket/短信通知用户]

# KEDA ScaledObject: 基于 RabbitMQ 队列长度自动伸缩
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: export-worker-scaler
spec:
  scaleTargetRef:
    name: export-worker-deployment
  pollingInterval: 15
  cooldownPeriod: 300
  minReplicaCount: 0  # 闲时缩容至 0，极致省钱
  maxReplicaCount: 50
  triggers:
  - type: rabbitmq
    metadata:
      host: amqp://user:pass@rabbitmq:5672
      queueName: export.high.priority
      queueLength: "10"  # 每 10 条消息触发 1 个 Pod
  - type: rabbitmq
    metadata:
      queueName: export.normal
      queueLength: "30"
  advanced:
    restoreToOriginalReplicaCount: false
    horizontalPodAutoscalerConfig:
      behavior:
        scaleDown:
          stabilizationWindowSeconds: 600
          policies:
          - type: Percent
            value: 10
            periodSeconds: 60

// 消费者测试 (Groovy Spock + Pact)
def "导出任务请求用户服务获取基础信息"() {
    given: "用户服务返回指定结构"
    service.uponReceiving("获取参会人基础信息")
        .path("/api/v1/users/batch")
        .method("POST")
        .body([
            user_ids: [1001, 1002],
            fields: ['name', 'phone', 'company']
        ])
        .willRespondWith(
            status: 200,
            headers: ['Content-Type': 'application/json'],
            body: eachLike([
                id: 1001,
                name: "张三",
                phone: "138****8888",
                company: "示例科技"
            ], minimum: 1)
        )

    when: "执行导出分片任务"
    def result = exportChunkJob.execute(meetingId: 1, chunkIndex: 0)

    then: "断言数据正确拼装"
    result.rows[0].name == "张三"
    result.rows[0].phone == "138****8888"
}

# 使用 Faker + 数据分布模型生成百万级真实感测试数据
from faker import Faker
import numpy as np

fake = Faker('zh_CN')

def generate_attendees(meeting_id, count=1_000_000):
    # 模拟真实分布：20% VIP、签到率 85%、手机号归属地分布
    vip_ids = set(random.sample(range(count), int(count * 0.2)))
    for i in range(count):
        yield {
            "meeting_id": meeting_id,
            "attendee_id": 1000000 + i,
            "name": fake.name(),
            "phone": fake.phone_number(),
            "company": fake.company(),
            "is_vip": i in vip_ids,
            "checkin_count": np.random.poisson(3) if random.random() < 0.85 else 0,
            "created_at": fake.date_time_between("-30d", "now")
        }

# 批量写入 ClickHouse / MySQL 分区表，分钟级造数百万行

# chaosmesh-network-delay.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: export-db-latency
spec:
  action: delay
  mode: one
  selector:
    namespaces: [production]
    labelSelectors:
      app: export-worker
  delay:
    latency: "500ms"
    correlation: "25"
    jitter: "100ms"
  duration: "5m"
  scheduler:
    cron: "@every 1h"

-- StarRocks / Doris / ClickHouse 物化视图自动维护
CREATE MATERIALIZED VIEW mv_meeting_attendee_rich
DISTRIBUTED BY HASH(meeting_id) BUCKETS 32
REFRESH ASYNC START '2024-01-01 00:00:00' EVERY 1 MINUTE
AS
SELECT 
    a.meeting_id,
    a.attendee_id,
    u.name, u.phone, u.company,
    c.checkin_count, c.first_checkin_time,
    s.survey_score,
    crm.lead_stage
FROM attendees a
JOIN users u ON a.user_id = u.id
LEFT JOIN checkin_agg c ON a.attendee_id = c.attendee_id
LEFT JOIN survey_agg s ON a.attendee_id = s.attendee_id
LEFT JOIN crm_leads crm ON u.phone = crm.phone;

// 直接走 OLAP 引擎，支持向量化执行、列裁剪、谓词下推
String sql = "SELECT " + columns + " FROM mv_meeting_attendee_rich WHERE meeting_id = ?";
try (ResultSet rs = olapClient.executeQuery(sql)) {
    // 直接流式写入 CSV/Parquet -> OSS
    CsvWriter.write(rs, ossOutputStream); 
}

用户输入："导出上周北京场会议的 VIP 客户签到率、满意度评分、意向等级分布，做成带图表的 PDF"
系统执行：
1. LLM 解析意图 -> 生成 SQL + 图表配置 (ECharts JSON) + 报表模板
2. 校验 SQL 安全性 (只读、无危险函数、行级权限)
3. 提交异步任务 -> 渲染图表 -> 填充模板 -> 生成 PDF
4. 返回下载链接