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

WFCatImportManifestAssetsProcessor.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2017 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.

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

	-- Rx
local Rx = require 'Rx'
local Observable = Rx.Observable
local Subject = Rx.Subject
local Disposable = Rx.Disposable

	-- WF.core.os
local KSDate = require 'KSDate'
local KSPathUtils = require 'KSPathUtils'
local WFFileUtils = require 'WFFileUtils'

	-- WF.core.utils
local WFRequestQueue = require 'WFRequestQueue'

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

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local KSJsonDecoder = require 'KSJsonDecoder'
local WFConfiguration = require 'WFConfiguration'

local WFRatioTime = require 'WFRatioTime'

	-- WF.oz_client.managed_catalog
local WFCatImport = require 'WFCatImport'

	-- WF.oz_client.managed_catalog.import
local WFCatImportConstants = require 'WFCatImportConstants'
local WFCatImportManifestConstants = require 'WFCatImportManifestConstants'
local WFCatImportManifestProcessorUtils = require 'WFCatImportManifestProcessorUtils'
local WFCatImportManifestProcessorBase = require 'WFCatImportManifestProcessorBase'

--[[
	Things to remember while touching this file:
		1. The import stream's observable can return an error synchronously (before the subject's :onNext() returns)
		2. The import stream's observable needs to be subscribed to before sending entries into the subject
		3. The import stream's subject should be closed when both of these occur (either can happen first)
			a. no more rows to process from the queue
			b. no more rows to process from the db
]]

--[[
	Strategy to avoid duplicates due to non-atomic update
	of docstore+manifeststatus db
		
		1. Before calling the docstore add api, we update the
			manifest status db, with pending status along with
			a generated uuid.
			
		2. If docstore got updated, but the app was killed
			before the status update, then the affected assets
			will have pending status in manifest-status db.
			
		3. On resume, we will go through the list of assets
			in pending status and will query the docstore
			to verify if we have updated the assets or not.
			Note that in the case of synced assets, it will
			not be sufficient to check the existence of the
			document. Hence we add the manifest source id as
			a _localOnly value and read the same to confirm.
			
		4. We will update all pending assets' status, to success
			if the manifest-source-id is found in the document.
			
		5. We are not removing the verification-failed records,
			since they will be retried and will be overwritten
			with the proper status after processing.
			
]]
--------------------------------------------------------------------------------

local logger = WFLogger 'WFCatImportManifestAssetsProcessor'
local derefOpt = KSTableUtils.derefOpt
local fakeRevisionExistsErrorUUID = WFConfiguration.Manifest.fakeRevisionExistsErrorUUIDForAsset

local kDbBatchSize = 100
local kMaxAssetsInFlight = 40
	-- How many assets are in the import pipeline at a time.
	-- Affects how many more failures we report when one asset fails
	-- Tune based on performance and UI considerations
	-- Currently set to 40 (the number of requests that the import pipeline
	-- 		keeps open. If lowered below 20, we may hit a deadlock because
	--		of the way Import batches doc store writes.

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

local kNotEnoughSpaceError = 'Unable to copy. Not enough space'

local _importErrorToManifestErrorMap = {}
_importErrorToManifestErrorMap[ WFCatImportConstants.kMissingOriginal ] = 
			WFCatImportManifestConstants.messages.assetImportErrorOriginalMissing.code
_importErrorToManifestErrorMap[ WFCatImportConstants.kMismatchWithExistingFile ] = 
			WFCatImportManifestConstants.messages.assetProcessingNonMatchingSyncdAsset.code
_importErrorToManifestErrorMap[ WFCatImportConstants.kErrorWhileCopying ] = 
			WFCatImportManifestConstants.messages.assetProcessingUnableToCopyFile.code
_importErrorToManifestErrorMap[ kNotEnoughSpaceError ] = 
			WFCatImportManifestConstants.messages.assetImportErrorNotEnoughSpace.code

local kUserIgnoredAsset = "UserIgnored"

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

local WFCatImportManifestSmartQueue = {
	Disposable,
	
	_requestFactory = false,
	_currentRequest = false,
	
	_queue = false,
	_queueSize = false,
	_numDataReceived = false,
	_noMoreData = false,
	
	_nextId = false,
	_nextIdGenerator = false,

	_dataSubject = false,
	_pendingDataCount = false,
}

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

function WFCatImportManifestSmartQueue:__init( requestFactory, firstId, nextIdGenerator )
	self._requestFactory = requestFactory
	
	self._queue = WFRequestQueue()
	self._queueSize = 0
	self._currentRequest = nil
	self._numDataReceived = 0
	self._noMoreData = false

	self._dataSubject = Subject()
	self._pendingDataCount = 0
	
	self._nextId = firstId
	self._nextIdGenerator = nextIdGenerator
	
end

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

function WFCatImportManifestSmartQueue:dispose()
	
	if self._currentRequest then
		self._currentRequest:dispose()
	end
	
	-- Though we may have been disposed, our source request may still
	-- be firing data. Make sure we don't respond to it.
	self._dataSubject:onCompleted()
	self._noMoreData = true
	self._queue:dispose()
	
	return self:Rx_Disposable_dispose()
end

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

function WFCatImportManifestSmartQueue:getObservable()
	return self._dataSubject
end

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

function WFCatImportManifestSmartQueue:_requestForMore()
	logger:debugf( "_requestForMore: requesting for more rows from the assets table. pending %s requests", self._pendingDataCount )
	
	self._currentRequest = self._requestFactory( self._nextId, kDbBatchSize )

		:tap( function( dataArray )
			local numEntries = #dataArray
			logger:debugf( "_requestForMore: received %s entries. total = %s", numEntries, self._numDataReceived + numEntries )

			self._numDataReceived = self._numDataReceived + numEntries
			
			if numEntries > 0 then
				self._nextId = self._nextIdGenerator( dataArray )
			end
			
			for i = 1, numEntries do
				
				if self._pendingDataCount > 0 then
					-- We have a pending request. Service it.
					self._dataSubject:onNext( dataArray[ i ] )
					self._pendingDataCount = self._pendingDataCount - 1
				else
					-- The rest can go into our buffer
					if not self._queue:isDisposed() then
						self._queue:addToQueue( dataArray[ i ] )
						self._queueSize = self._queueSize + 1
					end
				end
				
			end
			
			if numEntries < kDbBatchSize then
				logger:debugf( "_requestForMore: received all rows" )
				self._noMoreData = true
				
				-- This is the last batch of rows and the queue is empty. We're done.
				if self._queueSize == 0 then
					logger:debugf( "_requestForMore: closing subject" )
					self._dataSubject:onCompleted()
				end
				return
			end
		end )
		
		:tapOnCompleted( function()
			-- Clear out the current request
			self._currentRequest:dispose()
			self._currentRequest = nil

			-- Data from this request is not sufficient. Start another.
			if self._pendingDataCount > 0 and not self._noMoreData then
				logger:debugf( "_requestForMore: could not process all requests. Requesting once again." )
				self:_requestForMore()
			end
		end )
		
		:subscribe()
end

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

function WFCatImportManifestSmartQueue:sendOneMore()
	if self._queueSize > 0 then
		local row = self._queue:nextRequest()
		self._queueSize = self._queueSize - 1
		self._dataSubject:onNext( row )
		return
	end
	
	if self._noMoreData then
		-- The queue is empty and we can't expect any more from the db
		logger:debugf( "sendOneMore: all rows fetched, and nothing in the queue. Closing stream." )
		self._dataSubject:onCompleted()
		return
	end
	
	-- Save this request for us to fulfill later
	self._pendingDataCount = self._pendingDataCount + 1

	if not self._currentRequest then
		self:_requestForMore()
	end
end

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

WFCatImportManifestSmartQueue = Rx.Class 'WFCatImportManifestSmartQueue'( WFCatImportManifestSmartQueue )

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

local WFCatImportManifestAssetsProcessor = {

	WFCatImportManifestProcessorBase,
}

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

-- Converts a string representation of a ratioTime to an array of numerator
-- and denominator if possible. If not, return nil.
local function maybeReadRatioTime( ratioTimeString )
	
	local success, val = pcall( function()
		local t = WFRatioTime.makeRatioTime( )
		t:unarchive( ratioTimeString )
		
		return { t.numerator, t.denominator }
	end )
	
	if success then
		return val
	end
	
	return nil
end

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

local function readImportTimeStamp( maybeResumeInfo )
	
	-- If the context already has a importTimestamp, use
	-- the same time for importing, since this will
	-- ensure that the migrated images will be shown 
	-- as a group.
	local importTimestamp = derefOpt( maybeResumeInfo, "resumeContext", "importTimestamp" )
	if importTimestamp then
		
		return importTimestamp
	end

	return KSDate.formatDateTimeAsUTC( KSDate.timeNow() ) .. "Z"
end

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

local function readIgnoreErrors( maybeResumeInfo )
	local ignoreAll, ignoreIndividual = {}, {}
	
	if maybeResumeInfo and maybeResumeInfo.errorDecisions then
		
		if maybeResumeInfo.errorDecisions.ignoreAll then
			for messageCode, _ in pairs( maybeResumeInfo.errorDecisions.ignoreAll ) do
				ignoreAll[ messageCode ] = true
			end
		end
		
		if maybeResumeInfo.errorDecisions.ignoreIndividual then
			for messageCode, ignoreArr in pairs( maybeResumeInfo.errorDecisions.ignoreIndividual ) do
				for _, ignoreInfo in ipairs( ignoreArr ) do
					local local_id = ignoreInfo[ 2 ]
					ignoreIndividual[ local_id ] = true
				end
			end
		end
	end
	
	return ignoreAll, ignoreIndividual
end

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

local function getProfilesInfo( assetInfo )
	local profiles = {
		lensProfileDigest = assetInfo.lensProfileDigest,
	}
	
	if assetInfo.cameraProfileDigest then
		local ok, cameraProfileInfo = pcall( KSJsonDecoder.decodeWebSafeJson, assetInfo.cameraProfileDigest )
		
		if ok then
		
			profiles.isFallbackProfile = cameraProfileInfo.isFallbackProfile
			profiles.cameraProfileUniqueID = cameraProfileInfo.cameraProfileUniqueID
			profiles.cameraProfileDigest = cameraProfileInfo.cameraProfileDigest
			profiles.cameraProfileName = cameraProfileInfo.cameraProfileName
			profiles.cameraModelName = cameraProfileInfo.cameraModelName
		
		elseif type( assetInfo.cameraProfileDigest ) == "string" then
			
			profiles.cameraProfileUniqueID = assetInfo.cameraProfileDigest
		end

	end
	
	return profiles
end


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

function WFCatImportManifestAssetsProcessor:__init()
	self._stageName = WFCatImportManifestConstants.stageNames.assets

	self._messageCodeCompleted = WFCatImportManifestConstants.messages.assetProcessingComplete.code
	self._messageCodeErrorCount = WFCatImportManifestConstants.messages.assetProcessingErrorCount.code
	self._messageCodeGenericError = WFCatImportManifestConstants.messages.assetProcessingError.code
end

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

function WFCatImportManifestAssetsProcessor:_updateStatus( manifestStatusDatabase, local_id, oz_id, status )
	
	local statusData = { {
		local_id = local_id,
		status = status,
	} }
	
	return manifestStatusDatabase:writeData( 'updateAssetStatus', statusData )
		:ignoreElements()
end

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

function WFCatImportManifestAssetsProcessor:_addStatus( manifestStatusDatabase, local_id, oz_id, status )
	
	local statusData = { {
		local_id = local_id,
		oz_id = oz_id,
		status = status,
	} }
	
	return manifestStatusDatabase:writeData( 'addOrReplaceAssetStatus', statusData )
		:ignoreElements()
end

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

function WFCatImportManifestAssetsProcessor:_updateStatusBulk( manifestStatusDatabase, assets, status )
	
	local statuses = {}
	for i, asset in ipairs( assets ) do
		statuses[ #statuses + 1 ] = {
			local_id = asset.local_id,
			status = status,
		}
	end
	
	return manifestStatusDatabase:writeData( 'updateAssetStatus', statuses )
		:ignoreElements()
	
end

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

function WFCatImportManifestAssetsProcessor:_getLowestLocalId( manifestDatabase )
	
	return manifestDatabase:readData( 'minAssetId' )
		:map( function( results )
			if results[ 1 ] then
				return results[ 1 ].min_id
			end
		end )
	
end

function WFCatImportManifestAssetsProcessor:setup( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )

	return self:preProcessPending( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )
		:defaultIfEmpty()
		:flatMap( function()
		
			return self:WFCatImportManifestProcessorBase_setup( catalog, 
				manifestStatusDatabase, idCache, messageHandler, 'selectAssetsStatus', 
				WFCatImportManifestConstants.idCacheSets.assets )
		end )

end

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

--[[
	If there are pending records, we go through them and
	verify if they already exist. If they do, then we
	update the status of the same, so that it will get
	considered in the cache-setup phase.
]]
function WFCatImportManifestAssetsProcessor:preProcessPending( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )
	
	return manifestStatusDatabase:readData( "selectAssetStatusWithStatus" , { status = WFCatImportManifestConstants.status.pending } )
	
		:flatMap( function( pendingEntries )
		
			local docIds = {}
			local statusForDocId = {}
			
			for i, entry in ipairs( pendingEntries ) do
					
				docIds[ i ] = entry.oz_id
				statusForDocId[ entry.oz_id ] = entry
					
			end
		
			return WFCatImportManifestProcessorUtils.maybeReadManifestSourceId( manifestDatabase )
			
			:flatMap( function( sourceId )
				
				return catalog:docStore():bulkDocValue( docIds )
			
				:flatMap( function( docValues )
		
					local statusListToUpdate = {}
					
					for i, docValue in ipairs( docValues ) do
					
						if not docValue.empty then
							-- Document exists. 
							-- Check for import_source to reconfirm
							-- we have updated this doc.( synced albums
							-- can already be present. )
							
							local isManifestSourceIdPresent = not not derefOpt( docValue, 
																"_localOnly",
																"manifest",
																sourceId )
							
							if isManifestSourceIdPresent then
							
								local statusInfo = statusForDocId[ docValue._id ]
								statusListToUpdate[ #statusListToUpdate + 1 ] = {
									local_id = statusInfo.local_id,
								}
							
							end

						end
					end
					
					if #statusListToUpdate > 0 then
						return self:_updateStatusBulk( manifestStatusDatabase, statusListToUpdate, WFCatImportManifestConstants.status.success )
					else
						return Observable.empty()
					end
					
				end )
			end )
			
		end )

end

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

function WFCatImportManifestAssetsProcessor:_createImportInfo( assetInfo, lrcatStoreProviderId, manifestSourceId, importTimestamp )
	
	local localId = assetInfo.local_id
	local filePath = assetInfo.filePath
	local fileName = KSPathUtils.leafName( filePath )
				
	
	local fileAttrs = WFFileUtils.fileAttributes( filePath )
	local fileSize = fileAttrs.fileSize
	
	local touchTime, devSettingTime
	
	if assetInfo.touchTime and tonumber( assetInfo.touchTime ) ~= 0 then
		touchTime = KSDate.timeInIso8601Format( assetInfo.touchTime )
		devSettingTime = touchTime
	else
		touchTime = importTimestamp
	end
	
	local ratingTime = assetInfo.ratingTime or touchTime
	local reviewTime = assetInfo.reviewTime or touchTime
	
	local isVideo = assetInfo.subType == 'video'
	
	local additionalInfo = {}
	
	-- Storing the xmp sidecar for all assets (including JPEGs) as a backup of the metadata from the manifest since
	-- for all practical purposes, the information there is the original.
	additionalInfo.XMPSideCarData = assetInfo.xmp
	
	if isVideo then
		if assetInfo.duration then
			additionalInfo.duration = maybeReadRatioTime( assetInfo.duration )
		end

		if assetInfo.frameRate then
			additionalInfo.frameRate = maybeReadRatioTime( assetInfo.frameRate )
		end
	end
	
	local importSource = lrcatStoreProviderId and {
		lrcatStoreProviderId = lrcatStoreProviderId,
		lrcatImageId = assetInfo.global_id,
	}
	
	local profiles = getProfilesInfo( assetInfo )
	
	local pendingUUID = self._processedRows[ localId ] and self._processedRows[ localId ].oz_id
	local uuid = assetInfo.ozAssetId or pendingUUID or KSIdGenerator.uuidInLowercaseHex( )
	
	if fakeRevisionExistsErrorUUID and not pendingUUID then
		-- Use the value only once, so that the request
		-- succeeds next time.
		uuid = fakeRevisionExistsErrorUUID
	end
	
	return {
		identifier = localId,
		uuid = uuid,
		
		fileName = fileName,
		fileSize = fileSize,
		filePath = filePath,
		dateChanged = touchTime,
		importTimestamp = importTimestamp,
		ozAssetId = assetInfo.ozAssetId,
		
		ratingTime = ratingTime,
		reviewTime = reviewTime,
		devSettingTime = devSettingTime,
		
		importIfDuplicate = true,
		forceChangedDate = true,
		
		isVideo = isVideo,
		width = assetInfo.fileWidth,
		height = assetInfo.fileHeight,
		croppedWidth = assetInfo.croppedWidth,
		croppedHeight = assetInfo.croppedHeight,

		xmp = assetInfo.xmp,
		xmpCameraRaw = assetInfo.xmpCameraRaw,
		
		location = {
			city = assetInfo.city,
			state = assetInfo.state,
			country = assetInfo.country,
		},
	
		profiles = profiles,
		
		importSource = importSource,
		manifestSourceId = manifestSourceId,
		
		additionalInfo = additionalInfo,
	}
	
end

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

function WFCatImportManifestAssetsProcessor:processImpObservable( catalog, profileCatalog, wfCatSynchronizer, 
			manifestDatabase, manifestStatusDatabase, idCache, messageHandler, contextInfo )

	local lrcatStoreProviderId
	local manifestSourceId
		-- Will be set once the chain begins.
	
	local maybeResumeInfo = contextInfo.resumeInfo
	
	local importTimestamp = readImportTimeStamp( maybeResumeInfo )
	local ignoreAllErrors, ignoreIndividualErrors = readIgnoreErrors( maybeResumeInfo )
	
	local errorEntries = {}
	local rowsQueue
	
	local importSubject, importObservable = WFCatImport.addImagesToCatalog( catalog, profileCatalog, wfCatSynchronizer )
			
	local function shouldIgnoreError( messageCode, local_id )
		if ignoreAllErrors[ messageCode ] then
			return true
		elseif ignoreIndividualErrors[ local_id ] then
			return true
		end
		
		return false
	end
	
	local function shouldIgnoreAsset( local_id )
		if ignoreIndividualErrors[ local_id ] then
			return true
		end
		
		return false
	end
	
	local function ignoreAsset( local_id, filePath, messageCode )
		self:_sendMessage( messageHandler, WFCatImportManifestConstants.messages.assetIgnoredErrorWarning.code, { { filePath, local_id, messageCode } } )
		self:_updateError( messageHandler, 1 )
		return self:_updateStatus( manifestStatusDatabase, local_id, nil, WFCatImportManifestConstants.status.ignored )
			
			:defaultIfEmpty( nil )
			:tap( function()
				
				rowsQueue:sendOneMore()
			
			end )
	end 
	
	
	local function shouldPauseImport( messageCode )
		-- Pause the import only for actionable errors. For others, report the error, be calm and carry on.
		return messageCode == WFCatImportManifestConstants.messages.assetImportErrorOriginalMissing.code
			or messageCode == WFCatImportManifestConstants.messages.assetImportErrorNotEnoughSpace.code
			or messageCode == WFCatImportManifestConstants.messages.assetProcessingUnableToCopyFile.code
	end
	
	
	local function maybeTranslateResult( importedInfo )
		if importedInfo.err and importedInfo.err == WFCatImportConstants.kErrorWhileCopying then
			-- If we received a copy error, check for disk space. 
			-- If disk space is low, report as low-disk-space error
			
			local fileSpec = catalog:fileSpec()
			local originalsPathInfo = fileSpec:originalsPathInfo()
			
			if originalsPathInfo.volumeGuid then
				local volumeAgent = catalog:getVolumeAgentByVolumeGuid( originalsPathInfo.volumeGuid )
				
				if volumeAgent then
				
					return volumeAgent:getVolumeStatusStream()
						:take( 1 )
						:map( function( volumeStatus )
							
							if volumeStatus.isLowOnSpace then
								importedInfo.err = kNotEnoughSpaceError
							end
							
							return importedInfo
							
						end )
				end
			end
			
		end
		
		return Observable.just( importedInfo )
	end
	
		
	local function handleImportResult( importedInfo )
		local local_id = importedInfo.identifier
		
		logger:debugf( "Import request processed err = %s", importedInfo.err )
		
		if importedInfo.err then
			
			local messageCode = _importErrorToManifestErrorMap[ importedInfo.err ]
			local unknownError = false
			
			if not messageCode then
				messageCode = WFCatImportManifestConstants.messages.assetImportError.code
				unknownError = true
			end
			
			local shouldIgnoreError = shouldIgnoreError( messageCode, local_id )
			local shouldPauseImport = shouldPauseImport( messageCode )
			
			logger:debugf( "ignore err = %s, pause import = %s", shouldIgnoreError, shouldPauseImport )
			
			if shouldIgnoreError then
				
				return ignoreAsset( importedInfo.identifier, importedInfo.filePath, messageCode )
				
			else
				
				if not unknownError then
					self:_sendMessage( messageHandler, messageCode, { { importedInfo.filePath, importedInfo.identifier } } )
				else
					self:_sendMessage( messageHandler, messageCode, { { importedInfo.filePath, importedInfo.identifier, tostring( importedInfo.err ) } } )
				end
				
				self:_updateError( messageHandler, 1 )
				
				-- Do not save the status for failed imported assets, so we can try them again the next time.
				-- self:_updateStatus( manifestStatusDatabase, local_id, nil, WFCatImportManifestConstants.status.failure )
				
				if shouldPauseImport then
					-- Stop sending more entries into the import pipeline
					importSubject:onCompleted()
					
					-- Save the error so we can send it back as the paused status
					errorEntries[ messageCode ] = errorEntries[ messageCode ] or {}
					table.insert( errorEntries[ messageCode ], { importedInfo.filePath, local_id } )
					
					return Observable.empty()
					
				else
					
					return self:_updateStatus( manifestStatusDatabase, local_id, nil, WFCatImportManifestConstants.status.failure )
						:defaultIfEmpty( nil )
						:tap( function()
							
							rowsQueue:sendOneMore()
						
						end )
					
				end
				
			end
		
		elseif importedInfo.assetId then
			-- AssetId can be present even if the asset didn't get imported
			-- this time. This can be a matching asset, which can be considered
			-- same as the current asset for all practical purposes.
			idCache:setOzIdForLocalId( WFCatImportManifestConstants.idCacheSets.assets, local_id, importedInfo.assetId )
			self:_updateSuccess( messageHandler, 1 )
			
			return self:_updateStatus( manifestStatusDatabase, local_id, importedInfo.assetId, WFCatImportManifestConstants.status.success )
			
				:defaultIfEmpty( nil )
					:tap( function()
						
						-- Now that one asset has been imported, we can send one more into the import pipeline
						rowsQueue:sendOneMore()
					
					end )
			
		end
		
		return Observable.empty()
	end
	
	
	local function handleCompleted()
		logger:debugf( "handleCompleted: Import streams completed." )
		
		if rowsQueue then
			rowsQueue:dispose()
			rowsQueue = nil
		end

		if next( errorEntries ) then
			local pauseInfo = {
				pauseType = WFCatImportManifestConstants.pauseType.resumeUserChoice,
				errorEntries = errorEntries,
				resumeContext = {
					importTimestamp = importTimestamp,
				}
			}
			
			error( pauseInfo )
		end
	end
	
		
	importObservable = importObservable
		:flatMap( maybeTranslateResult )
		:flatMap( handleImportResult )
		:tapOnCompleted( handleCompleted )
		:catch( function( err )
			
			logger:errorf( "Error in import: %s", err )
			
			if WFCatImportManifestProcessorUtils.isDocAlreadyExistsError( err ) then
				-- Docstore seems to have the entry, but when we checked it was not there
				-- Ask client to resume, so that we can pick the correct path
				-- next time.
				self:_sendMessage( messageHandler, WFCatImportManifestConstants.messages.assetProcessingRevisionExists.code, { err } )
				
				local pauseInfo = {
					pauseType = WFCatImportManifestConstants.pauseType.resumeAuto,
					resumeContext = {
						importTimestamp = importTimestamp,
					}
				}
				return Observable.throw( pauseInfo )
			end
			
			return Observable.throw( err )
		end )
		
	
	logger:debugf( "Creating WFCatImportManifestAssetsProcessor processor observable" )
	
	local sourceObservable = Observable.zip(
			WFCatImportManifestProcessorUtils.maybeReadLrCatStoreProviderId( manifestDatabase ),
			WFCatImportManifestProcessorUtils.maybeReadManifestSourceId( manifestDatabase )
		)
		:tap( function( lrcatId, sourceId )
			
			lrcatStoreProviderId = lrcatId
			manifestSourceId = sourceId

		end )
		:flatMap( function()
		
			return self:_maybeSetCount( manifestDatabase, "countAssets" )
			
		end )
		
		:flatMap( function()
			return self:_getLowestLocalId( manifestDatabase )
		end )
		
		:flatMap( function( lowestLocalId )
			lowestLocalId = lowestLocalId or 0
			lowestLocalId = lowestLocalId - 1
			
			rowsQueue = WFCatImportManifestSmartQueue(
				-- Next request
				function( start, numToFetch )
					local params = {
						min_id = start,
						limit = numToFetch,
					}
					
					return manifestDatabase:readData( 'selectAssets', params )
					end,

				-- First id
				lowestLocalId,

				-- Generate the next lowest id given the results of the previous data
				function( data )
				
					local lastRow = data[ #data ]
					return lastRow.local_id
				
				end
			)
			
			-- Initialize the queue
			for _ = 1, kMaxAssetsInFlight do
				rowsQueue:sendOneMore()
			end
			
			return rowsQueue:getObservable()

		end )
		:flatMap( function( row )
			
			if self:_isAlreadyProcessed( row.local_id ) then
				self:_updateAlreadyProcessed( messageHandler, 1 )
				rowsQueue:sendOneMore()
				
				return Observable.empty()
			
			elseif shouldIgnoreAsset( row.local_id ) then
				
				return ignoreAsset( row.local_id, row.filePath, kUserIgnoredAsset )
				
			else
			
				local validateSyncedObservable
				
				if row.ozAssetId then
					validateSyncedObservable = catalog:docStore():docValue( row.ozAssetId )
						:map( function( docValue )
							if docValue._deleted then
								-- Treat deleted assets same as non-synced
								-- assets.
								row.ozAssetId = nil
								return false
							end
							
							return true
						end )
				else
					validateSyncedObservable = Observable.just( false )
				end
				
				return validateSyncedObservable
					:flatMap( function( isSyncedAsset )
						
						local importInfo = self:_createImportInfo( row, lrcatStoreProviderId, manifestSourceId, importTimestamp )
						local importUUID = importInfo.uuid
						
						-- Just so that we dont update the status db with
						-- this fake uuid. If updated once, we will always
						-- try to use the same id and hence even in resume
						-- we wont be able to recover.
						if importUUID == fakeRevisionExistsErrorUUID then
							importUUID = KSIdGenerator.uuidInLowercaseHex( )
							fakeRevisionExistsErrorUUID = nil
						end
						
						-- Update the status and then send the task for processing
						return self:_addStatus( manifestStatusDatabase, row.local_id, importUUID, WFCatImportManifestConstants.status.pending )
							:defaultIfEmpty()
							:tap( function()
								
								importSubject:onNext( importInfo )
							
							end )
					end )
					
			end
			
		end )
		
		:tapOnCompleted( function()
			logger:debugf( "Source stream completed. Completing import subject." )
			importSubject:onCompleted()
		end )
	
	return importObservable
		:merge( sourceObservable )
		:ignoreElements()
		:defaultIfEmpty( nil )
		:flatMap( function()
		
			local queryId = "countAssetsStatus"
			local dataType = WFCatImportManifestConstants.dataTypes.assets
			return self:_sendSummary( messageHandler, manifestStatusDatabase, queryId, dataType )
			
		end )
		:ignoreElements()
			-- Adding summary step to the chain and then explicitly
			-- ignoring elements to denote that we are not interested
			-- in the return values.
	
end

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

return Rx.Class 'WFCatImportManifestAssetsProcessor'( WFCatImportManifestAssetsProcessor )
