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

WFRxPlatformIndexStoreImp.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2020 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.logging
local WFLogger = require 'WFLogger'

local WFWorkerThread = require 'WFWorkerThread'
local WFRequestDispatcher = require 'WFRequestDispatcher'

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

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

local WFRxScheduler = require 'WFRxScheduler'


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

local log = WFLogger( "WFRxPlatformIndexStoreImp" )
		--:setLevel 'debug'

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

local kClosed = { __closed = "CLOSED" }

local gActiveIndexStores = {}

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

local function watchForUpdates( self, sql, calm )
	self._watchSet[ sql ] = self:watchForTableUpdatesForSql( sql, false )
		:calm( calm, WFRxScheduler.sharedScheduler() )
		:subscribe( function()
			log:debugf( "Table updated for sql: %s", sql )
			WFPlatformDataStoreNative.handleUpdates( self._path, sql )
		end )
end

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

local function watchHandler( path, sql, calm )
	log:debugf( "watchHandler called for path:%s, sql:%s", path, sql )
	local platformIndexStore = gActiveIndexStores[ path ]
	log:debugf( "watchHandler platformIndexStore:%s", platformIndexStore )
	if not platformIndexStore then
		return
	end
	
	watchForUpdates( platformIndexStore, sql, calm )
end

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

local function unwatchHandler( path, sql )
	log:debugf( "unwatchHandler called for path:%s, sql:%s", path, sql )
	local platformIndexStore = gActiveIndexStores[ path ]
	log:debugf( "unwatchHandler platformIndexStore:%s", platformIndexStore )
	if not platformIndexStore then
		return
	end
	
	local watchSubscription = platformIndexStore._watchSet[ sql ]
	log:debugf( "unwatchHandler watchSubscription:%s", watchSubscription )
	if watchSubscription then
		platformIndexStore._watchSet[ sql ] = nil
		watchSubscription:dispose()
	end
end

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

WFPlatformDataStoreNative.registerWatchHandler( watchHandler )
WFPlatformDataStoreNative.registerUnwatchHandler( unwatchHandler )

--============================================================================--
--[[!
	\class WFRxPlatformIndexStoreImp
	\brief The imp corresponding to WFRxPlatformIndexStore. 
--]]

local WFRxPlatformIndexStoreImp = {

	_path = false,
		-- (string) path to DB file
	
	_indexStoreImpThread = false,
		-- the real index store read/write thread, necessary to ensure queries can get notified of updates
		
	_watchSet = false,
	
	WFCloseSupport.addCloseSupportMethods,

}

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

	\param docStorePath (\c string) path to doc store to observe
	\param indexStoreSchema (\c WFRxIndexStoreSchema) index store schema
	
	The parameters should match the ones needed for WFRxIndexStoreImp.lua
--]]
function WFRxPlatformIndexStoreImp:__init( docStorePath, indexSpecScript )

	check( 'WFRxPlatformIndexStoreImp', '+string', '+string' )
		
	-- Create database in appropriate location.

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

	self._path = dbPath

	log:debugf( "WFRxPlatformIndexStore.open, dbPath = %s", dbPath )
	
	local platformIndexStoreData = WFPlatformDataStoreNative.open( self._path, "indexstore" )
	rawset( self, "platformIndexStoreData", platformIndexStoreData )
	
	local initArgs = string.format( '%s=%s', indexSpecScript, docStorePath )
	self._indexStoreImpThread = WFWorkerThread( 'WFRxIndexStoreImp.lua', initArgs )
	
	gActiveIndexStores[ self._path ] = self
	
	self._watchSet = {}

end

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

function WFRxPlatformIndexStoreImp:watchForTableUpdatesForSql( sql, quiesceUntilUpToDate )

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

end

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

function WFRxPlatformIndexStoreImp:close()
	gActiveIndexStores[ self._path ] = nil
	
	WFPlatformDataStoreNative.close( self._path )
	rawset( self, "platformIndexStoreData", nil )
	
	if self._watchSet then
		for k, v in pairs( self._watchSet ) do
			self._watchSet[ k ] = nil
			v:dispose()
		end
		self._watchSet = nil
	end
end

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

local WFRxPlatformIndexStoreImpClass = Rx.Class 'WFRxPlatformIndexStoreImp'( WFRxPlatformIndexStoreImp )

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

local initMessageProcessor = function(docStoreAndIndex)
	
	local _, _, indexSpecScript, docStoreId = string.find( docStoreAndIndex, "^([^=]+)=(.+)$" )
	
	local indexStore = WFRxPlatformIndexStoreImpClass( docStoreId, indexSpecScript )
	WFCloseSupport.mustCloseAtThreadTermination( indexStore )
	
	WFRequestDispatcher.onMessage( 'init', function()
		
		return true
		
	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 )

end

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

return {
	initMessageProcessor = initMessageProcessor,
}
