check if files are already processed if yes skip download also

Author: opticSquid

database/crud.go

@@ -2,7 +2,6 @@ package database
 
 import (
 	"context"
-	"errors"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -11,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")
@@ -21,25 +31,6 @@ func (dbConn *DatabaseConnectionPool) UpdateProcessingStatus(ctx context.Context
 	)
 	defer span.End()
 
-	// Check current process_status
-	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 {
-		span.RecordError(err)
-		span.SetStatus(codes.Error, err.Error())
-		log.Error(ctx, "could not fetch current process status", "error", err)
-		return err
-	}
-
-	if currentStatus == model.SUCCESS {
-		err := errors.New("cannot update process_status: already SUCCESS")
-		span.RecordError(err)
-		span.SetStatus(codes.Error, err.Error())
-		log.Error(ctx, "process_status is already SUCCESS, not updating", "filename", filename)
-		return err
-	}
-
 	query := `UPDATE media SET process_status = $1 where filename = $2`
 	cmdTag, err := dbConn.pool.Exec(ctx, query, processStatus, filename)
 	if err != nil {

worker/worker-pool.go

@@ -68,29 +68,35 @@ 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
-	cloudstorage.Download(ctx, s3Client, file, worker.cfg, workerLogger)
-	// process the file
-	err := file.Process(ctx, worker.cfg, worker.dbConnPool, workerLogger)
+	// Check if already processed
+	isProcessed, err := worker.dbConnPool.IsAlreadyProcessed(ctx, file.Filename)
 	if err != nil {
-		if err.Error() == "cannot update process_status: already SUCCESS" {
-			// Acknowledge and skip upload
-			if file.KafkaSession != nil && file.KafkaMessage != nil {
-				file.KafkaSession.MarkMessage(file.KafkaMessage, "")
-			}
-			workerLogger.Info(ctx, "file already processed, skipping upload", "file-name", file.Filename)
-		} else {
-			workerLogger.Error(ctx, "could not process file", "error", err.Error())
-			// do not acknowledge, so it can be retried
-		}
-	} else {
-		// Success: upload and acknowledge
-		cloudstorage.UploadDir(ctx, s3Client, file, worker.cfg, workerLogger)
+		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, "")
 		}
-		workerLogger.Info(ctx, "finished file processing", "file-name", file.Filename)
+		return
+	}
+
+	// Not processed: download, process, upload, then acknowledge
+	cloudstorage.Download(ctx, s3Client, file, worker.cfg, 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
+	}
+
+	cloudstorage.UploadDir(ctx, s3Client, file, worker.cfg, workerLogger)
+	if file.KafkaSession != nil && file.KafkaMessage != nil {
+		file.KafkaSession.MarkMessage(file.KafkaMessage, "")
 	}
+	workerLogger.Info(ctx, "finished file processing", "file-name", file.Filename)
 }
 
 // Wait ensures all workers complete processing before the program exits