How to Migrate Railway Redis to Amazon ElastiCache

// cart/cart.go
package cart

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	"encore.dev/beta/errs"
	"encore.dev/storage/cache"
	"github.com/redis/go-redis/v9"
)

// ElastiCache on AWS, an in-memory Redis with `encore run`.
var cluster = cache.NewCluster("carts", cache.ClusterConfig{
	EvictionPolicy: cache.VolatileLRU,
})

type Item struct {
	SKU string `json:"sku"`
	Qty int    `json:"qty"`
}

type Cart struct {
	Items     []Item    `json:"items"`
	UpdatedAt time.Time `json:"updated_at"`
}

const cartTTL = 30 * 24 * time.Hour

var Carts = cache.NewStructKeyspace[string, Cart](cluster, cache.KeyspaceConfig{
	KeyPattern:    "cart/:key",
	DefaultExpiry: cache.ExpireIn(cartTTL),
})

var secrets struct {
	// REDIS_PUBLIC_URL of the Railway Redis service. Remove this field and
	// the fallback in load once the migration window has passed.
	RailwayRedisURL string
}

//encore:service
type Service struct {
	railway *redis.Client
}

func initService() (*Service, error) {
	opt, err := redis.ParseURL(secrets.RailwayRedisURL)
	if err != nil {
		return nil, err
	}
	return &Service{railway: redis.NewClient(opt)}, nil
}

func (s *Service) Shutdown(force context.Context) {
	_ = s.railway.Close()
}

// load returns the user's cart from ElastiCache, moving it over from
// Railway Redis the first time it is read after the switch.
func (s *Service) load(ctx context.Context, userID string) (Cart, error) {
	c, err := Carts.Get(ctx, userID)
	if err == nil || !errors.Is(err, cache.Miss) {
		return c, err
	}

	oldKey := "cart:" + userID
	raw, err := s.railway.Get(ctx, oldKey).Bytes()
	if errors.Is(err, redis.Nil) {
		return Cart{}, nil // no cart in either place
	} else if err != nil {
		return Cart{}, err
	}
	if err := json.Unmarshal(raw, &c); err != nil {
		return Cart{}, err
	}
	ttl, err := s.railway.PTTL(ctx, oldKey).Result()
	if err != nil {
		return Cart{}, err
	}
	if ttl == -2 {
		return Cart{}, nil // expired between GET and PTTL
	}
	if ttl == -1 {
		ttl = cartTTL // no expiry on the old key
	}

	// SetIfNotExists lets one of two concurrent requests win the copy.
	err = Carts.With(cache.ExpireIn(ttl)).SetIfNotExists(ctx, userID, c)
	if errors.Is(err, cache.KeyExists) {
		return Carts.Get(ctx, userID)
	} else if err != nil {
		return Cart{}, err
	}
	// Delete the Railway key only if it still holds the snapshot we copied,
	// so a concurrent cart update is not discarded.
	kept, err := s.railway.Eval(ctx, `
		if redis.call("GET", KEYS[1]) == ARGV[1] then
			return redis.call("DEL", KEYS[1])
		end
		return 0
	`, []string{oldKey}, raw).Int()
	if err != nil {
		return Cart{}, err
	}
	if kept == 0 {
		// The Railway key changed. Keep a destination value only when it
		// already differs from the snapshot we wrote. A still-matching
		// copy is stale, so refresh from the current source value.
		dest, err := Carts.Get(ctx, userID)
		if err == nil && dest != c {
			return dest, nil
		}
		if err != nil && !errors.Is(err, cache.Miss) {
			return Cart{}, err
		}
		raw, err := s.railway.Get(ctx, oldKey).Bytes()
		if errors.Is(err, redis.Nil) {
			return Cart{}, nil
		} else if err != nil {
			return Cart{}, err
		}
		var newer Cart
		if err := json.Unmarshal(raw, &newer); err != nil {
			return Cart{}, err
		}
		ttl, err := s.railway.PTTL(ctx, oldKey).Result()
		if err != nil {
			return Cart{}, err
		}
		if ttl == -2 {
			return Cart{}, nil
		}
		if ttl == -1 {
			ttl = cartTTL
		}
		// Re-check the destination immediately before writing, so a cart
		// update that landed after the first Get is not overwritten.
		dest, err = Carts.Get(ctx, userID)
		if err == nil && dest != c {
			return dest, nil
		}
		if err != nil && !errors.Is(err, cache.Miss) {
			return Cart{}, err
		}
		if errors.Is(err, cache.Miss) {
			err = Carts.With(cache.ExpireIn(ttl)).SetIfNotExists(ctx, userID, newer)
			if errors.Is(err, cache.KeyExists) {
				return Carts.Get(ctx, userID)
			} else if err != nil {
				return Cart{}, err
			}
		} else {
			// Destination still holds the stale snapshot. Replace it only
			// after the re-check above; a later Get is the source of truth
			// if another writer won the race.
			if err := Carts.With(cache.ExpireIn(ttl)).Set(ctx, userID, newer); err != nil {
				return Cart{}, err
			}
			dest, err = Carts.Get(ctx, userID)
			if err == nil && dest != newer {
				return dest, nil
			} else if err != nil {
				return Cart{}, err
			}
		}
		kept, err = s.railway.Eval(ctx, `
			if redis.call("GET", KEYS[1]) == ARGV[1] then
				return redis.call("DEL", KEYS[1])
			end
			return 0
		`, []string{oldKey}, raw).Int()
		if err != nil {
			return Cart{}, err
		}
		if kept == 0 {
			dest, err = Carts.Get(ctx, userID)
			if err == nil && dest != newer {
				return dest, nil
			} else if err != nil && !errors.Is(err, cache.Miss) {
				return Cart{}, err
			}
			return Cart{}, errors.New("source changed during copy, retry")
		}
		return newer, nil
	}
	return c, nil
}

type GetParams struct {
	UserID string `query:"user_id"`
}

// Get is called by the storefront service, never directly by browsers.
//
//encore:api private method=GET path=/cart
func (s *Service) Get(ctx context.Context, p *GetParams) (*Cart, error) {
	c, err := s.load(ctx, p.UserID)
	return &c, err
}

type AddParams struct {
	UserID string `json:"user_id"`
	SKU    string `json:"sku"`
	Qty    int    `json:"qty"`
}

//encore:api private method=POST path=/cart/items
func (s *Service) Add(ctx context.Context, p *AddParams) (*Cart, error) {
	if p.UserID == "" || p.SKU == "" || p.Qty < 1 || p.Qty > 99 {
		return nil, &errs.Error{Code: errs.InvalidArgument, Message: "user_id, sku and qty 1-99 are required"}
	}
	c, err := s.load(ctx, p.UserID)
	if err != nil {
		return nil, err
	}
	merged := false
	for i := range c.Items {
		if c.Items[i].SKU == p.SKU {
			c.Items[i].Qty += p.Qty
			merged = true
		}
	}
	if !merged {
		c.Items = append(c.Items, Item{SKU: p.SKU, Qty: p.Qty})
	}
	c.UpdatedAt = time.Now()
	if err := Carts.Set(ctx, p.UserID, c); err != nil {
		return nil, err
	}
	return &c, nil
}