package events
import (
"context"
"crypto/tls"
"fmt"
"strconv"
"time"
"alin.ovh/x/log"
"github.com/Southclaws/fault"
"github.com/Southclaws/fault/fmsg"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
)
type RedisOptions struct {
Address string
Username string `conf:"default:default"`
Password string `conf:"default:default"`
TLSEnabled bool `conf:"default:true"`
}
type Logger struct {
log *zap.SugaredLogger
}
type RedisListener struct {
client *redis.Client
log *log.Logger
}
const (
db = 0
key = "run_id"
)
func (l *Logger) Printf(_ context.Context, format string, v ...any) {
l.log.Infof(format, v...)
}
func NewRedisListener(opts *RedisOptions, log *log.Logger) (*RedisListener, error) {
clientConfig := &redis.Options{
Addr: opts.Address,
Username: opts.Username,
Password: opts.Password,
}
if opts.TLSEnabled {
clientConfig.TLSConfig = &tls.Config{
InsecureSkipVerify: false,
MinVersion: tls.VersionTLS13,
}
}
redis.SetLogger(&Logger{log: log.GetLogger().Sugar()})
client := redis.NewClient(clientConfig)
log.Debug(
"connecting to redis",
"address",
opts.Address,
"username",
opts.Username,
"tls",
opts.TLSEnabled,
)
return &RedisListener{
client: client,
log: log,
}, nil
}
func (rl *RedisListener) GetLatestRunID() (uint64, error) {
ctx, cancel := context.WithTimeout(context.TODO(), 2*time.Second)
defer cancel()
payload, err := rl.client.Get(ctx, key).Result()
if err != nil {
return 1, fault.Wrap(err, fmsg.With("could not get latest run ID"))
}
runID, err := strconv.ParseUint(payload, 10, 64)
if err != nil {
return 1, fault.Wrap(err, fmsg.With("could not parse latest run ID"))
}
rl.log.Debug("redis response", "payload", payload, "run_id", runID)
return runID, nil
}
// requires `redis-cli config set notify-keyspace-events KEA`
func getKeyspaceName(key string) string {
return fmt.Sprintf("__keyspace@%d__:%s", db, key)
}
func (rl *RedisListener) Subscribe() (<-chan Event, error) {
events := make(chan Event, 1)
ctx := context.TODO()
channel := getKeyspaceName(key)
go func(ch <-chan *redis.Message) {
for msg := range ch {
rl.log.Debug("got event", "payload", msg.Payload)
runID, err := rl.GetLatestRunID()
if err != nil {
rl.log.Warn("could not get latest run ID")
}
events <- Event{
CIEvent: CIEvent{
RunID: runID,
},
}
}
}(rl.client.Subscribe(ctx, channel).Channel())
return events, nil
}
shared/events/redis.go (view raw)