sqlxplus.go

package outbox

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

	"github.com/bwmarrin/snowflake"
	"github.com/jmoiron/sqlx"

	"github.com/nilorg/eventbus"
	"github.com/nilorg/sqlxplus"
)

// sqlxPlusEngine ...
type sqlxPlusEngine struct {
	options             *EngineOptions
	db                  *sqlx.DB
	log                 sqlxplus.Logger
	node                *snowflake.Node
	eventBus            eventbus.EventBus
	subscribeItems      map[string]*subscribeItem
	subscribeItemsMutex sync.Mutex
	logger              Logger
}

// Subscribe ...
func (e *sqlxPlusEngine) Subscribe(ctx context.Context, topic string, h SubscribeHandler) (err error) {
	return e.subscribe(ctx, topic, h, false)
}

// SubscribeAsync ...
func (e *sqlxPlusEngine) SubscribeAsync(ctx context.Context, topic string, h SubscribeHandler) (err error) {
	return e.subscribe(ctx, topic, h, true)
}

func (e *sqlxPlusEngine) subscribe(ctx context.Context, topic string, h SubscribeHandler, async bool) (err error) {
	e.subscribeItemsMutex.Lock()
	defer e.subscribeItemsMutex.Unlock()

	group, _ := eventbus.FromGroupIDContext(ctx)

	hex := fmt.Sprintf("%s-%s", topic, group)
	e.subscribeItems[hex] = &subscribeItem{
		Topic: topic,
		Group: group,
		h:     h,
	}

	if async {
		return e.eventBus.SubscribeAsync(ctx, topic, e.newSubscribeHandler(topic, group, h))
	}
	return e.eventBus.Subscribe(ctx, topic, e.newSubscribeHandler(topic, group, h))
}

func (e *sqlxPlusEngine) newSubscribeHandler(topic, group string, h SubscribeHandler) eventbus.SubscribeHandler {
	return func(ctx context.Context, baseMsg *eventbus.Message) (err error) {
		id := e.node.Generate().Int64()
		var bytes []byte
		if bytes, err = json.Marshal(baseMsg); err != nil {
			return
		}
		r := &Received{
			ID:         id,
			Version:    MessageVersion,
			Topic:      topic,
			Group:      group,
			Value:      string(bytes),
			Retries:    0,
			CreatedAt:  time.Now(),
			StatusName: StatusNameScheduled,
		}
		msg := Message(*baseMsg)
		hErr := h(ctx, &msg)
		if hErr != nil {
			// 使用日志组件，打印日志
			if group != "" {
				e.logger.Errorf(ctx, "execute subscribe %s handler error: %s", topic, hErr)
			} else {
				e.logger.Errorf(ctx, "execute subscribe %s(%s) handler error: %s", topic, group, hErr)
			}
		} else {
			// 成功消息记录，到期时间
			exp := time.Now().Add(e.options.SucceedMessageExpiredAfter)
			r.ExpiresAt = &exp
			r.StatusName = StatusNameSucceeded
		}
		// 使用 ExecTransaction 记录收件箱
		return sqlxplus.ExecTransaction(ctx, func(txCtx context.Context) error {
			conn, err := sqlxplus.FromSqlxContext(txCtx)
			if err != nil {
				return err
			}
			return e.insertReceivedFromSqlxPlus(txCtx, conn, r)
		}, sqlxplus.WithExecTransactionDB(e.db), sqlxplus.WithExecTransactionLogger(e.log))
	}
}

// Publish ...
func (e *sqlxPlusEngine) Publish(ctx context.Context, topic string, v interface{}, callback ...string) (err error) {
	return e.publish(ctx, topic, v, false, callback...)
}

// PublishAsync ...
func (e *sqlxPlusEngine) PublishAsync(ctx context.Context, topic string, v interface{}, callback ...string) (err error) {
	return e.publish(ctx, topic, v, true, callback...)
}

func (e *sqlxPlusEngine) publish(ctx context.Context, topic string, v interface{}, async bool, callback ...string) (err error) {
	id := e.node.Generate().Int64()
	timeNow := time.Now()
	callbackName := ""
	if len(callback) > 0 {
		callbackName = callback[0]
	}
	var value string
	if value, err = EncodeValue(v); err != nil {
		return
	}
	msg := &eventbus.Message{
		Header: eventbus.MessageHeader{
			MessageHeaderMsgIDKey:       fmt.Sprint(id),
			MessageHeaderMsgTopicKey:    topic,
			MessageHeaderMsgTypeKey:     reflect.TypeOf(v).Name(),
			MessageHeaderMsgSendTimeKey: fmt.Sprint(timeNow.Unix()),
			MessageHeaderMsgCallbackKey: callbackName,
		},
		Value: value,
	}
	var bytes []byte
	if bytes, err = json.Marshal(msg); err != nil {
		return
	}
	p := &Published{
		ID:         id,
		Version:    MessageVersion,
		Topic:      topic,
		Value:      string(bytes),
		Retries:    0,
		CreatedAt:  timeNow,
		StatusName: StatusNameScheduled,
	}
	var pubErr error
	if async {
		pubErr = e.eventBus.PublishAsync(ctx, topic, msg)
	} else {
		pubErr = e.eventBus.Publish(ctx, topic, msg)
	}
	if pubErr != nil {
		// 使用日志组件，打印日志
		e.logger.Errorf(ctx, "execute async:%v publish %s error: %s", async, topic, pubErr)
	} else {
		// 成功消息记录，到期时间
		exp := time.Now().Add(e.options.SucceedMessageExpiredAfter)
		p.ExpiresAt = &exp
		p.StatusName = StatusNameSucceeded
	}
	// 使用 ExecTransaction 记录发件箱
	return sqlxplus.ExecTransaction(ctx, func(txCtx context.Context) error {
		conn, err := sqlxplus.FromSqlxContext(txCtx)
		if err != nil {
			return err
		}
		return e.insertPublishedFromSqlxPlus(txCtx, conn, p)
	}, sqlxplus.WithExecTransactionDB(e.db), sqlxplus.WithExecTransactionLogger(e.log))
}

func (e *sqlxPlusEngine) insertPublishedFromSqlxPlus(ctx context.Context, conn *sqlxplus.DBConnect, p *Published) error {
	if p.Version == "" {
		p.Version = MessageVersion
	}
	query := `INSERT INTO ` + PublishedTableName + ` (id, version, topic, value, retries, created_at, expires_at, status_name) 
			  VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
	_, err := sqlxplus.ExecContext(ctx, conn.Adapter(), query, p.ID, p.Version, p.Topic, p.Value, p.Retries, p.CreatedAt, p.ExpiresAt, p.StatusName)
	return err
}

func (e *sqlxPlusEngine) insertReceivedFromSqlxPlus(ctx context.Context, conn *sqlxplus.DBConnect, r *Received) error {
	if r.Version == "" {
		r.Version = MessageVersion
	}
	query := `INSERT INTO ` + ReceivedTableName + ` (id, version, name, ` + "`group`" + `, value, retries, created_at, expires_at, status_name) 
			  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
	_, err := sqlxplus.ExecContext(ctx, conn.Adapter(), query, r.ID, r.Version, r.Topic, r.Group, r.Value, r.Retries, r.CreatedAt, r.ExpiresAt, r.StatusName)
	return err
}

func (e *sqlxPlusEngine) changePublishState(ctx context.Context, conn *sqlxplus.DBConnect, msgID int64, state string) error {
	query := `UPDATE ` + PublishedTableName + ` SET status_name = ? WHERE id = ?`
	_, err := sqlxplus.ExecContext(ctx, conn.Adapter(), query, state, msgID)
	return err
}

func (e *sqlxPlusEngine) changePublishStateSucceeded(ctx context.Context, conn *sqlxplus.DBConnect, msgID int64) error {
	exp := time.Now().Add(e.options.SucceedMessageExpiredAfter)
	query := `UPDATE ` + PublishedTableName + ` SET status_name = ?, expires_at = ? WHERE id = ?`
	_, err := sqlxplus.ExecContext(ctx, conn.Adapter(), query, StatusNameSucceeded, exp, msgID)
	return err
}

func (e *sqlxPlusEngine) changeReceiveState(ctx context.Context, conn *sqlxplus.DBConnect, msgID int64, state string) error {
	query := `UPDATE ` + ReceivedTableName + ` SET status_name = ? WHERE id = ?`
	_, err := sqlxplus.ExecContext(ctx, conn.Adapter(), query, state, msgID)
	return err
}

func (e *sqlxPlusEngine) changeReceiveStateSucceeded(ctx context.Context, conn *sqlxplus.DBConnect, msgID int64) error {
	exp := time.Now().Add(e.options.SucceedMessageExpiredAfter)
	query := `UPDATE ` + ReceivedTableName + ` SET status_name = ?, expires_at = ? WHERE id = ?`
	_, err := sqlxplus.ExecContext(ctx, conn.Adapter(), query, StatusNameSucceeded, exp, msgID)
	return err
}

func (e *sqlxPlusEngine) findPublishByState(ctx context.Context, conn *sqlxplus.DBConnect, state string) (list []*Published, err error) {
	query := `SELECT * FROM ` + PublishedTableName + ` WHERE version = ? AND status_name = ?`
	err = sqlx.SelectContext(ctx, conn.Adapter(), &list, query, MessageVersion, state)
	return
}

func (e *sqlxPlusEngine) findReceiveByState(ctx context.Context, conn *sqlxplus.DBConnect, state string) (list []*Received, err error) {
	query := `SELECT * FROM ` + ReceivedTableName + ` WHERE version = ? AND status_name = ?`
	err = sqlx.SelectContext(ctx, conn.Adapter(), &list, query, MessageVersion, state)
	return
}

func (e *sqlxPlusEngine) deletePublished(ctx context.Context, conn *sqlxplus.DBConnect, state string, expired time.Time) (err error) {
	query := `DELETE FROM ` + PublishedTableName + ` WHERE version = ? AND status_name = ? AND expires_at <= ?`
	_, err = sqlxplus.ExecContext(ctx, conn.Adapter(), query, MessageVersion, state, expired)
	return
}

func (e *sqlxPlusEngine) deleteReceived(ctx context.Context, conn *sqlxplus.DBConnect, state string, expired time.Time) (err error) {
	query := `DELETE FROM ` + ReceivedTableName + ` WHERE version = ? AND status_name = ? AND expires_at <= ?`
	_, err = sqlxplus.ExecContext(ctx, conn.Adapter(), query, MessageVersion, state, expired)
	return
}

func (e *sqlxPlusEngine) retriePublished(ctx context.Context, conn *sqlxplus.DBConnect, msgID int64) (err error) {
	query := `UPDATE ` + PublishedTableName + ` SET retries = retries + 1 WHERE id = ?`
	_, err = sqlxplus.ExecContext(ctx, conn.Adapter(), query, msgID)
	return
}

func (e *sqlxPlusEngine) retrieReceived(ctx context.Context, conn *sqlxplus.DBConnect, msgID int64) (err error) {
	query := `UPDATE ` + ReceivedTableName + ` SET retries = retries + 1 WHERE id = ?`
	_, err = sqlxplus.ExecContext(ctx, conn.Adapter(), query, msgID)
	return err
}

// Begin ...
func (e *sqlxPlusEngine) Begin(ctx context.Context, opts ...*sql.TxOptions) (tx Transactioner, err error) {
	var sqlxTx *sqlx.Tx
	if len(opts) > 0 {
		sqlxTx, err = e.db.BeginTxx(ctx, opts[0])
	} else {
		sqlxTx, err = e.db.BeginTxx(ctx, nil)
	}
	if err != nil {
		return
	}
	tx = &sqlxPlusTransaction{
		tx:      &sqlxplus.Tx{SqlxTx: sqlxTx, Log: e.log},
		logger:  e.logger,
		publish: e.publishInTransaction,
	}
	return
}

// publishInTransaction 在事务中发布消息
func (e *sqlxPlusEngine) publishInTransaction(ctx context.Context, db interface{}, topic string, v interface{}, async bool, callback ...string) (err error) {
	// db 参数实际上是 *sqlxplus.Tx，但为了统一接口，我们通过上下文获取连接
	id := e.node.Generate().Int64()
	timeNow := time.Now()
	callbackName := ""
	if len(callback) > 0 {
		callbackName = callback[0]
	}
	var value string
	if value, err = EncodeValue(v); err != nil {
		return
	}
	msg := &eventbus.Message{
		Header: eventbus.MessageHeader{
			MessageHeaderMsgIDKey:       fmt.Sprint(id),
			MessageHeaderMsgTopicKey:    topic,
			MessageHeaderMsgTypeKey:     reflect.TypeOf(v).Name(),
			MessageHeaderMsgSendTimeKey: fmt.Sprint(timeNow.Unix()),
			MessageHeaderMsgCallbackKey: callbackName,
		},
		Value: value,
	}
	var bytes []byte
	if bytes, err = json.Marshal(msg); err != nil {
		return
	}
	p := &Published{
		ID:         id,
		Version:    MessageVersion,
		Topic:      topic,
		Value:      string(bytes),
		Retries:    0,
		CreatedAt:  timeNow,
		StatusName: StatusNameScheduled,
	}
	var pubErr error
	if async {
		pubErr = e.eventBus.PublishAsync(ctx, topic, msg)
	} else {
		pubErr = e.eventBus.Publish(ctx, topic, msg)
	}
	if pubErr != nil {
		e.logger.Errorf(ctx, "execute async:%v publish %s error: %s", async, topic, pubErr)
	} else {
		exp := time.Now().Add(e.options.SucceedMessageExpiredAfter)
		p.ExpiresAt = &exp
		p.StatusName = StatusNameSucceeded
	}

	// 从上下文获取连接
	var conn *sqlxplus.DBConnect
	if conn, err = sqlxplus.FromSqlxContext(ctx); err != nil {
		return
	}
	return e.insertPublishedFromSqlxPlus(ctx, conn, p)
}

func (e *sqlxPlusEngine) Transaction(ctx context.Context, h TransactionHandler, args ...*CommitMessage) (err error) {
	return sqlxplus.ExecTransaction(ctx, func(txCtx context.Context) error {
		conn, err := sqlxplus.FromSqlxContext(txCtx)
		if err != nil {
			return err
		}

		// 执行用户业务逻辑
		if err = h(txCtx, conn.Adapter()); err != nil {
			return err
		}

		// 发布消息
		if len(args) > 0 {
			for _, arg := range args {
				if pubErr := e.publishInTransaction(txCtx, conn.Transaction(), arg.Topic, arg.Value, false, arg.CallbackTopic); pubErr != nil {
					e.logger.Errorf(txCtx, "commit publish error: %s", pubErr)
					return pubErr
				}
			}
		}
		return nil
	}, sqlxplus.WithExecTransactionDB(e.db), sqlxplus.WithExecTransactionLogger(e.log))
}

// dataClean 数据清理
func (e *sqlxPlusEngine) dataClean() (err error) {
	ctx := context.Background()
	e.logger.Debugln(ctx, "execute dataClean")
	expd := time.Now()

	// 创建上下文并包装 DB
	ctx = sqlxplus.NewSqlxContext(ctx, &sqlxplus.DB{SqlxDB: e.db, Log: e.log})
	var dbConn *sqlxplus.DBConnect
	if dbConn, err = sqlxplus.FromSqlxContext(ctx); err != nil {
		return
	}

	if err = e.deletePublished(ctx, dbConn, StatusNameSucceeded, expd); err != nil {
		return
	}
	if err = e.deleteReceived(ctx, dbConn, StatusNameSucceeded, expd); err != nil {
		return
	}
	return
}

// failedRetryInterval 失败重试间隔
func (e *sqlxPlusEngine) failedRetryInterval() (err error) {
	ctx := context.Background()
	e.logger.Debugln(ctx, "execute failedRetryInterval")
	if err = e.failedPublishedRetryInterval(); err != nil {
		return
	}
	if err = e.failedReceivedRetryInterval(); err != nil {
		return
	}
	return
}

// failedPublishedRetryInterval 失败重试间隔
func (e *sqlxPlusEngine) failedPublishedRetryInterval() (err error) {
	ctx := context.Background()

	// 创建上下文并包装 DB
	ctx = sqlxplus.NewSqlxContext(ctx, &sqlxplus.DB{SqlxDB: e.db, Log: e.log})
	var conn *sqlxplus.DBConnect
	if conn, err = sqlxplus.FromSqlxContext(ctx); err != nil {
		return
	}

	var publisheds []*Published
	if publisheds, err = e.findPublishByState(ctx, conn, StatusNameScheduled); err != nil {
		return
	}

	for _, published := range publisheds {
		// 重试次数超过50，执行通知回调
		if published.Retries >= 50 {
			if err = e.changePublishState(ctx, conn, published.ID, StatusNameFailed); err != nil {
				return
			}
			if e.options.FailedThresholdCallback != nil {
				go e.options.FailedThresholdCallback(ctx, CallbackTypePublished, published)
			}
			continue
		}
		// 版本不相同，跳过执行
		if published.Version != MessageVersion {
			continue
		}
		// 添加重试次数
		if err = e.retriePublished(ctx, conn, published.ID); err != nil {
			return
		}
		var msg Message
		if msgErr := json.Unmarshal([]byte(published.Value), &msg); msgErr != nil {
			e.logger.Errorf(ctx, "failedRetryInterval-decodeValue error: %v", msgErr)
		} else {
			if pubErr := e.eventBus.Publish(ctx, published.Topic, msg); pubErr != nil {
				e.logger.Errorf(ctx, "failedRetryInterval-Publish error: %v", pubErr)
			} else {
				if err = e.changePublishStateSucceeded(ctx, conn, published.ID); err != nil {
					return
				}
			}
		}
	}
	return
}

// failedReceivedRetryInterval 失败重试间隔
func (e *sqlxPlusEngine) failedReceivedRetryInterval() (err error) {
	ctx := context.Background()

	// 创建上下文并包装 DB
	ctx = sqlxplus.NewSqlxContext(ctx, &sqlxplus.DB{SqlxDB: e.db, Log: e.log})
	var conn *sqlxplus.DBConnect
	if conn, err = sqlxplus.FromSqlxContext(ctx); err != nil {
		return
	}

	var receiveds []*Received
	if receiveds, err = e.findReceiveByState(ctx, conn, StatusNameScheduled); err != nil {
		return
	}
	e.subscribeItemsMutex.Lock()
	defer e.subscribeItemsMutex.Unlock()

	for _, received := range receiveds {
		// 重试次数超过50，执行通知回调
		if received.Retries >= 50 {
			if err = e.changeReceiveState(ctx, conn, received.ID, StatusNameFailed); err != nil {
				return
			}
			if e.options.FailedThresholdCallback != nil {
				go e.options.FailedThresholdCallback(ctx, CallbackTypeReceived, received)
			}
			continue
		}
		// 版本不相同，跳过执行
		if received.Version != MessageVersion {
			continue
		}
		// 添加重试次数
		if err = e.retrieReceived(ctx, conn, received.ID); err != nil {
			return
		}
		var msg Message
		if msgErr := json.Unmarshal([]byte(received.Value), &msg); msgErr != nil {
			e.logger.Errorf(ctx, "failedRetryInterval-decodeValue error: %v", msgErr)
		} else {
			hex := fmt.Sprintf("%s-%s", received.Topic, received.Group)
			if subscribeItem, subscribeItemOk := e.subscribeItems[hex]; !subscribeItemOk || subscribeItem.h == nil {
				e.logger.Warnf(ctx, "failedRetryInterval-subscribeItem %s error: %v", hex, msgErr)
			} else {
				if subErr := subscribeItem.h(ctx, &msg); subErr != nil {
					e.logger.Errorf(ctx, "failedRetryInterval-subscribeItem %s execute error: %v", hex, msgErr)
				} else {
					if err = e.changeReceiveStateSucceeded(ctx, conn, received.ID); err != nil {
						return
					}
				}
			}
		}
	}
	return
}

func (e *sqlxPlusEngine) sentry() {
	go func() {
		dataCleanTicker := time.NewTicker(e.options.DataCleanInterval)
		defer dataCleanTicker.Stop()
		failedRetryIntervalTicker := time.NewTicker(e.options.FailedRetryInterval)
		defer failedRetryIntervalTicker.Stop()
		for {
			select {
			case <-dataCleanTicker.C:
				recoverHandle(e.logger, e.dataClean)
			case <-failedRetryIntervalTicker.C:
				recoverHandle(e.logger, e.failedRetryInterval)
			}
		}
	}()
}

type sqlxPlusTransaction struct {
	tx      *sqlxplus.Tx
	logger  Logger
	publish func(ctx context.Context, db interface{}, topic string, v interface{}, async bool, callback ...string) (err error)
}

func (tran *sqlxPlusTransaction) Session() interface{} {
	return tran.tx
}

func (tran *sqlxPlusTransaction) Rollback(ctx context.Context) (err error) {
	// sqlxplus.Tx 没有直接的 Rollback 方法，需要通过底层 sqlx.Tx
	err = tran.tx.GetSqlxTx().Rollback()
	return
}

func (tran *sqlxPlusTransaction) Commit(ctx context.Context, args ...*CommitMessage) (err error) {
	if len(args) > 0 {
		for _, arg := range args {
			if pubErr := tran.publish(ctx, tran.tx, arg.Topic, arg.Value, false, arg.CallbackTopic); pubErr != nil {
				tran.logger.Errorf(ctx, "commit publish error: %s", pubErr)
				err = tran.Rollback(ctx)
				return
			}
		}
	}
	// sqlxplus.Tx 没有直接的 Commit 方法，需要通过底层 sqlx.Tx
	err = tran.tx.GetSqlxTx().Commit()
	return
}

// sqlxPlusLogger 适配 outbox.Logger 到 sqlxplus.Logger
type sqlxPlusLogger struct {
	logger Logger
}

func (l *sqlxPlusLogger) Printf(ctx context.Context, query string, args ...interface{}) {
	l.logger.Debugf(ctx, query, args...)
}

func (l *sqlxPlusLogger) Println(ctx context.Context, args ...interface{}) {
	l.logger.Debugln(ctx, args...)
}

func (l *sqlxPlusLogger) Errorf(ctx context.Context, format string, args ...interface{}) {
	l.logger.Errorf(ctx, format, args...)
}

func (l *sqlxPlusLogger) Errorln(ctx context.Context, args ...interface{}) {
	l.logger.Errorln(ctx, args...)
}