feat: initial commit

internal/store/redis/helper.go

@@ -0,0 +1,66 @@
+package redis
+
+import (
+	"encoding/json"
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+type jsonWrapper[T any] struct {
+	value T
+}
+
+func (w *jsonWrapper[T]) MarshalBinary() ([]byte, error) {
+	data, err := json.Marshal(w.value)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return data, nil
+}
+
+func (w *jsonWrapper[T]) UnmarshalBinary(data []byte) error {
+	if err := json.Unmarshal(data, &w.value); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
+func (w *jsonWrapper[T]) Value() T {
+	return w.value
+}
+
+func wrap[T any](v T) *jsonWrapper[T] {
+	return &jsonWrapper[T]{v}
+}
+
+func unwrap[T any](v any) (T, error) {
+	str, ok := v.(string)
+	if !ok {
+		return *new(T), errors.Errorf("could not unwrap value of type '%T'", v)
+	}
+
+	u := new(T)
+
+	if err := json.Unmarshal([]byte(str), u); err != nil {
+		return *new(T), errors.WithStack(err)
+	}
+
+	return *u, nil
+}
+
+func allNilValues(values []any) bool {
+	for _, v := range values {
+		if v != nil {
+			return false
+		}
+	}
+
+	return true
+}
+
+func key(parts ...string) string {
+	return strings.Join(parts, ":")
+}

internal/store/redis/layer_repository.go

@@ -0,0 +1,241 @@
+package redis
+
+import (
+	"context"
+	"time"
+
+	"forge.cadoles.com/cadoles/bouncer/internal/store"
+	"github.com/pkg/errors"
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	keyLayerName      = "name"
+	keyLayerProxy     = "proxy"
+	keyLayerType      = "type"
+	keyLayerOptions   = "options"
+	keyLayerUpdatedAt = "updated_at"
+	keyLayerCreatedAt = "created_at"
+	keyLayerWeight    = "weight"
+	keyPrefixLayer    = "layer:"
+)
+
+type LayerRepository struct {
+	client redis.UniversalClient
+}
+
+// CreateLayer implements store.LayerRepository
+func (r *LayerRepository) CreateLayer(ctx context.Context, proxyName store.ProxyName, layerName store.LayerName, layerType store.LayerType, options store.LayerOptions) (*store.Layer, error) {
+	now := time.Now().UTC()
+	key := layerKey(proxyName, layerName)
+
+	txf := func(tx *redis.Tx) error {
+		exists, err := tx.Exists(ctx, key).Uint64()
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		if exists > 0 {
+			return errors.WithStack(store.ErrAlreadyExist)
+		}
+
+		_, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error {
+			p.HMSet(ctx, key, keyLayerName, string(layerName))
+			p.HMSet(ctx, key, keyLayerType, string(layerType))
+			p.HMSet(ctx, key, keyLayerProxy, string(proxyName))
+			p.HMSet(ctx, key, keyLayerOptions, wrap(options))
+			p.HMSet(ctx, key, keyLayerWeight, wrap(0))
+			p.HMSet(ctx, key, keyLayerCreatedAt, wrap(now))
+			p.HMSet(ctx, key, keyLayerUpdatedAt, wrap(now))
+
+			return nil
+		})
+
+		return err
+	}
+
+	err := r.client.Watch(ctx, txf, key)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return &store.Layer{
+		LayerHeader: store.LayerHeader{
+			Name:      layerName,
+			Proxy:     proxyName,
+			Type:      layerType,
+			CreatedAt: now,
+			UpdatedAt: now,
+		},
+	}, nil
+}
+
+// DeleteLayer implements store.LayerRepository
+func (r *LayerRepository) DeleteLayer(ctx context.Context, proxyName store.ProxyName, layerName store.LayerName) error {
+	key := layerKey(proxyName, layerName)
+
+	if cmd := r.client.Del(ctx, key); cmd.Err() != nil {
+		return errors.WithStack(cmd.Err())
+	}
+
+	return nil
+}
+
+// GetLayer implements store.LayerRepository
+func (r *LayerRepository) GetLayer(ctx context.Context, proxyName store.ProxyName, layerName store.LayerName) (*store.Layer, error) {
+	var layer store.Layer
+
+	key := layerKey(proxyName, layerName)
+
+	cmd := r.client.HMGet(ctx, key, keyLayerType, keyLayerWeight, keyLayerOptions, keyLayerCreatedAt, keyLayerUpdatedAt)
+
+	values, err := cmd.Result()
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	if allNilValues(values) {
+		return nil, errors.WithStack(store.ErrNotFound)
+	}
+
+	layer.Name = layerName
+	layer.Proxy = proxyName
+
+	layerType, ok := values[0].(string)
+	if !ok {
+		return nil, errors.Errorf("unexpected '%s' value of type '%T'", keyLayerType, values[0])
+	}
+
+	layer.Type = store.LayerType(layerType)
+
+	weight, err := unwrap[int](values[1])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	layer.Weight = weight
+
+	options, err := unwrap[map[string]any](values[2])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	layer.Options = options
+
+	createdAt, err := unwrap[time.Time](values[3])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	layer.CreatedAt = createdAt
+
+	updatedAt, err := unwrap[time.Time](values[4])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	layer.UpdatedAt = updatedAt
+
+	return &layer, nil
+}
+
+// QueryLayers implements store.LayerRepository
+func (r *LayerRepository) QueryLayers(ctx context.Context, funcs ...store.QueryLayerOptionFunc) ([]*store.LayerHeader, error) {
+	opts := &store.QueryLayerOptions{}
+	for _, fn := range funcs {
+		fn(opts)
+	}
+
+	keyParts := []string{keyPrefixLayer}
+
+	if opts.Proxy != nil {
+		keyParts = append(keyParts, string(*opts.Proxy))
+	} else {
+		keyParts = append(keyParts, "*")
+	}
+
+	if opts.Name != nil {
+		keyParts = append(keyParts, string(*opts.Name))
+	} else {
+		keyParts = append(keyParts, "*")
+	}
+
+	key := key(keyParts...)
+
+	iter := r.client.Scan(ctx, uint64(*opts.Offset), key, int64(*opts.Limit)).Iterator()
+
+	headers := make([]*store.LayerHeader, 0)
+
+	for iter.Next(ctx) {
+		key := iter.Val()
+
+		cmd := r.client.HMGet(ctx, key, keyLayerName, keyLayerProxy, keyLayerType, keyLayerCreatedAt, keyLayerUpdatedAt)
+
+		values, err := cmd.Result()
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		if allNilValues(values) {
+			continue
+		}
+
+		layerName, ok := values[0].(string)
+		if !ok {
+			return nil, errors.Errorf("unexpected '%s' field value for key '%s': '%s'", keyLayerName, key, values[0])
+		}
+
+		proxyName, ok := values[1].(string)
+		if !ok {
+			return nil, errors.Errorf("unexpected '%s' field value for key '%s': '%s'", keyProxyName, key, values[1])
+		}
+
+		layerType, ok := values[2].(string)
+		if !ok {
+			return nil, errors.Errorf("unexpected '%s' field value for key '%s': '%s'", keyLayerType, key, values[1])
+		}
+
+		createdAt, err := unwrap[time.Time](values[3])
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		updatedAt, err := unwrap[time.Time](values[4])
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		h := &store.LayerHeader{
+			Name:      store.LayerName(layerName),
+			Proxy:     store.ProxyName(proxyName),
+			Type:      store.LayerType(layerType),
+			CreatedAt: createdAt,
+			UpdatedAt: updatedAt,
+		}
+
+		headers = append(headers, h)
+	}
+
+	if err := iter.Err(); err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return headers, nil
+}
+
+// UpdateLayer implements store.LayerRepository
+func (r *LayerRepository) UpdateLayer(ctx context.Context, proxyName store.ProxyName, layerName store.LayerName, options store.LayerOptions) error {
+	panic("unimplemented")
+}
+
+func NewLayerRepository(client redis.UniversalClient) *LayerRepository {
+	return &LayerRepository{
+		client: client,
+	}
+}
+
+var _ store.LayerRepository = &LayerRepository{}
+
+func layerKey(proxyName store.ProxyName, layerName store.LayerName) string {
+	return key(keyPrefixLayer, string(proxyName), string(layerName))
+}

internal/store/redis/layer_repository_test.go

@@ -0,0 +1,12 @@
+package redis
+
+import (
+	"testing"
+
+	"forge.cadoles.com/cadoles/bouncer/internal/store/testsuite"
+)
+
+func TestLayerRepository(t *testing.T) {
+	repository := NewLayerRepository(client)
+	testsuite.TestLayerRepository(t, repository)
+}

internal/store/redis/main_test.go

@@ -0,0 +1,58 @@
+package redis
+
+import (
+	"context"
+	"log"
+	"os"
+	"testing"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/pkg/errors"
+	"github.com/redis/go-redis/v9"
+)
+
+var client redis.UniversalClient
+
+func TestMain(m *testing.M) {
+	// uses a sensible default on windows (tcp/http) and linux/osx (socket)
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		log.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	// uses pool to try to connect to Docker
+	err = pool.Client.Ping()
+	if err != nil {
+		log.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	// pulls an image, creates a container based on it and runs it
+	resource, err := pool.Run("redis", "alpine3.17", []string{})
+	if err != nil {
+		log.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	if err := pool.Retry(func() error {
+		client = redis.NewUniversalClient(&redis.UniversalOptions{
+			Addrs: []string{resource.GetHostPort("6379/tcp")},
+		})
+
+		ctx := context.Background()
+
+		if cmd := client.Ping(ctx); cmd.Err() != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	}); err != nil {
+		log.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	code := m.Run()
+
+	if err := pool.Purge(resource); err != nil {
+		log.Fatalf("%+v", errors.WithStack(err))
+	}
+
+	os.Exit(code)
+}

internal/store/redis/proxy_repository.go

@@ -0,0 +1,211 @@
+package redis
+
+import (
+	"context"
+	"net/url"
+	"time"
+
+	"forge.cadoles.com/cadoles/bouncer/internal/store"
+	"github.com/pkg/errors"
+	"github.com/redis/go-redis/v9"
+)
+
+const (
+	keyProxyName      = "name"
+	keyProxyFrom      = "from"
+	keyProxyTo        = "to"
+	keyProxyUpdatedAt = "updated_at"
+	keyProxyCreatedAt = "created_at"
+	keyProxyWeight    = "weight"
+	keyPrefixProxy    = "proxy:"
+)
+
+type ProxyRepository struct {
+	client redis.UniversalClient
+}
+
+// GetProxy implements store.ProxyRepository
+func (r *ProxyRepository) GetProxy(ctx context.Context, name store.ProxyName) (*store.Proxy, error) {
+	var proxy store.Proxy
+
+	key := proxyKey(name)
+
+	cmd := r.client.HMGet(ctx, key, keyProxyFrom, keyProxyTo, keyProxyWeight, keyProxyCreatedAt, keyProxyUpdatedAt)
+
+	values, err := cmd.Result()
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	if allNilValues(values) {
+		return nil, errors.WithStack(store.ErrNotFound)
+	}
+
+	proxy.Name = name
+
+	from, err := unwrap[[]string](values[0])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	proxy.From = from
+
+	rawTo, ok := values[1].(string)
+	if !ok {
+		return nil, errors.Errorf("unexpected 'to' value of type '%T'", values[1])
+	}
+
+	to, err := url.Parse(rawTo)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	proxy.To = to
+
+	weight, err := unwrap[int](values[2])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	proxy.Weight = weight
+
+	createdAt, err := unwrap[time.Time](values[3])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	proxy.CreatedAt = createdAt
+
+	updatedAt, err := unwrap[time.Time](values[4])
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	proxy.UpdatedAt = updatedAt
+
+	return &proxy, nil
+}
+
+// CreateProxy implements store.ProxyRepository
+func (r *ProxyRepository) CreateProxy(ctx context.Context, name store.ProxyName, to *url.URL, from ...string) (*store.Proxy, error) {
+	now := time.Now().UTC()
+	key := proxyKey(name)
+
+	txf := func(tx *redis.Tx) error {
+		exists, err := tx.Exists(ctx, key).Uint64()
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		if exists > 0 {
+			return errors.WithStack(store.ErrAlreadyExist)
+		}
+
+		_, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error {
+			p.HMSet(ctx, key, keyProxyName, string(name))
+			p.HMSet(ctx, key, keyProxyFrom, wrap(from))
+			p.HMSet(ctx, key, keyProxyTo, to.String())
+			p.HMSet(ctx, key, keyProxyWeight, wrap(0))
+			p.HMSet(ctx, key, keyProxyCreatedAt, wrap(now))
+			p.HMSet(ctx, key, keyProxyUpdatedAt, wrap(now))
+
+			return nil
+		})
+
+		return err
+	}
+
+	err := r.client.Watch(ctx, txf, key)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return &store.Proxy{
+		ProxyHeader: store.ProxyHeader{
+			Name:      name,
+			CreatedAt: now,
+			UpdatedAt: now,
+		},
+		To:   to,
+		From: from,
+	}, nil
+}
+
+// DeleteProxy implements store.ProxyRepository
+func (r *ProxyRepository) DeleteProxy(ctx context.Context, name store.ProxyName) error {
+	key := proxyKey(name)
+
+	if cmd := r.client.Del(ctx, key); cmd.Err() != nil {
+		return errors.WithStack(cmd.Err())
+	}
+
+	return nil
+}
+
+// QueryProxy implements store.ProxyRepository
+func (r *ProxyRepository) QueryProxy(ctx context.Context, funcs ...store.QueryProxyOptionFunc) ([]*store.ProxyHeader, error) {
+	iter := r.client.Scan(ctx, 0, keyPrefixProxy+"*", 0).Iterator()
+
+	headers := make([]*store.ProxyHeader, 0)
+
+	for iter.Next(ctx) {
+		key := iter.Val()
+
+		cmd := r.client.HMGet(ctx, key, keyProxyName, keyProxyCreatedAt, keyProxyUpdatedAt)
+
+		values, err := cmd.Result()
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		if allNilValues(values) {
+			continue
+		}
+
+		name, ok := values[0].(string)
+		if !ok {
+			return nil, errors.Errorf("unexpected 'name' field value for key '%s': '%s'", key, values[0])
+		}
+
+		createdAt, err := unwrap[time.Time](values[1])
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		updatedAt, err := unwrap[time.Time](values[2])
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		h := &store.ProxyHeader{
+			Name:      store.ProxyName(name),
+			CreatedAt: createdAt,
+			UpdatedAt: updatedAt,
+		}
+
+		headers = append(headers, h)
+	}
+
+	if err := iter.Err(); err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return headers, nil
+}
+
+// UpdateProxy implements store.ProxyRepository
+func (*ProxyRepository) UpdateProxy(ctx context.Context, name store.ProxyName, funcs ...store.UpdateProxyOptionFunc) (*store.Proxy, error) {
+	panic("unimplemented")
+}
+
+func NewProxyRepository(client redis.UniversalClient) *ProxyRepository {
+	return &ProxyRepository{
+		client: client,
+	}
+}
+
+var _ store.ProxyRepository = &ProxyRepository{}
+
+func proxyKey(name store.ProxyName) string {
+	return key(keyPrefixProxy, string(name))
+}

internal/store/redis/proxy_repository_test.go

@@ -0,0 +1,12 @@
+package redis
+
+import (
+	"testing"
+
+	"forge.cadoles.com/cadoles/bouncer/internal/store/testsuite"
+)
+
+func TestProxyRepository(t *testing.T) {
+	repository := NewProxyRepository(client)
+	testsuite.TestProxyRepository(t, repository)
+}