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

WFRxIndexStoreImpUtils.lua

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

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

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

local Rx = require 'Rx'
local KSDate = require 'KSDate'
local KSStringUtils = require 'KSStringUtils'
local KSTableUtils = require 'KSTableUtils'
local WFCloseSupport = require 'WFCloseSupport'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.rxlua
local WFRxControlledObservableWithBuffer = require 'WFRxControlledObservableWithBuffer'
local WFRxDisposeOnUnsubscribeObservable = require 'WFRxDisposeOnUnsubscribeObservable'

local controlledWithBuffer = WFRxControlledObservableWithBuffer
local disposeOnUnsubscribe = WFRxDisposeOnUnsubscribeObservable

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

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

local check = assert( WFArguments.check )

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

local WFRxIndexStoreImpUtils = {}

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

--[[
	\brief This function will run a query function based on the updates emitted by the \c observable.
	However, it attempts to prevent this query from running 'too often'. This may naturally happen
	when a table is updated constantly or when a query runs slowly or some combination of those two
	conditions. In these cases we don't want the query to run every time it 'could' because it could
	significantly back up the index store and prevent access by other queries.

	\param observable (Observable) Observable that fires each time the query fn should fire (subject to
		additional throttling by this function
	\param timeRatio (number) This is a ratio of ~ how often a query should run relative to how long it
		takes to run the query. e.g. if a query takes 1s to run and the \c timeRatio is 4, then the query
		won't be run again for another 4s.
	\param queryFn (function) The function to run when the throttle conditions are met
--]]
local function throttledQuery( observable, timeRatio, queryFn )

	local getTimeNow = KSDate.timeNowForAnimation

	local lastDelayCount = 0
	local lastDelayCountProcessed = 0

	local noSoonerSb = Rx.Subject()

	local noSooner = noSoonerSb
		:replay( nil, 1 )
	noSooner:connect()

	noSoonerSb:onNext( getTimeNow() )

	-- noSooner is an Observable that handles onNext events containing a particular time
	-- and will emit when that time has been reached. If the time has already passed,
	-- noSooner will emit immediately. Otherwise it will schedule a timer to go off at
	-- the scheduled time and emit then
	noSooner = noSooner:map( function( nextTime )
		local now = getTimeNow()
		local delay = nextTime - now
		if delay > 0 then
			return Rx.Observable.timer( delay, WFRxScheduler.sharedScheduler() )
		else
			return Rx.Observable.just( true )
		end
	end )
	:switch()
	:map( function()
		lastDelayCount = lastDelayCount + 1
		return lastDelayCount
	end )

	local lastCount = 0
	local lastCountProcessed = 0

	local updateCount = observable:map( function()
		lastCount = lastCount + 1
		return lastCount
	end )

	local rateLimited = noSooner
		:combineLatest( function( delayCount, updateCount, ... )

			if updateCount <= lastCountProcessed then
				-- updateCount matches the last update we processed
				-- so the timer fired after the delay expired
				-- but there are no new updates to process
				return Rx.Observable.never()
			end
			if delayCount <= lastDelayCountProcessed then
				-- we have updates but the timer hasn't fired
				-- so we're still in the throttling period
				return Rx.Observable.never()
			end
			if delayCount < lastDelayCount then
				-- this probably shouldn't happen, but if we're
				-- somehow behind the timer, squelch the update
				-- because we'll be getting another timer update
				-- soon
				return Rx.Observable.never()
			end

			return Rx.Observable.just( ... )

		end, updateCount, observable )
		:switch()

	local function timeIt( start, ... )
		lastCountProcessed = lastCount
		lastDelayCountProcessed = lastDelayCount

		-- Schedule the next time we will process this query by taking
		-- calculating the duration of the query (now - start) and
		-- multiplying that by the timeRatio to get the delay before the
		-- next query execution. Since `noSoonerSb` takes absolute times
		-- add the current time to the delay to get it's next scheduled
		-- run.

		-- This prevents the query from monopolizing the index store thread
		-- and allows other queries opportunities to get their results
		local now = getTimeNow()
		local noSooner = now + ( now - start ) * timeRatio
		noSoonerSb:onNext( noSooner )

		return ...
	end

	return rateLimited:map( function( ... )

		local start = getTimeNow()
		return timeIt( start, queryFn( ... ) )

	end )

end

WFRxIndexStoreImpUtils.throttledQuery = throttledQuery

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

local function fireThenDebounce( observable, throttleTime )

	local refCountedObservable = observable:replay(nil, 1):refCount()

			--[[ In the case of throttleTime being present, we need to provide an
				immediate query response and convert to a debounce for the remaining
				lifetime of the observable. We do this by grabbing the first element
				of refCountedAnyTableUpdated, and fusing it with the debounced tail of
				refCountedAnyTableUpdated]]

	return Rx.Observable.concat(
		refCountedObservable:take(1),
		refCountedObservable
		:skip(1) -- Skip the head because we are passing in an "unchanged head"
		:debounce( throttleTime, WFRxScheduler.sharedScheduler() ))
end

WFRxIndexStoreImpUtils.fireThenDebounce = fireThenDebounce

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

local function runQueryWhenTablesUpdate( impObj, queryFn, sql, isQuiesceUntilUpToDate, isQueryOnce, optCalm, throttleTime )

	local anyTableUpdated = impObj:watchForTableUpdatesForSql( sql, isQuiesceUntilUpToDate )

	-- Raw query event stream. Fires whenever new data is available in any of the
	-- index tables, modulo the calming factor.

	-- The calming factor now scales with the performance of the query;
	-- as results get big, we calm more and more.
	local relevantTables = {}

	if isQuiesceUntilUpToDate then

		-- We consider ourselves "up to date" as soon as all the tables are
		-- current with the doc store rev counter as of the time the query
		-- started.
		-- If additional documents have arrived in the time it took us
		-- to bring the index store up to date, then we might not include those
		-- new documents in this query

		anyTableUpdated = anyTableUpdated:filter( function( docStoreRevCounter, ... )

			if not docStoreRevCounter then
				return
			end

			local n = select( '#', ... )

			for i = 1, n do

				local tableUpdateInfo = select( i, ... )

				if tableUpdateInfo.tableName then
					relevantTables[ #relevantTables + 1 ] = tableUpdateInfo.tableName
				end

				local tableUpToDateCounter =
						tonumber( tableUpdateInfo.tableUpToDateCounter ) or 0

				if tableUpToDateCounter <= docStoreRevCounter then
					return false
				end

			end

			return true

		end )
	end

	local needsToDeliverFirstResult = true

	anyTableUpdated = anyTableUpdated:where( function( ... )

		if needsToDeliverFirstResult then
			return true
		end

		local startIndex = 1
		if isQuiesceUntilUpToDate then
			startIndex = 2
		end

		local n = select( '#', ... )

		for i = startIndex, n do

			local tableUpdateInfo = select( i, ... )

			if tableUpdateInfo.tableName then
				relevantTables[ #relevantTables + 1 ] = tableUpdateInfo.tableName
			end

			if tableUpdateInfo.tableHasNewIndexData then

				return true

			end

		end

		return false

	end )

	if not isQueryOnce then

		if throttleTime then

			anyTableUpdated = fireThenDebounce( anyTableUpdated, throttleTime )

		else

			-- But we don't want to update too often in the non-calming case. Note that we start with the timer
			-- running so that we are calm from the get go, implemented via the :debounce() operator.
			-- In the case of a given calming factor we emit the first result immediately and start calming
			-- afterwards

			-- In order to speed up the WF unit tests, we remove all calming from index store queries.

			if optCalm and not WFPlatformInfo.isUnitTest() then

				anyTableUpdated = anyTableUpdated:calm( optCalm, WFRxScheduler.sharedScheduler() )

			else

				anyTableUpdated = anyTableUpdated:debounce( .125, WFRxScheduler.sharedScheduler() )

			end
		end
	end

	local wrappedQueryFn = function()
		needsToDeliverFirstResult = false
		return queryFn( sql )
	end

	-- This throttleQuery call will slow down how often a query runs. For example, if a query takes
	-- 1s to run, it won't be run again for another 4s even if there are updates. This is to help
	-- minimize the impact of slow/active queries on the index store and other queries.
	-- Note that this effectively means that the query will not update more often than the
	-- max( calm, 4 * query execution time ).
	local rawQuery = throttledQuery( anyTableUpdated, 4, wrappedQueryFn )

	return rawQuery
end

--------------------------------------------------------------------------------
--[[
	\brief Reports as an Observable the results of a SQL query on the index store.

	\param impObj (object) Object that implements the function 'watchForTableUpdatesForSQL( self, sql, quiesceUntilUpToDate )'
		that returns an Observable that will fire each time the relevant tables update and the SQL should be re-run
	\param queryFn (function) A function that executes a query and returns results as an array of arrays
	\param args (table) arguments for the query. See the WFRxIndexStore:query documentation for a list
		of valid arguments

	\return Observable< table > An Observable whose value tracks query results as an array of arrays.
--]]
function WFRxIndexStoreImpUtils.query( impObj, queryFn, args )

	local sql = args.sql

	local kSemiColonByte = 59
	if string.byte( sql, -1 ) ~= kSemiColonByte then
		sql = sql .. ';'
	end

	local rawQuery = runQueryWhenTablesUpdate( impObj, queryFn, sql, args.quiesceUntilUpToDate or false, args.queryOnce or false, args.calm, args.throttleTime or false )

	local previousResult, previousRowCount, previousSha256
	
	-- Squelch events if the actual query results are unchanged.
	
	local resultQuery = rawQuery
	
		:filter( function( newValue, rowCount )

			if type( newValue ) == 'table' then
		
				if newValue.sha256 then

					-- If the query result contains a sha256 checksum, we can use
					-- that for a quick, inexpensive result comparison.

					if previousSha256 == newValue.sha256 then
						return false
					else
						previousSha256 = newValue.sha256
						return true
					end

				else
				
					-- If the query does not contain a sha256 checksum, which is the
					-- case for set-logic queries, fallback to comparing the result
					-- sets directly.
				
					rowCount = rowCount or #newValue
					
					if previousRowCount == rowCount
							and KSTableUtils.deepEquals( previousResult, newValue ) then
							
						return false

					else
					
						previousResult = newValue
						previousRowCount = rowCount
						return true

					end
					
				end

			else
				return true
			end

		end )
		
		:map( function( ... )
		
			local newValue = ...
		
			if type( newValue ) == 'table' then
				newValue.sha256 = nil
			end
			
			return ...
		
		end )

	local bufferSize = 1
	local initialRequestCount = math.huge

	local controlledObservable = resultQuery:call( controlledWithBuffer, bufferSize, initialRequestCount )

	local unsubscribeObservable = controlledObservable:call( disposeOnUnsubscribe )

	args._controlledObservable = controlledObservable

	unsubscribeObservable:addDisposable( Rx.Disposable.create( function()

		args._controlledObservable = nil
		impObj._queryRequests[ unsubscribeObservable ] = nil

	end ) )

	-- Remember the query.

	impObj._queryRequests[ unsubscribeObservable ] = args

	return unsubscribeObservable

end

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

--[[
	\brief Loads index store schema from the index spec with the specified name

	\return WFRxIndexStoreSchema
]]--
function WFRxIndexStoreImpUtils.loadSchema( indexSpecScript )

	indexSpecScript = string.gsub( indexSpecScript, "%.lua$", "" )

	local schemaNamespace = require( indexSpecScript )
	local indexStoreSchema

	if type( schemaNamespace ) == 'table' then
		indexStoreSchema = schemaNamespace.indexStoreSchema
	end

	return indexStoreSchema

end

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

--[[
	\brief Identifies the tables in a SQL statement that match a list of possible tables

	\param tableNames (array) an array of table names that might be in the SQL
	\param sql (string) the SQL string

	\return map of table names from the \c tableNames parameter that were referenced in the SQL
--]]
function WFRxIndexStoreImpUtils.identifyRelevantTables( tableNames, sql )

	-- Identify all of our index tables that might be relevant for this query.
	-- Naive implementation: Simply search for each table name in query string.

	local result = {}

	for tableName in pairs( tableNames ) do

		-- Naive search, too: Since SQL table names can't contain Lua pattern
		-- special characters, just use the table name as search pattern.

		if sql:find( "[^a-zA-Z]" .. tableName .. "[^a-zA-Z]" ) then
			result[ tableName ] = true
		end

	end

	return result

end

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

--[[

	\brief Returns a table containing information about open index store SQLite
		queries. This function is only used by the diagnostic log.

	\param queryRequests (table) current query Observable requests.
		key: (Observable< table >) stream returned to caller
		value: (table) arguments to :query, plus the following fields:
			relevantTables: (table) set of table names to update
			nUpdatesSent: (number) how many times have we broadcasted results

	\return (table)

--]]

function WFRxIndexStoreImpUtils.filterAndFormatQueryRequestStats( queryRequests )

	local filteredQueryRequests = {}
	local nQueries = 0

	local function limitNestedQueryArgs( queryArgs )

		local kMaxQueryArgsToIncludeInStats = 100
			-- https://jira.corp.adobe.com/browse/WF-2299: Intentionally limited.

		if queryArgs[ 1 ] then

			local limitedArgs = {}

			for i = 1, kMaxQueryArgsToIncludeInStats do

				local queryArg = queryArgs[ i ]

				if type( queryArg ) == 'table' then

					limitedArgs[ i ] =  limitNestedQueryArgs( queryArg )

				else

					limitedArgs[ i ] = queryArg

				end

			end

			if #queryArgs > kMaxQueryArgsToIncludeInStats then

				limitedArgs[ '_warning_' ] = string.format( "Only %d of %d arguments are included in this log.",
						kMaxQueryArgsToIncludeInStats, #queryArgs )

			end

			return limitedArgs

		end

	end

	for resultE, queryInfo in pairs( queryRequests ) do

		if not WFCloseSupport.hasBeenOrWillBeClosed( resultE ) then

			nQueries = nQueries + 1

			local traceback = queryInfo.traceback

			if traceback then
				traceback = KSStringUtils.split( queryInfo.traceback, '\n' )
				table.remove( traceback, 1 )
			end

			filteredQueryRequests[ nQueries ] = {
				sql = queryInfo.sql,
				relevantTables = queryInfo.relevantTables,
				nUpdatesSent = queryInfo.nUpdatesSent,
				calm = queryInfo.calm,
				threadName = queryInfo.threadName,
				traceback = traceback,
			}

			filteredQueryRequests[ nQueries ].args = limitNestedQueryArgs( queryInfo )

		end

	end

	table.sort( filteredQueryRequests, function( a, b )
		return a.sql < b.sql
	end )

	return filteredQueryRequests

end

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

--[[

	\brief Returns a table containing information about open index store SQLite
		queries. This function is only used by the diagnostic log.

	\param tableName (string) name of index store table.
	\param columnName (string) name of column for the specified table.
	\param keys (table) values to match against the specified column.

	\return (table) Returns a tables that lists all matches, like this:

		{
			"<key1>": [
				"<assetId1>"
			],
			"<key2>": [
				"<assetId1>",
				"<assetId2>",
				"<assetId3>"
			]
		}

--]]

function WFRxIndexStoreImpUtils.genericIndexStoreQuery( catalog, tableName, columnName, keys )

	check( 'WFCatalog', 'string', 'string', 'table' )

	local sql = string.format(
			[[SELECT docId, %s
				FROM %s
				WHERE %s IN CARRAY( ?, ?, 'char*' )]],
			columnName,
			tableName,
			columnName )

	return catalog:indexStore():queryOnce {
			sql = sql,
			keys,
			#keys,
		}

		:map( function( rows )

			local result = {}

			for _, row in ipairs( rows ) do

				local docId = row[ 1 ]
				local key = row[ 2 ]

				local assetsForKey = result[ key ] or {}
				assetsForKey[ #assetsForKey + 1 ] = docId
				result[ key ] = assetsForKey

			end

			return result

		end )

end

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

return WFRxIndexStoreImpUtils
