import { flags , SfdxCommand } from "@salesforce/command" ; import { Messages , SfdxError } from "@salesforce/core" ; import { AnyJson } from "@salesforce/ts-types" ; import * as fs from "fs" ; import * as path from "path" ; import { Record , RecordResult , SuccessResult , ErrorResult , Connection } from 'jsforce' ; const util = require ( "util" ) ; const csv = require ( "csvtojson" ) ; Messages . importMessagesDirectory ( __dirname ) ; const messages = Messages . loadMessages ( "texei-sfdx-plugin" , "data-import" ) ; let conn ; let recordIdsMap ; let lookupOverrideMap ; let queriedLookupOverrideRecords ; let remainingDataFiles ; let isVerbose = false ; interface ErrorResultDetail { statusCode : string ; message : string ; fields : string [ ] ; } export default class Import extends SfdxCommand { public static description = messages . getMessage ( "commandDescription" ) ; public static examples = [ `$ sfdx texei:data:import --inputdir ./data --targetusername texei-scratch
  Data imported!
  ` ] ; protected static flagsConfig = { inputdir : flags . string ( { char : "d" , description : messages . getMessage ( "inputFlagDescription" ) , required : true } ) , allornone : flags . boolean ( { char : "a" , description : messages . getMessage ( "allOrNoneFlagDescription" ) , required : false } ) , verbose : flags . builtin ( { description : messages . getMessage ( 'verbose' ) , } ) } ; protected static requiresUsername = true ; protected static requiresDevhubUsername = false ; protected static requiresProject = false ; public async run ( ) { conn = await this . org . getConnection ( ) ; recordIdsMap = new Map < string , string > ( ) ; isVerbose = this . flags . verbose ; const scratchOrgUserId = ( ( await conn . query ( `Select Id from User where username = '${ this . org . getUsername ( ) }'` ) ) . records [ 0 ] as any ) . Id ; recordIdsMap . set ( "SfdxOrgUser" , scratchOrgUserId ) ; const filesPath = path . join ( process . cwd ( ) , this . flags . inputdir ) ; const readDir = util . promisify ( fs . readdir ) ; let dataFiles = ( await readDir ( filesPath , "utf8" ) ) . filter ( f => { return ! isNaN ( f . substr ( 0 , f . indexOf ( '-' ) ) ) ; } ) . sort ( function ( a , b ) { return a . substr ( 0 , a . indexOf ( '-' ) ) - b . substr ( 0 , b . indexOf ( '-' ) ) ; } ) ; remainingDataFiles = new Set ( dataFiles ) ; for ( let i = 0 ; i < dataFiles . length ; i ++ ) { const dataFile = dataFiles [ i ] ; if ( ! isNaN ( dataFile . substring ( 0 , 1 ) ) ) { const objectName = await this . getObjectNameFromFile ( dataFile ) ; this . ux . startSpinner ( `Importing ${ dataFile }` , null , { stdout : true } ) ; const objectData = ( await this . readFile ( dataFile ) ) ; const externalIdField = objectData ?. attributes ?. externalId ; const objectRecords = objectData . records ; await this . prepareDataForInsert ( objectName , objectRecords ) ; await this . upsertData ( objectRecords , objectName , externalIdField ) ; this . ux . stopSpinner ( `Done.` ) ; } remainingDataFiles . delete ( dataFile ) ; } return { message : "Data imported" } ; } private async prepareDataForInsert ( sobjectName , jsonData ) { const lookups = await this . getLookupsForObject ( sobjectName ) ; let recTypeInfos = new Map < string , string > ( ) ; recTypeInfos = await this . getRecordTypeMap ( sobjectName ) ; let standardPriceBookId = '' ; if ( sobjectName === 'PricebookEntry' ) { standardPriceBookId = ( ( await conn . query ( 'Select Id from Pricebook2 where IsStandard = true' ) ) . records [ 0 ] as any ) . Id ; } for ( const sobject of jsonData ) { for ( const lookup of lookups ) { if ( sobject [ lookup . name ] && ! ( sobjectName === 'PricebookEntry' && sobject . Pricebook2Id === 'StandardPriceBook' && lookup . name === 'Pricebook2Id' ) ) { sobject [ lookup . name ] = recordIdsMap . get ( sobject [ lookup . name ] ) ; } if ( sobject [ lookup . relationshipName ] ) { this . debug ( `### Overridden lookup found: ${ lookup . relationshipName }:` ) ; this . debug ( `Reference to: ${ lookup . referenceTo }` ) ; this . debug ( `fields to query: ${ JSON . stringify ( sobject [ lookup . relationshipName ] ) }` ) ; if ( lookupOverrideMap === undefined ) { lookupOverrideMap = await this . createLookupOverrideMap ( ) ; if ( lookupOverrideMap === undefined ) { throw new SfdxError ( `A lookupOverride is specified in your data plan for sObject ${ lookup . referenceTo }, but no field is associated to it. Either remove the lookupOverride for this sObject or add field(s).` ) ; } queriedLookupOverrideRecords = new Map < string , Record [ ] > ( ) ; } const sObjectName = sobject [ lookup . relationshipName ] . attributes . type ; let sObjectRecords ; if ( queriedLookupOverrideRecords [ sObjectName ] ) { sObjectRecords = queriedLookupOverrideRecords [ sObjectName ] ; } else { sObjectRecords = await this . getObjectRecords ( sObjectName , lookupOverrideMap [ sObjectName ] ) ; queriedLookupOverrideRecords [ sObjectName ] = sObjectRecords ; } let filterList = Object . assign ( { } , sobject [ lookup . relationshipName ] ) ; delete filterList . attributes ; const start = Date . now ( ) ; this . debug ( `Searching for record for lookup override` ) ; const foundRecord = sObjectRecords . find ( element => { for ( const [ key , value ] of Object . entries ( filterList ) ) { this . debug ( `Looking for value ${ value } for field ${ key }` ) ; this . debug ( `Value for current record: ${ element [ key ] }` ) ; if ( element [ key ] != value ) { return ; } } return element ; } ) ; const duration = Date . now ( ) - start ; this . debug ( `Finished searching record in ${ duration } milliseconds` ) ; if ( foundRecord === undefined ) { throw new SfdxError ( `No ${ lookup . referenceTo } record found for filter ${ Object . entries ( filterList ) }` ) ; } const queriedRecordId = foundRecord . Id ; if ( isVerbose ) { this . ux . log ( `RecordId found (${ queriedRecordId }) for sObject ${ sObjectName } and filter ${ Object . entries ( filterList ) }` ) ; } delete sobject [ lookup . relationshipName ] ; sobject [ lookup . name ] = queriedRecordId ; } } if ( recTypeInfos . size > 0 ) { sobject . RecordTypeId = recTypeInfos . get ( sobject . RecordTypeId ) ; } if ( sobjectName === 'PricebookEntry' && sobject . Pricebook2Id === 'StandardPriceBook' ) { sobject . Pricebook2Id = standardPriceBookId ; } if ( recordIdsMap . get ( sobject . attributes . referenceId ) ) { sobject . Id = recordIdsMap . get ( sobject . attributes . referenceId ) ; } } } private async upsertData ( records , sobjectName , externalIdField ) { let sobjectsResult = new Array < RecordResult > ( ) ; if ( externalIdField ) { this . debug ( `DEBUG upserting ${ sobjectName } records using external id field '${ externalIdField }'` ) ; const maxParallelUpsertRequests = 10 ; for ( var i = 0 ; i < records . length ; i += maxParallelUpsertRequests ) { const chunkResults = await conn . sobject ( sobjectName ) . upsert ( records . slice ( i , i + maxParallelUpsertRequests ) , externalIdField , { allowRecursive : true , allOrNone : this . flags . allornone } ) . catch ( ( err ) => { throw new SfdxError ( `Error upserting records: ${ err }` ) ; } ) ; sobjectsResult . push ( ... chunkResults ) ; } } else if ( records [ 0 ] && records [ 0 ] . Id ) { this . debug ( `DEBUG updating ${ sobjectName } records` ) ; sobjectsResult = await conn . sobject ( sobjectName ) . update ( records , { allowRecursive : true , allOrNone : this . flags . allornone } ) . catch ( err => { throw new SfdxError ( `Error importing records: ${ err }` ) ; } ) ; } else { this . debug ( `DEBUG inserting ${ sobjectName } records` ) ; sobjectsResult = await conn . sobject ( sobjectName ) . insert ( records , { allowRecursive : true , allOrNone : this . flags . allornone } ) . catch ( err => { throw new SfdxError ( `Error importing records: ${ err }` ) ; } ) ; } for ( let i = 0 ; i < sobjectsResult . length ; i ++ ) { if ( ! sobjectsResult [ i ] . success ) { const res = sobjectsResult [ i ] as ErrorResult ; const errors = res . errors [ 0 ] as any ; if ( externalIdField && errors . statusCode === 'METHOD_NOT_ALLOWED' ) { throw new SfdxError ( `Unable to upsert records. Make sure you're not importing records where ${ externalIdField } External Id field is missing.` ) ; } else if ( errors . statusCode !== 'ALL_OR_NONE_OPERATION_ROLLED_BACK' ) { throw new SfdxError ( `Error importing record ${ records [ i ] . attributes . referenceId }: ${ errors . statusCode }-${ errors . message }${ errors . fields ?. length > 0 ? '(' + errors . fields + ')' : '' }` ) ; } } } this . updateMapIdRef ( records , sobjectsResult , recordIdsMap ) ; } private async readFile ( fileName ) { let filePath = fileName ; if ( this . flags . inputdir ) { filePath = path . join ( this . flags . inputdir , fileName ) ; } filePath = path . join ( process . cwd ( ) , filePath ) ; const readFile = util . promisify ( fs . readFile ) ; return JSON . parse ( await readFile ( filePath , "utf8" ) ) ; } private async getRecordTypeMap ( sobjectName ) { let recTypesMap = new Map ( ) ; const conn = this . org . getConnection ( ) ; const recTypeResults = ( await conn . query ( `SELECT Id, DeveloperName FROM RecordType WHERE SobjectType = '${ sobjectName }'` ) ) . records as any ; for ( const recType of recTypeResults ) { recTypesMap . set ( recType . DeveloperName , recType . Id ) ; } return recTypesMap ; } private async updateMapIdRef ( inputRecords , inputResults , recordIdsMap ) { let index = 0 ; for ( let input of inputResults ) { input = input as SuccessResult ; recordIdsMap . set ( inputRecords [ index ] . attributes . referenceId , input . id ) ; index ++ ; } } private async getObjectNameFromFile ( filePath ) { if ( filePath . indexOf ( "-" ) === - 1 || filePath . indexOf ( ".json" ) === - 1 ) { throw new SfdxError ( `Invalid file name: ${ filePath }` ) ; } let fileName = '' ; fileName = filePath . substring ( filePath . indexOf ( "-" ) + 1 ) . replace ( ".json" , "" ) ; if ( fileName . indexOf ( "-" ) > 0 ) { fileName = fileName . substring ( 0 , fileName . indexOf ( "-" ) ) ; } return fileName ; } private async getLookupsForObject ( objectName ) { let lookups = [ ] ; const describeResult = await conn . sobject ( objectName ) . describe ( ) ; for ( const field of describeResult . fields ) { if ( field . createable && field . referenceTo && field . referenceTo . length > 0 && field . name != "OwnerId" && field . name != "RecordTypeId" ) { lookups . push ( field ) ; } } return lookups ; } private async createLookupOverrideMap ( ) { let overrideMap = new Map < string , Set < string > > ( ) ; for ( const dataFile of remainingDataFiles ) { if ( ! isNaN ( dataFile . substring ( 0 , 1 ) as any ) ) { const objectRecords = ( await this . readFile ( dataFile ) ) . records ; if ( objectRecords . length > 0 ) { const currentRec = objectRecords [ 0 ] ; for ( const [ key ] of Object . entries ( currentRec ) ) { if ( currentRec [ key ] ?. attributes ) { const sobjectName = currentRec [ key ] ?. attributes . type ; let fieldsToQuery = new Set < string > ( ) ; for ( const [ objectkey ] of Object . entries ( currentRec [ key ] ) ) { if ( objectkey !== 'attributes' ) { fieldsToQuery . add ( objectkey ) ; } } if ( overrideMap [ sobjectName ] ) { fieldsToQuery . forEach ( item => overrideMap [ sobjectName ] . add ( item ) ) ; } else { overrideMap [ sobjectName ] = new Set < string > ( ) ; fieldsToQuery . forEach ( item => overrideMap [ sobjectName ] . add ( item ) ) ; } } } } } } return overrideMap ; } private async getObjectRecords ( sObjectName , fieldsToQuery ) { const bulkQuery = async ( sObjectName , fieldsToQuery ) => new Promise < Array < Record > > ( async ( resolve , reject ) => { let retrievedRecords = new Array < Record > ( ) ; fieldsToQuery . add ( 'Id' ) ; conn . bulk . pollTimeout = 250000 ; this . debug ( `Querying ${ sObjectName } for lookup override` ) ; const recordStream = conn . bulk . query ( `Select ${ Array . from ( fieldsToQuery ) . join ( ',' ) } from ${ sObjectName }` ) ; const readStream = recordStream . stream ( ) ; const csvToJsonParser = csv ( { flatKeys : false , checkType : true } ) ; readStream . pipe ( csvToJsonParser ) ; csvToJsonParser . on ( "data" , ( data ) => { retrievedRecords . push ( JSON . parse ( data . toString ( 'utf8' ) ) ) ; } ) ; recordStream . on ( "error" , ( error ) => { reject ( error ) ; } ) ; csvToJsonParser . on ( "error" , ( error ) => { reject ( error ) ; } ) ; csvToJsonParser . on ( "done" , async ( ) => { resolve ( retrievedRecords ) ; } ) ; } ) ; return await bulkQuery ( sObjectName , fieldsToQuery ) ; } }