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

WFRxIndexStoreQueryServer.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain 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 may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  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'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFCurrentThread = require 'WFCurrentThread'
local WFDiagnosticLog = require 'WFDiagnosticLog'

local WFRxIndexStoreImpUtils = require 'WFRxIndexStoreImpUtils'

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

local Observable = Rx.Observable

local check = assert( WFArguments.check, "Cannot find WFArguments.check" )


--============================================================================--
--[[!
	\brief A \c WFRxIndexStoreQueryServer is used by the index store implementation
		to keep the index store tables up to date.

	It should only be accessed by \c WFRxIndexStoreImp.
--]]
local queryServerMethods = {

	_mapServer = false,
		-- (WFRxIndexStoreMapServer) map server for this index store

	_tableNames = false,
		-- (table) names of the index tables declared in this schema

	_manualTableNames = false,
		-- (table) names of the manual tables (no map function) declared in this schema

	_nColumnsPerManualTable = false,
		-- (table) how many columns are in each manual table

	_needsUpdateFromManualTable = false,
		-- (table) Array of Observable per manual table to trigger query execution

	_queryRequests = false,
		-- (table) current query Observable requests for the main read/write thread
		-- RO thread query Observables are tracked in the ReadOnly Imp
			-- key: (Observable< table >) stream returned to caller
			-- value: (table) arguments to :query, plus
				-- relevantTables: (table) set of table names to update
				-- nUpdatesSent: (number) how many times have we broadcasted results?

	_relevantTablesForSql = false,
		-- (table) maps sql the query server is watching to the relevant table names
			-- key: (string) sql string
			-- value: (table) array of table names that will be tracked for updates

	_virtualTables = false,
		-- (table) names and Observable< boolean > that updates whenever
		-- the virtual table has more data.  These are poked in by the imp.

	WFCloseSupport.addCloseSupportMethods,

}

--------------------------------------------------------------------------------
--[[!
	\brief Create a query server to support an index store implementation.

	\param mapServer (\c WFRxIndexStoreMapServer) map server providing the index tables
	\param tableNames (table) set of table names declared in this schema
--]]
function queryServerMethods:__init( mapServer, indexStoreSchema )

	check( 'WFRxIndexStoreQueryServer', 'WFRxIndexStoreMapServer', 'WFRxIndexStoreSchema' )

	self._virtualTables = {}
	self._needsUpdateFromManualTable = {}
	self._mapServer = mapServer
	self._mapServer:addCloseTarget( self )

	self._tableNames = indexStoreSchema:tableNames()
	self._manualTableNames = indexStoreSchema:manualTableNames()
	self._nColumnsPerManualTable = indexStoreSchema:nColumnsPerManualTable()

	self._queryRequests = {}
	self._relevantTablesForSql = {}

	for manualTableName in pairs( self._manualTableNames ) do
			self._needsUpdateFromManualTable[ manualTableName ] = Rx.ReplaySubject( 1 )

			self._needsUpdateFromManualTable[ manualTableName ]
				:onNext( {
					tableHasNewIndexData = true,
					tableUpToDateCounter = math.huge
				} )

	end

	WFDiagnosticLog.addLogProvider( self )

end

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

--[[!
	\brief Produces an Observable that watches for updates to tables referenced in
		the SQL and fires when those tables are updated.

	\param sql (string) the SQL to watch for table updates for
	\param quiesceUntilUpToDate (boolean) whether the Observable should fire only when
		the table is up to date or on all table updates

	\return Observable that fires when the table(s) are updated
--]]
function queryServerMethods:watchForTableUpdatesForSql( sql, quiesceUntilUpToDate )
	check( 'WFRxIndexStoreQueryServer', '+string', 'boolean' )

	if quiesceUntilUpToDate ~= nil then
		assert( type( quiesceUntilUpToDate ) == 'boolean', 'quiesceUntilUpToDate must be true or false' )
	end

	-- Identify relevant tables.
	local relevantTablesSet = WFRxIndexStoreImpUtils.identifyRelevantTables( self._tableNames, sql )

	local nTables = 0
	local relevantTables = {}

	for tableName in pairs( relevantTablesSet ) do
		nTables = nTables + 1
		relevantTables[ nTables ] = tableName
	end

	-- Housekeeping ...

	local mapServer = self._mapServer

	-- Do we have new data from any of the tables?

	local tablesUpdated = {}

	for i, relevantTable in ipairs( relevantTables ) do
		tablesUpdated[ i ] = mapServer:keepTableUpToDate( relevantTable )
	end

	for manualTableName in pairs( WFRxIndexStoreImpUtils.identifyRelevantTables( self._manualTableNames, sql ) ) do
		tablesUpdated[ #tablesUpdated + 1 ] = self._needsUpdateFromManualTable[ manualTableName ]
	end

	-- track relevant tables for sql for diagnostic purporses
	self._relevantTablesForSql[ sql ] = relevantTables

	for tableName in pairs( WFRxIndexStoreImpUtils.identifyRelevantTables( self._virtualTables, sql ) ) do
		local observable = self._virtualTables[ tableName ]
		tablesUpdated[ #tablesUpdated + 1 ] =
			observable
				:map(
					function( updateInfo )

						if type( updateInfo ) == "table" then

							return updateInfo

						end

						return {

							tableUpToDateCounter = tonumber( updateInfo ) or 0,

							-- currently virtual tables always return 0, so
							-- they are behind and should never cause a new
							-- query to fire. That may be incorrect behavior,
							-- but this code preserves it for now.
							tableHasNewIndexData = false,

						}

					end )

	end

	nTables = #tablesUpdated

	if nTables == 0 then
		error( "WFRxIndexStoreQueryServer:query: did not find any relevant tables in query " .. sql, 2 )
	end

	-- We use the laterE event to start the clock on calming even though
	-- we don't use this first event to trigger a re-query. This prevents us
	-- from re-running the query (and publishing results) any sooner than
	-- (interval) seconds after the query Observable is created.
	-- If you disable this logic, expect the test named testQuery_outOfDate_join
	-- to fail with a "too soon" error.

	local anyTableUpdated

	if quiesceUntilUpToDate then

		-- keep track of the revCounter for each table
		anyTableUpdated = Observable.combineLatest(
				self._mapServer:currentChangeCounter(),
				unpack( tablesUpdated, 1, nTables )
			 )

	else
		anyTableUpdated = Observable.combineLatest( unpack( tablesUpdated, 1, nTables ) )
	end

	return anyTableUpdated

end

--------------------------------------------------------------------------------
--[[!
	\brief Create an observable that tracks the results of a SQL query on the index store.

	\param args (table) arguments for query
		- \c sql: (string) SQL query to be performed
		- \c threadName: (optional, string) name of thread requesting the query
		- \c traceback: (optional, string) traceback of call site in requesting thread
		- \c calm: (optional, number) if present, calm query result updates to no more
			than one every _n_ seconds. Additional index store throttling may mean that
			the delay between updates is larger than the calm value here.
		- \c nArgs: (optional, number) number of query arguments (helps with \c nil
			values in the middle of the list)
		- 1 .. n: (optional) query arguments to be substituted for ? placeholders in query

	\return Observable< table > query results (array of arrays syntax)

	\b WARNING: Use this method sparingly. Creating many query Observables will adversely
		affect the overall performance of the application.

	\b WARNING: This Observable may, at times, return information that is based on
		incomplete index table results. It will eventually catch up and update
		results if needed. (Out of date information is especially likely for the
		first value returned.)
--]]
function queryServerMethods:query( args )

	check( 'WFRxIndexStoreQueryServer', 'table' )

	local encoder = args.encoder
	if encoder ~= nil then assert( type( encoder ) == 'string' ) end

	args.nUpdatesSent = 0
	local nQueryArgs = args.nArgs or #args
	local queryResultsNow = function( sql )
		args.nUpdatesSent = args.nUpdatesSent + 1
		local singleColumnQuery = args.singleColumnQuery
		return self._mapServer:query( sql, singleColumnQuery, encoder, args.tempIds, unpack( args, 1, nQueryArgs ) )
	end

	return WFRxIndexStoreImpUtils.query( self, queryResultsNow, args )

end

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

function queryServerMethods:bulkInsertManualTable( tableName, rows )

	check( 'WFRxIndexStoreQueryServer', 'string', 'table' )

	if #rows == 0 then
		return Observable.throw( "Empty value" )
	end

	if not self._manualTableNames[ tableName ] then
		return Observable.throw( "WFRxIndexStoreQueryServer:bulkInsertManualTable: did not find the manual table name" )
	end

	local mapServer = self._mapServer
	local db = mapServer._db
	local exec = db.exec

	local nColumns = self._nColumnsPerManualTable[ tableName ]
	local queryPlaceholders

	if nColumns > 1 then
		queryPlaceholders = string.format( '?%s', string.rep( ', ?', nColumns - 1 ) )
	else
		queryPlaceholders = '?'
	end

	local sql = string.format( 'INSERT INTO %s VALUES (%s)', tableName, queryPlaceholders )

	local result = db:withTransactionDo( function( db )

		local result = {}

		for i, row in ipairs( rows ) do

			local res = exec( db, sql, unpack( row ) )

			result[ i ] = res

		end

		return result

	end, db )

	self._needsUpdateFromManualTable[ tableName ]
		:onNext(
			{
				tableHasNewIndexData = true,
				tableUpToDateCounter = math.huge
			}
		)

	return Observable.just( result )

end

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

function queryServerMethods:clearManualTable( tableName )

	check( 'WFRxIndexStoreQueryServer', 'string' )

	if not self._manualTableNames[ tableName ] then
		error( "WFRxIndexStoreQueryServer:clearManualTable: did not find the manual table name" )
	end

	local db = self._mapServer._db
	local exec = db.exec

	exec( db, string.format( 'DELETE FROM %s', tableName ) )

	self._needsUpdateFromManualTable[ tableName ]
		:onNext(
			{
				tableHasNewIndexData = true,
				tableUpToDateCounter = math.huge
			}
		)

	return Observable.just( true )

end

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

function queryServerMethods:queryStats()

	return WFRxIndexStoreImpUtils.filterAndFormatQueryRequestStats( self._queryRequests )

end
--------------------------------------------------------------------------------

function queryServerMethods:sendDiagnosticLogData( )

	local sqlInfo = {}
	for sql, relevantTables in pairs( self._relevantTablesForSql ) do
		table.insert( sqlInfo,
			{
				sql = sql,
				relevantTables = relevantTables
			} )
	end

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

	return {

		id = 'WRxIndexStoreQueryServer-' .. WFCurrentThread.DEBUG_getThreadName():replace( ' :: ', '+' ),
		name = string.format( 'Query Server :: %s', WFCurrentThread.DEBUG_getThreadName() ),
		weight = 1750,
		parentSection = WFDiagnosticLog.Sections.INDEX_STORE,
		body = {
			[ "Query Details" ] = sqlInfo
		}
	}
end

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

return Rx.Class 'WFRxIndexStoreQueryServer'( queryServerMethods )
