package engine

import (
	"context"
	"errors"
	"testing"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/stretchr/testify/require"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
)

type scriptedS3ObjectLister struct {
	outputs []*s3.ListObjectsV2Output
	errors  []error
	inputs  []*s3.ListObjectsV2Input
}

func (f *scriptedS3ObjectLister) ListObjectsV2(
	_ context.Context,
	input *s3.ListObjectsV2Input,
	_ ...func(*s3.Options),
) (*s3.ListObjectsV2Output, error) {
	f.inputs = append(f.inputs, input)
	call := len(f.inputs) - 1
	if call < len(f.errors) && f.errors[call] != nil {
		return nil, f.errors[call]
	}
	if call >= len(f.outputs) || f.outputs[call] == nil {
		return &s3.ListObjectsV2Output{}, nil
	}
	return f.outputs[call], nil
}

func statementsDeltaDataSource() *dto.DataSourceInfo {
	return &dto.DataSourceInfo{
		Bucket: "o11y-bucket",
		Path:   "deltalake/100/o11y-id/persisted_statements_summary/",
	}
}

func TestHasDeltaLogFindsTransactionLogObject(t *testing.T) {
	client := &scriptedS3ObjectLister{
		outputs: []*s3.ListObjectsV2Output{
			{
				Contents:              []types.Object{{Key: aws.String("deltalake/100/o11y-id/persisted_statements_summary/_delta_log/")}},
				IsTruncated:           aws.Bool(true),
				NextContinuationToken: aws.String("next-page"),
			},
			{
				Contents: []types.Object{{Key: aws.String("deltalake/100/o11y-id/persisted_statements_summary/_delta_log/00000000000000000000.json")}},
			},
		},
	}

	available, err := hasDeltaLog(context.Background(), client, statementsDeltaDataSource())
	require.NoError(t, err)
	require.True(t, available)
	require.Len(t, client.inputs, 2)
	require.Equal(t, "deltalake/100/o11y-id/persisted_statements_summary/_delta_log/", aws.ToString(client.inputs[0].Prefix))
	require.Empty(t, aws.ToString(client.inputs[0].ContinuationToken))
	require.Equal(t, "next-page", aws.ToString(client.inputs[1].ContinuationToken))
}

func TestHasDeltaLogReturnsFalseForEmptyTableRoot(t *testing.T) {
	client := &scriptedS3ObjectLister{
		outputs: []*s3.ListObjectsV2Output{
			{
				Contents: []types.Object{{Key: aws.String("deltalake/100/o11y-id/persisted_statements_summary/_delta_log/")}},
			},
			{
				Contents: []types.Object{
					{Key: aws.String("deltalake/100/o11y-id/persisted_statements_summary/")},
					{Key: aws.String("deltalake/100/o11y-id/persisted_statements_summary/_delta_log/")},
				},
			},
		},
	}

	available, err := hasDeltaLog(context.Background(), client, statementsDeltaDataSource())
	require.NoError(t, err)
	require.False(t, available)
	require.Len(t, client.inputs, 2)
	require.Equal(t, "deltalake/100/o11y-id/persisted_statements_summary/", aws.ToString(client.inputs[1].Prefix))
}

func TestHasDeltaLogRejectsObjectsWithoutTransactionLog(t *testing.T) {
	client := &scriptedS3ObjectLister{
		outputs: []*s3.ListObjectsV2Output{
			{},
			{
				Contents: []types.Object{{Key: aws.String("deltalake/100/o11y-id/persisted_statements_summary/date=20260721/part-000.parquet")}},
			},
		},
	}

	available, err := hasDeltaLog(context.Background(), client, statementsDeltaDataSource())
	require.False(t, available)
	require.ErrorIs(t, err, ErrInvalidDeltaSource)
	require.Contains(t, err.Error(), "_delta_log/")
}

func TestHasDeltaLogPreservesS3Error(t *testing.T) {
	wantErr := errors.New("AccessDenied")
	client := &scriptedS3ObjectLister{
		errors: []error{wantErr},
	}

	available, err := hasDeltaLog(context.Background(), client, statementsDeltaDataSource())
	require.False(t, available)
	require.ErrorIs(t, err, wantErr)
	require.Same(t, wantErr, err)
	require.Len(t, client.inputs, 1)
}