func (q *TaskQueue) Init(ctx context.Context) error {
	return initTaskGroup(ctx, q.Redis, q.Group, q.Key)
}
Init initializes the task queue.
It must be called at least once before using the queue.
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
}
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
	}
}
type TaskQueue struct {
	Redis  WatchCmdable
	MaxLen int64
	Group  string
	Key    string
	consumerIDs sync.Map  
}