From 42eafd1fffe396fe5f527f952d4e3d61e4446e14 Mon Sep 17 00:00:00 2001
From: Roman Loginov <r.loginov@yadro.com>
Date: Thu, 25 Jan 2024 09:09:25 +0300
Subject: [PATCH] [#5] Change the filter operation when search objects

The delete function should work recursively
and delete all objects by path and by subpaths.

Signed-off-by: Roman Loginov <r.loginov@yadro.com>
---
 cmd/registry/config-dev-frostfs.yml        |  2 +-
 registry/storage/driver/frostfs/frostfs.go | 28 +++++++++++++++++++---
 2 files changed, 26 insertions(+), 4 deletions(-)

diff --git a/cmd/registry/config-dev-frostfs.yml b/cmd/registry/config-dev-frostfs.yml
index 4c9b7ce2b..f0463860e 100644
--- a/cmd/registry/config-dev-frostfs.yml
+++ b/cmd/registry/config-dev-frostfs.yml
@@ -55,7 +55,7 @@ storage:
     connection_timeout: 5s
     request_timeout: 5s
     rebalance_interval: 30s
-    rpc_endpoint: http://morph_chain.frostfs.devenv:30333
+    rpc_endpoint: http://morph-chain.frostfs.devenv:30333
 http:
   addr: :5000
   debug:
diff --git a/registry/storage/driver/frostfs/frostfs.go b/registry/storage/driver/frostfs/frostfs.go
index 473c527d9..7f2856ee4 100644
--- a/registry/storage/driver/frostfs/frostfs.go
+++ b/registry/storage/driver/frostfs/frostfs.go
@@ -699,7 +699,7 @@ func (d *driver) Move(ctx context.Context, sourcePath string, destPath string) e
 func (d *driver) Delete(ctx context.Context, path string) error {
 	filters := object.NewSearchFilters()
 	filters.AddRootFilter()
-	filters.AddFilter(attributeFilePath, path, object.MatchStringEqual)
+	filters.AddFilter(attributeFilePath, path, object.MatchCommonPrefix)
 
 	var prmSearch pool.PrmObjectSearch
 	prmSearch.SetContainerID(d.containerID)
@@ -714,10 +714,19 @@ func (d *driver) Delete(ctx context.Context, path string) error {
 
 	var inErr error
 	err = res.Iterate(func(id oid.ID) bool {
-		if err = d.delete(ctx, id); err != nil {
-			inErr = fmt.Errorf("couldn't delete object by path '%s': %w", path, err)
+		// Check if a key is a subpath (so that deleting "/a" does not delete "/ab").
+		isSubPath, err := d.checkIsSubPath(ctx, path, id)
+		if err != nil {
+			inErr = err
 			return true
 		}
+
+		if isSubPath {
+			if err = d.delete(ctx, id); err != nil {
+				inErr = fmt.Errorf("couldn't delete object by path '%s': %w", path, err)
+				return true
+			}
+		}
 		return false
 	})
 	if err == nil {
@@ -730,6 +739,19 @@ func (d *driver) Delete(ctx context.Context, path string) error {
 	return nil
 }
 
+func (d *driver) checkIsSubPath(ctx context.Context, path string, id oid.ID) (bool, error) {
+	var prmHead pool.PrmObjectHead
+	prmHead.SetAddress(d.objectAddress(id))
+
+	obj, err := d.sdkPool.HeadObject(ctx, prmHead)
+	if err != nil {
+		return false, fmt.Errorf("couldn't head object part '%s', id '%s': %w", path, id, err)
+	}
+
+	fileInf := newFileInfo(ctx, obj, "")
+	return fileInf.Path() <= path || fileInf.Path()[len(path)] == '/', nil
+}
+
 func (d *driver) delete(ctx context.Context, id oid.ID) error {
 	var prm pool.PrmObjectDelete
 	prm.SetAddress(d.objectAddress(id))