Implement list interface for filedb

pkg/datastore/filedb/BUILD.bazel

@@ -5,6 +5,8 @@ go_library(
     srcs = [
         "codec.go",
         "filedb.go",
+        "filter.go",
+        "iterator.go",
     ],
     importpath = "github.com/pipe-cd/pipecd/pkg/datastore/filedb",
     visibility = ["//visibility:public"],

pkg/datastore/filedb/filedb.go

@@ -17,6 +17,7 @@ package filedb
 import (
 	"context"
 	"fmt"
+	"path/filepath"
 
 	"go.uber.org/zap"
 
@@ -61,11 +62,80 @@ 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)
+
+			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
+			}
+
+			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 +211,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
+}