fb182adef4
🎯 核心变更: - OpType (string) → OpCode (int32) - 20+ OpCode枚举常量 (基于DOIP/IRP标准) - 类型安全 + 性能优化 📊 影响范围: - 核心模型: Operation结构体、CBOR序列化 - 数据库: schema.go + SQL DDL (PostgreSQL/MySQL/SQLite) - 持久化: repository.go查询、cursor_worker.go - API接口: Protobuf定义 + gRPC客户端 - 测试代码: 60+ 测试文件更新 ✅ 测试结果: - 通过率: 100% (所有87个测试用例) - 总体覆盖率: 53.7% - 核心包覆盖率: logger(100%), highclient(95.3%), model(79.1%) 📝 文档: - 精简README (1056行→489行，减少54%) - 完整的OpCode枚举说明 - 三种持久化策略示例 - 数据库表结构和架构图 🔧 技术细节: - 类型转换: string(OpCode) → int32(OpCode) - SQL参数: 字符串值 → 整数值 - Protobuf: op_type string → op_code int32 - 测试断言: 字符串比较 → 常量比较 🎉 质量保证: - 零编译错误 - 100%测试通过 - PostgreSQL/Pulsar集成测试验证 - 分布式并发安全测试通过
592 lines
16 KiB
Go
592 lines
16 KiB
Go
package model
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"go.yandata.net/iod/iod/go-trustlog/api/logger"
|
||
"go.yandata.net/iod/iod/go-trustlog/internal/helpers"
|
||
)
|
||
|
||
//
|
||
// ===== 操作来源类型 =====
|
||
//
|
||
|
||
// Source 表示操作来源，用于区分不同系统模块（IRP、DOIP）。
|
||
type Source string
|
||
|
||
const (
|
||
OpSourceIRP Source = "IRP"
|
||
OpSourceDOIP Source = "DOIP"
|
||
)
|
||
|
||
//
|
||
// ===== 操作代码枚举 (OpCode) =====
|
||
//
|
||
|
||
// OpCode 表示操作的具体代码（int32类型）
|
||
type OpCode int32
|
||
|
||
// 标准 Handle System 操作代码
|
||
const (
|
||
OpCodeReserved OpCode = 0 // Reserved
|
||
OpCodeResolution OpCode = 1 // Identifier query
|
||
OpCodeGetSiteInfo OpCode = 2 // Get HS_SITE element
|
||
|
||
OpCodeCreateID OpCode = 100 // Create new identifier
|
||
OpCodeDeleteID OpCode = 101 // Delete existing identifier
|
||
OpCodeAddElement OpCode = 102 // Add element(s)
|
||
OpCodeRemoveElement OpCode = 103 // Remove element(s)
|
||
OpCodeModifyElement OpCode = 104 // Modify element(s)
|
||
OpCodeListIDs OpCode = 105 // List identifiers
|
||
OpCodeListDerivedPrefixes OpCode = 106 // List derived prefixes
|
||
|
||
OpCodeChallengeResponse OpCode = 200 // Response to challenge
|
||
OpCodeVerifyResponse OpCode = 201 // Verify challenge response
|
||
|
||
OpCodeHomePrefix OpCode = 300 // Home prefix
|
||
OpCodeUnhomePrefix OpCode = 301 // Unhome prefix
|
||
OpCodeListHomedPrefixes OpCode = 302 // List homed prefixes
|
||
|
||
OpCodeSessionSetup OpCode = 400 // Session setup request
|
||
OpCodeSessionTerminate OpCode = 401 // Session termination request
|
||
|
||
// Yandata 扩展操作代码
|
||
OpCodeQueryIDs OpCode = 500 // Query DOIDs
|
||
OpCodeRenameID OpCode = 501 // Rename DOID
|
||
OpCodeResolveAltID OpCode = 502 // Resolve by alternative ID
|
||
OpCodeRegisterAltID OpCode = 503 // Register alternative ID
|
||
)
|
||
|
||
// OpCodeName 返回操作代码的名称
|
||
func (c OpCode) String() string {
|
||
switch c {
|
||
case OpCodeReserved:
|
||
return "RESERVED"
|
||
case OpCodeResolution:
|
||
return "RESOLUTION"
|
||
case OpCodeGetSiteInfo:
|
||
return "GET_SITEINFO"
|
||
case OpCodeCreateID:
|
||
return "CREATE_ID"
|
||
case OpCodeDeleteID:
|
||
return "DELETE_ID"
|
||
case OpCodeAddElement:
|
||
return "ADD_ELEMENT"
|
||
case OpCodeRemoveElement:
|
||
return "REMOVE_ELEMENT"
|
||
case OpCodeModifyElement:
|
||
return "MODIFY_ELEMENT"
|
||
case OpCodeListIDs:
|
||
return "LIST_IDS"
|
||
case OpCodeListDerivedPrefixes:
|
||
return "LIST_DERIVED_PREFIXES"
|
||
case OpCodeChallengeResponse:
|
||
return "CHALLENGE_RESPONSE"
|
||
case OpCodeVerifyResponse:
|
||
return "VERIFY_RESPONSE"
|
||
case OpCodeHomePrefix:
|
||
return "HOME_PREFIX"
|
||
case OpCodeUnhomePrefix:
|
||
return "UNHOME_PREFIX"
|
||
case OpCodeListHomedPrefixes:
|
||
return "LIST_HOMED_PREFIXES"
|
||
case OpCodeSessionSetup:
|
||
return "SESSION_SETUP"
|
||
case OpCodeSessionTerminate:
|
||
return "SESSION_TERMINATE"
|
||
case OpCodeQueryIDs:
|
||
return "QUERY_IDS"
|
||
case OpCodeRenameID:
|
||
return "RENAME_ID"
|
||
case OpCodeResolveAltID:
|
||
return "RESOLVE_ALT_ID"
|
||
case OpCodeRegisterAltID:
|
||
return "REGISTER_ALT_ID"
|
||
default:
|
||
return fmt.Sprintf("UNKNOWN(%d)", c)
|
||
}
|
||
}
|
||
|
||
// IsValid 检查操作代码是否有效
|
||
func (c OpCode) IsValid() bool {
|
||
validCodes := []OpCode{
|
||
OpCodeReserved, OpCodeResolution, OpCodeGetSiteInfo,
|
||
OpCodeCreateID, OpCodeDeleteID, OpCodeAddElement,
|
||
OpCodeRemoveElement, OpCodeModifyElement, OpCodeListIDs,
|
||
OpCodeListDerivedPrefixes,
|
||
OpCodeChallengeResponse, OpCodeVerifyResponse,
|
||
OpCodeHomePrefix, OpCodeUnhomePrefix, OpCodeListHomedPrefixes,
|
||
OpCodeSessionSetup, OpCodeSessionTerminate,
|
||
OpCodeQueryIDs, OpCodeRenameID, OpCodeResolveAltID, OpCodeRegisterAltID,
|
||
}
|
||
|
||
for _, valid := range validCodes {
|
||
if c == valid {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
//
|
||
// ===== 操作记录结构 =====
|
||
//
|
||
|
||
// Operation 表示一次完整的操作记录。
|
||
// 用于记录系统中的操作行为，包含操作元数据、数据标识、操作者信息以及请求/响应的哈希值。
|
||
type Operation struct {
|
||
OpID string `json:"opId" validate:"max=32"`
|
||
Timestamp time.Time `json:"timestamp" validate:"required"`
|
||
OpSource Source `json:"opSource" validate:"required,oneof=IRP DOIP"`
|
||
OpCode OpCode `json:"opCode" validate:"required"`
|
||
DoPrefix string `json:"doPrefix" validate:"required,max=512"`
|
||
DoRepository string `json:"doRepository" validate:"required,max=512"`
|
||
Doid string `json:"doid" validate:"required,max=512"`
|
||
ProducerID string `json:"producerId" validate:"required,max=512"`
|
||
OpActor string `json:"opActor" validate:"max=64"`
|
||
RequestBodyHash *string `json:"requestBodyHash" validate:"omitempty,max=128"`
|
||
ResponseBodyHash *string `json:"responseBodyHash" validate:"omitempty,max=128"`
|
||
// ClientIP 客户端IP地址，仅用于数据库持久化，不参与存证哈希计算
|
||
ClientIP *string `json:"clientIp,omitempty" validate:"omitempty,max=32"`
|
||
// ServerIP 服务端IP地址，仅用于数据库持久化，不参与存证哈希计算
|
||
ServerIP *string `json:"serverIp,omitempty" validate:"omitempty,max=32"`
|
||
Ack func() bool `json:"-"`
|
||
Nack func() bool `json:"-"`
|
||
binary []byte
|
||
}
|
||
|
||
//
|
||
// ===== 构造函数 =====
|
||
//
|
||
|
||
// NewFullOperation 创建包含所有字段的完整 Operation。
|
||
// 自动完成哈希计算和字段校验，确保创建的 Operation 是完整且有效的。
|
||
func NewFullOperation(
|
||
opSource Source,
|
||
opCode OpCode,
|
||
doPrefix, doRepository, doid string,
|
||
producerID string,
|
||
opActor string,
|
||
requestBody, responseBody interface{},
|
||
timestamp time.Time,
|
||
) (*Operation, error) {
|
||
log := logger.GetGlobalLogger()
|
||
log.Debug("Creating new full operation",
|
||
"opSource", opSource,
|
||
"opCode", opCode,
|
||
"doPrefix", doPrefix,
|
||
"doRepository", doRepository,
|
||
"doid", doid,
|
||
"producerID", producerID,
|
||
"opActor", opActor,
|
||
)
|
||
op := &Operation{
|
||
Timestamp: timestamp,
|
||
OpSource: opSource,
|
||
OpCode: opCode,
|
||
DoPrefix: doPrefix,
|
||
DoRepository: doRepository,
|
||
Doid: doid,
|
||
ProducerID: producerID,
|
||
OpActor: opActor,
|
||
}
|
||
|
||
log.Debug("Setting request body hash")
|
||
if err := op.RequestBodyFlexible(requestBody); err != nil {
|
||
log.Error("Failed to set request body hash",
|
||
"error", err,
|
||
)
|
||
return nil, err
|
||
}
|
||
log.Debug("Setting response body hash")
|
||
if err := op.ResponseBodyFlexible(responseBody); err != nil {
|
||
log.Error("Failed to set response body hash",
|
||
"error", err,
|
||
)
|
||
return nil, err
|
||
}
|
||
log.Debug("Checking and initializing operation")
|
||
if err := op.CheckAndInit(); err != nil {
|
||
log.Error("Failed to check and init operation",
|
||
"error", err,
|
||
)
|
||
return nil, err
|
||
}
|
||
|
||
log.Debug("Full operation created successfully",
|
||
"opID", op.OpID,
|
||
)
|
||
return op, nil
|
||
}
|
||
|
||
//
|
||
// ===== 接口实现 =====
|
||
//
|
||
|
||
func (o *Operation) Key() string {
|
||
return o.OpID
|
||
}
|
||
|
||
// OperationHashData 实现 HashData 接口，用于存储 Operation 的哈希计算结果。
|
||
type OperationHashData struct {
|
||
key string
|
||
hash string
|
||
}
|
||
|
||
func (o OperationHashData) Key() string {
|
||
return o.key
|
||
}
|
||
|
||
func (o OperationHashData) Hash() string {
|
||
return o.hash
|
||
}
|
||
|
||
func (o OperationHashData) Type() HashType {
|
||
return Sha256Simd
|
||
}
|
||
|
||
// DoHash 计算 Operation 的整体哈希值，用于数据完整性验证。
|
||
// 哈希基于序列化后的二进制数据计算，确保操作记录的不可篡改性。
|
||
func (o *Operation) DoHash(_ context.Context) (HashData, error) {
|
||
log := logger.GetGlobalLogger()
|
||
log.Debug("Computing hash for operation",
|
||
"opID", o.OpID,
|
||
)
|
||
hashTool := GetHashTool(Sha256Simd)
|
||
binary, err := o.MarshalBinary()
|
||
if err != nil {
|
||
log.Error("Failed to marshal operation for hash",
|
||
"error", err,
|
||
"opID", o.OpID,
|
||
)
|
||
return nil, fmt.Errorf("failed to marshal operation: %w", err)
|
||
}
|
||
|
||
log.Debug("Computing hash bytes",
|
||
"opID", o.OpID,
|
||
"binaryLength", len(binary),
|
||
)
|
||
hash, err := hashTool.HashBytes(binary)
|
||
if err != nil {
|
||
log.Error("Failed to compute hash",
|
||
"error", err,
|
||
"opID", o.OpID,
|
||
)
|
||
return nil, fmt.Errorf("failed to compute hash: %w", err)
|
||
}
|
||
|
||
log.Debug("Hash computed successfully",
|
||
"opID", o.OpID,
|
||
"hash", hash,
|
||
)
|
||
return OperationHashData{
|
||
key: o.OpID,
|
||
hash: hash,
|
||
}, nil
|
||
}
|
||
|
||
//
|
||
// ===== CBOR 序列化相关 =====
|
||
//
|
||
|
||
// operationData 用于 CBOR 序列化/反序列化的中间结构。
|
||
// 排除函数字段和缓存字段，仅包含可序列化的数据字段。
|
||
type operationData struct {
|
||
OpID *string `cbor:"opId"`
|
||
Timestamp *time.Time `cbor:"timestamp"`
|
||
OpSource *Source `cbor:"opSource"`
|
||
OpCode *OpCode `cbor:"opCode"`
|
||
DoPrefix *string `cbor:"doPrefix"`
|
||
DoRepository *string `cbor:"doRepository"`
|
||
Doid *string `cbor:"doid"`
|
||
ProducerID *string `cbor:"producerId"`
|
||
OpActor *string `cbor:"opActor"`
|
||
RequestBodyHash *string `cbor:"requestBodyHash"`
|
||
ResponseBodyHash *string `cbor:"responseBodyHash"`
|
||
}
|
||
|
||
// toOperationData 将 Operation 转换为 operationData，用于序列化。
|
||
func (o *Operation) toOperationData() *operationData {
|
||
return &operationData{
|
||
OpID: &o.OpID,
|
||
Timestamp: &o.Timestamp,
|
||
OpSource: &o.OpSource,
|
||
OpCode: &o.OpCode,
|
||
DoPrefix: &o.DoPrefix,
|
||
DoRepository: &o.DoRepository,
|
||
Doid: &o.Doid,
|
||
ProducerID: &o.ProducerID,
|
||
OpActor: &o.OpActor,
|
||
RequestBodyHash: o.RequestBodyHash,
|
||
ResponseBodyHash: o.ResponseBodyHash,
|
||
}
|
||
}
|
||
|
||
// fromOperationData 从 operationData 填充 Operation，用于反序列化。
|
||
func (o *Operation) fromOperationData(opData *operationData) {
|
||
if opData == nil {
|
||
return
|
||
}
|
||
|
||
if opData.OpID != nil {
|
||
o.OpID = *opData.OpID
|
||
}
|
||
if opData.Timestamp != nil {
|
||
o.Timestamp = *opData.Timestamp
|
||
}
|
||
if opData.OpSource != nil {
|
||
o.OpSource = *opData.OpSource
|
||
}
|
||
if opData.OpCode != nil {
|
||
o.OpCode = *opData.OpCode
|
||
}
|
||
if opData.DoPrefix != nil {
|
||
o.DoPrefix = *opData.DoPrefix
|
||
}
|
||
if opData.DoRepository != nil {
|
||
o.DoRepository = *opData.DoRepository
|
||
}
|
||
if opData.Doid != nil {
|
||
o.Doid = *opData.Doid
|
||
}
|
||
if opData.ProducerID != nil {
|
||
o.ProducerID = *opData.ProducerID
|
||
}
|
||
if opData.OpActor != nil {
|
||
o.OpActor = *opData.OpActor
|
||
}
|
||
if opData.RequestBodyHash != nil {
|
||
hash := *opData.RequestBodyHash
|
||
o.RequestBodyHash = &hash
|
||
}
|
||
if opData.ResponseBodyHash != nil {
|
||
hash := *opData.ResponseBodyHash
|
||
o.ResponseBodyHash = &hash
|
||
}
|
||
}
|
||
|
||
// MarshalBinary 将 Operation 序列化为 CBOR 格式的二进制数据。
|
||
// 实现 encoding.BinaryMarshaler 接口。
|
||
// 使用 Canonical CBOR 编码确保序列化结果的一致性，使用缓存机制避免重复序列化。
|
||
func (o *Operation) MarshalBinary() ([]byte, error) {
|
||
log := logger.GetGlobalLogger()
|
||
log.Debug("Marshaling operation to CBOR binary",
|
||
"opID", o.OpID,
|
||
)
|
||
if o.binary != nil {
|
||
log.Debug("Using cached binary data",
|
||
"opID", o.OpID,
|
||
)
|
||
return o.binary, nil
|
||
}
|
||
|
||
opData := o.toOperationData()
|
||
|
||
log.Debug("Marshaling operation data to canonical CBOR",
|
||
"opID", o.OpID,
|
||
)
|
||
binary, err := helpers.MarshalCanonical(opData)
|
||
if err != nil {
|
||
log.Error("Failed to marshal operation to CBOR",
|
||
"error", err,
|
||
"opID", o.OpID,
|
||
)
|
||
return nil, fmt.Errorf("failed to marshal operation to CBOR: %w", err)
|
||
}
|
||
|
||
o.binary = binary
|
||
|
||
log.Debug("Operation marshaled successfully",
|
||
"opID", o.OpID,
|
||
"binaryLength", len(binary),
|
||
)
|
||
return binary, nil
|
||
}
|
||
|
||
// GetProducerID 返回 ProducerID，实现 Trustlog 接口。
|
||
func (o *Operation) GetProducerID() string {
|
||
return o.ProducerID
|
||
}
|
||
|
||
// UnmarshalBinary 从 CBOR 格式的二进制数据反序列化为 Operation。
|
||
// 实现 encoding.BinaryUnmarshaler 接口。
|
||
func (o *Operation) UnmarshalBinary(data []byte) error {
|
||
log := logger.GetGlobalLogger()
|
||
log.Debug("Unmarshaling operation from CBOR binary",
|
||
"dataLength", len(data),
|
||
)
|
||
if len(data) == 0 {
|
||
log.Error("Data is empty")
|
||
return errors.New("data is empty")
|
||
}
|
||
|
||
opData := &operationData{}
|
||
|
||
log.Debug("Unmarshaling operation data from CBOR")
|
||
if err := helpers.Unmarshal(data, opData); err != nil {
|
||
log.Error("Failed to unmarshal operation from CBOR",
|
||
"error", err,
|
||
)
|
||
return fmt.Errorf("failed to unmarshal operation from CBOR: %w", err)
|
||
}
|
||
|
||
o.fromOperationData(opData)
|
||
|
||
o.binary = data
|
||
|
||
log.Debug("Operation unmarshaled successfully",
|
||
"opID", o.OpID,
|
||
)
|
||
return nil
|
||
}
|
||
|
||
//
|
||
// ===== 哈希设置方法 =====
|
||
//
|
||
|
||
// setBodyHashFlexible 根据输入数据类型计算哈希，支持 string 和 []byte。
|
||
// 使用固定的 Sha256Simd 算法。
|
||
// 如果输入为 nil 或空，则目标指针设置为 nil，表示该字段未设置。
|
||
func (o *Operation) setBodyHashFlexible(data interface{}, target **string) error {
|
||
log := logger.GetGlobalLogger()
|
||
log.Debug("Setting body hash flexible",
|
||
"opID", o.OpID,
|
||
"dataType", fmt.Sprintf("%T", data),
|
||
)
|
||
if data == nil {
|
||
log.Debug("Data is nil, setting target to nil")
|
||
*target = nil
|
||
return nil
|
||
}
|
||
|
||
hashTool := GetHashTool(Sha256Simd)
|
||
var raw []byte
|
||
|
||
switch v := data.(type) {
|
||
case string:
|
||
if v == "" {
|
||
log.Debug("String data is empty, setting target to nil")
|
||
*target = nil
|
||
return nil
|
||
}
|
||
raw = []byte(v)
|
||
log.Debug("Converting string to bytes",
|
||
"stringLength", len(v),
|
||
)
|
||
case []byte:
|
||
if len(v) == 0 {
|
||
log.Debug("Byte data is empty, setting target to nil")
|
||
*target = nil
|
||
return nil
|
||
}
|
||
raw = v
|
||
log.Debug("Using byte data directly",
|
||
"byteLength", len(v),
|
||
)
|
||
default:
|
||
log.Error("Unsupported data type",
|
||
"dataType", fmt.Sprintf("%T", v),
|
||
)
|
||
return fmt.Errorf("unsupported data type %T", v)
|
||
}
|
||
|
||
log.Debug("Computing hash for body data",
|
||
"dataLength", len(raw),
|
||
)
|
||
hash, err := hashTool.HashBytes(raw)
|
||
if err != nil {
|
||
log.Error("Failed to compute hash",
|
||
"error", err,
|
||
)
|
||
return err
|
||
}
|
||
|
||
*target = &hash
|
||
log.Debug("Body hash set successfully",
|
||
"hash", hash,
|
||
)
|
||
return nil
|
||
}
|
||
|
||
// RequestBodyFlexible 设置请求体哈希值。
|
||
// 支持 string 和 []byte 类型，nil 或空值会将 RequestBodyHash 设置为 nil。
|
||
func (o *Operation) RequestBodyFlexible(data interface{}) error {
|
||
return o.setBodyHashFlexible(data, &o.RequestBodyHash)
|
||
}
|
||
|
||
// ResponseBodyFlexible 设置响应体哈希值。
|
||
// 支持 string 和 []byte 类型，nil 或空值会将 ResponseBodyHash 设置为 nil。
|
||
func (o *Operation) ResponseBodyFlexible(data interface{}) error {
|
||
return o.setBodyHashFlexible(data, &o.ResponseBodyHash)
|
||
}
|
||
|
||
//
|
||
// ===== 链式调用支持 =====
|
||
//
|
||
|
||
// WithRequestBody 设置请求体哈希并返回自身，支持链式调用。
|
||
func (o *Operation) WithRequestBody(data []byte) *Operation {
|
||
_ = o.RequestBodyFlexible(data)
|
||
return o
|
||
}
|
||
|
||
// WithResponseBody 设置响应体哈希并返回自身，支持链式调用。
|
||
func (o *Operation) WithResponseBody(data []byte) *Operation {
|
||
_ = o.ResponseBodyFlexible(data)
|
||
return o
|
||
}
|
||
|
||
//
|
||
// ===== 初始化与验证 =====
|
||
//
|
||
|
||
// CheckAndInit 校验并初始化 Operation。
|
||
// 自动填充缺失字段（OpID、OpActor），执行业务逻辑验证（doid 格式），
|
||
// 字段非空验证由 validate 标签处理。
|
||
func (o *Operation) CheckAndInit() error {
|
||
log := logger.GetGlobalLogger()
|
||
log.Debug("Checking and initializing operation",
|
||
"opSource", o.OpSource,
|
||
"opCode", o.OpCode,
|
||
"doid", o.Doid,
|
||
)
|
||
if o.OpID == "" {
|
||
o.OpID = helpers.NewUUIDv7()
|
||
log.Debug("Generated new OpID",
|
||
"opID", o.OpID,
|
||
)
|
||
}
|
||
|
||
if o.OpActor == "" {
|
||
o.OpActor = "SYSTEM"
|
||
log.Debug("Set default OpActor to SYSTEM")
|
||
}
|
||
|
||
expectedPrefix := fmt.Sprintf("%s/%s", o.DoPrefix, o.DoRepository)
|
||
if !strings.HasPrefix(o.Doid, expectedPrefix) {
|
||
log.Error("Doid format validation failed",
|
||
"doid", o.Doid,
|
||
"expectedPrefix", expectedPrefix,
|
||
)
|
||
return fmt.Errorf("doid must start with '%s'", expectedPrefix)
|
||
}
|
||
|
||
log.Debug("Validating operation struct")
|
||
if err := helpers.GetValidator().Struct(o); err != nil {
|
||
log.Error("Operation validation failed",
|
||
"error", err,
|
||
"opID", o.OpID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
log.Debug("Operation checked and initialized successfully",
|
||
"opID", o.OpID,
|
||
)
|
||
return nil
|
||
}
|