🔀 CQRS 命令查询职责分离

// === Command Side ===
interface Command {
  type: string;
}

interface CreateOrderCommand extends Command {
  type: 'CreateOrder';
  userId: string;
  items: Array<{ productId: string; quantity: number }>;
}

// Command Handler - 写侧有业务逻辑
class OrderCommandHandler {
  constructor(private db: Database) {}

  async handle(cmd: CreateOrderCommand): Promise<string> {
    // 验证库存
    for (const item of cmd.items) {
      const stock = await this.db.query(
        'SELECT stock FROM products WHERE id = $1', [item.productId]
      );
      if (stock.rows[0].stock < item.quantity) {
        throw new Error(`库存不足: ${item.productId}`);
      }
    }

    // 创建订单
    const orderId = crypto.randomUUID();
    await this.db.query(
      `INSERT INTO orders (id, user_id, status, created_at)
       VALUES ($1, $2, 'pending', NOW())`,
      [orderId, cmd.userId]
    );

    // 扣减库存 + 添加订单项（事务）
    await this.db.transaction(async (tx) => {
      for (const item of cmd.items) {
        await tx.query(
          'UPDATE products SET stock = stock - $1 WHERE id = $2',
          [item.quantity, item.productId]
        );
        await tx.query(
          `INSERT INTO order_items (order_id, product_id, quantity)
           VALUES ($1, $2, $3)`,
          [orderId, item.productId, item.quantity]
        );
      }
    });

    return orderId;
  }
}

// === Query Side ===
// 查询侧只读，返回 DTO，没有业务逻辑
interface OrderDTO {
  id: string;
  userId: string;
  status: string;
  items: Array<{ productId: string; name: string; quantity: number; price: number }>;
  totalAmount: number;
  createdAt: string;
}

class OrderQueryHandler {
  constructor(private db: Database) {}

  async getOrderById(id: string): Promise<OrderDTO | null> {
    const result = await this.db.query(`
      SELECT o.id, o.user_id, o.status, o.created_at,
             json_agg(json_build_object(
               'productId', oi.product_id,
               'name', p.name,
               'quantity', oi.quantity,
               'price', p.price
             )) as items
      FROM orders o
      JOIN order_items oi ON o.id = oi.order_id
      JOIN products p ON oi.product_id = p.id
      WHERE o.id = $1
      GROUP BY o.id
    `, [id]);

    if (!result.rows[0]) return null;

    const row = result.rows[0];
    return {
      id: row.id,
      userId: row.user_id,
      status: row.status,
      items: row.items,
      totalAmount: row.items.reduce((sum: number, i: any) =>
        sum + i.price * i.quantity, 0),
      createdAt: row.created_at,
    };
  }
}

// 写侧：Command 处理后发布事件
class OrderCommandHandler {
  constructor(
    private db: Database,
    private eventBus: EventBus  // Kafka / RabbitMQ
  ) {}

  async handle(cmd: CreateOrderCommand): Promise<string> {
    const orderId = crypto.randomUUID();

    await this.db.transaction(async (tx) => {
      // ... 业务逻辑和写入 ...
      await tx.query(/* ... */);

      // 在事务内发布事件（Outbox Pattern 保证原子性）
      await tx.query(
        `INSERT INTO outbox_events (event_type, aggregate_id, payload)
         VALUES ($1, $2, $3)`,
        ['OrderCreated', orderId, JSON.stringify({
          orderId, userId: cmd.userId,
          items: cmd.items,
          timestamp: Date.now()
        })]
      );
    });

    return orderId;
  }
}

// Outbox Poller：将事件从 outbox 表推送到消息队列
class OutboxPoller {
  async poll() {
    const events = await this.db.query(
      `SELECT * FROM outbox_events
       WHERE processed = false
       ORDER BY created_at ASC LIMIT 100`
    );

    for (const event of events.rows) {
      await this.eventBus.publish(event.event_type, event.payload);

      await this.db.query(
        'UPDATE outbox_events SET processed = true WHERE id = $1',
        [event.id]
      );
    }
  }
}

// 读侧：事件处理器构建读模型
class OrderReadModelProjector {
  constructor(private esClient: ElasticsearchClient) {}

  async onOrderCreated(event: OrderCreatedEvent) {
    // 将订单数据写入 Elasticsearch，优化搜索体验
    await this.esClient.index({
      index: 'orders',
      id: event.orderId,
      body: {
        orderId: event.orderId,
        userId: event.userId,
        items: event.items,
        status: 'pending',
        createdAt: new Date(event.timestamp).toISOString()
      }
    });
  }

  async onOrderStatusChanged(event: OrderStatusChangedEvent) {
    await this.esClient.update({
      index: 'orders',
      id: event.orderId,
      body: { doc: { status: event.newStatus } }
    });
  }
}

// 读侧：查询处理器直接查读数据库
class OrderQueryHandler {
  constructor(private esClient: ElasticsearchClient) {}

  async searchOrders(query: string, userId: string): Promise<OrderDTO[]> {
    const result = await this.esClient.search({
      index: 'orders',
      body: {
        query: {
          bool: {
            must: [
              { term: { userId } },
              { multi_match: { query, fields: ['items.name', 'status'] } }
            ]
          }
        }
      }
    });
    return result.hits.hits.map((h: any) => h._source);
  }
}

// === 写侧：Agent 记忆写入 ===
interface AgentMemoryEvent {
  type: 'user_message' | 'assistant_reply' | 'tool_call' | 'tool_result' | 'thinking';
  sessionId: string;
  content: string;
  metadata?: Record<string, any>;
  timestamp: number;
}

class AgentMemoryWriteModel {
  constructor(private pg: Pool) {}

  // 写入是 append-only，类似 Event Sourcing
  async append(event: AgentMemoryEvent): Promise<void> {
    await this.pg.query(
      `INSERT INTO agent_events (session_id, event_type, content, metadata, created_at)
       VALUES ($1, $2, $3, $4, $5)`,
      [event.sessionId, event.type, event.content, event.metadata, new Date(event.timestamp)]
    );
  }

  // 批量写入（工具调用链）
  async appendBatch(events: AgentMemoryEvent[]): Promise<void> {
    const client = await this.pg.connect();
    try {
      await client.query('BEGIN');
      for (const event of events) {
        await client.query(
          `INSERT INTO agent_events (session_id, event_type, content, metadata, created_at)
           VALUES ($1, $2, $3, $4, $5)`,
          [event.sessionId, event.type, event.content, event.metadata, new Date(event.timestamp)]
        );
      }
      await client.query('COMMIT');
    } catch (e) {
      await client.query('ROLLBACK');
      throw e;
    } finally {
      client.release();
    }
  }
}

// === 读侧：Agent 记忆检索 ===
interface MemorySearchResult {
  content: string;
  score: number;
  source: 'semantic' | 'keyword' | 'recent';
  timestamp: number;
}

class AgentMemoryReadModel {
  constructor(
    private pg: Pool,           // 原始事件
    private vectorStore: VectorStore,  // pgvector / Pinecone
    private searchEngine: SearchEngine  // Meilisearch
  ) {}

  // 语义搜索：找到语义相关的历史记忆
  async searchSemantic(query: string, limit = 10): Promise<MemorySearchResult[]> {
    const queryEmbedding = await embed(query);
    return this.vectorStore.similaritySearch(queryEmbedding, limit);
  }

  // 关键词搜索：精确匹配
  async searchKeyword(query: string, sessionId?: string): Promise<MemorySearchResult[]> {
    const filter = sessionId ? { sessionId } : undefined;
    return this.searchEngine.search(query, { filter, limit: 10 });
  }

  // 最近上下文：构造 LLM 的上下文窗口
  async getRecentContext(sessionId: string, limit = 20): Promise<AgentMemoryEvent[]> {
    const result = await this.pg.query(
      `SELECT event_type, content, metadata, created_at
       FROM agent_events
       WHERE session_id = $1
       ORDER BY created_at DESC
       LIMIT $2`,
      [sessionId, limit]
    );
    return result.rows.reverse(); // 按时间正序
  }

  // 混合检索：语义 + 关键词 + 最近上下文
  async getFullContext(query: string, sessionId: string): Promise<{
    recent: AgentMemoryEvent[];
    semantic: MemorySearchResult[];
    keyword: MemorySearchResult[];
  }> {
    const [recent, semantic, keyword] = await Promise.all([
      this.getRecentContext(sessionId),
      this.searchSemantic(query),
      this.searchKeyword(query, sessionId)
    ]);
    return { recent, semantic, keyword };
  }
}

// === 同步层：写侧事件 → 读模型更新 ===
class MemoryProjector {
  constructor(
    private pg: Pool,
    private vectorStore: VectorStore,
    private searchEngine: SearchEngine
  ) {}

  // 订阅新事件，更新读模型
  async project(event: AgentMemoryEvent): Promise<void> {
    // 1. 生成 embedding 并写入向量库
    if (event.content.length > 10) {
      const embedding = await embed(event.content);
      await this.vectorStore.upsert({
        id: `${event.sessionId}:${event.timestamp}`,
        embedding,
        metadata: {
          sessionId: event.sessionId,
          type: event.type,
          timestamp: event.timestamp
        }
      });
    }

    // 2. 写入搜索索引
    await this.searchEngine.index({
      id: `${event.sessionId}:${event.timestamp}`,
      content: event.content,
      sessionId: event.sessionId,
      type: event.type
    });
  }
}

from dataclasses import dataclass
from typing import Any, Callable
import asyncio
import json
import asyncpg

# === 命令定义 ===
@dataclass
class Command:
    """所有命令的基类"""
    pass

@dataclass
class CreatePost(Command):
    title: str
    content: str
    author_id: str

@dataclass
class UpdatePost(Command):
    post_id: str
    title: str | None = None
    content: str | None = None

# === 查询定义 ===
@dataclass
class Query:
    """所有查询的基类"""
    pass

@dataclass
class GetPost(Query):
    post_id: str

@dataclass
class SearchPosts(Query):
    keyword: str
    limit: int = 10

# === 命令总线 ===
class CommandBus:
    def __init__(self):
        self._handlers: dict[type, Callable] = {}

    def register(self, cmd_type: type, handler: Callable):
        self._handlers[cmd_type] = handler

    async def dispatch(self, cmd: Command) -> Any:
        handler = self._handlers.get(type(cmd))
        if not handler:
            raise ValueError(f"No handler for {type(cmd).__name__}")
        return await handler(cmd)

# === 查询总线 ===
class QueryBus:
    def __init__(self):
        self._handlers: dict[type, Callable] = {}

    def register(self, query_type: type, handler: Callable):
        self._handlers[query_type] = handler

    async def dispatch(self, query: Query) -> Any:
        handler = self._handlers.get(type(query))
        if not handler:
            raise ValueError(f"No handler for {type(query).__name__}")
        return await handler(query)

# === 命令处理器 ===
class PostCommandHandler:
    def __init__(self, pg: asyncpg.Pool):
        self.pg = pg

    async def handle_create(self, cmd: CreatePost) -> str:
        post_id = str(__import__('uuid').uuid4())
        async with self.pg.acquire() as conn:
            async with conn.transaction():
                # 业务写入
                await conn.execute(
                    """INSERT INTO posts (id, title, content, author_id)
                       VALUES ($1, $2, $3, $4)""",
                    post_id, cmd.title, cmd.content, cmd.author_id
                )
                # Outbox 事件
                await conn.execute(
                    """INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
                       VALUES ($1, $2, $3, $4)""",
                    'post', post_id, 'PostCreated',
                    json.dumps({'id': post_id, 'title': cmd.title, 'content': cmd.content})
                )
        return post_id

    async def handle_update(self, cmd: UpdatePost) -> None:
        async with self.pg.acquire() as conn:
            async with conn.transaction():
                updates = []
                params = []
                if cmd.title is not None:
                    updates.append("title = $1")
                    params.append(cmd.title)
                if cmd.content is not None:
                    updates.append("content = $" + str(len(params) + 1))
                    params.append(cmd.content)

                params.append(cmd.post_id)
                await conn.execute(
                    f"UPDATE posts SET {', '.join(updates)} WHERE id = ${len(params)}",
                    *params
                )
                await conn.execute(
                    """INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
                       VALUES ($1, $2, $3, $4)""",
                    'post', cmd.post_id, 'PostUpdated',
                    json.dumps({'id': cmd.post_id, 'title': cmd.title, 'content': cmd.content})
                )

# === 查询处理器（读侧） ===
class PostQueryHandler:
    def __init__(self, pg: asyncpg.Pool, meili: Any):
        self.pg = pg
        self.meili = meili

    async def handle_get(self, query: GetPost) -> dict | None:
        """单条查询走写库（强一致）"""
        row = await self.pg.fetchrow(
            "SELECT * FROM posts WHERE id = $1", query.post_id
        )
        return dict(row) if row else None

    async def handle_search(self, query: SearchPosts) -> list[dict]:
        """搜索走读库（Meilisearch，最终一致）"""
        result = await self.meili.index('posts').search(
            query.keyword, {'limit': query.limit}
        )
        return result['hits']

# === 组装 ===
async def setup_cqrs(pg: asyncpg.Pool, meili: Any):
    cmd_bus = CommandBus()
    query_bus = QueryBus()

    cmd_handler = PostCommandHandler(pg)
    query_handler = PostQueryHandler(pg, meili)

    cmd_bus.register(CreatePost, cmd_handler.handle_create)
    cmd_bus.register(UpdatePost, cmd_handler.handle_update)
    query_bus.register(GetPost, query_handler.handle_get)
    query_bus.register(SearchPosts, query_handler.handle_search)

    return cmd_bus, query_bus