feat: create delete assets API

Makefile

@@ -26,7 +26,7 @@ help:           ##@help show this help
 NAME="github.com/goto/compass"
 VERSION=$(shell git describe --always --tags 2>/dev/null)
 COVERFILE="/tmp/compass.coverprofile"
-PROTON_COMMIT := "fe99cc96e060085d6052096e9ba59b4038c691c6"
+PROTON_COMMIT := "8375eddcb23d38f601f6036c676493d8feb84a7e"
 
 TOOLS_MOD_DIR = ./tools
 TOOLS_DIR = $(abspath ./.tools)

cli/server.go

@@ -148,12 +148,14 @@ func runServer(ctx context.Context, cfg *Config) error {
 		}
 	}()
 
-	assetService := asset.NewService(asset.ServiceDeps{
+	assetService, cancel := asset.NewService(asset.ServiceDeps{
 		AssetRepo:     assetRepository,
 		DiscoveryRepo: discoveryRepository,
 		LineageRepo:   lineageRepository,
 		Worker:        wrkr,
+		Logger:        logger,
 	})
+	defer cancel()
 
 	// init discussion
 	discussionRepository, err := postgres.NewDiscussionRepository(pgClient, 0)

compass.yaml.example

@@ -68,6 +68,7 @@ worker:
     sync_job_timeout: 15m
     index_job_timeout: 5s
     delete_job_timeout: 5s
+    max_attempt_retry: 3
 
 client:
     host: localhost:8081

core/asset/asset.go

@@ -6,12 +6,14 @@ import (
 	"time"
 
 	"github.com/goto/compass/core/user"
+	"github.com/goto/compass/pkg/queryexpr"
 	"github.com/r3labs/diff/v2"
 )
 
 type Repository interface {
 	GetAll(context.Context, Filter) ([]Asset, error)
 	GetCount(context.Context, Filter) (int, error)
+	GetCountByQueryExpr(ctx context.Context, queryExpr queryexpr.ExprStr) (int, error)
 	GetByID(ctx context.Context, id string) (Asset, error)
 	GetByURN(ctx context.Context, urn string) (Asset, error)
 	GetVersionHistory(ctx context.Context, flt Filter, id string) ([]Asset, error)
@@ -21,6 +23,7 @@ type Repository interface {
 	Upsert(ctx context.Context, ast *Asset) (string, error)
 	DeleteByID(ctx context.Context, id string) error
 	DeleteByURN(ctx context.Context, urn string) error
+	DeleteByQueryExpr(ctx context.Context, queryExpr queryexpr.ExprStr) ([]string, error)
 	AddProbe(ctx context.Context, assetURN string, probe *Probe) error
 	GetProbes(ctx context.Context, assetURN string) ([]Probe, error)
 	GetProbesWithFilter(ctx context.Context, flt ProbesFilter) (map[string][]Probe, error)
@@ -40,6 +43,7 @@ type Asset struct {
 	Owners      []user.User            `json:"owners,omitempty" diff:"owners"`
 	CreatedAt   time.Time              `json:"created_at" diff:"-"`
 	UpdatedAt   time.Time              `json:"updated_at" diff:"-"`
+	RefreshedAt *time.Time             `json:"refreshed_at" diff:"-"`
 	Version     string                 `json:"version" diff:"-"`
 	UpdatedBy   user.User              `json:"updated_by" diff:"-"`
 	Changelog   diff.Changelog         `json:"changelog,omitempty" diff:"-"`

core/asset/delete_asset_expr.go

@@ -0,0 +1,73 @@
+package asset
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/goto/compass/pkg/generichelper"
+	"github.com/goto/compass/pkg/queryexpr"
+)
+
+var (
+	assetJSONTagsSchema              = generichelper.GetJSONTags(Asset{})
+	errTypeOrServiceHasWrongOperator = errors.New("identifier type and service must be equals (==) or IN operator")
+	errMissRequiredIdentifier        = errors.New("must exists these identifiers: refreshed_at, type, and service")
+)
+
+type DeleteAssetExpr struct {
+	queryexpr.ExprStr
+}
+
+func (d DeleteAssetExpr) ToQuery() (string, error) {
+	return d.ExprStr.ToQuery()
+}
+
+func (d DeleteAssetExpr) Validate() error {
+	identifiersWithOperator, err := queryexpr.GetIdentifiersMap(d.ExprStr.String())
+	if err != nil {
+		return err
+	}
+
+	if err := d.isRequiredIdentifiersExist(identifiersWithOperator); err != nil {
+		return err
+	}
+
+	if err := d.isUsingRightOperator(identifiersWithOperator); err != nil {
+		return err
+	}
+
+	return d.isAllIdentifiersExistInStruct(identifiersWithOperator)
+}
+
+func (DeleteAssetExpr) isRequiredIdentifiersExist(identifiersWithOperator map[string]string) error {
+	isExist := func(jsonTag string) bool {
+		return identifiersWithOperator[jsonTag] != ""
+	}
+	mustExist := isExist("refreshed_at") && isExist("type") && isExist("service")
+	if !mustExist {
+		return errMissRequiredIdentifier
+	}
+	return nil
+}
+
+func (DeleteAssetExpr) isUsingRightOperator(identifiersWithOperator map[string]string) error {
+	isOperatorEqualsOrIn := func(jsonTag string) bool {
+		return identifiersWithOperator[jsonTag] == "==" || strings.ToUpper(identifiersWithOperator[jsonTag]) == "IN"
+	}
+	if !isOperatorEqualsOrIn("type") || !isOperatorEqualsOrIn("service") {
+		return errTypeOrServiceHasWrongOperator
+	}
+	return nil
+}
+
+func (DeleteAssetExpr) isAllIdentifiersExistInStruct(identifiersWithOperator map[string]string) error {
+	identifiers := generichelper.GetMapKeys(identifiersWithOperator)
+	for _, identifier := range identifiers {
+		isFieldValid := generichelper.Contains(assetJSONTagsSchema, identifier)
+		if !isFieldValid {
+			return fmt.Errorf("%s is not a valid identifier", identifier)
+		}
+	}
+	return nil
+}

core/asset/delete_asset_expr_test.go

@@ -0,0 +1,150 @@
+package asset_test
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/goto/compass/core/asset"
+	"github.com/goto/compass/pkg/queryexpr"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDeleteAssetExpr_ToQuery(t *testing.T) {
+	queryExp := `name == "John" || service not in ["test1","test2","test3"]`
+	sqlExpr := queryexpr.SQLExpr(queryExp)
+	esExpr := queryexpr.ESExpr(queryExp)
+	wrongExpr := queryexpr.SQLExpr("findLast(")
+	tests := []struct {
+		name    string
+		exprStr queryexpr.ExprStr
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "convert to SQL query",
+			exprStr: asset.DeleteAssetExpr{
+				ExprStr: sqlExpr,
+			},
+			want:    "((name = 'John') OR (service NOT IN ('test1', 'test2', 'test3')))",
+			wantErr: false,
+		},
+		{
+			name: "convert to ES query",
+			exprStr: asset.DeleteAssetExpr{
+				ExprStr: esExpr,
+			},
+			want:    `{"query":{"bool":{"should":[{"term":{"name":"John"}},{"bool":{"must_not":[{"terms":{"service.keyword":["test1","test2","test3"]}}]}}]}}}`,
+			wantErr: false,
+		},
+		{
+			name: "got error due to wrong syntax",
+			exprStr: asset.DeleteAssetExpr{
+				ExprStr: wrongExpr,
+			},
+			want:    "",
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			d := asset.DeleteAssetExpr{
+				ExprStr: tt.exprStr,
+			}
+			got, err := d.ToQuery()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("ToQuery() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("ToQuery() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestDeleteAssetExpr_Validate(t *testing.T) {
+	tests := []struct {
+		name      string
+		exprStrFn func() queryexpr.ExprStr
+		expectErr error
+		wantErr   bool
+	}{
+		{
+			name: "error get identifiers map",
+			exprStrFn: func() queryexpr.ExprStr {
+				wrongExpr := queryexpr.SQLExpr("findLast(")
+				return asset.DeleteAssetExpr{
+					ExprStr: wrongExpr,
+				}
+			},
+			expectErr: errors.New("error parsing expression"),
+			wantErr:   true,
+		},
+		{
+			name: "error miss refreshed_at not exist",
+			exprStrFn: func() queryexpr.ExprStr {
+				missRefreshedAt := queryexpr.SQLExpr(`updated_at < "2023-12-12 23:59:59" && type == "table" && service in ["test1","test2","test3"]`)
+				return asset.DeleteAssetExpr{
+					ExprStr: missRefreshedAt,
+				}
+			},
+			expectErr: errors.New("must exists these identifiers: refreshed_at, type, and service"),
+			wantErr:   true,
+		},
+		{
+			name: "error miss type not exist",
+			exprStrFn: func() queryexpr.ExprStr {
+				missType := queryexpr.SQLExpr(`refreshed_at < "2023-12-12 23:59:59" && service in ["test1","test2","test3"]`)
+				return asset.DeleteAssetExpr{
+					ExprStr: missType,
+				}
+			},
+			expectErr: errors.New("must exists these identifiers: refreshed_at, type, and service"),
+			wantErr:   true,
+		},
+		{
+			name: "error miss service not exist",
+			exprStrFn: func() queryexpr.ExprStr {
+				missService := queryexpr.SQLExpr(`refreshed_at < "2023-12-12 23:59:59" && type == "table"`)
+				return asset.DeleteAssetExpr{
+					ExprStr: missService,
+				}
+			},
+			expectErr: errors.New("must exists these identifiers: refreshed_at, type, and service"),
+			wantErr:   true,
+		},
+		{
+			name: "error wrong operator for type identifier",
+			exprStrFn: func() queryexpr.ExprStr {
+				wrongTypeOperator := queryexpr.SQLExpr(`refreshed_at < "2023-12-12 23:59:59" && type != "table" && service in ["test1","test2","test3"]`)
+				return asset.DeleteAssetExpr{
+					ExprStr: wrongTypeOperator,
+				}
+			},
+			expectErr: errors.New("identifier type and service must be equals (==) or IN operator"),
+			wantErr:   true,
+		},
+		{
+			name: "error wrong operator for service identifier",
+			exprStrFn: func() queryexpr.ExprStr {
+				wrongServiceOperator := queryexpr.SQLExpr(`refreshed_at < "2023-12-12 23:59:59" && type != "table" && service not in ["test1","test2","test3"]`)
+				return asset.DeleteAssetExpr{
+					ExprStr: wrongServiceOperator,
+				}
+			},
+			expectErr: errors.New("identifier type and service must be equals (==) or IN operator"),
+			wantErr:   true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.exprStrFn().Validate()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr)
+			}
+			if err != nil {
+				assert.ErrorContains(t, err, tt.expectErr.Error())
+			}
+		})
+	}
+}

core/asset/delete_assets.go

@@ -0,0 +1,6 @@
+package asset
+
+type DeleteAssetsRequest struct {
+	QueryExpr string
+	DryRun    bool
+}

core/asset/discovery.go

@@ -3,12 +3,14 @@ package asset
 //go:generate mockery --name=DiscoveryRepository -r --case underscore --with-expecter --structname DiscoveryRepository --filename discovery_repository.go --output=./mocks
 import (
 	"context"
+	"github.com/goto/compass/pkg/queryexpr"
 )
 
 type DiscoveryRepository interface {
 	Upsert(context.Context, Asset) error
 	DeleteByID(ctx context.Context, assetID string) error
 	DeleteByURN(ctx context.Context, assetURN string) error
+	DeleteByQueryExpr(ctx context.Context, queryExpr queryexpr.ExprStr) error
 	Search(ctx context.Context, cfg SearchConfig) (results []SearchResult, err error)
 	Suggest(ctx context.Context, cfg SearchConfig) (suggestions []string, err error)
 	GroupAssets(ctx context.Context, cfg GroupConfig) (results []GroupResult, err error)

core/asset/errors.go

@@ -10,6 +10,7 @@ var (
 	ErrEmptyID     = errors.New("asset does not have ID")
 	ErrProbeExists = errors.New("asset probe already exists")
 	ErrEmptyURN    = errors.New("asset does not have URN")
+	ErrEmptyQuery  = errors.New("query is empty")
 	ErrUnknownType = errors.New("unknown type")
 	ErrNilAsset    = errors.New("nil asset")
 )

core/asset/lineage.go

@@ -31,6 +31,7 @@ type LineageRepository interface {
 	GetGraph(ctx context.Context, urn string, query LineageQuery) (LineageGraph, error)
 	Upsert(ctx context.Context, urn string, upstreams, downstreams []string) error
 	DeleteByURN(ctx context.Context, urn string) error
+	DeleteByURNs(ctx context.Context, urns []string) error
 }
 
 type LineageGraph []LineageEdge