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

WFRxIndexStoreImp.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.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

	-- WF.core.logging
local WFLogger = require 'WFLogger'

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

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

	-- WF.core.utils
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCapabilities = require 'WFCapabilities'
local WFCloseSupport = require 'WFCloseSupport'
local WFPlatformInfo = require 'WFPlatformInfo'

	-- WF.sqlite.doc_store
local WFRxDocStore = require 'WFRxDocStore'
local WFRxIndexStoreMapServer = require 'WFRxIndexStoreMapServer'
local WFRxIndexStoreQueryServer = require 'WFRxIndexStoreQueryServer'
local WFSQLiteSetLogicImp = require 'WFSQLiteSetLogicImp'

local WFRxIndexStoreImpUtils = require 'WFRxIndexStoreImpUtils'


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

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

local kClosed = { __closed = "CLOSED" }


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

local WFRxIndexStoreImp = {

	_path = false,
		-- (string) path to DB file

	_indexSpecScript = false,
		-- (string) name of index store spec file

	_mapServer = false,
		-- (WFRxIndexStoreMapServer) map server

	_queryServer = false,
		-- (WFRxIndexStoreQueryServer) query server

	_log = false,
		-- (WFLogger)

	_nQuerysCreated = false,
		-- (number) how many query requests have ever been created on this thread

	_virtualSetTableCounter = false,
		-- (int) used for making up unique names for virtual tables

	_DEBUG_indexStoreSchema = false,
		-- (WFRxIndexStoreSchema) Only tracked and used by unit tests.

	WFCloseSupport.addCloseSupportMethods,

}

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

--
-- Use WFSQLiteSetLogic ResultSet object to combine the results of several queries.
--
local function booleanLogic( imp, operator, subqueries )

	local allQueries = {}
	for index, subquery in ipairs( subqueries ) do
		if subquery.operator then
			allQueries[ index ] = booleanLogic( imp, subquery.operator, subquery )
		else
			local copy = KSTableUtils.shallowcopy( subquery )
			copy.encoder = 'WFSQLiteSetLogicImp.encodeSet'

			allQueries[ index ] = imp:query( copy )
		end
	end

	return Observable.combineLatest( function( ... )
		local result = ( ... )
		for i = 2, select( '#', ... ) do
			local other = select( i, ... )
			local newResult
			if operator == 'and' then
				newResult = result:setWithIntersection( other )
			elseif operator == 'or' then
				newResult = result:setWithUnion( other )
			else
				error( 'Invalid operator ' .. operator )
			end
			result = newResult
		end
		return result
	end, unpack( allQueries ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Create a new index store implementation.

	Should only be used by index definition files.

	\param docStorePath (\c string) path to doc store to observe
	\param indexStoreSchema (\c WFRxIndexStoreSchema) index store schema
--]]

function WFRxIndexStoreImp:__init( docStorePath, indexSpecScript )

	check( 'WFRxIndexStoreImp', '+string', '+string' )

	local indexStoreSchema = WFRxIndexStoreImpUtils.loadSchema( indexSpecScript )

	assert( indexStoreSchema, "Cannot load schema" )

	self._indexSpecScript = indexSpecScript
	self._virtualSetTableCounter = 1
	self._nQuerysCreated = 0

	-- NOTE: Catalog ID should have been validated by WFRxIndexStore
	-- client-side code.

	self._log = WFLogger( string.format( 'WFRxIndexStoreImp %s', docStorePath ) )
		--:setLevel 'debug'

	self._log:debugf( "initializing with docStore = %s", docStorePath )

	local docStore = WFRxDocStore( docStorePath )

	-- self:addCloseTarget( docStore )
		-- map server will close

	WFCloseSupport.mustCloseAtThreadTermination( self )

	WFDiagnosticLog.addLogProvider( self )

	-- Register event processor event handlers.

	-- Create database in appropriate location.

	local dbPath

	if docStorePath:sub( 1, 8 ) == ':memory:' then

		dbPath = ':memory:'

	elseif docStorePath:sub( 1, 6 ) == ':temp:' then

		dbPath = ':temp:'

	else

		dbPath = docStore:path():gsub( '%.[^./\\]*$', '' ) .. '.wfindex'

	end

	self._path = dbPath

	self._log:debugf( "WFRxIndexStoreImp.start, dbPath = %s", dbPath )

	-- If we fail trying to open the database, just trash it and start again.
	-- [scouten 2012-03-06, PR# 3107228]

	-- Or if the capability is set to trigger this, e.g. for testing how
	-- fast we are at doing a rebuild.

	local forceRebuild = "rebuild" == WFCapabilities.getCapability "WFRxIndexStoreImp.rebuildIndexStore"

	local db
	if forceRebuild or not pcall( function()

		db = indexStoreSchema:openOrCreateDatabaseAtPath( dbPath )

	end ) then

		if not forceRebuild then
			self._log:warn( "WFRxIndexStore was corrupt. Deleting and rebuilding." )
		end

		-- Trash the index DB and its possible sidecars.

		WFFileUtils.deleteFile( dbPath )
		WFFileUtils.deleteFile( dbPath .. '-shm' )
		WFFileUtils.deleteFile( dbPath .. '-wal' )

		-- Try again. If we fail this time, that's it, man, game over, it's game over.

		db = indexStoreSchema:openOrCreateDatabaseAtPath( dbPath )

	end

	if WFPlatformInfo.isUnitTest() then
		self._DEBUG_indexStoreSchema = indexStoreSchema
	end

	-- Spin up map server and query server.

	self:setOwned { _mapServer = WFRxIndexStoreMapServer( docStore, db, indexStoreSchema ) }

	self:setOwned {
		_queryServer = WFRxIndexStoreQueryServer( self._mapServer, indexStoreSchema )
	}

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 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 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 nArgs: (optional, number) number of query arguments (helps with \c nil values
			in the middle of the list)
		- \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 B< table > query results (array of arrays syntax). If singleColumnQuery is set,
		the return value 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 WFRxIndexStoreImp:query( --[[table]] args )

	check( 'WFRxIndexStoreImp', 'table' )

	if args.sql == '< set logic query >' then
		return self:setLogicQuery( args )
	end

	self._nQuerysCreated = self._nQuerysCreated + 1

	return self._queryServer:query( args )

end

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

function WFRxIndexStoreImp:bulkInsertManualTable( tableName, rows )

	check( 'WFRxIndexStoreImp', 'string', 'table' )

	return self._queryServer:bulkInsertManualTable( tableName, rows )

end

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

function WFRxIndexStoreImp:clearManualTable( --[[string]] tableName )

	check( 'WFRxIndexStoreImp', 'string' )

	return self._queryServer:clearManualTable( tableName )

end

--------------------------------------------------------------------------------
--[[!
	\brief Return a set for some boolean combination of queries

	\return (<code>Observable< ResultSet or virtual table name ></code>) The result
		set or the name of the virtual table containing the results.
--]]

function WFRxIndexStoreImp:setLogicQuery( args )

	local setObject = booleanLogic( self, args[ 1 ].operator, args[ 1 ] )
		-- An Observable that vends the result set

	if args.asVirtualTable then

		local subscribed = false
		local virtualTableNumber = self._virtualSetTableCounter
		self._virtualSetTableCounter = self._virtualSetTableCounter + 1
		local setUsedByVirtualTable = WFSQLiteSetLogicImp.emptySet()
		local tableName = string.format( "temp.ResultSetVirtualTable_%d", virtualTableNumber )

		-- runWithNativeDb will throw if self._mapServer is closed. That's
		-- appropriate here. If some object is subscribing to this observable
		-- after mapserver is close, we want an exception
		local tableArg =
			self._mapServer
				:runWithNativeDb(
					function( db )

						return setUsedByVirtualTable:_prepareForVirtualTable( db )

					end )

		setObject = setObject:replay( nil, 1 ):refCount()

		local isUpToDate = setObject
				:map( function()
					return 0
						-- FIXME (arahn 07/14/2015) Need to track the change number
						-- for which subqueries are up to date and report the minimum
						-- of that here. Until that's done, quiesceUntilDone won't
						-- work properly on virtual tables.
				end )

		self._queryServer._virtualTables[ tableName ] = isUpToDate

		local virtualTable = Observable.create( function( observer )

			assert( not subscribed, 'Can only subscribe once' )
					-- In order to properly track the virtual table lifetime

			subscribed = true

			local createStmt = string.format(
					[[CREATE VIRTUAL TABLE %s USING WFSQLite_SetTable( %s )]],
					tableName,
					tableArg )

			-- runWithNativeDb will throw if self._mapServer is closed. That's
			-- appropriate here. If some object is subscribing to this observable
			-- after mapserver is close, we want an exception
			self._mapServer
				:runWithNativeDb(
					function( db )

						if not pcall( db.execOnce, db, createStmt ) then
							local dropStmt = string.format(
									[[DROP TABLE %s]],
									tableName )
							if pcall( db.execOnce, db, dropStmt ) then
								local success, maybeError = pcall( db.execOnce, db, createStmt )
								if not success then
									-- throw errorMessage when it fails to create virtual table.
									local errorMessage = string.format( "Failed to create virtual table %s: %s", tableName, maybeError )
									observer:onError( errorMessage )
									return
								end
							end
						end

					end )

			observer:onNext( tableName, setUsedByVirtualTable )


			return function()

				if WFCloseSupport.hasBeenOrWillBeClosed( self._mapServer ) then

					return

				end

				-- this may run at shutdown, so it might be called after the
				-- manpserver has closed. That's not a big deal: this code
				-- drops temp tables which would happen anyway when the mapserver
				-- closed the underlying db, so using maybeRunWithNativeDb is
				-- appropriate
				self._mapServer
					:maybeRunWithNativeDb(
						function( db )
							if not WFCloseSupport.hasBeenOrWillBeClosed( db ) then
								db:execOnce( [[DROP TABLE ]] .. tableName )
							end
						end )
			end

		end )

		local resultCount = 0

		return Observable.combineLatest( function( nextResultSet, tableName, setUsedByVirtualTable )
			if nextResultSet == setUsedByVirtualTable then
				-- skip copy
			else
				setUsedByVirtualTable:copyFrom( nextResultSet )
				resultCount = resultCount + 1
			end
			return tableName, resultCount
		end, setObject, virtualTable )

	else
		return setObject
	end

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 WFRxIndexStoreImp:watchForTableUpdatesForSql( sql, quiesceUntilUpToDate )
	check( 'WFRxIndexStoreImp', '+string', 'boolean' )

	return self._queryServer:watchForTableUpdatesForSql( sql, quiesceUntilUpToDate )
end

--------------------------------------------------------------------------------
--[[!
	\brief Return the number of documents left to index.

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

	\return (<code>Observable< number ></code>) how many documents left to index?
--]]

function WFRxIndexStoreImp:sizeOfIndexQueue()

	return self._mapServer:sizeOfIndexQueue()

end

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

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

function WFRxIndexStoreImp:getReindexingProgress()

	return self._mapServer:getReindexingProgress()

end

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

function WFRxIndexStoreImp:halt()

	self:close() -- should cause DB object and friends to close

	return Observable.empty()

end

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

function WFRxIndexStoreImp:haltAndDelete()

	local path = self._path

	self:close() -- should cause DB object and friends to close

	if path then
		pcall( WFFileUtils.delete, path )
		pcall( WFFileUtils.delete, path .. '-shm' )
		pcall( WFFileUtils.delete, path .. '-wal' )
	end

	return Observable.empty()

end

--------------------------------------------------------------------------------
--[[!
	\brief Creates a copy of the index store database to a new db at path.

	\param path (string) filename where the new db should be created.

	\return <code>Observable<number> stream of values relaying progress of
		the backup as a fraction indicating percent complete.
]]--
function WFRxIndexStoreImp:onlineBackup( path )

	check( 'WFRxIndexStoreImp', '+string')

	local WFSQLiteDatabaseUtils = require 'WFSQLiteDatabaseUtils'

	return self._mapServer:runWithNativeDb( function( db )
		return WFSQLiteDatabaseUtils.onlineBackupDbToPathWithProgressStream( db, path )
	end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Get a stream that fires with SQLite error codes when an index store operation
		encounters a critical error. Possible error codes include:

		10 : Some kind of disk I/O error occurred
		11 : The database disk image is malformed
		13 : Insertion failed because database is full

	\return <code>Observable<number>
]]--

function WFRxIndexStoreImp:getSQLiteFatalErrorCodeStream()

	check( 'WFRxIndexStoreImp' )

	return self._mapServer:getSQLiteFatalErrorCodeStream()

end

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

function WFRxIndexStoreImp:DEBUG_getMapFunctionHashes()

	return self._mapServer:DEBUG_getMapFunctionHashes()

end

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

function WFRxIndexStoreImp:DEBUG_getReindexingTableInfo()

	if self._DEBUG_indexStoreSchema then
	
		local info = {}
	
		for tableName, tableInfo in pairs( self._DEBUG_indexStoreSchema._tableInfo ) do
	
			local reindexingTableInfo = {
			
				sqlSchemaHash = tableInfo.sqlSchemaHash,
				mapFunctionHash = tableInfo.mapFunctionHash,
				docValueFilter = tableInfo.docValueFilterHash,
				relevantTypeSubtypeTuples = tableInfo.relevantTypeSubtypeTuples,
				includeDocumentsFromSource = tableInfo.includeDocumentsFromSource
				
			}
			
			info[ tableName ] = reindexingTableInfo
	
		end
	
		return Observable.just( info )
		
	else
	
		return Observable.just( {} )
		
	end

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 WFRxIndexStoreImp:DEBUG_getForcedReindexingStats()

	return self._mapServer: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 WFRxIndexStoreImp:DEBUG_setIndexStoreSchemaVersionInfo( currentVersionSequenceNumber, inProgressVersionSequenceNumber )

	return self._mapServer: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 WFRxIndexStoreImp:DEBUG_getIndexStoreSchemaVersionInfo()

	return self._mapServer: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 WFRxIndexStoreImp:DEBUG_getTypeSubtypeMapForTables( listOfTableNames )

	return self._DEBUG_indexStoreSchema
			and self._DEBUG_indexStoreSchema:getTypeSubtypeMapForTables( listOfTableNames )

end

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

function WFRxIndexStoreImp:_DEBUG_simulateCorruption()

	self._mapServer:_DEBUG_simulateCorruption()

end

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

function WFRxIndexStoreImp:sendDiagnosticLogData()

	-- Send information about query requests that are outstanding.

	local filteredQueryRequests = self._queryServer:queryStats()

	local nQuerysCreated = self._nQuerysCreated

	-- Send information about index update performance.

	return {

		name = string.format( "Index Store Query Info :: %s=%s", self._indexSpecScript, self._path ),
		id = 'WFRxIndexStoreImp-' .. WFCurrentThread.DEBUG_getThreadName():replace( ' :: ', '+' ),
		weight = 1800,
		parentSection = WFDiagnosticLog.Sections.INDEX_STORE,

		body = {

			path = self._path,
			["Total Query Count"] = nQuerysCreated,
			["Open Query Count"] = #filteredQueryRequests,
			["Open Query Details"] = filteredQueryRequests,

		},

	}

end

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

function WFRxIndexStoreImp:dbPath()
	return Observable.just( self._path )
end

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

WFRxIndexStoreImp = Rx.Class 'WFRxIndexStoreImp'( WFRxIndexStoreImp )

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

local initMessageProcessor = function(docStoreAndIndex)

	local _, _, indexSpecScript, docStoreId = string.find( docStoreAndIndex, "^([^=]+)=(.+)$" )

	local indexStore = WFRxIndexStoreImp( docStoreId, indexSpecScript )

	WFRequestDispatcher.onMessage( 'initIndexStore', function( docStoreId, indexSpecScript )

		return true
			-- noOp

	end )

	local messageForwarder = setmetatable({}, {
		__index = function( _, methodName )
			return function( self, ... )
				assert( indexStore, 'index store not yet inited' )
				assert( indexStore ~= kClosed, 'index store closed' )
				local method = assert( indexStore[ methodName ], 'No such method on index store' )
				return method( indexStore, ... )
			end
		end,
	})

	WFRequestDispatcher.onMessage( 'close', function( indexStoreIndex )

		assert( indexStore, 'index store not yet inited' )
		assert( indexStore ~= kClosed, 'index store closed' )
		local oldIndexStore = indexStore
		indexStore = kClosed
		return oldIndexStore:close()

	end )

	WFRequestDispatcher.onMessageDo {
		message = 'query',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'setLogicQuery',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'bulkInsertManualTable',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'clearManualTable',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'sizeOfIndexQueue',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'getReindexingProgress',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'onlineBackup',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'watchForTableUpdatesForSql',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'haltAndDelete',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = '_DEBUG_simulateCorruption',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_getForcedReindexingStats',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_setIndexStoreSchemaVersionInfo',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_getIndexStoreSchemaVersionInfo',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_getTypeSubtypeMapForTables',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'halt',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'dbPath',
		target = messageForwarder,
	}

	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_getMapFunctionHashes',
		target = messageForwarder,
	}
	
	WFRequestDispatcher.onMessageDo {
		message = 'DEBUG_getReindexingTableInfo',
		target = messageForwarder,
	}
	
	WFRequestDispatcher.onMessage( 'DEBUG_dumpCoverageP', function( path )
		require 'SetupCoverage'.dumpCoverage( path )

		return true
	end )

	WFRequestDispatcher.enableDiagnosticLogForRequestStats()

end

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

return {
	initMessageProcessor = initMessageProcessor,
}
