package systemtables

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/keyspace"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/typemap"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/types"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
)

// MySQLSchemaProvider supplies system_schema rows from information_schema.
type MySQLSchemaProvider struct {
	schemaManager *mysqlbackend.SchemaManager
}

type tableKey struct {
	schema string
	table  string
}

// NewMySQLSchemaProvider creates a MySQL-backed schema provider.
func NewMySQLSchemaProvider(schemaManager *mysqlbackend.SchemaManager) *MySQLSchemaProvider {
	return &MySQLSchemaProvider{schemaManager: schemaManager}
}

func (p *MySQLSchemaProvider) Keyspaces(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaKeyspaces, metaAvailable, err := p.schemaManager.CQLKeyspaces(ctx, user, password)
	if err != nil {
		return nil, err
	}
	schemas, err := p.schemaManager.ListSchemas(ctx, user, password)
	if err != nil {
		return nil, err
	}

	rows := make([]types.Row, 0, len(schemas))
	seen := make(map[string]struct{}, len(metaKeyspaces))
	if metaAvailable {
		for _, ks := range metaKeyspaces {
			if !isUserSchema(ks.Name) {
				continue
			}
			replication := decodeJSONMap(ks.ReplicationJSON)
			if replication == nil {
				replication = map[string]interface{}{
					"class":              "SimpleStrategy",
					"replication_factor": "1",
				}
			}
			rows = append(rows, buildRow(columns, map[string]interface{}{
				"keyspace_name":  ks.Name,
				"durable_writes": ks.DurableWrites,
				"replication":    replication,
				"fast_path":      false,
			}))
			seen[ks.Name] = struct{}{}
		}
	}
	for _, schema := range schemas {
		if !isUserSchema(schema) {
			continue
		}
		if _, ok := seen[schema]; ok {
			continue
		}
		rows = append(rows, buildRow(columns, map[string]interface{}{
			"keyspace_name":  schema,
			"durable_writes": true,
			"replication": map[string]string{
				"class":              "SimpleStrategy",
				"replication_factor": "1",
			},
			"fast_path": false,
		}))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Tables(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaTables, metaAvailable, err := p.schemaManager.CQLTables(ctx, user, password)
	if err != nil {
		return nil, err
	}
	tables, err := p.schemaManager.ListTables(ctx, user, password)
	if err != nil {
		return nil, err
	}
	rows := make([]types.Row, 0, len(tables))
	seen := make(map[tableKey]struct{})
	if metaAvailable {
		for _, table := range metaTables {
			if !isUserSchema(table.Keyspace) {
				continue
			}
			id, err := tableIDFor(table.Keyspace, table.Table)
			if err != nil {
				return nil, err
			}
			rows = append(rows, schemaTableRow(columns, table.Keyspace, table.Table, table.Comment, id))
			seen[tableKey{schema: table.Keyspace, table: table.Table}] = struct{}{}
		}
	}
	for _, table := range tables {
		if !isUserSchema(table.Schema) {
			continue
		}
		if _, ok := seen[tableKey{schema: table.Schema, table: table.Name}]; ok {
			continue
		}
		id, err := tableIDFor(table.Schema, table.Name)
		if err != nil {
			return nil, err
		}
		rows = append(rows, schemaTableRow(columns, table.Schema, table.Name, table.Comment, id))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Columns(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaCols, metaAvailable, err := p.schemaManager.CQLColumns(ctx, user, password)
	if err != nil {
		return nil, err
	}
	cols, err := p.schemaManager.ListColumns(ctx, user, password)
	if err != nil {
		return nil, err
	}
	indexes, err := p.schemaManager.ListIndexes(ctx, user, password)
	if err != nil {
		return nil, err
	}

	rows := make([]types.Row, 0, len(cols))
	pkMap := buildPrimaryKeyMap(indexes)

	if metaAvailable {
		for _, col := range metaCols {
			if !isUserSchema(col.Keyspace) {
				continue
			}
			rows = append(rows, buildRow(columns, map[string]interface{}{
				"keyspace_name":     col.Keyspace,
				"table_name":        col.Table,
				"column_name":       col.Name,
				"clustering_order":  col.ClusteringOrder,
				"column_name_bytes": []byte(col.Name),
				"default_expr":      col.DefaultExpr,
				"on_update_expr":    col.OnUpdateExpr,
				"kind":              col.Kind,
				"position":          col.Position,
				"type":              col.CQLType,
			}))
		}
	}

	metaTableSet := make(map[tableKey]struct{})
	if metaAvailable {
		for _, col := range metaCols {
			metaTableSet[tableKey{schema: col.Keyspace, table: col.Table}] = struct{}{}
		}
	}

	for _, col := range cols {
		if !isUserSchema(col.Schema) {
			continue
		}
		if _, ok := metaTableSet[tableKey{schema: col.Schema, table: col.Table}]; ok {
			continue
		}
		pk := pkMap[tableKey{schema: col.Schema, table: col.Table}]
		kind, position, order := columnKindForMySQL(col.Name, pk)
		rows = append(rows, buildRow(columns, map[string]interface{}{
			"keyspace_name":     col.Schema,
			"table_name":        col.Table,
			"column_name":       col.Name,
			"clustering_order":  order,
			"column_name_bytes": []byte(col.Name),
			"default_expr":      "",
			"on_update_expr":    "",
			"kind":              kind,
			"position":          position,
			"type":              typemap.MySQLTypeToCQLTypeStringWithPrecisionScale(col.DataType, col.NumericPrecision, col.NumericScale),
		}))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Indexes(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaIndexes, metaAvailable, err := p.schemaManager.CQLIndexes(ctx, user, password)
	if err != nil {
		return nil, err
	}
	indexes, err := p.schemaManager.ListIndexes(ctx, user, password)
	if err != nil {
		return nil, err
	}

	type idxKey struct {
		schema string
		table  string
		name   string
	}
	byKey := make(map[idxKey][]mysqlbackend.IndexInfo)
	for _, idx := range indexes {
		if idx.IsPrimary || !isUserSchema(idx.Schema) || idx.Column == "" {
			continue
		}
		key := idxKey{schema: idx.Schema, table: idx.Table, name: idx.Name}
		byKey[key] = append(byKey[key], idx)
	}

	keys := make([]idxKey, 0, len(byKey))
	for key := range byKey {
		keys = append(keys, key)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].schema == keys[j].schema {
			if keys[i].table == keys[j].table {
				return keys[i].name < keys[j].name
			}
			return keys[i].table < keys[j].table
		}
		return keys[i].schema < keys[j].schema
	})

	rows := make([]types.Row, 0, len(keys))
	seen := make(map[idxKey]struct{})
	if metaAvailable {
		for _, idx := range metaIndexes {
			if !isUserSchema(idx.Keyspace) || idx.Table == "" || idx.Name == "" {
				continue
			}
			options := decodeJSONMap(idx.OptionsJSON)
			rows = append(rows, buildRow(columns, map[string]interface{}{
				"keyspace_name": idx.Keyspace,
				"table_name":    idx.Table,
				"index_name":    idx.Name,
				"kind":          idx.Kind,
				"options":       options,
			}))
			seen[idxKey{schema: idx.Keyspace, table: idx.Table, name: idx.Name}] = struct{}{}
		}
	}
	for _, key := range keys {
		if _, ok := seen[key]; ok {
			continue
		}
		entries := append([]mysqlbackend.IndexInfo(nil), byKey[key]...)
		sort.Slice(entries, func(i, j int) bool { return entries[i].Seq < entries[j].Seq })
		target := make([]string, 0, len(entries))
		for _, entry := range entries {
			target = append(target, entry.Column)
		}
		rows = append(rows, buildRow(columns, map[string]interface{}{
			"keyspace_name": key.schema,
			"table_name":    key.table,
			"index_name":    key.name,
			"kind":          "CUSTOM",
			"options": map[string]interface{}{
				"target": target,
			},
		}))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Types(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaTypes, metaAvailable, err := p.schemaManager.CQLTypes(ctx, user, password)
	if err != nil {
		return nil, err
	}
	if !metaAvailable {
		return nil, nil
	}
	rows := make([]types.Row, 0, len(metaTypes))
	for _, typ := range metaTypes {
		if !isUserSchema(typ.Keyspace) {
			continue
		}
		rows = append(rows, buildRow(columns, map[string]interface{}{
			"keyspace_name": typ.Keyspace,
			"type_name":     typ.Name,
			"field_names":   typ.FieldNames,
			"field_types":   typ.FieldTypes,
		}))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Views(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaViews, metaAvailable, err := p.schemaManager.CQLViews(ctx, user, password)
	if err != nil {
		return nil, err
	}
	if !metaAvailable {
		return nil, nil
	}

	rows := make([]types.Row, 0, len(metaViews))
	baseIDs := make(map[tableKey]string)
	for _, view := range metaViews {
		if !isUserSchema(view.Keyspace) {
			continue
		}
		baseKey := tableKey{schema: view.Keyspace, table: view.BaseTable}
		baseTableID, ok := baseIDs[baseKey]
		if !ok {
			id, err := tableIDFor(view.Keyspace, view.BaseTable)
			if err != nil {
				return nil, err
			}
			baseTableID = id
			baseIDs[baseKey] = baseTableID
		}
		options := decodeJSONMap(view.OptionsJSON)
		comment := commentFromOptionsMap(options)
		viewID, err := viewIDFor(view.Keyspace, view.Name)
		if err != nil {
			return nil, err
		}
		rows = append(rows, schemaViewRow(columns, view.Keyspace, view.Name, view.BaseTable, view.WhereClause, baseTableID, viewID, comment))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Triggers(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	metaTriggers, metaAvailable, err := p.schemaManager.CQLTriggers(ctx, user, password)
	if err != nil {
		return nil, err
	}
	if !metaAvailable {
		return nil, nil
	}
	rows := make([]types.Row, 0, len(metaTriggers))
	for _, trigger := range metaTriggers {
		if !isUserSchema(trigger.Keyspace) {
			continue
		}
		options := decodeJSONMap(trigger.OptionsJSON)
		rows = append(rows, buildRow(columns, map[string]interface{}{
			"keyspace_name": trigger.Keyspace,
			"table_name":    trigger.Table,
			"trigger_name":  trigger.Name,
			"options":       options,
		}))
	}
	return rows, nil
}

func (p *MySQLSchemaProvider) Functions(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	return nil, nil
}

func (p *MySQLSchemaProvider) Aggregates(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	return nil, nil
}

func (p *MySQLSchemaProvider) Users(ctx context.Context, columns []types.ColumnInfo, user, password string) ([]types.Row, error) {
	if p == nil || p.schemaManager == nil {
		return nil, nil
	}
	users, err := p.schemaManager.ListUsers(ctx, user, password)
	if err != nil {
		return nil, err
	}
	rows := make([]types.Row, 0, len(users))
	for _, account := range users {
		rows = append(rows, buildRow(columns, map[string]interface{}{
			"role":         account.Name,
			"is_superuser": false,
			"can_login":    true,
			"salted_hash":  nil,
			"member_of":    nil,
		}))
	}
	return rows, nil
}

func decodeJSONMap(raw []byte) map[string]interface{} {
	if len(raw) == 0 {
		return nil
	}
	var out map[string]interface{}
	if err := json.Unmarshal(raw, &out); err != nil {
		return nil
	}
	return out
}

func isUserSchema(name string) bool {
	return !keyspace.IsInternalSchema(name) && !keyspace.IsBuiltIn(name)
}

func schemaTableRow(columns []types.ColumnInfo, keyspaceName, tableName, comment, id string) types.Row {
	values := map[string]interface{}{
		"keyspace_name":               keyspaceName,
		"table_name":                  tableName,
		"allow_auto_snapshot":         true,
		"bloom_filter_fp_chance":      0.01,
		"caching":                     map[string]string{"keys": "ALL"},
		"comment":                     comment,
		"compaction":                  map[string]string{"class": "LeveledCompactionStrategy"},
		"compression":                 map[string]string{"class": "LZ4Compressor"},
		"memtable":                    map[string]string{},
		"crc_check_chance":            1.0,
		"dclocal_read_repair_chance":  0.0,
		"default_time_to_live":        0,
		"extensions":                  map[string]string{},
		"flags":                       []string{"compound"},
		"gc_grace_seconds":            0,
		"incremental_backups":         false,
		"id":                          id,
		"max_index_interval":          2048,
		"memtable_flush_period_in_ms": 0,
		"min_index_interval":          128,
		"read_repair_chance":          0.0,
		"speculative_retry":           "99PERCENTILE",
		"additional_write_policy":     "99PERCENTILE",
		"cdc":                         false,
		"read_repair":                 "NONE",
		"fast_path":                   false,
		"auto_repair":                 false,
	}
	return buildRow(columns, values)
}

func schemaViewRow(columns []types.ColumnInfo, keyspaceName, viewName, baseTable, whereClause, baseTableID, viewID, comment string) types.Row {
	values := map[string]interface{}{
		"keyspace_name":               keyspaceName,
		"view_name":                   viewName,
		"base_table_id":               baseTableID,
		"base_table_name":             baseTable,
		"where_clause":                whereClause,
		"allow_auto_snapshot":         true,
		"bloom_filter_fp_chance":      0.01,
		"caching":                     map[string]string{"keys": "ALL"},
		"comment":                     comment,
		"compaction":                  map[string]string{"class": "LeveledCompactionStrategy"},
		"compression":                 map[string]string{"class": "LZ4Compressor"},
		"memtable":                    map[string]string{},
		"crc_check_chance":            1.0,
		"dclocal_read_repair_chance":  0.0,
		"default_time_to_live":        0,
		"extensions":                  map[string]string{},
		"gc_grace_seconds":            0,
		"incremental_backups":         false,
		"id":                          viewID,
		"include_all_columns":         true,
		"max_index_interval":          2048,
		"memtable_flush_period_in_ms": 0,
		"min_index_interval":          128,
		"read_repair_chance":          0.0,
		"speculative_retry":           "99PERCENTILE",
		"additional_write_policy":     "99PERCENTILE",
		"cdc":                         false,
		"read_repair":                 "NONE",
		"auto_repair":                 map[string]string{},
	}
	return buildRow(columns, values)
}

func buildPrimaryKeyMap(indexes []mysqlbackend.IndexInfo) map[tableKey][]mysqlbackend.IndexInfo {
	pkMap := make(map[tableKey][]mysqlbackend.IndexInfo)
	for _, idx := range indexes {
		if !idx.IsPrimary {
			continue
		}
		key := tableKey{schema: idx.Schema, table: idx.Table}
		pkMap[key] = append(pkMap[key], idx)
	}
	return pkMap
}

func columnKindForMySQL(column string, pk []mysqlbackend.IndexInfo) (string, int, string) {
	if len(pk) == 0 {
		return "regular", -1, "NONE"
	}
	if pk[0].Column == column {
		return "partition_key", 0, "NONE"
	}
	for i := 1; i < len(pk); i++ {
		if pk[i].Column == column {
			order := "ASC"
			if pk[i].Collation == "D" {
				order = "DESC"
			}
			return "clustering", i - 1, order
		}
	}
	return "regular", -1, "NONE"
}

func commentFromOptionsMap(options map[string]interface{}) string {
	if len(options) == 0 {
		return ""
	}
	raw, ok := options["comment"]
	if !ok {
		return ""
	}
	switch v := raw.(type) {
	case string:
		return v
	default:
		return fmt.Sprint(v)
	}
}