package engine

import (
	"context"
	"errors"
	"sync"
	"testing"

	"google.golang.org/api/iterator"

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

type fakeGCSObjectNameIterator struct {
	names []string
	err   error
	index int
}

func (it *fakeGCSObjectNameIterator) Next() (string, error) {
	if it.index < len(it.names) {
		name := it.names[it.index]
		it.index++
		return name, nil
	}
	if it.err != nil {
		err := it.err
		it.err = nil
		return "", err
	}
	return "", iterator.Done
}

func TestInspectGCSDeltaRoot(t *testing.T) {
	root := "deltalake/100/o11y/persisted_statements_summary/"
	deltaRoot := root + "_delta_log/"
	tests := []struct {
		name          string
		deltaObjects  []string
		deltaTerminal error
		rootObjects   []string
		rootTerminal  error
		available     bool
		wantErr       error
	}{
		{
			name: "empty root",
		},
		{
			name: "directory markers are empty",
			deltaObjects: []string{
				deltaRoot,
			},
			rootObjects: []string{
				root,
				deltaRoot,
				root + "date=20260721/",
			},
		},
		{
			name: "commit json is a delta log",
			deltaObjects: []string{
				deltaRoot + "00000000000000000000.json",
			},
			available: true,
		},
		{
			name: "unrecognized metadata before commit does not hide valid delta log",
			deltaObjects: []string{
				deltaRoot + "README",
				deltaRoot + "00000000000000000000.json",
			},
			available: true,
		},
		{
			name: "checkpoint is a delta log",
			deltaObjects: []string{
				deltaRoot + "00000000000000000010.checkpoint.0000000001.0000000002.parquet",
			},
			available: true,
		},
		{
			name: "last checkpoint is a delta log",
			deltaObjects: []string{
				deltaRoot + "_last_checkpoint",
			},
			available: true,
		},
		{
			name: "data without transaction log is invalid",
			rootObjects: []string{
				root + "date=20260721/part-00000.parquet",
			},
			wantErr: ErrInvalidDeltaSource,
		},
		{
			name: "unrecognized delta log object is invalid",
			deltaObjects: []string{
				deltaRoot + "README",
			},
			wantErr: ErrInvalidDeltaSource,
		},
		{
			name:          "permission error is preserved",
			deltaTerminal: errTestGCSAccessDenied,
			wantErr:       errTestGCSAccessDenied,
		},
		{
			name:         "timeout while checking an otherwise empty root is preserved",
			rootTerminal: context.DeadlineExceeded,
			wantErr:      context.DeadlineExceeded,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			available, err := inspectGCSDeltaRoot(
				&fakeGCSObjectNameIterator{names: tt.deltaObjects, err: tt.deltaTerminal},
				&fakeGCSObjectNameIterator{names: tt.rootObjects, err: tt.rootTerminal},
				deltaRoot,
			)
			if available != tt.available {
				t.Fatalf("available = %v, want %v", available, tt.available)
			}
			if tt.wantErr == nil && err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
			if tt.wantErr != nil && !errors.Is(err, tt.wantErr) {
				t.Fatalf("error = %v, want errors.Is(_, %v)", err, tt.wantErr)
			}
			if tt.deltaTerminal != nil && err != tt.deltaTerminal {
				t.Fatalf("delta list error = %v, want original error %v", err, tt.deltaTerminal)
			}
			if tt.rootTerminal != nil && err != tt.rootTerminal {
				t.Fatalf("root list error = %v, want original error %v", err, tt.rootTerminal)
			}
		})
	}
}

var errTestGCSAccessDenied = errors.New("gcs access denied")

type fakeGCSObjectLister struct {
	mu           sync.Mutex
	objects      map[string][]string
	prefixes     map[string][]string
	prefixErrors map[string]error
	listCalls    int
	prefixCalls  int
	lastBucket   string
	lastPrefix   string
}

func (l *fakeGCSObjectLister) List(_ context.Context, bucket, prefix string) gcsObjectNameIterator {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.listCalls++
	l.lastBucket = bucket
	l.lastPrefix = prefix
	return &fakeGCSObjectNameIterator{names: append([]string(nil), l.objects[prefix]...)}
}

func (l *fakeGCSObjectLister) ListPrefixes(_ context.Context, bucket, prefix string) gcsObjectNameIterator {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.prefixCalls++
	l.lastBucket = bucket
	l.lastPrefix = prefix
	return &fakeGCSObjectNameIterator{names: append([]string(nil), l.prefixes[prefix]...), err: l.prefixErrors[prefix]}
}

func TestBigQueryHasDeltaLogInitializesGCSLazilyOnce(t *testing.T) {
	lister := &fakeGCSObjectLister{
		objects: map[string][]string{
			"deltalake/100/o11y/persisted_statements_summary/_delta_log/": {
				"deltalake/100/o11y/persisted_statements_summary/_delta_log/00000000000000000000.json",
			},
		},
	}
	factoryCalls := 0
	queryEngine := &BigQueryEngine{
		newGCSObjectLister: func(context.Context) (gcsObjectLister, error) {
			factoryCalls++
			return lister, nil
		},
	}
	dataSource := &dto.DataSourceInfo{
		Bucket: "o11y-prod-shared-us-central1",
		Path:   "/deltalake/100/o11y/persisted_statements_summary/",
	}

	for range 2 {
		available, err := queryEngine.HasDeltaLog(context.Background(), dataSource)
		if err != nil || !available {
			t.Fatalf("HasDeltaLog() = %v, %v; want true, nil", available, err)
		}
	}

	if factoryCalls != 1 {
		t.Fatalf("GCS factory calls = %d, want 1", factoryCalls)
	}
	if lister.listCalls != 4 {
		t.Fatalf("GCS list calls = %d, want 4", lister.listCalls)
	}
	if lister.lastBucket != dataSource.Bucket {
		t.Fatalf("bucket = %q, want %q", lister.lastBucket, dataSource.Bucket)
	}
	wantPrefix := "deltalake/100/o11y/persisted_statements_summary/"
	if lister.lastPrefix != wantPrefix {
		t.Fatalf("prefix = %q, want %q", lister.lastPrefix, wantPrefix)
	}
}

func TestBigQueryHasDeltaLogPreservesFactoryError(t *testing.T) {
	wantErr := errors.New("ADC unavailable")
	queryEngine := &BigQueryEngine{
		newGCSObjectLister: func(context.Context) (gcsObjectLister, error) {
			return nil, wantErr
		},
	}

	available, err := queryEngine.HasDeltaLog(context.Background(), &dto.DataSourceInfo{Bucket: "bucket"})
	if available || !errors.Is(err, wantErr) {
		t.Fatalf("HasDeltaLog() = %v, %v; want false, %v", available, err, wantErr)
	}
}

func TestBigQueryHasDeltaLogAddsLocationToInvalidDeltaError(t *testing.T) {
	lister := &fakeGCSObjectLister{
		objects: map[string][]string{
			"root/": {"root/date=20260721/part-00000.parquet"},
		},
	}
	queryEngine := &BigQueryEngine{
		gcsObjectLister: lister,
	}

	available, err := queryEngine.HasDeltaLog(context.Background(), &dto.DataSourceInfo{
		Bucket: "test-bucket",
		Path:   "root",
	})
	if available || !errors.Is(err, ErrInvalidDeltaSource) {
		t.Fatalf("HasDeltaLog() = %v, %v; want false, ErrInvalidDeltaSource", available, err)
	}
	if got := err.Error(); got != ErrInvalidDeltaSource.Error()+" at gs://test-bucket/root" {
		t.Fatalf("error = %q", got)
	}
}