flag assets as deleted

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/golang/glog v1.2.0
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/livepeer/catalyst-api v0.1.2-0.20230925142340-c311569665b4
-	github.com/livepeer/go-api-client v0.4.19-0.20240308102249-998a7fa05134
+	github.com/livepeer/go-api-client v0.4.19-0.20240311145302-1abd53df256c
 	github.com/livepeer/go-tools v0.3.2
 	github.com/livepeer/livepeer-data v0.7.5-0.20230927031152-b938ac1dc665
 	github.com/peterbourgon/ff v1.7.1

go.sum

@@ -281,10 +281,8 @@ github.com/libp2p/go-netroute v0.2.0 h1:0FpsbsvuSnAhXFnCY0VLFbJOzaK0VnP0r1QT/o4n
 github.com/libp2p/go-openssl v0.1.0 h1:LBkKEcUv6vtZIQLVTegAil8jbNpJErQ9AnT+bWV+Ooo=
 github.com/livepeer/catalyst-api v0.1.2-0.20230925142340-c311569665b4 h1:UfiMdEDGa88yqYD9+i1+ldAex9Kf1+3jbq+wBrmZccM=
 github.com/livepeer/catalyst-api v0.1.2-0.20230925142340-c311569665b4/go.mod h1:Ybiub5AGDrDfvyh1MWdIa551LAwhx/6lSpbQlgb1W1Q=
-github.com/livepeer/go-api-client v0.4.17 h1:/VKSUTn0hP8dHuAyA2He0kf/f0Av9iOLFnAeNStol6s=
-github.com/livepeer/go-api-client v0.4.17/go.mod h1:Jdb+RI7JyzEZOHd1GUuKofwFDKMO/btTa80SdpUpYQw=
-github.com/livepeer/go-api-client v0.4.19-0.20240308102249-998a7fa05134 h1:HBMmeVR7QBH7FWJDUFEHwdNf2Krh0qGou3KK+EqA1sY=
-github.com/livepeer/go-api-client v0.4.19-0.20240308102249-998a7fa05134/go.mod h1:Jdb+RI7JyzEZOHd1GUuKofwFDKMO/btTa80SdpUpYQw=
+github.com/livepeer/go-api-client v0.4.19-0.20240311145302-1abd53df256c h1:KPGkwuKvAbHCADy3hssTCfJVh0wxYMlXVTehEikljCc=
+github.com/livepeer/go-api-client v0.4.19-0.20240311145302-1abd53df256c/go.mod h1:Jdb+RI7JyzEZOHd1GUuKofwFDKMO/btTa80SdpUpYQw=
 github.com/livepeer/go-tools v0.3.2 h1:5pOUrOmkkGbbcWnpCt2yrSD6cD85G4GcpO4B25NpMJM=
 github.com/livepeer/go-tools v0.3.2/go.mod h1:qs31y68b3PQPmSr8nR8l5WQiIWI623z6pqOccqebjos=
 github.com/livepeer/livepeer-data v0.7.5-0.20230927031152-b938ac1dc665 h1:EXlI922Fsv9lyIw1LQ7pZN6slCuYya8NQrCFWN8INg4=

task/runner.go

@@ -581,60 +581,77 @@ func (r *runner) CronJobForAssetDeletion(ctx context.Context) error {
 				continue
 			}
 			for _, asset := range assets {
-				_, _, assetOS, err := r.getAssetAndOS(asset.ID)
+				err := deleteAsset(asset, r, ctx)
 				if err != nil {
-					glog.Errorf("Error getting asset object store session: %v", err)
-					continue
+					glog.Errorf("Error deleting asset %v: %v", asset.ID, err)
 				}
+			}
+		}
+	}
+}
 
-				directory := asset.PlaybackID
-				var totalDeleted int
-
-				// Initially list files
-				pi, err := assetOS.ListFiles(ctx, directory, "/")
-				if err != nil {
-					glog.Errorf("Error listing files for asset %v: %v", asset.ID, err)
-					continue
-				}
+func deleteAsset(asset *api.Asset, r *runner, ctx context.Context) error {
+	_, _, assetOS, err := r.getAssetAndOS(asset.ID)
+	if err != nil {
+		glog.Errorf("Error getting asset object store session: %v", err)
+		return err
+	}
 
-				for pi != nil {
-					for _, file := range pi.Files() {
-						err := assetOS.DeleteFile(ctx, file.Name)
-						if err != nil {
-							glog.Errorf("Error deleting file %v: %v", file.Name, err)
-							continue
-						}
-						totalDeleted++
-					}
-
-					if pi.HasNextPage() {
-						pi, err = pi.NextPage()
-						if err != nil {
-							glog.Errorf("Failed to load next page of files for asset %v: %v", asset.ID, err)
-							break
-						}
-					} else {
-						break // No more pages
-					}
-				}
+	directory := asset.PlaybackID
+	var totalDeleted int
 
-				glog.Infof("Deleted %v files from asset=%v", totalDeleted, asset.ID)
+	// Initially list files
+	pi, err := assetOS.ListFiles(ctx, directory, "/")
+	if err != nil {
+		glog.Errorf("Error listing files for asset %v: %v", asset.ID, err)
+		return err
+	}
 
-				if ipfs := asset.AssetSpec.Storage.IPFS; ipfs != nil {
-					err = r.UnpinFromIpfs(ctx, ipfs.CID, "cid")
-					if err != nil {
-						glog.Errorf("Error unpinning from IPFS %v", ipfs.CID)
-					}
-					err = r.UnpinFromIpfs(ctx, ipfs.NFTMetadata.CID, "nftMetadataCid")
-					if err != nil {
-						glog.Errorf("Error unpinning metadata from IPFS %v", ipfs.NFTMetadata.CID)
-					}
+	for pi != nil {
+		for _, file := range pi.Files() {
+			err := assetOS.DeleteFile(ctx, file.Name)
+			if err != nil {
+				glog.Errorf("Error deleting file %v: %v", file.Name, err)
+				continue
+			}
+			totalDeleted++
+		}
 
-					glog.Infof("Unpinned asset=%v from IPFS", asset.ID)
-				}
+		if pi.HasNextPage() {
+			pi, err = pi.NextPage()
+			if err != nil {
+				glog.Errorf("Failed to load next page of files for asset %v: %v", asset.ID, err)
+				break
 			}
+		} else {
+			break // No more pages
+		}
+	}
+
+	glog.Infof("Deleted %v files from asset=%v", totalDeleted, asset.ID)
+
+	if ipfs := asset.AssetSpec.Storage.IPFS; ipfs != nil {
+		err = r.UnpinFromIpfs(ctx, ipfs.CID, "cid")
+		if err != nil {
+			glog.Errorf("Error unpinning from IPFS %v", ipfs.CID)
+			return err
+		}
+		err = r.UnpinFromIpfs(ctx, ipfs.NFTMetadata.CID, "nftMetadataCid")
+		if err != nil {
+			glog.Errorf("Error unpinning metadata from IPFS %v", ipfs.NFTMetadata.CID)
+			return err
 		}
+
+		glog.Infof("Unpinned asset=%v from IPFS", asset.ID)
 	}
+
+	err = r.lapi.FlagAssetAsDeleted(asset.ID)
+	if err != nil {
+		glog.Errorf("Error flagging asset as deleted: %v", err)
+		return err
+	}
+
+	return nil
 }
 
 type taskErrInfo struct {