Merge pull request #82 from opticSquid/fix/process-bugs

Skips already processed files even if evet is replayed, deletes local directory after s3 upload

Author: opticSquid

cloudstorage/fileIO.go

@@ -176,12 +176,35 @@ func UploadDir(ctx context.Context, s3Client *s3.Client, event *files.File, cfg
 	})
 
 	if err != nil {
-		log.Error(ctx, "Error walking the folder", "folder", currentDir, "error", err)
+		log.Error(ctx, "Error walking the directory", "directory", currentDir, "error", err)
 	} else {
 		log.Info(ctx, "Folder uploaded successfully", "directory", currentDir)
+		span.SetStatus(codes.Ok, "Folder uploaded successfully")
+		log.Info(ctx, "Deleting the local directory", "directory", currentDir)
+		deleteUploadedDirFromLocal(ctx, event, log)
 	}
 }
 
+func deleteUploadedDirFromLocal(ctx context.Context, file *files.File, log logger.Log) {
+	tr := otel.Tracer("hangout.storage.cloudstorage")
+	ctx, span := tr.Start(ctx, "DeleteUploadedFiles")
+	defer span.End()
+
+	span.SetAttributes(
+		attribute.String("file.name", file.Filename),
+		attribute.Int("file.userId", int(file.UserId)),
+	)
+
+	baseFilename := strings.Split(file.Filename, ".")[0]
+	dirPath := "/tmp/" + baseFilename
+	err := os.RemoveAll(dirPath)
+	if err != nil {
+		log.Error(ctx, "Could not delete local directory", "directory", dirPath, "error", err)
+		span.RecordError(err)
+	}
+	log.Info(ctx, "Local directory deleted successfully", "file", file.Filename, "directory", dirPath)
+}
+
 func getContentType(extension string) string {
 	switch extension {
 	case ".mpd":

database/crud.go

@@ -10,6 +10,17 @@ import (
 	"hangout.com/core/storage-service/logger"
 )
 
+// IsAlreadyProcessed checks if the file with the given filename has process_status SUCCESS.
+func (dbConn *DatabaseConnectionPool) IsAlreadyProcessed(ctx context.Context, filename string) (bool, error) {
+	var currentStatus model.ProcessStatus
+	row := dbConn.pool.QueryRow(ctx, "SELECT process_status FROM media WHERE filename = $1", filename)
+	err := row.Scan(&currentStatus)
+	if err != nil {
+		return false, err
+	}
+	return currentStatus == model.SUCCESS, nil
+}
+
 func (dbConn *DatabaseConnectionPool) UpdateProcessingStatus(ctx context.Context, filename string, processStatus model.ProcessStatus, log logger.Log) error {
 	tr := otel.Tracer("hangout.storage.database")
 	ctx, span := tr.Start(ctx, "UpdateProcessingStatus")

files/file.go

@@ -51,8 +51,14 @@ func (f *File) Process(workerContext context.Context, cfg *koanf.Koanf, dbConnPo
 	} else {
 		mediaFile := &pipeline.Video{Filename: f.Filename}
 		log.Info(ctx, "marking file status as PROCESSING in db", "filename", f.Filename)
-		dbConnPool.UpdateProcessingStatus(ctx, f.Filename, model.PROCESSING, log)
-		err := mediaFile.ProcessMedia(ctx, cfg, log)
+		err := dbConnPool.UpdateProcessingStatus(ctx, f.Filename, model.PROCESSING, log)
+		if err != nil {
+			log.Error(ctx, "could not mark file as PROCESSING in db", "filename", f.Filename)
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
+			return err
+		}
+		err = mediaFile.ProcessMedia(ctx, cfg, log)
 		if err != nil {
 			log.Error(ctx, "marking file status as FAILED in db", "filename", f.Filename)
 			dbConnPool.UpdateProcessingStatus(ctx, f.Filename, model.FAIL, log)

worker/worker-pool.go

@@ -68,17 +68,33 @@ func (worker *WorkerPool) do(workerId int, file *files.File, workerLogger logger
 
 	workerLogger.Info(ctx, "starting file processing", "file-name", file.Filename, "user-id", file.UserId)
 
-	// download the given file from cloud storage
+	// Check if already processed
+	isProcessed, err := worker.dbConnPool.IsAlreadyProcessed(ctx, file.Filename)
+	if err != nil {
+		workerLogger.Error(ctx, "error checking process status", "error", err.Error())
+		// Optionally: do not acknowledge, so it can be retried
+		return
+	}
+	if isProcessed {
+		workerLogger.Info(ctx, "file already processed, acknowledging and skipping", "file-name", file.Filename)
+		if file.KafkaSession != nil && file.KafkaMessage != nil {
+			file.KafkaSession.MarkMessage(file.KafkaMessage, "")
+		}
+		return
+	}
+
+	// Not processed: download, process, upload, then acknowledge
 	cloudstorage.Download(ctx, s3Client, file, worker.cfg, workerLogger)
-	// process the file
-	err := file.Process(ctx, worker.cfg, worker.dbConnPool, workerLogger)
+	err = file.Process(ctx, worker.cfg, worker.dbConnPool, workerLogger)
 	if err != nil {
 		workerLogger.Error(ctx, "could not process file", "error", err.Error())
+		// Optionally: do not acknowledge, so it can be retried
+		return
 	}
-	// upload the given file to cloud storage
+
 	cloudstorage.UploadDir(ctx, s3Client, file, worker.cfg, workerLogger)
 	if file.KafkaSession != nil && file.KafkaMessage != nil {
-		file.KafkaSession.MarkMessage(file.KafkaMessage, "") // Acknowledge the Kafka event and mark as completed
+		file.KafkaSession.MarkMessage(file.KafkaMessage, "")
 	}
 	workerLogger.Info(ctx, "finished file processing", "file-name", file.Filename)
 }