go-trustlog/api/highclient/client.go

package highclient

import (
	"errors"
	"fmt"

	"github.com/ThreeDotsLabs/watermill/message"

	"go.yandata.net/iod/iod/go-trustlog/api/adapter"
	"go.yandata.net/iod/iod/go-trustlog/api/logger"
	"go.yandata.net/iod/iod/go-trustlog/api/model"
)

type Client struct {
	publisher      message.Publisher
	logger         logger.Logger
	envelopeConfig model.EnvelopeConfig
}

// NewClient 创建HighClient,使用Envelope序列化方式.
// publisher可以使用任意（包含forwarder）创建的publisher,但是我们所有的订阅者必须可以处理Envelope格式的消息.
// 参数:
//   - publisher: 消息发布器
//   - logger: 日志记录器
//   - envelopeConfig: SM2密钥配置,用于签名和序列化
func NewClient(publisher message.Publisher, logger logger.Logger, envelopeConfig model.EnvelopeConfig) *Client {
	return &Client{
		publisher:      publisher,
		logger:         logger,
		envelopeConfig: envelopeConfig,
	}
}

func (c *Client) GetLow() message.Publisher {
	return c.publisher
}

func (c *Client) OperationPublish(operation *model.Operation) error {
	if operation == nil {
		c.logger.Error("operation publish failed: operation is nil")
		return errors.New("operation cannot be nil")
	}

	c.logger.Debug("publishing operation",
		"opID", operation.OpID,
		"opType", operation.OpType,
		"doPrefix", operation.DoPrefix,
	)

	err := publish(operation, adapter.OperationTopic, c.publisher, c.envelopeConfig, c.logger)
	if err != nil {
		c.logger.Error("operation publish failed",
			"opID", operation.OpID,
			"error", err,
		)
		return err
	}

	c.logger.Info("operation published successfully",
		"opID", operation.OpID,
		"opType", operation.OpType,
	)
	return nil
}

func (c *Client) RecordPublish(record *model.Record) error {
	if record == nil {
		c.logger.Error("record publish failed: record is nil")
		return errors.New("record cannot be nil")
	}

	c.logger.Debug("publishing record",
		"recordID", record.ID,
		"rcType", record.RCType,
		"doPrefix", record.DoPrefix,
	)

	err := publish(record, adapter.RecordTopic, c.publisher, c.envelopeConfig, c.logger)
	if err != nil {
		c.logger.Error("record publish failed",
			"recordID", record.ID,
			"error", err,
		)
		return err
	}

	c.logger.Info("record published successfully",
		"recordID", record.ID,
		"rcType", record.RCType,
	)
	return nil
}

func (c *Client) Close() error {
	c.logger.Info("closing high client")
	err := c.publisher.Close()
	if err != nil {
		c.logger.Error("failed to close publisher", "error", err)
		return err
	}
	c.logger.Info("high client closed successfully")
	return nil
}

// publish 通用的发布函数，支持任何实现了 Trustlog 接口的类型。
// 使用 Envelope 格式序列化并发布到指定 topic。
func publish(
	data model.Trustlog,
	topic string,
	publisher message.Publisher,
	config model.EnvelopeConfig,
	logger logger.Logger,
) error {
	messageKey := data.Key()

	logger.Debug("starting envelope serialization",
		"messageKey", messageKey,
		"topic", topic,
	)

	// 使用 Envelope 序列化（MarshalTrustlog 会自动提取 producerID）
	envelopeData, err := model.MarshalTrustlog(data, config)
	if err != nil {
		logger.Error("envelope serialization failed",
			"messageKey", messageKey,
			"error", err,
		)
		return fmt.Errorf("failed to marshal envelope: %w", err)
	}

	logger.Debug("envelope serialized successfully",
		"messageKey", messageKey,
		"envelopeSize", len(envelopeData),
	)

	msg := message.NewMessage(messageKey, envelopeData)
	logger.Debug("publishing message to topic",
		"messageKey", messageKey,
		"topic", topic,
	)

	if publishErr := publisher.Publish(topic, msg); publishErr != nil {
		logger.Error("failed to publish to topic",
			"messageKey", messageKey,
			"topic", topic,
			"error", publishErr,
		)
		return fmt.Errorf("failed to publish message to topic %s: %w", topic, publishErr)
	}

	logger.Debug("message published to topic successfully",
		"messageKey", messageKey,
		"topic", topic,
	)
	return nil
}