日均3000+查询零丢失：AI对话工具的异步执行与独立会话管理实战

// 格式：{userId}:{timestamp}:{randomSuffix}
// 示例：u_10086:1725984000000:a3f8k2
function createSessionId(userId: string): string {
  return `${userId}:${Date.now()}:${nanoid(6)}`;
}

interface SessionState {
  id: string;
  status: 'streaming' | 'executing' | 'completed' | 'failed';
  messages: Message[];        // 对话历史
  sql?: string;               // 生成的SQL
  result?: QueryResult;       // 查询结果
  error?: string;
  createdAt: number;
}

// 核心：Map结构，key=sessionId，各会话完全隔离
const sessions = useRef<Map<string, SessionState>>(new Map());

// 切换当前活跃会话 ≠ 销毁其他会话
function switchSession(sessionId: string) {
  activeSessionId.current = sessionId;
  // 其他会话的状态完整保留，SSE连接保持或按需重连
}

import { Queue } from 'bullmq';

const queryQueue = new Queue('sql-execution', { connection: redisConnection });

async function submitQuery(sessionId: string, sql: string, userId: string) {
  await queryQueue.add(
    'execute',
    { sessionId, sql, userId },
    {
      jobId: sessionId,          // 用sessionId做jobId，天然幂等
      attempts: 3,               // 失败重试3次
      backoff: { type: 'exponential', delay: 2000 },
      removeOnComplete: 1000,    // 成功后保留1000条记录用于排查
      removeOnFail: 5000,        // 失败保留更多
    }
  );
}

import { Worker } from 'bullmq';

const worker = new Worker('sql-execution', async (job) => {
  const { sessionId, sql, userId } = job.data;

  // 1. 通知前端"开始执行"
  await eventBus.publish(sessionId, {
    event: 'exec_status',
    data: { status: 'running', progress: 0 }
  });

  // 2. 执行SQL（带超时保护）
  const result = await executeWithTimeout(sql, {
    timeout: 30000,
    maxRows: 10000,
    userId, // 用于行级权限过滤
  });

  // 3. 结果落库（不通过SSE直传大数据）
  const resultId = await storeResult(sessionId, result);

  // 4. 通知前端"执行完成"，只传resultId
  await eventBus.publish(sessionId, {
    event: 'result',
    data: { resultId, rowCount: result.rows.length }
  });

  return { resultId }; // Bull记录返回值
}, { concurrency: 20 }); // 控制并发，防止打爆数据库

async function executeWithTimeout(sql: string, opts: TimeoutOpts) {
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), opts.timeout);

  try {
    return await db.query(sql, { signal: controller.signal, maxRows: opts.maxRows });
  } catch (err) {
    if (controller.signal.aborted) {
      throw new Error(`Query timeout after ${opts.timeout}ms`);
    }
    throw err;
  } finally {
    clearTimeout(timer);
  }
}

// 事件总线基于Redis Pub/Sub，跨进程/跨实例通信
class EventBus {
  async publish(sessionId: string, message: SSEMessage) {
    // 写入Redis Stream作为持久化备份
    await redis.xadd(`session:${sessionId}:events`, '*', ...serialize(message));
    // 同时Pub/Sub实时推送
    await redis.publish(`channel:session:${sessionId}`, JSON.stringify(message));
  }
}

// SSE Server订阅对应channel
redis.subscribe(`channel:session:${sessionId}`, (msg) => {
  const sseRes = sessionManager.get(sessionId)?.res;
  if (sseRes && !sseRes.writableEnded) {
    sseRes.write(`event: ${msg.event}\ndata: ${JSON.stringify(msg.data)}\n\n`);
  }
});

// 用户重新打开页面时调用
app.get('/api/session/:sessionId/events', async (req, res) => {
  const events = await redis.xrange(
    `session:${req.params.sessionId}:events`,
    req.query.lastEventId || '-', // 支持续传
    '+',
    'COUNT', 100 // 单次最多100条，分页拉取
  );
  res.json(events.map(deserialize));
});

// 前端收到result事件后
async function onResult(sessionId: string, resultId: string) {
  // 不立即拉取，等用户切换到该会话或展开结果区域时才请求
  const result = await fetch(`/api/result/${resultId}`).then(r => r.json());
  renderChartOrTable(sessionId, result);
}