2025-11-24 ImportData Document

Process ImportData

ProcessCsvFile

1. Receive bucketName and fieldName.
2. Define the range e.g. range = "bytes=0-5000".
3. Fetch data from S3 with:

    {
  		Bucket: 'izara-importdata',
  		Key: 'csv/xxxx.csv',
  		Range: 'bytes=0-5000'
	}

1. Take the importBatchId from the S3 metadata and use it to get ImportBatchMain to retrieve importConfigId.
2. Use the retrieved importConfigId to get CsvImportConfig.
3. Use the CsvImportConfig as the main configuration for processing.
4. Loop through the S3 data to extract the titleRow, collecting all field names.
5. After extracting all title rows, update fieldNames in ImportBatchMain.
6. If S3 data is not fully read, continue with pagination to fetch the next portion.
7. When the file ends, send importBatchId to ProcessRowData.

ProcessRowData

1. Receive importBatchId.
2. Use the received importBatchId to get ImportBatchMain and extract importConfigId.
3. Use the retrieved importConfigId to get CsvImportConfig.
4. Fetch data from S3 with:

    {
  		Bucket: 'izara-importdata',
  		Key: 'csv/xxxx.csv',
  		Range: 'bytes=0-5000'
	}

1. Loop through the fetched data and store each row into dataRow, including pendingObjects and tempRelationship.
2. Validate Schema, Validate Type, Check Required fields, and Check Action (create/update/reference/error).
3. Check ImportUserReference to see if this object was already created; if yes, change Action from create → reference.
4. Store PendingObjectMain, PendingObjectReference, PendingLink, and FloatingRelationships.
5. If the file is not finished, re-invoke itself with the new startByte:
   • If not timed out → continue looping.
   • If timed out → perform pagination.
6. If the file is finished, send importBatchId to ProcessFloatingRelationshipsHdrDsq.

ProcessFloatingRelationshipsHdrDsq

1. Query FloatingRelationships using importBatchId.
2. Process based on action type (create or update):
   • create
     • Create PendingLink for relationships where action = create, and update FloatingStatus: "complete".
   • update
     • Update the existing relationship using data from object A/B.
     • Save AwaitingStep.
     • Call sendMessageToExternalForUpdateRelationship, sending: firstObject, secondObject, relType, relId, relationshipProperties, relationshipDirection.
3. If not finished, continue with pagination.
4. If fully processed, send importBatchId to ProcessPendingLinkHdrDsq.

ProcessPendingLink

1. Query PendingLink using importBatchId.
2. Loop to process each PendingLink.
3. Run validateLinkConfig to check RequiredOnCreate.
4. If no RequiredOnCreate, create AwaitingMultipleSteps:

await asyncFlowSharedLib.createAwaitingMultipleSteps(
 _izContext,
  [
    asyncFlowSharedLib.createAwaitingStepId(pendingObjectMainOfObjectA.pendingObjectId, consts.PREFIX_CREATE_PENDINGLINK), // object A
    asyncFlowSharedLib.createAwaitingStepId(pendingObjectMainOfObjectB.pendingObjectId, consts.PREFIX_CREATE_PENDINGLINK) // object B
  ],
    asyncFlowSharedLib.createPendingStepId(queryPendingLink.pendingLinkId, consts.PREFIX_CREATE_PENDINGLINK) // pendingLink
 );

1. If RequiredOnCreate exists, create two AwaitingMultipleSteps:

   • await asyncFlowSharedLib.createAwaitingMultipleSteps(
      _izContext,
       [asyncFlowSharedLib.createAwaitingStepId(objectRequiredOnCreate.pendingObjectId, consts.PREFIX_IMPORTBATCHMAIN)],
       asyncFlowSharedLib.createPendingStepId(importBatchId, consts.PREFIX_IMPORTBATCHMAIN)
      );
     

   • await asyncFlowSharedLib.createAwaitingMultipleSteps(
      _izContext,
       [asyncFlowSharedLib.createAwaitingStepId(otherPendingObject.pendingObjectId, consts.PREFIX_CREATE_PENDINGOBJECT)],
       asyncFlowSharedLib.createPendingStepId(objectRequiredOnCreate.pendingObjectId, consts.PREFIX_CREATE_PENDINGOBJECT)
      );
     

2. Update the PendingObjectMain of the RequiredOnCreate object:

       {
     		"attributeName": "requireOnCreateLinkIds",
    		"value": [pendingLinkId],
     		"action": "SET",
    		"listAppend": true
   	}
   

3. If more data remains, continue pagination.
4. If all processed, send importBatchId to ProcessPendingObjectHdrDsq.

ProcessPendingObjectHdrDsq

1. Query PendingObjectMain using importBatchId.
2. Loop through each PendingObjectMain.
3. If the object has RequiredOnCreate, do not send to ExternalService.
4. If it has no RequiredOnCreate:
   • Create AwaitingMultipleSteps using PREFIX_IMPORTBATCHMAIN + pendingObjectId.
   • Send to ExternalService according to Action (create, get, update), including:
     • callingFlow = ExternalActionCompleteHdrSqs
     • callingFlowProperties = PREFIX_IMPORTBATCHMAIN + pendingObjectId

ExternalActionComplete

1. Receive objType, objInstanceFull, errorFound, passBackProperty from ExternalService.
2. Extract pendingObjectId from passBackProperty, locate AwaitingMultipleSteps, and remove it.
3. If errorFound, run recursiveErrorPendingObject.
4. If no error, update status to "complete".
5. Create pendingObjectAwaitingId to find objects that were waiting for this one.
6. If there are objects waiting:
   • If they still wait for others → remove their AwaitingMultipleSteps.
   • If not → execute the RequiredOnCreate object and create its relationship.
7. Check if all objects are completed. If yes → remove awaitingStep of ImportBatchMain and call clearAllAwaitingSteps.
8. Return { returnValue: "externalActionComplete finish" }.

CreateLinkExternalComplete

1. Receive firstObject, secondObject, relType, errorFounds, passBackProperties from ExternalService.
2. Extract pendingLinkId from passBackProperties, find AwaitingMultipleSteps, and remove it.
3. If there are errorFounds → update PendingLink as error; otherwise → mark it complete.
4. Remove AwaitingMultipleSteps and call clearAllAwaitingSteps.