AgentServer/apps/server/src/services/event-store.ts

// 事件管道：分配 seq → 落库（瞬时事件除外）→ 内存缓冲 → 广播（docs/design/02-architecture.md §3.3）
import { EventEmitter } from 'node:events'
import { and, asc, eq, gt, sql } from 'drizzle-orm'
import { TRANSIENT_EVENT_TYPES, type AgentEvent, type AgentEventType, type EventPayloads } from '@agent-server/shared'
import type { DB } from '../db'
import { events, sessions } from '../db/schema'
import { nowIso } from '../lib/ids'

const BUFFER_SIZE = 2000

export class EventStore {
  private bus = new EventEmitter()
  /** 每个会话最近的事件（含瞬时事件），用于 SSE 断线续传 */
  private buffers = new Map<string, AgentEvent[]>()

  constructor(private db: DB) {
    this.bus.setMaxListeners(0)
  }

  append<T extends AgentEventType>(sessionId: string, runId: string | null, type: T, payload: EventPayloads[T]): AgentEvent {
    const ts = nowIso()
    const persist = !TRANSIENT_EVENT_TYPES.has(type)
    const event = this.db.transaction(tx => {
      const row = tx.update(sessions)
        .set({ lastSeq: sql`${sessions.lastSeq} + 1`, ...(persist ? { updatedAt: ts } : {}) })
        .where(eq(sessions.id, sessionId))
        .returning({ seq: sessions.lastSeq })
        .get()
      if (!row) throw new Error(`会话不存在：${sessionId}`)
      if (persist) tx.insert(events).values({ sessionId, seq: row.seq, runId, type, payload, ts }).run()
      return { seq: row.seq, type, sessionId, runId, ts, payload } as AgentEvent
    })

    const buf = this.buffers.get(sessionId) ?? []
    buf.push(event)
    if (buf.length > BUFFER_SIZE) buf.splice(0, buf.length - BUFFER_SIZE)
    this.buffers.set(sessionId, buf)

    this.bus.emit(sessionId, event)
    this.bus.emit('*', event)
    return event
  }

  /** 读取 seq > after 的事件：优先使用内存缓冲（包含瞬时事件），不足时回落到数据库 */
  list(sessionId: string, after = 0, limit = 5000): AgentEvent[] {
    const buf = this.buffers.get(sessionId)
    if (buf?.length && buf[0]!.seq <= after + 1) return buf.filter(e => e.seq > after).slice(0, limit)
    const rows = this.db.select().from(events)
      .where(and(eq(events.sessionId, sessionId), gt(events.seq, after)))
      .orderBy(asc(events.seq))
      .limit(limit)
      .all()
    const persisted = rows.map(r => ({ seq: r.seq, type: r.type, sessionId: r.sessionId, runId: r.runId, ts: r.ts, payload: r.payload }) as AgentEvent)
    if (!buf?.length) return persisted
    // 数据库部分 + 缓冲中更新的部分（缓冲区里才有瞬时事件）
    const last = persisted.at(-1)?.seq ?? after
    return [...persisted, ...buf.filter(e => e.seq > last)].slice(0, limit)
  }

  subscribe(sessionId: string | '*', fn: (e: AgentEvent) => void): () => void {
    this.bus.on(sessionId, fn)
    return () => this.bus.off(sessionId, fn)
  }

  dropBuffer(sessionId: string) {
    this.buffers.delete(sessionId)
  }
}