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

WFRxIndexStoreReadOnlyImp.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.

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

	-- Rx
local Rx = require 'Rx'

	-- WF.core.diagnostic_log
local WFDiagnosticLog = require 'WFDiagnosticLog'

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

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

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

	-- WF.sqlite.doc_store
local WFRxIndexStoreImpUtils = require 'WFRxIndexStoreImpUtils'

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

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

local kClosed = { __closed = "CLOSED" }

local encoderFn = KSFunctionUtils.memoizeFn( function( encoder )

	local _, _, ns, field = string.find( encoder, "([^.]+)%.([^.]+)" )
	assert( ns, "malformed encoder : no namespace" )
	assert( field, "malformed encoder : no field" )

	return assert( require( ns )[ field ], "No such field in namespace" )

end )

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

local WFRxIndexStoreReadOnlyImp = {

	_path = false,
		-- (string) path to DB file
	
	_db = false,
		-- (WFSQLiteDatabase) read only connection to the index store
	
	_indexSpecScript = false,
		-- (string) name of index store spec file
	
	_readOnlyThreadId = false,
		-- (number) simple integer identifying the read only thread
		
	_indexStoreSchema = false,
		-- (WFRxIndexStoreSchema)
		
	_indexStoreImpThread = false,
		-- the real index store read/write thread, necessary to ensure queries can get notified of updates
	
	_queryRequests = false,
		-- (table) current query Observable requests
			-- 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?
	
	_nQuerysCreated = 0,
		-- (number) how many query requests have ever been created on this thread
		
	_log = false,
		-- (WFLogger)
		
	WFCloseSupport.addCloseSupportMethods,

}

--[[!
	\brief Create a new read only index store thread.

	\param docStorePath (\c string) path to doc store to observe
	\param indexStoreSchema (\c WFRxIndexStoreSchema) index store schema
	\param readOnlyThreadId (\c number) thread id to uniquely identify the thread
--]]

function WFRxIndexStoreReadOnlyImp:__init( docStorePath, indexSpecScript, readOnlyThreadId )

	check( 'WFRxIndexStoreReadOnlyImp', '+string', '+string' )
	
	local indexStoreSchema = WFRxIndexStoreImpUtils.loadSchema( indexSpecScript )
	
	assert( indexStoreSchema, "Cannot load schema" )
	
	self._readOnlyThreadId = readOnlyThreadId
	
	self._indexStoreSchema = indexStoreSchema
	
	self._indexSpecScript = indexSpecScript
	
	-- NOTE: Catalog ID should have been validated by WFRxIndexStore
	-- client-side code.

	self._log = WFLogger( string.format( 'WFRxIndexStoreReadOnlyImp%s %s', readOnlyThreadId or 0, docStorePath ) )
		--:setLevel 'debug'

	WFCloseSupport.mustCloseAtThreadTermination( self )

	WFDiagnosticLog.addLogProvider( self )

	self._queryRequests = {}
	
	local initArgs = string.format( '%s=%s', indexSpecScript, docStorePath )
	self._indexStoreImpThread = WFWorkerThread( 'WFRxIndexStoreImp.lua', initArgs )
	
end

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

local function executeQuery( self, sql, singleColumnQuery, encoder, ... )
	
	if encoder then

		return self._db:encodeResults( sql, encoderFn( encoder ), ... )

	elseif singleColumnQuery then
	
		return self._db:gatherArrayWithChecksum( sql, ... )
	
	else
	
		return self._db:gatherArrayOfArrayWithChecksum( sql, ... )
		
	end

end

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

function WFRxIndexStoreReadOnlyImp:watchForTableUpdatesForSql( sql, quiesceUntilUpToDate )

	return self._indexStoreImpThread:request( 'watchForTableUpdatesForSql', sql, quiesceUntilUpToDate )

end

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

--[[!
	\brief Reports as an Observable the results of a SQL query on the index store.
	
	\b WARNING: Read only thread implementation does not support the use of tempIds or
	any temp/virtual tables. Those queries should be routed to the regular 
	WFRxIndexStoreImp worker thread.
	
	
	\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.
		- 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.

	\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 WFRxIndexStoreReadOnlyImp:query( args )

	check( 'WFRxIndexStoreReadOnlyImp', 'table' )
	
	local encoder = args.encoder
	if encoder ~= nil then assert( type( encoder ) == 'string' ) end
	
	-- Fetch query results
	args.nUpdatesSent = 0

	local nQueryArgs = args.nArgs or #args
	local queryResultsNow = function( sql )
		args.nUpdatesSent = args.nUpdatesSent + 1
		local singleColumnQuery = args.singleColumnQuery
		return executeQuery( self, sql, singleColumnQuery, encoder, unpack( args, 1, nQueryArgs ) )
	end
	
	self._nQuerysCreated = self._nQuerysCreated + 1
	
	return WFRxIndexStoreImpUtils.query( self, queryResultsNow, args )
	
end

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

function WFRxIndexStoreReadOnlyImp:close()
	
	self._indexStoreImpThread:close()
	self._indexStoreImpThread = nil
	
end

--------------------------------------------------------------------------------
--[[!
	\brief Initializes the read only implementation. This must be called by the owning thread
	before any queries are sent to this worker.
	
	\return Observable< boolean > returns true once initialization is complete
--]]
function WFRxIndexStoreReadOnlyImp:initializeDb()
	
	return self._indexStoreImpThread:request( 'dbPath' )
		:map( function( dbPath )
			
			self:setOwned {
				_db = require 'WFSQLiteDatabase'( dbPath, self._indexStoreSchema:schema(), { readOnly = true } )
			}
	
			self._path = dbPath
			
			return true
		end )
		
end

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

function WFRxIndexStoreReadOnlyImp:sendDiagnosticLogData()

	-- Send information about query requests that are outstanding.

	local filteredQueryRequests = WFRxIndexStoreImpUtils.filterAndFormatQueryRequestStats( self._queryRequests )

	local nQuerysCreated = self._nQuerysCreated

	-- Send information about index update performance.
	
	return {

		name = string.format( "Read-Only Index Store Query Info :: %s=%s=%s", self._readOnlyThreadId, self._indexSpecScript, self._path ),
		id = 'WFRxIndexStoreReadOnlyImp-' .. WFCurrentThread.DEBUG_getThreadName():replace( ' :: ', '+' ),
		weight = 1850,
		parentSection = WFDiagnosticLog.Sections.INDEX_STORE,
		
		body = {

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

		},

	}

end

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

WFRxIndexStoreReadOnlyImp = Rx.Class 'WFRxIndexStoreReadOnlyImp'( WFRxIndexStoreReadOnlyImp )

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

local initMessageProcessor = function(docStoreAndIndex)

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

	local indexStore = WFRxIndexStoreReadOnlyImp( docStoreId, indexSpecScript, readOnlyThreadId )
	
	local isInitialized = false
	WFRequestDispatcher.onMessage( 'initIndexStoreReadOnly', function( docStoreId, indexSpecScript )
		
		if not isInitialized then
			isInitialized = true
			return indexStore:initializeDb()
		else
			return Rx.Observable.just( true)
		end
		
	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 read only threadnot yet inited' )
		assert( indexStore ~= kClosed, 'index store closed' )
		local oldIndexStore = indexStore
		indexStore = kClosed
		return oldIndexStore:close()

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

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

return {
	initMessageProcessor = initMessageProcessor,
}
