summercms/modules/pier/store_redis.go

package pier

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

// Redis key and channel layout under cooler's application prefix.
const (
	redisSessionKey = "mcp:s:"     // <prefix>mcp:s:<id> holds the JSON record
	redisBindingKey = "mcp:b:"     // <prefix>mcp:b:<binding> is the set of its session ids
	redisEventsKey  = "mcp:events" // <prefix>mcp:events carries JSON events
)

// redisStore keeps session records in Redis with PX TTLs and fans events
// out through pub/sub, so several instances share sessions and react to
// each other's events.
type redisStore struct {
	client *redis.Client
	prefix string
}

func newRedisStore(client *redis.Client, prefix string) *redisStore {
	return &redisStore{client: client, prefix: prefix}
}

func (s *redisStore) sessionKey(id string) string { return s.prefix + redisSessionKey + id }

func (s *redisStore) bindingKey(binding string) string { return s.prefix + redisBindingKey + binding }

func (s *redisStore) channel() string { return s.prefix + redisEventsKey }

func (s *redisStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error {
	raw, err := json.Marshal(rec)
	if err != nil {
		return err
	}
	pipe := s.client.TxPipeline()
	pipe.Set(ctx, s.sessionKey(id), raw, ttl)
	pipe.SAdd(ctx, s.bindingKey(rec.Binding), id)
	pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl)
	_, err = pipe.Exec(ctx)
	return err
}

// Update rewrites an existing session record (SET XX) and refreshes the
// binding set; a record that is gone stays gone.
func (s *redisStore) Update(ctx context.Context, id string, rec Record, ttl time.Duration) (bool, error) {
	raw, err := json.Marshal(rec)
	if err != nil {
		return false, err
	}
	err = s.client.SetArgs(ctx, s.sessionKey(id), raw, redis.SetArgs{Mode: "XX", TTL: ttl}).Err()
	if errors.Is(err, redis.Nil) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	pipe := s.client.TxPipeline()
	pipe.SAdd(ctx, s.bindingKey(rec.Binding), id)
	pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl)
	_, err = pipe.Exec(ctx)
	return true, err
}

func (s *redisStore) Get(ctx context.Context, id string) (Record, bool, error) {
	raw, err := s.client.Get(ctx, s.sessionKey(id)).Bytes()
	if errors.Is(err, redis.Nil) {
		return Record{}, false, nil
	}
	if err != nil {
		return Record{}, false, err
	}
	var rec Record
	if err := json.Unmarshal(raw, &rec); err != nil {
		return Record{}, false, err
	}
	return rec, true, nil
}

// Touch slides the record's TTL and its binding set's with it.
func (s *redisStore) Touch(ctx context.Context, id string, ttl time.Duration) error {
	rec, ok, err := s.Get(ctx, id)
	if err != nil || !ok {
		return err
	}
	pipe := s.client.TxPipeline()
	pipe.PExpire(ctx, s.sessionKey(id), ttl)
	pipe.PExpire(ctx, s.bindingKey(rec.Binding), ttl)
	_, err = pipe.Exec(ctx)
	return err
}

func (s *redisStore) Delete(ctx context.Context, id string) error {
	rec, ok, err := s.Get(ctx, id)
	if err != nil {
		return err
	}
	pipe := s.client.TxPipeline()
	pipe.Del(ctx, s.sessionKey(id))
	if ok {
		pipe.SRem(ctx, s.bindingKey(rec.Binding), id)
	}
	_, err = pipe.Exec(ctx)
	return err
}

// Count answers the live sessions of a binding, pruning members whose
// record has expired from the set.
func (s *redisStore) Count(ctx context.Context, binding string) (int, error) {
	members, err := s.client.SMembers(ctx, s.bindingKey(binding)).Result()
	if err != nil {
		return 0, err
	}
	n := 0
	for _, id := range members {
		exists, err := s.client.Exists(ctx, s.sessionKey(id)).Result()
		if err != nil {
			return 0, err
		}
		if exists > 0 {
			n++
			continue
		}
		_ = s.client.SRem(ctx, s.bindingKey(binding), id).Err()
	}
	return n, nil
}

func (s *redisStore) Publish(ctx context.Context, ev Event) error {
	raw, err := json.Marshal(ev)
	if err != nil {
		return err
	}
	return s.client.Publish(ctx, s.channel(), raw).Err()
}

// Subscribe delivers every event published on the channel to fn, on its
// own goroutine, until ctx ends.
func (s *redisStore) Subscribe(ctx context.Context, fn func(Event)) error {
	sub := s.client.Subscribe(ctx, s.channel())
	if _, err := sub.Receive(ctx); err != nil {
		_ = sub.Close()
		return err
	}
	go func() {
		defer sub.Close()
		ch := sub.Channel()
		for {
			select {
			case <-ctx.Done():
				return
			case msg, ok := <-ch:
				if !ok {
					return
				}
				var ev Event
				if err := json.Unmarshal([]byte(msg.Payload), &ev); err != nil {
					continue
				}
				fn(ev)
			}
		}
	}()
	return nil
}

// lazyStore opens the Redis store on first use, so nothing dials at mount
// (route:list and tests never dial) and a missing redis.url is a 503 on the
// first request, logged once, not a boot failure (cooler's rule).
type lazyStore struct {
	open   func(ctx context.Context) (Store, error)
	mu     sync.Mutex
	store  Store
	warned atomic.Bool
}

// get returns the store, opening it on the first call and retrying on a
// later call after a failed open.
func (l *lazyStore) get(ctx context.Context) (Store, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.store != nil {
		return l.store, nil
	}
	s, err := l.open(ctx)
	if err != nil {
		if !l.warned.Swap(true) {
			logger().Error("pier: session store could not be opened", "error", err)
		}
		return nil, err
	}
	l.store = s
	return s, nil
}

func (l *lazyStore) Put(ctx context.Context, id string, rec Record, ttl time.Duration) error {
	s, err := l.get(ctx)
	if err != nil {
		return err
	}
	return s.Put(ctx, id, rec, ttl)
}

func (l *lazyStore) Get(ctx context.Context, id string) (Record, bool, error) {
	s, err := l.get(ctx)
	if err != nil {
		return Record{}, false, err
	}
	return s.Get(ctx, id)
}

// Update forwards to the opened store's Update (Put for a store without one).
func (l *lazyStore) Update(ctx context.Context, id string, rec Record, ttl time.Duration) (bool, error) {
	s, err := l.get(ctx)
	if err != nil {
		return false, err
	}
	if u, ok := s.(updater); ok {
		return u.Update(ctx, id, rec, ttl)
	}
	return true, s.Put(ctx, id, rec, ttl)
}

func (l *lazyStore) Touch(ctx context.Context, id string, ttl time.Duration) error {
	s, err := l.get(ctx)
	if err != nil {
		return err
	}
	return s.Touch(ctx, id, ttl)
}

func (l *lazyStore) Delete(ctx context.Context, id string) error {
	s, err := l.get(ctx)
	if err != nil {
		return err
	}
	return s.Delete(ctx, id)
}

func (l *lazyStore) Count(ctx context.Context, binding string) (int, error) {
	s, err := l.get(ctx)
	if err != nil {
		return 0, err
	}
	return s.Count(ctx, binding)
}

func (l *lazyStore) Publish(ctx context.Context, ev Event) error {
	s, err := l.get(ctx)
	if err != nil {
		return err
	}
	return s.Publish(ctx, ev)
}

func (l *lazyStore) Subscribe(ctx context.Context, fn func(Event)) error {
	s, err := l.get(ctx)
	if err != nil {
		return err
	}
	return s.Subscribe(ctx, fn)
}

// redisURLError names the key a redis store needs.
func redisURLError(url string) error {
	if strings.TrimSpace(url) == "" {
		return fmt.Errorf("pier: redis.url is required when %s is %s (set it in config/redis.yaml or SUMMER_REDIS__URL)", SessionStoreKey, StoreRedis)
	}
	return nil
}