1 files changed, 170 insertions, 0 deletions
diff --git a/internal/controlplane/store.go b/internal/controlplane/store.go
new file mode 100644
index 0000000..833f846
--- /dev/null
+++ b/internal/controlplane/store.go
@@ -0,0 +1,170 @@
+package controlplane
+
+import (
+	"context"
+	_ "embed"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"aigw/internal/security"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/redis/go-redis/v9"
+)
+
+//go:embed schema.sql
+var schemaSQL string
+
+var ErrRedisDisabled = errors.New("Redis propagation is disabled")
+
+type Options struct {
+	DatabaseURL     string
+	RedisURL        string
+	CredentialKey   string
+	RedisChannel    string
+	VersionCacheKey string
+}
+
+type Store struct {
+	db              *pgxpool.Pool
+	redis           *redis.Client
+	cipher          *security.CredentialCipher
+	redisChannel    string
+	versionCacheKey string
+}
+
+func NewStore(ctx context.Context, options Options) (*Store, error) {
+	cipher, err := security.NewCredentialCipher(options.CredentialKey)
+	if err != nil {
+		return nil, err
+	}
+	db, err := pgxpool.New(ctx, options.DatabaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("configure PostgreSQL: %w", err)
+	}
+	if err := db.Ping(ctx); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("connect PostgreSQL: %w", err)
+	}
+	var redisClient *redis.Client
+	if strings.TrimSpace(options.RedisURL) != "" {
+		redisOptions, err := redis.ParseURL(options.RedisURL)
+		if err != nil {
+			db.Close()
+			return nil, fmt.Errorf("parse Redis URL: %w", err)
+		}
+		redisClient = redis.NewClient(redisOptions)
+	}
+	return &Store{
+		db: db, redis: redisClient, cipher: cipher,
+		redisChannel: options.RedisChannel, versionCacheKey: options.VersionCacheKey,
+	}, nil
+}
+
+func (s *Store) Close() error {
+	s.db.Close()
+	if s.redis == nil {
+		return nil
+	}
+	return s.redis.Close()
+}
+
+func (s *Store) RedisEnabled() bool {
+	return s.redis != nil
+}
+
+func (s *Store) Migrate(ctx context.Context) error {
+	if _, err := s.db.Exec(ctx, schemaSQL); err != nil {
+		return fmt.Errorf("apply control-plane schema: %w", err)
+	}
+	return nil
+}
+
+func MigrateDatabase(ctx context.Context, databaseURL string) error {
+	db, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		return fmt.Errorf("configure PostgreSQL: %w", err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(ctx, schemaSQL); err != nil {
+		return fmt.Errorf("apply control-plane schema: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) DatabaseGeneration(ctx context.Context) (int64, error) {
+	var generation int64
+	err := s.db.QueryRow(ctx, `SELECT generation FROM control_state WHERE singleton = TRUE`).Scan(&generation)
+	if err != nil {
+		return 0, fmt.Errorf("read control-plane generation: %w", err)
+	}
+	return generation, nil
+}
+
+func (s *Store) RedisGeneration(ctx context.Context) (int64, error) {
+	if s.redis == nil {
+		return 0, ErrRedisDisabled
+	}
+	generation, err := s.redis.Get(ctx, s.versionCacheKey).Int64()
+	if errors.Is(err, redis.Nil) {
+		return 0, nil
+	}
+	return generation, err
+}
+
+func (s *Store) PublishChange(ctx context.Context, event ChangeEvent) error {
+	if s.redis == nil {
+		return ErrRedisDisabled
+	}
+	payload, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	pipeline := s.redis.TxPipeline()
+	pipeline.Set(ctx, s.versionCacheKey, event.Generation, 0)
+	pipeline.Publish(ctx, s.redisChannel, payload)
+	_, err = pipeline.Exec(ctx)
+	if err != nil {
+		return fmt.Errorf("publish control-plane change: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) Subscribe(ctx context.Context) (<-chan ChangeMessage, func() error, error) {
+	if s.redis == nil {
+		return nil, nil, ErrRedisDisabled
+	}
+	pubsub := s.redis.Subscribe(ctx, s.redisChannel)
+	if _, err := pubsub.Receive(ctx); err != nil {
+		_ = pubsub.Close()
+		return nil, nil, fmt.Errorf("subscribe control-plane changes: %w", err)
+	}
+	messages := make(chan ChangeMessage)
+	redisMessages := pubsub.Channel()
+	go func() {
+		defer close(messages)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case message, ok := <-redisMessages:
+				if !ok {
+					return
+				}
+				select {
+				case messages <- ChangeMessage{Payload: message.Payload}:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+	return messages, pubsub.Close, nil
+}
+
+func newChange(generation int64, resource, id string) ChangeEvent {
+	return ChangeEvent{Generation: generation, Resource: resource, ID: id, ChangedAt: time.Now().UTC().Format(time.RFC3339Nano)}
+}