package resource

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	badger "github.com/dgraph-io/badger/v4"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"

	authlib "github.com/grafana/authlib/types"
	"github.com/grafana/dskit/services"

	"github.com/grafana/grafana/pkg/apimachinery/identity"
	"github.com/grafana/grafana/pkg/apimachinery/utils"
	"github.com/grafana/grafana/pkg/infra/log"
	"github.com/grafana/grafana/pkg/infra/tracing"
	"github.com/grafana/grafana/pkg/storage/unified/resourcepb"
	"github.com/grafana/grafana/pkg/util/scheduler"
)

func TestSimpleServer(t *testing.T) {
	testUserA := &identity.StaticRequester{
		Type:           authlib.TypeUser,
		Login:          "testuser",
		UserID:         123,
		UserUID:        "u123",
		OrgRole:        identity.RoleAdmin,
		IsGrafanaAdmin: true, // can do anything
	}
	ctx := authlib.WithAuthInfo(context.Background(), testUserA)

	// Create in-memory BadgerDB for testing
	db, err := badger.Open(badger.DefaultOptions("").
		WithInMemory(true).
		WithLogger(nil))
	require.NoError(t, err)
	defer func() {
		err := db.Close()
		require.NoError(t, err)
	}()

	kv := NewBadgerKV(db)
	store, err := NewKVStorageBackend(KVBackendOptions{
		KvStore: kv,
	})
	require.NoError(t, err)

	server, err := NewResourceServer(ResourceServerOptions{
		Backend: store,
	})
	require.NoError(t, err)

	t.Run("playlist happy CRUD paths", func(t *testing.T) {
		raw := []byte(`{
    		"apiVersion": "playlist.grafana.app/v0alpha1",
			"kind": "Playlist",
			"metadata": {
				"name": "fdgsv37qslr0ga",
				"uid": "xyz",
				"namespace": "default",
				"annotations": {
					"grafana.app/repoName": "elsewhere",
					"grafana.app/repoPath": "path/to/item",
					"grafana.app/repoTimestamp": "2024-02-02T00:00:00Z"
				}
			},
			"spec": {
				"title": "hello",
				"interval": "5m",
				"items": [
					{
						"type": "dashboard_by_uid",
						"value": "vmie2cmWz"
					}
				]
			}
		}`)

		key := &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "fdgsv37qslr0ga",
		}

		// Should be empty when we start
		all, err := server.List(ctx, &resourcepb.ListRequest{Options: &resourcepb.ListOptions{
			Key: &resourcepb.ResourceKey{
				Group:    key.Group,
				Resource: key.Resource,
			},
		}})
		require.NoError(t, err)
		require.Len(t, all.Items, 0)

		// should return 404 if not found
		found, err := server.Read(ctx, &resourcepb.ReadRequest{Key: key})
		require.NoError(t, err)
		require.NotNil(t, found.Error)
		require.Equal(t, int32(http.StatusNotFound), found.Error.Code)

		created, err := server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})
		require.NoError(t, err)
		require.Nil(t, created.Error)
		require.True(t, created.ResourceVersion > 0)

		// The key does not include resource version
		found, err = server.Read(ctx, &resourcepb.ReadRequest{Key: key})
		require.NoError(t, err)
		require.Nil(t, found.Error)
		require.Equal(t, created.ResourceVersion, found.ResourceVersion)

		// Now update the value
		tmp := &unstructured.Unstructured{}
		err = json.Unmarshal(found.Value, tmp)
		require.NoError(t, err)

		now := time.Now().UnixMilli()
		obj, err := utils.MetaAccessor(tmp)
		require.NoError(t, err)
		obj.SetAnnotation("test", "hello")
		obj.SetUpdatedTimestampMillis(now)
		obj.SetUpdatedBy(testUserA.GetUID())
		obj.SetLabels(map[string]string{
			utils.LabelKeyGetTrash: "", // should not be allowed to save this!
		})
		raw, err = json.Marshal(tmp)
		require.NoError(t, err)
		updated, err := server.Update(ctx, &resourcepb.UpdateRequest{
			Key:             key,
			Value:           raw,
			ResourceVersion: created.ResourceVersion})
		require.NoError(t, err)
		require.Equal(t, int32(400), updated.Error.Code) // bad request

		// remove the invalid labels
		obj.SetLabels(nil)
		raw, err = json.Marshal(tmp)
		require.NoError(t, err)
		updated, err = server.Update(ctx, &resourcepb.UpdateRequest{
			Key:             key,
			Value:           raw,
			ResourceVersion: created.ResourceVersion})
		require.NoError(t, err)
		require.Nil(t, updated.Error)
		require.True(t, updated.ResourceVersion > created.ResourceVersion)

		// We should still get the latest
		found, err = server.Read(ctx, &resourcepb.ReadRequest{Key: key})
		require.NoError(t, err)
		require.Nil(t, found.Error)
		require.Equal(t, updated.ResourceVersion, found.ResourceVersion)

		all, err = server.List(ctx, &resourcepb.ListRequest{Options: &resourcepb.ListOptions{
			Key: &resourcepb.ResourceKey{
				Group:    key.Group,
				Resource: key.Resource,
			},
		}})
		require.NoError(t, err)
		require.Len(t, all.Items, 1)
		require.Equal(t, updated.ResourceVersion, all.Items[0].ResourceVersion)

		// Try again with a direct query
		all, err = server.List(ctx, &resourcepb.ListRequest{Options: &resourcepb.ListOptions{
			Key: &resourcepb.ResourceKey{
				Namespace: key.Namespace,
				Group:     key.Group,
				Resource:  key.Resource,
			},
			Fields: []*resourcepb.Requirement{{
				Key:      "metadata.name",
				Operator: "=",
				Values:   []string{"not-matching"},
			}},
		}})
		require.NoError(t, err)
		require.Len(t, all.Items, 0)

		// This time matching
		all, err = server.List(ctx, &resourcepb.ListRequest{Options: &resourcepb.ListOptions{
			Key: &resourcepb.ResourceKey{
				Namespace: key.Namespace,
				Group:     key.Group,
				Resource:  key.Resource,
			},
			Fields: []*resourcepb.Requirement{{
				Key:      "metadata.name",
				Operator: "=",
				Values:   []string{"fdgsv37qslr0ga"},
			}},
		}})
		require.NoError(t, err)
		require.Len(t, all.Items, 1)
		require.Equal(t, raw, all.Items[0].Value)

		deleted, err := server.Delete(ctx, &resourcepb.DeleteRequest{Key: key, ResourceVersion: updated.ResourceVersion})
		require.NoError(t, err)
		require.True(t, deleted.ResourceVersion > updated.ResourceVersion)

		// We should get not found status when trying to read the latest value
		found, err = server.Read(ctx, &resourcepb.ReadRequest{Key: key})
		require.NoError(t, err)
		require.NotNil(t, found.Error)
		require.Equal(t, int32(404), found.Error.Code)

		// And the deleted value should not be in the results
		all, err = server.List(ctx, &resourcepb.ListRequest{Options: &resourcepb.ListOptions{
			Key: &resourcepb.ResourceKey{
				Group:    key.Group,
				Resource: key.Resource,
			},
		}})
		require.NoError(t, err)
		require.Len(t, all.Items, 0) // empty
	})

	t.Run("playlist FAIL CRUD paths due to invalid key", func(t *testing.T) {
		raw := []byte(`{
    		"apiVersion": "playlist.grafana.app/v0alpha1",
			"kind": "Playlist",
			"metadata": {
				"name": "fdgsv37#qslr0ga",
				"uid": "xyz",
				"namespace": "default",
				"annotations": {
					"grafana.app/repoName": "elsewhere",
					"grafana.app/repoPath": "path/to/item",
					"grafana.app/repoTimestamp": "2024-02-02T00:00:00Z"
				}
			},
			"spec": {
				"title": "hello",
				"interval": "5m",
				"items": [
					{
						"type": "dashboard_by_uid",
						"value": "vmie2cmWz"
					}
				]
			}
		}`)

		// invalid group
		key := &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app###",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "fdgsv37qslr0ga",
		}

		created, err := server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})
		require.Error(t, err)
		require.Nil(t, created)

		// invalid resource
		key = &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr###", // can be anything :(
			Namespace: "default",
			Name:      "fdgsv37qslr0ga",
		}

		created, err = server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})
		require.Error(t, err)
		require.Nil(t, created)

		// invalid namespace
		key = &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default###",
			Name:      "fdgsv37qslr0ga",
		}

		created, err = server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})
		require.Error(t, err)
		require.Nil(t, created)

		// invalid name
		key = &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "fdgsv37qslr0g###",
		}

		created, err = server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})
		require.Error(t, err)
		require.Nil(t, created)

		// legacy name - valid
		key = &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "2c7e5361-7360-4d2a-ae45-5e79bba458d6",
		}

		created, err = server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})

		require.NoError(t, err)
		require.NotNil(t, created)

		// legacy name - also valid
		key = &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "IvIsO_YGz",
		}

		created, err = server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})

		require.NoError(t, err)
		require.NotNil(t, created)

		// legacy name - also valid
		key = &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "_IvIsOYGz",
		}

		created, err = server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})

		require.NoError(t, err)
		require.NotNil(t, created)

		invalidQualifiedNames := []string{
			"",                                     // empty
			strings.Repeat("1", 260),               // too long
			"    ",                                 // only spaces
			"f8cc010c.ee72.4681;89d2+d46e1bd47d33", // invalid chars
		}

		// group
		for _, invalidGroup := range invalidQualifiedNames {
			key = &resourcepb.ResourceKey{
				Group:     invalidGroup,
				Resource:  "rrrr", // can be anything :(
				Namespace: "default",
				Name:      "_IvIsOYGz",
			}

			created, err = server.Create(ctx, &resourcepb.CreateRequest{
				Value: raw,
				Key:   key,
			})

			require.Error(t, err)
			require.Nil(t, created)
		}

		// resource
		for _, invalidResource := range invalidQualifiedNames {
			key = &resourcepb.ResourceKey{
				Group:     "playlist.grafana.app",
				Resource:  invalidResource,
				Namespace: "default",
				Name:      "_IvIsOYGz",
			}

			created, err = server.Create(ctx, &resourcepb.CreateRequest{
				Value: raw,
				Key:   key,
			})

			require.Error(t, err)
			require.Nil(t, created)
		}

		// namespace
		for _, invalidNamespace := range invalidQualifiedNames {
			if invalidNamespace == "" {
				// empty namespace is allowed
				continue
			}

			key = &resourcepb.ResourceKey{
				Group:     "playlist.grafana.app",
				Resource:  "rrrr", // can be anything :(
				Namespace: invalidNamespace,
				Name:      "_IvIsOYGz",
			}

			created, err = server.Create(ctx, &resourcepb.CreateRequest{
				Value: raw,
				Key:   key,
			})

			require.Error(t, err)
			require.Nil(t, created)
		}
	})

	t.Run("playlist update optimistic concurrency check", func(t *testing.T) {
		raw := []byte(`{
    	"apiVersion": "playlist.grafana.app/v0alpha1",
			"kind": "Playlist",
			"metadata": {
				"name": "fdgsv37qslr0ga",
				"namespace": "default",
				"uid": "xyz",
				"annotations": {
					"grafana.app/repoName": "elsewhere",
					"grafana.app/repoPath": "path/to/item",
					"grafana.app/repoTimestamp": "2024-02-02T00:00:00Z"
				}
			},
			"spec": {
				"title": "hello",
				"interval": "5m",
				"items": [
					{
						"type": "dashboard_by_uid",
						"value": "vmie2cmWz"
					}
				]
			}
		}`)

		key := &resourcepb.ResourceKey{
			Group:     "playlist.grafana.app",
			Resource:  "rrrr", // can be anything :(
			Namespace: "default",
			Name:      "fdgsv37qslr0ga",
		}

		created, err := server.Create(ctx, &resourcepb.CreateRequest{
			Value: raw,
			Key:   key,
		})
		require.NoError(t, err)

		// Update should return an ErrOptimisticLockingFailed the second time

		_, err = server.Update(ctx, &resourcepb.UpdateRequest{
			Key:             key,
			Value:           raw,
			ResourceVersion: created.ResourceVersion})
		require.NoError(t, err)

		rsp, _ := server.Update(ctx, &resourcepb.UpdateRequest{
			Key:             key,
			Value:           raw,
			ResourceVersion: created.ResourceVersion})
		require.Equal(t, rsp.Error.Code, ErrOptimisticLockingFailed.Code)
		require.Equal(t, rsp.Error.Message, ErrOptimisticLockingFailed.Message)
	})
}

func TestRunInQueue(t *testing.T) {
	const testTenantID = "test-tenant"
	t.Run("should execute successfully when queue has capacity", func(t *testing.T) {
		s, _ := newTestServerWithQueue(t, 1, 1)
		executed := make(chan bool, 1)

		runnable := func(ctx context.Context) {
			executed <- true
		}

		err := s.runInQueue(context.Background(), testTenantID, runnable)
		require.NoError(t, err)
		assert.True(t, <-executed, "runnable should have been executed")
	})

	t.Run("should time out if a task is sitting in the queue beyond the timeout", func(t *testing.T) {
		s, _ := newTestServerWithQueue(t, 1, 1)
		executed := make(chan struct{}, 1)
		runnable := func(ctx context.Context) {
			time.Sleep(1 * time.Second)
			executed <- struct{}{}
		}

		err := s.runInQueue(context.Background(), testTenantID, runnable)
		require.Error(t, err)
		assert.Equal(t, context.DeadlineExceeded, err)
		<-executed
	})

	t.Run("should return an error if queue is consistently full after retrying", func(t *testing.T) {
		s, q := newTestServerWithQueue(t, 1, 1)
		// Task 1: This will be picked up by the worker and block it.
		blocker := make(chan struct{})
		defer close(blocker)
		blockingRunnable := func() {
			<-blocker
		}
		err := q.Enqueue(context.Background(), testTenantID, blockingRunnable)
		require.NoError(t, err)
		for q.Len() > 0 {
			time.Sleep(100 * time.Millisecond)
		}
		err = q.Enqueue(context.Background(), testTenantID, blockingRunnable)
		require.NoError(t, err)

		// Task 2: This runnable should never execute because the queue is full.
		mu := sync.Mutex{}
		executed := false
		runnable := func(ctx context.Context) {
			mu.Lock()
			defer mu.Unlock()
			executed = true
		}

		err = s.runInQueue(context.Background(), testTenantID, runnable)
		require.Error(t, err)
		require.ErrorIs(t, err, scheduler.ErrTenantQueueFull)
		require.False(t, executed, "runnable should not have been executed")
	})
}

// newTestServerWithQueue creates a server with a real scheduler.Queue for testing.
// It also sets up a worker to consume items from the queue.
func newTestServerWithQueue(t *testing.T, maxSizePerTenant int, numWorkers int) (*server, *scheduler.Queue) {
	t.Helper()
	q := scheduler.NewQueue(&scheduler.QueueOptions{
		MaxSizePerTenant: maxSizePerTenant,
		Registerer:       prometheus.NewRegistry(),
		Logger:           log.NewNopLogger(),
	})
	err := services.StartAndAwaitRunning(context.Background(), q)
	require.NoError(t, err)
	t.Cleanup(func() {
		err := services.StopAndAwaitTerminated(context.Background(), q)
		require.NoError(t, err)
	})

	// Create a worker to consume from the queue
	worker, err := scheduler.NewScheduler(q, &scheduler.Config{
		Logger:     log.NewNopLogger(),
		NumWorkers: numWorkers,
	})
	require.NoError(t, err)
	err = services.StartAndAwaitRunning(context.Background(), worker)
	require.NoError(t, err)
	t.Cleanup(func() {
		err := services.StopAndAwaitTerminated(context.Background(), worker)
		require.NoError(t, err)
	})

	s := &server{
		queue: q,
		queueConfig: QueueConfig{
			Timeout:    500 * time.Millisecond,
			MaxRetries: 2,
			MinBackoff: 10 * time.Millisecond,
		},
		log: log.NewNopLogger(),
	}
	return s, q
}

func TestArtificialDelayAfterSuccessfulOperation(t *testing.T) {
	s := &server{
		artificialSuccessfulWriteDelay: 1 * time.Millisecond,
		log:                            log.NewNopLogger(),
	}

	check := func(t *testing.T, expectedSleep bool, res responseWithErrorResult, err error) {
		slept := s.sleepAfterSuccessfulWriteOperation("test", &resourcepb.ResourceKey{}, res, err)
		require.Equal(t, expectedSleep, slept)
	}

	// Successful responses should sleep
	check(t, true, nil, nil)

	check(t, true, (responseWithErrorResult)((*resourcepb.CreateResponse)(nil)), nil)
	check(t, true, &resourcepb.CreateResponse{}, nil)

	check(t, true, (responseWithErrorResult)((*resourcepb.UpdateResponse)(nil)), nil)
	check(t, true, &resourcepb.UpdateResponse{}, nil)

	check(t, true, (responseWithErrorResult)((*resourcepb.DeleteResponse)(nil)), nil)
	check(t, true, &resourcepb.DeleteResponse{}, nil)

	// Failed responses should return without sleeping
	check(t, false, nil, errors.New("some error"))
	check(t, false, &resourcepb.CreateResponse{Error: AsErrorResult(errors.New("some error"))}, nil)
	check(t, false, &resourcepb.UpdateResponse{Error: AsErrorResult(errors.New("some error"))}, nil)
	check(t, false, &resourcepb.DeleteResponse{Error: AsErrorResult(errors.New("some error"))}, nil)
}

func TestGetQuotaUsage(t *testing.T) {
	ctx := context.Background()

	t.Run("returns error when overrides service is not configured", func(t *testing.T) {
		s := &server{
			overridesService: nil,
			log:              log.NewNopLogger(),
		}

		resp, err := s.GetQuotaUsage(ctx, &resourcepb.QuotaUsageRequest{
			Key: &resourcepb.ResourceKey{
				Namespace: "stacks-123",
				Group:     "dashboard.grafana.app",
				Resource:  "dashboards",
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp.Error)
		assert.Equal(t, int32(http.StatusNotImplemented), resp.Error.Code)
		assert.Equal(t, "overrides service not configured on resource server", resp.Error.Message)
	})

	t.Run("returns usage and limit successfully", func(t *testing.T) {
		// Create a temporary overrides config file
		tmpFile := filepath.Join(t.TempDir(), "overrides.yaml")
		content := `overrides:
  "123":
    quotas:
      dashboard.grafana.app/dashboards:
        limit: 500
`
		require.NoError(t, os.WriteFile(tmpFile, []byte(content), 0644))

		// Create a real OverridesService with the temp file
		overridesService, err := NewOverridesService(ctx, log.NewNopLogger(), prometheus.NewRegistry(), tracing.NewNoopTracerService(), ReloadOptions{
			FilePath: tmpFile,
		})
		require.NoError(t, err)
		require.NoError(t, overridesService.init(ctx))
		defer func() {
			_ = overridesService.stop(ctx)
		}()

		// Create a mock backend that returns resource stats (reusing mockStorageBackend from search_test.go)
		mockBackend := &mockStorageBackend{
			resourceStats: []ResourceStats{{Count: 42}},
		}

		s := &server{
			backend:          mockBackend,
			overridesService: overridesService,
			log:              log.NewNopLogger(),
		}

		resp, err := s.GetQuotaUsage(ctx, &resourcepb.QuotaUsageRequest{
			Key: &resourcepb.ResourceKey{
				Namespace: "stacks-123",
				Group:     "dashboard.grafana.app",
				Resource:  "dashboards",
			},
		})
		require.NoError(t, err)
		require.Nil(t, resp.Error)
		assert.Equal(t, int64(42), resp.Usage)
		assert.Equal(t, int64(500), resp.Limit)
	})
}