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

WFRxPlatformIndexStoreImp.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.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,
}
