// Package deltautils provides a centralized Delta Lake table manager that allows
// multiple tasks to share a single open table instance. Since all NextGen tasks
// (metering, reporter, maintenance) run serially with a Delta Lake lock, sharing
// a single table instance avoids redundant open operations and reduces memory usage.
package deltautils

import (
	"context"
	"fmt"
	"net/url"
	"os"
	"strings"
	"sync"

	deltalake "github.com/tidbcloud/metering/delta-rs-golang"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/models"
	"github.com/tidbcloud/metering/pkg/common/models/deltaschema"
	"go.uber.org/zap"
)

// TableManager manages a globally shared Delta Lake table instance.
// It provides thread-safe access to the table and handles lifecycle management.
type TableManager struct {
	mu             sync.Mutex
	table          *deltalake.DeltaTable
	tablePath      string
	storageOptions map[string]string
	log            *zap.Logger
}

var (
	globalManager   *TableManager
	globalManagerMu sync.Mutex
)

// DefaultTableConfig returns the optimized table configuration for opening tables
func DefaultTableConfig() deltalake.TableConfig {
	return deltalake.TableConfig{
		RequireFiles:  true,
		LogBufferSize: 4,
	}
}

// GetGlobalTableManager returns the global TableManager singleton.
// The manager must be initialized with InitGlobalTableManager before calling this.
func GetGlobalTableManager() *TableManager {
	globalManagerMu.Lock()
	defer globalManagerMu.Unlock()
	return globalManager
}

// InitGlobalTableManager initializes the global TableManager with the given configuration.
// This should be called once during application startup.
// It automatically prepares storage options based on the vendor (e.g., Alicloud).
func InitGlobalTableManager(conf *config.ServiceConfig, log *zap.Logger) (*TableManager, error) {
	globalManagerMu.Lock()
	defer globalManagerMu.Unlock()

	if globalManager != nil {
		return globalManager, nil
	}

	if conf == nil || conf.NextGenMeteringConfig == nil {
		return nil, fmt.Errorf("NextGenMeteringConfig is required")
	}

	// Prepare storage options based on vendor
	if err := PrepareStorageOptions(conf); err != nil {
		return nil, fmt.Errorf("failed to prepare storage options: %w", err)
	}

	globalManager = &TableManager{
		tablePath:      conf.NextGenMeteringConfig.DeltaTablePath,
		storageOptions: conf.NextGenMeteringConfig.StorageOptions,
		log:            log.With(zap.String("module", "delta-table-manager")),
	}

	return globalManager, nil
}

// ResetGlobalTableManager resets the global manager (primarily for testing)
func ResetGlobalTableManager() {
	globalManagerMu.Lock()
	defer globalManagerMu.Unlock()

	if globalManager != nil {
		globalManager.Close()
		globalManager = nil
	}
}

// NewTableManager creates a new TableManager instance (for use cases that don't need the global singleton)
func NewTableManager(tablePath string, storageOptions map[string]string, log *zap.Logger) *TableManager {
	return &TableManager{
		tablePath:      tablePath,
		storageOptions: storageOptions,
		log:            log.With(zap.String("module", "delta-table-manager")),
	}
}

// OpenTable explicitly opens the Delta Lake table.
// If a table is already open, it returns the existing instance.
func (m *TableManager) OpenTable(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.table != nil {
		m.log.Debug("table already open", zap.String("path", m.tablePath))
		return nil
	}

	_, err := m.openTableLocked()
	return err
}

// openTableLocked opens the table (must be called with lock held)
func (m *TableManager) openTableLocked() (*deltalake.DeltaTable, error) {
	if m.tablePath == "" {
		return nil, fmt.Errorf("delta table path not configured")
	}

	m.log.Info("opening Delta table", zap.String("path", m.tablePath))

	config := DefaultTableConfig()

	var table *deltalake.DeltaTable
	var err error

	// Recover from panics during table open
	func() {
		defer func() {
			if r := recover(); r != nil {
				m.log.Error("recovered from panic while opening Delta table",
					zap.String("path", m.tablePath),
					zap.Any("panic", r))
				err = fmt.Errorf("table does not exist or is corrupted: %v", r)
			}
		}()
		table, err = deltalake.OpenDeltaLakeTableWithConfig(m.tablePath, m.storageOptions, config)
	}()

	if err != nil || table == nil {
		return nil, fmt.Errorf("failed to open delta table at %s: %w", m.tablePath, err)
	}

	m.table = table
	m.log.Info("Delta table opened successfully",
		zap.String("path", m.tablePath),
		zap.Bool("require_files", config.RequireFiles),
		zap.Int("log_buffer_size", config.LogBufferSize))

	return m.table, nil
}

// CreateTable creates a new Delta Lake table at the configured path.
func (m *TableManager) CreateTable(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.tablePath == "" {
		return fmt.Errorf("delta table path not configured")
	}
	// if a table is already open, return error
	if m.table != nil {
		return fmt.Errorf("delta table already opened at %s", m.tablePath)
	}

	m.log.Info("creating new Delta table", zap.String("path", m.tablePath))

	// Use unified schema definition from models package
	columns := deltaschema.NextGenMeteringDeltaSchema
	partitionColumns := deltaschema.NextGenMeteringPartitionColumns
	tableName := deltaschema.NextGenMeteringTableName

	// Default optimized table configuration
	tableConfiguration := map[string]string{
		"delta.logRetentionDuration":    "interval 1 days",
		"delta.checkpointInterval":      "50",
		"delta.enableExpiredLogCleanup": "true",
	}

	createOptions := deltalake.CreateTableOptions{
		TablePath:          m.tablePath,
		SaveMode:           deltalake.WriteModeError,
		Columns:            columns,
		PartitionColumns:   partitionColumns,
		TableName:          &tableName,
		StorageOptions:     m.storageOptions,
		TableConfiguration: tableConfiguration,
	}

	table, err := deltalake.CreateDeltaLakeTable(createOptions)
	if err != nil {
		return fmt.Errorf("failed to create delta table: %w", err)
	}

	m.table = table

	m.log.Info("Delta table created successfully",
		zap.String("path", m.tablePath),
		zap.Int("checkpoint_interval", 50),
		zap.String("log_retention", "1 days"))

	return nil
}

// TableExists checks if a Delta Lake table exists at the configured path.
func (m *TableManager) TableExists(ctx context.Context) (bool, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.tablePath == "" {
		return false, fmt.Errorf("delta table path not configured")
	}

	exists, err := deltalake.TableExists(m.tablePath, m.storageOptions)
	if err != nil {
		m.log.Error("failed to check table existence",
			zap.String("path", m.tablePath),
			zap.Error(err))
		return false, fmt.Errorf("failed to check table existence: %w", err)
	}

	m.log.Debug("checked table existence",
		zap.String("path", m.tablePath),
		zap.Bool("exists", exists))

	return exists, nil
}

// GetTableVersion returns the current version of the Delta Lake table.
func (m *TableManager) GetTableVersion(ctx context.Context) (int64, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	// if table is not opened, return error
	if m.table == nil {
		return 0, fmt.Errorf("delta table not opened")
	}

	version, err := m.table.Version()
	if err != nil {
		m.log.Error("failed to get table version from opened table", zap.Error(err))
		return 0, fmt.Errorf("failed to get table version: %w", err)
	}
	m.log.Debug("retrieved table version from opened table", zap.Int64("version", version))
	return version, nil

}

// WriteToDeltaTable writes billing records to the delta table.
// This method handles the complexities of Rust FFI memory management.
func (m *TableManager) WriteToDeltaTable(ctx context.Context, records []models.BillingReportV2Deltalake, saveMode string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if len(records) == 0 {
		m.log.Debug("no records to write")
		return nil
	}

	// if table is not opened, return error
	if m.table == nil {
		return fmt.Errorf("delta table not opened")
	}

	// Convert records to WriteData format
	data := make([][]interface{}, 0, len(records))
	for _, record := range records {
		row := []interface{}{
			record.SchemeVersion,
			int64(record.TenantID),
			record.CostID,
			record.CostTimestamp,
			record.Unit,
			record.UnitValue,
			record.Cost,
			record.Labels,
			record.CreatedAt,
			record.SKUType,
			record.HourDimension,
		}
		data = append(data, row)
	}

	schema := deltaschema.NextGenMeteringDeltaSchema
	writeData := deltalake.WriteData{
		Schema: schema,
		Data:   data,
	}

	// Convert saveMode to WriteMode
	var mode deltalake.WriteMode
	switch saveMode {
	case "Append":
		mode = deltalake.WriteModeAppend
	case "Overwrite":
		mode = deltalake.WriteModeOverwrite
	default:
		mode = deltalake.WriteModeAppend
	}

	m.log.Info("writing records to Delta table",
		zap.Int("count", len(records)),
		zap.String("mode", saveMode))

	// Write to delta table
	// Note: Rust FFI now properly consumes the old table pointer and returns a new one
	updatedTable, err := m.table.Write(writeData, mode)

	// CRITICAL: Always update table pointer, even on error
	m.table = updatedTable

	if err != nil {
		return fmt.Errorf("failed to write to delta table: %w", err)
	}
	for i, record := range records {
		m.log.Debug("Writing records to delta table",
			zap.Int("record_index", i+1),
			zap.Int("total_records", len(records)),
			zap.String("mode", saveMode),
			zap.Any("record", record))
	}
	m.log.Info("successfully wrote records to Delta table", zap.Int("count", len(records)))
	return nil
}

// Query executes a SQL query against the Delta Lake table.
func (m *TableManager) Query(ctx context.Context, sql string, options *deltalake.QueryOptions) ([]map[string]interface{}, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// if table is not opened, return error
	if m.table == nil {
		return nil, fmt.Errorf("delta table not opened")
	}

	m.log.Debug("executing query", zap.String("sql", sql))

	rows, err := m.table.Query(sql, options)
	if err != nil {
		return nil, fmt.Errorf("failed to execute query: %w", err)
	}

	return rows, nil
}

// DeleteResult contains the metrics from a delete operation.
type DeleteResult struct {
	NumRemovedFiles int64
	NumDeletedRows  int64
	NumCopiedRows   int64
}

// Delete removes rows matching the predicate from the Delta Lake table.
// This method handles table pointer updates after the delete operation.
func (m *TableManager) Delete(ctx context.Context, predicate string) (*DeleteResult, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.table == nil {
		return nil, fmt.Errorf("delta table not opened")
	}

	m.log.Info("executing delete operation", zap.String("predicate", predicate))

	result, err := m.table.DeleteWhere(predicate)
	if err != nil {
		return nil, fmt.Errorf("failed to delete records: %w", err)
	}

	if result == nil || result.Table == nil {
		return nil, fmt.Errorf("delete operation returned an empty result")
	}

	// Update table pointer with the new table returned by delete
	m.table = result.Table

	metrics := &DeleteResult{
		NumRemovedFiles: result.Metrics.NumRemovedFiles,
		NumDeletedRows:  result.Metrics.NumDeletedRows,
		NumCopiedRows:   result.Metrics.NumCopiedRows,
	}

	m.log.Info("delete completed",
		zap.Int64("num_removed_files", metrics.NumRemovedFiles),
		zap.Int64("num_deleted_rows", metrics.NumDeletedRows),
		zap.Int64("num_copied_rows", metrics.NumCopiedRows))

	return metrics, nil
}

// CompactResult contains the metrics from a compact operation.
type CompactResult struct {
	NumFilesAdded       int64
	NumFilesRemoved     int64
	PartitionsOptimized int64
}

// Compact performs file compaction on the Delta Lake table.
// This method handles table pointer updates after the compact operation.
func (m *TableManager) Compact(ctx context.Context) (*CompactResult, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.table == nil {
		return nil, fmt.Errorf("delta table not opened")
	}

	m.log.Info("executing compact operation")

	builder := m.table.Optimize()
	if builder == nil {
		return nil, fmt.Errorf("failed to create optimize builder")
	}
	defer builder.Free()

	builder.SetOptimizeType(deltalake.OptimizeTypeCompact)
	result, err := builder.Execute()
	if err != nil {
		return nil, fmt.Errorf("compact operation failed: %w", err)
	}

	// Update table pointer with the new table returned by optimize
	if result != nil && result.Table != nil {
		m.table = result.Table
	}

	metrics := &CompactResult{}
	if result != nil {
		metrics.NumFilesAdded = result.Metrics.NumFilesAdded
		metrics.NumFilesRemoved = result.Metrics.NumFilesRemoved
		metrics.PartitionsOptimized = result.Metrics.PartitionsOptimized
	}

	m.log.Info("compact completed",
		zap.Int64("files_added", metrics.NumFilesAdded),
		zap.Int64("files_removed", metrics.NumFilesRemoved),
		zap.Int64("partitions_optimized", metrics.PartitionsOptimized))

	return metrics, nil
}

// VacuumResult contains the metrics from a vacuum operation.
type VacuumResult struct {
	TotalFilesDeleted int64
	DryRun            bool
}

// VacuumOptions contains configuration for the vacuum operation.
type VacuumOptions struct {
	RetentionHours int  // Number of hours to retain files (default: 168 = 7 days)
	DryRun         bool // If true, only simulate the vacuum without deleting files
}

// Vacuum removes old files that are no longer referenced by the Delta Lake table.
// This method handles table pointer updates after the vacuum operation.
func (m *TableManager) Vacuum(ctx context.Context, opts *VacuumOptions) (*VacuumResult, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.table == nil {
		return nil, fmt.Errorf("delta table not opened")
	}

	if opts == nil {
		opts = &VacuumOptions{RetentionHours: 168} // Default 7 days
	}

	m.log.Info("executing vacuum operation",
		zap.Int("retention_hours", opts.RetentionHours),
		zap.Bool("dry_run", opts.DryRun))

	builder := m.table.Vacuum()
	if builder == nil {
		return nil, fmt.Errorf("failed to create vacuum builder")
	}
	defer builder.Free()

	if opts.RetentionHours > 0 {
		builder.SetRetentionHours(opts.RetentionHours)
	}
	builder.SetDryRun(opts.DryRun)

	result, err := builder.Execute()
	if err != nil {
		return nil, fmt.Errorf("vacuum operation failed: %w", err)
	}

	// Update table pointer with the new table returned by vacuum
	if result != nil && result.Table != nil {
		m.table = result.Table
	}

	metrics := &VacuumResult{}
	if result != nil {
		metrics.TotalFilesDeleted = result.Metrics.TotalFilesDeleted
		metrics.DryRun = result.Metrics.DryRun
	}

	m.log.Info("vacuum completed",
		zap.Int64("total_files_deleted", metrics.TotalFilesDeleted),
		zap.Bool("dry_run", metrics.DryRun))

	return metrics, nil
}

// Close closes the current table and releases resources.
func (m *TableManager) Close() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.table != nil {
		m.log.Info("closing Delta table", zap.String("path", m.tablePath))
		m.table.Close()
		m.table = nil
	}
	return nil
}

// GetTablePath returns the configured table path.
func (m *TableManager) GetTablePath() string {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.tablePath
}

// GetStorageOptions returns a copy of the storage options.
func (m *TableManager) GetStorageOptions() map[string]string {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.storageOptions == nil {
		return nil
	}

	// Return a copy to prevent external modification
	opts := make(map[string]string, len(m.storageOptions))
	for k, v := range m.storageOptions {
		opts[k] = v
	}
	return opts
}

// IsOpen returns whether a table is currently open.
func (m *TableManager) IsOpen() bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.table != nil
}

// PrepareStorageOptions prepares storage options based on the vendor configuration.
// This function handles vendor-specific setup (e.g., Alicloud OSS credentials).
// It modifies conf.NextGenMeteringConfig.StorageOptions in place.
func PrepareStorageOptions(conf *config.ServiceConfig) error {
	if conf == nil || conf.NextGenMeteringConfig == nil {
		return nil
	}

	switch strings.ToLower(conf.Vendor) {
	case strings.ToLower(constants.VendorAlicloud):
		return prepareAlicloudStorageOptions(conf)
	case strings.ToLower(constants.VendorAzure):
		return prepareAzureStorageOptions(conf)
	default:
		// No special preparation needed for other vendors (AWS, GCP, etc.)
		return nil
	}
}

// prepareAlicloudStorageOptions prepares storage options for Alicloud OSS.
func prepareAlicloudStorageOptions(conf *config.ServiceConfig) error {
	accessKey := strings.TrimSpace(os.Getenv(constants.AlicloudAccessKeyIDEnv))
	secretKey := strings.TrimSpace(os.Getenv(constants.AlicloudAccessKeySecretEnv))
	if accessKey == "" || secretKey == "" {
		return fmt.Errorf("%s and %s environment variables are required for Alicloud",
			constants.AlicloudAccessKeyIDEnv, constants.AlicloudAccessKeySecretEnv)
	}

	bucketEndpoint := strings.TrimSpace(conf.NextGenMeteringConfig.BucketEndpoint)
	if bucketEndpoint == "" {
		return fmt.Errorf("bucketEndpoint is required in NextGenMeteringConfig for Alicloud")
	}
	bucket := strings.TrimSpace(conf.Bucket)
	if bucket == "" {
		return fmt.Errorf("bucket is required in ServiceConfig for Alicloud")
	}

	parsedEndpoint, err := url.Parse(bucketEndpoint)
	if err != nil {
		return fmt.Errorf("invalid bucketEndpoint %q: %w", bucketEndpoint, err)
	}

	if parsedEndpoint.Scheme == "" {
		parsedEndpoint.Scheme = "https"
	}

	host := strings.TrimSpace(parsedEndpoint.Host)
	if host == "" {
		return fmt.Errorf("invalid bucketEndpoint %q: missing host", bucketEndpoint)
	}

	if !strings.HasPrefix(host, bucket+".") {
		host = bucket + "." + host
	}

	parsedEndpoint.Host = host
	parsedEndpoint.Path = ""
	parsedEndpoint.RawQuery = ""
	parsedEndpoint.Fragment = ""

	bucketEndpoint = parsedEndpoint.String()

	if conf.NextGenMeteringConfig.StorageOptions == nil {
		conf.NextGenMeteringConfig.StorageOptions = make(map[string]string)
	}

	conf.NextGenMeteringConfig.StorageOptions["AWS_ACCESS_KEY_ID"] = accessKey
	conf.NextGenMeteringConfig.StorageOptions["AWS_SECRET_ACCESS_KEY"] = secretKey
	conf.NextGenMeteringConfig.StorageOptions["AWS_ENDPOINT_URL"] = bucketEndpoint
	conf.NextGenMeteringConfig.StorageOptions["AWS_VIRTUAL_HOSTED_STYLE_REQUEST"] = "true"
	conf.NextGenMeteringConfig.StorageOptions["AWS_COPY_IF_NOT_EXISTS"] = "header-with-status:x-oss-forbid-overwrite:true:409"
	if conf.Region != "" {
		conf.NextGenMeteringConfig.StorageOptions["region"] = conf.Region
		conf.NextGenMeteringConfig.StorageOptions["AWS_REGION"] = conf.Region
	}

	return nil
}

// prepareAzureStorageOptions prepares storage options for Azure Blob Storage.
// Credential sources (in precedence order):
// 1. Storage account key (AZURE_STORAGE_ACCOUNT_NAME + AZURE_STORAGE_ACCOUNT_KEY)
// 2. SAS token (AZURE_STORAGE_SAS_TOKEN)
// 3. Managed identity (default when no explicit credentials are set)
func prepareAzureStorageOptions(conf *config.ServiceConfig) error {
	accountName := strings.TrimSpace(os.Getenv(constants.AzureStorageAccountNameEnv))
	if accountName == "" {
		return fmt.Errorf("%s environment variable is required for Azure", constants.AzureStorageAccountNameEnv)
	}

	accountKey := strings.TrimSpace(os.Getenv(constants.AzureStorageAccountKeyEnv))
	sasToken := strings.TrimSpace(os.Getenv(constants.AzureStorageSASTokenEnv))

	if conf.NextGenMeteringConfig.StorageOptions == nil {
		conf.NextGenMeteringConfig.StorageOptions = make(map[string]string)
	}

	conf.NextGenMeteringConfig.StorageOptions["AZURE_STORAGE_ACCOUNT_NAME"] = accountName
	if accountKey != "" {
		conf.NextGenMeteringConfig.StorageOptions["AZURE_STORAGE_ACCOUNT_KEY"] = accountKey
	}
	if sasToken != "" {
		conf.NextGenMeteringConfig.StorageOptions["AZURE_STORAGE_SAS_TOKEN"] = sasToken
	}

	return nil
}