package meta

import (
	"context"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"
)

func TestStorageNamespaceAppendLogBindingRoundTrip(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	now := time.Now().UTC()
	want := &StorageNamespace{
		ID:                    "append-binding-ns",
		OwnerTenantID:         "append-binding-tenant",
		Backend:               "s3",
		Bucket:                "normal-bucket",
		Prefix:                "tenant/append-binding-tenant/",
		AppendLogBindingState: AppendLogBindingBound,
		AppendLogBucket:       "directory-bucket--use1-az1--x-s3",
		AppendLogPrefix:       "/append-root",
		AppendLogRegion:       "us-east-1",
		State:                 StorageNamespaceActive,
		CreatedAt:             now,
		UpdatedAt:             now,
	}
	if err := s.UpsertStorageNamespace(ctx, want); err != nil {
		t.Fatal(err)
	}
	got, err := s.GetStorageNamespace(ctx, want.ID)
	if err != nil {
		t.Fatal(err)
	}
	if got.AppendLogBindingState != AppendLogBindingBound ||
		got.AppendLogBucket != want.AppendLogBucket ||
		got.AppendLogPrefix != "append-root/" ||
		got.AppendLogRegion != want.AppendLogRegion {
		t.Fatalf("append-log binding = %+v, want bound %q/%q/%q", got,
			want.AppendLogBucket, "append-root/", want.AppendLogRegion)
	}
}

func TestStorageNamespaceAppendLogBindingValidation(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	cases := []struct {
		name string
		ns   StorageNamespace
		want string
	}{
		{
			name: "unbound target",
			ns: StorageNamespace{
				ID: "unbound-target", OwnerTenantID: "tenant", Backend: "s3", Bucket: "normal", Prefix: "tenant/",
				AppendLogBindingState: AppendLogBindingUnbound, AppendLogBucket: "directory",
			},
			want: "unbound append-log namespace",
		},
		{
			name: "bound missing region",
			ns: StorageNamespace{
				ID: "bound-missing-region", OwnerTenantID: "tenant", Backend: "s3", Bucket: "normal", Prefix: "tenant/",
				AppendLogBindingState: AppendLogBindingBound, AppendLogBucket: "directory",
			},
			want: "requires bucket and region",
		},
		{
			name: "bound local backend",
			ns: StorageNamespace{
				ID: "bound-local", OwnerTenantID: "tenant", Backend: "local", Prefix: "tenant/",
				AppendLogBindingState: AppendLogBindingBound, AppendLogBucket: "directory", AppendLogRegion: "us-east-1",
			},
			want: "requires s3 storage backend",
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := s.UpsertStorageNamespace(ctx, &tc.ns)
			if err == nil || !strings.Contains(err.Error(), tc.want) {
				t.Fatalf("UpsertStorageNamespace error = %v, want %q", err, tc.want)
			}
		})
	}
}

func TestEnsureTenantStorageNamespaceStartsAppendLogUnbound(t *testing.T) {
	s := newControlStore(t)
	const tenantID = "append-binding-unbound"
	ns := newAppendLogBindingNamespace(t, s, tenantID)
	if ns.AppendLogBindingState != AppendLogBindingUnbound || ns.AppendLogBucket != "" || ns.AppendLogPrefix != "" || ns.AppendLogRegion != "" {
		t.Fatalf("new namespace append-log binding = %+v, want unbound", ns)
	}
}

func TestMigrateStorageNamespaceAddsAppendLogBindingColumns(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	const namespaceID = "append-binding-migrate"
	if err := s.UpsertStorageNamespace(ctx, &StorageNamespace{
		ID: namespaceID, OwnerTenantID: "tenant", Backend: "s3", Bucket: "normal-bucket", Prefix: "tenant/migrate/",
		State: StorageNamespaceActive,
	}); err != nil {
		t.Fatal(err)
	}
	for _, column := range []string{
		"append_log_binding_state",
		"append_log_bucket",
		"append_log_prefix",
		"append_log_region",
	} {
		if _, err := s.DB().ExecContext(ctx, "ALTER TABLE storage_namespaces DROP COLUMN "+column); err != nil {
			t.Fatalf("drop %s: %v", column, err)
		}
	}
	if err := s.Close(); err != nil {
		t.Fatal(err)
	}
	migrated, err := Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = migrated.Close() })
	ns, err := migrated.GetStorageNamespace(ctx, namespaceID)
	if err != nil {
		t.Fatal(err)
	}
	if ns.AppendLogBindingState != AppendLogBindingUnbound || ns.AppendLogBucket != "" || ns.AppendLogPrefix != "" || ns.AppendLogRegion != "" {
		t.Fatalf("migrated namespace append-log binding = %+v, want unbound", ns)
	}
}

func TestEnsureAppendLogStorageNamespaceBindingIsImmutable(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	ns := newAppendLogBindingNamespace(t, s, "append-binding-immutable")

	bound, err := s.EnsureAppendLogStorageNamespaceBinding(ctx, ns.ID, "directory-a", "/root", "us-east-1")
	if err != nil {
		t.Fatal(err)
	}
	if bound.AppendLogBindingState != AppendLogBindingBound || bound.AppendLogBucket != "directory-a" || bound.AppendLogPrefix != "root/" || bound.AppendLogRegion != "us-east-1" {
		t.Fatalf("bound namespace = %+v, want target A", bound)
	}
	if _, err := s.EnsureAppendLogStorageNamespaceBinding(ctx, ns.ID, "directory-a", "root/", "us-east-1"); err != nil {
		t.Fatalf("repeat same binding: %v", err)
	}
	if _, err := s.EnsureAppendLogStorageNamespaceBinding(ctx, ns.ID, "directory-b", "root/", "us-east-1"); !errors.Is(err, ErrAppendLogBindingMismatch) {
		t.Fatalf("replace binding error = %v, want ErrAppendLogBindingMismatch", err)
	}
	stored, err := s.GetStorageNamespace(ctx, ns.ID)
	if err != nil {
		t.Fatal(err)
	}
	if stored.AppendLogBucket != "directory-a" || stored.AppendLogPrefix != "root/" || stored.AppendLogRegion != "us-east-1" {
		t.Fatalf("stored binding = %+v, want target A", stored)
	}
}

func TestEnsureAppendLogStorageNamespaceBindingRejectsInactiveLifecycle(t *testing.T) {
	cases := []struct {
		name    string
		prepare func(context.Context, *Store, *StorageNamespace) error
	}{
		{
			name: "namespace deleting",
			prepare: func(ctx context.Context, s *Store, ns *StorageNamespace) error {
				return s.UpdateStorageNamespaceState(ctx, ns.ID, StorageNamespaceDeleting)
			},
		},
		{
			name: "namespace deleted",
			prepare: func(ctx context.Context, s *Store, ns *StorageNamespace) error {
				return s.UpdateStorageNamespaceState(ctx, ns.ID, StorageNamespaceDeleted)
			},
		},
		{
			name: "tenant suspended",
			prepare: func(ctx context.Context, s *Store, ns *StorageNamespace) error {
				return s.UpdateTenantStatus(ctx, ns.OwnerTenantID, TenantSuspended)
			},
		},
		{
			name: "tenant deleting",
			prepare: func(ctx context.Context, s *Store, ns *StorageNamespace) error {
				return s.UpdateTenantStatus(ctx, ns.OwnerTenantID, TenantDeleting)
			},
		},
		{
			name: "tenant deleted",
			prepare: func(ctx context.Context, s *Store, ns *StorageNamespace) error {
				return s.UpdateTenantStatus(ctx, ns.OwnerTenantID, TenantDeleted)
			},
		},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			s := newControlStore(t)
			ctx := context.Background()
			ns := newAppendLogBindingNamespace(t, s, "append-binding-inactive-"+strings.ReplaceAll(tc.name, " ", "-"))
			if err := tc.prepare(ctx, s, ns); err != nil {
				t.Fatal(err)
			}

			if _, err := s.EnsureAppendLogStorageNamespaceBinding(ctx, ns.ID, "directory-a", "root", "us-east-1"); !errors.Is(err, ErrNotFound) {
				t.Fatalf("bind inactive namespace error = %v, want ErrNotFound", err)
			}
			stored, err := s.GetStorageNamespace(ctx, ns.ID)
			if err != nil {
				t.Fatal(err)
			}
			if stored.AppendLogBindingState != AppendLogBindingUnbound {
				t.Fatalf("inactive namespace binding state = %q, want unbound", stored.AppendLogBindingState)
			}
		})
	}
}

func TestEnsureAppendLogStorageNamespaceBindingConcurrentTargets(t *testing.T) {
	s := newControlStore(t)
	ctx := context.Background()
	ns := newAppendLogBindingNamespace(t, s, "append-binding-concurrent")
	targets := []struct {
		bucket string
		prefix string
		region string
	}{
		{bucket: "directory-a", prefix: "a", region: "us-east-1"},
		{bucket: "directory-b", prefix: "b", region: "us-west-2"},
	}
	start := make(chan struct{})
	errs := make([]error, len(targets))
	var wg sync.WaitGroup
	for i := range targets {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			<-start
			_, errs[i] = s.EnsureAppendLogStorageNamespaceBinding(ctx, ns.ID, targets[i].bucket, targets[i].prefix, targets[i].region)
		}(i)
	}
	close(start)
	wg.Wait()

	succeeded := 0
	for _, err := range errs {
		if err == nil {
			succeeded++
			continue
		}
		if !errors.Is(err, ErrAppendLogBindingMismatch) {
			t.Fatalf("concurrent binding error = %v, want mismatch", err)
		}
	}
	if succeeded != 1 {
		t.Fatalf("successful concurrent bindings = %d, want 1 (errors=%v)", succeeded, errs)
	}
	stored, err := s.GetStorageNamespace(ctx, ns.ID)
	if err != nil {
		t.Fatal(err)
	}
	if stored.AppendLogBindingState != AppendLogBindingBound {
		t.Fatalf("stored binding state = %q, want bound", stored.AppendLogBindingState)
	}
	matched := false
	for _, target := range targets {
		if stored.AppendLogBucket == target.bucket && stored.AppendLogPrefix == target.prefix+"/" && stored.AppendLogRegion == target.region {
			matched = true
		}
	}
	if !matched {
		t.Fatalf("stored concurrent binding = %+v, want one target", stored)
	}
}

func newAppendLogBindingNamespace(t *testing.T, s *Store, tenantID string) *StorageNamespace {
	t.Helper()
	ctx := context.Background()
	now := time.Now().UTC()
	if err := s.InsertTenant(ctx, &Tenant{
		ID: tenantID, Status: TenantActive, Provider: tidbCloudNativeProvider,
		DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	ns, err := s.EnsureTenantStorageNamespace(ctx, tenantID, "s3", "normal-bucket", "tenant/"+tenantID+"/")
	if err != nil {
		t.Fatal(err)
	}
	return ns
}