package lake

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestFakeClientLifecycleAndObservation(t *testing.T) {
	fake := NewFakeClient()
	fake.SetClock(func() time.Time { return testNow })
	datasourceSpec := testDataSourceSpec("test-datasource")
	taskSpec := testTaskSpec("test-task", "test-warehouse")

	require.NoError(t, fake.TestDataSource(context.Background(), testInvocation, datasourceSpec))
	require.NoError(t, fake.CreateDataSource(context.Background(), testInvocation, datasourceSpec))
	taskSpec.DataSourceName = datasourceSpec.Name
	require.NoError(t, fake.CreateTask(context.Background(), testInvocation, taskSpec))
	task, err := fake.GetTask(context.Background(), testInvocation, taskSpec.Name)
	require.NoError(t, err)
	assert.Equal(t, TaskStatusStopped, task.Status)
	assert.Equal(t, "test-warehouse", task.Warehouse)

	require.NoError(t, fake.StartTask(context.Background(), testInvocation, taskSpec.Name))
	resolvedTS := uint64(testNow.Add(-2*time.Second).UnixMilli()) << 18
	fake.PutTaskRun(TaskRun{
		ExecuteID: "test-run", TaskID: 1, Status: TaskRunStatusRunning,
		Progress: TiDBProgress{LastResolvedTS: resolvedTS, RowsSynced: 12}, CreatedAt: testNow,
	})
	observation, err := fake.ObserveTask(context.Background(), testInvocation, taskSpec.Name)
	require.NoError(t, err)
	assert.Equal(t, TaskStatusRunning, observation.TaskStatus)
	assert.Equal(t, resolvedTS, observation.LastResolvedTS)
	assert.Equal(t, testNow.Add(-2*time.Second), observation.LastResolvedAt)
	assert.Equal(t, "test-run", observation.LatestRunID)

	foundTask, err := fake.GetTask(context.Background(), testInvocation, "test-task")
	require.NoError(t, err)
	assert.Equal(t, "test-task", foundTask.Name)
	foundDataSource, err := fake.GetDataSource(context.Background(), testInvocation, "test-datasource")
	require.NoError(t, err)
	assert.Equal(t, "test-datasource", foundDataSource.Name)

	require.NoError(t, fake.StopTask(context.Background(), testInvocation, taskSpec.Name))
	require.NoError(t, fake.DropTask(context.Background(), testInvocation, taskSpec.Name))
	_, err = fake.GetTask(context.Background(), testInvocation, taskSpec.Name)
	var apiErr *APIError
	require.ErrorAs(t, err, &apiErr)
	assert.Equal(t, 404, apiErr.StatusCode)
	require.NoError(t, fake.DeleteDataSource(context.Background(), testInvocation, datasourceSpec.Name))
}

func TestFakeClientFaultInjectionAndDatabaseValidation(t *testing.T) {
	fake := NewFakeClient()
	fake.FailNext(OperationTestDataSource, errors.New("test injected failure"))
	err := fake.TestDataSource(context.Background(), testInvocation, testDataSourceSpec("test-datasource"))
	assert.EqualError(t, err, "test injected failure")
	require.NoError(t, fake.TestDataSource(context.Background(), testInvocation, testDataSourceSpec("test-datasource")))
	assert.Equal(t, []Operation{OperationTestDataSource, OperationTestDataSource}, fake.Calls())

	// A datasource still needs a name and a bucket; a spec without one never
	// reaches the wire.
	missingName := testDataSourceSpec("unnamed")
	missingName.Name = ""
	require.Error(t, fake.TestDataSource(context.Background(), testInvocation, missingName))
}

func TestFakeClientReturnsCopies(t *testing.T) {
	fake := NewFakeClient()
	err := fake.CreateDataSource(context.Background(), testInvocation, testDataSourceSpec("test-datasource"))
	require.NoError(t, err)
	first, err := fake.GetDataSource(context.Background(), testInvocation, "test-datasource")
	require.NoError(t, err)
	assert.True(t, first.S3.AccessKey.Empty())
	assert.True(t, first.S3.SecretKey.Empty())
	assert.True(t, first.S3.ExternalID.Empty())
	first.S3.Bucket = "mutated"
	second, err := fake.GetDataSource(context.Background(), testInvocation, "test-datasource")
	require.NoError(t, err)
	assert.Equal(t, "test-bucket", second.S3.Bucket)
}

func TestDeterministicResourceNames(t *testing.T) {
	first, err := DeterministicResourceNames("pipeline/test/001")
	require.NoError(t, err)
	second, err := DeterministicResourceNames("pipeline/test/001")
	require.NoError(t, err)
	other, err := DeterministicResourceNames("pipeline-test-001")
	require.NoError(t, err)
	assert.Equal(t, first, second)
	assert.NotEqual(t, first, other, "hash suffix must prevent sanitized-name collisions")
	assert.Regexp(t, `^tidb-dp-ds-[A-Za-z0-9_-]+-[0-9a-f]{12}$`, first.DataSource)
	assert.Regexp(t, `^tidb-dp-task-[A-Za-z0-9_-]+-[0-9a-f]{12}$`, first.Task)
	_, err = DeterministicResourceNames(" ")
	assert.Error(t, err)
}

// TestObserveTaskSeparatesRunCountersFromTaskFacts pins the split the review
// asked for. /task/runs returns history, so a value that describes one
// execution must come from one execution, while a value that describes the
// task may be gathered from any run that reported it.
func TestObserveTaskSeparatesRunCountersFromTaskFacts(t *testing.T) {
	fake := NewFakeClient()
	fake.SetClock(func() time.Time { return testNow })
	spec := testDataSourceSpec("test-datasource")
	require.NoError(t, fake.CreateDataSource(context.Background(), testInvocation, spec))
	task := testTaskSpec("test-task", "test-warehouse")
	task.DataSourceName = spec.Name
	require.NoError(t, fake.CreateTask(context.Background(), testInvocation, task))

	// An old execution that finished with an error and a big row count.
	fake.PutTaskRun(TaskRun{
		ExecuteID: "old", TaskID: 1, Status: TaskRunStatusFail,
		CreatedAt: testNow.Add(-time.Hour),
		Progress: TiDBProgress{
			RowsSynced: 9000, TablesError: 3, LastResolvedTS: 100 << 18,
		},
	})
	// The dumpling run of the current execution: carries only the task facts.
	fake.PutTaskRun(TaskRun{
		ExecuteID: "dumpling", TaskID: 1, Status: TaskRunStatusSuccess,
		CreatedAt: testNow.Add(-time.Minute),
		Progress: TiDBProgress{
			FullSyncCompleted: true, BackupTSO: 150 << 18, SyncMode: SyncModeSnapshotAndCDC,
		},
	})
	// The CDC run of the current execution: carries only the counters.
	fake.PutTaskRun(TaskRun{
		ExecuteID: "cdc", TaskID: 1, Status: TaskRunStatusRunning, CreatedAt: testNow,
		Progress: TiDBProgress{
			RowsSynced: 12, InsertCount: 7, UpdateCount: 4, DeleteCount: 1,
			TablesActive: 2, TablesError: 0, LastResolvedTS: 200 << 18,
		},
	})

	observation, err := fake.ObserveTask(context.Background(), testInvocation, "test-task")
	require.NoError(t, err)

	// Facts about the task survive across runs.
	assert.True(t, observation.Progress.FullSyncCompleted,
		"the dumpling run reports completion once and the CDC run never repeats it")
	assert.Equal(t, uint64(150)<<18, observation.Progress.BackupTSO)
	assert.Equal(t, uint64(200)<<18, observation.LastResolvedTS, "resolved position only moves forward")

	// Counters describe the newest run only, so a long-finished failure cannot
	// pin the task to "erroring" or inflate the row count forever.
	assert.Equal(t, uint64(12), observation.Progress.RowsSynced)
	assert.Equal(t, int32(0), observation.Progress.TablesError)
	assert.Equal(t, int32(2), observation.Progress.TablesActive)
}

// TestDeleteDataSourceIsIdempotent pins that deleting something already absent
// is the desired end state, not a failure: a cleanup retry after a delete whose
// result was lost must converge.
func TestDeleteDataSourceIsIdempotent(t *testing.T) {
	fake := NewFakeClient()
	require.NoError(t, fake.DeleteDataSource(context.Background(), testInvocation, "never-existed"))
}

// TestFakeClientSetupInfoMirrorsTheContract keeps the fake from being easier
// than Lake: it must report both principals, and it must fill the external ID
// the way the HTTP client does - from the organization the token exchange
// resolves when the caller did not name one.
func TestFakeClientSetupInfoMirrorsTheContract(t *testing.T) {
	fake := NewFakeClient()

	info, err := fake.SetupInfo(context.Background(), testInvocation)
	require.NoError(t, err)
	require.Len(t, info.TrustPrincipals, 2,
		"a single principal would let a caller that handles one of them pass")
	require.NotEqual(t, info.TrustPrincipals[0].ARN, info.TrustPrincipals[1].ARN)
	require.NotEmpty(t, info.ExternalID)

	// An invocation that names no organization still gets one, as it would
	// through the token exchange.
	anonymous := Invocation{TiDBOrgID: "1369847559692509704", LakeOrgID: "7"}
	info, err = fake.SetupInfo(context.Background(), anonymous)
	require.NoError(t, err)
	require.Equal(t, anonymous.TiDBOrgID, info.ExternalID)

	fake.PutSetupInfo(SetupInfo{ExternalID: "override", TrustPrincipals: nil})
	info, err = fake.SetupInfo(context.Background(), testInvocation)
	require.NoError(t, err)
	require.Equal(t, "override", info.ExternalID)
	require.Empty(t, info.TrustPrincipals)
}

// TestFakeClientUpdateTaskResolvesTheDataSource pins the resolution the HTTP
// client performs. Without it a test could move a task to another datasource
// and watch it succeed, which production cannot do: the adapter resolves the
// name the spec carries and fails when it names nothing.
func TestFakeClientUpdateTaskResolvesTheDataSource(t *testing.T) {
	fake := NewFakeClient()
	ctx := context.Background()
	require.NoError(t, fake.CreateDataSource(ctx, testInvocation, testDataSourceSpec("first")))
	require.NoError(t, fake.CreateDataSource(ctx, testInvocation, testDataSourceSpec("second")))

	spec := testTaskSpec("test-task", "test-warehouse")
	spec.DataSourceName = "first"
	require.NoError(t, fake.CreateTask(ctx, testInvocation, spec))
	created, err := fake.GetTask(ctx, testInvocation, "test-task")
	require.NoError(t, err)

	// Moving the task to another datasource is reflected, not ignored.
	spec.DataSourceName = "second"
	require.NoError(t, fake.UpdateTask(ctx, testInvocation, "test-task", spec))
	moved, err := fake.GetTask(ctx, testInvocation, "test-task")
	require.NoError(t, err)
	require.NotEqual(t, created.DataSourceID, moved.DataSourceID)

	// A datasource that does not exist fails here, as it does on the wire.
	spec.DataSourceName = "absent"
	require.Error(t, fake.UpdateTask(ctx, testInvocation, "test-task", spec))
}