package controlplane

import (
	"context"
	"crypto/sha256"
	_ "embed"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"aigw/internal/security"

	"github.com/jackc/pgx/v5"
	"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")

const migrationVersion int64 = 2026080504

type Options struct {
	DatabaseURL            string
	RedisURL               string
	CredentialKey          string
	PreviousCredentialKeys []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.NewCredentialKeyring(options.CredentialKey, options.PreviousCredentialKeys)
	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) Ping(ctx context.Context) error { return s.db.Ping(ctx) }

func (s *Store) Migrate(ctx context.Context) error {
	return applySchema(ctx, s.db)
}

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()
	return applySchema(ctx, db)
}

func MigrationStatusDatabase(ctx context.Context, databaseURL string) (MigrationStatus, error) {
	db, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		return MigrationStatus{}, fmt.Errorf("configure PostgreSQL: %w", err)
	}
	defer db.Close()
	var result MigrationStatus
	err = db.QueryRow(ctx, `SELECT version,name,checksum,applied_at FROM schema_migrations ORDER BY version DESC LIMIT 1`).Scan(&result.Version, &result.Name, &result.Checksum, &result.AppliedAt)
	if err != nil {
		return MigrationStatus{}, fmt.Errorf("read migration status: %w", err)
	}
	return result, nil
}

func applySchema(ctx context.Context, db *pgxpool.Pool) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return fmt.Errorf("begin migration: %w", err)
	}
	defer tx.Rollback(ctx)
	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, migrationVersion); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, schemaSQL); err != nil {
		return fmt.Errorf("apply control-plane schema: %w", err)
	}
	hash := sha256.Sum256([]byte(schemaSQL))
	checksum := hex.EncodeToString(hash[:])
	var existing string
	err = tx.QueryRow(ctx, `SELECT checksum FROM schema_migrations WHERE version=$1`, migrationVersion).Scan(&existing)
	if err == nil && existing != checksum {
		return fmt.Errorf("migration %d checksum changed; deploy an explicit new migration version", migrationVersion)
	}
	if !errors.Is(err, pgx.ErrNoRows) && err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations(version,name,checksum) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING`, migrationVersion, "commercial-control-plane", checksum); err != nil {
		return err
	}
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("commit migration: %w", err)
	}
	return nil
}

type MigrationStatus struct {
	Version   int64     `json:"version"`
	Name      string    `json:"name"`
	Checksum  string    `json:"checksum"`
	AppliedAt time.Time `json:"applied_at"`
}

func (s *Store) MigrationStatus(ctx context.Context) (MigrationStatus, error) {
	var result MigrationStatus
	err := s.db.QueryRow(ctx, `SELECT version,name,checksum,applied_at FROM schema_migrations ORDER BY version DESC LIMIT 1`).Scan(&result.Version, &result.Name, &result.Checksum, &result.AppliedAt)
	return result, err
}

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