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)}
}