mirror of
https://github.com/TecharoHQ/anubis.git
synced 2026-04-14 20:48:44 +00:00
feat(lib/store): implement valkey backend
Signed-off-by: Xe Iaso <me@xeiaso.net>
This commit is contained in:
@@ -6,4 +6,5 @@ package all
|
||||
import (
|
||||
_ "github.com/TecharoHQ/anubis/lib/store/bbolt"
|
||||
_ "github.com/TecharoHQ/anubis/lib/store/memory"
|
||||
_ "github.com/TecharoHQ/anubis/lib/store/valkey"
|
||||
)
|
||||
|
||||
84
lib/store/valkey/factory.go
Normal file
84
lib/store/valkey/factory.go
Normal file
@@ -0,0 +1,84 @@
|
||||
package valkey
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/TecharoHQ/anubis/lib/store"
|
||||
valkey "github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNoURL = errors.New("valkey.Config: no URL defined")
|
||||
ErrBadURL = errors.New("valkey.Config: URL is invalid")
|
||||
)
|
||||
|
||||
func init() {
|
||||
store.Register("valkey", Factory{})
|
||||
}
|
||||
|
||||
type Factory struct{}
|
||||
|
||||
func (Factory) Build(ctx context.Context, data json.RawMessage) (store.Interface, error) {
|
||||
var config Config
|
||||
|
||||
if err := json.Unmarshal([]byte(data), &config); err != nil {
|
||||
return nil, fmt.Errorf("%w: %w", store.ErrBadConfig, err)
|
||||
}
|
||||
|
||||
if err := config.Valid(); err != nil {
|
||||
return nil, fmt.Errorf("%w: %w", store.ErrBadConfig, err)
|
||||
}
|
||||
|
||||
opts, err := valkey.ParseURL(config.URL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: %w", store.ErrBadConfig, err)
|
||||
}
|
||||
|
||||
rdb := valkey.NewClient(opts)
|
||||
|
||||
if _, err := rdb.Ping(ctx).Result(); err != nil {
|
||||
return nil, fmt.Errorf("can't ping valkey instance: %w", err)
|
||||
}
|
||||
|
||||
return &Store{
|
||||
rdb: rdb,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (Factory) Valid(data json.RawMessage) error {
|
||||
var config Config
|
||||
if err := json.Unmarshal([]byte(data), &config); err != nil {
|
||||
return fmt.Errorf("%w: %w", store.ErrBadConfig, err)
|
||||
}
|
||||
|
||||
if err := config.Valid(); err != nil {
|
||||
return fmt.Errorf("%w: %w", store.ErrBadConfig, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
URL string `json:"url"`
|
||||
}
|
||||
|
||||
func (c Config) Valid() error {
|
||||
var errs []error
|
||||
|
||||
if c.URL == "" {
|
||||
errs = append(errs, ErrNoURL)
|
||||
}
|
||||
|
||||
if _, err := valkey.ParseURL(c.URL); err != nil {
|
||||
errs = append(errs, ErrBadURL)
|
||||
}
|
||||
|
||||
if len(errs) != 0 {
|
||||
return fmt.Errorf("valkey.Config: invalid config: %w", errors.Join(errs...))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
49
lib/store/valkey/valkey.go
Normal file
49
lib/store/valkey/valkey.go
Normal file
@@ -0,0 +1,49 @@
|
||||
package valkey
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/TecharoHQ/anubis/lib/store"
|
||||
valkey "github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type Store struct {
|
||||
rdb *valkey.Client
|
||||
}
|
||||
|
||||
func (s *Store) Delete(ctx context.Context, key string) error {
|
||||
n, err := s.rdb.Del(ctx, key).Result()
|
||||
if err != nil {
|
||||
return fmt.Errorf("can't delete from valkey: %w", err)
|
||||
}
|
||||
|
||||
switch n {
|
||||
case 0:
|
||||
return fmt.Errorf("%w: %d key(s) deleted", store.ErrNotFound, n)
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Store) Get(ctx context.Context, key string) ([]byte, error) {
|
||||
result, err := s.rdb.Get(ctx, key).Result()
|
||||
if err != nil {
|
||||
if valkey.HasErrorPrefix(err, "redis: nil") {
|
||||
return nil, fmt.Errorf("%w: %w", store.ErrNotFound, err)
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("can't fetch from valkey: %w", err)
|
||||
}
|
||||
|
||||
return []byte(result), nil
|
||||
}
|
||||
|
||||
func (s *Store) Set(ctx context.Context, key string, value []byte, expiry time.Duration) error {
|
||||
if _, err := s.rdb.Set(ctx, key, string(value), expiry).Result(); err != nil {
|
||||
return fmt.Errorf("can't set %q in valkey: %w", key, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
61
lib/store/valkey/valkey_test.go
Normal file
61
lib/store/valkey/valkey_test.go
Normal file
@@ -0,0 +1,61 @@
|
||||
package valkey
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"testing"
|
||||
|
||||
"github.com/TecharoHQ/anubis/lib/store/storetest"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
"github.com/testcontainers/testcontainers-go/wait"
|
||||
)
|
||||
|
||||
func TestImpl(t *testing.T) {
|
||||
if os.Getenv("DONT_USE_NETWORK") != "" {
|
||||
t.Skip("test requires network egress")
|
||||
return
|
||||
}
|
||||
|
||||
req := testcontainers.ContainerRequest{
|
||||
Image: "valkey/valkey:8",
|
||||
WaitingFor: wait.ForLog("Ready to accept connections"),
|
||||
}
|
||||
valkeyC, err := testcontainers.GenericContainer(t.Context(), testcontainers.GenericContainerRequest{
|
||||
ContainerRequest: req,
|
||||
Started: true,
|
||||
})
|
||||
testcontainers.CleanupContainer(t, valkeyC)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
containerIP, err := valkeyC.ContainerIP(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// XXX(Xe): This is bad code. Do not do this.
|
||||
//
|
||||
// I have to do this because I'm running from inside the context of a dev
|
||||
// container. This dev container runs in a different docker network than
|
||||
// the valkey test container runs in. In order to let my dev container
|
||||
// connect to the test container, they need to share a network in common.
|
||||
// The easiest network to use for this is the default "bridge" network.
|
||||
//
|
||||
// This is a horrifying monstrosity, but the part that scares me the most
|
||||
// is the fact that it works.
|
||||
if hostname, err := os.Hostname(); err == nil {
|
||||
exec.Command("docker", "network", "connect", "bridge", hostname).Run()
|
||||
}
|
||||
|
||||
data, err := json.Marshal(Config{
|
||||
URL: fmt.Sprintf("redis://%s:6379/0", containerIP),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
storetest.Common(t, Factory{}, json.RawMessage(data))
|
||||
}
|
||||
Reference in New Issue
Block a user