remark42/backend/vendor/github.com/redis/go-redis/v9/tx.go

package redis

import (
	"context"
	"errors"

	"github.com/redis/go-redis/v9/internal/proto"
)

// TxFailedErr transaction redis failed.
const TxFailedErr = proto.RedisError("redis: transaction failed")

// Tx implements Redis transactions as described in
// https://redis.io/docs/latest/develop/using-commands/transactions. It's NOT safe for concurrent use
// by multiple goroutines, because Exec resets list of watched keys.
//
// If you don't need WATCH, use Pipeline instead.
type Tx struct {
	baseClient
	cmdable
	statefulCmdable

	// watchArmed reports whether a WATCH issued through Tx.Watch may still be
	// active on the connection. It is set on a successful WATCH and cleared once
	// the watched keys are discarded server-side: on UNWATCH (Tx.Unwatch) and on
	// EXEC, including an aborted EXEC (the TxPipeline closure). Close uses it to
	// skip an otherwise redundant UNWATCH round trip.
	//
	// Only Tx.Watch, Tx.Unwatch and the TxPipeline EXEC closure maintain this
	// flag. go-redis never issues a standalone DISCARD (Pipeline.Discard is a
	// client-side buffer reset, not a server command). Issuing a WATCH directly
	// via Process bypasses tracking, so Close would not release it and the watch
	// would leak onto the pooled connection; that is the one case where skipping
	// UNWATCH is unsafe, and using raw Process for WATCH/EXEC/UNWATCH is therefore
	// unsupported.
	//
	// Tx is not safe for concurrent use (see above), so watchArmed is accessed
	// only from the goroutine that owns the Tx and needs no synchronization.
	watchArmed bool
}

func (c *Client) newTx() *Tx {
	tx := Tx{
		baseClient: baseClient{
			opt:           c.cloneOpt(), // Clone options under optLock to avoid race with initConn
			connPool:      c.baseClient.newStickyConnPool(),
			hooksMixin:    c.hooksMixin.clone(),
			pushProcessor: c.pushProcessor, // Copy push processor from parent client
			onClose:       &onCloseHooks{},
			// Share the HIMPORT fieldset registry: the sticky pool borrows
			// connections from the parent client's pool, so fieldsets
			// prepared on them stay valid after the connections are
			// returned.
			himport: c.himport,
			// Carry the shared eviction hook (not csc: a sticky Tx must not serve
			// cached reads) so close/reinit hooks on a Watch-initialized conn still
			// evict from the parent cache.
			cscPoolHook: c.cscPoolHook,
			cscActive:   c.cscActive,
		},
	}
	tx.init()
	return &tx
}

func (c *Tx) init() {
	c.cmdable = c.Process
	c.statefulCmdable = c.Process

	c.initHooks(hooks{
		dial:       c.baseClient.dial,
		process:    c.baseClient.process,
		pipeline:   c.baseClient.processPipeline,
		txPipeline: c.baseClient.processTxPipeline,
	})
}

func (c *Tx) Process(ctx context.Context, cmd Cmder) error {
	err := c.processHook(ctx, cmd)
	cmd.SetErr(err)
	return err
}

// Watch prepares a transaction and marks the keys to be watched
// for conditional execution if there are any keys.
//
// The transaction is automatically closed when fn exits.
func (c *Client) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
	tx := c.newTx()
	defer tx.Close(ctx)
	if len(keys) > 0 {
		if err := tx.Watch(ctx, keys...).Err(); err != nil {
			return err
		}
	}
	return fn(tx)
}

// Close closes the transaction, releasing any open resources.
func (c *Tx) Close(ctx context.Context) error {
	// UNWATCH is only needed while a WATCH is still active. EXEC discards the
	// watched keys server-side on both commit and abort, so the common
	// WATCH/.../EXEC paths leave nothing to release and avoid the extra round
	// trip.
	if c.watchArmed {
		_ = c.Unwatch(ctx).Err()
	}
	return c.baseClient.Close()
}

// Watch marks the keys to be watched for conditional execution
// of a transaction.
func (c *Tx) Watch(ctx context.Context, keys ...string) *StatusCmd {
	args := make([]interface{}, 1+len(keys))
	args[0] = "watch"
	for i, key := range keys {
		args[1+i] = key
	}
	cmd := NewStatusCmd(ctx, args...)
	_ = c.Process(ctx, cmd)
	// A successful WATCH leaves keys watched on the connection that Close must
	// later release with UNWATCH.
	if cmd.Err() == nil {
		c.watchArmed = true
	}
	return cmd
}

// Unwatch flushes all the previously watched keys for a transaction.
func (c *Tx) Unwatch(ctx context.Context, keys ...string) *StatusCmd {
	args := make([]interface{}, 1+len(keys))
	args[0] = "unwatch"
	for i, key := range keys {
		args[1+i] = key
	}
	cmd := NewStatusCmd(ctx, args...)
	_ = c.Process(ctx, cmd)
	// The watched keys have been released, so Close need not UNWATCH again.
	if cmd.Err() == nil {
		c.watchArmed = false
	}
	return cmd
}

// Pipeline creates a pipeline. Usually it is more convenient to use Pipelined.
func (c *Tx) Pipeline() Pipeliner {
	pipe := Pipeline{
		exec: func(ctx context.Context, cmds []Cmder) error {
			return c.processPipelineHook(ctx, cmds)
		},
	}
	pipe.init()
	return &pipe
}

// Pipelined executes commands queued in the fn outside of the transaction.
// Use TxPipelined if you need transactional behavior.
func (c *Tx) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
	return c.Pipeline().Pipelined(ctx, fn)
}

// TxPipelined executes commands queued in the fn in the transaction.
//
// When using WATCH, EXEC will execute commands only if the watched keys
// were not modified, allowing for a check-and-set mechanism.
//
// Exec always returns list of commands. If transaction fails
// TxFailedErr is returned. Otherwise Exec returns an error of the first
// failed command or nil.
func (c *Tx) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
	return c.TxPipeline().Pipelined(ctx, fn)
}

// TxPipeline creates a pipeline. Usually it is more convenient to use TxPipelined.
func (c *Tx) TxPipeline() Pipeliner {
	pipe := Pipeline{
		exec: func(ctx context.Context, cmds []Cmder) error {
			cmds = wrapMultiExec(ctx, cmds)
			err := c.processTxPipelineHook(ctx, cmds)
			// EXEC discards the watched keys server-side, so the watch is
			// cleared only when EXEC actually ran: a nil error (committed),
			// TxFailedErr (a watched key changed, EXEC returned nil), or an
			// EXECABORT error (a queued command was rejected, so EXEC discarded
			// the transaction). All three release the watched keys. Any other
			// error may be reported before EXEC executes (for example -LOADING on
			// the MULTI reply) or on a broken connection, so leave watchArmed set
			// and let Close send UNWATCH rather than risk leaving a watch on a
			// pooled connection.
			if err == nil || errors.Is(err, TxFailedErr) || IsExecAbortError(err) {
				c.watchArmed = false
			}
			return err
		},
	}
	pipe.init()
	return &pipe
}

func wrapMultiExec(ctx context.Context, cmds []Cmder) []Cmder {
	if len(cmds) == 0 {
		panic("not reached")
	}
	cmdsCopy := make([]Cmder, len(cmds)+2)
	cmdsCopy[0] = NewStatusCmd(ctx, "multi")
	copy(cmdsCopy[1:], cmds)
	cmdsCopy[len(cmdsCopy)-1] = NewSliceCmd(ctx, "exec")
	return cmdsCopy
}