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

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'
local WFPlatformInfo = require 'WFPlatformInfo'

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

	-- WF.oz_client.managed_catalog
local WFCatAlbums = require 'WFCatAlbums'
local WFCatAssetAux = require 'WFCatAssetAux'
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'
local WFCatVersions = require 'WFCatVersions'

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

	-- WF.oz_client.managed_catalog.profiles
local NmbProfileCatalogUtils = require 'NmbProfileCatalogUtils'
local NmbCrPathOverrides = require 'NmbCrPathOverrides'
local NmbAcr = require "NmbAcr"

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

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

local check = WFArguments.check

local derefOpt = KSTableUtils.derefOpt

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

local WFCrContext = NmbAcr.WFCrContext
local WFVideoFoundationContext = NmbAcr.WFVideoFoundationContext

local kDontDuplicateCheck = WFConfiguration.Import.DontDuplicateCheck
local kDontGenerateSha256 = WFConfiguration.Import.DontGenerateSha256
local kDontGenerateOriginalDigest = WFConfiguration.Import.DontGenerateOriginalDigest
local kDontApplyLongPathFix = WFConfiguration.Import.DontApplyLongPathFix

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 handleMergeOrPassOn
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 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 = 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
	
	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.transferFunction = videoInfo.transferFunction
	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 )
	
	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 WFCatDevelopSettings.developVideoSettingsOverlay( 
						imageId, temporal, orientedCroppedWidth,
						orientedCroppedHeight,
						importInfo.settingsVideoFoundation, devSettingTime )
			: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 auxilaryFilePath = NmbCrPathOverrides.getAuxParentPath( wfCatalog )
	
	local maybeAuxInfo, xmp = {}, nil
	local needsUpdateAISettings = nil
	if importInfo.xmpCameraRaw then
		maybeAuxInfo, xmp, needsUpdateAISettings = NmbAcr.WFCrDevelopSettings.extractAuxFilesFromXMP( importInfo.xmpCameraRaw, auxilaryFilePath , importInfo.filePath )

		importInfo.xmpCameraRaw = xmp
	end
	
	local assetDocValueObs = WFCatDevelopSettings.developSettingsOverlay( wfCatalog, imageId, 
						orientation, orientedCroppedWidth, orientedCroppedHeight, importInfo.xmpCameraRaw, 
						maybeAuxInfo, needsUpdateAISettings,
						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 )
	
	local auxDataList =  maybeAuxInfo or {}

	local auxOverlayObs = WFCatAssetAux.getAuxOverlayForAuxInfo(
			wfCatalog, auxDataList )
	
	return Observable.combineLatest( 
				assetDocValueObs,
				auxOverlayObs )
			:map( 
				function( assetDocValue, auxOverlay )
					local assetDocValueWithAux = 
							WFTableMerge.applyOverlay( assetDocValue, auxOverlay )
					return assetDocValueWithAux
				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

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

-- [jjeyakar 2022-06-27]: See NMB-21469
-- On WAS builds, we found several reports of users not being able to import. Traced
-- this down to users not being able to copy files into the catalog folder ( error code
-- ERROR_PATH_NOT_FOUND ). On WAS builds (packaged apps), the AppData folder can be
-- virtualized i.e., the real folder is in an app-specific path ( Eg C:\Users\kmoch\AppData\Local\Packages\AdobeSystemsIncorporated.AdobeLightroom_ynb6jyjzte8ga\LocalCache\Local\Adobe\Lightroom CC\Data\5d5498d657c24099b9caa5f504fc46b0\originals\2020\2020-12-12\_DSC0117.ARW )
-- However, this folder is exposed to us as C:\Users\kmoch\AppData\Local\Adobe\Lightroom CC\Data\5d5498d657c24099b9caa5f504fc46b0\originals\2020\2020-12-12\_DSC0117.ARW
-- Looks like there is a bug on Win while copying to this virtualized path. And it looks
-- like some rate-limiting is happening beacuse a few images do get copied into the same
-- path. But when we try to copy more (often 7th image), the copy fails. This happens
-- for importing, copying from device into the staging folder as well as moving originals.
-- An engineer from MS suggested we prefix the path with "\\?\" to bypass the MAX_PATH
-- limit. Even though the non-virtualized path is shorter than MAX_PATH, this fix seems
-- to be effective.

local APPLY_LONG_PATH_FIX = require "WFPlatformInfo".isWinAppStore and ( not kDontApplyLongPathFix )

local kLongPathPrefix = '\\\\?\\'

local function hasLongPathPrefix( str )
	return string.sub( str, 1, #kLongPathPrefix ) == kLongPathPrefix
end

local function maybeApplyWinAppStoreLongPathFix( str )
	if APPLY_LONG_PATH_FIX and not hasLongPathPrefix( str ) then
		return kLongPathPrefix .. str
	end
	
	return str
end

local function maybeRevertWinAppStoreLongPathFix( str )

	if APPLY_LONG_PATH_FIX and hasLongPathPrefix( str ) then
		return string.sub( str, #kLongPathPrefix + 1 )
	end

	return str
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

		sidecarPath = maybeApplyWinAppStoreLongPathFix( sidecarPath )
		
		return WFAsyncFileUtils.copyToUniquelyNamedFile( sidecarImportPath, sidecarPath )
			:catch( function( copyOrMoveError, platformError )
				return Observable.throw( {
					failure = WFCatImportConstants.kErrorWhileCopying,
					message = WFCatImportConstants.kErrorWhileCopying,
					errInfo = {
						platformError = platformError,
						dstPath = sidecarPath,
					},
					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.
				
				copiedFilePath = maybeRevertWinAppStoreLongPathFix( copiedFilePath )

				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,
				originalDigest = importInfo.originalDigest,
        
				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 getOriginalDigestObservable( importInfo )

	local originalDigestObservable = Observable.just( importInfo.originalDigest )
	
	if importInfo.originalDigest == nil and not kDontGenerateOriginalDigest then
		
		local success, context
		
		if importInfo.isVideo then
			success, context = pcall( WFVideoFoundationContext, importInfo.filePath )
		else
			success, context = pcall( WFCrContext, importInfo.filePath )
		end
		
		if success then

			originalDigestObservable = WFPerformanceReport.wrapObservable(
				'WFCrContext: getOriginalDigest',
				context:getOriginalDigest()
					:catch( function()
						logger:warnf( 'WFCrContext getOriginalDigest failed for %s', importInfo.filePath )
						return Observable.just( nil )
					end )
					:finally( function()
						context:close()
					end )
			)
	
		else
			logger:warnf( 'WFCrContext init failed for %s', importInfo.filePath )
		end
		
	end
	
	return originalDigestObservable
		:map( function( originalDigest )
			
			if originalDigest then
				importInfo.originalDigest = originalDigest
			end
			
			return importInfo
		end )
end

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

local function findDuplicatesObservable( importSessionShaCache, catalog, importOptions, 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
		
		local function handleDuplicate( existingFile )
			if importOptions.updateDuplicates then
				importInfo.matchedAssetToUpdate = existingFile
				importInfo.devSettingTime = KSDate.timeInIso8601FormatNearestSecond()
				return Observable.just( importInfo )
			elseif importOptions.albumId then
				importInfo.addDuplicateToAlbum = true
				importInfo.matchedAsset = existingFile
				return Observable.just( importInfo )
			else
				return Observable.throw( {
					failure = WFCatImportConstants.kErrorDuplicate,
					message = WFCatImportConstants.kMatchesExistingFile,
					filePath = importInfo.filePath,
					identifier = importInfo.identifier,
				} )
			end
		end
		
		local duplicatesByOriginalDigest = WFPerformanceReport.wrapObservable(
				'WFCatImport: findExistingFilesByOriginalDigests',
				WFCatImportDuplicateFiles.findExistingFilesByOriginalDigests( catalog, { importInfo.originalDigest } )
			)
			:flatMapLatest( function( foundHashes )
			
				local foundHash, existingFile = next( foundHashes )
				
				if not foundHash then
					return Observable.just( importInfo )
				end
				
				logger:debugf( "found existing image with original digest %s", importInfo.originalDigest )
			
				return handleDuplicate( existingFile )
			
			end )

		local duplicatesBySha256 = 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 handleDuplicate( fileExistsWithSha256 )
				end

				return Observable.just( importInfo )

			end )
		
		return importOptions.useOriginalDigestToCheckDupes
			and duplicatesByOriginalDigest
			or duplicatesBySha256

	end
end

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

local function computeDiff( old, new )
	
	local overlay = WFTableMerge.createOverlay( old, new )
	
	-- This overlay may include aux sections that only differ by their id
	-- We need to re-use them
	local auxTable = overlay.aux
	if auxTable then

		-- We don't delete any entries from the aux table, nor the aux section itself
		auxTable._delete = nil
		
		for id, aux in pairs( auxTable ) do
			if aux._delete then
				auxTable[ id ] = nil
			end
		end

		for id, aux in pairs( auxTable ) do
		
			-- Search for this aux in the old document
			local tryingToReplace = false
			for id2, aux2 in pairs( old.aux or {} ) do
				if aux.digest == aux2.digest then
					tryingToReplace = true
					break
				end
			end
			
			-- remove the new one with the changed id
			if tryingToReplace then
				auxTable[ id ] = nil
			end
		end
		
	end
	
	-- We don't delete the aux local section as well
	local auxLocal = derefOpt( overlay, '_localOnly', 'files', 'aux' )
	if auxLocal then
		auxLocal._delete = nil
		for id, aux in pairs( auxLocal ) do
			if aux._delete then
				auxLocal[ id ] = nil
			end
		end
	end
	
	-- Repeat the same with masks. The overlay may be just renaming the ids of existing masks.
	-- We don't want to do that. Instead, reuse the existing mask wherever possible.
	local maskTable = derefOpt( overlay, 'develop', 'masks' )
	if maskTable and not maskTable._delete then
		for id, mask in pairs( maskTable ) do
		
			-- Search for this mask in the old document
			local oldId = false
			for id2, mask2 in pairs( derefOpt( old, 'develop', 'masks' ) or {} ) do
				if mask.digest == mask2.digest then
					oldId = id2
					break
				end
			end
			
			-- See if we are trying to delete the old id in the overlay
			local tryingToReplace = false
			if oldId then
				for id2, mask2 in pairs( maskTable ) do
					if id2 == oldId and mask2._delete then
						tryingToReplace = true
						break
					end
				end
			end
			
			-- remove both
			if tryingToReplace then
				maskTable[ id ] = nil
				maskTable[ oldId ] = nil
			end
		end
		
	end

	-- Also repeat the same with filters. The overlay may be just renumbering the auxIds of existing filters.
	-- We don't want to just renumber them as once an aux file is put on oz it is never deleted.
	-- Instead, reuse the existing filter wherever possible.

	local filterTable = derefOpt( overlay, 'develop', 'filters' )
	local oldFilterTable = derefOpt( old, 'develop', 'filters' )

	if oldFilterTable and filterTable and not filterTable._delete then

		for groupDigest, groupInfo in pairs( filterTable ) do

			for auxId, filterInfo in pairs( groupInfo ) do

				local oldGroupInfo = oldFilterTable[ groupDigest ]
				if oldGroupInfo then

					-- Search for this filter in the old document
					local oldAuxId = false
					for auxId2, filterInfo2 in pairs( oldGroupInfo ) do
						if filterInfo.digest == filterInfo2.digest then
							oldAuxId = auxId2
							break
						end
					end

					-- See if we are trying to delete the old auxId in the overlay
					local tryingToReplace = false
					if oldAuxId then
						for auxId2, filterInfo2 in pairs( groupInfo ) do
							if auxId2 == oldAuxId and filterInfo2._delete then
								tryingToReplace = true
								break
							end
						end
					end

					-- remove both
					if tryingToReplace then
						groupInfo[ auxId ] = nil
						groupInfo[ oldAuxId ] = nil
					end

				end

			end

		end

	end

	return overlay

end

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

local function maybeRemoveUserUpdated( sectionDiff )

	if not sectionDiff then
		return false
	end

	local hasChanges = false
	
	for k, v in pairs( sectionDiff ) do
		
		if k ~= 'userUpdated' then
			
			if type( v ) == 'table' then
				hasChanges = maybeRemoveUserUpdated( v )
			else
				hasChanges = true
			end
			
		end
		
	end
	
	if not hasChanges then
		sectionDiff.userUpdated = nil
	end
	
	return hasChanges

end

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

local function cleanupEmptyChanges( overlay )
	
	maybeRemoveUserUpdated( overlay.develop )
	
	-- Remove empty sections
	local function removeEmptyImp( t, key )
		local val = t[ key ]
		
		if type( val ) == 'table' then
		
			for k, v in pairs( val ) do
				removeEmptyImp( val, k )
			end
		
			if not next( val ) then
				t[ key ] = nil
			end
		end
	end
	
	for k in pairs( overlay ) do
		removeEmptyImp( overlay, k )
	end	
	
	if next( overlay ) then
		overlay.userUpdated = KSDate.timeInIso8601Format()
	end
end

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

local function makeUpdateDuplicateOverlay( catalog, updateMetadata, createVersion, newDocValue, importInfo )

	return catalog:docStore():docValue( importInfo.matchedAssetToUpdate )
		:take( 1 )
		:flatMap( function( existingDocValue )
		
			local existingDocCopy = KSTableUtils.deepCopy( existingDocValue )
		
			KSTableUtils.setNested( existingDocCopy, derefOpt( newDocValue, '_localOnly', 'files', 'aux' ), '_localOnly', 'files', 'aux' )
			KSTableUtils.setNested( existingDocCopy, derefOpt( newDocValue, '_localOnly', 'files', 'xmp_develop' ), '_localOnly', 'files', 'xmp_develop' )
			existingDocCopy.aux = newDocValue.aux
			existingDocCopy.develop = newDocValue.develop
			
			if updateMetadata then
				existingDocCopy.xmp = newDocValue.xmp
				existingDocCopy.location = newDocValue.location
				existingDocCopy.ratings = newDocValue.ratings
				existingDocCopy.reviews = newDocValue.reviews
				existingDocCopy.video = newDocValue.video
			end
			
			local overlay = computeDiff( existingDocValue, existingDocCopy )
			
			cleanupEmptyChanges( overlay )
			
			overlay._id = existingDocValue._id
			overlay._localId = existingDocValue._localId
			
			local assetDevelopUpdated = not not derefOpt( overlay, 'develop' )
			
			if createVersion and not importInfo.isVideo then
			
				local function createVersionForCurrentSettings()
				
					local sha256 = derefOpt( newDocValue, 'develop', 'xmpCameraRaw', 'sha256' )
					
					if sha256 then
					
						local pathForSettingsXmp = catalog:fileSpec():pathForSettingsXmp( sha256 )
						
						local developSettingsParam = KSTableUtils.deepCopy( newDocValue.develop )
						developSettingsParam.cameraRawXmpFilePath = pathForSettingsXmp
						developSettingsParam.cameraRawXmpSha256 = sha256
						
						return WFAsyncFileUtils.calculatePathSize( pathForSettingsXmp )
							:flatMapLatest( function( fileSize )
								developSettingsParam.cameraRawXmpSize = fileSize
								
								return WFCatVersions.createVersionForAssetWithExternalSettings( catalog,
									importInfo.matchedAssetToUpdate,
									false,
									nil,
									developSettingsParam )
							end )
					
					else
					
						local developSettingsParam = KSTableUtils.deepCopy( newDocValue.develop )
						developSettingsParam.cameraRawXmpStr = developSettingsParam.xmpCameraRaw
						developSettingsParam.xmpCameraRaw = nil
						
						return WFCatVersions.createVersionForAssetWithSettings( catalog,
							importInfo.matchedAssetToUpdate,
							false,
							nil,
							developSettingsParam )
					end
				
				end
				
				-- The first time that an asset is imported, we don't create a version. However,
				-- if it is being updated, we create two versions - one for the existing settings
				-- and one the current settings. The version for existing settings is created only
				-- if there is no auto-version already for the asset. This is the same logic used
				-- by the client when assets are edited and we'd like to retain the same behaviour
				-- here.
				local createTwoVersions = Observable.just( true )
				
				-- But we need to do anything only if we are going to update the dev settings
				if derefOpt( overlay, 'develop' ) then
				
					createTwoVersions = WFCatVersions.getAutoVersionCountForAsset( catalog, existingDocValue._id )
						
						:flatMapLatest( function( numAutoVersions )
						
							if numAutoVersions == 0 then
								
								return WFCatVersions.createVersionForAsset( catalog, existingDocValue._id, false, nil, nil )
									
									:take( 1 )
									
									:flatMapLatest( function()
										
										return createVersionForCurrentSettings()
										
									end )
								
							end
							
							return createVersionForCurrentSettings()
						
						end )
				
				end
				
				return createTwoVersions
					:take( 1 )
					:map( function()
						return overlay, assetDevelopUpdated
					end )
			
			end
			
			return Observable.just( overlay, assetDevelopUpdated )
				
		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 assembleSyncedAssetOverlayObservable( catalog, profileCatalog, importInfo, maybeSyncedDoc )

	return assembleDocObservableImp( catalog, profileCatalog, importInfo )
		:map( function( assetDocValue )

			-- Receives the assembled docValue and the syncedDocValue
			-- Merge both
			
			local overlayDoc, docValueIsOverlay = handleMergeOrPassOn( 
					catalog:ozUserId(), importInfo.ozAssetId, 
					importInfo.importTimestamp, assetDocValue, maybeSyncedDoc )
			
			return importInfo, assetDocValue.captureDate, overlayDoc, docValueIsOverlay

		end )

end

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

local function assembleUpdateDuplicateOverlayObservable( catalog, profileCatalog, updateMetadata, createVersion, importInfo )

	return assembleDocObservableImp( catalog, profileCatalog, importInfo )

		:flatMapLatest( function( assetDocValue )

			return makeUpdateDuplicateOverlay( catalog, updateMetadata, createVersion, assetDocValue, importInfo )
				:map( function( overlayDoc, assetDevelopUpdated )
					return importInfo, nil, overlayDoc, true, assetDevelopUpdated
				end )

		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
	
	originalPath = maybeApplyWinAppStoreLongPathFix( originalPath )

	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 = platformError,
					dstPath = originalPath,
				},
				filePath = importInfo.filePath,
				identifier = importInfo.identifier,
			} )
		end )
		:map( function( copiedFilePath )
			copiedFilePath = maybeRevertWinAppStoreLongPathFix( copiedFilePath )
			
			return importInfo, copiedFilePath, assetDocValue, docValueIsOverlay
		end )
end

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

local function formDocInfo( catalog, importInfo, copiedFilePath, assetDocValue, docValueIsOverlay, assetDevelopUpdated )
	
	if copiedFilePath then
	
		local volumeGuid = WFFileUtils.getVolumeGuidForPath( copiedFilePath )
		local relativeFilePath = 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 )
		
		-- LrClassic don't have getDeletionBacklog function
		if not WFPlatformInfo.isClassic() then
			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.
		end
	end
		
	local docInfo = {
		identifier = importInfo.identifier,
		filePath = importInfo.filePath,
		docValue = assetDocValue
	}

	
	if importInfo.addDuplicateToAlbum then
	
		docInfo.operation = WFCatImportConstants.kDocStoreOperation_DoNotAddThisDuplicate
	
	elseif docValueIsOverlay then

		if assetDevelopUpdated then
			docInfo.operation = WFCatImportConstants.kDocStoreOperation_UpdateAssetDevelop
		else
			docInfo.operation = WFCatImportConstants.kDocStoreOperation_UpdateAsset
		end

	else

		docInfo.operation = WFCatImportConstants.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.addHashesForRecentImage( docStore, indexStore, importInfo.sha256, importInfo.originalDigest, docInfo.docValue._id )
	end
end

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

local function addToAlbumObservable( catalog, albumId, importInfo, docInfo )
	
	-- This function passes the docInfo along to be added to the doc store. In addition, it adds another
	-- entry into the output stream for the album-asset to be added. If the image is a duplicate, don't add
	-- it to the output stream, but throw an appropriate error, but still add an album-asset entry.
	
	local docInfoPassThrough
	local assetUuid
	
	if docInfo.operation == WFCatImportConstants.kDocStoreOperation_DoNotAddThisDuplicate then
	
		docInfoPassThrough = Observable.just( {
			failure = WFCatImportConstants.kErrorDuplicate,
			message = WFCatImportConstants.kMatchesExistingFileButAddedToAlbum,
			filePath = importInfo.filePath,
			identifier = importInfo.identifier,
		} )
		
		assetUuid = importInfo.matchedAsset
		
	else
	
		docInfoPassThrough = Observable.just( docInfo )
		assetUuid = docInfo.docValue._id
		
	end
	
	if albumId then

		return docInfoPassThrough
			:merge( WFCatIdUtils.albumUuidForMaybeLocalId( catalog, albumId )
				:map( function( albumUuid )

					local albumAssetDocValue = WFCatAlbums.createAlbumAssetDocument( albumUuid, assetUuid, importInfo.spaceId )
					
					local albumDocInfo = {
						identifier = importInfo.identifier,
						filePath = importInfo.filePath,
						operation = WFCatImportConstants.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 docInfoPassThrough
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 appendSyncedDocObservable( catalog, importInfo, docIdWithoutMaster )

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

			return importInfo, docIdWithoutMaster, maybeSyncedDoc

		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 importInfo.matchedAssetToUpdate then
		return passType == "update_matched_asset"
	end
	
	if importInfo.addDuplicateToAlbum then
		return passType == "add_to_album_only"
	end

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

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

handleMergeOrPassOn = function( 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 addToAlbumPathFilterFn = KSFunctionUtils.bindAllToFront( importPathFilterObservable, catalog, "add_to_album_only" )
	local updateMatchedAssetFilterFn = KSFunctionUtils.bindAllToFront( importPathFilterObservable, catalog, "update_matched_asset" )
	
	local genShaObservableFn = KSFunctionUtils.bindAllToFront( genShaObservable, importSessionShaCache )
	local findDuplicatesObservableFn = KSFunctionUtils.bindAllToFront( findDuplicatesObservable, importSessionShaCache, catalog, importOptions )
	local appendSyncedDocObservableFn = KSFunctionUtils.bindAllToFront( appendSyncedDocObservable, catalog )
	
	local assembleDocObservableFn = KSFunctionUtils.bindAllToFront( assembleDocObservable, catalog, profileCatalog )
	local assembleSyncedAssetOverlayObservableFn = KSFunctionUtils.bindAllToFront( assembleSyncedAssetOverlayObservable, catalog, profileCatalog )
	local assembleMissingOriginalOverlayObservableFn = KSFunctionUtils.bindAllToFront( assembleMissingOriginalOverlayObservable, catalog )
	local assembleUpdateDuplicateOverlayObservableFn = KSFunctionUtils.bindAllToFront( assembleUpdateDuplicateOverlayObservable, catalog, profileCatalog, importOptions.updateMetadata, importOptions.createVersionForUpdate )
	
	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, Rx.Scheduler.currentThread() )
		:flatMapLatest( takeUntilCancelled )
		:flatMapLatest( checkOriginalObservable )
		:flatMapLatest( genShaObservableFn )
		:flatMapLatest( getOriginalDigestObservable )
		:flatMapLatest( findDuplicatesObservableFn )
		:flatMapLatest( appendSyncedDocObservableFn )
		: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( assembleSyncedAssetOverlayObservableFn )
		:flatMapLatest( takeUntilCancelled )
		:flatMap( copyXmpSidecarObservableFn )
		:flatMap( copyOriginalObservableFn )
		:map( formDocInfoFn )
		
	local addDuplicatesToAlbum = importInfoObservable
		:filter( addToAlbumPathFilterFn )
		:flatMapLatest( takeUntilCancelled )
		:map( formDocInfoFn )
		
	local updateMatchedAssets = importInfoObservable
		:filter( updateMatchedAssetFilterFn )
		:flatMapLatest( assembleUpdateDuplicateOverlayObservableFn )
		:flatMapLatest( takeUntilCancelled )
		:map( formDocInfoFn )
		
		
	local allDocValues = docValueForNewAssets
		:merge( docValueForMissingOriginals, docValueForSyncedAssets, addDuplicatesToAlbum, updateMatchedAssets )
		
	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()
	
	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 == WFCatImportConstants.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 == WFCatImportConstants.kDocStoreOperation_UpdateAsset or docInfo.operation == WFCatImportConstants.kDocStoreOperation_UpdateAssetDevelop
		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 == WFCatImportConstants.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 )
		:map( function( docInfo )
			-- Form the result for the client
			return {
				filePath = docInfo.filePath,
				identifier = docInfo.identifier,
				assetId = docInfo.docValue._id,
				localId = docInfo.docValue._localId,
				operation = docInfo.operation,
			}
		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.assetId 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.assetId }, 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
