feat: add sync metadata to sftp destination file path

router/batchrouter/asyncdestinationmanager/common/common.go

@@ -137,6 +137,7 @@ type AsyncDestinationStruct struct {
 	AttemptNums           map[int64]int
 	FirstAttemptedAts     map[int64]time.Time
 	OriginalJobParameters map[int64]stdjson.RawMessage
+	PartFileNumber        int
 }
 
 type AsyncFailedPayload struct {

router/batchrouter/asyncdestinationmanager/sftp/manager.go

@@ -27,9 +27,16 @@ func (d *defaultManager) Upload(asyncDestStruct *common.AsyncDestinationStruct)
 		return generateErrorOutput(fmt.Sprintf("error marshalling destination config: %v", err.Error()), asyncDestStruct.ImportingJobIDs, destinationID)
 	}
 
+	jobRunID := gjson.GetBytes(asyncDestStruct.OriginalJobParameters[0], "source_job_run_id").String()
+	metadata := map[string]any{
+		"destinationID":  destinationID,
+		"sourceJobRunID": jobRunID,
+		"partFileNumber": asyncDestStruct.PartFileNumber,
+	}
+
 	result := gjson.ParseBytes(destConfigJSON)
 	uploadFilePath := result.Get("filePath").String()
-	uploadFilePath = getUploadFilePath(uploadFilePath)
+	uploadFilePath = getUploadFilePath(uploadFilePath, metadata)
 	fileFormat := result.Get("fileFormat").String()
 
 	// Generate temporary file based on the destination's file format

router/batchrouter/asyncdestinationmanager/sftp/sftp_test.go

@@ -282,33 +282,39 @@ var _ = Describe("SFTP test", func() {
 
 		It("TestReplaceDynamicVariables", func() {
 			initSFTP()
-			input := "/path/to/{YYYY}/{MM}/{DD}/{hh}/{mm}/{ss}/{ms}/{timestampInSec}/{timestampInMS}"
-			expected := fmt.Sprintf("/path/to/%d/%02d/%02d/%02d/%02d/%02d/%03d/%d/%d", time.Now().Year(), time.Now().Month(), time.Now().Day(), time.Now().Hour(), time.Now().Minute(), time.Now().Second(), time.Now().Nanosecond()/1e6, time.Now().Unix(), time.Now().UnixNano()/1e6)
-			received := getUploadFilePath(input)
+			now := time.Now()
+			input := "/path/to/{destinationID}_{jobRunID}/{YYYY}/{MM}/{DD}/{hh}/{mm}/{ss}/{ms}/{timestampInSec}/{timestampInMS}"
+			expected := fmt.Sprintf("/path/to/%s_%s/%d/%02d/%02d/%02d/%02d/%02d/%03d/%d/%d_1", "some_destination_id", "some_source_job_run_id", now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), now.Second(), now.Nanosecond()/1e6, now.Unix(), now.UnixNano()/1e6)
+			metadata := map[string]any{
+				"destinationID":  "some_destination_id",
+				"sourceJobRunID": "some_source_job_run_id",
+				"partFileNumber": 1,
+			}
+			received := getUploadFilePath(input, metadata)
 			Expect(received).To(Equal(expected))
 		})
 
 		It("TestNoDynamicVariables", func() {
 			initSFTP()
 			input := "/path/to/file.txt"
 			expected := "/path/to/file.txt"
-			received := getUploadFilePath(input)
+			received := getUploadFilePath(input, nil)
 			Expect(received).To(Equal(expected))
 		})
 
 		It("TestEmptyInputPath", func() {
 			initSFTP()
 			input := ""
 			expected := ""
-			received := getUploadFilePath(input)
+			received := getUploadFilePath(input, nil)
 			Expect(received).To(Equal(expected))
 		})
 
 		It("TestInvalidDynamicVariables", func() {
 			initSFTP()
 			input := "/path/to/{invalid}/file.txt"
 			expected := "/path/to/{invalid}/file.txt"
-			received := getUploadFilePath(input)
+			received := getUploadFilePath(input, nil)
 			Expect(received).To(Equal(expected))
 		})
 	})

router/batchrouter/asyncdestinationmanager/sftp/utils.go

@@ -222,7 +222,7 @@ func getTempFilePath() (string, error) {
 	return tmpFilePath, nil
 }
 
-func getUploadFilePath(path string) string {
+func getUploadFilePath(path string, metadata map[string]any) string {
 	// Get the current date and time
 	now := time.Now()
 	// Replace dynamic variables with their actual values
@@ -246,13 +246,24 @@ func getUploadFilePath(path string) string {
 			return strconv.FormatInt(now.Unix(), 10)
 		case "{timestampInMS}":
 			return strconv.FormatInt(now.UnixNano()/1e6, 10)
+		case "{destinationID}":
+			return metadata["destinationID"].(string)
+		case "{jobRunID}":
+			return metadata["sourceJobRunID"].(string)
 		default:
 			// If the dynamic variable is not recognized, keep it unchanged
 			return match
 		}
 	})
 
-	return result
+	partFileNumber, ok := metadata["partFileNumber"].(int)
+	if !ok {
+		return result
+	}
+
+	ext := filepath.Ext(result)
+	base := strings.TrimSuffix(result, ext)
+	return fmt.Sprintf("%s_%d%s", base, partFileNumber, ext)
 }
 
 func generateErrorOutput(err string, importingJobIds []int64, destinationID string) common.AsyncUploadOutput {

router/batchrouter/handle_async.go

@@ -338,6 +338,7 @@
 				timeout := uploadIntervalMap[destinationID]
 				if brt.asyncDestinationStruct[destinationID].Exists && (brt.asyncDestinationStruct[destinationID].CanUpload || timeElapsed > timeout) {
 					brt.asyncDestinationStruct[destinationID].CanUpload = true
+					brt.asyncDestinationStruct[destinationID].PartFileNumber++
 					uploadResponse := brt.asyncDestinationStruct[destinationID].Manager.Upload(brt.asyncDestinationStruct[destinationID])
 					if uploadResponse.ImportingParameters != nil && len(uploadResponse.ImportingJobIDs) > 0 {
 						brt.asyncDestinationStruct[destinationID].UploadInProgress = true