package gcp

import (
	"context"
	"errors"
	"fmt"
	"os"
	"testing"

	"github.com/tidbcloud/observability/pkg/common/message"

	"cloud.google.com/go/iam"
	"cloud.google.com/go/storage"
	. "github.com/onsi/gomega"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/gcp/gcs"
	mocks "github.com/tidbcloud/observability/test/mockresourceworker/providers/gcp/gcs"
	"go.uber.org/mock/gomock"
	pb "google.golang.org/genproto/googleapis/iam/v1"
	"google.golang.org/genproto/googleapis/type/expr"
)

func newMockGCS(mock gcs.CloudStorageClient) *CloudStorageClient {
	return &CloudStorageClient{
		cli:     mock,
		project: "project",
		region:  "region",
	}
}

func TestGCSNew(t *testing.T) {
	g := NewGomegaWithT(t)
	os.Setenv("GOOGLE_APPLICATION_CREDENTIALS", "../../../../test/files/application_default_credentials.json")
	defer os.Unsetenv("GOOGLE_APPLICATION_CREDENTIALS")
	g.Expect(NewCloudStorage("project1", "region")).Should(Equal(&CloudStorageClient{
		cli:     gcs.StorageClientWarp{Client: cscli},
		project: "project1",
		region:  "region",
	}))
}

func TestGCSCreateBucket(t *testing.T) {
	g := NewGomegaWithT(t)
	ctl := gomock.NewController(t)
	defer ctl.Finish()
	mockclient := mocks.NewMockCloudStorageClient(ctl)
	handleclient := mocks.NewMockBucketHandle(ctl)
	gcsclient := newMockGCS(mockclient)
	//successs
	mockclient.EXPECT().Bucket(gomock.Any()).Return(handleclient).AnyTimes()
	handleclient.EXPECT().Create(gomock.Any(), gcsclient.project, &storage.BucketAttrs{
		Location: gcsclient.region,
		UniformBucketLevelAccess: storage.UniformBucketLevelAccess{
			Enabled: true,
		},
	}).Return(nil)
	g.Expect(gcsclient.CreateBucket(context.Background(), "test-1")).Should(Succeed())
	//error
	handleclient.EXPECT().Create(gomock.Any(), gcsclient.project, &storage.BucketAttrs{
		Location: gcsclient.region,
		UniformBucketLevelAccess: storage.UniformBucketLevelAccess{
			Enabled: true,
		},
	}).Return(errors.New(ErrNotFound))
	g.Expect(gcsclient.CreateBucket(context.Background(), "test-1")).Should(HaveOccurred())
}

func TestGCSBucketExists(t *testing.T) {
	g := NewGomegaWithT(t)
	ctl := gomock.NewController(t)
	defer ctl.Finish()
	mockclient := mocks.NewMockCloudStorageClient(ctl)
	handleclient := mocks.NewMockBucketHandle(ctl)
	gcsclient := newMockGCS(mockclient)
	//bucket exists
	mockclient.EXPECT().Bucket(gomock.Any()).Return(handleclient).AnyTimes()
	handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{}, nil)
	g.Expect(gcsclient.BucketExists(context.Background(), "test-1")).Should(BeTrue())
	//error
	handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{}, errors.New("errors"))
	g.Expect(gcsclient.BucketExists(context.Background(), "test-1")).Error().Should(HaveOccurred())
	//not found
	handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{}, storage.ErrBucketNotExist)
	exist, err := gcsclient.BucketExists(context.Background(), "test-1")
	g.Expect(exist).Should(BeFalse())
	g.Expect(err).Should(Succeed())
}

func TestGCSSetBucketPolicy(t *testing.T) {
	g := NewGomegaWithT(t)
	ctl := gomock.NewController(t)
	defer ctl.Finish()
	mockclient := mocks.NewMockCloudStorageClient(ctl)
	handleclient := mocks.NewMockBucketHandle(ctl)
	iamhandleclient := mocks.NewMockIamHandle3(ctl)
	gcsclient := newMockGCS(mockclient)
	//setbucket policy success
	mockclient.EXPECT().Bucket(gomock.Any()).Return(handleclient).AnyTimes()
	handleclient.EXPECT().IAMV3().Return(iamhandleclient).AnyTimes()
	iamhandleclient.EXPECT().SetPolicy(gomock.Any(), gomock.Any()).Return(nil)
	g.Expect(gcsclient.SetBucketPolicy(context.Background(), "test-1", &iam.Policy3{})).Should(Succeed())
	//error
	iamhandleclient.EXPECT().SetPolicy(gomock.Any(), gomock.Any()).Return(errors.New("errors"))
	g.Expect(gcsclient.SetBucketPolicy(context.Background(), "test-1", &iam.Policy3{})).Should(HaveOccurred())
}

func TestGCSGetBucketPolicy(t *testing.T) {
	g := NewGomegaWithT(t)
	ctl := gomock.NewController(t)
	defer ctl.Finish()
	mockclient := mocks.NewMockCloudStorageClient(ctl)
	handleclient := mocks.NewMockBucketHandle(ctl)
	iamhandleclient := mocks.NewMockIamHandle3(ctl)
	gcsclient := newMockGCS(mockclient)
	//setbucket policy success
	mockclient.EXPECT().Bucket(gomock.Any()).Return(handleclient).AnyTimes()
	handleclient.EXPECT().IAMV3().Return(iamhandleclient).AnyTimes()
	iamhandleclient.EXPECT().Policy(gomock.Any()).Return(&iam.Policy3{}, nil)
	policy, exists, err := gcsclient.GetBucketPolicy(context.Background(), "test-1")
	g.Expect(policy).Should(Equal(&iam.Policy3{}))
	g.Expect(exists).Should(BeTrue())
	g.Expect(err).Should(Succeed())
	//error
	iamhandleclient.EXPECT().Policy(gomock.Any()).Return(nil, errors.New("error"))
	g.Expect(gcsclient.GetBucketPolicy(context.Background(), "test-1")).Error().Should(HaveOccurred())
}

func TestGCSDeleteBucketPolicy(t *testing.T) {
	g := NewGomegaWithT(t)
	ctl := gomock.NewController(t)
	defer ctl.Finish()
	mockclient := mocks.NewMockCloudStorageClient(ctl)
	handleclient := mocks.NewMockBucketHandle(ctl)
	iamhandleclient := mocks.NewMockIamHandle3(ctl)
	gcsclient := newMockGCS(mockclient)
	//setbucket policy success
	mockclient.EXPECT().Bucket(gomock.Any()).Return(handleclient).AnyTimes()
	handleclient.EXPECT().IAMV3().Return(iamhandleclient).AnyTimes()
	iamhandleclient.EXPECT().Policy(gomock.Any()).Return(&iam.Policy3{
		Bindings: []*pb.Binding{
			{
				Condition: &expr.Expr{
					Title: "p1",
				},
			},
			{
				Condition: &expr.Expr{
					Title: "p2",
				},
			},
		},
	}, nil)
	iamhandleclient.EXPECT().SetPolicy(gomock.Any(), &iam.Policy3{
		Bindings: []*pb.Binding{
			{
				Condition: &expr.Expr{
					Title: "p2",
				},
			},
		},
	}).Return(nil)
	g.Expect(gcsclient.DeleteBucketPolicy(context.Background(), "test-1", "p1")).Should(Succeed())
	//get error
	iamhandleclient.EXPECT().Policy(gomock.Any()).Return(nil, errors.New("error"))
	g.Expect(gcsclient.DeleteBucketPolicy(context.Background(), "test-1", "p1")).Should(HaveOccurred())
	// not found error
	iamhandleclient.EXPECT().Policy(gomock.Any()).Return(&iam.Policy3{
		Bindings: []*pb.Binding{
			{
				Condition: &expr.Expr{
					Title: "p1",
				},
			},
			{
				Condition: &expr.Expr{
					Title: "p2",
				},
			},
		},
	}, nil)
	g.Expect(gcsclient.DeleteBucketPolicy(context.Background(), "test-1", "p3")).Should(Equal(errors.New(ErrNotFound)))
	// set error
	iamhandleclient.EXPECT().Policy(gomock.Any()).Return(&iam.Policy3{
		Bindings: []*pb.Binding{
			{
				Condition: &expr.Expr{
					Title: "p1",
				},
			},
			{
				Condition: &expr.Expr{
					Title: "p2",
				},
			},
		},
	}, nil)
	iamhandleclient.EXPECT().SetPolicy(gomock.Any(), &iam.Policy3{
		Bindings: []*pb.Binding{
			{
				Condition: &expr.Expr{
					Title: "p2",
				},
			},
		},
	}).Return(errors.New("errors"))
	g.Expect(gcsclient.DeleteBucketPolicy(context.Background(), "test-1", "p1")).Should(HaveOccurred())
}

// TestAddLifecycle tests the AddLifecycle method.
func TestAddLifecycle(t *testing.T) {
	g := NewGomegaWithT(t)
	ctl := gomock.NewController(t)
	defer ctl.Finish()

	mockclient := mocks.NewMockCloudStorageClient(ctl)
	handleclient := mocks.NewMockBucketHandle(ctl)
	gcsclient := newMockGCS(mockclient)

	mockclient.EXPECT().Bucket(gomock.Any()).Return(handleclient).AnyTimes()

	t.Run("add new rules when no existing lifecycle rules", func(t *testing.T) {
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle).ToNot(BeNil())
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(2))
				g.Expect(attrs.Lifecycle.Rules[0].Action.Type).To(Equal(storage.DeleteAction))
				g.Expect(attrs.Lifecycle.Rules[0].Condition.AgeInDays).To(Equal(int64(30)))
				g.Expect(attrs.Lifecycle.Rules[1].Action.Type).To(Equal(storage.SetStorageClassAction))
				g.Expect(attrs.Lifecycle.Rules[1].Action.StorageClass).To(Equal("COLDLINE"))
				return &storage.BucketAttrs{}, nil
			},
		)

		rules := []message.LifecycleRuleParam{
			{Prefix: "test/", ExpirationDays: 30, ColdDays: 90},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

	t.Run("add new rules with existing policy and no conflicts", func(t *testing.T) {
		existingRules := []storage.LifecycleRule{
			{
				Action:    storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{MatchesPrefix: []string{"existing/"}},
			},
		}
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: existingRules},
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(3))
				return &storage.BucketAttrs{}, nil
			},
		)

		rules := []message.LifecycleRuleParam{
			{Prefix: "new/", ExpirationDays: 10, ColdDays: 20},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

	t.Run("replace existing rules with the same prefix and action type", func(t *testing.T) {
		existingRules := []storage.LifecycleRule{
			{
				Action:    storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{MatchesPrefix: []string{"data/"}, AgeInDays: 5},
			},
			{
				Action:    storage.LifecycleAction{Type: storage.SetStorageClassAction, StorageClass: "COLDLINE"},
				Condition: storage.LifecycleCondition{MatchesPrefix: []string{"data/"}, AgeInDays: 15},
			},
		}
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: existingRules},
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(2))
				g.Expect(attrs.Lifecycle.Rules[0].Action.Type).To(Equal(storage.DeleteAction))
				g.Expect(attrs.Lifecycle.Rules[0].Condition.AgeInDays).To(Equal(int64(10)))
				g.Expect(attrs.Lifecycle.Rules[1].Action.Type).To(Equal(storage.SetStorageClassAction))
				g.Expect(attrs.Lifecycle.Rules[1].Condition.AgeInDays).To(Equal(int64(20)))
				return &storage.BucketAttrs{}, nil
			},
		)

		rules := []message.LifecycleRuleParam{
			{Prefix: "data/", ExpirationDays: 10, ColdDays: 20},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

	t.Run("filter out rules with unused prefixes", func(t *testing.T) {
		existingRules := []storage.LifecycleRule{
			{
				Action:    storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{MatchesPrefix: []string{"used/"}},
			},
			{
				Action:    storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{MatchesPrefix: []string{"unused/"}},
			},
		}
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: existingRules},
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(2))
				g.Expect(attrs.Lifecycle.Rules[0].Condition.MatchesPrefix).To(ConsistOf("new/"))
				g.Expect(attrs.Lifecycle.Rules[1].Condition.MatchesPrefix).To(ConsistOf("used/"))
				return &storage.BucketAttrs{}, nil
			},
		)

		rules := []message.LifecycleRuleParam{
			{Prefix: "new/", ExpirationDays: 10},
		}
		unusedRulePrefixes := []string{"unused/"}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, unusedRulePrefixes)
		g.Expect(err).Should(Succeed())
	})

	t.Run("get existing rules error", func(t *testing.T) {
		handleclient.EXPECT().Attrs(gomock.Any()).Return(nil, errors.New("get attrs error"))
		rules := []message.LifecycleRuleParam{
			{Prefix: "test/", ExpirationDays: 30},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(HaveOccurred())
		g.Expect(err.Error()).To(ContainSubstring("failed to get existing lifecycle rules for bucket test-bucket: get attrs error"))
	})

	t.Run("update lifecycle rules error", func(t *testing.T) {
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).Return(nil, errors.New("update error"))
		rules := []message.LifecycleRuleParam{
			{Prefix: "test/", ExpirationDays: 30},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(HaveOccurred())
		g.Expect(err.Error()).To(ContainSubstring("failed to set lifecycle rules for bucket test-bucket: update error"))
	})

	t.Run("existing lifecycle is nil", func(t *testing.T) {
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{}, // Lifecycle is a struct, so it's never truly nil, but its Rules can be.
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle).ToNot(BeNil())
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(1))
				return &storage.BucketAttrs{}, nil
			},
		)
		rules := []message.LifecycleRuleParam{
			{Prefix: "test/", ExpirationDays: 30},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

	t.Run("existing lifecycle rules is empty", func(t *testing.T) {
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: []storage.LifecycleRule{}},
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(1))
				return &storage.BucketAttrs{}, nil
			},
		)
		rules := []message.LifecycleRuleParam{
			{Prefix: "test/", ExpirationDays: 30},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

	t.Run("skip update when lifecycle rules are unchanged", func(t *testing.T) {
		existingRules := []storage.LifecycleRule{
			{
				Action: storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{
					AgeInDays:     30,
					MatchesPrefix: []string{"test/"},
				},
			},
		}
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: existingRules},
		}, nil)

		rules := []message.LifecycleRuleParam{
			{Prefix: "test/", ExpirationDays: 30},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

	t.Run("drop stale managed cluster lifecycle rules not in new config", func(t *testing.T) {
		existingRules := []storage.LifecycleRule{
			{
				Action: storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{
					AgeInDays:     32,
					MatchesPrefix: []string{"0/tenant/project/oldapp/profiles/"},
				},
			},
			{
				Action: storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{
					AgeInDays:     32,
					MatchesPrefix: []string{"0/tenant/project/oldapp/oom_record/"},
				},
			},
			{
				Action: storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{
					AgeInDays:     2,
					MatchesPrefix: []string{"diagnosis/tasks/processor-dedicated/"},
				},
			},
		}
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: existingRules},
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(2))
				g.Expect(attrs.Lifecycle.Rules[0].Condition.MatchesPrefix).To(ConsistOf("0/tenant/project/newapp/profiles/"))
				g.Expect(attrs.Lifecycle.Rules[1].Condition.MatchesPrefix).To(ConsistOf("diagnosis/tasks/processor-dedicated/"))
				return &storage.BucketAttrs{}, nil
			},
		)

		rules := []message.LifecycleRuleParam{
			{Prefix: "0/tenant/project/newapp/profiles/", ExpirationDays: 32},
			{Prefix: "diagnosis/tasks/processor-dedicated/", ExpirationDays: 2},
		}
		unusedRulePrefixes := []string{"profile_deletion_", "oom_record_deletion_", "keyviz_record_deletion_"}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, unusedRulePrefixes)
		g.Expect(err).Should(Succeed())
	})

	t.Run("truncate rules when exceeding gcs max lifecycle rules", func(t *testing.T) {
		existingRules := make([]storage.LifecycleRule, 0, gcsMaxLifecycleRules)
		for i := 0; i < gcsMaxLifecycleRules-1; i++ {
			existingRules = append(existingRules, storage.LifecycleRule{
				Action: storage.LifecycleAction{Type: storage.DeleteAction},
				Condition: storage.LifecycleCondition{
					AgeInDays:     10,
					MatchesPrefix: []string{fmt.Sprintf("legacy/data-%d/", i)},
				},
			})
		}
		handleclient.EXPECT().Attrs(gomock.Any()).Return(&storage.BucketAttrs{
			Lifecycle: storage.Lifecycle{Rules: existingRules},
		}, nil)
		handleclient.EXPECT().Update(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, attrs storage.BucketAttrsToUpdate) (*storage.BucketAttrs, error) {
				g.Expect(attrs.Lifecycle.Rules).To(HaveLen(gcsMaxLifecycleRules))
				return &storage.BucketAttrs{}, nil
			},
		)

		rules := []message.LifecycleRuleParam{
			{Prefix: "0/tenant/project/newapp/profiles/", ExpirationDays: 32},
			{Prefix: "diagnosis/tasks/processor-dedicated/", ExpirationDays: 2},
		}
		err := gcsclient.AddLifecycle(context.Background(), "test-bucket", rules, []string{})
		g.Expect(err).Should(Succeed())
	})

}