fix: insert logic and make it single responsibility for upsert patch and upsert post

Author: Muhammad Luthfi Fahlevi

internal/store/postgres/asset_repository.go

@@ -322,22 +322,61 @@ func (r *AssetRepository) getByVersion(
 	return ast.toVersionedAsset(latest)
 }
 
+// UpsertPatch creates a new asset if it does not exist yet.
+// It updates if asset does exist.
+// Checking existence is done using "urn", "type", and "service" fields
+// And will revalidate again with additional: "data" and "name" fields.
 func (r *AssetRepository) UpsertPatch(ctx context.Context, ast *asset.Asset, patchData map[string]interface{}) (asset.Asset, error) {
-	return r.upsertWithPatchOption(ctx, ast, patchData)
-}
+	var upsertedAsset asset.Asset
+	err := r.client.RunWithinTx(ctx, func(tx *sqlx.Tx) error {
+		fetchedAsset, err := r.GetByURNWithTx(ctx, tx, ast.URN)
+		if errors.As(err, new(asset.NotFoundError)) {
+			err = nil
+		}
+		if err != nil {
+			return fmt.Errorf("error getting asset by URN: %w", err)
+		}
 
-func (r *AssetRepository) Upsert(ctx context.Context, ast *asset.Asset) (asset.Asset, error) {
-	return r.upsertWithPatchOption(ctx, ast, nil)
+		if fetchedAsset.ID == "" {
+			// insert flow
+			upsertedAsset, err = r.insert(ctx, tx, ast)
+			if err != nil {
+				return fmt.Errorf("error inserting asset to DB: %w", err)
+			}
+			return nil
+		}
+
+		// update flow
+		if err := copier.CopyWithOption(&ast, &fetchedAsset, copier.Option{DeepCopy: true}); err != nil {
+			return err
+		}
+		ast.Patch(patchData)
+		if err := r.validateAsset(*ast); err != nil {
+			return err
+		}
+		changelog, err := fetchedAsset.Diff(ast)
+		if err != nil {
+			return fmt.Errorf("error diffing two assets: %w", err)
+		}
+
+		upsertedAsset, err = r.update(ctx, tx, ast, &fetchedAsset, changelog)
+		if err != nil {
+			return fmt.Errorf("error updating asset to DB: %w", err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return asset.Asset{}, err
+	}
+
+	return upsertedAsset, nil
 }
 
 // Upsert creates a new asset if it does not exist yet.
 // It updates if asset does exist.
-// Checking existence is done using "urn", "type", and "service" fields.
-func (r *AssetRepository) upsertWithPatchOption( //nolint:gocognit
-	ctx context.Context,
-	ast *asset.Asset,
-	patchData map[string]interface{},
-) (asset.Asset, error) {
+// Checking existence is done using "urn", "type", "name", "data", and "service" fields.
+func (r *AssetRepository) Upsert(ctx context.Context, ast *asset.Asset) (asset.Asset, error) {
 	var upsertedAsset asset.Asset
 	err := r.client.RunWithinTx(ctx, func(tx *sqlx.Tx) error {
 		fetchedAsset, err := r.GetByURNWithTx(ctx, tx, ast.URN)
@@ -347,27 +386,17 @@ func (r *AssetRepository) upsertWithPatchOption( //nolint:gocognit
 		if err != nil {
 			return fmt.Errorf("error getting asset by URN: %w", err)
 		}
-		if err := r.validateAsset(fetchedAsset); err != nil {
-			return err
-		}
 
 		if fetchedAsset.ID == "" {
 			// insert flow
 			upsertedAsset, err = r.insert(ctx, tx, ast)
 			if err != nil {
 				return fmt.Errorf("error inserting asset to DB: %w", err)
 			}
+
 			return nil
 		}
 
-		// update flow
-		if patchData != nil {
-			err := copier.CopyWithOption(&ast, &fetchedAsset, copier.Option{DeepCopy: true})
-			if err != nil {
-				return err
-			}
-			ast.Patch(patchData)
-		}
 		changelog, err := fetchedAsset.Diff(ast)
 		if err != nil {
 			return fmt.Errorf("error diffing two assets: %w", err)
@@ -614,6 +643,7 @@ func (r *AssetRepository) insert(ctx context.Context, tx *sqlx.Tx, ast *asset.As
 			"created_at", "updated_by", "updated_at", "refreshed_at", "version").
 		Values(ast.URN, ast.Type, ast.Service, ast.Name, ast.Description, ast.Data, ast.URL, ast.Labels,
 			ast.CreatedAt, ast.UpdatedBy.ID, ast.UpdatedAt, currentTime, asset.BaseVersion).
+		Suffix("RETURNING \"id\"").
 		PlaceholderFormat(sq.Dollar).
 		ToSql()
 	if err != nil {
@@ -622,9 +652,12 @@ func (r *AssetRepository) insert(ctx context.Context, tx *sqlx.Tx, ast *asset.As
 
 	ast.Version = asset.BaseVersion
 
-	if err := r.execContext(ctx, tx, query, args...); err != nil {
+	var id string
+	err = tx.QueryRowContext(ctx, query, args...).Scan(&id)
+	if err != nil {
 		return asset.Asset{}, fmt.Errorf("run insert query: %w", err)
 	}
+	ast.ID = id
 
 	users, err := r.createOrFetchUsers(ctx, tx, ast.Owners)
 	if err != nil {

internal/store/postgres/asset_repository_test.go

@@ -87,7 +87,7 @@ func (r *AssetRepositoryTestSuite) createUsers(userRepo user.Repository) []user.
 	return users
 }
 
-func (r *AssetRepositoryTestSuite) BeforeTest() {
+func (r *AssetRepositoryTestSuite) BeforeTest(suiteName, testName string) {
 	err := testutils.RunMigrationsWithClient(r.T(), r.client)
 	r.NoError(err)
 
@@ -607,18 +607,17 @@ func (r *AssetRepositoryTestSuite) TestGetByID() {
 		}
 
 		var err error
-		upsertedAsset, err := r.repository.Upsert(r.ctx, &asset1)
+		upsertedAsset1, err := r.repository.Upsert(r.ctx, &asset1)
 		r.Require().NoError(err)
-		r.NotEmpty(upsertedAsset.ID)
+		r.NotEmpty(upsertedAsset1.ID)
 
-		upsertedAsset, err = r.repository.Upsert(r.ctx, &asset2)
+		upsertedAsset2, err := r.repository.Upsert(r.ctx, &asset2)
 		r.Require().NoError(err)
-		r.NotEmpty(upsertedAsset.ID)
+		r.NotEmpty(upsertedAsset2.ID)
 
-		result, err := r.repository.GetByID(r.ctx, asset2.ID)
+		result, err := r.repository.GetByID(r.ctx, upsertedAsset2.ID)
 		r.NoError(err)
-		asset2.UpdatedBy = r.users[1]
-		r.assertAsset(&asset2, &result)
+		r.assertAsset(&upsertedAsset2, &result)
 	})
 
 	r.Run("return owners if any", func() {
@@ -637,10 +636,8 @@ func (r *AssetRepositoryTestSuite) TestGetByID() {
 		r.Require().NoError(err)
 		r.Require().NotEmpty(upsertedAsset.ID)
 
-		result, err := r.repository.GetByID(r.ctx, ast.ID)
-		r.NoError(err)
-		r.Len(result.Owners, len(ast.Owners))
-		for i, owner := range result.Owners {
+		r.Len(upsertedAsset.Owners, len(ast.Owners))
+		for i, owner := range upsertedAsset.Owners {
 			r.Equal(ast.Owners[i].ID, owner.ID)
 		}
 	})
@@ -1471,7 +1468,7 @@ func (r *AssetRepositoryTestSuite) TestAddProbe() {
 	})
 
 	r.Run("should populate CreatedAt and persist probe", func() {
-		r.BeforeTest()
+		r.BeforeTest("", "")
 		ast := asset.Asset{
 			URN:       "urn-add-probe-1",
 			Type:      typeJob,