事件溯源与 CQRS 在 Go 中的实践：复杂业务系统的架构升级

package events

import (
	"encoding/json"
	"time"
)

// EventMeta 所有领域事件的公共元信息
type EventMeta struct {
	EventID       string    `json:"event_id"`
	AggregateID   string    `json:"aggregate_id"`
	AggregateType string    `json:"aggregate_type"`
	EventType     string    `json:"event_type"`
	Version       int       `json:"version"`
	OccurredAt    time.Time `json:"occurred_at"`
}

// OrderCreated 订单创建事件 V1
type OrderCreated struct {
	EventMeta
	UserID      string   `json:"user_id"`
	Items       []Item   `json:"items"`
	TotalAmount float64  `json:"total_amount"`
	Address     Address  `json:"address"`
}

type Item struct {
	ProductID string  `json:"product_id"`
	Quantity  int     `json:"quantity"`
	UnitPrice float64 `json:"unit_price"`
}

type Address struct {
	Province string `json:"province"`
	City     string `json:"city"`
	Detail   string `json:"detail"`
}

// OrderPaid 订单支付完成事件
type OrderPaid struct {
	EventMeta
	PaymentID   string    `json:"payment_id"`
	PaidAmount  float64   `json:"paid_amount"`
	PaidAt      time.Time `json:"paid_at"`
}

// EventUpcaster 事件向上转换器接口
type EventUpcaster interface {
	// SupportedEventType 返回支持处理的事件类型名
	SupportedEventType() string
	// CanUpcast 判断给定版本是否能转换
	CanUpcast(fromVersion int) bool
	// Upcast 将旧版本 JSON 转换为当前版本 JSON
	Upcast(fromVersion int, data []byte) ([]byte, error)
}

// UpcasterRegistry 转换器注册表
type UpcasterRegistry struct {
	upcasters []EventUpcaster
}

func (r *UpcasterRegistry) Register(u EventUpcaster) {
	r.upcasters = append(r.upcasters, u)
}

func (r *UpcasterRegistry) Upcast(eventType string, version int, data []byte) ([]byte, error) {
	for _, u := range r.upcasters {
		if u.SupportedEventType() == eventType && u.CanUpcast(version) {
			return u.Upcast(version, data)
		}
	}
	return data, nil
}

CREATE TABLE event_store (
    id              BIGSERIAL PRIMARY KEY,
    aggregate_id    VARCHAR(255) NOT NULL,
    aggregate_type  VARCHAR(255) NOT NULL,
    event_type      VARCHAR(255) NOT NULL,
    event_version   INT NOT NULL,
    event_data      JSONB NOT NULL,
    metadata        JSONB,
    sequence        BIGSERIAL NOT NULL,
    occurred_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE (aggregate_id, event_version)
);

CREATE INDEX idx_event_store_aggregate ON event_store(aggregate_id, event_version);
CREATE INDEX idx_event_store_sequence ON event_store(sequence);

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/EventStore/EventStore-Client-Go/v3/esdb"
)

func main() {
	// 创建 EventStoreDB 连接
	settings, err := esdb.ParseConnectionString("esdb://admin:changeit@localhost:2113?tls=false")
	if err != nil {
		log.Fatal(err)
	}

	client, err := esdb.NewClient(settings)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	// 读取某个聚合根的事件流
	streamID := "order-12345"
	readResult, err := client.ReadStream(
		context.Background(),
		streamID,
		esdb.ReadStreamOptions{Direction: esdb.Forwards, From: esdb.Start{}},
		100,
	)
	if err != nil {
		log.Fatal(err)
	}
	defer readResult.Close()

	for {
		event, err := readResult.Recv()
		if err != nil {
			break
		}
		fmt.Printf("Event: %s, Data: %s\n", event.Event.EventType, string(event.Event.Data))
	}
}

package domain

import (
	"context"
	"fmt"
)

// AggregateRoot 是事件溯源聚合根的接口
type AggregateRoot interface {
	// AggregateID 返回聚合根的唯一标识
	AggregateID() string
	// AggregateType 返回聚合根类型名
	AggregateType() string
	// Version 返回当前版本号（未提交事件的基准版本）
	Version() int
	// SetVersion 设置版本号（由仓库调用）
	SetVersion(int)
	// ApplyEvent 将事件应用到聚合根状态（仅状态变更，无业务校验）
	ApplyEvent(event DomainEvent) error
	// UncommittedEvents 返回待提交的事件列表
	UncommittedEvents() []DomainEvent
	// ClearUncommittedEvents 清空待提交事件（保存成功后调用）
	ClearUncommittedEvents()
}

// DomainEvent 领域事件接口
type DomainEvent interface {
	EventMeta() EventMeta
}

// EventMeta 领域事件元信息
type EventMeta struct {
	EventID       string
	AggregateID   string
	AggregateType string
	EventType     string
	Version       int
	OccurredAt    int64 // Unix timestamp ms
}

package infra

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"

	"github.com/yourcompany/es-order/domain"
)

// EventStore 基于 PostgreSQL 的事件存储实现
type EventStore struct {
	db *sql.DB
}

func NewEventStore(db *sql.DB) *EventStore {
	return &EventStore{db: db}
}

// Save 保存聚合根的未提交事件，使用乐观并发控制
func (s *EventStore) Save(ctx context.Context, aggregate domain.AggregateRoot) error {
	uncommitted := aggregate.UncommittedEvents()
	if len(uncommitted) == 0 {
		return nil
	}

	currentVersion := aggregate.Version()
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelSerializable})
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	// 乐观并发检查：确认聚合根当前版本与数据库一致
	var latestVersion int
	row := tx.QueryRowContext(ctx,
		`SELECT COALESCE(MAX(event_version), 0) FROM event_store WHERE aggregate_id = $1`,
		aggregate.AggregateID(),
	)
	if err := row.Scan(&latestVersion); err != nil {
		return fmt.Errorf("check version: %w", err)
	}

	if latestVersion != currentVersion {
		return fmt.Errorf("concurrency conflict: expected version %d, got %d",
			currentVersion, latestVersion)
	}

	// 写入新事件
	for i, evt := range uncommitted {
		meta := evt.EventMeta()
		meta.Version = currentVersion + i + 1
		meta.OccurredAt = time.Now().UnixMilli()

		data, err := json.Marshal(evt)
		if err != nil {
			return fmt.Errorf("marshal event: %w", err)
		}

		_, err = tx.ExecContext(ctx, `
			INSERT INTO event_store (aggregate_id, aggregate_type, event_type, event_version, event_data, occurred_at)
			VALUES ($1, $2, $3, $4, $5, to_timestamp($6::bigint / 1000.0))
		`, aggregate.AggregateID(), aggregate.AggregateType(), meta.EventType, meta.Version, data, meta.OccurredAt)
		if err != nil {
			return fmt.Errorf("insert event: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit tx: %w", err)
	}

	// 更新内存中的版本号并清空未提交事件
	aggregate.SetVersion(currentVersion + len(uncommitted))
	aggregate.ClearUncommittedEvents()
	return nil
}

// Load 加载聚合根：读取历史事件并重放
func (s *EventStore) Load(ctx context.Context, aggregate domain.AggregateRoot, aggregateID string) error {
	rows, err := s.db.QueryContext(ctx, `
		SELECT event_data FROM event_store
		WHERE aggregate_id = $1
		ORDER BY event_version ASC
	`, aggregateID)
	if err != nil {
		return fmt.Errorf("query events: %w", err)
	}
	defer rows.Close()

	version := 0
	for rows.Next() {
		var raw []byte
		if err := rows.Scan(&raw); err != nil {
			return fmt.Errorf("scan event: %w", err)
		}

		evt, err := s.deserialize(raw)
		if err != nil {
			return fmt.Errorf("deserialize: %w", err)
		}

		if err := aggregate.ApplyEvent(evt); err != nil {
			return fmt.Errorf("apply event: %w", err)
		}
		version++
	}

	aggregate.SetVersion(version)
	return rows.Err()
}

// deserialize 将 JSON 反序列化为具体的领域事件（此处简化，实际可用工厂模式或反射注册）
func (s *EventStore) deserialize(data []byte) (domain.DomainEvent, error) {
	// 先解析 event_type 字段，再做二次反序列化
	var wrapper struct {
		EventType string `json:"event_type"`
	}
	if err := json.Unmarshal(data, &wrapper); err != nil {
		return nil, err
	}

	// 实际项目中使用注册表映射 event_type -> 具体类型
	// 此处为示例简化
	switch wrapper.EventType {
	case "OrderCreated":
		var evt domain.OrderCreated
		if err := json.Unmarshal(data, &evt); err != nil {
			return nil, err
		}
		return &evt, nil
	// ... 其他事件类型
	default:
		return nil, fmt.Errorf("unknown event type: %s", wrapper.EventType)
	}
}

package serialization

import (
	"encoding/json"
	"fmt"
	"reflect"
)

// Registry 事件类型注册表
type Registry struct {
	typeMap map[string]reflect.Type
}

func NewRegistry() *Registry {
	return &Registry{typeMap: make(map[string]reflect.Type)}
}

func (r *Registry) Register(eventType string, prototype interface{}) {
	r.typeMap[eventType] = reflect.TypeOf(prototype).Elem()
}

func (r *Registry) Deserialize(eventType string, data []byte) (interface{}, error) {
	t, ok := r.typeMap[eventType]
	if !ok {
		return nil, fmt.Errorf("event type %s not registered", eventType)
	}
	ev := reflect.New(t).Interface()
	if err := json.Unmarshal(data, ev); err != nil {
		return nil, err
	}
	return ev, nil
}

package projection

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
)

// OrderSummary 读模型：订单摘要
type OrderSummary struct {
	OrderID       string  `json:"order_id"`
	UserID        string  `json:"user_id"`
	Status        string  `json:"status"`
	TotalAmount   float64 `json:"total_amount"`
	ItemCount     int     `json:"item_count"`
	CreatedAt     string  `json:"created_at"`
	LastUpdatedAt string  `json:"last_updated_at"`
}

// OrderProjector 订单投影器
type OrderProjector struct {
	readDB *sql.DB
}

func NewOrderProjector(readDB *sql.DB) *OrderProjector {
	return &OrderProjector{readDB: readDB}
}

// HandleEvent 处理单个领域事件以更新读模型
func (p *OrderProjector) HandleEvent(ctx context.Context, eventType string, data []byte) error {
	switch eventType {
	case "OrderCreated":
		var evt struct {
			AggregateID string  `json:"aggregate_id"`
			UserID      string  `json:"user_id"`
			Items       []struct {
				Quantity int `json:"quantity"`
			} `json:"items"`
			TotalAmount float64 `json:"total_amount"`
			OccurredAt  int64   `json:"occurred_at"`
		}
		if err := json.Unmarshal(data, &evt); err != nil {
			return err
		}

		itemCount := 0
		for _, item := range evt.Items {
			itemCount += item.Quantity
		}

		_, err := p.readDB.ExecContext(ctx, `
			INSERT INTO order_summary (order_id, user_id, status, total_amount, item_count, created_at, last_updated_at)
			VALUES ($1, $2, 'created', $3, $4, to_timestamp($5::bigint/1000.0), to_timestamp($5::bigint/1000.0))
		`, evt.AggregateID, evt.UserID, evt.TotalAmount, itemCount, evt.OccurredAt)
		return err

	case "OrderPaid":
		var evt struct {
			AggregateID string `json:"aggregate_id"`
			OccurredAt  int64  `json:"occurred_at"`
		}
		if err := json.Unmarshal(data, &evt); err != nil {
			return err
		}
		_, err := p.readDB.ExecContext(ctx, `
			UPDATE order_summary
			SET status = 'paid', last_updated_at = to_timestamp($2::bigint/1000.0)
			WHERE order_id = $1
		`, evt.AggregateID, evt.OccurredAt)
		return err

	case "OrderShipped":
		var evt struct {
			AggregateID string `json:"aggregate_id"`
			OccurredAt  int64  `json:"occurred_at"`
		}
		if err := json.Unmarshal(data, &evt); err != nil {
			return err
		}
		_, err := p.readDB.ExecContext(ctx, `
			UPDATE order_summary
			SET status = 'shipped', last_updated_at = to_timestamp($2::bigint/1000.0)
			WHERE order_id = $1
		`, evt.AggregateID, evt.OccurredAt)
		return err

	default:
		return nil
	}
}

// ReplayAll 从事件存储中从头重放所有事件以重建读模型
func (p *OrderProjector) ReplayAll(ctx context.Context, eventStore *infra.EventStore) error {
	// 先清空读模型
	if _, err := p.readDB.ExecContext(ctx, `TRUNCATE TABLE order_summary`); err != nil {
		return fmt.Errorf("truncate: %w", err)
	}

	// 分页读取所有事件并重放
	batchSize := 1000
	lastSequence := 0

	for {
		rows, err := eventStore.QueryBySequence(ctx, lastSequence, batchSize)
		if err != nil {
			return fmt.Errorf("query events: %w", err)
		}

		count := 0
		for rows.Next() {
			var eventType string
			var data []byte
			var sequence int
			if err := rows.Scan(&sequence, &eventType, &data); err != nil {
				rows.Close()
				return err
			}
			if err := p.HandleEvent(ctx, eventType, data); err != nil {
				rows.Close()
				return fmt.Errorf("handle event seq %d: %w", sequence, err)
			}
			lastSequence = sequence
			count++
		}
		rows.Close()

		if count < batchSize {
			break
		}
	}
	return nil
}

package app

import (
	"context"
	"fmt"
	"time"

	"github.com/yourcompany/es-order/domain"
	"github.com/yourcompany/es-order/infra"
)

// CreateOrderCommand 创建订单命令
type CreateOrderCommand struct {
	OrderID     string
	UserID      string
	Items       []domain.Item
	TotalAmount float64
	Address     domain.Address
}

// CommandHandler 命令处理器
type CommandHandler struct {
	eventStore *infra.EventStore
	eventBus   infra.EventBus
}

func NewCommandHandler(es *infra.EventStore, bus infra.EventBus) *CommandHandler {
	return &CommandHandler{eventStore: es, eventBus: bus}
}

// HandleCreateOrder 处理创建订单命令
func (h *CommandHandler) HandleCreateOrder(ctx context.Context, cmd CreateOrderCommand) error {
	order := domain.NewOrder()

	err := order.Create(cmd.OrderID, cmd.UserID, cmd.Items, cmd.TotalAmount, cmd.Address)
	if err != nil {
		return fmt.Errorf("domain error: %w", err)
	}

	// 保存事件到事件存储
	if err := h.eventStore.Save(ctx, order); err != nil {
		return fmt.Errorf("save events: %w", err)
	}

	// 发布事件到总线
	for _, evt := range order.RecentEvents() {
		if err := h.eventBus.Publish(ctx, evt); err != nil {
			// 记录日志，发送失败不影响已持久化的一致性
			// 生产环境可使用发件箱模式保证最终一致性
			return fmt.Errorf("publish event: %w", err)
		}
	}

	return nil
}

package query

import (
	"context"
	"database/sql"
	"fmt"
)

// OrderQueryService 订单查询服务
type OrderQueryService struct {
	readDB *sql.DB
}

func NewOrderQueryService(readDB *sql.DB) *OrderQueryService {
	return &OrderQueryService{readDB: readDB}
}

// OrderDTO 订单数据传输对象
type OrderDTO struct {
	OrderID       string  `json:"order_id"`
	UserID        string  `json:"user_id"`
	Status        string  `json:"status"`
	TotalAmount   float64 `json:"total_amount"`
	ItemCount     int     `json:"item_count"`
	CreatedAt     string  `json:"created_at"`
}

// GetOrderByID 根据订单 ID 查询
func (s *OrderQueryService) GetOrderByID(ctx context.Context, orderID string) (*OrderDTO, error) {
	row := s.readDB.QueryRowContext(ctx, `
		SELECT order_id, user_id, status, total_amount, item_count, created_at
		FROM order_summary
		WHERE order_id = $1
	`, orderID)

	var dto OrderDTO
	if err := row.Scan(&dto.OrderID, &dto.UserID, &dto.Status, &dto.TotalAmount, &dto.ItemCount, &dto.CreatedAt); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	return &dto, nil
}

// ListOrdersByUser 查询用户的订单列表，支持分页
func (s *OrderQueryService) ListOrdersByUser(ctx context.Context, userID string, page, pageSize int) ([]OrderDTO, int, error) {
	offset := (page - 1) * pageSize

	// 查询总数
	var total int
	if err := s.readDB.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM order_summary WHERE user_id = $1`, userID,
	).Scan(&total); err != nil {
		return nil, 0, err
	}

	// 查询分页数据
	rows, err := s.readDB.QueryContext(ctx, `
		SELECT order_id, user_id, status, total_amount, item_count, created_at
		FROM order_summary
		WHERE user_id = $1
		ORDER BY created_at DESC
		LIMIT $2 OFFSET $3
	`, userID, pageSize, offset)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()

	var orders []OrderDTO
	for rows.Next() {
		var dto OrderDTO
		if err := rows.Scan(&dto.OrderID, &dto.UserID, &dto.Status, &dto.TotalAmount, &dto.ItemCount, &dto.CreatedAt); err != nil {
			return nil, 0, err
		}
		orders = append(orders, dto)
	}

	return orders, total, rows.Err()
}

package api

import (
	"encoding/json"
	"net/http"
	"strings"

	"github.com/yourcompany/es-order/app"
	"github.com/yourcompany/es-order/query"
)

// CommandServer 命令端 HTTP 服务
type CommandServer struct {
	cmdHandler *app.CommandHandler
}

func (s *CommandServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method == http.MethodPost && strings.HasPrefix(r.URL.Path, "/orders") {
		var req struct {
			OrderID string `json:"order_id"`
			UserID  string `json:"user_id"`
		}
		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		// 实际调用命令处理器...
		w.WriteHeader(http.StatusAccepted)
		return
	}
	http.NotFound(w, r)
}

// QueryServer 查询端 HTTP 服务
type QueryServer struct {
	queryService *query.OrderQueryService
}

func (s *QueryServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/orders/") {
		orderID := strings.TrimPrefix(r.URL.Path, "/orders/")
		dto, err := s.queryService.GetOrderByID(r.Context(), orderID)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		if dto == nil {
			http.NotFound(w, r)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(dto)
		return
	}
	http.NotFound(w, r)
}

package saga

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// SagaStep Saga 中的单步定义
type SagaStep struct {
	Name        string
	Action      func(ctx context.Context) error
	Compensate  func(ctx context.Context) error
}

// Saga 编排器
type Saga struct {
	name  string
	steps []SagaStep
}

func NewSaga(name string) *Saga {
	return &Saga{name: name}
}

func (s *Saga) AddStep(step SagaStep) {
	s.steps = append(s.steps, step)
}

// Execute 按顺序执行 Saga，任一失败时触发补偿
func (s *Saga) Execute(ctx context.Context) error {
	completed := []int{}

	for i, step := range s.steps {
		if err := step.Action(ctx); err != nil {
			// 触发已执行步骤的补偿
			for j := len(completed) - 1; j >= 0; j-- {
				compStep := s.steps[completed[j]]
				if compStep.Compensate != nil {
					// 补偿失败通常需要人工介入，记录日志即可
					if compErr := compStep.Compensate(ctx); compErr != nil {
						fmt.Printf("compensation failed for step %s: %v\n", compStep.Name, compErr)
					}
				}
			}
			return fmt.Errorf("saga failed at step %s: %w", step.Name, err)
		}
		completed = append(completed, i)
	}

	return nil
}

// OrderCreationSaga 订单创建 Saga：创建订单 -> 扣库存 -> 创建支付单
func BuildOrderCreationSaga(
	orderSvc OrderService,
	inventorySvc InventoryService,
	paymentSvc PaymentService,
	orderID string,
	items []OrderItem,
) *Saga {
	saga := NewSaga("CreateOrder")

	var deductedInventory map[string]int

	saga.AddStep(SagaStep{
		Name: "CreateOrder",
		Action: func(ctx context.Context) error {
			return orderSvc.CreateDraftOrder(ctx, orderID, items)
		},
		Compensate: func(ctx context.Context) error {
			return orderSvc.CancelOrder(ctx, orderID)
		},
	})

	saga.AddStep(SagaStep{
		Name: "DeductInventory",
		Action: func(ctx context.Context) error {
			result, err := inventorySvc.Deduct(ctx, items)
			if err == nil {
				deductedInventory = result
			}
			return err
		},
		Compensate: func(ctx context.Context) error {
			if deductedInventory != nil {
				return inventorySvc.Restore(ctx, deductedInventory)
			}
			return nil
		},
	})

	saga.AddStep(SagaStep{
		Name: "CreatePayment",
		Action: func(ctx context.Context) error {
			return paymentSvc.CreatePaymentOrder(ctx, orderID, calculateTotal(items))
		},
		Compensate: func(ctx context.Context) error {
			return paymentSvc.CancelPaymentOrder(ctx, orderID)
		},
	})

	return saga
}

type OrderService interface {
	CreateDraftOrder(ctx context.Context, orderID string, items []OrderItem) error
	CancelOrder(ctx context.Context, orderID string) error
}

type InventoryService interface {
	Deduct(ctx context.Context, items []OrderItem) (map[string]int, error)
	Restore(ctx context.Context, items map[string]int) error
}

type PaymentService interface {
	CreatePaymentOrder(ctx context.Context, orderID string, amount float64) error
	CancelPaymentOrder(ctx context.Context, orderID string) error
}

type OrderItem struct {
	ProductID string
	Quantity  int
	UnitPrice float64
}

func calculateTotal(items []OrderItem) float64 {
	total := 0.0
	for _, item := range items {
		total += item.UnitPrice * float64(item.Quantity)
	}
	return total
}

package infra

import (
	"context"
	"fmt"
	"sync"
)

// EventBus 事件总线接口
type EventBus interface {
	Publish(ctx context.Context, event interface{}) error
	Subscribe(eventType string, handler EventHandler)
}

// EventHandler 事件处理器函数类型
type EventHandler func(ctx context.Context, eventType string, data []byte) error

// InMemoryEventBus 内存事件总线实现
type InMemoryEventBus struct {
	mu        sync.RWMutex
	handlers  map[string][]EventHandler
	ser       EventSerializer
}

func NewInMemoryEventBus(ser EventSerializer) *InMemoryEventBus {
	return &InMemoryEventBus{
		handlers: make(map[string][]EventHandler),
		ser:      ser,
	}
}

func (b *InMemoryEventBus) Publish(ctx context.Context, event interface{}) error {
	b.mu.RLock()
	defer b.mu.RUnlock()

	eventType, data, err := b.ser.Serialize(event)
	if err != nil {
		return fmt.Errorf("serialize: %w", err)
	}

	// 异步调用处理器，避免阻塞发布方
	for _, h := range b.handlers[eventType] {
		go func(handler EventHandler) {
			if err := handler(ctx, eventType, data); err != nil {
				// 记录错误日志，生产环境应有重试和死信队列机制
				fmt.Printf("handler error: %v\n", err)
			}
		}(h)
	}
	return nil
}

func (b *InMemoryEventBus) Subscribe(eventType string, handler EventHandler) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.handlers[eventType] = append(b.handlers[eventType], handler)
}

// EventSerializer 事件序列化接口
type EventSerializer interface {
	Serialize(event interface{}) (eventType string, data []byte, err error)
}