func (ps *PubSubStore) FindRelated(ctx context.Context, correlationID string) ([]events.Event, error) {
	uids, err := ps.client.LRange(ctx, ps.eventIndexKey(ctx, correlationID), 0, -1).Result()
	if err != nil {
		return nil, err
	}
	evtPBs := make([]*ttnpb.Event, len(uids))
	for i, uid := range uids {
		evtPBs[i] = &ttnpb.Event{UniqueId: uid}
	}
	if err = ps.loadEventData(ctx, evtPBs...); err != nil {
		return nil, err
	}
	evts := make([]events.Event, len(evtPBs))
	for i, evtPB := range evtPBs {
		if evtPB.Name == "" {
			continue
		}
		evts[i], err = events.FromProto(evtPB)
		if err != nil {
			return nil, err
		}
	}
	return evts, nil
}
FindRelated finds events with matching correlation IDs.
package redis
import (
	"context"
	"crypto/sha1"
	"encoding/base32"
	"strconv"
	"strings"
	"time"
	"github.com/go-redis/redis/v8"
	"go.thethings.network/lorawan-stack/v3/pkg/events"
	"go.thethings.network/lorawan-stack/v3/pkg/log"
	ttnredis "go.thethings.network/lorawan-stack/v3/pkg/redis"
	"go.thethings.network/lorawan-stack/v3/pkg/ttnpb"
	"go.thethings.network/lorawan-stack/v3/pkg/unique"
)
type PubSubStore struct {
	*PubSub
	historyTTL                time.Duration
	entityHistoryCount        int
	entityHistoryTTL          time.Duration
	correlationIDHistoryCount int
}
func (ps *PubSubStore) eventDataKey(_ context.Context, uid string) string {
	return ps.client.Key("event_data", uid)
}
func (ps *PubSubStore) eventIndexKey(_ context.Context, correlationID string) string {
	h := sha1.New()
	h.Write([]byte(correlationID))
	return ps.client.Key("event_index", base32.StdEncoding.EncodeToString(h.Sum(nil)))
}
func (ps *PubSubStore) eventStream(ctx context.Context, ids *ttnpb.EntityIdentifiers) string {
	if ids == nil {
		return ps.client.Key("event_stream")
	}
	return ps.client.Key("event_stream", ids.EntityType(), unique.ID(ctx, ids))
}
func (ps *PubSubStore) storeEvent(ctx context.Context, evt events.Event) error {
	b, err := encodeEventData(evt)
	if err != nil {
		return err
	}
	_, err = ps.client.TxPipelined(ctx, func(tx redis.Pipeliner) error {
		tx.Set(ps.ctx, ps.eventDataKey(evt.Context(), evt.UniqueID()), b, ps.historyTTL)
		for _, cid := range evt.CorrelationIDs() {
			key := ps.eventIndexKey(evt.Context(), cid)
			tx.LPush(ps.ctx, key, evt.UniqueID())
			tx.LTrim(ps.ctx, key, 0, int64(ps.correlationIDHistoryCount))
			tx.Expire(ps.ctx, key, ps.historyTTL)
		}
		return nil
	})
	return ttnredis.ConvertError(err)
}
func (ps *PubSubStore) loadEventData(ctx context.Context, evts ...*ttnpb.Event) error {
	if len(evts) == 0 {
		return nil
	}
	keys := make([]string, 0, len(evts))
	for _, evt := range evts {
		keys = append(keys, ps.eventDataKey(ctx, evt.UniqueId))
	}
	bs, err := ps.client.MGet(ctx, keys...).Result()
	if err != nil {
		return ttnredis.ConvertError(err)
	}
	for i, b := range bs {
		switch b := b.(type) {
		case nil:
			continue  
		case string:
			if err = decodeEventData(b, evts[i]); err != nil {
				log.FromContext(ctx).WithError(err).Warnf("Failed to decode event payload")
				continue
			}
		default:
			log.FromContext(ctx).Warnf("Invalid %T element in event payloads", b)
			continue
		}
	}
	return nil
}
func (ps *PubSubStore) LoadEvent(ctx context.Context, uid string) (*ttnpb.Event, error) {
	data, err := ps.client.Get(ctx, ps.eventDataKey(ctx, uid)).Result()
	if err != nil {
		return nil, ttnredis.ConvertError(err)
	}
	var evtPB ttnpb.Event
	if err = decodeEventData(data, &evtPB); err != nil {
		return nil, err
	}
	return &evtPB, nil
}
func xMessageHasEventName(names ...string) func(msg redis.XMessage) bool {
	if len(names) == 0 {
		return func(msg redis.XMessage) bool {
			return true
		}
	}
	nameMap := make(map[string]struct{}, len(names))
	for _, name := range names {
		nameMap[name] = struct{}{}
	}
	return func(msg redis.XMessage) bool {
		name, ok := msg.Values[eventNameKey].(string)
		if !ok {
			return false
		}
		_, match := nameMap[name]
		return match
	}
}
func eventsFromXMessages(msgs []redis.XMessage, match func(msg redis.XMessage) bool) []*ttnpb.Event {
	evts := make([]*ttnpb.Event, 0, len(msgs))
	for _, msg := range msgs {
		if match != nil && !match(msg) {
			continue
		}
		if evt, err := decodeEventMeta(msg.Values); err == nil {
			evts = append(evts, evt)
		}
	}
	return evts
}
func reverseEvents(slice []*ttnpb.Event) {
	for i := 0; i < len(slice)/2; i++ {
		j := len(slice) - i - 1
		slice[i], slice[j] = slice[j], slice[i]
	}
}
func (ps *PubSubStore) tailStream(ctx context.Context, names []string, ids *ttnpb.EntityIdentifiers, start string, tail int) (events []*ttnpb.Event, nextStart string, err error) {
	matchNames := xMessageHasEventName(names...)
	if tail > 0 {
		msgs, err := ps.client.XRevRangeN(ctx, ps.eventStream(ctx, ids), "+", start, int64(tail)).Result()
		if err != nil {
			return nil, "", ttnredis.ConvertError(err)
		}
		if len(msgs) == 0 {
			return nil, start, nil
		}
		events = eventsFromXMessages(msgs, matchNames)
		reverseEvents(events)
		nextStart = msgs[0].ID
	} else {
		msgs, err := ps.client.XRange(ctx, ps.eventStream(ctx, ids), start, "+").Result()
		if err != nil {
			return nil, "", ttnredis.ConvertError(err)
		}
		if len(msgs) == 0 {
			return nil, start, nil
		}
		events = eventsFromXMessages(msgs, matchNames)
		nextStart = msgs[len(msgs)-1].ID
	}
	if err = ps.loadEventData(ctx, events...); err != nil {
		return nil, "", err
	}
	return events, nextStart, nil
}
func (ps *PubSubStore) FetchHistory(ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, after *time.Time, tail int) ([]events.Event, error) {
	start := "-"
	switch {
	case after == nil:
	case after.IsZero():
		after = nil
	default:
		afterMS := after.Truncate(time.Millisecond)
		after = &afterMS
		start = strconv.FormatInt(after.Add(-1*time.Second).UnixNano()/1_000_000, 10)
	}
	var evts []events.Event
	for _, id := range ids {
		evtPBs, _, err := ps.tailStream(ctx, names, id, start, tail)
		if err != nil {
			return nil, err
		}
		for _, evtPB := range evtPBs {
			if after != nil && !evtPB.GetTime().Truncate(time.Millisecond).After(*after) {
				continue
			}
			evt, err := events.FromProto(evtPB)
			if err != nil {
				return nil, err
			}
			evts = append(evts, evt)
		}
	}
	return evts, nil
}
func (ps *PubSubStore) SubscribeWithHistory(ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, after *time.Time, tail int, hdl events.Handler) error {
	start := "0-0"
	switch {
	case after == nil:
	case after.IsZero():
		after = nil
	default:
		afterMS := after.Truncate(time.Millisecond)
		after = &afterMS
		start = strconv.FormatInt(after.Add(-1*time.Second).UnixNano()/1_000_000, 10)
	}
	type streamState struct {
		id     *ttnpb.EntityIdentifiers
		stream string
		start  string
	}
	state := make([]*streamState, len(ids))
	for i, id := range ids {
		state[i] = &streamState{
			id:     id,
			stream: ps.eventStream(ctx, id),
			start:  start,
		}
	}
	getState := func(stream string) *streamState {
		for _, state := range state {
			if state.stream == stream {
				return state
			}
		}
		return nil
	}
	for _, s := range state {
		if ctx.Err() != nil {
			return ctx.Err()
		}
		evtPBs, nextStart, err := ps.tailStream(ctx, names, s.id, s.start, tail)
		if err != nil {
			return err
		}
		for _, evtPB := range evtPBs {
			if after != nil && !evtPB.GetTime().Truncate(time.Millisecond).After(*after) {
				continue
			}
			evt, err := events.FromProto(evtPB)
			if err != nil {
				return err
			}
			hdl.Notify(evt)
		}
		s.start = nextStart
	}
	matchNames := xMessageHasEventName(names...)
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			args := &redis.XReadArgs{
				Count: 10,
				Block: time.Second,
			}
			for _, s := range state {
				args.Streams = append(args.Streams, s.stream)
			}
			for _, s := range state {
				args.Streams = append(args.Streams, s.start)
			}
			streams, err := ps.client.XRead(ctx, args).Result()
			if err != nil && err != redis.Nil {
				return ttnredis.ConvertError(err)
			}
			for _, stream := range streams {
				if len(stream.Messages) == 0 {
					continue
				}
				state := getState(stream.Stream)
				if state == nil {
					continue
				}
				state.start = stream.Messages[len(stream.Messages)-1].ID
				evtPBs := eventsFromXMessages(stream.Messages, matchNames)
				if len(evtPBs) == 0 {
					continue
				}
				if err = ps.loadEventData(ctx, evtPBs...); err != nil {
					return err
				}
				for _, evtPB := range evtPBs {
					evt, err := events.FromProto(evtPB)
					if err != nil {
						return err
					}
					hdl.Notify(evt)
				}
			}
		}
	}
}