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

WFCatAssets.lua

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

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

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

	-- wf.core.utils
local WFFunctionUtils = require "WFFunctionUtils"
local WFCloseSupport = require 'WFCloseSupport'

	-- wf.oz_client.managed_catalog.pixel_provider
local WFCatRenderCapabilities = require "WFCatRenderCapabilities"

	-- rxlua
local Rx = require 'Rx'

	-- WF.sqlite.database
local WFSQLiteUtils = require 'WFSQLiteUtils'

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

local WFFunctionUtils_reduce = assert( WFFunctionUtils.reduce,
			"WFFunctionUtils.reduce does not exist" )

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

local function addLocalAssetIdToSet( set, localAssetIdArray )

	local localAssetId = localAssetIdArray[ 1 ]

	if not localAssetId then

		return set

	end

	set[ localAssetId ] = true

	return set

end

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

local function convertLocalAssetIdTupleArrayToSet( localAssetIdTupleArrays )

	return WFFunctionUtils_reduce( {}, addLocalAssetIdToSet,
										localAssetIdTupleArrays )

end

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

local WFCatRenderCapabilities_renderableAssetSubtypesArray =
		require "WFCatRenderCapabilities".renderableAssetSubtypesArray

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

assert( type( WFCatRenderCapabilities_renderableAssetSubtypesArray ) == "table" and
				#WFCatRenderCapabilities_renderableAssetSubtypesArray > 0,
			"renderableAssetSubtypesArray must be non empty array" )

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

local _numRenderableSubtypes = #WFCatRenderCapabilities_renderableAssetSubtypesArray

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

local _renderableAssetIdsSQL =
	string.format( "SELECT docId FROM assets WHERE subtype IN ( %s )",
					WFSQLiteUtils.createVariableArgList( _numRenderableSubtypes ) )

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

local function getAllRenderableLocalAssetIdsAsSetStream( wfCatalog )

	local changeCounter = wfCatalog:docStore():changeCounter():take( 1 )

	local allRenderableLocalAssetIdsSet =
			Rx.Observable.create(

				function( observer )

					if WFCloseSupport.hasBeenOrWillBeClosed( wfCatalog ) or
							WFCloseSupport.hasBeenOrWillBeClosed( wfCatalog:indexStore() ) then

						return

					end

					return wfCatalog
							:indexStore()
							:queryOnce( { 

										sql = _renderableAssetIdsSQL,

										quiesceUntilUpToDate = true,

										unpack( WFCatRenderCapabilities_renderableAssetSubtypesArray ),

									} )
							:map( convertLocalAssetIdTupleArrayToSet )
							:subscribe( observer )

				end )

	return Rx.Observable.concat( changeCounter, allRenderableLocalAssetIdsSet )
				:toArray()
				:map( unpack )	

end

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

local function makeAccumulatedLocalAssetIdUpdateStream( ozCatalogId, stream,
															interval )

	local scheduler = require 'WFRxScheduler'.sharedScheduler()

	local adds = {}

	local deletes = {}

	local kIgnoredFirstResult = {}

	-- kIgnoredFirstResult is used to make sure that :calm doesn't emit until interval
	-- passes
	return Rx.Observable.concat( Rx.Observable.just( kIgnoredFirstResult ), stream )
			:map(
				function( updateType, localAssetId )

					if updateType == kIgnoredFirstResult then

						adds[ kIgnoredFirstResult ] = true

					elseif updateType == "add" then

						if not deletes[ localAssetId ] then

							adds[ localAssetId ] = true

						end

					elseif updateType == "delete" then

						adds[ localAssetId ] = nil

						deletes[ localAssetId ] = true

					end

				end )
			:calm( interval, scheduler )
			:map(
				function()

					local currAdds, currDeletes = adds, deletes

					adds, deletes = {}, {}

					currAdds[ kIgnoredFirstResult ] = nil

					return ozCatalogId, currAdds, currDeletes

				end )
			:where(
				function( _, currAdds, currDeletes )

					return next( currAdds ) ~= nil or next( currDeletes ) ~= nil

				end )

end

--------------------------------------------------------------------------------
-- maybeReportAllLocalAssetIdChange is called from
-- 	docStore:scannedFilteredMappedChangesFeedSince on a separate worker thread.
--------------------------------------------------------------------------------

local function maybeReportAllLocalAssetIdChange( allRenderableLocalAssetIds, docValue )

	if not WFCatRenderCapabilities.isRenderableAsset( docValue ) then

		return allRenderableLocalAssetIds

	end

	local localAssetId = docValue._localId

	if not localAssetId then

		return allRenderableLocalAssetIds

	end

	if docValue._deleted then

		allRenderableLocalAssetIds[ localAssetId ] = nil

		return allRenderableLocalAssetIds, "delete", localAssetId

	elseif not allRenderableLocalAssetIds[ localAssetId ] then

		allRenderableLocalAssetIds[ localAssetId ] = true

		return allRenderableLocalAssetIds, "add", localAssetId

	end

	return allRenderableLocalAssetIds

end

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

local function makeLocalAssetIdsDeltaStream( wfCatalog, ozCatalogId,
												changeCounter,
												allRenderableLocalAssetIdsSet,
												bufferInterval )

	local unbufferedLocalAssetIdUpdateStream = 
				wfCatalog
					:docStore()
					:scannedFilteredMappedChangesFeedSince( 
						"WFCatAssets", "maybeReportAllLocalAssetIdChange", 
						allRenderableLocalAssetIdsSet, changeCounter )

	return makeAccumulatedLocalAssetIdUpdateStream( ozCatalogId,
				unbufferedLocalAssetIdUpdateStream, bufferInterval )

end

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

local function allRenderableLocalAssetIdsStreamImp( wfCatalog, bufferInterval )

	bufferInterval = type( bufferInterval ) == "number" and 
						bufferInterval > 0 and bufferInterval or 0.5

	local ozCatalogId = wfCatalog:ozCatalogId()

	return getAllRenderableLocalAssetIdsAsSetStream( wfCatalog )
			:take( 1 )
			:flatMapLatest(
				function( changeCounter, allRenderableLocalAssetIdsSet )

					local initialDeltaStream =
							Rx.Observable.just( ozCatalogId,
												allRenderableLocalAssetIdsSet, {} )

					local ongoingLocalAssetIdsDeltaStream =
							makeLocalAssetIdsDeltaStream( 
								wfCatalog, ozCatalogId, changeCounter + 1,
								allRenderableLocalAssetIdsSet, bufferInterval )

					return Rx.Observable.concat( 
								initialDeltaStream,
								ongoingLocalAssetIdsDeltaStream )

				end )

end

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

local function pack( ... )

	return { ... }

end

--------------------------------------------------------------------------------
-- allRenderableLocalAssetIdsStream
--
-- returns a stream of lua tuples, 
--
-- 	{ ozCatalogId, addedLocalAssetIds, deletedLocalAssetIds }, 
-- 	{ ozCatalogId, addedLocalAssetIds, deletedLocalAssetIds }, 
-- 	...
--
-- 	The first pair will have
--
-- 		addedLocalAssetIds = set of all localAssetIds in catalog at
-- 		                     that point in time
--
-- 		deletedLocalAssetIds = {}
--
-- 	All subsequent pairs will be the deltas since the previous pair
-- 	was sent. i.e.
--
-- 		addedLocalAssetIds = set of all new localAssetIds since the
-- 		                     last pair
--
-- 		deletedLocalAssetIds = set of all removed localAssetIds
-- 		                       since the last pair
--------------------------------------------------------------------------------

local function allRenderableLocalAssetIdsStream( wfCatalog, bufferInterval )

	return Rx.Observable.create(

		function( observer )

			return allRenderableLocalAssetIdsStreamImp( wfCatalog, bufferInterval )
					:map( pack )
					:subscribe( observer )

		end )

end


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

return {

	allRenderableLocalAssetIdsStream = allRenderableLocalAssetIdsStream,

	maybeReportAllLocalAssetIdChange = maybeReportAllLocalAssetIdChange,

	privateUnitTestAPI = {

		getAllRenderableLocalAssetIdsAsSetStream = getAllRenderableLocalAssetIdsAsSetStream

	}

}
