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

WFCatImport.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated 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 Systems Incorporated.

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

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

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

	-- WF.core.os
local KSDate = require 'KSDate'
local KSFunctionUtils = require 'KSFunctionUtils'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
local WFDeviceInfo = require 'WFDeviceInfo'
local WFFileUtils = require 'WFFileUtils'
local WFRxScheduler = require 'WFRxScheduler'

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

	-- WF.core.rxlua
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

	-- 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'

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

	-- WF.oz_client.managed_catalog
local WFCatAlbums = require 'WFCatAlbums'
local WFCatImportConstants = require 'WFCatImportConstants'
local WFCatImportDuplicateFiles = require 'WFCatImportDuplicateFiles'
local WFCatImportMetadata = require 'WFCatImportMetadata'
local WFCatImportManifestMergeAssetUtils = require 'WFCatImportManifestMergeAssetUtils'
local WFCatIdUtils = require 'WFCatIdUtils'
local WFCatDevelopSettings = require 'WFCatDevelopSettings'
local WFCatSharedWithYouAlbums = require 'WFCatSharedWithYouAlbums'
local WFImageOrientation = require 'WFImageOrientation'
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatUtils = require 'WFCatUtils'

	-- WF.oz_client.managed_catalog.metadata
local WFXMLNS = require 'WFXMLNS'
local WFCatGPSFormatUtils = require 'WFCatGPSFormatUtils'

	-- WF.oz_client.managed_catalog.profiles
local NmbProfileCatalogUtils = require 'NmbProfileCatalogUtils'

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

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

local check = WFArguments.check

local derefOpt = KSTableUtils.derefOpt

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

local kDontDuplicateCheck = WFConfiguration.Import.DontDuplicateCheck
local kDontGenerateSha256 = WFConfiguration.Import.DontGenerateSha256

local kDocStoreOperation_AddAsset = 1
local kDocStoreOperation_UpdateAsset = 2
local kDocStoreOperation_AddAlbumAsset = 3

local kDocStoreBatchSize = WFConfiguration.Import.BatchSize or 10
local kNumParallelImports = WFConfiguration.Import.NumImagesProcessedInParallel or 3
	-- Tune with experience
	--[[
	** kNumParallelImports currently set to 3 based on testing with stamina.
		Related bugs NMB-4955, NMB-3865, NMB-4404, NMB-4405, NMB-4760, NMB-5272
		The performance also varies based on the type of disk (SSD vs HDD etc)
		But I don't think we're at point where we can reliably differenciate
		between them and fork our functionality based on hardware. This is 
		currently optimized for HDDs and Win+SSD while taking a hit on Mac+SSD
		on which import is the fastest anyway.
	** kDocStoreBatchSize controls the number of images that are processed and
		are waiting only for an entry in the doc store (done in batches). If
		reduced, we will be touching the doc store and index stores often - 
		which hurts performance due to disk thrashing.
	** As of 2017-08-07, kNumParallelImports also determines how many files we
		attempt to copy in parallel (WFRxNativeObservable uses a thread pool
		whose size is decided by the OS). While SSDs seem to handle this well
		(performance degrades on decreasing kNumParallelImports), it does not 
		make a difference for HDDs (anything more than 1 is the same - some
		parallelism is needed since we have both a SHA gen and copy - and doing
		one is almost as expensive as doing both and doing 2 in parallel ensures
		that we are not waiting for both to happen serially). I'm also guessing
		that scheduling too many up front hurts performance outside the app.
	** Decreasing kNumParallelImports will make the system wait for less assets
		to complete if the import is canceled, making it more responsive.
	** Both kNumParallelImports and kDocStoreBatchSize affect the wait to see the
		first images appear as imported.
	--]]

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

	-- 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 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.assetId then
						sessionStats.numCompleted = sessionStats.numCompleted + 1
					else
						sessionStats.numError = sessionStats.numError + 1
					end
					logger:debugf( "Completed import: filePath = %s, assetId = %s, err = %s", importedInfo.filePath, importedInfo.assetId, 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

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

local WFCatImport = {}

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

local function getDate( metadata, namespace, key )
	local dateString = metadata:get( namespace, key )
	
	return KSDate.maybeKSDateFromString( dateString )
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 limitToPeriodOf( input, maxValue )
	-- convert the input value to be within [-maxValue, maxValue]
	
	local period = 2 * maxValue;
	
	if input < -period or input > period then
		input = math.fmod( input, period )
	end

	if input < -maxValue then
		input = input + period
	elseif input > maxValue then
		input = input - period
	end

	return input
end

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

local function normalizeCoord( input, maxValue )

	if not input then
		return
	end

	local pat = '[0-9.]+'
	local value = 0
	local divisor = 1

	for x in string.gfind( input, pat ) do
		value = value + tonumber( x ) / divisor
		divisor = divisor * 60
	end

	local ref = string.gfind( input, '[NSWE]' )()
	if ref == 'S' or ref == 'W' then
		value = - value
	end
	
	local output = tonumber( string.format( "%.6f", value ) )
	
	return limitToPeriodOf( output, maxValue )

end

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

local function convertLocation( metadata, locationInfo )
	local location = {} 
		
	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

	local longitude = normalizeCoord( metadata:get( WFXMLNS.exif, "GPSLongitude" ), 180)
	local latitude = normalizeCoord( metadata:get( WFXMLNS.exif, "GPSLatitude" ), 90 )

	
	if longitude and latitude then
		location.longitude = longitude
		location.latitude = latitude
	end
	
	if altitude then
		location.altitude = altitude
	end
 
	local directionRef = metadata:get( WFXMLNS.exif, "GPSImgDirectionRef" )
	if directionRef then
		location.reference = WFCatGPSFormatUtils.convertGPSDirectionRefToOzRef( directionRef )
	end
	
	local direction = metadata:get( WFXMLNS.exif, "GPSImgDirection" )
	if direction then 
		location.direction = WFCatGPSFormatUtils.convertGPSImgDirectionToNumber( direction )
	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 convertRating( userId, metadata, dateChanged, forceChangedDate )

	local rating = tonumber( metadata:get( WFXMLNS.xap, "Rating" ) )
	
	if forceChangedDate and not rating then
		rating = 0
	end

	if userId and rating and 
		( 0 <= rating ) and 
		( rating <= 5 ) then
		return {
			[ userId ] = {
				rating = math.floor( rating ),
				date = dateChanged,
			}
		}
	end

end
--------------------------------------------------------------------------------

local b = {}
b[ "0" ] = false
b[ "false" ] = false

b[ "1" ] = true
b[ "true" ] = true

-- ashj: This function returns either true, false or nil.
-- False has to be differentaited from nil.
-- Due to the above requirement `not not value`
-- won't work

local function toBooleanOrNil( value )
	local strValue = tostring( value ):lower()
	return b[ strValue ]
end
--------------------------------------------------------------------------------

local function convertReviews( userId, metadata, dateChanged, forceSet )
	local good = metadata:get( WFXMLNS.dm, "good" )
	good = toBooleanOrNil( good )
	
	if good ~= nil then
		good = good and "pick" or "reject"
	elseif forceSet then
		good = 'unflagged'
	end

	if good ~= nil then
		return {
			[ userId ] = {
				flag = good,
				date = dateChanged,
			}
		}
	end

end

--------------------------------------------------------------------------------
local function setRatioValueToDoc( param, doc, src )

	local numerator = derefOpt( src, param, 1 )
	local denominator = derefOpt( src, param, 2 )
	
	if numerator and denominator then
		doc[ param ] = { numerator, denominator } 
	end
end
--------------------------------------------------------------------------------

local function convertVideoInfo( importInfo )

	local videoInfo = importInfo.isVideo and importInfo.additionalInfo
	if not videoInfo then
		return nil
	end
	
	local doc = {}
	
	setRatioValueToDoc( 'duration', doc, videoInfo )
	setRatioValueToDoc( 'frameRate', doc, videoInfo )
	
	doc.audioSampleRate = tonumber( videoInfo.audioSampleRate )
	
	setRatioValueToDoc( 'displayAspectRatio', doc, videoInfo )
	doc.videoBitrate = tonumber( videoInfo.videoBitrate )
	doc.audioBitrate = tonumber( videoInfo.audioBitrate )
	doc.audioSampleSize = tonumber( videoInfo.audioSampleSize )
	doc.videoCodec = videoInfo.videoCodec
	doc.audioCodec = videoInfo.audioCodec
	doc.audioChannel = videoInfo.audioChannel
	doc.fieldOrder = videoInfo.fieldOrder
	doc.storageWidth = tonumber( videoInfo.storageWidth )
	doc.storageHeight = tonumber( videoInfo.storageHeight )
	
	if not next( doc ) then
		return nil -- don't provide an empty table for video
	end
	
	return doc

end

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

local function maybeAddCameraAndLensProfileInfo( profileCatalog, wfCatalog, importInfo, assetDocValue )
	local profiles = importInfo.profiles
	-- profiles.cameraProfileUniqueID should have the UUID of the
	-- camera profile or of the fallback camera profile, if the
	-- profile is not present locally.
	
	if profileCatalog and profiles then 
		
		local maybeIsFallbackProfile = derefOpt( importInfo, "profiles", "isFallbackProfile" )
		local maybeCameraProfileDigest = derefOpt( importInfo, "profiles", "cameraProfileDigest" )
		local cameraModelName = derefOpt( importInfo, "profiles", "cameraModelName" )
	
		return NmbProfileCatalogUtils.addCameraAndLensProfilesToDoc( profileCatalog,
																	wfCatalog,
																	assetDocValue,
																	profiles.cameraProfileUniqueID, 
																	profiles.lensProfileDigest,
																	cameraModelName, 
																	maybeIsFallbackProfile, 
																	maybeCameraProfileDigest )
	else 
		return Observable.just( assetDocValue )
	end	

end 

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

local function addCustomCameraProfileInfo( wfCatalog, profileCatalog, importInfo, assetDocValue )
	-- isFallbackProfile - Whether the cameraProfileUniqueID corresponds to a fallback profile
	-- cameraProfileDigest - The profile digest of the actual camera profile( not the fallback one )
	-- cameraProfileName - The name of the actual camera profile( not the fallback one )
	
	local maybeIsFallbackProfile = derefOpt( importInfo, "profiles", "isFallbackProfile" )
	local maybeCameraProfileDigest = derefOpt( importInfo, "profiles", "cameraProfileDigest" )
	local maybeCameraProfileName = derefOpt( importInfo, "profiles", "cameraProfileName" )
	local cameraModelName = derefOpt( importInfo, "profiles", "cameraModelName" )
	
	return NmbProfileCatalogUtils.getCameraProfileInfoOverlay( wfCatalog:ozCatalogId(), 
			profileCatalog, assetDocValue, maybeIsFallbackProfile, maybeCameraProfileDigest, 
			maybeCameraProfileName, cameraModelName )
		
		:map( function( customProfileOverlay )
			return WFTableMerge.applyOverlay( assetDocValue, customProfileOverlay or {} )
		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 addDevelopSettings( wfCatalog, profileCatalog, imageId, importInfo, orientation, assetDocValue, devSettingTime )
	
	-- SPAM [matthewj 2016-06-22]: Currently video assets should have no 
	-- develop section. The Oz data model for the develop section currently only
	-- supports trim points here and we don't have that sophistication on import
	-- so drop the entire section if it's a video
	if importInfo.isVideo then
		return Observable.just( assetDocValue )
	end
	
	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 )
	
	return WFCatDevelopSettings.developSettingsOverlay( wfCatalog, imageId, 
						orientation, orientedCroppedWidth, orientedCroppedHeight, importInfo.xmpCameraRaw, devSettingTime,
						importInfo.fromDefaults )
		:map( function( developSettingsOverlay )
			return WFTableMerge.applyOverlay( assetDocValue, developSettingsOverlay )
		end )
		:flatMap( function( assetDocValue )		
			return maybeAddCameraAndLensProfileInfo( profileCatalog, wfCatalog, importInfo, assetDocValue )
				:flatMap( function( updatedAssetDocValue ) 
					return addCustomCameraProfileInfo( wfCatalog, profileCatalog, importInfo, updatedAssetDocValue )
				end )
			
		end )
			
end

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

local function updateSidecarMetadataDateAndSaveToFile( sidecarMetadata, sidecarPath )

	local sidecarXmp = WFXmp.WFXmp()
	sidecarXmp:parse( sidecarMetadata )

	local dateTimeNow = KSDate.timeInIso8601FormatNearestSecond()

	sidecarXmp:set( WFXMLNS.xap, "MetadataDate", dateTimeNow )

	local updatedSidecarData = sidecarXmp:serialize()

	return WFAsyncFileUtils.saveToFile( updatedSidecarData, sidecarPath )
		:map( function()
			return sidecarPath
		end )

end

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

local function updateSidecarMetadataDate( sidecarPath )

	return WFAsyncFileUtils.readFile( sidecarPath )

		:flatMap( function( data )

			if data then

				local sidecarMetadata = tostring( data )
				data:close()

				return updateSidecarMetadataDateAndSaveToFile( sidecarMetadata, sidecarPath )

			else

				-- Nothing in file continue.
				return Observable.just( sidecarPath )

			end

		end )

		:catch( function( errorMessage )

			-- Proceed on if we can't update the metadataDate.

			return Observable.just( sidecarPath )

		end )

end

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

local function applyXmpSidecarToDocValue( sidecarPath, assetDocValue )

	local sidecarFileSize = WFFileUtils.fileAttributes( sidecarPath ).fileSize

	local xmpSidecarOverlay = {
		_localOnly = {
			files = {
				original = {
					xmp = {
						path = sidecarPath,
						fileSize = sidecarFileSize,
					}
				}
			},
			xmpSidecarExists = true,
		}
	}

	return WFTableMerge.applyOverlay( assetDocValue, xmpSidecarOverlay )

end

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

local function maybeApplyXmpSidecarToDocValue( wfCatalog, importInfo, imageId, sidecarImportData, sidecarImportPath, assetDocValue )

	if sidecarImportPath == nil and sidecarImportData == nil then

		return Observable.just( assetDocValue )

	end

	local sidecarPath = wfCatalog:fileSpec():pathForXmpSidecar( imageId )

	-- if sidecarImportData is specfied ignore sidecarImportPath as it should override the contents of any sidecar .xmp file

	if sidecarImportData then

		return updateSidecarMetadataDateAndSaveToFile( sidecarImportData, sidecarPath )
			:map( function( savedFilePath )

				return applyXmpSidecarToDocValue( savedFilePath, assetDocValue )

			end )

	else

		return WFAsyncFileUtils.copyToUniquelyNamedFile( sidecarImportPath, sidecarPath )
			:catch( function( copyOrMoveError, platformError )
				return Observable.throw( {
					failure = WFCatImportConstants.kErrorWhileCopying,
					message = WFCatImportConstants.kErrorWhileCopying,
					errInfo = platformError,
					filePath = importInfo.filePath,
					identifier = importInfo.identifier,
				} )
			end )
			:flatMap( function( copiedFilePath )

				-- Update/add metadataDate to force the sidecar XMP on oz to have a different/newer
				-- date than what is in the originals embedded metadata. So that we
				-- can tell if the xmp is from a sidecar.

				return updateSidecarMetadataDate( copiedFilePath )

			end )
			:map( function( copiedFilePath )

				return applyXmpSidecarToDocValue( copiedFilePath, assetDocValue )

			end )

	end

end

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

local function makeUniquePath( importSessionFileNameCache, originalPath )
	
	if importSessionFileNameCache[ originalPath ] then
	
		local ext = originalPath:match( "%.[^%.]+$" )
		local pathWithoutExt = originalPath:sub( 1, #originalPath - #ext )
		local i = 1
		
		while importSessionFileNameCache[ originalPath ] do
			i = i + 1
			originalPath = pathWithoutExt .. '-' .. tostring( i ) .. ext
		end
	end
	
	importSessionFileNameCache[ originalPath ] = true
	return originalPath
	
end

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

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

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

function WFCatImport.convertImportInfoToAssetDocument( wfCatalog, profileCatalog, ozUserId, importInfo )

	check( 'WFCatalog', 'WFCatalog|nil', '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 dateChanged = importInfo.dateChanged
	local forceChangedDate = importInfo.forceChangedDate
	local ratingTime = importInfo.ratingTime or dateChanged
	local reviewTime = importInfo.reviewTime or dateChanged
	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 = getOrientation( metadata )
	local location = convertLocation( metadata, importInfo.location )
	local ratings = convertRating( ozUserId, metadata, ratingTime, forceChangedDate )
	local reviews = convertReviews( ozUserId, metadata, reviewTime, forceChangedDate )

	local xmpJson = WFCatImportMetadata.convertXmpInfo( metadata )
	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
	
	local lrcatStoreProviderId = derefOpt( importInfo, 'importSource', 'lrcatStoreProviderId' )
	local lrcatImageId = derefOpt( importInfo, 'importSource', 'lrcatImageId' )
	
	-- should be sent only if it is true
	local hasEmbeddedDepthMap = importInfo.hasEmbeddedDepthMap or nil
	
	return Observable.just( {

			_id = docId,
			type = 'asset',
			subtype = subtype,
			_spaceId = maybeSpaceId,
	
			captureDate = assert( captureDate, "Capture date is required on asset" ),
	
			xmp = xmpJson,

			location = location,
			ratings = ratings,
			reviews = reviews,

			video = convertVideoInfo( importInfo ),
	
			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,
				
				lrcatStoreProviderId = lrcatStoreProviderId,
				lrcatImageId = lrcatImageId,
			},
			
			_localOnly = {
				xmpSidecarExists = false,
					-- this may be overridden by copyXmpSidecarAndApplyToDocValue below
				
				manifest = importInfo.manifestSourceId and {
					[ importInfo.manifestSourceId ] = true
				},
			},
			
			mergedAssets = maybeGetMergedAssets( importInfo )

		} )

		:flatMap( function( assetDocValue )
			return addDevelopSettings( wfCatalog, profileCatalog, docId, importInfo, orientation, assetDocValue, devSettingTime )
		end )
		
end

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

local function checkOriginalObservable( importInfo )
	local exists = WFFileUtils.exists( importInfo.filePath )
	
	if exists ~= 'file' then
		-- This error is thrown on a scheduler to avoid the processing of this entry, and
		-- starting the next on the same call stack. This throw will result in this
		-- instance of `createDocumentFromImportInfo` to complete. And `flatMerge` will
		-- subscribe to the next on the same call stack. `LUAI_MAXCCALLS` is currently 200
		-- and is not very far away.
		return Observable.throw( {
			failure = WFCatImportConstants.kMissingOriginal,
			message = WFCatImportConstants.kMissingOriginal,
			filePath = importInfo.filePath,
			identifier = importInfo.identifier,
		}, WFRxScheduler.sharedScheduler() )
	end
	
	return Observable.just( importInfo )
end

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

local function genShaObservable( importSessionShaCache, importInfo )

	local providedSha256 = importInfo.sha256

	local shaObservable
	
	if providedSha256 == nil and not kDontGenerateSha256 then

		shaObservable = WFPerformanceReport.wrapObservable(
			'WFCatImport: digestFileSHA256',
			WFCrypto.asyncDigestFile( importInfo.filePath, 'SHA256' )
		)
	
	elseif providedSha256 == "dont-generate" then
	
		shaObservable = Observable.just( nil )
		
	else
	
		shaObservable = Observable.just( providedSha256 )
		
	end


	return shaObservable
		:map( function( sha256 )
			importInfo.sha256 = sha256
			
			if sha256 and
				not importInfo.spaceId and
				importSessionShaCache and
				not importSessionShaCache[ sha256 ] then
				importSessionShaCache[ sha256 ] = importInfo
			end
			
			return importInfo
		end )
end

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

local function findDuplicatesObservable( importSessionShaCache, catalog, importInfo )
	-- Does two things:
	-- 1. Finds if a duplicate exists ( if so, return an error )
	-- 2. If no duplicate, returns assets referring to the same master, but with the master file missing

	local docStore = catalog:docStore()
	local indexStore = catalog:indexStore()

	if importInfo.importIfDuplicate or kDontDuplicateCheck then

		return Observable.just( importInfo )

	else
	
		if importSessionShaCache and
			importSessionShaCache[ importInfo.sha256 ] ~= importInfo then
			
			logger:debugf( "found existing image with sha %s within session", importInfo.sha256 )
			return Observable.throw( {
				failure = WFCatImportConstants.kErrorDuplicate,
				message = WFCatImportConstants.kMatchesExistingFileWithinSession,
				filePath = importInfo.filePath,
				identifier = importInfo.identifier,
			} )
		end

		return WFPerformanceReport.wrapObservable(
			'WFCatImport: findExistingFileBySha256',
			WFCatImportDuplicateFiles.findExistingFileBySha256( docStore, indexStore, importInfo.sha256 )
		)

		:flatMapLatest( function( fileExistsWithSha256, hashOfDocsWithoutMaster )

			if fileExistsWithSha256 then

				if hashOfDocsWithoutMaster then

					local docIdsWithoutMaster = WFCatImportDuplicateFiles.getDocIdsFromHash( hashOfDocsWithoutMaster )

					logger:debugf( "found existing %s image(s) with sha %s but without master", #docIdsWithoutMaster, importInfo.sha256 )
					
					return Observable.combineLatest( function( importInfo, docIdWithoutMaster )

							return importInfo, docIdWithoutMaster

						end,
						Observable.just( importInfo ),
						Observable.from( docIdsWithoutMaster ) )
						
				end
				
				logger:debugf( "found existing image with sha %s", importInfo.sha256 )

				return Observable.throw( {
					failure = WFCatImportConstants.kErrorDuplicate,
					message = WFCatImportConstants.kMatchesExistingFile,
					filePath = importInfo.filePath,
					identifier = importInfo.identifier,
				} )

			end

			return Observable.just( importInfo )

		end )

	end
end

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

local function assembleDocObservableImp( catalog, profileCatalog, importInfo )

	local ozUserId = catalog:ozUserId()

	return WFCatImport.convertImportInfoToAssetDocument( catalog, profileCatalog, ozUserId, importInfo )

end

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

local function assembleDocObservable( catalog, profileCatalog, importInfo )

	return assembleDocObservableImp( catalog, profileCatalog, importInfo )
		:map( function( assetDocValue )
			return importInfo, assetDocValue.captureDate, assetDocValue, false
		end )

end

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

local function copyXmpSidecarObservable( catalog, importInfo, captureDate, assetDocValue, docValueIsOverlay )

	local sidecarData = derefOpt( importInfo, 'additionalInfo', 'XMPSideCarData' )
		-- This will be set by the lightroom migrator when saving xmp data from the catalog database.

	local sidecarFilePath = derefOpt( importInfo, 'additionalInfo', 'XMPSideCarPath' )
		-- This will be set by the importer only when ACR actually uses a sidecar file.

	return maybeApplyXmpSidecarToDocValue( catalog, importInfo, assetDocValue._id, sidecarData, sidecarFilePath, assetDocValue )
		:map( function( assetDocValue )
			return importInfo, captureDate, assetDocValue, docValueIsOverlay
		end )
end
-------------------------------------------------------------------------------

local function assembleMissingOriginalDeleteOverlay( importTimestamp, doc )
	
	doc.importSource = doc.importSource or {}
	doc.importSource.willNotUploadMaster = { _delete = true }
	doc.importSource.importTimestamp = importTimestamp
				or KSDate.formatDateTimeAsUTC( KSDate.timeNow() ) .. "Z"

end

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

local function assembleMissingOriginalOverlayObservable( catalog, importInfo, docIdWithoutMaster )
	
	local docStore = catalog:docStore()
	
	return docStore:docValue( docIdWithoutMaster )
		:map( function( docValue )

			local assetOverlay = {
				_id = docIdWithoutMaster,
			}
			
			assembleMissingOriginalDeleteOverlay( importInfo.importTimestamp, assetOverlay )

			logger:debugf( "updating master in document %s", docIdWithoutMaster )

			return importInfo, docValue.captureDate, assetOverlay, true

		end )
end

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

local function copyOriginalObservable( catalog, importSessionFileNameCache, importInfo, captureDate, assetDocValue, docValueIsOverlay )

	local fileSpec = catalog:fileSpec()
	
	local originalPath =
		fileSpec:pathForOriginal( importInfo.fileName, captureDate, true )
	
	-- copy/moveToUniquelyNamedFile isn't thread safe for collisions. Handle collision edge-cases here instead.
	if importSessionFileNameCache then
		originalPath = makeUniquePath( importSessionFileNameCache, originalPath )
	end

	local transferOriginal
	if importInfo.moveOrCopyOriginal then
		transferOriginal = WFPerformanceReport.wrapObservable( 'WFCatImport: moveToUniquelyNamedFile',
			WFAsyncFileUtils.moveToUniquelyNamedFile( importInfo.filePath, originalPath )
				:catch( function( err ) 
					return WFAsyncFileUtils.copyToUniquelyNamedFile( importInfo.filePath, originalPath, false )
				end )
			)
			
	else
		transferOriginal = WFPerformanceReport.wrapObservable( 'WFCatImport: copyToUniquelyNamedFile',
			WFAsyncFileUtils.copyToUniquelyNamedFile( importInfo.filePath, originalPath, false ) )
	end
	
	return transferOriginal
		:catch( function( copyError, platformError )
			return Observable.throw( {
				failure = WFCatImportConstants.kErrorWhileCopying,
				message = WFCatImportConstants.kErrorWhileCopying,
				errInfo = platformError,
				filePath = importInfo.filePath,
				identifier = importInfo.identifier,
			} )
		end )
		:map( function( copiedFilePath )
			return importInfo, copiedFilePath, assetDocValue, docValueIsOverlay
		end )
end

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

local function formDocInfo( catalog, importInfo, copiedFilePath, assetDocValue, docValueIsOverlay )
	
	local volumeGuid = copiedFilePath and WFFileUtils.getVolumeGuidForPath( copiedFilePath )
	local relativeFilePath = copiedFilePath and WFFileUtils.removeVolumeFromPath( copiedFilePath )
	
	local overlay = WFCatUtils.createBaseOverlayWithUserCreatedFields()
	
	overlay._localOnly = {
		files = {
			original = {
				master = {
					relativePath = relativeFilePath,
					volumeGuid = volumeGuid,
					sha256 = derefOpt( assetDocValue, 'importSource', 'sha256' ) or importInfo.sha256,
					size = derefOpt( assetDocValue, 'importSource', 'fileSize' ) or importInfo.fileSize,
				}
			}
		},
	}
	
	WFTableMerge.applyOverlay( assetDocValue, overlay )

	catalog:getDeletionBacklog():removeFromDeletionBacklog( volumeGuid, relativeFilePath )
		-- If the file has been previously added to the deletion backlog, clear it out
		-- so we don't delete it.
		
	local docInfo = {
		identifier = importInfo.identifier,
		filePath = importInfo.filePath,
		docValue = assetDocValue
	}

	if docValueIsOverlay then

		docInfo.operation = kDocStoreOperation_UpdateAsset

	else

		docInfo.operation = kDocStoreOperation_AddAsset
		
	end
	
	return importInfo, docInfo
end

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

local function saveSHA( catalog, importInfo, docInfo )
	local docStore = catalog:docStore()
	local indexStore = catalog:indexStore()
	
	if not importInfo.spaceId then
		WFCatImportDuplicateFiles.addRecentSha256( docStore, indexStore, importInfo.sha256, docInfo.docValue._id )
	end
end

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

local function addToAlbumObservable( catalog, albumId, importInfo, docInfo )
	
	if albumId then

		return Observable.just( docInfo )
			:merge( WFCatIdUtils.albumUuidForMaybeLocalId( catalog, albumId )
				:map( function( albumUuid )

					local albumAssetDocValue = WFCatAlbums.createAlbumAssetDocument( albumUuid, docInfo.docValue._id, importInfo.spaceId )
					
					local albumDocInfo = {
						identifier = importInfo.identifier,
						filePath = importInfo.filePath,
						operation = kDocStoreOperation_AddAlbumAsset,
						docValue = albumAssetDocValue
					}

					return albumDocInfo
				end )
				:catch( function( err )
					return Observable.throw( {
						failure = WFCatImportConstants.kErrorAddingToAlbum,
						message = WFCatImportConstants.kErrorAddingToAlbum,
						filePath = importInfo.filePath,
						identifier = importInfo.identifier,
					} )
				end )
			)				
	end

	return Observable.just( docInfo )
end

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

local function sanitizeErrors( importInfo, message )
	
	if type( message ) == 'table' then
		return Observable.throw( message )
	else
		return Observable.throw( {
			failure = WFCatImportConstants.kError,
			message = message,
			filePath = importInfo.filePath,
			identifier = importInfo.identifier,
		} )
	end
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 )
		:map( function( docValue )
			
			if docValue._empty or docValue.type ~= 'asset' then
				return nil
			end
			
			return docValue
		end )
end

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

local function importPathFilterObservable( catalog, passType, importInfo, docIdWithoutMaster, mayBeSyncedDoc )
	-- Each asset, should pass through exactly one path.
	-- Deciding which path to take is through this fn.

	if mayBeSyncedDoc or importInfo.ozAssetId then
		return passType == "synced_asset" 
	end
	
	if docIdWithoutMaster then
		return passType == "missing_original"
	end
	
	return passType == "default" 
		
end

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

local function handleMergeOrPassOn( ozUserId, ozAssetId, importTimestamp, assetDocValue, syncedDocValue )
	-- Receives the assembled docValue and the syncedDocValue
	-- Merge both
	
	local overlayDoc
	local docValueIsOverlay = 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
		
		assembleMissingOriginalDeleteOverlay( importTimestamp, overlayDoc )
	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
	
	return overlayDoc, docValueIsOverlay

end

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

local function isMissingOriginal( docValue )

	return derefOpt( docValue, 'importSource', 'willNotUploadMaster' )
		and not WFCatFileSpec.getOriginalLinkFromDocValue( docValue )
		and ( 
			not WFCatFileSpec.getOriginalVolumeGuidFromDocValue( docValue )
			or not WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue )
		)
		
	-- If the doc is marked with willNotUploadMaster check to see if it hasn't
	-- been re-imported already by looking at local path and oz master link.
	-- This is needed because any future edits in Lightroom will update the
	-- importSource adding willNotUploadMaster back.

end

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

local function validateMatchingSyncedFile( importInfo, _, mayBeSyncedDoc )
			
	-- Check if the SHAs match
	if mayBeSyncedDoc then
		if importInfo.sha256 ~= mayBeSyncedDoc.importSource.sha256 then
		
			-- SHAs are different. If master link is present, then
			-- ignore this asset. Else continue.
			
			if not isMissingOriginal( mayBeSyncedDoc ) then
				return Observable.throw( {
					failure = WFCatImportConstants.kMismatchWithExistingFile,
					message = WFCatImportConstants.kMismatchWithExistingFile,
					filePath = importInfo.filePath,
					identifier = importInfo.identifier,
					assetId = mayBeSyncedDoc._id, -- Set the matching asset id.
				} )
			end
		end
	end
	
	return Observable.just( importInfo, mayBeSyncedDoc )
	
end
--------------------------------------------------------------------------------

function WFCatImport.createDocumentFromImportInfo( catalog, profileCatalog, importInfo, importOptions, importSessionCache, takeUntilCancelled )

	local importSessionShaCache = importSessionCache and importSessionCache.shaCache
	local importSessionFileNameCache = importSessionCache and importSessionCache.fileNameCache

	logger:debugf( "createDocumentForFileAtPath %s", importInfo.filePath )

	local syncedAssetPathFilterFn = KSFunctionUtils.bindAllToFront( importPathFilterObservable, catalog, "synced_asset" )
	local missingOriginalPathFilterFn = KSFunctionUtils.bindAllToFront( importPathFilterObservable, catalog, "missing_original" )
	local defaultPathFilterFn = KSFunctionUtils.bindAllToFront( importPathFilterObservable, catalog, "default" )
	
	local genShaObservableFn = KSFunctionUtils.bindAllToFront( genShaObservable, importSessionShaCache )
	local findDuplicatesObservableFn = KSFunctionUtils.bindAllToFront( findDuplicatesObservable, importSessionShaCache, catalog )
	local assembleDocObservableFn = KSFunctionUtils.bindAllToFront( assembleDocObservable, catalog, profileCatalog )
	local assembleDocObservableImpFn = KSFunctionUtils.bindAllToFront( assembleDocObservableImp, catalog, profileCatalog )
	local assembleMissingOriginalOverlayObservableFn = KSFunctionUtils.bindAllToFront( assembleMissingOriginalOverlayObservable, catalog )
	local copyOriginalObservableFn = KSFunctionUtils.bindAllToFront( copyOriginalObservable, catalog, importSessionFileNameCache )
	local copyXmpSidecarObservableFn = KSFunctionUtils.bindAllToFront( copyXmpSidecarObservable, catalog )
	local addToAlbumObservableFn = KSFunctionUtils.bindAllToFront( addToAlbumObservable, catalog, importOptions.albumId )
	local formDocInfoFn = KSFunctionUtils.bindAllToFront( formDocInfo, catalog )
	local saveSHAFn = KSFunctionUtils.bindAllToFront( saveSHA, catalog )
	local sanitizeErrorsFn = KSFunctionUtils.bindAllToFront( sanitizeErrors, importInfo )


	local importInfoObservable = Observable.just( importInfo )
		:flatMapLatest( takeUntilCancelled )
		:flatMapLatest( checkOriginalObservable )
		:flatMapLatest( genShaObservableFn )
		:flatMapLatest( findDuplicatesObservableFn )
		:flatMapLatest( function( importInfo, docIdWithoutMaster )
			return getSyncedAsset( catalog, importInfo )
				:map( function( maybeSyncedDoc )
					return importInfo, docIdWithoutMaster, maybeSyncedDoc
				end )
		end )
		:share()
		
	local docValueForNewAssets = importInfoObservable
		:filter( defaultPathFilterFn )
		:flatMapLatest( assembleDocObservableFn )
		:flatMapLatest( takeUntilCancelled )
		:flatMap( copyXmpSidecarObservableFn )
		:flatMap( copyOriginalObservableFn )
		:map( formDocInfoFn )
		:tap( saveSHAFn )
		
	local docValueForMissingOriginals = importInfoObservable
		:filter( missingOriginalPathFilterFn )
		:flatMap( assembleMissingOriginalOverlayObservableFn )
		:flatMapLatest( takeUntilCancelled )
		:flatMap( copyXmpSidecarObservableFn )
		:flatMap( copyOriginalObservableFn )
		:map( formDocInfoFn )
	
	local docValueForSyncedAssets = importInfoObservable
		:filter( syncedAssetPathFilterFn )
		:flatMapLatest( validateMatchingSyncedFile )
		:flatMapLatest( function( importInfo, mayBeSyncedDoc )
			return assembleDocObservableImpFn( importInfo )
				:map( function( assetDocValue )
					return importInfo, assetDocValue.captureDate, assetDocValue, mayBeSyncedDoc
				end )
		end )
		:map( function( importInfo, captureDate, assetDocValue, syncedDocValue )
			-- Receives the assembled docValue and the syncedDocValue
			-- Merge both
			
			local overlayDoc, docValueIsOverlay = handleMergeOrPassOn( 
					catalog:ozUserId(), importInfo.ozAssetId, 
					importInfo.importTimestamp, assetDocValue, syncedDocValue )
			
			return importInfo, captureDate, overlayDoc, docValueIsOverlay

		end )
		:flatMapLatest( takeUntilCancelled )
		:flatMap( copyXmpSidecarObservableFn )
		:flatMap( copyOriginalObservableFn )
		:map( formDocInfoFn )
		
	local allDocValues = docValueForNewAssets
		:merge( docValueForMissingOriginals, docValueForSyncedAssets )
		
	local returnValue = allDocValues
		:flatMap( addToAlbumObservableFn )
		:catch( sanitizeErrorsFn )

	return WFPerformanceReport.wrapObservable(
		'WFCatImport: createDocumentFromImportInfo', returnValue )

end

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

function WFCatImport.addDocumentsInBatches( wfCatalog, docValuesObservable, optBatchSize )

	local batchSize = optBatchSize or kDocStoreBatchSize

	local docStore = wfCatalog:docStore()
	local fileAccessDb = wfCatalog:fileAccessDb()
	
	docValuesObservable = docValuesObservable
		:share()
		
	-- Filter out errors
	local errorsObservable = docValuesObservable
		:filter( function( docInfo )
			return docInfo.failure ~= nil
		end )
		:map( function( docInfo )
			return {
				filePath = docInfo.filePath,
				identifier = docInfo.identifier,
				err = docInfo.message,
				errInfo = docInfo.errInfo,
				assetId = docInfo.assetId, -- Matching asset-id
			}
		end )
	
	
	-- Filter out new assets
	local newAssetsObservable = docValuesObservable
		:filter( function( docInfo )		
			return docInfo.operation == kDocStoreOperation_AddAsset
		end )
		:bufferWithCount( batchSize )
		:flatMap( function( docInfoArray )
		
			if #docInfoArray > 0 then
				local docValArray = {}
				
				for _, docInfo in ipairs( docInfoArray ) do
					table.insert( docValArray, docInfo.docValue )
				end
				
				return WFPerformanceReport.wrapObservable( 'WFCatImport: bulkSetDocValue', 
					docStore:bulkSetDocValue( docValArray, nil, "import_addPhotos" )
						:map( function( bulkSetResults )
							for i, docInfo in ipairs( docInfoArray ) do
								docInfo.docValue._localId = bulkSetResults[ i ].localId
							end
							return docInfoArray
						end )
				)
			end
			
			return Observable.empty()
		end )
		
		
	-- Filter out updated assets
	local updatedAssetsObservable = docValuesObservable
		:filter( function( docInfo )
			return docInfo.operation == kDocStoreOperation_UpdateAsset
		end )
		:bufferWithCount( batchSize )
		:flatMap( function( docInfoArray )
			
			if #docInfoArray > 0 then
				local docValArray = {}
				
				for _, docInfo in ipairs( docInfoArray ) do
					table.insert( docValArray, docInfo.docValue )
				end
				
				return WFPerformanceReport.wrapObservable(  'WFCatImport: bulkApplyOverlay', 
					docStore:bulkApplyOverlay( docValArray )
						:map( function( ... )
							return docInfoArray
						end )
				)
			end
			
			return Observable.empty()
		end )

	
	-- Filter out album_asset documents which don't need to return anything to the client
	local albumAssetsObservable = docValuesObservable
		:filter( function( docInfo )
			return docInfo.operation == kDocStoreOperation_AddAlbumAsset
		end )
		:bufferWithCount( batchSize )
		:flatMap( function( docInfoArray )
			
			if #docInfoArray > 0 then
				local docValArray = {}
				
				for _, docInfo in ipairs( docInfoArray ) do
					table.insert( docValArray, docInfo.docValue )
				end
				
				return WFPerformanceReport.wrapObservable( 'WFCatImport: bulkSetDocValue', 
					docStore:bulkSetDocValue( docValArray, nil, "import_addToAlbum" )
						:map( function( ... )
							return docInfoArray
						end )
				)
			end
			
			return Observable.empty()
		end )
		:ignoreElements()
	
	
	-- Merge all three streams that update the doc store
	local docUpdatesObservable = newAssetsObservable
		:merge( updatedAssetsObservable )
		:merge( albumAssetsObservable )
		:flatMap( function( buffer )
			return Observable.from( buffer )
		end )
		:tap( function( docInfo )
			-- Mark this entry as accessed
			local idToUse = docInfo.docValue._localId or docInfo.docValue._id
			fileAccessDb:touch( idToUse )
		end )
		:map( function( docInfo )
			-- Form the result for the client
			return {
				filePath = docInfo.filePath,
				identifier = docInfo.identifier,
				assetId = docInfo.docValue._id,
				localId = docInfo.docValue._localId,
			}
		end )
		
	-- Merge the errors stream
	return docUpdatesObservable
		:merge( errorsObservable )		

end

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

local function maybeContributeAssetToAlbum( wfCatalog, assetStream, importOptions )

	local contributeToAlbumId = importOptions.contributeToAlbumId
	local redactionOptions = importOptions.redactionOptions or {}
	
	local retObservable
	
	if contributeToAlbumId then
		
		retObservable = assetStream
			:flatMap( function( importedInfo )
			
				if importedInfo.localId then
				
					return WFCatSharedWithYouAlbums.getSpaceIdOfAlbumId( wfCatalog, contributeToAlbumId )
						:flatMap( function( spaceId )
						
							local retObservable = Observable.just( importedInfo )
							
							if spaceId then
							
								retObservable = retObservable
									:merge( 
										WFCatSharedWithYouAlbums.addAssetsToAlbum( wfCatalog, spaceId, contributeToAlbumId, { importedInfo.localId }, redactionOptions )
											:ignoreElements()
											:catch( function( err )
												return Observable.just( {
													filePath = importedInfo.filePath,
													identifier = importedInfo.identifier,
													assetId = importedInfo.assetId,
													localId = importedInfo.localId,
													
													err = WFCatImportConstants.kErrorWhileContributing,
												} )
											end )
									)
							
							end
							
							return retObservable
						
						end )
				
				else
					
					return Observable.just( importedInfo )
				
				end
			
			end )
		
	else
	
		retObservable = assetStream
		
	end
	
	return retObservable
	
end

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

local function addSpaceInfoToImportInfoStream( assetStream, wfCatalog, importOptions )

	local albumId = importOptions.albumId

	if albumId and importOptions.redactionOptions == false then
		-- Importing into a SharedWithYou album doesn't support redactions as yet. But we don't have a workflow
		-- where that is needed since in all cases, they will be an effect of processing another image from the
		-- same album (eg. HDR, pano merge, NND, edit in PS). Just ensure that the user of this API understands
		-- that and sets redactionOptions = false
		
		local spaceIdObservable = WFCatSharedWithYouAlbums.getSpaceIdOfAlbumId( wfCatalog, albumId )
		
		return Observable.combineLatest( assetStream, spaceIdObservable )
			:map( function( importInfo, maybeSpaceId )
				if maybeSpaceId then
					importInfo.spaceId = maybeSpaceId
					importInfo.importIfDuplicate = true
				end
				return importInfo
			end )
	else
		return assetStream
	end
end

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

local function addImageStreamToCatalog( wfCatalog, profileCatalog, assetStream, maybeSynchronizer, optImportOptions )

	local importOptions = optImportOptions or {}
	
	local batchSize = importOptions.batchSize or kDocStoreBatchSize
	local numParallel = importOptions.numParallel or kNumParallelImports

	if maybeSynchronizer and maybeSynchronizer:isSyncRunningNow() then
		maybeSynchronizer:stopSyncing()
	end
	
	local maybeRestartSync = function()
	
		if maybeSynchronizer then
			maybeSynchronizer:startSyncing()
		end
	end
		
	local addImageStreamObservable = Observable.create( function( observer )

		local importSessionCache = {
			shaCache = {},
			fileNameCache = {},
		}
		
		local isCancelled = false
		local function takeUntilCancelled( ... )
			
			if isCancelled then
				-- Put this on a scheduler so that the whole set of cancelled images is not unwrapped on the same call stack.
				return Observable.empty( WFRxScheduler.sharedScheduler() )
			else
				return Observable.just( ... )
			end
			
		end

		assetStream = addSpaceInfoToImportInfoStream( assetStream, wfCatalog, importOptions )

		local addImagesObservable = assetStream
			:map( function( importInfo )

				return WFCatImport.createDocumentFromImportInfo( wfCatalog, profileCatalog, importInfo, importOptions, importSessionCache, takeUntilCancelled )
					:catch( function( errInfo )

						return Observable.just( errInfo )
							-- don't let one image failure abort the array

					end )

			end )
			:flatMerge( numParallel )
				-- flatMerge used to limit how many items are actually started at once as cancel
				-- will let the inflight items complete.

		addImagesObservable = WFCatImport.addDocumentsInBatches( wfCatalog, addImagesObservable, batchSize )
		
		addImagesObservable = maybeContributeAssetToAlbum( wfCatalog, addImagesObservable, importOptions )
			:tap( function( ... )
					observer:onNext( ... )
				end,
				function( ... )
					observer:onError( ... )
				end,
				function()
					observer:onCompleted()
				end )
			:finally( maybeRestartSync )
			:call( WFRxRunUntilDone )

		return function()
		    maybeRestartSync()
			isCancelled = true
		end

	end )

	return instrumentObservableForHttpReport( addImageStreamObservable )

end

--------------------------------------------------------------------------------
--[[!
	\brief Adds an array of images from disk to the catalog.
	
	\param wfCatalog (WFCatalog) catalog to receive the images
	\param profileCatalog (WFCatalog) catalog containing camera profiles
	\param assetArray (array of strings) paths of files to add
	\param maybeSynchronizer (WFCatSynchronizer or nil) used to cancel the sync during import to enhance performance (needed?)
	\param optImportOptions ( table or nil ) used to specify the target albumId, batchSize, numParallel and contribution options

	\return Observable sends onCompleted when done
]]--

function WFCatImport.addImageArrayToCatalog( wfCatalog, profileCatalog, assetArray, maybeSynchronizer, optImportOptions )

	local assetStream = Observable.from( assetArray )
	
	local addImageStreamObservable = addImageStreamToCatalog( wfCatalog, profileCatalog, assetStream, maybeSynchronizer, optImportOptions )

	return WFPerformanceReport
		.wrapObservable( "WFCatImport: addImageArrayToCatalog", addImageStreamObservable )

end

--------------------------------------------------------------------------------
--[[!
	\brief Adds images from disk to the catalog. Returns a Subject into which images to be added can be sent.
	
	\param wfCatalog (WFCatalog) catalog to receive the images
	\param profileCatalog (WFCatalog) catalog containing camera profiles
	\param maybeSynchronizer (WFCatSynchronizer or nil) used to cancel the sync during import to enhance performance (needed?)
	\param optImportOptions ( table or nil ) used to specify the target albumId, batchSize, numParallel and contribution options

	\return Subject, Observable Subject that is the source of images to import. Observable that sends the results of the import. Sends onCompleted when done
]]--

function WFCatImport.addImagesToCatalog( wfCatalog, profileCatalog, maybeSynchronizer, optImportOptions )

	local assetSubject = Rx.Subject()
	
	local addImageStreamObservable = addImageStreamToCatalog( wfCatalog, profileCatalog, assetSubject, maybeSynchronizer, optImportOptions )
	
	addImageStreamObservable = addImageStreamObservable
		:finally( function()
			-- We need to explicitly complete the input stream here since we cannot guarantee
			-- that it is closed by the client ( this observable can be disposed if import
			-- is canceled )
			assetSubject:onCompleted()
			
		end )

	return assetSubject, WFPerformanceReport
		.wrapObservable( "WFCatImport: addImagesToCatalog", addImageStreamObservable )

end

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

function WFCatImport.makeAddImagesModel( wfCatalogStream, profileCatalogStream, wfCatSynchronizerStream, optImportOptions )

	local imagesSourceSubject

	return {
			
		value = Observable.combineLatest( wfCatalogStream,
										profileCatalogStream, 
										wfCatSynchronizerStream )
					:take( 1 )
					:flatMapLatest( function( wfCatalog, profileCatalog, wfCatSynchronizer )
						logger:debug( "Creating import model" )
						
						local WFCatImport = require 'WFCatImport'
						local sourceSubject, retObservable = WFCatImport.addImagesToCatalog( wfCatalog, profileCatalog, wfCatSynchronizer, optImportOptions )
						
						imagesSourceSubject = sourceSubject
						
						return Observable.just( { status = WFCatImportConstants.kImportSessionStarted } )
							:merge( retObservable )
						
					end ),
					
		events = {
			addImageArray = function( _, imageArray )
				check( '_', 'table' )
				logger:debugf( "Received %d files", #imageArray )
				
				if imagesSourceSubject then
					for _, importInfo in ipairs( imageArray ) do
						imagesSourceSubject:onNext( importInfo )
					end
				end
			end,
			
			endStream = function()
				logger:debug( "Ending import stream" )
			
				if imagesSourceSubject then
					imagesSourceSubject:onCompleted()
					imagesSourceSubject = nil
				end
			end
		},
	}
end

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

return WFCatImport
