func popTask(ctx context.Context, r redis.Cmdable, group, id string, maxLen int64, f func(p redis.Pipeliner, payload string, startAt time.Time) error, k string) (err error) {
	var (
		readyStream   = ReadyTaskKey(k)
		inputStream   = InputTaskKey(k)
		waitingStream = WaitingTaskKey(k)
	)
	for {
		vs, err := RunInterfaceSliceScript(ctx, r, popTaskScript, []string{readyStream, inputStream, waitingStream}, group, id, time.Now().UnixNano(), maxLen).Result()
		if err != nil && err != redis.Nil {
			return ConvertError(err)
		}
		typ, ok := vs[0].(string)
		if !ok {
			panic(fmt.Sprintf("invalid type of entry at index %d of result returned by Redis task pop script: %T", 0, vs[0]))
		}
		var fields map[string]string
		if len(vs) > 1 {
			fields = make(map[string]string, (len(vs)-1)/2)
			for i := 1; i < len(vs); i += 2 {
				k, ok := vs[i].(string)
				if !ok {
					panic(fmt.Sprintf("invalid type of entry at index %d of result returned by Redis task pop script: %T", i, vs[i]))
				}
				v, ok := vs[i+1].(string)
				if !ok {
					panic(fmt.Sprintf("invalid type of entry at index %d of result returned by Redis task pop script: %T", i+1, vs[i+1]))
				}
				fields[k] = v
			}
		}
		switch typ {
		case "ready":
		case "waiting":
			xs, err := r.XPendingExt(ctx, &redis.XPendingExtArgs{
				Stream: readyStream,
				Group:  group,
				Start:  "-",
				End:    "+",
				Count:  1,
			}).Result()
			if err != nil && err != redis.Nil {
				return ConvertError(err)
			}
			if len(xs) > 0 {
			}
			var block time.Duration
			if s, ok := fields[nextAtKey]; ok {
				nextAt, err := parseTime(s)
				if err != nil {
					return errInvalidKeyValueType.WithAttributes("key", nextAtKey).WithCause(err)
				}
				if nextAt.IsZero() {
					block = -1
				} else {
					now := time.Now()
					if nextAt.Before(now) {
						continue
					} else {
						block = nextAt.Sub(now)
					}
				}
			}
			var id string
			if s, ok := fields[lastIDKey]; ok {
				id = s
			} else {
				id = "0-0"
			}
			_, err = r.XRead(ctx, &redis.XReadArgs{
				Streams: []string{inputStream, id},
				Count:   1,
				Block:   block,
			}).Result()
			if err != nil && err != redis.Nil {
				return ConvertError(err)
			}
			continue
		default:
			panic(fmt.Sprintf("unknown result type received `%s`", typ))
		}
		var startAt time.Time
		if s, ok := fields[startAtKey]; ok {
			startAt, err = parseTime(s)
			if err != nil {
				return errInvalidKeyValueType.WithAttributes("key", startAtKey).WithCause(err)
			}
		}
		p := r.Pipeline()
		defer func() {
			_, pErr := p.Exec(ctx)
			if err == nil && pErr != nil {
				err = ConvertError(pErr)
			}
		}()
		if err = f(p, fields["payload"], startAt); err != nil {
			return err
		}
		p.XAck(ctx, readyStream, group, fields["id"])
		p.XDel(ctx, readyStream, fields["id"])
		return nil
	}
}
popTask calls f on the most recent task in the queue, for which timestamp is in range [0, time.Now()] or blocks until such is available or context is done.
If there are no tasks available for immediate processing, popTask lazily dispatches available tasks for itself and all other callers of popTask.
group is the consumer group name.
id is the consumer group ID.
k is the keys to pop from.
Pipeline is executed even if f returns an error.
Tasks are acked only if f returns without error.
package redis
import (
	"context"
	"crypto/rand"
	"crypto/tls"
	"encoding/base64"
	"fmt"
	"net"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/go-redis/redis/v8"
	"github.com/gogo/protobuf/proto"
	"github.com/oklog/ulid/v2"
	"go.thethings.network/lorawan-stack/v3/pkg/config/tlsconfig"
	"go.thethings.network/lorawan-stack/v3/pkg/log"
)
const (
	separator = ':'
)
var encoding = base64.RawStdEncoding
type WatchCmdable interface {
	redis.Cmdable
	Watch(ctx context.Context, fn func(*redis.Tx) error, keys ...string) error
}
func MarshalProto(pb proto.Message) (string, error) {
	b, err := proto.Marshal(pb)
	if err != nil {
		return "", errEncode.WithCause(err)
	}
	protosMarshaled.Inc()
	return encoding.EncodeToString(b), nil
}
func UnmarshalProto(s string, pb proto.Message) error {
	b, err := encoding.DecodeString(s)
	if err != nil {
		return errDecode.WithCause(err)
	}
	if err = proto.Unmarshal(b, pb); err != nil {
		return errDecode.WithCause(err)
	}
	protosUnmarshaled.Inc()
	return nil
}
func Key(ks ...string) string {
	return strings.Join(ks, string(separator))
}
type Client struct {
	*redis.Client
	namespace string
}
type Config struct {
	Address       string         `name:"address" description:"Address of the Redis server"`
	Password      string         `name:"password" description:"Password of the Redis server"`
	Database      int            `name:"database" description:"Redis database to use"`
	RootNamespace []string       `name:"namespace" description:"Namespace for Redis keys"`
	PoolSize      int            `name:"pool-size" description:"The maximum number of database connections"`
	Failover      FailoverConfig `name:"failover" description:"Redis failover configuration"`
	TLS           struct {
		Require          bool `name:"require" description:"Require TLS"`
		tlsconfig.Client `name:",squash"`
	} `name:"tls"`
	namespace []string
}
func (c Config) WithNamespace(namespace ...string) *Config {
	deriv := c
	deriv.namespace = namespace
	return &deriv
}
func (c Config) IsZero() bool {
	if c.Failover.Enable {
		return c.Failover.MasterName == "" && len(c.Failover.Addresses) == 0
	}
	return c.Address == ""
}
type FailoverConfig struct {
	Enable     bool     `name:"enable" description:"Enable failover using Redis Sentinel"`
	Addresses  []string `name:"addresses" description:"Redis Sentinel server addresses"`
	MasterName string   `name:"master-name" description:"Redis Sentinel master name"`
}
func (c Config) makeDialer() func(ctx context.Context, network, addr string) (net.Conn, error) {
	var (
		tlsConfig    *tls.Config
		tlsConfigErr error
	)
	if c.TLS.Require {
		tlsConfig = &tls.Config{
			MinVersion: tls.VersionTLS12,
		}
		tlsConfigErr = c.TLS.Client.ApplyTo(tlsConfig)
	}
	return func(ctx context.Context, network, addr string) (net.Conn, error) {
		var dialer interface {
			DialContext(ctx context.Context, network, addr string) (net.Conn, error)
		} = &net.Dialer{}
		if c.TLS.Require {
			if tlsConfigErr != nil {
				return nil, tlsConfigErr
			}
			dialer = &tls.Dialer{NetDialer: dialer.(*net.Dialer), Config: tlsConfig}
		}
		conn, err := dialer.DialContext(ctx, network, addr)
		if err != nil {
			return nil, err
		}
		return &observableConn{addr: addr, Conn: conn}, nil
	}
}
type logFunc func(context.Context, string, ...interface{})
func (f logFunc) Printf(ctx context.Context, format string, v ...interface{}) {
	f(ctx, format, v...)
}
func debugLogFunc(ctx context.Context, format string, v ...interface{}) {
	log.FromContext(ctx).WithField("origin", "go-redis").Debugf(format, v...)
}
func newRedisClient(conf *Config) *redis.Client {
	if conf.Failover.Enable {
		return redis.NewFailoverClient(&redis.FailoverOptions{
			Dialer:        conf.makeDialer(),
			MasterName:    conf.Failover.MasterName,
			SentinelAddrs: conf.Failover.Addresses,
			Password:      conf.Password,
			DB:            conf.Database,
			PoolSize:      conf.PoolSize,
		})
	}
	return redis.NewClient(&redis.Options{
		Dialer:   conf.makeDialer(),
		Addr:     conf.Address,
		Password: conf.Password,
		DB:       conf.Database,
		PoolSize: conf.PoolSize,
	})
}
func New(conf *Config) *Client {
	return &Client{
		namespace: Key(append(conf.RootNamespace, conf.namespace...)...),
		Client:    newRedisClient(conf),
	}
}
func (cl *Client) Key(ks ...string) string {
	return Key(append([]string{cl.namespace}, ks...)...)
}
type ProtoCmd struct {
	result func() (string, error)
}
func (cmd ProtoCmd) ScanProto(pb proto.Message) error {
	s, err := cmd.result()
	if err != nil {
		return ConvertError(err)
	}
	return UnmarshalProto(s, pb)
}
func GetProto(ctx context.Context, r redis.Cmdable, k string) *ProtoCmd {
	return &ProtoCmd{r.Get(ctx, k).Result}
}
func SetProto(ctx context.Context, r redis.Cmdable, k string, pb proto.Message, expiration time.Duration) (*redis.StatusCmd, error) {
	s, err := MarshalProto(pb)
	if err != nil {
		return nil, err
	}
	return r.Set(ctx, k, s, expiration), nil
}
func FindProto(ctx context.Context, r WatchCmdable, k string, keyCmd func(string) (string, error)) *ProtoCmd {
	var result func() (string, error)
	if err := r.Watch(ctx, func(tx *redis.Tx) error {
		id, err := tx.Get(ctx, k).Result()
		if err != nil {
			return err
		}
		ik, err := keyCmd(id)
		if err != nil {
			return err
		}
		result = tx.Get(ctx, ik).Result
		return nil
	}, k); err != nil {
		return &ProtoCmd{result: func() (string, error) { return "", err }}
	}
	return &ProtoCmd{result: result}
}
type stringSliceCmd struct {
	result func() ([]string, error)
}
type ProtosCmd stringSliceCmd
func (cmd ProtosCmd) Range(f func() (proto.Message, func() (bool, error))) error {
	ss, err := cmd.result()
	if err != nil {
		return err
	}
	for _, s := range ss {
		if s == "" {
			continue
		}
		pb, cb := f()
		if pb == nil && cb == nil {
			continue
		}
		if err := UnmarshalProto(s, pb); err != nil {
			return err
		}
		if ok, err := cb(); err != nil {
			return err
		} else if !ok {
			return nil
		}
	}
	return nil
}
type ProtosWithKeysCmd stringSliceCmd
func (cmd ProtosWithKeysCmd) Range(f func(string) (proto.Message, func() (bool, error))) error {
	ss, err := cmd.result()
	if err != nil {
		return err
	}
	if len(ss)%2 != 0 {
		panic(fmt.Sprintf("odd slice length: %d", len(ss)))
	}
	for i := 0; i < len(ss); i += 2 {
		if ss[i+1] == "" {
			continue
		}
		pb, cb := f(ss[i])
		if pb == nil && cb == nil {
			continue
		}
		if err := UnmarshalProto(ss[i+1], pb); err != nil {
			return err
		}
		if ok, err := cb(); err != nil {
			return err
		} else if !ok {
			return nil
		}
	}
	return nil
}
type redisSort struct {
	*redis.Sort
}
type FindProtosOption func(redisSort)
func FindProtosSorted(alpha bool) FindProtosOption {
	return func(s redisSort) {
		s.Alpha = alpha
		s.By = ""
	}
}
func FindProtosWithOffsetAndCount(offset, count int64) FindProtosOption {
	return func(s redisSort) {
		s.Offset, s.Count = offset, count
	}
}
func findProtos(ctx context.Context, r redis.Cmdable, k string, keyCmd func(string) string, opts ...FindProtosOption) stringSliceCmd {
	s := &redis.Sort{
		Get: []string{keyCmd("*")},
		By:  "nosort",  
	}
	for _, opt := range opts {
		opt(redisSort{s})
	}
	return stringSliceCmd{
		result: r.Sort(ctx, k, s).Result,
	}
}
func FindProtos(ctx context.Context, r redis.Cmdable, k string, keyCmd func(string) string, opts ...FindProtosOption) ProtosCmd {
	return ProtosCmd(findProtos(ctx, r, k, keyCmd, opts...))
}
func FindProtosWithKeys(ctx context.Context, r redis.Cmdable, k string, keyCmd func(string) string, opts ...FindProtosOption) ProtosWithKeysCmd {
	return ProtosWithKeysCmd(findProtos(ctx, r, k, keyCmd, append([]FindProtosOption{func(s redisSort) { s.Get = append([]string{"#"}, s.Get...) }}, opts...)...))
}
func ListProtos(ctx context.Context, r redis.Cmdable, k string) ProtosCmd {
	return ProtosCmd{
		result: r.LRange(ctx, k, 0, -1).Result,
	}
}
type InterfaceSliceCmd struct {
	*redis.Cmd
}
func (cmd InterfaceSliceCmd) Result() ([]interface{}, error) {
	v, err := cmd.Cmd.Result()
	if err != nil {
		return nil, err
	}
	vs, ok := v.([]interface{})
	if !ok {
		return nil, errDecode.New()
	}
	return vs, nil
}
func RunInterfaceSliceScript(ctx context.Context, r Scripter, s *redis.Script, keys []string, args ...interface{}) *InterfaceSliceCmd {
	return &InterfaceSliceCmd{s.Run(ctx, r, keys, args...)}
}
const (
	payloadKey = "payload"
	replaceKey = "replace"
	startAtKey = "start_at"
	nextAtKey  = "next_at"
	lastIDKey  = "last_id"
)
func InputTaskKey(k string) string {
	return Key(k, "input")
}
func ReadyTaskKey(k string) string {
	return Key(k, "ready")
}
func WaitingTaskKey(k string) string {
	return Key(k, "waiting")
}
func IsConsumerGroupExistsErr(err error) bool {
	return err != nil && err.Error() == "BUSYGROUP Consumer Group name already exists"
}
func initTaskGroup(ctx context.Context, r redis.Cmdable, group, k string) error {
	_, err := r.Pipelined(ctx, func(p redis.Pipeliner) error {
		p.XGroupCreateMkStream(ctx, InputTaskKey(k), group, "$")
		p.XGroupCreateMkStream(ctx, ReadyTaskKey(k), group, "$")
		return nil
	})
	if IsConsumerGroupExistsErr(err) {
		return nil
	}
	return ConvertError(err)
}
func addTask(ctx context.Context, r redis.Cmdable, k string, maxLen int64, payload string, startAt time.Time, replace bool) error {
	m := make(map[string]interface{}, 2)
	m[payloadKey] = payload
	if replace {
		m[replaceKey] = replace
	}
	if !startAt.IsZero() {
		m[startAtKey] = startAt.UnixNano()
	}
	return ConvertError(r.XAdd(ctx, &redis.XAddArgs{
		Stream:       InputTaskKey(k),
		MaxLenApprox: maxLen,
		Values:       m,
	}).Err())
}
func parseTime(s string) (time.Time, error) {
	nsec, err := strconv.ParseFloat(s, 64)
	if err != nil {
		return time.Time{}, err
	}
	return time.Unix(0, int64(nsec)).UTC(), nil
}