构建基于 Vercel, Couchbase 与 Scikit-learn 的离线优先事件溯源架构

事件溯源 (Event Sourcing) 的核心思想并非记录系统的当前状态，而是将引发状态变更的一系列领域事件 (Domain Events) 作为唯一事实源 (Single Source of Truth) 进行持久化。状态只是这些事件顺序聚合的结果。一个常见的误解是将其等同于操作日志或审计追踪，但其本质是一种截然不同的持久化范式。

一个典型的领域事件接口，它不是简单的数据传输对象，而是承载了业务语义和不变性的契约。

// src/domain/events.ts

/**
 * 所有领域事件的基础接口
 * @interface IDomainEvent
 * @template T - 事件负载的数据类型
 * @template V - 事件版本
 */
export interface IDomainEvent<T extends object = object, V extends number = number> {
  /**
   * 事件的唯一标识符
   */
  readonly eventId: string;

  /**
   * 事件所属的聚合根ID (例如, 用户ID)
   */
  readonly aggregateId: string;

  /**
   * 事件类型，用于反序列化和路由
   */
  readonly eventType: string;

  /**
   * 事件负载，包含所有与事件相关的数据
   */
  readonly payload: T;

  /**
   * 事件发生的时间戳 (ISO 8601格式)
   */
  readonly timestamp: string;

  /**
   * 事件在聚合流中的序列号，用于保证顺序和乐观并发
   */
  readonly version: V;

  /**
   * 触发此事件的元数据，如用户IP、会话ID等
   */
  readonly metadata: {
    correlationId: string; // 用于追踪跨服务调用的ID
    causationId: string;   // 触发此事件的命令ID
    userId?: string;       // 操作者ID
  };
}

// 示例：用户资料已更新事件
export interface UserProfileUpdatedPayload {
  displayName?: string;
  bio?: string;
}

export type UserProfileUpdatedEvent = IDomainEvent<UserProfileUpdatedPayload, 1>;

这种范式天然地解决了传统CRUD模型中的一些棘手问题，例如，我们不再丢失信息——每一次变更都被完整记录，使得回溯任何时间点的系统状态成为可能。它也是实现命令查询职责分离 (CQRS) 模式的理想基础，写模型专注于处理命令并生成事件，读模型（也称投影）则消费这些事件来构建和维护为查询优化的数据视图。

架构设计：Serverless 与边缘优先的事件溯源

在一个现代化的技术栈中，我们将挑战在无服务器 (Serverless) 环境下实现一个完整的事件溯源系统。该系统不仅要处理命令、存储事件，还要利用机器学习模型进行实时数据扩充，并通过 Service Workers 在客户端实现强大的离线读能力。

我们的技术选型如下：

• 命令处理与投影: Vercel Functions (Edge)
• 事件存储 (Event Store) 与读模型 (Read Model): Couchbase Capella (DBaaS)
• 实时机器学习投影: Scikit-learn 模型部署于 Python Vercel Function
• 客户端读模型缓存与离线同步: Service Workers

整体架构的数据流如下所示：

sequenceDiagram
    participant Client as 客户端 (Browser)
    participant SW as Service Worker
    participant VercelCmd as Vercel 命令处理函数 (Node.js)
    participant CouchbaseES as Couchbase (事件存储)
    participant VercelProj as Vercel 投影函数 (Node.js)
    participant VercelML as Vercel ML投影函数 (Python)
    participant CouchbaseRM as Couchbase (读模型)

    Note over Client, VercelCmd: 写路径 (Command Path)
    Client->>VercelCmd: POST /api/commands/updateProfile { userId, displayName }
    VercelCmd->>CouchbaseES: 1. 读取用户事件流 (Get events for userId)
    CouchbaseES-->>VercelCmd: 2. 返回事件列表
    VercelCmd->>VercelCmd: 3. 重放事件构建聚合状态 (Apply events)
    VercelCmd->>VercelCmd: 4. 验证命令并生成新事件 (UserProfileUpdatedEvent)
    VercelCmd->>CouchbaseES: 5. 原子性追加新事件 (Optimistic Locking)
    CouchbaseES-->>VercelCmd: 6. 确认写入
    VercelCmd-->>Client: 202 Accepted { commandId }

    Note over VercelProj, CouchbaseRM: 读路径 (Query Path) & 投影
    CouchbaseES->>VercelProj: [触发] Couchbase Eventing Service or Cron Job
    VercelProj->>CouchbaseES: 1. 读取新事件批次
    CouchbaseES-->>VercelProj: 2. 返回新事件
    VercelProj->>CouchbaseRM: 3. 更新用户资料读模型
    CouchbaseRM-->>VercelProj: 4. 确认更新

    Note over VercelML, CouchbaseRM: 机器学习增强投影
    CouchbaseES->>VercelML: [触发]
    VercelML->>CouchbaseES: 1. 读取新事件 (如 UserActivityLoggedEvent)
    VercelML->>VercelML: 2. 加载 Scikit-learn 模型
    VercelML->>VercelML: 3. 基于事件负载进行预测 (e.g., 用户流失风险)
    VercelML->>CouchbaseRM: 4. 更新用户风险评分读模型
    CouchbaseRM-->>VercelML: 5. 确认更新

    Note over Client, SW: 客户端离线读取
    Client->>SW: fetch('/api/query/userProfile/userId')
    SW->>SW: 1. 检查 Cache API
    alt 缓存命中 (Cache Hit)
        SW-->>Client: 2. 返回缓存的读模型数据
    else 缓存未命中 (Cache Miss)
        SW->>CouchbaseRM: 2. network.fetch()
        CouchbaseRM-->>SW: 3. 返回最新读模型
        SW->>SW: 4. 存入缓存
        SW-->>Client: 5. 返回数据
    end

关键实现：Couchbase 作为事件存储

Couchbase 的多模型特性使其非常适合同时作为事件存储和读模型存储。我们将使用其 Key-Value 操作的亚毫秒级延迟和 N1QL 查询的灵活性。

事件流将存储为单独的文档，文档ID遵循 aggregate_type::aggregate_id::version 格式，这使得按聚合ID和版本范围查询事件流变得高效。

// 文档 ID: user::user-123::1
{
  "eventId": "evt-uuid-001",
  "aggregateId": "user-123",
  "eventType": "UserRegistered",
  "payload": {
    "email": "[email protected]",
    "hashedPassword": "..."
  },
  "timestamp": "2023-10-27T10:00:00Z",
  "version": 1,
  "metadata": { ... }
}

// 文档 ID: user::user-123::2
{
  "eventId": "evt-uuid-002",
  "aggregateId": "user-123",
  "eventType": "UserProfileUpdated",
  "payload": {
    "displayName": "Test User"
  },
  "timestamp": "2023-10-27T10:05:00Z",
  "version": 2,
  "metadata": { ... }
}

为了管理乐观并发，我们需要一个单独的 “head” 文档来追踪每个聚合的最新版本。

// 文档 ID: user_head::user-123
{
  "aggregateId": "user-123",
  "version": 2,
  "type": "aggregate_head"
}

在命令处理函数中，保存事件的操作必须是事务性的。Couchbase 7.0+ 支持分布式 ACID 事务，这对于确保 head 文档和新事件文档的一致性至关重要。

// src/services/couchbaseEventStore.ts
import { Cluster, Transaction, GetResult, MutationResult } from 'couchbase';
import { IDomainEvent } from '../domain/events';

// 从环境变量初始化 Couchbase 连接
const cluster = await Cluster.connect(process.env.CB_HOST!, {
  username: process.env.CB_USER!,
  password: process.env.CB_PASSWORD!,
});
const bucket = cluster.bucket(process.env.CB_BUCKET!);
const collection = bucket.scope('event_sourcing').collection('events');
const headCollection = bucket.scope('event_sourcing').collection('aggregate_heads');

export class CouchbaseEventStore {
  public async saveEvents(
    aggregateType: string,
    aggregateId: string,
    expectedVersion: number,
    events: IDomainEvent[]
  ): Promise<void> {
    const headDocId = `${aggregateType}_head::${aggregateId}`;

    try {
      // 启动 Couchbase 事务
      await cluster.transactions().run(async (ctx: Transaction) => {
        // 1. 获取当前的 head 文档并进行 CAS 检查
        let headDoc: GetResult | null = null;
        try {
          headDoc = await ctx.get(headCollection, headDocId);
        } catch (e) {
          // 文档不存在是正常情况，对于新的聚合
          if (expectedVersion !== 0) throw new Error('Concurrency conflict: Head document not found for existing aggregate.');
        }

        const currentVersion = headDoc ? headDoc.content.version : 0;
        if (currentVersion !== expectedVersion) {
          throw new Error(`Concurrency conflict: Expected version ${expectedVersion}, but found ${currentVersion}.`);
        }

        // 2. 在事务中插入所有新事件
        let newVersion = currentVersion;
        for (const event of events) {
          newVersion++;
          if (event.version !== newVersion) {
             throw new Error(`Event version mismatch. Expected ${newVersion}, got ${event.version}`);
          }
          const eventDocId = `${aggregateType}::${aggregateId}::${event.version}`;
          await ctx.insert(collection, eventDocId, event);
        }

        // 3. 更新 head 文档
        const newHeadContent = { aggregateId, version: newVersion, type: 'aggregate_head' };
        if (headDoc) {
          await ctx.replace(headDoc, newHeadContent);
        } else {
          await ctx.insert(headCollection, headDocId, newHeadContent);
        }
      });
    } catch (err: any) {
      // 事务会自动回滚
      console.error('Failed to save events transactionally', { error: err.message, aggregateId });
      // 在真实项目中，这里应该根据错误类型进行更细致的处理
      throw err;
    }
  }

  public async getEventsForAggregate(
    aggregateType: string,
    aggregateId: string
  ): Promise<IDomainEvent[]> {
    // N1QL 查询效率不如 KV，但对于重放整个事件流是可行的。
    // 在生产中，可以考虑快照（Snapshotting）来优化。
    const query = `
      SELECT e.*
      FROM \`${bucket.name}\`.\`event_sourcing\`.\`events\` e
      WHERE e.aggregateId = $1
      ORDER BY e.version ASC;
    `;
    try {
      const result = await cluster.query(query, { parameters: [aggregateId] });
      return result.rows as IDomainEvent[];
    } catch (err) {
      console.error('Failed to get events for aggregate', { err, aggregateId });
      return [];
    }
  }
}

Vercel Functions: 无状态的命令处理与投影

Vercel Functions 的无状态特性与事件溯源的聚合模型完美契合。每个命令处理函数都是一个独立的计算单元，它加载事件、重建状态、执行业务逻辑、然后持久化新事件。

// pages/api/commands/updateUserProfile.ts
import type { VercelRequest, VercelResponse } from '@vercel/node';
import { CouchbaseEventStore } from '../../../src/services/couchbaseEventStore';
import { UserAggregate } from '../../../src/domain/userAggregate';
import { UserProfileUpdatedEvent } from '../../../src/domain/events';

// 单元测试思路：
// 1. Mock CouchbaseEventStore.
// 2. 测试输入验证失败场景 (400 Bad Request).
// 3. 测试成功场景：验证 store.saveEvents 被正确调用，参数正确。
// 4. 测试并发冲突场景：模拟 store.saveEvents 抛出并发异常，验证返回 409 Conflict.

export default async function handler(req: VercelRequest, res: VercelResponse) {
  if (req.method !== 'POST') {
    return res.status(405).json({ message: 'Method Not Allowed' });
  }

  const { aggregateId, expectedVersion, displayName, bio } = req.body;

  // 在生产环境中，应该使用更健壮的验证库，如 Zod 或 Joi
  if (!aggregateId || typeof expectedVersion !== 'number') {
    return res.status(400).json({ message: 'Invalid input' });
  }

  const store = new CouchbaseEventStore();
  const user = new UserAggregate(aggregateId);

  try {
    // 1. 加载历史事件
    const events = await store.getEventsForAggregate('user', aggregateId);
    
    // 2. 重建聚合状态
    user.loadFromHistory(events);

    // 3. 执行命令
    const newEvents: UserProfileUpdatedEvent[] = user.updateProfile({ displayName, bio });

    // 4. 保存新事件
    await store.saveEvents('user', aggregateId, expectedVersion, newEvents);

    // 返回 202 Accepted，表示命令已被接受处理，但结果是异步的
    res.status(202).json({ message: 'Command accepted' });

  } catch (error: any) {
    // 精细化错误处理
    if (error.message.includes('Concurrency conflict')) {
      res.status(409).json({ message: error.message });
    } else {
      console.error(`Error processing updateUserProfile command for ${aggregateId}:`, error);
      res.status(500).json({ message: 'Internal Server Error' });
    }
  }
}

投影函数同样可以部署为 Vercel Function。触发机制可以是 Couchbase Eventing Service 发出的 webhook，也可以是 Vercel Cron Jobs 定期轮询新事件。定期轮询更简单，但会有延迟。

Scikit-learn 赋能的智能投影

事件流是机器学习模型的绝佳数据源。我们可以创建一个 Python Vercel Function，它消费特定类型的事件（例如 UserActivityLogged），并使用预训练的 Scikit-learn 模型来更新读模型，比如预测用户流失风险或计算用户参与度得分。

首先，你需要一个训练好的模型。假设我们已经训练了一个简单的逻辑回归模型来根据用户行为预测其是否为高价值用户，并将其保存为 model.pkl。

# requirements.txt
# vercel-python
# scikit-learn
# pandas
# joblib
# couchbase

# /api/projections/updateUserValueScore.py
from http.server import BaseHTTPRequestHandler
from vercel_app.app import VercelApp
import joblib
import pandas as pd
from couchbase.cluster import Cluster, ClusterOptions
from couchbase.auth import PasswordAuthenticator
import os
import json

# 在函数外部加载模型，以便在热函数中复用
# 模型文件需要和函数一起部署
try:
    MODEL_PATH = os.path.join(os.path.dirname(__file__), 'model.pkl')
    model = joblib.load(MODEL_PATH)
except Exception as e:
    # 关键的错误处理：如果模型加载失败，函数将无法工作
    print(f"FATAL: Could not load Scikit-learn model from {MODEL_PATH}. Error: {e}")
    model = None

# Couchbase 连接同样应在外部初始化
try:
    cluster = Cluster(
        os.environ.get("CB_HOST"),
        ClusterOptions(PasswordAuthenticator(os.environ.get("CB_USER"), os.environ.get("CB_PASSWORD")))
    )
    bucket = cluster.bucket(os.environ.get("CB_BUCKET"))
    read_model_collection = bucket.scope("read_models").collection("user_profiles")
except Exception as e:
    print(f"FATAL: Could not connect to Couchbase. Error: {e}")
    cluster = None

app = VercelApp()

@app.route('/', methods=['POST'])
def handler(request: BaseHTTPRequestHandler):
    if not model or not cluster:
        return {'status': 500, 'body': 'ML model or DB connection not initialized'}

    # 假设触发器（如Couchbase Eventing）将事件作为POST body发送
    try:
        event = json.loads(request.body)
        if event.get('eventType') != 'UserActivityLogged':
            # 只处理我们关心的事件
            return {'status': 200, 'body': 'Event type skipped'}

        # 1. 从事件负载中提取特征
        payload = event.get('payload', {})
        features = {
            'login_count_last_30d': payload.get('loginCount'),
            'feature_x_usage': payload.get('featureXUsage'),
            'session_duration_avg': payload.get('avgSessionDuration'),
        }
        
        # 2. 使用 Pandas DataFrame 准备模型输入
        # Scikit-learn 模型期望的输入格式
        input_df = pd.DataFrame([features])

        # 3. 进行预测
        # predict_proba 返回每个类别的概率，我们取高价值类别（假设是第1类）的概率
        value_score = model.predict_proba(input_df)[0][1]

        # 4. 更新 Couchbase 中的读模型
        # 使用 upsert，无论文档是否存在都会写入
        user_id = event.get('aggregateId')
        read_model_collection.upsert(
            f"user_profile::{user_id}",
            {'valueScore': round(value_score, 4)},
        )

        return {'status': 200, 'body': f'Updated value score for user {user_id}'}

    except Exception as e:
        print(f"Error processing ML projection: {e}")
        return {'status': 500, 'body': 'Internal Server Error'}

Service Workers：实现离线优先的读模型

客户端体验是 CQRS 模式的最终受益者。Service Workers 允许我们在浏览器端为读模型实现一个强大的、离线优先的缓存层。

Service Worker 会拦截对读模型 API 的 fetch 请求。如果网络可用且缓存过期，它会从网络获取最新数据，更新缓存，并返回给应用。如果网络不可用，它会直接从缓存中返回数据，保证了应用的离线可用性。

// public/sw.js

const CACHE_NAME = 'read-model-cache-v1';
const READ_MODEL_API_PREFIX = '/api/query/';

// 安装时预缓存核心资源
self.addEventListener('install', (event) => {
  event.waitUntil(
    caches.open(CACHE_NAME).then((cache) => {
      // 可以预缓存一些静态外壳
      return cache.addAll(['/', '/styles/main.css', '/app.js']);
    })
  );
});

// 拦截 fetch 请求
self.addEventListener('fetch', (event) => {
  const { request } = event;

  // 只处理我们关心的读模型API请求
  if (request.url.includes(READ_MODEL_API_PREFIX) && request.method === 'GET') {
    event.respondWith(
      // 策略：Stale-While-Revalidate
      // 优先从缓存返回，然后异步更新缓存
      caches.open(CACHE_NAME).then((cache) => {
        return cache.match(request).then((cachedResponse) => {
          const fetchPromise = fetch(request).then((networkResponse) => {
            // 确保我们有一个有效的响应
            if (networkResponse && networkResponse.status === 200) {
              cache.put(request, networkResponse.clone());
            }
            return networkResponse;
          }).catch(error => {
            // 网络请求失败时，什么都不做，让缓存继续服务
            console.warn('Network request failed, serving from cache.', error);
          });

          // 如果缓存中有数据，立即返回，否则等待网络响应
          return cachedResponse || fetchPromise;
        });
      })
    );
  }
  // 对于其他请求，使用默认的网络优先策略
});

这种架构的实现并非没有挑战。Serverless 函数的冷启动会增加命令处理的延迟，这对于需要即时反馈的场景是个问题。Vercel 的 Edge Functions 可以缓解部分问题，但不能完全消除。其次，最终一致性是这套架构的固有属性，UI/UX 设计必须考虑到读模型数据可能存在的短暂延迟。最后，事件的 schema 演进需要一套严格的管理流程，因为旧事件需要能够被新版本的代码正确地重放。

尽管存在这些权衡，这种基于 Vercel、Couchbase、Scikit-learn 和 Service Workers 的事件溯源架构，为构建高可伸缩、富有弹性且具备离线能力的现代应用提供了一个强大的、非传统的实现路径。它将后端的不可变事实源与前端的极致用户体验通过一系列解耦的、功能单一的服务连接了起来。未来的优化方向可能包括引入快照机制来减少聚合重放时间，以及使用 WebSockets 或 Server-Sent Events 来实现读模型的实时推送，进一步降低数据延迟。