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

WFRxIndexStore.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'
local Observable = Rx.Observable

	-- WF.core.lua.threading
local WFCurrentThread = require 'WFCurrentThread'
local WFWorkerThread = require 'WFWorkerThread'

	-- WF.core.os
local WFPlatformInfo = require 'WFPlatformInfo'

	-- WF.core.reactive
local KSReactive = require 'KSReactive'

	-- WF.core.rxlua
local WFRxEventBridge = require 'WFRxEventBridge'
local WFRxPromiseBridge = require 'WFRxPromiseBridge'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.utils
local KSBucketBrigade = require 'KSBucketBrigade'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFLogger = require 'WFLogger'
local WFUnknown = require 'WFUnknown'

	-- WF.sqlite.doc_store
local WFRxDocStore = require 'WFRxDocStore'
local WFRxPlatformIndexStore = require 'WFRxPlatformIndexStore'

local WFRxRunUntilDone = require 'WFRxRunUntilDone'

local WFRxIndexStoreImpUtils = require 'WFRxIndexStoreImpUtils'

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

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

local asKSEvent = WFRxEventBridge
local asKSPromise = WFRxPromiseBridge

local DEBUG = _G.DEBUG

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

local kSetLogicQuery = '< set logic query >'

local logger = WFLogger( 'WFRxIndexStore' )

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

local function containsOnlyStandardIndexStoreTables( self, sql )

	-- Don't send queries for temp tables created with virtualTableForSetQuery or
	-- for tables named with TEMP_
	if sql:match( "temp%.ResultSetVirtualTable_" ) or sql:match( "TEMP_" ) then
		return false
	end

	-- Don't send queries for manual tables as defined by the schema
	local manualTableNames = self._indexStoreSchema:manualTableNames()
	local relevantManualTables = WFRxIndexStoreImpUtils.identifyRelevantTables( manualTableNames, sql )
	if next( relevantManualTables ) then
		return false
	end

	return true
end

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

local function dispatchToWorkerThread( self, selector, ... )
	return self._indexStoreWorkerThread:request( selector, ... )
end

local readOnlyThreadRotation = 0

local function dispatchToReadOnlyWorkerThreadIfAvailableAndSupportsQuery( self, selector, args )

	if type( self._indexStoreReadOnlyWorkerThreads ) == 'table' and
		not args.tempIds and -- no queries with temp ids since we can't write table
		not ( args.sql == kSetLogicQuery ) and -- no set logic queries since we can't write table

		containsOnlyStandardIndexStoreTables( self, args.sql ) then

		local readOnlyThreadCount = #self._indexStoreReadOnlyWorkerThreads
		readOnlyThreadRotation = ( readOnlyThreadRotation % readOnlyThreadCount ) + 1
		local workerThread = self._indexStoreReadOnlyWorkerThreads[ readOnlyThreadRotation ]
		return workerThread:request( selector, args )

	else

		return dispatchToWorkerThread( self, selector, args )

	end
end

--============================================================================--
--[[!
	\class WFRxIndexStore
	\brief Tracks an \c WFRxDocStore and provides SQL-style indexing of its contents.

	An index store is created using an index schema (a list of tables and indices)
	and a reference to a doc store. It provides ability to run SQL queries against
	the tables.
--]]

-- @mark WFRxIndexStore

local indexStoreMethods = {

	_indexSpecScript = false,
		-- (string) the name of the index definition script used by the index store

	_indexStoreSchema = false,
		-- (WFRxIndexStoreSchema)

	_indexStoreWorkerThread = false,
		-- (WFWorkerThread) reference to the event processor that does the indexing
		-- and handles queries requiring writing to the database

	_indexStoreReadOnlyWorkerThreads = false,
		-- (table) table of WFWorkerThreads that can handle read only queries

	_areWorkerThreadsInitialized = false,
		-- (ReplaySubject) subject indicating whether the worker threads have been initalized

	_platformIndexStore = false,
		-- (WFRxPlatformIndexStore) The reference to the platform based indexstore.
		-- Will remain false if it is not enabled.


	_docStoreId = false,
		-- (string) doc store ID, used for debugging

	_queryResultsCache = false,
		-- (fully weak table) a mapping from cache keys to query results

	_queryResultsBrigade = false,
		-- (KSBucketBrigade) used to avoid having GC wipe things out too fast

	WFCloseSupport.addCloseSupportMethods,

}

--------------------------------------------------------------------------------
--[[!
	\brief Creates an index store object that tracks the contents of an Oz document store.

	\param docStore (\c WFRxDocStore or \c WFDocStore or \c OzDocStore) The document store object.
	\param indexSpecScript (string) The name of an index definition script for this index
	\param optIndexStoreReadOnlyThreadCount (number) the number of read-only threads to create to handle queries
	\param optEnablePlatformIndexStore (boolean) Boolean indicating whether platform based indexstore should be enabled
--]]

function indexStoreMethods:__init( docStore, indexSpecScript, optIndexStoreReadOnlyThreadCount, optEnablePlatformIndexStore )

	check( 'WFRxIndexStore', 'WFRxDocStore', '+string', '?number', '?boolean' )

	self._indexSpecScript = indexSpecScript

	local indexStoreSchema = WFRxIndexStoreImpUtils.loadSchema( indexSpecScript )
	assert( indexStoreSchema, "Cannot load schema" )

	self._indexStoreSchema = indexStoreSchema

	local docStoreId = docStore:path()
	local isMemoryDb = docStoreId:sub( 1, 8 ) == ':memory:'
	local isTempDb = docStoreId:sub( 1, 6 ) == ':temp:'

	if isMemoryDb or isTempDb then

		if optIndexStoreReadOnlyThreadCount and optIndexStoreReadOnlyThreadCount > 0 then
			logger:errorf( "Unable to use multiple index store threads with in-memory or temp databases, will restrict %s to a single thread", docStoreId )
		end

		-- in memory dbs should not use multiple read threads because they can end up with table
		-- locks in the shared cache (even if SQLITE_OPEN_PRIVATECACHE)
		optIndexStoreReadOnlyThreadCount = 0

	end

	assert( WFRxDocStore:isClassOf( docStore ), "Requires a WFRxDocStore docStore" )

	self._docStoreId = docStoreId

	self._queryResultsCache = setmetatable( { }, { __mode = 'kv' } )

	local cacheBrigadeDepth = WFPlatformInfo.isConstrainedMemoryDevice and 5 or 10

	self._queryResultsBrigade = KSBucketBrigade.makeBrigade( cacheBrigadeDepth )

	-- Cannot use :setOwned because we need to use self._indexStoreWorkerThread to send a close
	-- message to the imp.
	--	self:setOwned {
	--	}


	self._areWorkerThreadsInitialized = Rx.ReplaySubject( 1 )
	self._areWorkerThreadsInitialized:onNext( false )

	local initArgs = string.format( '%s=%s', indexSpecScript, docStoreId )
	self._indexStoreWorkerThread = WFWorkerThread( 'WFRxIndexStoreImp.lua', initArgs )

	local init = self._indexStoreWorkerThread:request( 'initIndexStore', docStoreId, indexSpecScript )
		:subscribe(function()

			if optIndexStoreReadOnlyThreadCount and optIndexStoreReadOnlyThreadCount > 0 then

				self._indexStoreReadOnlyWorkerThreads = {}
				local initializedThreads = 0

				for i = 1, optIndexStoreReadOnlyThreadCount do
					local workerThread = WFWorkerThread( 'WFRxIndexStoreReadOnlyImp.lua', i .. "=" .. initArgs )

					local workerThreadInitRequest = workerThread:request( 'initIndexStoreReadOnly', docStoreId, indexSpecScript )
							:subscribe( function()

								table.insert( self._indexStoreReadOnlyWorkerThreads, workerThread )

								-- Lazily assuming that if the last worker thread is initialized they all are.
								-- It's not perfect here, but it's close enough to get started passing out
								-- query requests
								initializedThreads = initializedThreads + 1
								if initializedThreads == optIndexStoreReadOnlyThreadCount then
									self._areWorkerThreadsInitialized:onNext( true )
								end


							end)

					self:addCloseAction( function()
						workerThreadInitRequest:dispose()
					end )

				end

			else

				-- no read only threads to init and the main r/w thread is initialized so mark
				-- initialization complete
				self._areWorkerThreadsInitialized:onNext( true )

			end

			if optEnablePlatformIndexStore then
				self._platformIndexStore = WFRxPlatformIndexStore( docStore:path(), indexSpecScript )
			end
		end)

	self:addCloseAction( function()
		init:dispose()
	end )



end

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

--[[!
	\brief Returns the name of the index spec script for this index store

	\return string the name of the index spec script
--]]
function indexStoreMethods:indexSpecScript()
	return self._indexSpecScript
end

--------------------------------------------------------------------------------
--[[!
	\brief Performs a single SQL query on the index store and returns the result
		as a Promise.

		Before performing the query, the method brings all relevant index tables
		up to date. If data is added to the doc store while this request
		is being processed, the newer data might or might not appear in the query results.

	\param sql (string) The SQL query to be performed. OR a table:
		- \c sql: (string) SQL query to be performed
		- \c tempIds: (optional, array of doc ids) if present, the table TEMP_ids will
			be populated with these ids for use in the query, e.g. as part of
			an join such as "INNER JOIN TEMP_ids ON TEMP_ids.docId = yourTable.docId"
			or via a where in clause such as "WHERE docId IN ( SELECT docId from TEMP_ids )"
			TEMP_ids is a temporary table with only one column, docId, and no special indexes.
		- 1 .. n: (optional) query arguments to be substituted for ? placeholders in query
	\param ... (optional) The query arguments to be substituted for ? placeholders in the query.
		ONLY if sql was provided as a string, not as the table form.

	\return <code>P< table ></code> A Promise that resolves to the query results,
		as an array of arrays.
--]]

function indexStoreMethods:queryP( sql, ... )

	check( 'WFRxIndexStore', 'string|table' )

	local args

	if type( sql ) == 'string' then
		args = {
			sql = sql,
			nArgs = select( '#', ... ),
			...
		}
	else
		args = {}
		for k,v in pairs( sql ) do
			args[ k ] = v
		end
	end
	args.quiesceUntilUpToDate = true
	args.calm = 1 -- deliver first (and only) result immediately

	return self:query( args )
		:call( asKSPromise )

end

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

local function getCachedResult( self, cacheKey, ... )

	if not cacheKey then
		return ...
	end

	local entry = self._queryResultsCache[ cacheKey ]

	if not entry then
		return ...
	end

	self._queryResultsBrigade( entry )

	return entry.data

end

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

local function storeCachedResult( self, cacheKey, value )

	assert( value ~= nil, "Cannot cache nil" )

	local entry = self._queryResultsCache[ cacheKey ]

	if entry then
		entry.data = value
	else
		entry = { data = value }
		self._queryResultsCache[ cacheKey ] = entry
	end

	self._queryResultsBrigade( entry )

end

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

local enqueue2selector = {
	normal = 'query',
	fast = '+query',
	slow = '-query'
}

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

--[[!
	\brief DEPRECATED. Reports as a KSReactive Behavior 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 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 cacheKey: (optional, string) if supplied, this will be used as a cache
			value to look up cached query results
		- \c enqueue: (optional, string) enqueue policy for this query. Default is 'normal'.
			other values are 'fast' and 'slow'. The former will push the query to the front
			of the list. The latter will cause even the initial response to be handled
			in order.
		- \c nArgs: (optional, number) if present, specifies number of arguments
			(overrides #args)
		- \c tempIds: (optional, array of doc ids) if present, the table TEMP_ids will
			be populated with these ids for use in the query, e.g. as part of
			an join such as "INNER JOIN TEMP_ids ON TEMP_ids.docId = yourTable.docId"
			or via a where in clause such as "WHERE docId IN ( SELECT docId from TEMP_ids )"
			TEMP_ids is a temporary table with only one column, docId, and no special indexes.
		- 1 .. n: (optional) query arguments to be substituted for ? placeholders in query

	For compatibility with an older version of this API, this function also permits
	the first argument to be a string (interpreted as the SQL query string) followed
	by any number of query parameters as ... arguments. Future code should use the
	new named-argument syntax. Note that \c calm is inaccessible
	through the old syntax.

	You may use a behavior for \c sql and any of the query arguments. In this case
	the actual query performed will update in response to changes to the behavior(s).

	\return B< table > A Behavior whose value tracks query results, as an array of arrays.

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

	\b WARNING: This behavior 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 indexStoreMethods:queryB( args, ... )

	check( 'WFRxIndexStore', 'string|table' )

	-- Retrofit for old syntax.

	assert( not KSReactive.isBehavior( args ), 'KSReactive.isBehavior sql no longer supported' )
	if type( args ) == 'string' then
		local nArgs = select( '#', ... )
		args = { sql = args, nArgs = nArgs, ... }
	end

	-- Fill in thread and traceback info if in debug build.

	args.threadName = DEBUG and WFCurrentThread.DEBUG_getThreadName() or nil
	args.traceback = DEBUG and debug.traceback( '', 2 ) or nil

	-- Are there any behaviors in params?

	local nArgs = args.nArgs or #args

	for i = 1, nArgs do
		assert( not KSReactive.isBehavior( args[ i ] ), 'KSReactive.isBehavior sql no longer supported' )
	end

	-- Figure out the initial value. This will bias how we handle the
	-- enqueue logic.

	local cacheKey = args.cacheKey
	local initialValue = getCachedResult( self, cacheKey, WFUnknown.value() )
	args.initialValue = initialValue

	-- If we've got a value already, we don't need to rush to compute a new one
	-- But note that we only want to do this once.

	if WFUnknown.isKnown( initialValue ) then
--		args.enqueue = args.enqueue or 'slow'
	end

	return self:query( args, ... )
		:call( asKSEvent )
		:closingHoldB( initialValue )

end

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

local function assertOptionalBoolField( args, fieldName )
	local value = args[ fieldName ]
	if value == nil or value == true or value == false then
		return value
	end
	assert( false, "args." .. fieldName .. " must be a bool" )
end

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

local function assertOptionalPositiveNumberField( args, fieldName )
	local value = args[ fieldName ]
	if value == nil or ( type( value ) == 'number' and value > 0 ) then
		return value
	end
	assert( false, "args." .. fieldName .. " must be a positive number" )
end

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

local function assertOptionalStringField( args, fieldName )
	local value = args[ fieldName ]
	if value == nil or type( value ) == 'string' then
		return value
	end
	assert( false, "args." .. fieldName .. " must be a string" )
end

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

local function assertOptionalTableField( args, fieldName )
	local value = args[ fieldName ]
	if value == nil or type( value ) == 'table' then
		return value
	end
	assert( false, "args." .. fieldName .. " must be a table" )
end

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

local function waitForWorkerThreadsToInitialize( self )

	return self._areWorkerThreadsInitialized
		:filter( function( value )
			return value
		end )
		:take( 1 )

end

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

--[[!
	\brief Reports as an Observable 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 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 cacheKey: (optional, string) if supplied, this will be used as a cache
			value to look up cached query results
		- \c enqueue: (optional, string) enqueue policy for this query. Default is 'normal'.
			other values are 'fast' and 'slow'. The former will push the query to the front
			of the list. The latter will cause even the initial response to be handled
			in order.
		- \c initialValue: (optional, table) default starting value if no cached value
			is found.
		- \c quiesceUntilUpToDate: (optional, boolean) if true, then cached and intermediate
			values will be skipped. Observable will only produce an initial notification once
			the index store tables are up to date with the doc store at the time \c query is
			called. All subsequent updates will occur as changes come in.
		- \c nArgs: (optional, number) if present, specifies number of arguments
			(overrides #args)
		- \c encoder: (optional, string) The namespace and function to use to encode
			the sql results. E.g. to produce a binary asset info array.
		- \c tempIds: (optional, array of doc ids) if present, the table TEMP_ids will
			be populated with these ids for use in the query, e.g. as part of
			an join such as "INNER JOIN TEMP_ids ON TEMP_ids.docId = yourTable.docId"
			or via a where in clause such as "WHERE docId IN ( SELECT docId from TEMP_ids )"
			TEMP_ids is a temporary table with only one column, docId, and no special indexes.
		- \c singleColumnQuery: (optional, boolean) if present, the query will run via gatherArray,
			instead of gatherArrayOfArray, which is slightly more performant. The result will
			be a single array instead of an array of arrays, which should be more desirable for
			the calling code.
		- 1 .. n: (optional) query arguments to be substituted for ? placeholders in query

	\return Observable< table > An Observable whose value tracks query results, as an array of arrays.
		If singleColumnQuery is set, the return value that tracks the query results will be
		an array of values.

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

	\b WARNING: This behavior 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 indexStoreMethods:query( inArgs )

	check( 'WFRxIndexStore', 'table' )

	local enqueue = inArgs.enqueue

	local selector = enqueue2selector[ enqueue ]

	if not selector then
		if enqueue then
			error( "invalid enqueue parameter: " .. tostring( enqueue ) )
		end
		selector = enqueue2selector[ 'normal' ]
	end

	-- Fill in thread and traceback info if in debug build.

	local sql = assert( inArgs.sql, 'args.sql is required for :query()' )
	local nArgs = inArgs.nArgs or #inArgs
	local quiesceUntilUpToDate = assertOptionalBoolField( inArgs, 'quiesceUntilUpToDate' )
	local singleColumnQuery = assertOptionalBoolField( inArgs, 'singleColumnQuery' )

	local newArgs = {
		sql = sql,
		calm = assertOptionalPositiveNumberField( inArgs, 'calm' ),
		encoder = assertOptionalStringField( inArgs, 'encoder' ),
		asVirtualTable = assertOptionalBoolField( inArgs, 'asVirtualTable' ),
		tempIds = assertOptionalTableField( inArgs, 'tempIds' ),
		queryOnce = assertOptionalBoolField( inArgs, 'queryOnce' ),
		quiesceUntilUpToDate = quiesceUntilUpToDate,
		singleColumnQuery = singleColumnQuery,
		nArgs = nArgs,
		unpack( inArgs, 1, nArgs )
	}

	newArgs.threadName = DEBUG and WFCurrentThread.DEBUG_getThreadName() or nil
	newArgs.traceback = DEBUG and debug.traceback( '', 2 ) or nil

	local cacheKey = inArgs.cacheKey
	local cachedValue = getCachedResult( self, cacheKey )
	local haveCachedValue = cachedValue ~= nil
	local initialValue = cachedValue or inArgs.initialValue

	local function makeQueryRequest()

		if WFCloseSupport.hasBeenOrWillBeClosed( self ) then

			error( string.format( "query %s running after index store closed", sql ) )

		end

		local request = waitForWorkerThreadsToInitialize( self )
			:flatMap( function()
				return dispatchToReadOnlyWorkerThreadIfAvailableAndSupportsQuery( self, selector, newArgs )
			end )

		-- If we have a cache key, then we want to cache results.

		if cacheKey then
			request = request:tap( function( ... )
				storeCachedResult( self, cacheKey, ( ... ) )
			end )
		end

		return request

	end

	local function makeRequestWithDelay( delay )
		local obs
		if delay then
			obs = Observable.timer( 0.25, WFRxScheduler.sharedScheduler() )
		else
			obs = Observable.just()
		end

		return obs:map( function()
				return makeQueryRequest()
			end )
			:switch()
	end

	local delay = haveCachedValue and not quiesceUntilUpToDate

	if initialValue then
		return Observable.just( initialValue )
			:concat( makeRequestWithDelay( delay ) )
	else
		return makeRequestWithDelay( delay )
	end

end

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

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

	\param args (table) Please see the query API for a complete list of the supported arguments
--]]

function indexStoreMethods:queryOnce( args )

	check( 'WFRxIndexStore', 'table' )

	if args.calm then

		return Observable.throw( "Calm is not allowed when using the queryOnce API." )

	else

		args.queryOnce = true

		return self:query( args ):take( 1 )

	end
end

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

--[[!
	\brief Reports the result of a SQL query on the index store,
		and continues returning results every n seconds as an
		Observable.

	\param args (table) Please see the query API for a complete list of the supported arguments

	\param throttleTime <number> (seconds) The periodic time for the observable to check
		if it should yield new values

	\return (Observable< table > or Observable.throw) An Observable which tracks query results,
		as an array of arrays. The first return value of this observable will be present as soon as
		the provided query returns - subsequent returns are debounced. If singleColumnQuery is set,
		the return value that tracks the query results will be an array of values.

--]]

function indexStoreMethods:queryThenDebounce( --[[number (seconds)]] throttleTime,  --[[query args]] args)

	check( 'WFRxIndexStore', 'number', 'table' )

	if args.calm then

		return Observable.throw( "Calm is not allowed when using the queryThenDebounce API." )

	elseif args.throttleTime then

		return Observable.throw( "Please use queryThenDebounce instead of setting throttleTime directly" )

	elseif throttleTime < 0.125 then
		return Observable.throw( "To prevent misuse, the lower threshold for receiving updates with this API is 0.125 seconds" )

	else
		args.throttleTime = throttleTime

		return self:query( args )

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Inserts data into a manual table

	\param args (table) arguments for query
		- \c tableName: (string) name of the manual table
		- \c rows: (table) Array of values to be inserted into the manual table

	\return Observable< table > An Observable whose value tracks the successful
		insertion of the given rows

--]]

function indexStoreMethods:bulkInsertManualTable( tableName, rows )

	check( 'WFRxIndexStore', 'string', 'table' )

	assert( not WFCloseSupport.hasBeenOrWillBeClosed( self ),
			"query running after index store closed" )

	return dispatchToWorkerThread( self, 'bulkInsertManualTable', tableName, rows )
		:call( runUntilDone )

end

--------------------------------------------------------------------------------
--[[!
	\brief Clears the content of a manual table

	\param args (table) arguments for query
		- \c tableName: (string) name of the manual table

	\return Observable< table > An Observable whose value tracks where the manual
		table content got deleted

--]]

function indexStoreMethods:clearManualTable( tableName )

	check( 'WFRxIndexStore', 'string')

	assert( not WFCloseSupport.hasBeenOrWillBeClosed( self ),
			"query running after index store closed" )

	return dispatchToWorkerThread( self, 'clearManualTable', tableName )
		:call( runUntilDone )

end

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

--[[!
	\brief Reports as an Observable the results of a doing arbitrary boolean logic
		to combine a set of subqueries.  Each subquery must return a single
		column of integer results.  The column must be named 'id'.  The results
		must be sorted by id.

	\param args (table) arguments for query
		- \c operator: (string) either 'or' or 'and'
		- 1 .. n: subqueries to combine (as complete query arguments for :query() above)
			OR a table containing the key "operator" and array arguments for further
			boolean processing.

	\return Observable< table > An Observable whose value tracks query results, just
		like :query(), above.

--]]

function indexStoreMethods:querySetLogic( inArgs )

	return self:query {
		sql = kSetLogicQuery,
		calm = assertOptionalBoolField( inArgs, 'calm' ),
		inArgs,
	}

end

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

--[[!
	\brief Reports as an Observable the results of a doing arbitrary boolean logic
		to combine a set of subqueries.  Each subquery must return a single
		column of integer results.  The column must be named 'id'.  The results
		must be sorted by id.

	\param args (table) arguments for query
		- \c operator: (string) either 'or' or 'and'
		- 1 .. n: subqueries to combine (as complete query arguments for :query() above)
			OR a table containing the key "operator" and array arguments for further
			boolean processing.

	\return Observable< string > The name of the virtual table that contains the results.
		This table is updated whenever any of the subqueries generates new values.
		Furthermore, any query made against this virtual table will also update.

	Important: Any subscription to this must be disposed to avoid leaking virtual tables.

--]]

function indexStoreMethods:newVirtualTableForSetQuery( inArgs )

	return self:query {
		sql = kSetLogicQuery,
		calm = assertOptionalBoolField( inArgs, 'calm' ),
		asVirtualTable = true,
		inArgs,
	}

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves (as an Observable) the number of indexed and the total
	number needing indexing. By subtracting one from the other you can derive
	the same value reported by the old sizeOfIndexQueueB API.

	This can be used to approximate the time left until the index store
	catches up.

	\return (<code>Observable< number, number ></code>) A Behavior whose value is
		the number of revisions.
--]]

function indexStoreMethods:sizeOfIndexQueue()
	return dispatchToWorkerThread( self, 'sizeOfIndexQueue' )
end

--------------------------------------------------------------------------------
--[[!
	\brief Returns reindexing progress during index store initialization.

	\return (<code>WFProgressReporter</code>)
--]]

function indexStoreMethods:getReindexingProgress()
	return dispatchToWorkerThread( self, 'getReindexingProgress' )
end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves (as a Behavior) the number of documents left to index.

	This can be used to approximate the time left until the index store
	catches up.

	\return (<code>B< number ></code>) A Behavior whose value is the number of documents.
--]]

function indexStoreMethods:sizeOfIndexQueueB()

	return self:sizeOfIndexQueue()
		:map( function( progress, total )
			return total - progress
		end )
		:call( asKSEvent )
		:closingHoldB( 0 )

end

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

local function maybeCloseReadOnlyThreads( self )

	if self._indexStoreReadOnlyWorkerThreads then
		local roThreadCount = #self._indexStoreReadOnlyWorkerThreads
		for i = 1, roThreadCount do
			local workerThread = self._indexStoreReadOnlyWorkerThreads[ i ]
			workerThread:close()
			self._indexStoreReadOnlyWorkerThreads[ i ] = nil
		end
		self._indexStoreReadOnlyWorkerThreads = nil
	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Close this instance.

	\return An Observable which sends onCompleted when it is safe to reopen.
--]]

function indexStoreMethods:close()

	maybeCloseReadOnlyThreads( self )

	if self._indexStoreWorkerThread then
		self._indexStoreWorkerThread:close()
		self._indexStoreWorkerThread = nil
	end

	if self._platformIndexStore then
		self._platformIndexStore:close()
		self._platformIndexStore = nil
	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Permanently shuts down this index store.
--]]

function indexStoreMethods:halt()

	local result = dispatchToWorkerThread( self, 'halt' )

	local disposable
	disposable = result:subscribeOnCompleted( function()

		disposable:dispose()
		disposable = nil

		self:close()

	end )

	return result

end

--------------------------------------------------------------------------------
--[[!
	\brief Permanently shuts down this index store and deletes all related files.
--]]

function indexStoreMethods:haltAndDelete()

	local result = dispatchToWorkerThread( self, 'haltAndDelete' )

	local disposable
	disposable = result:subscribeOnCompleted( function()
		disposable:dispose()
		disposable = nil

		self:close()

	end )

	return result

end

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

--[[!
	\brief Retrieves the hashes of the map functions.
--]]

function indexStoreMethods:DEBUG_getMapFunctionHashes()

	-- Should be used only by WFRxIndexStoreTest.lua.

	return dispatchToWorkerThread( self, 'DEBUG_getMapFunctionHashes' )

end

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

--[[!
	\brief Retrieves the table info all tables.
--]]

function indexStoreMethods:DEBUG_getReindexingTableInfo()

	-- Should be used only by WFRxIndexStoreTest.lua.

	return dispatchToWorkerThread( self, 'DEBUG_getReindexingTableInfo' )

end

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

function indexStoreMethods:_DEBUG_simulateCorruption()

	return dispatchToWorkerThread( self, '_DEBUG_simulateCorruption' )

end

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

--[[!
	\brief Retrieves a table of metrics associated with version list forced
		reindexing. This function is only used by unit tests.

	\return (<code>Observable< table ></code>) An observable whose value is a
		table of metrics. If there was no forced reindexing, an empty table
		will be returned.
--]]

function indexStoreMethods:DEBUG_getForcedReindexingStats()

	return dispatchToWorkerThread( self, 'DEBUG_getForcedReindexingStats' )

end

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

--[[!
	\brief Set the tracked schema version info to specific sequence numbers.
		This function is only used by unit tests.

	\param currentVersionSequenceNumber (number) The tracked sequence number.
	\param inProgressVersionSequenceNumber (number) The in-progress sequence number;
		this is used to resume partial reindexing.

	\return (<code>Observable< true ></code>) An observable that fires with true.
--]]

function indexStoreMethods:DEBUG_setIndexStoreSchemaVersionInfo( currentVersionSequenceNumber, inProgressVersionSequenceNumber )

	return dispatchToWorkerThread( self, 'DEBUG_setIndexStoreSchemaVersionInfo', currentVersionSequenceNumber, inProgressVersionSequenceNumber )

end

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

--[[!
	\brief Retrieves the tracked schema version info from the index store.
		This function is only used by unit tests.

	\return (<code>Observable< number, number ></code>) An observable whose values
		are the current tracked sequence number, and the in-progress sequence number.
--]]

function indexStoreMethods:DEBUG_getIndexStoreSchemaVersionInfo()

	return dispatchToWorkerThread( self, 'DEBUG_getIndexStoreSchemaVersionInfo' )

end

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

--[[!
	\brief For the given list of tables, retrieves all relevant doc types/subtypes.
		This function is only used by unit tests.

	\return (<code>Observable< table ></code>) A map with doc types as
		keys and a list of subtypes as values.
--]]

function indexStoreMethods:DEBUG_getTypeSubtypeMapForTables( listOfTableNames )

	return dispatchToWorkerThread( self, 'DEBUG_getTypeSubtypeMapForTables', listOfTableNames )

end

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

function indexStoreMethods:DEBUG_dumpCoverage( path )

	-- Should be used only by WFRxIndexStoreTest.lua.

	local result = self._indexStoreWorkerThread:request( 'DEBUG_dumpCoverageP', path )

	local disposable
	disposable = result:subscribeOnCompleted( function()
		disposable:dispose()
	end )

	return result

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the platformIndexStore instance
--]]

function indexStoreMethods:DEBUG_getPlatformIndexStore()

	-- Should be used only by WFRxIndexStoreTest.lua.

	return self._platformIndexStore

end

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

return Rx.Class 'WFRxIndexStore'( indexStoreMethods )
