package systemtables

import (
	"context"
	"crypto/sha1"
	"encoding/binary"
	"fmt"
	"hash"
	"strings"
	"sync"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

// SchemaVersionProvider supplies schema_version values.
type SchemaVersionProvider interface {
	Current(ctx context.Context, user, password string) (string, error)
	Bump(ctx context.Context, user, password string) (string, error)
}

func newSchemaVersionProvider(cfg config.SchemaVersionConfig, schemaManager *mysqlbackend.SchemaManager) (SchemaVersionProvider, error) {
	strategy := strings.ToLower(strings.TrimSpace(cfg.Strategy))
	switch strategy {
	case "fixed":
		if cfg.Fixed == "" {
			return nil, fmt.Errorf("schema_version.fixed is required for fixed strategy")
		}
		if _, err := parseUUIDBytes(cfg.Fixed); err != nil {
			return nil, fmt.Errorf("invalid schema_version.fixed: %w", err)
		}
		return fixedSchemaVersionProvider{value: cfg.Fixed}, nil
	case "":
		strategy = "cql_meta"
		fallthrough
	case "cql_meta":
		if schemaManager == nil {
			return newInMemorySchemaVersionProvider(), nil
		}
		return &cqlMetaSchemaVersionProvider{schemaManager: schemaManager, fallback: newInMemorySchemaVersionProvider()}, nil
	case "hash":
		if schemaManager == nil {
			return newInMemorySchemaVersionProvider(), nil
		}
		return &hashSchemaVersionProvider{schemaManager: schemaManager}, nil
	default:
		return nil, fmt.Errorf("unknown schema_version strategy %q", cfg.Strategy)
	}
}

type fixedSchemaVersionProvider struct {
	value string
}

func (p fixedSchemaVersionProvider) Current(_ context.Context, _ string, _ string) (string, error) {
	return p.value, nil
}

func (p fixedSchemaVersionProvider) Bump(_ context.Context, _ string, _ string) (string, error) {
	return p.value, nil
}

type inMemorySchemaVersionProvider struct {
	mu    sync.Mutex
	value string
}

func newInMemorySchemaVersionProvider() *inMemorySchemaVersionProvider {
	value, err := uuidV4()
	if err != nil {
		value = "00000000-0000-0000-0000-000000000001"
	}
	return &inMemorySchemaVersionProvider{value: value}
}

func (p *inMemorySchemaVersionProvider) Current(_ context.Context, _ string, _ string) (string, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.value, nil
}

func (p *inMemorySchemaVersionProvider) Bump(_ context.Context, _ string, _ string) (string, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	next, err := uuidV4()
	if err != nil {
		return "", err
	}
	p.value = next
	return next, nil
}

type cqlMetaSchemaVersionProvider struct {
	schemaManager *mysqlbackend.SchemaManager
	fallback      *inMemorySchemaVersionProvider
}

func (p *cqlMetaSchemaVersionProvider) Current(ctx context.Context, user, password string) (string, error) {
	if p.schemaManager == nil {
		return "", fmt.Errorf("schema manager not initialized")
	}
	if err := p.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		if mysqlbackend.IsAccessDenied(err) && p.fallback != nil {
			return p.fallback.Current(ctx, user, password)
		}
		return "", err
	}
	version, err := p.schemaManager.SchemaVersion(ctx, user, password)
	if err != nil {
		if mysqlbackend.IsAccessDenied(err) && p.fallback != nil {
			return p.fallback.Current(ctx, user, password)
		}
		return "", err
	}
	if version != "" {
		return version, nil
	}
	next, err := uuidV4()
	if err != nil {
		return "", err
	}
	if err := p.schemaManager.SetSchemaVersion(ctx, user, password, next); err != nil {
		if mysqlbackend.IsAccessDenied(err) && p.fallback != nil {
			return p.fallback.Current(ctx, user, password)
		}
		return "", err
	}
	return next, nil
}

func (p *cqlMetaSchemaVersionProvider) Bump(ctx context.Context, user, password string) (string, error) {
	if p.schemaManager == nil {
		return "", fmt.Errorf("schema manager not initialized")
	}
	if err := p.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		if mysqlbackend.IsAccessDenied(err) && p.fallback != nil {
			return p.fallback.Bump(ctx, user, password)
		}
		return "", err
	}
	next, err := uuidV4()
	if err != nil {
		return "", err
	}
	if err := p.schemaManager.SetSchemaVersion(ctx, user, password, next); err != nil {
		if mysqlbackend.IsAccessDenied(err) && p.fallback != nil {
			return p.fallback.Bump(ctx, user, password)
		}
		return "", err
	}
	return next, nil
}

type hashSchemaVersionProvider struct {
	schemaManager *mysqlbackend.SchemaManager
}

func (p *hashSchemaVersionProvider) Current(ctx context.Context, user, password string) (string, error) {
	return p.compute(ctx, user, password)
}

func (p *hashSchemaVersionProvider) Bump(ctx context.Context, user, password string) (string, error) {
	return p.compute(ctx, user, password)
}

func (p *hashSchemaVersionProvider) compute(ctx context.Context, user, password string) (string, error) {
	if p.schemaManager == nil {
		return "", fmt.Errorf("schema manager not initialized")
	}
	if err := p.schemaManager.EnsureCQLMeta(ctx, user, password); err != nil {
		return "", err
	}
	hasher := sha1.New()

	keyspaces, _, err := p.schemaManager.CQLKeyspaces(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "keyspaces", len(keyspaces))
	for _, ks := range keyspaces {
		hashString(hasher, ks.Name)
		hashBool(hasher, ks.DurableWrites)
		hashBytes(hasher, ks.ReplicationJSON)
	}

	tables, _, err := p.schemaManager.CQLTables(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "tables", len(tables))
	for _, tbl := range tables {
		hashString(hasher, tbl.Keyspace)
		hashString(hasher, tbl.Table)
		hashBytes(hasher, tbl.OptionsJSON)
		hashString(hasher, tbl.Comment)
	}

	columns, _, err := p.schemaManager.CQLColumns(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "columns", len(columns))
	for _, col := range columns {
		hashString(hasher, col.Keyspace)
		hashString(hasher, col.Table)
		hashString(hasher, col.Name)
		hashString(hasher, col.CQLType)
		hashString(hasher, col.Kind)
		hashInt(hasher, int64(col.Position))
		hashString(hasher, col.ClusteringOrder)
	}

	typesList, _, err := p.schemaManager.CQLTypes(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "types", len(typesList))
	for _, typ := range typesList {
		hashString(hasher, typ.Keyspace)
		hashString(hasher, typ.Name)
		hashStringSlice(hasher, typ.FieldNames)
		hashStringSlice(hasher, typ.FieldTypes)
	}

	indexes, _, err := p.schemaManager.CQLIndexes(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "indexes", len(indexes))
	for _, idx := range indexes {
		hashString(hasher, idx.Keyspace)
		hashString(hasher, idx.Table)
		hashString(hasher, idx.Name)
		hashString(hasher, idx.Kind)
		hashBytes(hasher, idx.OptionsJSON)
	}

	views, _, err := p.schemaManager.CQLViews(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "views", len(views))
	for _, view := range views {
		hashString(hasher, view.Keyspace)
		hashString(hasher, view.Name)
		hashString(hasher, view.BaseTable)
		hashString(hasher, view.WhereClause)
		hashBytes(hasher, view.OptionsJSON)
	}

	triggers, _, err := p.schemaManager.CQLTriggers(ctx, user, password)
	if err != nil {
		return "", err
	}
	hashLabel(hasher, "triggers", len(triggers))
	for _, trig := range triggers {
		hashString(hasher, trig.Keyspace)
		hashString(hasher, trig.Table)
		hashString(hasher, trig.Name)
		hashBytes(hasher, trig.OptionsJSON)
	}

	sum := hasher.Sum(nil)
	version := uuidFromHash(sum)
	return formatUUIDBytes(version), nil
}

func hashLabel(h hash.Hash, label string, count int) {
	hashString(h, label)
	hashInt(h, int64(count))
}

func hashString(h hash.Hash, value string) {
	hashBytes(h, []byte(value))
}

func hashStringSlice(h hash.Hash, values []string) {
	hashInt(h, int64(len(values)))
	for _, value := range values {
		hashString(h, value)
	}
}

func hashBool(h hash.Hash, value bool) {
	if value {
		hashBytes(h, []byte{1})
	} else {
		hashBytes(h, []byte{0})
	}
}

func hashInt(h hash.Hash, value int64) {
	var buf [8]byte
	binary.BigEndian.PutUint64(buf[:], uint64(value))
	hashBytes(h, buf[:])
}

func hashBytes(h hash.Hash, value []byte) {
	var lenBuf [4]byte
	binary.BigEndian.PutUint32(lenBuf[:], uint32(len(value)))
	_, _ = h.Write(lenBuf[:])
	if len(value) > 0 {
		_, _ = h.Write(value)
	}
}