package tenant

import (
	"context"
	"errors"
	"net/url"
	"testing"
	"time"

	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/s3client"
	"github.com/mem9-ai/drive9/pkg/testtidb"
)

func TestS3ExpressForAppendLogNamespaceRequiresSafeNamespace(t *testing.T) {
	p := &Pool{cfg: PoolConfig{S3ExpressBucket: "directory-bucket", S3Region: "us-east-1"}}
	ns := &meta.StorageNamespace{
		ID:                    "tenant/a",
		AppendLogBindingState: meta.AppendLogBindingBound,
		AppendLogBucket:       "directory-bucket",
		AppendLogRegion:       "us-east-1",
	}
	if _, err := p.S3ExpressForAppendLogNamespace(context.Background(), ns); err == nil {
		t.Fatal("unsafe append-log namespace succeeded")
	}
}

func TestS3ExpressCleanupPrefixMatchesWriter(t *testing.T) {
	for _, prefix := range []string{"", "/", "///", "archive", "archive/", "archive//", "/archive///", "archive//nested//"} {
		t.Run(prefix, func(t *testing.T) {
			ctx := context.Background()
			p := &Pool{cfg: PoolConfig{
				S3ExpressBucket: "test-bucket", S3ExpressPrefix: prefix,
				S3Region: "us-east-1", S3AccessKeyID: "test-key", S3SecretAccessKey: "test-secret",
			}}
			ns := &meta.StorageNamespace{
				ID:                    "namespace-a",
				AppendLogBindingState: meta.AppendLogBindingBound,
				AppendLogBucket:       "test-bucket",
				AppendLogPrefix:       meta.NormalizeAppendLogBasePrefix(prefix),
				AppendLogRegion:       "us-east-1",
			}
			cleanup, err := p.S3ExpressForAppendLogNamespace(ctx, ns)
			if err != nil {
				t.Fatal(err)
			}
			writer, err := s3client.New(ctx, s3client.AWSConfig{
				Bucket: "test-bucket", Prefix: prefix, Region: "us-east-1",
				AccessKeyID: "test-key", SecretAccessKey: "test-secret",
			})
			if err != nil {
				t.Fatal(err)
			}
			// Presigning is local: compare the actual SDK keys without contacting AWS.
			writerURL, err := writer.PresignGetObject(ctx, "append-log/namespace-a/generation", time.Minute)
			if err != nil {
				t.Fatal(err)
			}
			cleanupURL, err := cleanup.PresignGetObject(ctx, "generation", time.Minute)
			if err != nil {
				t.Fatal(err)
			}
			written, err := url.Parse(writerURL)
			if err != nil {
				t.Fatal(err)
			}
			cleaned, err := url.Parse(cleanupURL)
			if err != nil {
				t.Fatal(err)
			}
			if written.Path != cleaned.Path {
				t.Fatalf("writer key=%q cleanup key=%q", written.Path, cleaned.Path)
			}
			if prefix == "archive//" && written.Path != "/archive//append-log/namespace-a/generation" {
				t.Fatalf("existing writer key changed: %q", written.Path)
			}
		})
	}
}

func TestS3ExpressForAppendLogNamespaceIsNoopWhenDisabled(t *testing.T) {
	p := &Pool{}
	client, err := p.S3ExpressForAppendLogNamespace(context.Background(), &meta.StorageNamespace{
		ID: "namespace-a", AppendLogBindingState: meta.AppendLogBindingUnbound,
	})
	if err != nil {
		t.Fatalf("disabled append-log cleanup client: %v", err)
	}
	if client != nil {
		t.Fatalf("disabled append-log cleanup client = %T, want nil", client)
	}
}

func TestS3ExpressForBoundAppendLogNamespaceFailsClosedOnConfigMismatch(t *testing.T) {
	ns := &meta.StorageNamespace{
		ID:                    "namespace-a",
		AppendLogBindingState: meta.AppendLogBindingBound,
		AppendLogBucket:       "directory-a",
		AppendLogPrefix:       "root/",
		AppendLogRegion:       "us-east-1",
	}
	for _, tc := range []struct {
		name string
		cfg  PoolConfig
	}{
		{name: "disabled", cfg: PoolConfig{}},
		{name: "bucket mismatch", cfg: PoolConfig{S3ExpressBucket: "directory-b", S3ExpressPrefix: "root/", S3Region: "us-east-1"}},
		{name: "prefix mismatch", cfg: PoolConfig{S3ExpressBucket: "directory-a", S3ExpressPrefix: "other", S3Region: "us-east-1"}},
		{name: "region mismatch", cfg: PoolConfig{S3ExpressBucket: "directory-a", S3ExpressPrefix: "root/", S3Region: "us-west-2"}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			p := &Pool{cfg: tc.cfg}
			if _, err := p.S3ExpressForAppendLogNamespace(context.Background(), ns); err == nil {
				t.Fatal("bound cleanup client succeeded with mismatched configuration")
			}
		})
	}
}

func TestAppendLogClientForStorageNamespaceBindsThenRejectsChangedConfig(t *testing.T) {
	ctx := context.Background()
	metaStore, err := meta.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = metaStore.Close() })
	testtidb.ResetMetaDB(t, metaStore.DB())
	now := time.Now().UTC()
	const tenantID = "append-log-binding-pool"
	if err := metaStore.InsertTenant(ctx, &meta.Tenant{
		ID: tenantID, Status: meta.TenantActive, Provider: ProviderTiDBCloudNative,
		DBPasswordCipher: []byte{}, CreatedAt: now, UpdatedAt: now,
	}); err != nil {
		t.Fatal(err)
	}
	ns, err := metaStore.EnsureTenantStorageNamespace(ctx, tenantID, "s3", "normal-bucket", "tenant/append-log-binding-pool/")
	if err != nil {
		t.Fatal(err)
	}
	p := &Pool{cfg: PoolConfig{
		S3ExpressBucket: "directory-a", S3ExpressPrefix: "/root", S3Region: "us-east-1",
		S3AccessKeyID: "test-key", S3SecretAccessKey: "test-secret",
	}, metaStore: metaStore}
	client, err := p.appendLogClientForStorageNamespace(ctx, ns)
	if err != nil {
		t.Fatal(err)
	}
	if client == nil {
		t.Fatal("append-log client = nil after binding")
	}
	bound, err := metaStore.GetStorageNamespace(ctx, ns.ID)
	if err != nil {
		t.Fatal(err)
	}
	if bound.AppendLogBindingState != meta.AppendLogBindingBound || bound.AppendLogBucket != "directory-a" || bound.AppendLogPrefix != "root/" || bound.AppendLogRegion != "us-east-1" {
		t.Fatalf("bound namespace = %+v, want target A", bound)
	}
	p.cfg.S3ExpressBucket = "directory-b"
	client, err = p.appendLogClientForStorageNamespace(ctx, bound)
	if err != nil {
		t.Fatal(err)
	}
	if client != nil {
		t.Fatalf("client with changed configuration = %T, want nil", client)
	}
	stillBound, err := metaStore.GetStorageNamespace(ctx, ns.ID)
	if err != nil {
		t.Fatal(err)
	}
	if stillBound.AppendLogBucket != "directory-a" || stillBound.AppendLogPrefix != "root/" || stillBound.AppendLogRegion != "us-east-1" {
		t.Fatalf("changed config rewrote binding: %+v", stillBound)
	}

	p.cfg.S3ExpressBucket = "directory-a"
	if err := metaStore.UpdateTenantStatus(ctx, tenantID, meta.TenantDeleting); err != nil {
		t.Fatal(err)
	}
	if _, err := p.appendLogClientForStorageNamespace(ctx, bound); !errors.Is(err, meta.ErrNotFound) {
		t.Fatalf("stale bound namespace client error = %v, want ErrNotFound", err)
	}
}