fix: soft delete by ID succeeded

Author: Muhammad Luthfi Fahlevi

core/asset/asset.go

@@ -49,19 +49,19 @@ type Asset struct {
 	RefreshedAt *time.Time             `json:"refreshed_at" diff:"-"`
 	Version     string                 `json:"version" diff:"-"`
 	UpdatedBy   user.User              `json:"updated_by" diff:"-"`
-	IsDeleted   bool                   `json:"is_deleted"`
+	IsDeleted   bool                   `json:"is_deleted" diff:"is_deleted"`
 	Changelog   diff.Changelog         `json:"changelog,omitempty" diff:"-"`
 	Probes      []Probe                `json:"probes,omitempty"`
 }
 
 type SoftDeleteAsset struct {
-	URN         string         `json:"urn"`
-	UpdatedAt   time.Time      `json:"updated_at"`
-	RefreshedAt time.Time      `json:"refreshed_at"`
-	Version     string         `json:"version"`
-	UpdatedBy   string         `json:"updated_by"`
-	IsDeleted   bool           `json:"is_deleted"`
-	Changelog   diff.Changelog `json:"changelog"`
+	URN         string         `json:"urn" diff:"-"`
+	UpdatedAt   time.Time      `json:"updated_at" diff:"-"`
+	RefreshedAt time.Time      `json:"refreshed_at" diff:"-"`
+	Version     string         `json:"version" diff:"-"`
+	UpdatedBy   string         `json:"updated_by" diff:"-"`
+	IsDeleted   bool           `json:"is_deleted" diff:"is_deleted"`
+	Changelog   diff.Changelog `json:"changelog,omitempty" diff:"-"`
 }
 
 func NewSoftDeleteAsset(updatedAt, refreshedAt time.Time, updatedBy string) SoftDeleteAsset {

core/asset/service.go

@@ -185,32 +185,28 @@ func (s *Service) DeleteAsset(ctx context.Context, id string) (err error) {
 }
 
 // SoftDeleteAsset is soft-deletion that can accept ID or URN of asset
-func (s *Service) SoftDeleteAsset(ctx context.Context, id, updatedBy string) (err error) {
+func (s *Service) SoftDeleteAsset(ctx context.Context, id, updatedBy string) (urn string, err error) {
 	defer func() {
 		s.instrumentAssetOp(ctx, "SoftDeleteAsset", id, err)
 	}()
 
 	currentTime := time.Now()
 	softDeleteAsset := NewSoftDeleteAsset(currentTime, currentTime, updatedBy)
 
-	urn := id
+	urn = id
 	if isValidUUID(id) {
 		urn, err = s.assetRepository.SoftDeleteByID(ctx, id, softDeleteAsset)
 		if err != nil {
-			return err
+			return urn, err
 		}
 	} else {
 		if err := s.assetRepository.SoftDeleteByURN(ctx, urn, softDeleteAsset); err != nil {
-			return err
+			return urn, err
 		}
 	}
 
 	softDeleteAsset.URN = urn
-	if err := s.worker.EnqueueSoftDeleteAssetJob(ctx, softDeleteAsset); err != nil {
-		return err
-	}
-
-	return s.lineageRepository.DeleteByURN(ctx, urn)
+	return urn, s.worker.EnqueueSoftDeleteAssetJob(ctx, softDeleteAsset)
 }
 
 func (s *Service) DeleteAssets(ctx context.Context, request DeleteAssetsRequest) (affectedRows uint32, err error) {

internal/server/v1beta1/asset.go

@@ -32,7 +32,7 @@ type AssetService interface {
 	UpsertPatchAsset(ctx context.Context, ast *asset.Asset, upstreams, downstreams []string, patchData map[string]interface{}) (string, error)
 	UpsertPatchAssetWithoutLineage(ctx context.Context, ast *asset.Asset, patchData map[string]interface{}) (string, error)
 	DeleteAsset(ctx context.Context, id string) error
-	SoftDeleteAsset(ctx context.Context, id, updatedBy string) error
+	SoftDeleteAsset(ctx context.Context, id, updatedBy string) (string, error)
 	DeleteAssets(ctx context.Context, request asset.DeleteAssetsRequest) (uint32, error)
 
 	GetLineage(ctx context.Context, urn string, query asset.LineageQuery) (asset.Lineage, error)
@@ -299,7 +299,8 @@ func (server *APIServer) DeleteAsset(ctx context.Context, req *compassv1beta1.De
 		return nil, err
 	}
 
-	if err := server.assetService.SoftDeleteAsset(ctx, req.GetId(), userID); err != nil {
+	_, err = server.assetService.SoftDeleteAsset(ctx, req.GetId(), userID)
+	if err != nil {
 		if errors.As(err, new(asset.InvalidError)) {
 			return nil, status.Error(codes.InvalidArgument, err.Error())
 		}

internal/server/v1beta1/mocks/asset_service.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/goto/compass/core/asset"
+	"github.com/goto/compass/core/user"
 	"github.com/goto/compass/pkg/queryexpr"
 	"github.com/goto/salt/log"
 )
@@ -216,6 +217,19 @@ func (repo *DiscoveryRepository) softDeleteAsset(ctx context.Context, discoveryO
 		})
 	}(time.Now())
 
+	// First get the current version
+	currentVersion, err := repo.getCurrentAssetVersion(ctx, softDeleteAsset.URN)
+	if err != nil {
+		return asset.DiscoveryError{
+			Op:  "GetCurrentVersion",
+			Err: fmt.Errorf("failed to get current version for URN %s: %w", softDeleteAsset.URN, err),
+		}
+	}
+	newVersion, err := asset.IncreaseMinorVersion(currentVersion)
+	if err != nil {
+		return err
+	}
+
 	// Create the update request body
 	body := map[string]interface{}{
 		"query": map[string]interface{}{
@@ -228,33 +242,36 @@ func (repo *DiscoveryRepository) softDeleteAsset(ctx context.Context, discoveryO
                 ctx._source.is_deleted = true;
                 ctx._source.updated_at = params.updated_at;
                 ctx._source.refreshed_at = params.refreshed_at;
-                ctx._source.updated_by = params.updated_by
+                ctx._source.updated_by = params.updated_by;
+                ctx._source.version = params.version;
             `,
 			"lang": "painless",
 			"params": map[string]interface{}{
 				"updated_at":   softDeleteAsset.UpdatedAt,
 				"refreshed_at": softDeleteAsset.RefreshedAt,
-				"updated_by":   softDeleteAsset.UpdatedBy,
+				"updated_by":   user.User{ID: softDeleteAsset.UpdatedBy},
+				"version":      newVersion,
 			},
 		},
 	}
 
-	// Convert body to JSON
-	var buf bytes.Buffer
-	if err := json.NewEncoder(&buf).Encode(body); err != nil {
+	buf, err := encodeBodyRequest(body)
+	if err != nil {
 		return asset.DiscoveryError{
 			Op:  "SoftDeleteByURN",
-			Err: fmt.Errorf("failed to encode request body: %w", err),
+			Err: err,
 		}
 	}
 
 	// Execute UpdateByQuery
 	res, err := repo.cli.client.UpdateByQuery(
 		[]string{defaultSearchIndex},
 		repo.cli.client.UpdateByQuery.WithContext(ctx),
-		repo.cli.client.UpdateByQuery.WithBody(&buf),
+		repo.cli.client.UpdateByQuery.WithBody(buf),
 		repo.cli.client.UpdateByQuery.WithRefresh(true),
 		repo.cli.client.UpdateByQuery.WithIgnoreUnavailable(true),
+		repo.cli.client.UpdateByQuery.WithWaitForCompletion(true),
+		repo.cli.client.UpdateByQuery.WithConflicts("proceed"),
 	)
 	if err != nil {
 		return asset.DiscoveryError{
@@ -276,6 +293,58 @@ func (repo *DiscoveryRepository) softDeleteAsset(ctx context.Context, discoveryO
 	return nil
 }
 
+// Helper function to get current version
+func (repo *DiscoveryRepository) getCurrentAssetVersion(ctx context.Context, urn string) (string, error) {
+	query := map[string]interface{}{
+		"query": map[string]interface{}{
+			"term": map[string]interface{}{
+				"urn.keyword": urn,
+			},
+		},
+		"size":    1,
+		"_source": []string{"version"},
+	}
+
+	var buf bytes.Buffer
+	if err := json.NewEncoder(&buf).Encode(query); err != nil {
+		return "", fmt.Errorf("failed to encode query: %w", err)
+	}
+
+	res, err := repo.cli.client.Search(
+		repo.cli.client.Search.WithContext(ctx),
+		repo.cli.client.Search.WithIndex(defaultSearchIndex),
+		repo.cli.client.Search.WithBody(&buf),
+	)
+	if err != nil {
+		return "", fmt.Errorf("search failed: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		return "", fmt.Errorf("search error: %s", res.String())
+	}
+
+	var result struct {
+		Hits struct {
+			Hits []struct {
+				Source struct {
+					Version string `json:"version"`
+				} `json:"_source"`
+			} `json:"hits"`
+		} `json:"hits"`
+	}
+
+	if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
+		return "", fmt.Errorf("failed to decode response: %w", err)
+	}
+
+	if len(result.Hits.Hits) == 0 {
+		return "", fmt.Errorf("asset with URN %s not found", urn)
+	}
+
+	return result.Hits.Hits[0].Source.Version, nil
+}
+
 func (repo *DiscoveryRepository) indexAsset(ctx context.Context, ast asset.Asset) (err error) {
 	defer func(start time.Time) {
 		const op = "index"
@@ -287,7 +356,7 @@ func (repo *DiscoveryRepository) indexAsset(ctx context.Context, ast asset.Asset
 		})
 	}(time.Now())
 
-	body, err := createUpsertBody(ast)
+	body, err := encodeBodyRequest(ast)
 	if err != nil {
 		return asset.DiscoveryError{
 			Op:  "EncodeAsset",
@@ -327,10 +396,10 @@ func (repo *DiscoveryRepository) indexAsset(ctx context.Context, ast asset.Asset
 	return nil
 }
 
-func createUpsertBody(ast asset.Asset) (io.Reader, error) {
+func encodeBodyRequest(body interface{}) (io.Reader, error) {
 	var buf bytes.Buffer
-	if err := json.NewEncoder(&buf).Encode(ast); err != nil {
-		return nil, fmt.Errorf("encode asset: %w", err)
+	if err := json.NewEncoder(&buf).Encode(body); err != nil {
+		return nil, fmt.Errorf("encode request body: %w", err)
 	}
 
 	return &buf, nil

internal/store/elasticsearch/discovery_repository.go

@@ -23,6 +23,7 @@ type AssetModel struct {
 	Data        JSONMap    `db:"data"`
 	URL         string     `db:"url"`
 	Labels      JSONMap    `db:"labels"`
+	IsDeleted   bool       `db:"is_deleted"`
 	Version     string     `db:"version"`
 	UpdatedBy   UserModel  `db:"updated_by"`
 	CreatedAt   time.Time  `db:"created_at"`
@@ -44,6 +45,7 @@ func (a *AssetModel) toAsset(owners []user.User) asset.Asset {
 		Data:        a.Data,
 		URL:         a.URL,
 		Labels:      a.buildLabels(),
+		IsDeleted:   a.IsDeleted,
 		Owners:      owners,
 		Version:     a.Version,
 		UpdatedBy:   a.UpdatedBy.toUser(),
@@ -65,6 +67,7 @@ func (a *AssetModel) toAssetVersion() (asset.Asset, error) {
 		URN:         a.URN,
 		Type:        asset.Type(a.Type),
 		Service:     a.Service,
+		IsDeleted:   a.IsDeleted,
 		Version:     a.Version,
 		Changelog:   clog,
 		UpdatedBy:   a.UpdatedBy.toUser(),
@@ -96,6 +99,7 @@ func (a *AssetModel) toVersionedAsset(latestAssetVersion asset.Asset) (asset.Ass
 		Description: a.Description,
 		Data:        a.Data,
 		Labels:      a.buildLabels(),
+		IsDeleted:   a.IsDeleted,
 		Owners:      owners,
 		Version:     a.Version,
 		UpdatedBy:   a.UpdatedBy.toUser(),