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

PixelProviderExternalXmpDevelopPathCheckTask.lua

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

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

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

	-- Rx
local Rx = require "Rx"

	-- WF.core.utils
local WFFunctionUtils = require 'WFFunctionUtils'
local WFAsyncFileUtils = require "WFAsyncFileUtils"
local WFCrypto = require "WFCrypto"

	-- WF.oz_client.managed_catalog.pixel_provider
local WFCatAssetAux = require 'WFCatAssetAux'

	-- pixel_provider
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local WFCatPixelFileCacheCheck = require 'WFCatPixelFileCacheCheck'
local NmbDocValueToRenderInfo = require 'NmbDocValueToRenderInfo'
local PixelPathProviderSchedulerPrivateMethods =
		require "PixelPathProviderSchedulerPrivateMethods"
local PixelProviderSharedTaskUtils = require 'PixelProviderSharedTaskUtils'

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

local logger = require 'WFLogger'( 'PixelProviderLocalBinaryPathCheckTask' )

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

local function reduceT( acc, f, t, ... )

	for k, v in pairs( t ) do

		acc = f( acc, k, v, ... )

	end

	return acc

end

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

local function countT( t )

	return reduceT( 0, function( count ) return count + 1 end, t )

end

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

local _maxActiveFileChecks = 32

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

local function getFileCheckTransitionForRequest( 
				pixStreamId, localAssetId, fileCheckSubscrPerAssetId,
				baseRenderInfo )

	if fileCheckSubscrPerAssetId[ localAssetId ] then

		-- already checking, join it
		return 'start_local_develop_files_check'
		
	end
	
	if type( baseRenderInfo ) ~= "table" then

		logger:warnf( "Can't start file check for local external xmp develop %s.%s: no baseRenderInfo",
						pixStreamId, localAssetId )

		return 'missing_external_develop_file_docstore_info'

	end

	if type( baseRenderInfo.externalDevelopFiles ) ~= "table" or
			#baseRenderInfo.externalDevelopFiles == 0 then

		logger:warnf( "Can't start file check for local external xmp develop %s.%s: missing_expected_external_develop_file_info",
						pixStreamId, localAssetId )

		return 'missing_expected_external_develop_file_info'

	end

	return 'start_local_develop_files_check'

end

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

local function getFilePathsForFileType( fileSpec, baseAuxFilePath, fileType, uuid )

	if fileType == "develop" then

		return fileSpec:pathForSettingsXmp( uuid )

	end

	if fileType == "aux" then

		return WFCatAssetAux.calculatePathUsingDigest( 
					baseAuxFilePath, uuid )

	end

	error( string.format( "don't know how to make file paths for %s",
							fileType ) )

end

--------------------------------------------------------------------------------
-- makeExternalDevelopFileCheckResultStream emits exactly one tuple of form:
--
-- 	fileExistsResult, fileType, uuid, uri, filePath

local function makeExternalDevelopFileCheckResultStream( 
				externalDevelopFileInfo, fileChecker, baseAuxFilePath,
				fileSpec )

	local fileType, uuid, uri = unpack( externalDevelopFileInfo )

	local filePath, relativePath =
			getFilePathsForFileType( fileSpec, baseAuxFilePath, fileType, uuid )

	if not filePath then

		return Rx.Observable.throw( 
				string.format( "Don't know how to make localDevelopPath for type %s uuid %s",
								fileType, uuid ) )

	end

	local fileCheckResultStream =
			fileChecker
				:checkPath( filePath )
				:startWith( false )

	return Rx.Observable.combineLatest(
				fileCheckResultStream, 
				Rx.Observable.just( fileType ), 
				Rx.Observable.just( uuid ), 
				Rx.Observable.just( uri ), 
				Rx.Observable.just( filePath ),
				Rx.Observable.just( relativePath ) )
			:takeLast( 1 )

end

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

local function gatherFileStatuses( fileStatuses, exists, fileType, uuid,
										uri, filePath, relativePath )

	if fileType == 'develop' then

		fileStatuses.xmpDevelopPath = filePath

		fileStatuses.xmpDevelopPathIsMissing = not exists

		fileStatuses.xmpDevelopCanBeDownloaded = uri ~= nil

		return fileStatuses

	end

	if fileType == 'aux' and not exists then

		fileStatuses.missingDevelopAuxFiles = fileStatuses.missingDevelopAuxFiles or {}

		fileStatuses.missingDevelopAuxFiles[ #fileStatuses.missingDevelopAuxFiles + 1 ] = 
			{ uuid, uri, filePath, relativePath }

	end

	return fileStatuses

end

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

local function makeFileChecksResultStream( externalDevelopFiles, fileChecker,
											baseAuxFilePath, fileSpec )

	local externalDevelopFileCheckResultStreams = 
		WFFunctionUtils.map(
			makeExternalDevelopFileCheckResultStream,
				externalDevelopFiles, fileChecker, baseAuxFilePath, 
				fileSpec )

	-- allExternalDevelopFileCheckResultsStream fires off all the request at
	-- once and combines the results into a stream of tuples
	--
	-- 	fileExistsResult1, fileType1, uuid1, uri1, filePath1, relFilePath1
	-- 	fileExistsResult2, fileType2, uuid2, uri2, filePath2, relFilePath2
	-- 	...
	-- 	fileExistsResultN, fileTypeN, uuidN, uriN, filePathN, relFilePathN
	--
	local allExternalDevelopFileCheckResultsStream =
			Rx.Observable.from( externalDevelopFileCheckResultStreams )
				:flatMergeAll()

	-- :scan() plus :takeLast( 1 ) waits for all of the
	-- allExternalDevelopFileCheckResultsStream tuples and then emits a
	-- single combined missing file list

	return allExternalDevelopFileCheckResultsStream
			:scan( gatherFileStatuses, {} )
			:takeLast( 1 )

end

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

local PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId =
		assert( PixelPathProviderSchedulerPrivateMethods.maybeApplyTransitionForLocalAssetId,
				"missing PixelPathProviderSchedulerPrivateMethods.maybeApplyTransitionForLocalAssetId" )

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

local _checkingIfExternalXmpDevelopIsLocalStates = {
		PixelPathProviderStateMachine.LocalBinaryPathCheckStates.CheckingIfExternalXmpDevelopIsLocal,
		PixelPathProviderStateMachine.LocalBinaryPathCheckStates.RenditionStrategyCheckingIfExternalXmpDevelopIsLocal,
	}

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

local function getFileCheckResultTransitionStream( 
				scheduler, localAssetId, origBaseRenderInfo, baseRenderInfoCache, localBinaryPathCache,
				needDevelopAuxFileDownloadCache, fileChecker, baseAuxFilePath,
				fileSpec, docStore, makeAsyncFileSizeStream, makeAsyncFileDigestStream )

	local externalDevelopFileCheckStream =
			makeFileChecksResultStream( 
				origBaseRenderInfo.externalDevelopFiles, fileChecker,
				baseAuxFilePath, fileSpec )

	return externalDevelopFileCheckStream
			:flatMap( 
				function( fileStatuses )

					local currBaseRenderInfo = 
							baseRenderInfoCache:peekCacheItem( localAssetId )

					if not currBaseRenderInfo then

						return Rx.Observable.just( 'need_new_xmp_develop_info_from_docstore' )

					end

					if not NmbDocValueToRenderInfo.areExternalDevelopFilesEqual(
							origBaseRenderInfo.externalDevelopFiles, 
							currBaseRenderInfo.externalDevelopFiles ) then

						return Rx.Observable.just( 'redo_xmp_develop_check' )

					end

					if not fileStatuses.xmpDevelopPathIsMissing then

						-- add the xmp develop path to localBinaryPathCache for all
						-- requests that are checking for it.

						local requestsCheckingLocalXmpDevelopPath =
								scheduler:_getLocalAssetIdRequestsWithState( 
									localAssetId, 
									unpack( _checkingIfExternalXmpDevelopIsLocalStates ) )

						WFFunctionUtils.map(
							function( request )

								local pixStreamId, theLocalAssetId = unpack( request )

								localBinaryPathCache
									:pushCacheItem( 
										pixStreamId, theLocalAssetId,
										{ fileStatuses.xmpDevelopPath, "xmp_develop" } )

							end, requestsCheckingLocalXmpDevelopPath )
						
					end

					-- check if we've untracked existing xmp_develop
					local haveUntrackedLocalXmpDevelopPath =
							not fileStatuses.xmpDevelopPathIsMissing and
								origBaseRenderInfo.localXmpDevelopPathFromDocstore ~= fileStatuses.xmpDevelopPath

					-- define docStoreUpdateCompletionStream which will run
					-- if haveUntrackedLocalXmpDevelopPath
					local rawDocStoreUpdateStream =
							Rx.Observable.combineLatest(
									makeAsyncFileSizeStream( fileStatuses.xmpDevelopPath ),
									makeAsyncFileDigestStream( fileStatuses.xmpDevelopPath, 'SHA256' ) )
								:flatMap(
									function( fileSize, sha256 )

										return PixelProviderSharedTaskUtils.updateDocStoreXmpDevelopInfoStream(
												docStore, localAssetId,
												fileStatuses.xmpDevelopPath,
												fileSize, sha256 )

									end )

					local docStoreUpdateCompletionStream =
							rawDocStoreUpdateStream
								-- emit no results
								:takeLast( 0 )
								:catch(
									function( errStr )

										-- log, but do not propagate exceptions
										-- updating the docstore
										logger:warnf( "Error updating docstore xmp develop for %s at %s: %s",
														localAssetId, fileStatuses.xmpDevelopPath, errStr )

										-- emit nothing
										return Rx.Observable.empty()

									end )

					-- set up a stream that either updates the docstore or does
					-- nothing
					local maybeUpdateDocStoreCompletionStream =
						haveUntrackedLocalXmpDevelopPath and
							docStoreUpdateCompletionStream or
							-- no-op stream
							Rx.Observable.empty()

					-- from this point on, run
					-- maybeUpdateDocStoreCompletionStream before returning the
					-- transition
					if not fileStatuses.xmpDevelopPathIsMissing and
							not fileStatuses.missingDevelopAuxFiles then

						return Rx.Observable.concat( 
									maybeUpdateDocStoreCompletionStream, 
									Rx.Observable.just( 'has_all_local_develop_files' ) )

					end

					if fileStatuses.xmpDevelopPathIsMissing and
							fileStatuses.xmpDevelopCanBeDownloaded then
					
						return Rx.Observable.concat( 
								maybeUpdateDocStoreCompletionStream, 
								Rx.Observable.just( 'need_download_external_develop_xmp' ) )

					end

					if fileStatuses.xmpDevelopPathIsMissing and
							not fileStatuses.xmpDevelopCanBeDownloaded then

						return Rx.Observable.concat( 
								maybeUpdateDocStoreCompletionStream, 
								Rx.Observable.just( 'missing_needed_develop_xmp_link' ) )

					end

					needDevelopAuxFileDownloadCache
						:pushCacheItem( 
							localAssetId,
							{

								externalDevelopFiles = 
									currBaseRenderInfo.externalDevelopFiles,

								missingDevelopAuxFiles = fileStatuses.missingDevelopAuxFiles,

							} )

					return Rx.Observable.concat( 
							maybeUpdateDocStoreCompletionStream, 
							Rx.Observable.just( 'need_download_one_or_more_external_aux_files' ) )

				end )

end

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

local function startFileCheckForRequest( 
				fileCheckSubscrPerAssetId, request, scheduler,
				baseRenderInfoCache, localBinaryPathCache, needDevelopAuxFileDownloadCache, fileChecker,
				baseAuxFilePath, fileSpec, docStore, makeAsyncFileSizeStream,
				makeAsyncFileDigestStream )

	local pixStreamId, localAssetId = unpack( request )
	
	local origBaseRenderInfo = baseRenderInfoCache:peekCacheItem( localAssetId )

	local transition = 
			getFileCheckTransitionForRequest( 
				pixStreamId, localAssetId, fileCheckSubscrPerAssetId,
				origBaseRenderInfo )

	if transition == "start_local_develop_files_check" and 
			not fileCheckSubscrPerAssetId[ localAssetId ] then

		fileCheckSubscrPerAssetId[ localAssetId ] =
			getFileCheckResultTransitionStream( 
					scheduler, localAssetId, origBaseRenderInfo, baseRenderInfoCache, localBinaryPathCache,
					needDevelopAuxFileDownloadCache, fileChecker, baseAuxFilePath,
					fileSpec, docStore, makeAsyncFileSizeStream, makeAsyncFileDigestStream )
			:subscribe(
				function( fileCheckResultTransition )

					PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId(
						scheduler, localAssetId, fileCheckResultTransition )

				end,
				function( errStr )

					logger:warnf( "unexpected exception checking external develop file existence for %s: %s",
									localAssetId, errStr )

					PixelPathProviderSchedulerPrivateMethods_maybeApplyTransitionForLocalAssetId(
						scheduler, localAssetId, 
						'unexpected_external_develop_file_check_exception' )

					fileCheckSubscrPerAssetId[ localAssetId ] = nil
					
				end,
				function()

					fileCheckSubscrPerAssetId[ localAssetId ] = nil

				end)
			
	end

	scheduler:_applyTransitionToRequest( request, transition )

	return fileCheckSubscrPerAssetId

end

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

local function maybeJoinCurrentSyncTasks( scheduler, syncingStates, 
											needsCheckStates, joinTransition )

	-- first gather all the localAssetIds that are in one of the syncingStates
	local localAssetIdsWithXmpDevelopSyncTasks =
			PixelPathProviderSchedulerPrivateMethods.reduceAllRequestsWithStates( 
				scheduler, {}, syncingStates,
				function( localAssetIds, request )

					local _, localAssetId = unpack( request )

					localAssetIds[ localAssetId ] = true

					return localAssetIds

				end )

	-- Now go through all requests with the
	-- _needCheckIfExternalXmpDevelopIsLocalStates and if any of those also
	-- we seen above, apply the joinTransition
	PixelPathProviderSchedulerPrivateMethods.reduceAllRequestsWithStates( 
		scheduler, localAssetIdsWithXmpDevelopSyncTasks, needsCheckStates,
		function( localAssetIdsWithSyncTask, request )

			local _, localAssetId = unpack( request )

			if localAssetIdsWithSyncTask[ localAssetId ] then

				scheduler
					:_applyTransitionToRequest( 
						request, joinTransition )

			end

			return localAssetIdsWithSyncTask

		end )

end

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

local _externalXmpDevelopIsDownloadingStates = {
		PixelPathProviderStateMachine.DownloadStates.NeedXmpDevelopDownload,
		PixelPathProviderStateMachine.DownloadStates.RenderNeedXmpDevelopDownload,
		PixelPathProviderStateMachine.DownloadStates.DownloadingXmpDevelop,
		PixelPathProviderStateMachine.DownloadStates.RenderDownloadingXmpDevelop,
	}

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

local _needCheckIfExternalXmpDevelopIsLocalStates = {
		PixelPathProviderStateMachine.LocalBinaryPathCheckStates.NeedCheckIfExternalXmpDevelopIsLocal,
		PixelPathProviderStateMachine.LocalBinaryPathCheckStates.RenditionStrategyNeedCheckIfExternalXmpDevelopIsLocal,
	}

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

local function maybeJoinCurrentExternalXmpDevelopDownloadTasks( scheduler )

	-- Join all requests that need to check external xmp develop status with
	-- any same localAssetId requests that are currently downloading xmp
	-- develop
	maybeJoinCurrentSyncTasks( 
		scheduler, _externalXmpDevelopIsDownloadingStates,
		_needCheckIfExternalXmpDevelopIsLocalStates,
		'join_existing_xmp_develop_download' ) 

end

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

local _externalXmpDevelopIsSavingStates = {
		PixelPathProviderStateMachine.SaveDownloadStates.NeedToSaveXmpDevelop,
		PixelPathProviderStateMachine.SaveDownloadStates.SavingXmpDevelop,
		PixelPathProviderStateMachine.SaveDownloadStates.RenderNeedToSaveXmpDevelop,
		PixelPathProviderStateMachine.SaveDownloadStates.RenderSavingXmpDevelop,
	}

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

local function maybeJoinCurrentExternalXmpDevelopSaveTasks( scheduler )

	-- Join all requests that need to check external xmp develop status with
	-- any same localAssetId requests that are currently saving xmp
	-- develop
	--
	-- note: it's fine to put the state in need save vs saving, the
	-- save task ensures ensures all requests with same local asset id are
	-- joined later
	maybeJoinCurrentSyncTasks( 
		scheduler, _externalXmpDevelopIsSavingStates,
		_needCheckIfExternalXmpDevelopIsLocalStates,
		'join_existing_xmp_develop_save' ) 

end

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

local _externalDevelopAuxFilesAreDownloadingStates = {
		PixelPathProviderStateMachine.DownloadStates.NeedExternalDevelopAuxFilesDownload,
		PixelPathProviderStateMachine.DownloadStates.RenderNeedExternalDevelopAuxFilesDownload,
		PixelPathProviderStateMachine.DownloadStates.DownloadingExternalAuxFiles,
		PixelPathProviderStateMachine.DownloadStates.RenderDownloadingExternalAuxFiles,
	}

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

local _needCheckIfExternalAuxFilesLocalStates = {
		PixelPathProviderStateMachine.LocalBinaryPathCheckStates.NeedCheckIfExternalDevelopAuxFilesAreLocal,
		PixelPathProviderStateMachine.LocalBinaryPathCheckStates.RenditionStrategyNeedCheckIfExternalDevelopAuxFilesAreLocal,
	}

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

local function maybeJoinCurrentExternalDevelopAuxDownloadTasks( scheduler )

	-- Join all requests that need to check external aux files with all
	-- localAssetId requests that are currently downloading external aux files
	--
	-- note: it's fine to put the state in download vs downloading, the
	-- download task ensures ensures all requests with same local asset id are
	-- joined later
	maybeJoinCurrentSyncTasks( 
		scheduler, _externalDevelopAuxFilesAreDownloadingStates,
		_needCheckIfExternalAuxFilesLocalStates,
		'join_existing_external_aux_file_download' ) 

end

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

local function prepareForExternalDevelopAuxFileChecks( scheduler, baseRenderInfoCache,
									localBinaryPathCache )

	-- we are looping through all requests that might need external aux file
	-- checks. Requests get in this state after freshly downloading an
	-- xmpDevelop file
	scheduler:_mapAllRequestsWithStates( _needCheckIfExternalAuxFilesLocalStates,
		function( request )

			local pixStreamId, localAssetId = unpack( request )

			local baseRenderInfo = 
					baseRenderInfoCache:peekCacheItem( localAssetId )

			if not baseRenderInfo then

				scheduler
					:_applyTransitionToRequest( 
						request, 'missing_expected_external_develop_file_info' )

				return

			end

			local localXmpDevelopPath =
				unpack( localBinaryPathCache
							:peekCacheItem( pixStreamId, localAssetId )  or {} )

			if not localXmpDevelopPath then

				logger:warnf( "when starting external aux file check for %s, missing local xmp develop path",
								localAssetId )

				-- this is unexpected and must be treated at error, we
				-- shouldn't be doing the external aux file check until we have a
				-- localBinaryPathCache item to send back to the caller
				--
				-- 	The only recourse is transition to error state
				scheduler
					:_applyTransitionToRequest( 
						request, 'missing_expected_external_develop_file_info' )

				return

			end

			-- at this point we know we have a valid local xmp develop path, now
			-- we decide if we need to check for local external aux files
			if not baseRenderInfo.hasDevelopAuxFiles then

				scheduler
					:_applyTransitionToRequest( request, 'skip_aux_file_check' )

				return

			end

		end )

end

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

local function appendT( ret, t )

	local nRet = #ret

	for i, v in ipairs( t ) do

		ret[ nRet + i ] = v

	end

	return ret

end

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

local _needCheckIfDevelopFilesLocalStates =
	WFFunctionUtils.reduce( {},
		appendT,
		{
			_needCheckIfExternalXmpDevelopIsLocalStates,
			_needCheckIfExternalAuxFilesLocalStates,
		} )

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

local function processRequests( scheduler, taskState, baseRenderInfoCache,
									localBinaryPathCache,
									needDevelopAuxFileDownloadCache )

	maybeJoinCurrentExternalXmpDevelopDownloadTasks( scheduler )

	maybeJoinCurrentExternalXmpDevelopSaveTasks( scheduler )

	maybeJoinCurrentExternalDevelopAuxDownloadTasks( scheduler )

	prepareForExternalDevelopAuxFileChecks( scheduler, baseRenderInfoCache,
									localBinaryPathCache )

	-- now start file checks as needed
	repeat 

		local unusedFileChecks =
			_maxActiveFileChecks - countT( taskState.fileCheckSubscrPerAssetId )
					
		if unusedFileChecks <= 0 then

			return

		end
		
		local requests =
				PixelPathProviderSchedulerPrivateMethods.getNextNRequestsWithState( 
					scheduler, unusedFileChecks, unpack( _needCheckIfDevelopFilesLocalStates ) )

		if not next( requests ) then

			return

		end

		taskState.fileCheckSubscrPerAssetId =
			WFFunctionUtils.reduce(
				taskState.fileCheckSubscrPerAssetId,
					startFileCheckForRequest,
						requests, scheduler, baseRenderInfoCache, localBinaryPathCache,
						needDevelopAuxFileDownloadCache, taskState.fileChecker,
						taskState.baseAuxFilePath, taskState.fileSpec,
						taskState.docStore, taskState.makeAsyncFileSizeStream,
						taskState.makeAsyncFileDigestStream )

	until false
	
end

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

local function setupTask( pixelsPath, wfCatalog, docStore, fileSpec )

	return {
		
		fileChecker = WFCatPixelFileCacheCheck( pixelsPath ),

		baseAuxFilePath = WFCatAssetAux.getBasePath( wfCatalog ),

		docStore = docStore,

		fileSpec = fileSpec,

		fileCheckSubscrPerAssetId = {},
		
		-- making these functions part of the task state allows tests to
		-- override them
		makeAsyncFileSizeStream = WFAsyncFileUtils.calculatePathSize,

		makeAsyncFileDigestStream = WFCrypto.asyncDigestFile,
	}

end

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

local function cleanup( taskState )

	for _, subscr in pairs( taskState.fileCheckSubscrPerAssetId ) do

		subscr:dispose()

	end

	taskState.fileCheckSubscrPerAssetId = {}

end

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

return {

	processRequests = processRequests,

	setupTask = setupTask,

	cleanup = cleanup,

	privateUnitTestExport = {

		needCheckIfExternalXmpDevelopIsLocalStates =
			_needCheckIfExternalXmpDevelopIsLocalStates,

		externalXmpDevelopIsDownloadingStates = 
			_externalXmpDevelopIsDownloadingStates,

		externalXmpDevelopIsSavingStates = 
			_externalXmpDevelopIsSavingStates,

		needCheckIfExternalAuxFilesLocalStates =
			_needCheckIfExternalAuxFilesLocalStates, 

		externalDevelopAuxFilesAreDownloadingStates = _externalDevelopAuxFilesAreDownloadingStates,

	}

}
