package deltautils

import (
	"context"
	"encoding/json"
	"os"
	"sync"
	"testing"
	"time"

	deltalake "github.com/tidbcloud/metering/delta-rs-golang"
	"github.com/tidbcloud/metering/pkg/common/constants"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/metering/pkg/common/config"
	"github.com/tidbcloud/metering/pkg/common/models"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest"
	"go.uber.org/zap/zaptest/observer"
)

func TestNewTableManager(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path/to/table", map[string]string{"key": "value"}, log)

	assert.NotNil(t, manager)
	assert.Equal(t, "/path/to/table", manager.tablePath)
	assert.Equal(t, "value", manager.storageOptions["key"])
	assert.Nil(t, manager.table)
}

func TestTableManager_GetTablePath(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path/to/table", nil, log)

	assert.Equal(t, "/path/to/table", manager.GetTablePath())
}

func TestTableManager_GetStorageOptions(t *testing.T) {
	log := zaptest.NewLogger(t)

	t.Run("nil options", func(t *testing.T) {
		manager := NewTableManager("/path", nil, log)
		assert.Nil(t, manager.GetStorageOptions())
	})

	t.Run("with options", func(t *testing.T) {
		manager := NewTableManager("/path", map[string]string{"key": "value"}, log)
		opts := manager.GetStorageOptions()
		assert.Equal(t, "value", opts["key"])

		// Verify it's a copy
		opts["key"] = "modified"
		assert.Equal(t, "value", manager.storageOptions["key"])
	})
}

func TestTableManager_IsOpen(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	assert.False(t, manager.IsOpen())
}

func TestTableManager_Close(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	// Close should not error even when no table is open
	err := manager.Close()
	assert.NoError(t, err)
}

func TestTableManager_TableExists_NoPath(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("", nil, log)

	_, err := manager.TableExists(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table path not configured")
}

func TestTableManager_GetTableVersion_NotOpened(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	_, err := manager.GetTableVersion(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_OpenTable_NoPath(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("", nil, log)

	err := manager.OpenTable(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table path not configured")
}

func TestTableManager_CreateTable_NoPath(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("", nil, log)

	err := manager.CreateTable(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table path not configured")
}

func TestTableManager_WriteToDeltaTable_EmptyRecords(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	err := manager.WriteToDeltaTable(context.Background(), []models.BillingReportV2Deltalake{}, "Append")
	assert.NoError(t, err)
}

func TestGlobalTableManager(t *testing.T) {
	// Reset any existing global manager
	ResetGlobalTableManager()

	log := zaptest.NewLogger(t)
	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/test/path",
			StorageOptions: map[string]string{"key": "value"},
		},
	}

	// Initialize global manager
	manager, err := InitGlobalTableManager(conf, log)
	require.NoError(t, err)
	assert.NotNil(t, manager)

	// Get global manager
	retrieved := GetGlobalTableManager()
	assert.Equal(t, manager, retrieved)

	// Initialize again should return same instance
	manager2, err := InitGlobalTableManager(conf, log)
	require.NoError(t, err)
	assert.Equal(t, manager, manager2)

	// Reset and verify
	ResetGlobalTableManager()
	assert.Nil(t, GetGlobalTableManager())
}

func TestInitGlobalTableManager_NilConfig(t *testing.T) {
	ResetGlobalTableManager()

	log := zaptest.NewLogger(t)

	_, err := InitGlobalTableManager(nil, log)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "NextGenMeteringConfig is required")
}

func TestInitGlobalTableManager_NilNextGenConfig(t *testing.T) {
	ResetGlobalTableManager()

	log := zaptest.NewLogger(t)
	conf := &config.ServiceConfig{}

	_, err := InitGlobalTableManager(conf, log)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "NextGenMeteringConfig is required")
}

func TestTableManager_ConcurrentAccess(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	var wg sync.WaitGroup
	numGoroutines := 10

	// Concurrent reads should not cause issues
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = manager.GetTablePath()
			_ = manager.IsOpen()
			_ = manager.GetStorageOptions()
		}()
	}

	wg.Wait()
}

func TestDefaultTableConfig(t *testing.T) {
	config := DefaultTableConfig()

	assert.True(t, config.RequireFiles)
	assert.Equal(t, 4, config.LogBufferSize)
}

func TestTableManager_Delete_NotOpened(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	_, err := manager.Delete(context.Background(), "predicate")
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_Compact_NotOpened(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	_, err := manager.Compact(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_Vacuum_NotOpened(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	_, err := manager.Vacuum(context.Background(), nil)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_Vacuum_DefaultOptions(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	// Call with nil options - should use defaults
	_, err := manager.Vacuum(context.Background(), nil)
	assert.Error(t, err) // Still error because table not opened
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_Query_NotOpened(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", nil, log)

	_, err := manager.Query(context.Background(), "SELECT * FROM table", nil)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_UpdateConfig(t *testing.T) {
	log := zaptest.NewLogger(t)

	manager := NewTableManager("/path", map[string]string{"key": "old"}, log)
	assert.Equal(t, "/path", manager.GetTablePath())
	assert.Equal(t, "old", manager.GetStorageOptions()["key"])
}

// Integration-style tests with real Delta table operations

func TestTableManager_WriteToDeltaTable_NotOpened(t *testing.T) {
	log := zaptest.NewLogger(t)
	manager := NewTableManager("/path", nil, log)

	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: time.Now().Unix(),
			Unit:          "GB",
			UnitValue:     "100.0",
			Cost:          10,
			Labels:        "{}",
			CreatedAt:     time.Now().Unix(),
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
	}

	err := manager.WriteToDeltaTable(context.Background(), records, "Append")
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table not opened")
}

func TestTableManager_OpenTable_AlreadyOpen(t *testing.T) {
	log := zaptest.NewLogger(t)
	manager := NewTableManager("/tmp/test-table", nil, log)

	// Mock that table is already open
	manager.mu.Lock()
	manager.table = &deltalake.DeltaTable{} // Mock table
	manager.mu.Unlock()

	err := manager.OpenTable(context.Background())
	assert.NoError(t, err)
}

func TestTableManager_CreateTable_AlreadyOpened(t *testing.T) {
	log := zaptest.NewLogger(t)
	manager := NewTableManager("/tmp/test-table", nil, log)

	// Mock that table is already open
	manager.mu.Lock()
	manager.table = &deltalake.DeltaTable{} // Mock table
	manager.mu.Unlock()

	err := manager.CreateTable(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "delta table already opened")
}

func TestTableManager_Close_WithTable(t *testing.T) {
	log := zaptest.NewLogger(t)
	manager := NewTableManager("/tmp/test-table", nil, log)

	// Set a mock table
	manager.mu.Lock()
	manager.table = &deltalake.DeltaTable{} // Mock table
	manager.mu.Unlock()

	err := manager.Close()
	assert.NoError(t, err)
	assert.False(t, manager.IsOpen())
}

func TestPrepareStorageOptions(t *testing.T) {
	tests := []struct {
		name       string
		conf       *config.ServiceConfig
		setupEnv   func()
		cleanupEnv func()
		wantErr    bool
		errMsg     string
	}{
		{
			name:    "nil config",
			conf:    nil,
			wantErr: false,
		},
		{
			name:    "nil NextGenMeteringConfig",
			conf:    &config.ServiceConfig{},
			wantErr: false,
		},
		{
			name: "non-alicloud vendor",
			conf: &config.ServiceConfig{
				Vendor: "AWS",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
				},
			},
			wantErr: false,
		},
		{
			name: "alicloud without credentials",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
				},
			},
			setupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: true,
			errMsg:  "environment variables are required",
		},
		{
			name: "alicloud without bucket endpoint",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "",
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: true,
			errMsg:  "bucketEndpoint is required",
		},
		{
			name: "alicloud without bucket",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				Bucket: "",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "oss-cn-hangzhou.aliyuncs.com",
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: true,
			errMsg:  "bucket is required",
		},
		{
			name: "alicloud invalid endpoint",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				Bucket: "test-bucket",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "://invalid-url",
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: true,
			errMsg:  "invalid bucketEndpoint",
		},
		{
			name: "alicloud with valid config",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				Bucket: "test-bucket",
				Region: "cn-hangzhou",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "https://oss-cn-hangzhou.aliyuncs.com",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: false,
		},
		{
			name: "alicloud with endpoint including scheme",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				Bucket: "test-bucket",
				Region: "cn-hangzhou",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "https://oss-cn-hangzhou.aliyuncs.com",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: false,
		},
		{
			name: "alicloud with endpoint without bucket prefix",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				Bucket: "my-bucket",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "https://oss-cn-hangzhou.aliyuncs.com",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: false,
		},
		{
			name: "alicloud with endpoint already having bucket prefix",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAlicloud,
				Bucket: "my-bucket",
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					BucketEndpoint: "https://my-bucket.oss-cn-hangzhou.aliyuncs.com",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
				os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
				os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
			},
			wantErr: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if tt.setupEnv != nil {
				tt.setupEnv()
			}
			if tt.cleanupEnv != nil {
				defer tt.cleanupEnv()
			}

			err := PrepareStorageOptions(tt.conf)

			if tt.wantErr {
				require.Error(t, err)
				if tt.errMsg != "" {
					assert.Contains(t, err.Error(), tt.errMsg)
				}
			} else {
				require.NoError(t, err)
				// Verify Alicloud-specific options are set correctly
				if tt.conf != nil && tt.conf.NextGenMeteringConfig != nil && tt.conf.Vendor == constants.VendorAlicloud {
					opts := tt.conf.NextGenMeteringConfig.StorageOptions
					assert.NotEmpty(t, opts["AWS_ACCESS_KEY_ID"])
					assert.NotEmpty(t, opts["AWS_SECRET_ACCESS_KEY"])
					assert.NotEmpty(t, opts["AWS_ENDPOINT_URL"])
					assert.Equal(t, "true", opts["AWS_VIRTUAL_HOSTED_STYLE_REQUEST"])
					assert.Equal(t, "header-with-status:x-oss-forbid-overwrite:true:409", opts["AWS_COPY_IF_NOT_EXISTS"])
					if tt.conf.Region != "" {
						assert.Equal(t, tt.conf.Region, opts["region"])
						assert.Equal(t, tt.conf.Region, opts["AWS_REGION"])
					}
				}
			}
		})
	}
}

func TestInitGlobalTableManager_WithAlicloud(t *testing.T) {
	ResetGlobalTableManager()
	defer ResetGlobalTableManager()

	log := zaptest.NewLogger(t)

	// Setup Alicloud environment
	os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
	os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
	defer func() {
		os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
		os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
	}()

	conf := &config.ServiceConfig{
		Vendor: constants.VendorAlicloud,
		Bucket: "test-bucket",
		Region: "cn-hangzhou",
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/test/path",
			BucketEndpoint: "https://oss-cn-hangzhou.aliyuncs.com",
			StorageOptions: map[string]string{},
		},
	}

	manager, err := InitGlobalTableManager(conf, log)
	require.NoError(t, err)
	assert.NotNil(t, manager)

	// Verify storage options were prepared
	opts := manager.GetStorageOptions()
	assert.Equal(t, "test-key", opts["AWS_ACCESS_KEY_ID"])
	assert.Equal(t, "test-secret", opts["AWS_SECRET_ACCESS_KEY"])
	assert.Contains(t, opts["AWS_ENDPOINT_URL"], "test-bucket")
}

func TestTableManager_WriteToDeltaTable_Modes(t *testing.T) {
	log := zaptest.NewLogger(t)
	manager := NewTableManager("/path", nil, log)

	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
		},
	}

	tests := []struct {
		name     string
		saveMode string
	}{
		{"append mode", "Append"},
		{"overwrite mode", "Overwrite"},
		{"unknown mode defaults to append", "Unknown"},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := manager.WriteToDeltaTable(context.Background(), records, tt.saveMode)
			// All should fail since table is not opened
			assert.Error(t, err)
			assert.Contains(t, err.Error(), "delta table not opened")
		})
	}
}

// Integration Tests - These tests create real Delta tables in temp directories

func createTestTable(t *testing.T, tempDir string) *TableManager {
	t.Helper()
	log := zaptest.NewLogger(t)
	tablePath := tempDir + "/test_table"

	// Create the directory first (required by delta-rs)
	err := os.MkdirAll(tablePath, 0755)
	require.NoError(t, err)

	manager := NewTableManager(tablePath, nil, log)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)
	require.True(t, manager.IsOpen())

	return manager
}

func TestTableManager_CreateAndOpenTable_Integration(t *testing.T) {
	tempDir := t.TempDir()
	log := zaptest.NewLogger(t)
	tablePath := tempDir + "/test_table"

	// Create directory first
	err := os.MkdirAll(tablePath, 0755)
	require.NoError(t, err)

	manager := NewTableManager(tablePath, nil, log)

	// Create table
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)
	assert.True(t, manager.IsOpen())

	// Get version
	version, err := manager.GetTableVersion(context.Background())
	require.NoError(t, err)
	assert.Equal(t, int64(0), version)

	// Close table
	err = manager.Close()
	require.NoError(t, err)
	assert.False(t, manager.IsOpen())

	// Open table again
	err = manager.OpenTable(context.Background())
	require.NoError(t, err)
	assert.True(t, manager.IsOpen())

	// Verify table version after reopen
	version, err = manager.GetTableVersion(context.Background())
	require.NoError(t, err)
	assert.Equal(t, int64(0), version)

	// Clean up
	manager.Close()
}

func TestTableManager_TableExists_Integration(t *testing.T) {
	tempDir := t.TempDir()
	log := zaptest.NewLogger(t)

	// Test non-existent table
	manager := NewTableManager(tempDir+"/non_existent_table", nil, log)
	exists, err := manager.TableExists(context.Background())
	require.NoError(t, err)
	assert.False(t, exists)

	// Create table directory
	tablePath := tempDir + "/test_table"
	err = os.MkdirAll(tablePath, 0755)
	require.NoError(t, err)

	// Create table
	manager2 := NewTableManager(tablePath, nil, log)
	err = manager2.CreateTable(context.Background())
	require.NoError(t, err)
	manager2.Close()

	// Check existence
	manager3 := NewTableManager(tablePath, nil, log)
	exists, err = manager3.TableExists(context.Background())
	require.NoError(t, err)
	assert.True(t, exists)
}

func TestTableManager_WriteToDeltaTable_Integration(t *testing.T) {
	tempDir := t.TempDir()
	manager := createTestTable(t, tempDir)
	defer manager.Close()

	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
		{
			SchemeVersion: "1.0",
			TenantID:      456,
			CostID:        "cost2",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "200",
			Cost:          2000,
			Labels:        `{"cluster":"test"}`,
			CreatedAt:     1704067200,
			SKUType:       "compute",
			HourDimension: "2024010100",
		},
	}

	// Write records
	err := manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	// Verify version incremented
	version, err := manager.GetTableVersion(context.Background())
	require.NoError(t, err)
	assert.Equal(t, int64(1), version)

	// Query to verify data (delta_table is the default table name in the query context)
	query := "SELECT * FROM delta_table"
	rows, err := manager.Query(context.Background(), query, nil)
	require.NoError(t, err)
	assert.Len(t, rows, 2)
}

func TestTableManager_WriteToDeltaTable_LogsEachRecordAsJSON(t *testing.T) {
	tempDir := t.TempDir()
	core, observedLogs := observer.New(zap.DebugLevel)
	logger := zap.New(core)
	tablePath := tempDir + "/test_table"

	err := os.MkdirAll(tablePath, 0755)
	require.NoError(t, err)

	manager := NewTableManager(tablePath, nil, logger)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)
	defer manager.Close()

	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
		{
			SchemeVersion: "1.0",
			TenantID:      456,
			CostID:        "cost2",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "200",
			Cost:          2000,
			Labels:        `{"cluster":"test"}`,
			CreatedAt:     1704067200,
			SKUType:       "compute",
			HourDimension: "2024010100",
		},
	}

	err = manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	entries := observedLogs.FilterMessage("Writing records to delta table").AllUntimed()
	require.Len(t, entries, len(records))

	for i, entry := range entries {
		fields := entry.ContextMap()
		assert.EqualValues(t, i+1, fields["record_index"])
		assert.EqualValues(t, len(records), fields["total_records"])
		assert.Equal(t, "Append", fields["mode"])

		recordJSON, err := json.Marshal(fields["record"])
		require.NoError(t, err)

		expectedJSON, err := json.Marshal(records[i])
		require.NoError(t, err)

		assert.JSONEq(t, string(expectedJSON), string(recordJSON))
	}
}

func TestTableManager_Query_Integration(t *testing.T) {
	tempDir := t.TempDir()
	manager := createTestTable(t, tempDir)
	defer manager.Close()

	// Write test data
	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
	}

	err := manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	// Test basic query (delta_table is the default table name)
	rows, err := manager.Query(context.Background(), "SELECT * FROM delta_table", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 1)

	// Test query with filter
	rows, err = manager.Query(context.Background(), "SELECT * FROM delta_table WHERE tenant_id = 123", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 1)

	// Test query with no results
	rows, err = manager.Query(context.Background(), "SELECT * FROM delta_table WHERE tenant_id = 999", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 0)

	// Test aggregation query
	rows, err = manager.Query(context.Background(), "SELECT COUNT(*) as cnt FROM delta_table", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 1)
}

func TestTableManager_Delete_Integration(t *testing.T) {
	tempDir := t.TempDir()
	manager := createTestTable(t, tempDir)
	defer manager.Close()

	// Write test data
	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
		{
			SchemeVersion: "1.0",
			TenantID:      456,
			CostID:        "cost2",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "200",
			Cost:          2000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
	}

	err := manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	// Delete one record
	result, err := manager.Delete(context.Background(), "tenant_id = 123")
	require.NoError(t, err)
	assert.NotNil(t, result)
	assert.Equal(t, int64(1), result.NumDeletedRows)

	// Verify remaining records (delta_table is the default table name)
	rows, err := manager.Query(context.Background(), "SELECT * FROM delta_table", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 1)
}

func TestTableManager_Compact_Integration(t *testing.T) {
	tempDir := t.TempDir()
	manager := createTestTable(t, tempDir)
	defer manager.Close()

	// Write multiple batches to create multiple files
	for i := 0; i < 3; i++ {
		records := []models.BillingReportV2Deltalake{
			{
				SchemeVersion: "1.0",
				TenantID:      uint64(100 + i),
				CostID:        "cost" + string(rune('0'+i)),
				CostTimestamp: 1704067200,
				Unit:          "GB",
				UnitValue:     "100",
				Cost:          1000,
				Labels:        "{}",
				CreatedAt:     1704067200,
				SKUType:       "storage",
				HourDimension: "2024010100",
			},
		}
		err := manager.WriteToDeltaTable(context.Background(), records, "Append")
		require.NoError(t, err)
	}

	// Run compact
	result, err := manager.Compact(context.Background())
	require.NoError(t, err)
	assert.NotNil(t, result)
	// Compact might not remove files if data is small
	t.Logf("Compact result: files_added=%d, files_removed=%d, partitions_optimized=%d",
		result.NumFilesAdded, result.NumFilesRemoved, result.PartitionsOptimized)
}

func TestTableManager_Vacuum_Integration(t *testing.T) {
	tempDir := t.TempDir()
	manager := createTestTable(t, tempDir)
	defer manager.Close()

	// Write some data
	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
	}
	err := manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	// Test dry run vacuum
	opts := &VacuumOptions{
		RetentionHours: 0, // Set to 0 for immediate cleanup
		DryRun:         true,
	}
	result, err := manager.Vacuum(context.Background(), opts)
	require.NoError(t, err)
	assert.NotNil(t, result)
	assert.True(t, result.DryRun)
	t.Logf("Vacuum dry run result: files_deleted=%d", result.TotalFilesDeleted)

	// Test actual vacuum (may not delete anything since files are recent)
	opts = &VacuumOptions{
		RetentionHours: 0,
		DryRun:         false,
	}
	result, err = manager.Vacuum(context.Background(), opts)
	require.NoError(t, err)
	assert.NotNil(t, result)
	assert.False(t, result.DryRun)
}

func TestTableManager_OpenTable_NonExistentPath(t *testing.T) {
	log := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := tempDir + "/non_existent_table"

	manager := NewTableManager(tablePath, nil, log)

	err := manager.OpenTable(context.Background())
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "failed to open delta table")
}

func TestTableManager_VacuumOptions_DefaultRetention(t *testing.T) {
	// This tests the default retention hours case
	opts := &VacuumOptions{
		DryRun: true,
	}
	// Retention hours should be 0 by default, but in the Vacuum function
	// it will only set if > 0
	assert.Equal(t, 0, opts.RetentionHours)
}

func TestTableManager_MultipleWrites_Integration(t *testing.T) {
	tempDir := t.TempDir()
	manager := createTestTable(t, tempDir)
	defer manager.Close()

	// Write multiple times with different modes
	records1 := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "storage",
			HourDimension: "2024010100",
		},
	}

	// Append mode
	err := manager.WriteToDeltaTable(context.Background(), records1, "Append")
	require.NoError(t, err)

	// Verify data (delta_table is the default table name)
	rows, err := manager.Query(context.Background(), "SELECT * FROM delta_table", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 1)

	// Append more records
	records2 := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      456,
			CostID:        "cost2",
			CostTimestamp: 1704067200,
			Unit:          "GB",
			UnitValue:     "200",
			Cost:          2000,
			Labels:        "{}",
			CreatedAt:     1704067200,
			SKUType:       "compute",
			HourDimension: "2024010100",
		},
	}

	err = manager.WriteToDeltaTable(context.Background(), records2, "Append")
	require.NoError(t, err)

	// Verify all data present (delta_table is the default table name)
	rows, err = manager.Query(context.Background(), "SELECT * FROM delta_table", nil)
	require.NoError(t, err)
	assert.Len(t, rows, 2)

	// Verify version
	version, err := manager.GetTableVersion(context.Background())
	require.NoError(t, err)
	assert.Equal(t, int64(2), version)
}

func TestPrepareStorageOptions_AlicloudNoRegion(t *testing.T) {
	// Setup Alicloud environment
	os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
	os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
	defer func() {
		os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
		os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
	}()

	conf := &config.ServiceConfig{
		Vendor: constants.VendorAlicloud,
		Bucket: "test-bucket",
		Region: "", // No region
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/test/path",
			BucketEndpoint: "https://oss-cn-hangzhou.aliyuncs.com",
			StorageOptions: nil, // nil storage options
		},
	}

	err := PrepareStorageOptions(conf)
	require.NoError(t, err)

	// Verify storage options were created
	assert.NotNil(t, conf.NextGenMeteringConfig.StorageOptions)
	assert.Equal(t, "test-key", conf.NextGenMeteringConfig.StorageOptions["AWS_ACCESS_KEY_ID"])
	// Region should not be set
	assert.Empty(t, conf.NextGenMeteringConfig.StorageOptions["region"])
	assert.Empty(t, conf.NextGenMeteringConfig.StorageOptions["AWS_REGION"])
}

func TestPrepareStorageOptions_AlicloudEmptyHost(t *testing.T) {
	// Setup Alicloud environment
	os.Setenv(constants.AlicloudAccessKeyIDEnv, "test-key")
	os.Setenv(constants.AlicloudAccessKeySecretEnv, "test-secret")
	defer func() {
		os.Unsetenv(constants.AlicloudAccessKeyIDEnv)
		os.Unsetenv(constants.AlicloudAccessKeySecretEnv)
	}()

	conf := &config.ServiceConfig{
		Vendor: constants.VendorAlicloud,
		Bucket: "test-bucket",
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/test/path",
			BucketEndpoint: "https://", // Empty host
			StorageOptions: nil,
		},
	}

	err := PrepareStorageOptions(conf)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "missing host")
}

func TestPrepareStorageOptions_Azure(t *testing.T) {
	tests := []struct {
		name       string
		conf       *config.ServiceConfig
		setupEnv   func()
		cleanupEnv func()
		wantErr    bool
		errMsg     string
	}{
		{
			name: "azure with account key",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAzure,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AzureStorageAccountNameEnv, "teststorageaccount")
				os.Setenv(constants.AzureStorageAccountKeyEnv, "test-account-key")
				os.Unsetenv(constants.AzureStorageSASTokenEnv)
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AzureStorageAccountNameEnv)
				os.Unsetenv(constants.AzureStorageAccountKeyEnv)
			},
			wantErr: false,
		},
		{
			name: "azure with SAS token",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAzure,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AzureStorageAccountNameEnv, "teststorageaccount")
				os.Unsetenv(constants.AzureStorageAccountKeyEnv)
				os.Setenv(constants.AzureStorageSASTokenEnv, "sv=2023-01-01&ss=b&srt=sco&sp=rwdlac")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AzureStorageAccountNameEnv)
				os.Unsetenv(constants.AzureStorageSASTokenEnv)
			},
			wantErr: false,
		},
		{
			name: "azure with managed identity (no key or SAS)",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAzure,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					StorageOptions: map[string]string{},
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AzureStorageAccountNameEnv, "teststorageaccount")
				os.Unsetenv(constants.AzureStorageAccountKeyEnv)
				os.Unsetenv(constants.AzureStorageSASTokenEnv)
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AzureStorageAccountNameEnv)
			},
			wantErr: false,
		},
		{
			name: "azure without account name",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAzure,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
				},
			},
			setupEnv: func() {
				os.Unsetenv(constants.AzureStorageAccountNameEnv)
			},
			wantErr: true,
			errMsg:  "environment variable is required for Azure",
		},
		{
			name: "azure with nil storage options",
			conf: &config.ServiceConfig{
				Vendor: constants.VendorAzure,
				NextGenMeteringConfig: &config.NextGenMeteringConfig{
					DeltaTablePath: "/test/path",
					StorageOptions: nil,
				},
			},
			setupEnv: func() {
				os.Setenv(constants.AzureStorageAccountNameEnv, "teststorageaccount")
				os.Setenv(constants.AzureStorageAccountKeyEnv, "test-key")
			},
			cleanupEnv: func() {
				os.Unsetenv(constants.AzureStorageAccountNameEnv)
				os.Unsetenv(constants.AzureStorageAccountKeyEnv)
			},
			wantErr: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if tt.setupEnv != nil {
				tt.setupEnv()
			}
			if tt.cleanupEnv != nil {
				defer tt.cleanupEnv()
			}

			err := PrepareStorageOptions(tt.conf)

			if tt.wantErr {
				require.Error(t, err)
				if tt.errMsg != "" {
					assert.Contains(t, err.Error(), tt.errMsg)
				}
			} else {
				require.NoError(t, err)
				opts := tt.conf.NextGenMeteringConfig.StorageOptions
				assert.NotNil(t, opts)
				assert.Equal(t, "teststorageaccount", opts["AZURE_STORAGE_ACCOUNT_NAME"])
			}
		})
	}
}

func TestPrepareStorageOptions_AzureWithAccountKey(t *testing.T) {
	os.Setenv(constants.AzureStorageAccountNameEnv, "mystorage")
	os.Setenv(constants.AzureStorageAccountKeyEnv, "mykey")
	os.Unsetenv(constants.AzureStorageSASTokenEnv)
	defer func() {
		os.Unsetenv(constants.AzureStorageAccountNameEnv)
		os.Unsetenv(constants.AzureStorageAccountKeyEnv)
	}()

	conf := &config.ServiceConfig{
		Vendor: constants.VendorAzure,
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/test/path",
			StorageOptions: map[string]string{},
		},
	}

	err := PrepareStorageOptions(conf)
	require.NoError(t, err)

	opts := conf.NextGenMeteringConfig.StorageOptions
	assert.Equal(t, "mystorage", opts["AZURE_STORAGE_ACCOUNT_NAME"])
	assert.Equal(t, "mykey", opts["AZURE_STORAGE_ACCOUNT_KEY"])
	_, hasSAS := opts["AZURE_STORAGE_SAS_TOKEN"]
	assert.False(t, hasSAS, "SAS token should not be set when only account key is provided")
}

func TestPrepareStorageOptions_AzureWithSASToken(t *testing.T) {
	os.Setenv(constants.AzureStorageAccountNameEnv, "mystorage")
	os.Unsetenv(constants.AzureStorageAccountKeyEnv)
	os.Setenv(constants.AzureStorageSASTokenEnv, "sv=2023-01-01&ss=b")
	defer func() {
		os.Unsetenv(constants.AzureStorageAccountNameEnv)
		os.Unsetenv(constants.AzureStorageSASTokenEnv)
	}()

	conf := &config.ServiceConfig{
		Vendor: constants.VendorAzure,
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/test/path",
			StorageOptions: map[string]string{},
		},
	}

	err := PrepareStorageOptions(conf)
	require.NoError(t, err)

	opts := conf.NextGenMeteringConfig.StorageOptions
	assert.Equal(t, "mystorage", opts["AZURE_STORAGE_ACCOUNT_NAME"])
	assert.Equal(t, "sv=2023-01-01&ss=b", opts["AZURE_STORAGE_SAS_TOKEN"])
	_, hasKey := opts["AZURE_STORAGE_ACCOUNT_KEY"]
	assert.False(t, hasKey, "Account key should not be set when only SAS token is provided")
}