package nextgenmaintenance

import (
	"context"
	"os"
	"testing"
	"time"

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

func TestNewNextGenMaintenance_ConfigValidation(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	_, err := NewNextGenMaintenance(nil, logger)
	assert.Error(t, err)

	_, err = NewNextGenMaintenance(&config.ServiceConfig{}, logger)
	assert.Error(t, err)

	_, err = NewNextGenMaintenance(&config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{}}, logger)
	assert.Error(t, err)

	_, err = NewNextGenMaintenance(&config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{MaintenanceConfig: &config.NextGenMaintenanceConfig{}}}, logger)
	assert.Error(t, err)

	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath:    "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{Enable: true},
	}}
	svc, err := NewNextGenMaintenance(conf, logger)
	assert.NoError(t, err)
	assert.NotNil(t, svc)
}

func TestNewNextGenMaintenance_Disabled(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath:    "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{Enable: false},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	assert.Error(t, err)
	assert.Nil(t, svc)
}

func TestResolveDeleteCutoff_RetentionDays(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath: "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{
			Enable:              true,
			EnableDelete:        true,
			DeleteRetentionDays: 30,
		},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	fixedNow := time.Date(2024, time.March, 1, 12, 0, 0, 0, time.UTC)
	svc.now = func() time.Time { return fixedNow }

	cutoff, err := svc.resolveDeleteCutoff()
	require.NoError(t, err)

	expectedCutoff := fixedNow.Add(-30 * 24 * time.Hour)
	assert.Equal(t, expectedCutoff, cutoff)
}

func TestResolveDeleteCutoff_AbsoluteTimestamp(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	cutoffTimestamp := int64(1700000000)
	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath: "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{
			Enable:                true,
			EnableDelete:          true,
			DeleteBeforeTimestamp: cutoffTimestamp,
		},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	// Set now to after the cutoff
	svc.now = func() time.Time { return time.Unix(cutoffTimestamp+3600, 0).UTC() }

	cutoff, err := svc.resolveDeleteCutoff()
	require.NoError(t, err)

	expectedCutoff := time.Unix(cutoffTimestamp, 0).UTC()
	assert.Equal(t, expectedCutoff, cutoff)
}

func TestResolveDeleteCutoff_FutureTimestampError(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	futureTimestamp := time.Now().Add(24 * time.Hour).Unix()
	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath: "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{
			Enable:                true,
			EnableDelete:          true,
			DeleteBeforeTimestamp: futureTimestamp,
		},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	_, err = svc.resolveDeleteCutoff()
	assert.Error(t, err)
}

func TestResolveDeleteCutoff_NoConfigError(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath: "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{
			Enable:       true,
			EnableDelete: true,
			// No DeleteBeforeTimestamp or DeleteRetentionDays
		},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	_, err = svc.resolveDeleteCutoff()
	assert.Error(t, err)
}

func TestBuildDeletePredicate(t *testing.T) {
	cutoff := time.Date(2024, time.January, 15, 10, 30, 0, 0, time.UTC)
	predicate := buildDeletePredicate(cutoff)

	expectedHour := utils.GenerateHourDimension(cutoff.Unix())
	expected := "hour_dimension < '" + expectedHour + "'"
	assert.Equal(t, expected, predicate)
}

func TestBuildDeletePredicate_DifferentTimes(t *testing.T) {
	tests := []struct {
		name   string
		cutoff time.Time
	}{
		{
			name:   "start of day",
			cutoff: time.Date(2024, time.March, 1, 0, 0, 0, 0, time.UTC),
		},
		{
			name:   "end of day",
			cutoff: time.Date(2024, time.March, 1, 23, 59, 59, 0, time.UTC),
		},
		{
			name:   "mid year",
			cutoff: time.Date(2024, time.June, 15, 12, 30, 45, 0, time.UTC),
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			predicate := buildDeletePredicate(tc.cutoff)
			expectedHour := utils.GenerateHourDimension(tc.cutoff.Unix())
			expected := "hour_dimension < '" + expectedHour + "'"
			assert.Equal(t, expected, predicate)
		})
	}
}

func TestNewNextGenMaintenance_InitializesTableManager(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath:    "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{Enable: true},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)
	assert.NotNil(t, svc.tableManager)
	assert.Equal(t, "s3://bucket/table", svc.tableManager.GetTablePath())
}

func TestNewNextGenMaintenance_UsesExistingTableManager(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	// Initialize global table manager first
	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath:    "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{Enable: true},
	}}

	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)

	// Create maintenance instance - should use existing manager
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)
	assert.Same(t, manager, svc.tableManager)
}

func TestRun_DeleteConfigValidation(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{NextGenMeteringConfig: &config.NextGenMeteringConfig{
		DeltaTablePath: "s3://bucket/table",
		MaintenanceConfig: &config.NextGenMaintenanceConfig{
			Enable:       true,
			EnableDelete: true,
			EnableVacuum: false,
			// Missing DeleteRetentionDays and DeleteBeforeTimestamp
		},
	}}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	// Test resolveDeleteCutoff directly - should fail because delete config is invalid
	_, err = svc.resolveDeleteCutoff()
	require.Error(t, err)
	assert.Contains(t, err.Error(), "deleteBeforeTimestamp or deleteRetentionDays")
}

// Integration tests for Run method

func createTestDeltaTable(t *testing.T, tempDir string) string {
	t.Helper()
	tablePath := tempDir + "/maintenance_test_table"

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

	return tablePath
}

func TestRun_Integration_AllOperationsEnabled(t *testing.T) {
	logger := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := createTestDeltaTable(t, tempDir)

	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: tablePath,
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable:               true,
				EnableDelete:         true,
				DeleteRetentionDays:  30,
				EnableVacuum:         true,
				VacuumRetentionHours: 0,
				VacuumDryRun:         true,
			},
		},
	}

	// Initialize table manager and create table
	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)

	// Write some test data
	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: time.Now().Add(-40 * 24 * time.Hour).Unix(), // 40 days ago
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     time.Now().Unix(),
			SKUType:       "storage",
			HourDimension: utils.GenerateHourDimension(time.Now().Add(-40 * 24 * time.Hour).Unix()),
		},
	}
	err = manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	// Create maintenance instance and run
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	err = svc.Run(context.Background())
	require.NoError(t, err)
}

func TestRun_Integration_DeleteDisabled(t *testing.T) {
	logger := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := createTestDeltaTable(t, tempDir)

	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: tablePath,
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable:       true,
				EnableDelete: false, // Delete disabled
				EnableVacuum: false, // Vacuum disabled
			},
		},
	}

	// Initialize table manager and create table
	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)

	// Create maintenance instance and run
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	err = svc.Run(context.Background())
	require.NoError(t, err)
}

func TestRun_Integration_VacuumDisabled(t *testing.T) {
	logger := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := createTestDeltaTable(t, tempDir)

	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: tablePath,
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable:              true,
				EnableDelete:        true,
				DeleteRetentionDays: 30,
				EnableVacuum:        false, // Vacuum disabled
			},
		},
	}

	// Initialize table manager and create table
	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)

	// Write some test data (required before delete operation)
	records := []models.BillingReportV2Deltalake{
		{
			SchemeVersion: "1.0",
			TenantID:      123,
			CostID:        "cost1",
			CostTimestamp: time.Now().Add(-40 * 24 * time.Hour).Unix(),
			Unit:          "GB",
			UnitValue:     "100",
			Cost:          1000,
			Labels:        "{}",
			CreatedAt:     time.Now().Unix(),
			SKUType:       "storage",
			HourDimension: utils.GenerateHourDimension(time.Now().Add(-40 * 24 * time.Hour).Unix()),
		},
	}
	err = manager.WriteToDeltaTable(context.Background(), records, "Append")
	require.NoError(t, err)

	// Create maintenance instance and run
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	err = svc.Run(context.Background())
	require.NoError(t, err)
}

func TestRun_Integration_VacuumWithDefaultRetention(t *testing.T) {
	logger := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := createTestDeltaTable(t, tempDir)

	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: tablePath,
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable:               true,
				EnableDelete:         false,
				EnableVacuum:         true,
				VacuumRetentionHours: 0, // Should default to 168 (7 days)
				VacuumDryRun:         false,
			},
		},
	}

	// Initialize table manager and create table
	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)

	// Create maintenance instance and run
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	err = svc.Run(context.Background())
	require.NoError(t, err)
}

func TestRun_OpenTableError(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	// Use a non-existent path that will fail to open
	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/non/existent/path/that/should/fail",
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable:       true,
				EnableDelete: false,
				EnableVacuum: false,
			},
		},
	}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	// Run should fail because table cannot be opened
	err = svc.Run(context.Background())
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed to open Delta Lake table")
}

func TestRun_InvalidDeleteConfig(t *testing.T) {
	logger := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := createTestDeltaTable(t, tempDir)

	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: tablePath,
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable:       true,
				EnableDelete: true,
				// Missing DeleteRetentionDays and DeleteBeforeTimestamp
				EnableVacuum: false,
			},
		},
	}

	// Initialize table manager and create table
	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)

	// Create maintenance instance and run
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	// Run should fail because delete config is invalid
	err = svc.Run(context.Background())
	require.Error(t, err)
	assert.Contains(t, err.Error(), "deleteBeforeTimestamp or deleteRetentionDays")
}

func TestGetTableVersion_Integration(t *testing.T) {
	logger := zaptest.NewLogger(t)
	tempDir := t.TempDir()
	tablePath := createTestDeltaTable(t, tempDir)

	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: tablePath,
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable: true,
			},
		},
	}

	// Initialize table manager and create table
	manager, err := deltautils.InitGlobalTableManager(conf, logger)
	require.NoError(t, err)
	err = manager.CreateTable(context.Background())
	require.NoError(t, err)

	// Create maintenance instance
	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

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

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

	// Get version again - should be incremented
	version, err = svc.GetTableVersion(context.Background())
	require.NoError(t, err)
	assert.Equal(t, int64(1), version)
}

func TestGetTableVersion_OpenTableError(t *testing.T) {
	logger := zaptest.NewLogger(t)
	deltautils.ResetGlobalTableManager()
	defer deltautils.ResetGlobalTableManager()

	conf := &config.ServiceConfig{
		NextGenMeteringConfig: &config.NextGenMeteringConfig{
			DeltaTablePath: "/non/existent/path",
			MaintenanceConfig: &config.NextGenMaintenanceConfig{
				Enable: true,
			},
		},
	}

	svc, err := NewNextGenMaintenance(conf, logger)
	require.NoError(t, err)

	// GetTableVersion should fail because table cannot be opened
	_, err = svc.GetTableVersion(context.Background())
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed to open Delta Lake table")
}