unistore: add LastResourceVersion to the datastore (#107303)

* Add datastore

* too many slashes

* lint

* add metadata store

* simplify meta

* Add eventstore

* golint

* lint

* Add datastore

* too many slashes

* lint

* pr comments

* extract ParseKey

* readcloser

* remove get prefix

* use dedicated keys

* parsekey

* sameresource

* unrelated

* name

* renmae tests

* add key validation

* fix tests

* refactor a bit

* lint

* allow empty ns

* get keys instead of list

* rename the functions

* refactor yield candidate

* update test

* unistore: add LastResourceVersion to datastore

* lint

* use map string

* missing err check

* fix

pkg/storage/unified/resource/datastore.go

@@ -49,6 +49,10 @@ func (k DataKey) String() string {
 	return fmt.Sprintf("%s/%s/%s/%s/%d~%s", k.Namespace, k.Group, k.Resource, k.Name, k.ResourceVersion, k.Action)
 }
 
+func (k DataKey) Equals(other DataKey) bool {
+	return k.Namespace == other.Namespace && k.Group == other.Group && k.Resource == other.Resource && k.Name == other.Name && k.ResourceVersion == other.ResourceVersion && k.Action == other.Action
+}
+
 func (k DataKey) Validate() error {
 	if k.Namespace == "" {
 		if k.Group != "" || k.Resource != "" || k.Name != "" {
@@ -207,6 +211,29 @@ func (d *dataStore) Keys(ctx context.Context, key ListRequestKey) iter.Seq2[Data
 	}
 }
 
+// LastResourceVersion returns the last key for a given resource
+func (d *dataStore) LastResourceVersion(ctx context.Context, key ListRequestKey) (DataKey, error) {
+	if err := key.Validate(); err != nil {
+		return DataKey{}, fmt.Errorf("invalid data key: %w", err)
+	}
+	if key.Group == "" || key.Resource == "" || key.Namespace == "" || key.Name == "" {
+		return DataKey{}, fmt.Errorf("group, resource, namespace or name is empty")
+	}
+	prefix := key.Prefix()
+	for key, err := range d.kv.Keys(ctx, dataSection, ListOptions{
+		StartKey: prefix,
+		EndKey:   PrefixRangeEnd(prefix),
+		Limit:    1,
+		Sort:     SortOrderDesc,
+	}) {
+		if err != nil {
+			return DataKey{}, err
+		}
+		return ParseKey(key)
+	}
+	return DataKey{}, ErrNotFound
+}
+
 func (d *dataStore) Get(ctx context.Context, key DataKey) (io.ReadCloser, error) {
 	if err := key.Validate(); err != nil {
 		return nil, fmt.Errorf("invalid data key: %w", err)

pkg/storage/unified/resource/datastore_test.go

@@ -3,6 +3,7 @@ package resource
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"testing"
 
@@ -1353,3 +1354,105 @@ func TestListRequestKey_Prefix(t *testing.T) {
 		})
 	}
 }
+
+func TestDataStore_LastResourceVersion(t *testing.T) {
+	ds := setupTestDataStore(t)
+	ctx := context.Background()
+
+	t.Run("returns last resource version for existing data", func(t *testing.T) {
+		resourceKey := ListRequestKey{
+			Namespace: "test-namespace",
+			Group:     "test-group",
+			Resource:  "test-resource",
+			Name:      "test-name",
+		}
+
+		// Create test data with multiple versions
+		rv1 := node.Generate()
+		rv2 := node.Generate()
+		rv3 := node.Generate()
+
+		versions := []int64{
+			rv1.Int64(),
+			rv2.Int64(),
+			rv3.Int64(),
+		}
+
+		// Save all versions
+		for _, version := range versions {
+			dataKey := DataKey{
+				Namespace:       resourceKey.Namespace,
+				Group:           resourceKey.Group,
+				Resource:        resourceKey.Resource,
+				Name:            resourceKey.Name,
+				ResourceVersion: version,
+				Action:          DataActionCreated,
+			}
+
+			err := ds.Save(ctx, dataKey, bytes.NewReader([]byte(fmt.Sprintf("version-%d", version))))
+			require.NoError(t, err)
+		}
+
+		// Get the last resource version
+		lastKey, err := ds.LastResourceVersion(ctx, resourceKey)
+		require.NoError(t, err)
+
+		// Verify the result
+		require.Equal(t, resourceKey.Namespace, lastKey.Namespace)
+		require.Equal(t, resourceKey.Group, lastKey.Group)
+		require.Equal(t, resourceKey.Resource, lastKey.Resource)
+		require.Equal(t, resourceKey.Name, lastKey.Name)
+		require.Equal(t, DataActionCreated, lastKey.Action)
+
+		require.Equal(t, rv3.Int64(), lastKey.ResourceVersion)
+	})
+
+	t.Run("returns error for non-existent resource", func(t *testing.T) {
+		nonExistentKey := ListRequestKey{
+			Namespace: "non-existent-namespace",
+			Group:     "non-existent-group",
+			Resource:  "non-existent-resource",
+			Name:      "non-existent-name",
+		}
+
+		_, err := ds.LastResourceVersion(ctx, nonExistentKey)
+		require.Error(t, err)
+		require.Equal(t, ErrNotFound, err)
+	})
+
+	t.Run("returns error for empty required fields", func(t *testing.T) {
+		testCases := map[string]ListRequestKey{
+			"empty namespace": {
+				Namespace: "",
+				Group:     "test-group",
+				Resource:  "test-resource",
+				Name:      "test-name",
+			},
+			"empty group": {
+				Namespace: "test-namespace",
+				Group:     "",
+				Resource:  "test-resource",
+				Name:      "test-name",
+			},
+			"empty resource": {
+				Namespace: "test-namespace",
+				Group:     "test-group",
+				Resource:  "",
+				Name:      "test-name",
+			},
+			"empty name": {
+				Namespace: "test-namespace",
+				Group:     "test-group",
+				Resource:  "test-resource",
+				Name:      "",
+			},
+		}
+
+		for name, key := range testCases {
+			t.Run(name, func(t *testing.T) {
+				_, err := ds.LastResourceVersion(ctx, key)
+				require.Error(t, err)
+			})
+		}
+	})
+}