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

WFCatImportManifestStacksProcessor.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 ReplaySubject = Rx.ReplaySubject

local WFConfiguration = require "WFConfiguration"

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

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

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

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

local function addStackAssetStatus( manifestStatusDatabase, stackAssets, status )
	
	local statusData = {}
	
	for _, stackAssetId in ipairs( stackAssets ) do
		
		table.insert( statusData, {
			local_id = stackAssetId,
			status = status,
		} )
	end
	
	return manifestStatusDatabase:writeData( 'addOrReplaceStackAssetStatus', statusData )
		:ignoreElements()
end

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

local function updateStackStatus( manifestStatusDatabase, stackId, ozId, status )
	
	local statusData = {
		local_id = stackId,
		status = status,
	}
	
	return manifestStatusDatabase:writeData( 'updateStackStatus', { statusData } )
		:ignoreElements()
end

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

-- We add pending status only to the stack, not to the stack-assets
-- Once we are done with stack and stack-assets processing, we will
-- add/update the status to both the tables. Once a stack is processed
-- we can be sure that all its associated stack-assets are processed
-- since the operation is atomic.

local function addStackStatus( manifestStatusDatabase, stackId, uuid, status )
	
	local statusData = {
		local_id = stackId,
		oz_id = uuid,
		status = status,
	}
	
	return manifestStatusDatabase:writeData( 'addOrReplaceStackStatus', { statusData } )
		:ignoreElements()
end

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

local function addStackAndStackAssetStatus( manifestStatusDatabase, 
												stackId, stackOzId, stackAssetIds, status )
	
	return addStackAssetStatus( manifestStatusDatabase, stackAssetIds, status )
	
		:defaultIfEmpty( nil )
		:flatMap( function()
			
			-- Update the status of the stack only after
			-- stack-asset update since we are assuming
			-- stack-asset as done, if stack status is
			-- updated.
			return addStackStatus( manifestStatusDatabase, stackId, stackOzId, status )
		end )
end

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

local function updateStackAndStackAssetStatus( manifestStatusDatabase, 
												stackId, stackOzId, stackAssetIds, status )
	
	return addStackAssetStatus( manifestStatusDatabase, stackAssetIds, status )
	
		:defaultIfEmpty( nil )
		:flatMap( function()
			
			-- Update the status of the stack only after
			-- stack-asset update since we are assuming
			-- stack-asset as done, if stack status is
			-- updated.
			return updateStackStatus( manifestStatusDatabase, stackId, stackOzId, status )
		end )
end


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

local WFCatImportManifestStacksProcessor = {
	
	WFCatImportManifestProcessorBase,
}

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

function WFCatImportManifestStacksProcessor:__init()
	self._stageName = WFCatImportManifestConstants.stageNames.stacks
	
	self._messageCodeCompleted = WFCatImportManifestConstants.messages.stacksProcessingComplete.code
	self._messageCodeErrorCount = WFCatImportManifestConstants.messages.stacksProcessingErrorCount.code
	self._messageCodeGenericError = WFCatImportManifestConstants.messages.stacksProcessingError.code
end

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

function WFCatImportManifestStacksProcessor:setup( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )
	
	-- Though we are storing both Stack status as well as Stack-asset status, 
	-- we will be using only the stack status ( for now ), to figure out
	-- if we have processed since stack-asset is atomically processed along 
	-- with stack.
	-- But stack-asset entries need to be populated for generating summary-info
	-- and for analyzing asset-missing entries.
	
	return self:preProcessPending( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )
		:takeLast( 1 )
		:flatMap( function()
			
			return self:WFCatImportManifestProcessorBase_setup( catalog, 
				manifestStatusDatabase, idCache, messageHandler, 'selectStackStatus', 
				WFCatImportManifestConstants.idCacheSets.stacks )
		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 WFCatImportManifestStacksProcessor:preProcessPending( catalog, manifestDatabase, manifestStatusDatabase, idCache, messageHandler )
	
	return manifestStatusDatabase:readData( "selectStackStatusWithStatus" , { 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
			
			if #docIds == 0 then
				return Observable.just()
			end
		
			return catalog:docStore():bulkDocValue( docIds )
			
				:flatMap( function( docValues )
		
					local statusListToUpdate = {}
					
					for i, docValue in ipairs( docValues ) do
					
						if not docValue.empty then
							-- Document exists. 
							-- Mark the status as done.
							
							local statusInfo = statusForDocId[ docValue._id ]
							
							statusListToUpdate[ #statusListToUpdate + 1 ] = {
								local_id = statusInfo.local_id,
								oz_id = statusInfo.oz_id,
							}

						end
					end
					
					if #statusListToUpdate > 0 then
						-- Update the status for each stack and 
						-- update the status of all the stack assets
						-- if the status is not updated for them yet.
						-- ( If the asset is missing, we will update
						-- the stack-asset status directly, before
						-- creation of the stack. In those cases,
						-- it is possible to have stack-asset status
						-- entries to be present )
						-- The stack-asset update query will ignore 
						-- the rows which have been already updated.
						
						-- We are chaining observables below and it
						-- can become really long if not broken in between.
						-- The below chain gets broken at "readData"
						-- since the read happens on a different thread.
						
						local updateObs = Observable.from( statusListToUpdate )
							:flatMap( function( stackStatus )
							
								return manifestDatabase:readData( "selectStackAssetsForStack", 
										{ stack = stackStatus.local_id} )
							
									:flatMap( function( rows )
									
										if #rows == 0 then
											return Observable.just()
										end
										
										local stackAssetIds = {}
										for i, row in ipairs( rows ) do
											stackAssetIds[ i ] = row.local_id
										end
										
										return updateStackAndStackAssetStatus( manifestStatusDatabase, 
											stackStatus.local_id, stackStatus.oz_id, 
											stackAssetIds, WFCatImportManifestConstants.status.success )
											
											:defaultIfEmpty()
												
									end )	
								
							end )
						
						return updateObs
						
					else
						return Observable.just()
					end
					
				end )
		end )

end

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

function WFCatImportManifestStacksProcessor:process( ... )

	if WFConfiguration.Manifest.disableStacks then
		local retSubject = ReplaySubject( 1 )
		retSubject:onNext( true )
		retSubject:onCompleted()
		return retSubject
	end
	
	return self:WFCatImportManifestProcessorBase_process( ... )
end

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

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

	local stackInfoMap = {}
	
	return self:_maybeSetCount( manifestDatabase, 'countStacks' )
	
		:flatMap( function()
		
			return manifestDatabase:readData( 'selectStacks' )
			
		end )
		
		:flatMap( function( rows ) 
		
			return Observable.create( function( observer ) 
				
				for i, row in ipairs( rows ) do
					observer:onNext( row )
				end
				
				observer:onCompleted()
				
			end )
		end )
		
		:map( function( row ) 
		
			local stackInfo = stackInfoMap[ row.stack_id ] or {}
			stackInfoMap[ row.stack_id ] = stackInfo
			
			stackInfo.cover = row.cover
			stackInfo.stack_assets = stackInfo.stack_assets or {}
			table.insert( stackInfo.stack_assets, { asset = row.asset, stack_asset_id = row.stack_asset_id } )
			
			return stackInfoMap
			
		end )

		:takeLast( 1 )
		
		:flatMap( function( stackInfoMap )
		
			return Observable.create( function( observer )
				
				for stackId, stackInfo in pairs( stackInfoMap ) do
					observer:onNext( stackId, stackInfo )
				end
				
				observer:onCompleted()
				
			end )
		end )
		
		:flatMap( function( stackId, stackInfo )
		
			if self:_isAlreadyProcessed( stackId ) then
				
				-- The stack has been processed already. Skip.
				self:_updateAlreadyProcessed( messageHandler, 1 )
				return Observable.empty()
				
			end
			
			local pendingUUID = self._processedRows[ stackId ] and self._processedRows[ stackId ].oz_id
			local uuid = pendingUUID or KSIdGenerator.uuidInLowercaseHex( )
						
			
			local coverOzId = idCache:getOzIdForLocalId( WFCatImportManifestConstants.idCacheSets.assets, stackInfo.cover )
			
			if not coverOzId then
				
				-- Send a message, but continue adding the stack
				self:_sendMessage(
					messageHandler,
					WFCatImportManifestConstants.messages.stacksProcessingMissingCoverError.code,
					{ { stackId, stackInfo.cover } }
				)
			end
			
			local assetIds = {}
			local inProcessStackAssetIds = {}
			local skippedStackAssets = {}
			
			for i, stackAssetInfo in pairs( stackInfo.stack_assets ) do
				local assetOzId = idCache:getOzIdForLocalId( WFCatImportManifestConstants.idCacheSets.assets, stackAssetInfo.asset )
			
				if not assetOzId then
				
					-- Mark for updating status later
					skippedStackAssets[ #skippedStackAssets + 1 ] = stackAssetInfo.stack_asset_id
					
					-- Send a message about stack_asset addition error
					self:_sendMessage(
					messageHandler,
						WFCatImportManifestConstants.messages.stackProcessingMissingAssetError.code,
						{ { stackAssetInfo.stack_asset_id, stackAssetInfo.asset } }
					)
				else
					table.insert( assetIds, assetOzId )
					table.insert( inProcessStackAssetIds, stackAssetInfo.stack_asset_id )
				end
			end
			
			local skippedEntriesUpdateObs
			if #skippedStackAssets > 0 then
				skippedEntriesUpdateObs = addStackAssetStatus( manifestStatusDatabase, skippedStackAssets, WFCatImportManifestConstants.status.failure )
			else
				skippedEntriesUpdateObs = Observable.empty()
			end
			
			local processObservable
			
			if #assetIds < 2 then
				
				-- Stack has only 0 or 1 element.
				-- Invalid stack. Report and exit this stack creation.
				
				self:_updateError( messageHandler, 1 )
				self:_sendMessage(
					messageHandler,
					WFCatImportManifestConstants.messages.stacksProcessingInsufficientElements.code,
					{ { stackId } }
				)
				
				-- Update the status of the stack
				processObservable = addStackAndStackAssetStatus( manifestStatusDatabase, 
												stackId, nil, inProcessStackAssetIds, 
												WFCatImportManifestConstants.status.failure )

			else
				-- The stack seems to be valid. Add it.
				processObservable = addStackStatus( manifestStatusDatabase, stackId, uuid, WFCatImportManifestConstants.status.pending )
					:defaultIfEmpty()
					
					:flatMap( function()
						return WFCatStacks.createStackWithId( catalog, uuid, coverOzId, assetIds )
					end )
					
					:flatMap( function(  )
					
						idCache:setOzIdForLocalId( WFCatImportManifestConstants.idCacheSets.stacks, stackId, uuid )
						
						self:_updateSuccess( messageHandler, 1 )
						
						return updateStackAndStackAssetStatus( manifestStatusDatabase, 
												stackId, uuid, inProcessStackAssetIds, 
												WFCatImportManifestConstants.status.success )
						
					end )
					
					:catch( function( err )

						self:_updateError( messageHandler, 1 )
						self:_sendMessage(
							messageHandler,
							WFCatImportManifestConstants.messages.stacksProcessingDocError.code,
							{ { stackId, err } }
						)
						
						return updateStackAndStackAssetStatus( manifestStatusDatabase, 
												stackId, uuid, inProcessStackAssetIds, 
												WFCatImportManifestConstants.status.failure )

					end )
			end
			
			return skippedEntriesUpdateObs
				:defaultIfEmpty( nil )
				:flatMap( function()
					
					return processObservable
				
				end )
		end )
		:ignoreElements()
		:defaultIfEmpty( nil )
		:flatMap( function()
		
			local queryId = "countStacksStatus"
			local dataType = WFCatImportManifestConstants.dataTypes.stacks
			
			return self:_sendSummary( messageHandler, manifestStatusDatabase, queryId, dataType )
				:ignoreElements()
				:defaultIfEmpty( nil )
				:flatMap( function()
					
					queryId = "countStackAssetsStatus"
					dataType = WFCatImportManifestConstants.dataTypes.stackAssets
					return self:_sendSummary( messageHandler, manifestStatusDatabase, queryId, dataType )
				end )
			
		end )
		:ignoreElements()
			-- Adding summary step to the chain and then explicitely
			-- ignoring elements to denote that we are not interested
			-- in the return values.
end

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

return Rx.Class 'WFCatImportManifestStacksProcessor'( WFCatImportManifestStacksProcessor )
