--[[----------------------------------------------------------------------------

LrClassicImport.lua

--------------------------------------------------------------------------------

ADOBE CONFIDENTIAL
------------------
Copyright 2023 Adobe
All Rights Reserved.

NOTICE: All information contained herein is, and remains
the property of Adobe and its suppliers, if any. The intellectual
and technical concepts contained herein are proprietary to Adobe
and its suppliers and are protected by all applicable intellectual
property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from Adobe.

----------------------------------------------------------------------------]]--

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable

	-- WF.core.os
local KSDate = require 'KSDate'
local WFDeviceInfo = require 'WFDeviceInfo'

	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.core.performance
local WFPerformanceReport = require 'WFPerformanceReport'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFCatImportConstants = require 'WFCatImportConstants'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCrypto = require 'WFCrypto'
local WFConfiguration = require 'WFConfiguration'
local WFTableMerge = require 'WFTableMerge'
local _ = require 'WFArguments_uuid'
local WFRxThrottledBatchedBulkApplyOverlay = require 'WFRxThrottledBatchedBulkApplyOverlay'
local WFCapabilities = require 'WFCapabilities'

	-- WF.oz_client.managed_catalog
local WFCatAssetAux = require 'WFCatAssetAux'
local WFXMLNS = require 'WFXMLNS'
local WFCatGPSFormatUtils = require 'WFCatGPSFormatUtils'
local WFImageOrientation = require 'WFImageOrientation'
local WFCatImportMetadata = require 'WFCatImportMetadata'
local WFCatImportManifestMergeAssetUtils = require 'WFCatImportManifestMergeAssetUtils'

	-- WF.imaging.metadata
local WFXmp = require 'WFXmp'
local LrClassicDevelopSettings = require "LrClassicDevelopSettings"

	-- WF.oz_client.catalog
local WFOzContentTypes = require 'WFOzContentTypes'
local WFFileUtils = require 'WFFileUtils'
local WFCatUtils = require 'WFCatUtils'

--------------------------------------------------------------------------------

local kDocStoreBatchSize = WFConfiguration.Import.BatchSize or 10
local kNumParallelImports = WFConfiguration.Import.NumImagesProcessedInParallel or 100

--------------------------------------------------------------------------------

	-- Merge schema for merging synced assets
local kMergeTypes = WFCatImportManifestMergeAssetUtils.kMergeTypes
local kUserIdPlaceHolder = WFCatImportManifestMergeAssetUtils.kUserIdPlaceHolder
local kMergeSchema = {
	xmp = {
		dc = {
			description = kMergeTypes.override,
			title = kMergeTypes.override,
			rights = kMergeTypes.override,
			subject = kMergeTypes.merge,
		}
	},
	develop = kMergeTypes.latest,
	_localOnly = kMergeTypes.ignore,
	
	-- [ NMB-5046 ] Add to importSource only if it is not present,
	-- except for the lrcat details.
	importSource = {
		[ WFCatImportManifestMergeAssetUtils.kKeyDefaultMergeType ] = kMergeTypes.add,
		lrcatStoreProviderId = kMergeTypes.override,
		lrcatImageId = kMergeTypes.override,
		sha256 = kMergeTypes.override,
		fileSize = kMergeTypes.override,
	},

}
kMergeSchema.ratings = {}
kMergeSchema.ratings[ kUserIdPlaceHolder ] = kMergeTypes.latest
kMergeSchema.reviews = {}
kMergeSchema.reviews[ kUserIdPlaceHolder ] = kMergeTypes.latest

--==============================================================================

local check = WFArguments.check

local derefOpt = KSTableUtils.derefOpt

local logger = WFLogger 'LrClassicImport'
--	:setLevel 'debug'

--------------------------------------------------------------------------------

local LrClassicImport = {}

--------------------------------------------------------------------------------

local function isNonEmpty( maybeString )

	return type( maybeString ) == "string"

end

--------------------------------------------------------------------------------

local function getDate( metadata, namespace, key )
	local dateString = metadata:get( namespace, key )
	
	return KSDate.maybeKSDateFromString( dateString )
end

--------------------------------------------------------------------------------

local function maybeSetNestedFieldToNil( xmp, ... )

	local maybeValue = derefOpt( xmp, ... )

	if maybeValue then
		KSTableUtils.setNested( xmp, nil, ... )
	end

	return xmp
end

--------------------------------------------------------------------------------

local function getCaptureDate( metadata, fileCreationDate )
	local captureDate = getDate( metadata, WFXMLNS.exif, "DateTimeOriginal" )
						or getDate( metadata, WFXMLNS.photoshop, "DateCreated" )
						or getDate( metadata, WFXMLNS.exif, "DateTimeDigitized" )
						or getDate( metadata, WFXMLNS.xap, "CreateDate" )
						or getDate( metadata, WFXMLNS.xap, "ModifyDate" )
						or getDate( metadata, WFXMLNS.dm, "shotDate" )
						or fileCreationDate

	if not captureDate then
		-- if there is no capture date, the Oz data model expects 
		-- the 'zero' date: 0000-00-00T00:00:00
		captureDate = "0000-00-00T00:00:00" 
	end
	
	return captureDate
end

--------------------------------------------------------------------------------

local function getOrientation( metadata )
	return tonumber( metadata:get( WFXMLNS.tiff, "Orientation") or 1 )
end

--------------------------------------------------------------------------------

local function getOrientedDimensions( tiffOrientation, width, height )
	
	if width and height and WFImageOrientation.swapDimensions( tiffOrientation ) then
		return height, width
	else 
		return width, height
	end
end

--------------------------------------------------------------------------------

local function getOriginalCrop( orientation, additionalInfo )
	
	if additionalInfo and additionalInfo.defaultCropRight and additionalInfo.defaultCropLeft and
		additionalInfo.defaultCropTop and additionalInfo.defaultCropBottom then
		
		return {
			top = additionalInfo.defaultCropTop,
			bottom = additionalInfo.defaultCropBottom,
			left = additionalInfo.defaultCropLeft,
			right = additionalInfo.defaultCropRight,
			orientation = orientation or 1
		}
		
	end

end

--------------------------------------------------------------------------------

local function getAltitiudeFromMetadata( metadata )

	local altitudeRatio = metadata:get( WFXMLNS.exif, "GPSAltitude" )
	local altitude
	if altitudeRatio then
		local _, _, num, den = string.find( altitudeRatio, "(%--%d+)/(%--%d+)" )
		num = num and tonumber( num )
		den = den and tonumber( den )
		if num and den and den > 0 then
			altitude = num / den
		end
		local belowSeaLevel = tonumber( metadata:get( WFXMLNS.exif, "GPSAltitudeRef" ) ) == 1

		if altitude then
			if altitude > 0 and belowSeaLevel then
				altitude = altitude * -1
			end
			altitude = tonumber( string.format( "%.1f", altitude ) )
		end
	end

	return altitude
end

--------------------------------------------------------------------------------

local function convertLocation( metadata, locationInfo )
	local location = {}

	local longitude = WFCatGPSFormatUtils.GPSCoordStringToNumber( metadata:get( WFXMLNS.exif, "GPSLongitude" ), 180)
	local latitude = WFCatGPSFormatUtils.GPSCoordStringToNumber( metadata:get( WFXMLNS.exif, "GPSLatitude" ), 90 )
	
	if longitude and latitude then
		location.longitude = longitude
		location.latitude = latitude
	end
	
	local altitude = getAltitiudeFromMetadata( metadata )
	if altitude then
		location.altitude = altitude
	end

	if locationInfo then
		location.city = locationInfo.city
		location.state = locationInfo.state
		location.country = locationInfo.country
	end

	if not next( location ) then
		-- empty location table, return nil
		return nil
	end

	return location

end

--------------------------------------------------------------------------------

local function getLocationChange( metadata, docValue )
	local location = {}

	local altitude = getAltitiudeFromMetadata( metadata )

	local mayBeAltitude = derefOpt( docValue, "location", "altitude" )
	if altitude ~= mayBeAltitude then
		if altitude then
			KSTableUtils.setNested( location, altitude, 'altitude')
		else
			KSTableUtils.setNested( location, { _delete = true }, 'altitude')
		end
	end

	local lng = WFCatGPSFormatUtils.GPSCoordStringToNumber( metadata:get( WFXMLNS.exif, "GPSLongitude" ), 180)
	local lat = WFCatGPSFormatUtils.GPSCoordStringToNumber( metadata:get( WFXMLNS.exif, "GPSLatitude" ), 90 )

	local maybeLatitude = derefOpt( docValue, "location", "latitude" )
	local maybeLongitude = derefOpt( docValue, "location", "longitude" )

	if lng ~= maybeLongitude or lat ~= maybeLatitude then
		if lat and lng then
			KSTableUtils.setNested( location, lng, 'longitude')
			KSTableUtils.setNested( location, lat, 'latitude')
			KSTableUtils.setNested( location, { _delete = true }, 'city')
			KSTableUtils.setNested( location, { _delete = true }, 'state')
			KSTableUtils.setNested( location, { _delete = true }, 'country')
			KSTableUtils.setNested( location, { _delete = true }, 'isoCountryCode')
			KSTableUtils.setNested( location, { _delete = true }, 'sublocation')
		else
			KSTableUtils.setNested( location, { _delete = true }, 'longitude')
			KSTableUtils.setNested( location, { _delete = true }, 'latitude')
		end
	end

	if not next( location ) then
		-- empty location table, return nil
		return nil
	end

	return location
end

--------------------------------------------------------------------------------

local function convertXmpPhotoshop( importInfo, xmp )

	--Do not sync manually entered location fields from classic

	xmp = maybeSetNestedFieldToNil( xmp, "photoshop", "City" )

	xmp = maybeSetNestedFieldToNil( xmp, "photoshop", "State" )

	xmp = maybeSetNestedFieldToNil( xmp, "photoshop", "Country" )

	return xmp
end
--------------------------------------------------------------------------------


local function convertRating( userId, ratingData )

	if userId and ratingData and type(ratingData) == "table" and next(ratingData) ~= nil and
		( 0 <= ratingData.rating ) and
		( ratingData.rating <= 5 ) then
		return {
			[ userId ] = {
				rating = math.floor( ratingData.rating ),
				date = ratingData.dateChanged,
			}
		}
	end

end

--------------------------------------------------------------------------------

local function convertReviews( userId, reviewData )

	if userId and reviewData and type(reviewData) == "table" and next(reviewData) ~= nil then
		return {
			[ userId ] = {
				flag = reviewData.flag,
				date = reviewData.dateChanged,
			}
		}
	end

end

--------------------------------------------------------------------------------

local function convertXmpDc( importInfo, xmp )

	if isNonEmpty( importInfo.copyright ) then
		KSTableUtils.setNested( xmp, importInfo.copyright, 'dc', 'rights')
	end

	if isNonEmpty( importInfo.title ) then
		KSTableUtils.setNested( xmp, importInfo.title, 'dc', 'title')
	end

	if isNonEmpty( importInfo.caption ) then
		KSTableUtils.setNested( xmp, importInfo.caption, 'dc', 'description')
	end

	if isNonEmpty( importInfo.creator ) then
		KSTableUtils.setNested( xmp, importInfo.creator, 'dc', 'creator')
	end

	--LRD-4216689: Do not sync keywords from classic
	xmp = maybeSetNestedFieldToNil( xmp, 'dc', 'subject' )

	return xmp
end

--------------------------------------------------------------------------------

local function maybeGetMergedAssets( importInfo )
	if importInfo.mergeSourceImageIds and next( importInfo.mergeSourceImageIds ) then
		return importInfo.mergeSourceImageIds
	end
end

--------------------------------------------------------------------------------

-- Utility function for rounding cropped width/cropped height, which are always
-- positive.

local function roundPositive( x )
	return math.floor( x + 0.5 )
end

--------------------------------------------------------------------------------

local function calculateCroppedDimensions( width, height, left, right, top, bottom )
	
	local x = right - left
	local y = bottom - top
	
	-- Use 'round', rather than math.floor, as x and/or y are likely to be only
	-- slightly erroneous, if at all.
	local cropWidth = roundPositive( x * width )
	local cropHeight = roundPositive( y * height )
	
	local minDim = math.min( cropWidth, cropHeight )
	-- Oz doesn't allows sizes less than 1, if we find an invalid size, just return
	-- width, height instead
	if minDim < 1 then
		return width, height
	else
		return cropWidth, cropHeight 
	end
		
end

--------------------------------------------------------------------------------

local function maybeAddc2paIngredients( wfCatalog, importInfo, assetDocValue )

	local digest = derefOpt( importInfo, "c2paIngredients", "digest" )
	local state = derefOpt( importInfo, "c2paIngredients", "state" )
	local version = derefOpt( importInfo, "c2paIngredients", "version" )

	if digest and state and version then
	
		return WFCatAssetAux.getC2paOverlay( wfCatalog, derefOpt( importInfo, "c2paIngredients") )
				:flatMap( function( overlay )
					WFTableMerge.applyOverlay( assetDocValue, overlay )
					return Observable.just( assetDocValue )

				end ) 
	end
	
	return Observable.just( assetDocValue )

end

--------------------------------------------------------------------------------

function LrClassicImport.DEBUG_maybeAddc2paIngredients( wfCatalog, importInfo, assetDocValue )
	return maybeAddc2paIngredients( wfCatalog, importInfo, assetDocValue )
end

--------------------------------------------------------------------------------

local function addDevelopSettings( wfCatalog, imageId, importInfo, orientation, assetDocValue, devSettingTime, crVersion )
	
	local addlInfo = importInfo.additionalInfo
	
	local croppedWidth, croppedHeight
	
	if importInfo.croppedWidth and importInfo.croppedHeight then
		
		-- pre-existing ACR style crop
		croppedWidth, croppedHeight = importInfo.croppedWidth, importInfo.croppedHeight
		
	elseif addlInfo and addlInfo.defaultCropLeft and addlInfo.defaultCropRight 
		and addlInfo.defaultCropTop and addlInfo.defaultCropBottom then
		
		-- in camera crop
		croppedWidth, croppedHeight = calculateCroppedDimensions( importInfo.width, importInfo.height, 
															addlInfo.defaultCropLeft, addlInfo.defaultCropRight, 
															addlInfo.defaultCropTop, addlInfo.defaultCropBottom )
		
	else
		
		-- no crop
		croppedWidth, croppedHeight = importInfo.width, importInfo.height
		
	end
	
	local orientedCroppedWidth, orientedCroppedHeight = getOrientedDimensions( orientation, croppedWidth, croppedHeight )
	
	if importInfo.isVideo then 

		local temporal = importInfo.temporal
		
		return LrClassicDevelopSettings.developVideoSettingsOverlay( 
						imageId, temporal, orientedCroppedWidth,
						orientedCroppedHeight,
						importInfo.settingsVideoFoundation, devSettingTime )
			:map( function( developSettingsOverlay )
				return WFTableMerge.applyOverlay( assetDocValue, developSettingsOverlay )
				end )
		
	end
	
	local assetDocValueObs = LrClassicDevelopSettings.developSettingsOverlay( wfCatalog, imageId, 
								orientation, orientedCroppedWidth, orientedCroppedHeight, importInfo.xmpCameraRaw, devSettingTime, 
								importInfo.auxInfo, importInfo.needsUpdateAISettings,
								crVersion, importInfo.revId )
		:map( function( developSettingsOverlay )
			return WFTableMerge.applyOverlay( assetDocValue, developSettingsOverlay )
		end )
	
	-- get aux overlay

	local auxDataList =  importInfo.auxInfo or {}

	local auxOverlayObs = WFCatAssetAux.getAuxOverlayForAuxInfo( wfCatalog, auxDataList )

		:catch( function( errMessage, optDigest, optFilePath )

			if errMessage == "AuxNotFound" then

				local digest = optDigest
				local filePath = optFilePath

				return Observable.throw( {
					failure = "missingAux",
					message = "Auxiliary binaries are missing",
					Digest = digest,
					FilePath = filePath,
				})

			end

			return Observable.throw( errMessage )

		end )

	return Observable.combineLatest( 
			assetDocValueObs,
			auxOverlayObs )

		:map( function( assetDocValue, auxOverlay )

			local assetDocValueWithAux
					= WFTableMerge.applyOverlay( assetDocValue, auxOverlay )

			return assetDocValueWithAux

		end )

end

--------------------------------------------------------------------------------

-- Generate original or proxy sha256
local function genShaObservable( importInfo, rendition )
	
	local providedSha256

	if rendition == 'master' then
		providedSha256 = importInfo.sha256
	elseif rendition == 'proxy' then
		providedSha256 = importInfo.proxySha256
	end
	
	local shaObservable
	
	if providedSha256 == nil then
	
		local filePath
		if rendition == 'master' then
			filePath = importInfo.filePath
		elseif rendition == 'proxy' then
			filePath = importInfo.proxyFilePath
		end
		
		if not filePath or not WFFileUtils.exists( filePath ) then
			shaObservable = Observable.just( nil )
		else
			shaObservable = WFPerformanceReport.wrapObservable(
				'WFCatImport: digestFileSHA256',
				WFCrypto.asyncDigestFile( filePath, 'SHA256' )
			)
		end
	
	elseif providedSha256 == "dont-generate" then
	
		shaObservable = Observable.just( nil )
		
	else
	
		shaObservable = Observable.just( providedSha256 )
		
	end

	return shaObservable
	
end

--------------------------------------------------------------------------------

local function applyFilesOverlay( asset, renditionName, filePath, size, sha256 )
	
	local fileInfo
			
	if filePath then

		local volumeGuid = WFFileUtils.getVolumeGuidForPath( filePath )
		local relativePath = WFFileUtils.removeVolumeFromPath( filePath )
		
		if renditionName == 'master' then
		
			fileInfo = {
				original = {
					master = {
						relativePath = relativePath,
						size = size,
						sha256 = sha256,
						volumeGuid = volumeGuid
					}
				}
			}
			
		elseif renditionName == 'proxy' then
			
			fileInfo = {
				proxy = {
					[ '2560' ] = {
						relativePath = relativePath,
						size = size,
						sha256 = sha256,
						volumeGuid = volumeGuid
					}
				}
			}
			
		end

	end

	return fileInfo

end

--------------------------------------------------------------------------------

local function getSyncedAsset( catalog, importInfo )
	
	if not importInfo.ozAssetId then
		-- No corresponding synced asset.
		return Observable.just( nil )
	end
	
	return catalog:docStore():docValue( importInfo.ozAssetId )
		:flatMap( function( docValue )

			if docValue._empty or docValue.type ~= 'asset' then
				return Observable.just( nil )
			end

			if docValue._deleted then

				return Observable.throw( {
					failure = "errorAlreadyDeletedOnOz",
					message = "errorAlreadyDeletedOnOz",
					classicLocalId = derefOpt( importInfo, 'importSource', 'lrcatImageId' ),
					assetId = importInfo.ozAssetId,
				} )

			end
			
			return Observable.just( docValue )

		end )
end

--------------------------------------------------------------------------------

local function appendSyncedDocObservable( catalog, importInfo, assetDocValue )

	return getSyncedAsset( catalog, importInfo )
		:map( function( maybeSyncedDoc )

			return assetDocValue, maybeSyncedDoc

		end )
end

--------------------------------------------------------------------------------

local function handleMergeOrPassOnObv( ozUserId, ozAssetId, importTimestamp, assetDocValue, syncedDocValue )
	-- Receives the assembled docValue and the syncedDocValue
	-- Merge both

	local overlayDoc
	local isAssetUpdate = not not syncedDocValue
	
	if syncedDocValue then	
		
		-- SyncedDocValue is present. We will have to merge the doc.
		local mergedDoc = WFCatImportManifestMergeAssetUtils.mergeDoc( ozUserId, kMergeSchema, assetDocValue, syncedDocValue )
		overlayDoc = WFTableMerge.createOverlay( syncedDocValue, mergedDoc )
		overlayDoc._id = ozAssetId -- createOverlay wont add the id. Add it explicitly
		
		
		-- NMB-5779: If the sha256 changes, we need to ensure that the master we upload 
		-- goes to the right revision so that Oz will correctly mark is as valid again
		-- after the sha256 change.
		local maybeSha256 = derefOpt( overlayDoc, 'importSource', 'sha256' )
		local willNotUploadMaster = derefOpt( mergedDoc, 'importSource', 'willNotUploadMaster' )
		if maybeSha256 and willNotUploadMaster then
			-- We check willNotUploadMaster in the merged doc here to verify that it's an 
			-- asset that doesn't already have it's master though this removed via the overlay 
			-- in the lines below this if statement.
			
			overlayDoc._newRevId = KSIdGenerator.uuidInLowercaseHex()
			local localOnly = overlayDoc._localOnly or {}
			-- annotate that we need to wait until Oz marks the binaries invalid (i.e. it's 
			-- received the sha256 changes
			localOnly.binaryUpload = {
				waitForInvalid = maybeSha256 
			}
			
			overlayDoc._localOnly = localOnly
			
		end
		
		local now = KSDate.timeInIso8601Format()
		overlayDoc.userUpdated = now
		
		overlayDoc.importSource = overlayDoc.importSource or {}
		overlayDoc.importSource.importTimestamp = importTimestamp or KSDate.formatDateTimeAsUTC( KSDate.timeNow() ) .. "Z"
		
	else
		
		-- If not present, then this might be the not-yet-synced case or
		-- already-deleted case. We will create the asset with the same
		-- ozAssetId
		
		overlayDoc = assetDocValue
	end
	
	overlayDoc.isAssetUpdate = isAssetUpdate
	
	return Observable.just( overlayDoc )

end

--------------------------------------------------------------------------------

local function convertImportInfoToAssetDocument( wfCatalog, ozUserId, importInfo )

	check( 'WFCatalog', 'uuid', 'table' )

	local docId = importInfo.ozAssetId or importInfo.uuid or KSIdGenerator.uuidInLowercaseHex( )
	local maybeSpaceId = importInfo.spaceId

	logger:debugf( "generating document %s", docId )

	local metadata = WFXmp.WFXmp( )
	metadata:parse( importInfo.xmp )
	
	local devSettingTime = importInfo.devSettingTime -- or dateChanged
		-- NMB-8550: Do not default devSettingTime to dateChanged, if not
		-- explicitly set in importInfo, since we only want the 'userUpdated'
		-- field set in the develop overlay in the case where it represents an
		-- actual edit, rather than just an import. One such scenario is an import
		-- that is part of a migration from Lightroom Classic. In that case, we
		-- want the userUpdated field in the develop overlayto represent edits
		-- made in Lightroom Classic.

	local captureDate = getCaptureDate( metadata, importInfo.fileCreationDate )
	local orientation = importInfo.userOrientation or getOrientation( metadata )
	
	local location = convertLocation( metadata, importInfo.location )

	local ratings = convertRating( ozUserId, importInfo.rating )
	local reviews = convertReviews( ozUserId, importInfo.review )

	local xmpJson = WFCatImportMetadata.convertXmpInfo( metadata )
	xmpJson = convertXmpDc(importInfo, xmpJson)
	xmpJson = convertXmpPhotoshop(importInfo, xmpJson)
	if type( xmpJson ) == 'table' and not next( xmpJson ) then
		-- If there is no content in the XMP section, remove it.
		xmpJson = nil
	end

	metadata:close()
	metadata = nil

	local orientedWidth, orientedHeight = getOrientedDimensions( orientation, importInfo.width, importInfo.height  )
	
	local originalCrop = getOriginalCrop( orientation, importInfo.additionalInfo )
	
	local subtype = importInfo.isVideo and 'video' or 'image'
	
	local extension = importInfo.fileName:match( "%.([^%.]+)$" ):lower()
	local contentType = WFOzContentTypes.contentTypeForExtention( extension ) or WFOzContentTypes.kRawContentType
	
	-- should be sent only if it is true
	local hasEmbeddedDepthMap = importInfo.hasEmbeddedDepthMap or nil
	
	local classicLocalId = derefOpt( importInfo, 'importSource', 'lrcatImageId' )
	local now = KSDate.timeInIso8601Format()
	
	return Observable.just( {

			_id = docId,
			type = 'asset',
			subtype = subtype,
			_spaceId = maybeSpaceId,
			_newRevId = importInfo.revId,
			classicLocalId = classicLocalId,
			
			userUpdated = now,
			userCreated = now,
			
			captureDate = assert( captureDate, "Capture date is required on asset" ),
	
			xmp = xmpJson,

			location = location,
			ratings = ratings,
			reviews = reviews,
	
			importSource = {
				fileName = importInfo.fileName,
				fileSize = importInfo.fileSize,
				importTimestamp = importInfo.importTimestamp 
								or KSDate.formatDateTimeAsUTC( KSDate.timeNow() ) .. "Z",
				importedBy = ozUserId,
				importedOnDevice = WFDeviceInfo.deviceName() or 'unknown',
				sha256 = importInfo.sha256,
				originalWidth = assert( orientedWidth, "Width is required on asset" ),
				originalHeight = assert( orientedHeight, "Height is required on asset" ),
				originalCrop = originalCrop,
				contentType = contentType,
				hasEmbeddedDepthMap = hasEmbeddedDepthMap,
				originalDigest = importInfo.originalDigest,
        
				willNotUploadMaster = importInfo.willNotUploadMaster
			},
			
			mergedAssets = maybeGetMergedAssets( importInfo )

		} )
		:flatMap( function( assetDocValue )
			return addDevelopSettings( wfCatalog, docId, importInfo, orientation, assetDocValue, devSettingTime, importInfo.crVersion )
		end )
		:flatMap( function( assetDocValue )
			return maybeAddc2paIngredients( wfCatalog, importInfo, assetDocValue )
		end )
		:flatMap( function( assetDocValue )
			return appendSyncedDocObservable( wfCatalog, importInfo, assetDocValue )
		end )
		:flatMap( function( assetDocValue, maybeSyncedDoc )
			return handleMergeOrPassOnObv( 
					wfCatalog:ozUserId(), importInfo.ozAssetId, 
					importInfo.importTimestamp, assetDocValue, maybeSyncedDoc )
		end )
		-- generate sha256 
		:flatMap( function ( assetDocValue )
			
			if not importInfo.proxyFilePath or not WFFileUtils.exists( importInfo.proxyFilePath ) then 
				return Observable.throw( {
					failure = WFCatImportConstants.kMissingProxy,
					message = WFCatImportConstants.kMissingProxy,
					filePath = importInfo.filePath,
					classicLocalId = assetDocValue.classicLocalId,
					assetId = assetDocValue._id,
				} )
			end
			
			local masterShaObs = genShaObservable( importInfo , 'master' )
			local proxyShaObs = genShaObservable( importInfo , 'proxy' )
			
			return Observable.combineLatest( masterShaObs, proxyShaObs )
				
				:map( function( masterSha, proxySha )
					
					importInfo.sha256 = masterSha
					importInfo.proxySha256 = proxySha
					
					-- If original filePath doesn't exist, and there's no existing importSource sha256 in synced doc,
					-- use dummy sha256.
					-- Cannot just set nil because version creation requires source asset sha256 for reference
					local maybeSha256 = derefOpt( assetDocValue, 'importSource', 'sha256' )
					assetDocValue.importSource.sha256 = masterSha or maybeSha256 or string.rep( 'd', 64 )
					assetDocValue.importSource.proxySHA256 = proxySha
					
					return assetDocValue, importInfo
					
				end )
				
		end )
		:map( function( assetDocValue, importInfo )
			
			-- proxy localOnly
			local proxyFileName = derefOpt( importInfo, 'proxyFileName' )
			local proxyFilePath = derefOpt( importInfo, 'proxyFilePath' )
			local proxyFileSize = derefOpt( importInfo, 'proxyFileSize' )
			local proxySha256 =   derefOpt( importInfo, 'proxySha256' )
			
			-- original localOnly
			local originalFileName = derefOpt( importInfo, 'fileName' )
			local originalFilePath = derefOpt( importInfo, 'filePath' )
			local originalFileSize = derefOpt( importInfo, 'fileSize' )
			local originalSha256 =   derefOpt( importInfo, 'sha256' )
			
			if proxyFileName and proxyFilePath and proxyFileSize then 
				local fileInfo = applyFilesOverlay( importInfo, 'proxy', proxyFilePath, proxyFileSize, proxySha256 )
				WFTableMerge.applyOverlay( assetDocValue, {
					_localOnly = {
						files = fileInfo
					}
				} )
			end
			
			if originalFileName and originalFilePath and originalFileSize then 
				local fileInfo = applyFilesOverlay( importInfo, 'master', originalFilePath, originalFileSize, originalSha256 )
				WFTableMerge.applyOverlay( assetDocValue, {
					_localOnly = {
						files = fileInfo
					}
				} )
			end
			
			assetDocValue.classicLocalId = derefOpt( importInfo, 'importSource', 'lrcatImageId' )
		
			return assetDocValue
			
		end )
		
end

--==============================================================================

local httpStats

local function instrumentObservableForHttpReport( importObservable )
	if not httpStats then
		httpStats = {}
		
		WFDiagnosticLog.addLogSectionFunc( function()
			if #httpStats == 0 then
				return
			end
			
			return {
		
				name = "Import",
				id = 'WFCatImport',
				weight = 500,
				outputForAutomation = true,
		
				body = httpStats,
			}

		end )		
	end
	
	local sessionStats = {}
	table.insert( httpStats, sessionStats )
	
	sessionStats.numCompleted = 0
	sessionStats.numError = 0
	sessionStats.startTime = KSDate.timeInIso8601FormatNearestSecond( nil, true )
	
	local startTime = KSDate.timeNowForAnimation()
	
	return importObservable
		:tap(
			-- onNext
			function( importedInfo )
				if importedInfo then
					if importedInfo.id and not importedInfo.err then
						sessionStats.numCompleted = sessionStats.numCompleted + 1
					else
						sessionStats.numError = sessionStats.numError + 1
					end
					logger:debugf( "Completed import: assetId = %s, revId = %s, classicLocalId = %s, err = %s", importedInfo.id, importedInfo.revId, importedInfo.classicLocalId, importedInfo.err )
				end
			end,
			
			-- onError
			function( ... )
				sessionStats.importFailure = { ... }
				logger:debug( "Import error in stream" )
			end,
			
			-- onCompleted
			function()
				sessionStats.totalTime = KSDate.timeNowForAnimation() - startTime
				sessionStats.endTime = KSDate.timeInIso8601FormatNearestSecond( nil, true )
				logger:debug( "Completing import stream" )
			end
		)
end


--------------------------------------------------------------------------------
--[[!
	\brief add many asset values and store in docStore at once.

	\param wfCatalog (WFCatalog) catalog to receive the images
	\param assetArray (array of asset ) 
	
	The array may contain zero or more document values. Each value entry
	looks like:
	{  	_id = 10,
		importSource = { 
			lrcatImageId = 123,
			filePath = /path/to/original,
			fileSize = 234,
			fileName = origianlMaster,
			proxyFilePath = /path/to/proxy,
			proxyFileSize = 345,
			proxyFileName = proxyName,
		},
		...
	}
	
	Note: lrcatImageId is classic localId, it'll return back in the end  
	proxyFilePath, proxyFileSize, proxyFileName are a field in oz data-model, we put it here 
	just to form proxy's _localOnly section
	
	\param optImportOptions(optional table) import options, something like below:
		{ batchSize = 10 }

	\return <code>Observable< table ></code> resolves to a single value containing
		an array of the following tables with one entry for each of the supplied doc
		values in order:
			id - the supplied doc _id.
			ok - success or failure, currently always true.
			revId - the revId associated with the document.
			localId - the integer based local doc id that was just allocated.
			classicLocalId - classic localId 
		
]]--

function LrClassicImport.addAssetArray( wfCatalog, assetArray, optImportOptions )
	
	check( 'WFCatalog', 'table', '?table' )
	
	local batchSize = optImportOptions and optImportOptions.batchSize or kDocStoreBatchSize
	local numParallel = kNumParallelImports
	
	local docStore = wfCatalog:docStore()
	local ozUserId = wfCatalog:ozUserId()
	
	local docValuesObservable = Observable.from( assetArray )
		:map( function( importInfo )
			return convertImportInfoToAssetDocument( wfCatalog, ozUserId, importInfo )
				:catch( function( errInfo )

					return Observable.just( errInfo )
							-- don't let one image failure abort the array
				end )
		end )
		:flatMerge( numParallel )
		
	docValuesObservable = docValuesObservable
		:share()
		
	-- Filter out errors
	local errorsObservable = docValuesObservable
		:filter( function( docInfo )
			return docInfo.failure ~= nil
		end )
		:map( function( errorDocInfo )
			return {
				filePath = errorDocInfo.filePath,
				err = errorDocInfo.message,
				id = errorDocInfo.assetId, -- Matching asset-id
				classicLocalId = errorDocInfo.classicLocalId,
			}
		end )

	-- Filter out new assets
	local newAssetsObservable = docValuesObservable
		:filter( function( docInfo )		
			return ( not not docInfo._id ) and not docInfo.isAssetUpdate
		end )
		:bufferWithCount( batchSize )
		:flatMap( function( docInfoArray )
			
			if #docInfoArray > 0 then
			
				local docValArray = {}
				local docIdToClassicLocalIds = {}
				
				for _, docInfo in ipairs( docInfoArray ) do

					local docId = derefOpt( docInfo, '_id' )
					
					if docId then
						local classicLocalId = derefOpt( docInfo, 'classicLocalId' )
						docIdToClassicLocalIds[ docId ] =  classicLocalId
						--reset it to avoid payload validation error
						docInfo.classicLocalId = nil
						--In case of merged Doc setting it to nil befoe apply overlay
						docInfo._changeCounter = nil
						docInfo.isAssetUpdate = nil
					end
					
					docValArray[ #docValArray + 1 ] = docInfo

				end
				
				return WFPerformanceReport.wrapObservable( 'LrClassicImport: bulkSetDocValue', 
					docStore:bulkSetDocValue( docValArray )
						:map( function( bulkSetResults )
						
							-- tables with each entry have { id, ok, revId, localId }
							local resultArray = {}
							for i, result in ipairs( bulkSetResults ) do
							
								resultArray [i] = result
								local docId = result and result.id
								if docId then
									local returnClassicLocalId = docIdToClassicLocalIds [ docId ]
									resultArray[ i ].classicLocalId = returnClassicLocalId
								end
								
							end
							
							-- Each entry have { id, ok, revId, localId, classicLocalId }
							return resultArray
						end )
						:catch( function( errInfo )
						
							logger:warnf( "Fail to import assets, the eror message was: %s", errInfo )
							
							return Observable.just( {
								err = errInfo
							} )
								-- don't let one image failure abort the array

						end ) 
				)	
			end 
			
			return Observable.empty()
			
		end )
		
	-- Filter out updated assets
	local updatedAssetsObservable = docValuesObservable
		:filter( function( docInfo )		
			return ( not not docInfo._id ) and docInfo.isAssetUpdate
		end )
		:bufferWithCount( batchSize )
		:flatMap( function( docInfoArray )
			
			if #docInfoArray > 0 then
			
				local docValArray = {}
				local docIdToClassicLocalIds = {}
				
				for _, docInfo in ipairs( docInfoArray ) do

					local docId = derefOpt( docInfo, '_id' )
					
					if docId then
						local classicLocalId = derefOpt( docInfo, 'classicLocalId' )
						docIdToClassicLocalIds[ docId ] =  classicLocalId
						--reset it to avoid payload validation error
						docInfo.classicLocalId = nil
						--In case of merged Doc setting it to nil befoe apply overlay
						docInfo._changeCounter = nil
						docInfo._labels = nil
						docInfo._revId = nil
						docInfo.isAssetUpdate = nil
					end
					
					docValArray[ #docValArray + 1 ] = docInfo

				end
				
				return WFPerformanceReport.wrapObservable( 'LrClassicImport: bulkApplyOverlay', 
					docStore:bulkApplyOverlay( docValArray )
						:map( function( bulkApplyOverlayResults )
						
							-- tables with each entry have revId
							local resultArray = {}
							for i, docVal in ipairs( docValArray ) do
								local docId = docVal._id
								
								resultArray[ i ] = {
									id = docId,
									ok = true,
									revId = bulkApplyOverlayResults[ i ],
								}
								if docId then
									local returnClassicLocalId = docIdToClassicLocalIds [ docId ]
									resultArray[ i ].classicLocalId = returnClassicLocalId
								end
								
							end
							
							-- Each entry have { id, ok, revId, classicLocalId }
							return resultArray
						end )
						:catch( function( errInfo )
						
							logger:warnf( "Fail to import assets, the eror message was: %s", errInfo )
							
							return Observable.just( {
								err = errInfo
							} )
								-- don't let one image failure abort the array

						end ) 
				)
			end 
			
			return Observable.empty()
			
		end )
		
	local docUpdatesObservable = newAssetsObservable
		:merge( updatedAssetsObservable )
		:flatMap( function( buffer )
			return Observable.from( buffer )
		end )
		:merge( errorsObservable )
	
	return instrumentObservableForHttpReport( docUpdatesObservable )

end

--------------------------------------------------------------------------------

--[[!
	\brief Set original and proxy file info

	\param wfCatalog (WFCatalog) catalog to update binary info
	\param docId (string) docId to update binary info
	
	\param originalFilePath (string) filePath of update original
	\param originalFileSize (number) fileSize of update original
	
	\param proxyFilePath (string) filePath of update proxy
	\param proxyFileSize (number) fileSize of update proxy
	
	\return Observable sends onCompleted when done
]]--

function LrClassicImport.setOriginalAndProxyInfo( wfCatalog, docId, originalFilePath, originalFileSize, proxyFilePath, proxyFileSize )

	check( 'WFCatalog', '+string', '+string', 'number', '+string', 'number' )
	
	local originalShaObs = genShaObservable( { filePath = originalFilePath }, 'master' )
	local proxyShaObs = genShaObservable( { proxyFilePath = proxyFilePath }, 'proxy' )
	local docValueObs = wfCatalog:docStore():docValue( docId )
	
	return Observable.combineLatest( originalShaObs, proxyShaObs, docValueObs )
				
		:flatMap( function( originalSha256, proxySha256, docValue )
			
			local overlay = {
				_id = docId,
				importSource = {
					fileSize = originalFileSize,
					sha256 = originalSha256,
					proxySHA256 = proxySha256
				}
			}
			
			local proxyFileInfo = applyFilesOverlay( nil, 'proxy', proxyFilePath, proxyFileSize, proxySha256 )
			WFTableMerge.applyOverlay( overlay, {
				_localOnly = {
					files = proxyFileInfo
				}
			} )
			
			local originalFileInfo = applyFilesOverlay( nil, 'master', originalFilePath, originalFileSize, originalSha256 )
			WFTableMerge.applyOverlay( overlay, {
				_localOnly = {
					files = originalFileInfo
				}
			} )
			
			local hadC2paInImportSource = derefOpt( docValue, "importSource", "c2paIngredients" )
			
			if hadC2paInImportSource then
				KSTableUtils.setNested( overlay, { _delete = true }, 'importSource', 'c2paIngredients' )
			end
			
			return wfCatalog:docStore():applyOverlay( overlay )
				
		end )
	
end

--------------------------------------------------------------------------------

--[[!
	\brief Update Aux file info

	\param wfCatalog (WFCatalog) catalog to update binary info
	\param docId (string) docId to update binary info
	
	\param binaryTypes (table) binary types to delete from _localOnly files section
		{
			original = true,
			proxy = true,
			xmp_develop = true,
			aux = {
				[ digest ] = true
			}
		}
	
	\return Observable sends onCompleted when done
]]--

function LrClassicImport.deleteBinaryInfoFromLocalOnlySection( wfCatalog, docId, binaryTypes )

	check( 'WFCatalog', '+string', 'table' )
	
	local overlay = { _id = docId }
	local deleteOverlay = { _delete = true }
	
	if binaryTypes.original then
		KSTableUtils.setNested( overlay, deleteOverlay, '_localOnly', 'files', 'original', 'master' )
	end
	
	if binaryTypes.proxy then
		KSTableUtils.setNested( overlay, deleteOverlay, '_localOnly', 'files', 'proxy', '2560' )
	end
	
	if binaryTypes.xmp_develop then
		KSTableUtils.setNested( overlay, deleteOverlay, '_localOnly', 'files', 'xmp_develop' )
	end
	
	if binaryTypes.aux and type( binaryTypes.aux ) == 'table' then
		for digest, _ in pairs( binaryTypes.aux ) do
			KSTableUtils.setNested( overlay, deleteOverlay, '_localOnly', 'files', 'aux', digest )
		end
	end
	
	return wfCatalog:docStore():applyOverlay( overlay )
	
end

--------------------------------------------------------------------------------

local function verifyMetadataHasChanged( currentMetadata, metadataOverlay, ozUserId )

	local copiedCurrentMetadata = KSTableUtils.deepCopy( currentMetadata )
	local copiedMetadataOverlay = KSTableUtils.deepCopy( metadataOverlay )

	-- Strip dates in order to compare actual content changes

	KSTableUtils.setNested( copiedCurrentMetadata, nil, 'ratings', ozUserId, 'date' )
	KSTableUtils.setNested( copiedCurrentMetadata, nil, 'reviews', ozUserId, 'date' )
	KSTableUtils.setNested( copiedMetadataOverlay, nil, 'ratings', ozUserId, 'date' )
	KSTableUtils.setNested( copiedMetadataOverlay, nil, 'reviews', ozUserId, 'date' )
	
	local _, metadataHasChanged = WFTableMerge.applyOverlay( copiedCurrentMetadata, copiedMetadataOverlay )

	return metadataHasChanged

end

--------------------------------------------------------------------------------

--[[!
	\brief Set metadata for assets

	\param wfCatalog (WFCatalog) catalog to update metadata info
	\param assetsMetadata (table) metadata of assets

	\return Observable sends onCompleted when done
]]--

function LrClassicImport.updateAssetMetadata( wfCatalog, assetsMetadata )

	check( 'WFCatalog', 'table' )
	local ozUserId = wfCatalog:ozUserId()

	return Observable.from( assetsMetadata )

		:filter( function( assetMetadata )
			return assetMetadata.ozAssetId
		end )

		:flatMap( function( assetMetadata ) 
			return wfCatalog:docStore():docValue( assetMetadata.ozAssetId )
				:map( function( docValue )
					local canUpdate = not docValue._empty and docValue.type == 'asset' and not docValue._deleted
					return assetMetadata, docValue, canUpdate
				end )
		end )
		
		:filter( function( assetMetadata, docValue, canUpdate)
			return canUpdate
		end )
		
		:map( function( assetMetadata, docValue )
		
			local id_local = assetMetadata.ozAssetId
			local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( id_local )
	
			local metadata = WFXmp.WFXmp( )
		    metadata:parse( assetMetadata.xmp )
	
		    local captureDate = getCaptureDate( metadata, assetMetadata.fileCreationDate )
			local location = getLocationChange( metadata, docValue )
			local ratings = convertRating( ozUserId, assetMetadata.rating )
			local reviews = convertReviews( ozUserId, assetMetadata.review )
	
			local xmpJson = WFCatImportMetadata.convertXmpInfo( metadata )
			xmpJson = convertXmpDc(assetMetadata, xmpJson)
			xmpJson = convertXmpPhotoshop(assetMetadata, xmpJson)
			if type( xmpJson ) == 'table' and not next( xmpJson ) then
				-- If there is no content in the XMP section, remove it.
				xmpJson = nil
			end
			metadata:close()
			metadata = nil
	
			local overlayToApply = {
				_id = id_local,
				captureDate = captureDate,
				xmp = xmpJson,
				location = location,
				ratings = ratings,
				reviews = reviews,
			}
	
			-- Check if overlay makes change to existing docValue

			local currentMetadata = {
				_id = docValue._id,
				captureDate = docValue.captureDate,
				xmp = docValue.xmp,
				location = docValue.location,
				ratings = docValue.ratings,
				reviews = docValue.reviews
			}

			if WFCapabilities.getCapability( 'LrClassic_avoidEmptyContentUpdate' ) then
				
				local metadataHasChanged = verifyMetadataHasChanged( currentMetadata, overlayToApply, ozUserId )

				if not metadataHasChanged then
					return
				end

			end

			overlayToApply._newRevId = assetMetadata.revId
			overlayToApply = WFTableMerge.threeWayMerge( {}, overlayToApply, baseOverlay )
			
			return overlayToApply
			
		end )

		:filter( function( overlayToApply )
			return overlayToApply and next( overlayToApply )
		end )
		
		:toArray()
		
		:flatMap( function( overlays )
			return WFRxThrottledBatchedBulkApplyOverlay( "updateAssetMetadata", wfCatalog:docStore(), overlays, 100, 10 )
		end )
		
end

--------------------------------------------------------------------------------

return LrClassicImport
