Implement list interface for filedb

pkg/datastore/filedb/BUILD.bazel

@@ -5,10 +5,13 @@ go_library(
     srcs = [
         "codec.go",
         "filedb.go",
+        "filter.go",
+        "iterator.go",
     ],
     importpath = "github.com/pipe-cd/pipecd/pkg/datastore/filedb",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/cache:go_default_library",
         "//pkg/datastore:go_default_library",
         "//pkg/filestore:go_default_library",
         "@org_uber_go_zap//:go_default_library",

pkg/datastore/filedb/filedb.go

@@ -17,15 +17,18 @@ package filedb
 import (
 	"context"
 	"fmt"
+	"path/filepath"
 
 	"go.uber.org/zap"
 
+	"github.com/pipe-cd/pipecd/pkg/cache"
 	"github.com/pipe-cd/pipecd/pkg/datastore"
 	"github.com/pipe-cd/pipecd/pkg/filestore"
 )
 
 type FileDB struct {
 	backend filestore.Store
+	cache   cache.Cache
 	logger  *zap.Logger
 }
 
@@ -37,9 +40,10 @@ func WithLogger(logger *zap.Logger) Option {
 	}
 }
 
-func NewFileDB(fs filestore.Store, opts ...Option) (*FileDB, error) {
+func NewFileDB(fs filestore.Store, c cache.Cache, opts ...Option) (*FileDB, error) {
 	fd := &FileDB{
 		backend: fs,
+		cache:   c,
 		logger:  zap.NewNop(),
 	}
 	for _, opt := range opts {
@@ -61,11 +65,100 @@ func (f *FileDB) fetch(ctx context.Context, path string) ([]byte, error) {
 }
 
 func (f *FileDB) Find(ctx context.Context, col datastore.Collection, opts datastore.ListOptions) (datastore.Iterator, error) {
-	_, ok := col.(datastore.ShardStorable)
+	scol, ok := col.(datastore.ShardStorable)
 	if !ok {
 		return nil, datastore.ErrUnsupported
 	}
-	return nil, datastore.ErrUnimplemented
+
+	var (
+		kind   = col.Kind()
+		shards = scol.ListInUsedShards()
+		// Map of objects values with the first key is the object id.
+		objects map[string][][]byte
+	)
+
+	// Fetch the first part of all objects under a specified "directory".
+	for _, shard := range shards {
+		dpath := makeHotStorageDirPath(kind, shard)
+		parts, err := f.backend.List(ctx, dpath)
+		if err != nil {
+			f.logger.Error("failed to find entities",
+				zap.String("kind", kind),
+				zap.Error(err),
+			)
+			return nil, err
+		}
+
+		if objects == nil {
+			objects = make(map[string][][]byte, len(parts))
+		}
+		for _, obj := range parts {
+			id := filepath.Base(obj.Path)
+
+			// Check if the raw data stored under this path is fetched or not
+			// based on its etag value stored as key in `f.cache` store.
+			cdata, err := f.cache.Get(obj.Etag)
+			if err == nil {
+				objects[id] = append(objects[id], cdata.([]byte))
+				continue
+			}
+
+			// If there is no value attached with the given etag key, fetch the
+			// content under the object path.
+			data, err := f.fetch(ctx, obj.Path)
+			if err != nil {
+				f.logger.Error("failed to fetch entity part",
+					zap.String("kind", kind),
+					zap.String("id", id),
+					zap.Error(err),
+				)
+				return nil, err
+			}
+
+			// Store fetched data to cache.
+			if err = f.cache.Put(obj.Etag, data); err != nil {
+				f.logger.Error("failed to store entity part to cache",
+					zap.String("kind", kind),
+					zap.String("id", id),
+					zap.String("key", obj.Etag),
+					zap.Error(err),
+				)
+			}
+
+			objects[id] = append(objects[id], data)
+		}
+	}
+
+	entities := make([]interface{}, 0, len(objects))
+	for id, obj := range objects {
+		e := col.Factory()()
+		if err := decode(col, e, obj...); err != nil {
+			f.logger.Error("failed to build entity",
+				zap.String("kind", kind),
+				zap.String("id", id),
+				zap.Error(err),
+			)
+			return nil, err
+		}
+
+		pass, err := filter(col, e, opts.Filters)
+		if err != nil {
+			f.logger.Error("failed to filter entity",
+				zap.String("kind", kind),
+				zap.String("id", id),
+				zap.Error(err),
+			)
+			return nil, err
+		}
+
+		if pass {
+			entities = append(entities, e)
+		}
+	}
+
+	return &Iterator{
+		data: entities,
+	}, nil
 }
 
 func (f *FileDB) Get(ctx context.Context, col datastore.Collection, id string, v interface{}) error {
@@ -141,3 +234,7 @@ func makeHotStorageFilePath(kind, id string, shard datastore.Shard) string {
 	// TODO: Find a way to separate files by project to avoid fetch resources cross project.
 	return fmt.Sprintf("%s/%s/%s.json", kind, shard, id)
 }
+
+func makeHotStorageDirPath(kind string, shard datastore.Shard) string {
+	return fmt.Sprintf("%s/%s/", kind, shard)
+}

pkg/datastore/filedb/filter.go

@@ -0,0 +1,33 @@
+// Copyright 2022 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filedb
+
+import (
+	"github.com/pipe-cd/pipecd/pkg/datastore"
+)
+
+// TODO: Implement filterable interface for each collection.
+type filterable interface {
+	Match(e interface{}, filters []datastore.ListFilter) (bool, error)
+}
+
+func filter(col datastore.Collection, e interface{}, filters []datastore.ListFilter) (bool, error) {
+	fcol, ok := col.(filterable)
+	if !ok {
+		return false, datastore.ErrUnsupported
+	}
+
+	return fcol.Match(e, filters)
+}

pkg/datastore/filedb/iterator.go

@@ -0,0 +1,29 @@
+// Copyright 2022 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filedb
+
+import "github.com/pipe-cd/pipecd/pkg/datastore"
+
+type Iterator struct {
+	data []interface{}
+}
+
+func (it *Iterator) Next(dst interface{}) error {
+	return datastore.ErrUnimplemented
+}
+
+func (it *Iterator) Cursor() (string, error) {
+	return "", datastore.ErrUnimplemented
+}