Adds backup path list method to FilesystemSink

This method needs to cover two use-cases:

1. When creating an incremental backup, find the latest bundle and refs.
2. When restoring an incremental backup, find the latest or specified
   series of incremental backups starting with a full backup.

The filenames will need to be devised in a way such that the prefix
filtering provided here is efficient and will not cause too many paths
to be returned.

2 files changed, 68 insertions, 0 deletions

diff --git a/internal/backup/filesystem_sink.go b/internal/backup/filesystem_sink.go
index 27749c47f..a987cc47a 100644
--- a/internal/backup/filesystem_sink.go
+++ b/internal/backup/filesystem_sink.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 )
 
 // FilesystemSink is a sink for creating and restoring backups from the local filesystem.
@@ -66,3 +67,40 @@ func (fs *FilesystemSink) GetReader(ctx context.Context, relativePath string) (i
 	}
 	return f, nil
 }
+
+// List returns the relative path for each data where the relative path matches
+// the given prefix
+func (fs *FilesystemSink) List(ctx context.Context, prefix string) ([]string, error) {
+	var relativePaths []string
+	prefixDir := filepath.Dir(prefix)
+	err := filepath.Walk(filepath.Join(fs.path, prefixDir), func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+
+		relativePath, err := filepath.Rel(fs.path, path)
+		if err != nil {
+			return err
+		}
+
+		if info.IsDir() {
+			// skip directories when they have no possibility of matching the prefix
+			if len(relativePath) > len(prefix) && !strings.HasPrefix(relativePath, prefix) {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+
+		if !strings.HasPrefix(relativePath, prefix) {
+			return nil
+		}
+
+		relativePaths = append(relativePaths, relativePath)
+
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("list: %w", err)
+	}
+	return relativePaths, nil
+}
diff --git a/internal/backup/filesystem_sink_test.go b/internal/backup/filesystem_sink_test.go
index c7a295da1..ed467abd1 100644
--- a/internal/backup/filesystem_sink_test.go
+++ b/internal/backup/filesystem_sink_test.go
@@ -1,6 +1,7 @@
 package backup
 
 import (
+	"bytes"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -109,3 +110,32 @@ func TestFilesystemSink_Write(t *testing.T) {
 		require.EqualError(t, err, fmt.Sprintf(`create directory structure %[1]q: mkdir %[1]s: not a directory`, filepath.Join(dir, "nested")))
 	})
 }
+
+func TestFilesystemSink_List(t *testing.T) {
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	dir := testhelper.TempDir(t)
+	fsSink := NewFilesystemSink(dir)
+
+	data := []byte("test")
+
+	for _, relativePath := range []string{
+		"a/a_pineapple",
+		"b/a_apple",
+		"b/a_carrot",
+		"b/a_cucumber",
+	} {
+		require.NoError(t, fsSink.Write(ctx, relativePath, bytes.NewReader(data)))
+	}
+
+	expectedPaths := []string{
+		"b/a_carrot",
+		"b/a_cucumber",
+	}
+
+	paths, err := fsSink.List(ctx, "b/a_c")
+	require.NoError(t, err)
+
+	require.ElementsMatch(t, expectedPaths, paths)
+}