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

WFRxIndexStoreMapServer.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 BehaviorSubject = Rx.BehaviorSubject
local Observable = Rx.Observable

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

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

	-- WF.core.utils
local KSFunctionUtils = require 'KSFunctionUtils'
local KSIdGenerator = require 'KSIdGenerator'
local KSLuaUtils = require 'KSLuaUtils'
local KSTableUtils = require 'KSTableUtils'
local WFAnalytics = require 'WFAnalytics'
local WFArguments = require 'WFArguments'
local WFCapabilities = require 'WFCapabilities'
local WFCloseSupport = require 'WFCloseSupport'
local WFConfiguration = require 'WFConfiguration'
local WFCurrentThread = require 'WFCurrentThread'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFPerformanceUtils = require 'WFPerformanceUtils'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxScheduler = require 'WFRxScheduler'
local WFProgressReporter = require 'WFProgressReporter'
local WFTableArchiving = require 'WFTableArchiving'

	-- WF.sqlite
local WFRxIndexStoreSchema = require 'WFRxIndexStoreSchema'
local WFRxIndexStoreSchemaVersionList = require 'WFRxIndexStoreSchemaVersionList'
local WFSQLiteUtils = require 'WFSQLiteUtils'

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

local WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithDynamicDescription =
		assert( WFPerformanceUtils.wrapInLogLongExecutionInDiagnosticLogWithDynamicDescription,
				 "No WFPerformanceUtils.wrapInLogLongExecutionInDiagnosticLogWithDynamicDescription" )

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

local kDefaultBatchSizeForDocMapping = 500
local kMaxBatchSizeForDocMapping = 1000

local kLocalState_SchemaVersionSequence =  'schemaVersionSequence'
local kLocalState_InProgressReindexSchemaVersionSequence = 'inProgressReindexSchemaVersionSequence'

local kResumeReindexingEvent = 'resume-reindexing-event'
local kInvalidIndexStoreReindexingEvent = 'invalid-or-missing-indexstore-reindexing-event'
local kStandardReindexingEvent = 'standard-reindexing-event'
local kCompletedReindexingEvent = 'completed-reindexing-event'
local kPercentageToReindexingEvent = 'percentage-to-reindexing-event'

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

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

local WFFunctionUtils_reduce = WFFunctionUtils.reduce

local KSTableUtils_derefOpt = KSTableUtils.derefOpt

local logger = WFLogger 'WFRxIndexStoreMapServer'
	--:setLevel 'debug'

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

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 bringTableUpToDateWithChangedDocs 
	-- Forward declaration
local transactInDB
	-- Forward declaration
local function accumulateAndRemoveTables()

	local t = {}

	return function( addTable, removeTable )
		if addTable then
			t[ addTable ] = true
		end
		if removeTable then
			t[ removeTable ] = nil
		end
		local result = {}
		for k,v in pairs( t ) do
			result[ k ] = v
		end
		return result
	end

end

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

-- startChangeCounter is an Observable which fires when last batch was done
-- being processed
local function getChangedDocsThrough( docStore, startChangeCounter, endChangeCounter, 
										batchSize, cache, shouldIncludeDocumentsWithSpaceIds,
										shouldIncludeDocumentsWithoutSpaceIds,
										maybeFilterByTypeSubtypeMap )
	
	local includeDocumentsFromSource = WFRxIndexStoreSchema.sourceCatalogDocuments
	
	if shouldIncludeDocumentsWithSpaceIds and shouldIncludeDocumentsWithoutSpaceIds then
	
		includeDocumentsFromSource = WFRxIndexStoreSchema.sourceAllDocuments
	
	elseif shouldIncludeDocumentsWithSpaceIds then
	
		includeDocumentsFromSource = WFRxIndexStoreSchema.sourceSpaceFeedDocuments
	
	elseif shouldIncludeDocumentsWithoutSpaceIds then
	
		includeDocumentsFromSource = WFRxIndexStoreSchema.sourceCatalogDocuments
	
	end
	
	return startChangeCounter:selectMany( function( fromChangeCounter )

		local result
		local prefetchChangeCounter = cache.fromChangeCounter

		if prefetchChangeCounter and prefetchChangeCounter == fromChangeCounter then
			result = cache.changedDocsPrefetch
		else
			result = docStore:changedDocs( {
				fromChangeCounter = fromChangeCounter,
				limit = batchSize,
				includeDocumentsFromSource = includeDocumentsFromSource,
				filterByTypeSubtypeMap = maybeFilterByTypeSubtypeMap,
			}, 'IndexStore_reindex' )
		end

		cache.fromChangeCounter, cache.changedDocsPrefetch = nil, nil

		return result
			:tap( function( changedDocs )
				local continueFrom = changedDocs.continueFrom
				if continueFrom and continueFrom < endChangeCounter then

					local changedDocsPrefetch = docStore:changedDocs( {
							fromChangeCounter = continueFrom,
							limit = batchSize,
							includeDocumentsFromSource = includeDocumentsFromSource,
							filterByTypeSubtypeMap = maybeFilterByTypeSubtypeMap,
						}, 'IndexStore_reindex_cont' )
						:replay( nil, 1 )
							-- Cache the result; we'll subscribe right away
							-- (as a consequence of calling :connect()) and
							-- later if this Observable is pulled from the 
							-- cache, this :selectMany() will also subscribe.

					changedDocsPrefetch:connect()
						-- Activate it now so it starts the prefetch

					changedDocsPrefetch = changedDocsPrefetch
						:refCount()
							-- Finish enabling the caching and multiple subscriptions.
							-- If we don't end up using this from the cache then
							-- this was just wasted effort.

					cache.fromChangeCounter = continueFrom
					cache.changedDocsPrefetch = changedDocsPrefetch

				end
			end )
	end )

end

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

local function parseTablesNeedingIndexing( self, tablesNeedingIndexing,
											changeCounter,
											tablesThatNeedAggressiveUpdates,
											ignoreTablesThatDontExist,
											needsUpdateForTable, fastReindexingLimit )

	-- The goal of this logic is to ensure that we issue as few changedDocs
	-- requests as possible. We only allow one changedDocs request to be
	-- issued at a time, so we examine the state of each table that needs
	-- updating and back up to its starting point for the next request.

	local minNeedsUpdateFrom
	local combinedIndexingTable = {}
	local tableHasNewIndexData = {}
	
	for _, tableName in ipairs( tablesNeedingIndexing ) do
		
		-- There may be old tables in the index store on disk that are no 
		-- longer in the schema. If they're not, we have no way to update 
		-- them so don't list them for update
		
		if self._tableNames[ tableName ] then
		
			combinedIndexingTable[ tableName ] = true
			tableHasNewIndexData[ tableName ] = true
			
		else
		
			logger:debugf( "skipping updates for table = %s, which no longer exists.", tableName )
		
		end
		
	end
	
	for tableName in pairs( tablesThatNeedAggressiveUpdates or {} ) do
		combinedIndexingTable[ tableName ] = true
		tableHasNewIndexData[ tableName ] = true
	end
	
	local shouldIncludeDocumentsWithSpaceIds = false
	local shouldIncludeDocumentsWithoutSpaceIds = false
	
	for tableName in pairs( combinedIndexingTable ) do

		local needsUpdateFrom = needsUpdateForTable[ tableName ]
		
		if needsUpdateFrom then
		
			local needsUpdate = type( needsUpdateFrom ) == 'number' and needsUpdateFrom <= changeCounter
			
			tableHasNewIndexData[ tableName ] = needsUpdate
			
			if needsUpdate then

				logger:debugf( "table %s needs update from %s", tableName, needsUpdateFrom )

				shouldIncludeDocumentsWithSpaceIds = shouldIncludeDocumentsWithSpaceIds 
						or self._tablesThatShouldIncludeDocumentsWithSpaceIds[ tableName ]
						
				shouldIncludeDocumentsWithoutSpaceIds = shouldIncludeDocumentsWithoutSpaceIds 
						or self._tablesThatShouldIncludeDocumentsWithoutSpaceIds[ tableName ]

				if not minNeedsUpdateFrom or minNeedsUpdateFrom > needsUpdateFrom then
					minNeedsUpdateFrom = needsUpdateFrom
				end

			end
		
		else
		
			local tableDoesntExistWarning = string.format( "table %s needs update but it doesn't exist within needsUpdateForTable.", tableName )
		
			if ignoreTablesThatDontExist then
				logger:infof( tableDoesntExistWarning )
			else
				logger:errorf( tableDoesntExistWarning )
			end
			
		end

	end

	local canSkipRowLookup = false

	if minNeedsUpdateFrom and minNeedsUpdateFrom <= fastReindexingLimit then
		-- We are in the (forced) re-indexing phase and we can and need to skip row lookups
		-- due to the dropped SQL indicies.
		canSkipRowLookup = true
		
	end

	return combinedIndexingTable, tableHasNewIndexData, minNeedsUpdateFrom, canSkipRowLookup,
				shouldIncludeDocumentsWithSpaceIds, shouldIncludeDocumentsWithoutSpaceIds

end

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

local function _fetchChangedDocsAndBringUpToDate( self, tablesNeedingIndexing, 
												changeCounter, needsUpdateForTable,
												tablesThatNeedAggressiveUpdates,
												batchSize, ignoreTablesThatDontExist,
												fastReindexingLimit, toBeReindexedTypeSubtypeMap,
												forcedReindexViaSchemaVersionList )

	-- The goal of this logic is to ensure that we issues as few changedDocs
	-- requests as possible. We only allow one changedDocs request to be
	-- issued at a time, so we examine the state of each table that needs
	-- updating and back up to its starting point for the next request.
	
	batchSize = batchSize or kDefaultBatchSizeForDocMapping
	
	fastReindexingLimit = fastReindexingLimit or 0
	
	local combinedIndexingTable, tableHasNewIndexData, minNeedsUpdateFrom, canSkipRowLookup,
			shouldIncludeDocumentsWithSpaceIds, shouldIncludeDocumentsWithoutSpaceIds 
					= parseTablesNeedingIndexing( self, tablesNeedingIndexing,
													changeCounter,
													tablesThatNeedAggressiveUpdates,
													ignoreTablesThatDontExist,
													needsUpdateForTable, fastReindexingLimit )

	-- Nothing needs updating? Woot!

	if not minNeedsUpdateFrom then

		if not forcedReindexViaSchemaVersionList then
			self._nDocumentsToIndex:onNext( changeCounter, changeCounter )
		end
		
		return Observable.just( needsUpdateForTable, tableHasNewIndexData )

	end

	if not forcedReindexViaSchemaVersionList then
		self._nDocumentsToIndex:onNext( minNeedsUpdateFrom - 1, changeCounter )
			-- We're behind. Signal how far behind and then issue next request.
	end
	
	local startChangeCounter = BehaviorSubject( minNeedsUpdateFrom )

	local cache = {}
		-- used to prefetch the changed docs.
		-- however, between cycles we don't want them cached, otherwise we might miss new changes.
	
	if self._isUnitTest then

		self._forcedReindexingStats.minNeedsUpdateFrom = minNeedsUpdateFrom
		self._forcedReindexingStats.numberOfTablesToReindex = #tablesNeedingIndexing
		
	end
	
	local changedDocs = getChangedDocsThrough( self._docStore, startChangeCounter, changeCounter, 
												batchSize, cache, shouldIncludeDocumentsWithSpaceIds,
												shouldIncludeDocumentsWithoutSpaceIds, toBeReindexedTypeSubtypeMap )

	return changedDocs
		
		:flatMap( function( changedDocs )

			if self._isUnitTest then
			
				local numberOfDocsToReindex = self._forcedReindexingStats.numberOfDocsToReindex or 0
				self._forcedReindexingStats.numberOfDocsToReindex = numberOfDocsToReindex + #changedDocs
				
			end

			if canSkipRowLookup and changedDocs.continueFrom > fastReindexingLimit + 1 then
			
				-- We are still in the (forced) re-indexing phase but batch contains changes
				-- beyond the fastReindexingLimit. End this cycle to restore SQL indices.
				startChangeCounter:onCompleted()
				return Observable.just( needsUpdateForTable, tableHasNewIndexData )
				
			else
			
				return Observable.just( self, changedDocs, combinedIndexingTable,
					minNeedsUpdateFrom, needsUpdateForTable, changeCounter, tableHasNewIndexData, canSkipRowLookup, forcedReindexViaSchemaVersionList )
					:map( bringTableUpToDateWithChangedDocs )
					:map( function( needsUpdateForTable, continueFrom, tableHasNewIndexData )
						-- We need to process documents until the continueFrom (the index store's
						-- indexStatus.needsUpdateFrom column) is > than the changeCounter (the doc store's
						-- revs.sequence column)
						if continueFrom > changeCounter then
							startChangeCounter:onCompleted()
						else
							startChangeCounter:onNext( continueFrom )
						end
						
						if self._isUnitTest then
						
							local reindexedDocCount = self._forcedReindexingStats.reindexedDocCount or 0
							self._forcedReindexingStats.reindexedDocCount = reindexedDocCount + #changedDocs
						
						end
						
						return needsUpdateForTable, tableHasNewIndexData
						
					end )
					
			end

		end )
		
end

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

local function fetchChangedDocsAndBringUpToDate( self, tablesNeedingIndexing, 
												changeCounter, needsUpdateForTable, 
												batchSize, ignoreTablesThatDontExist,
												fastReindexingLimit )

	-- fetchChangedDocsAndBringUpToDate is used for hash-based forced reindexing and for
	-- non-forced reindexing workflows. When forced reindexing, it's not very performant
	-- as it has to fetch and process all documents from the doc store.
	
	self._forcedReindexingStats = { method = 'fetchChangedDocsAndBringUpToDate' }
	
	local tablesThatNeedAggressiveUpdates = self._tablesThatNeedAggressiveUpdates

	local toBeReindexedTables = tablesNeedingIndexing and KSTableUtils.extractKeys( tablesNeedingIndexing )

	return _fetchChangedDocsAndBringUpToDate( self, toBeReindexedTables, 
												changeCounter, 
												needsUpdateForTable,
												tablesThatNeedAggressiveUpdates,
												batchSize, ignoreTablesThatDontExist,
												fastReindexingLimit )
		
end

--------------------------------------------------------------------------------
	
local function fetchChangedDocsByTypeAndBringUpToDate( self, toBeReindexedTables, 
												toBeReindexedTypeSubtypeMap, 
												currentChangeCounter, 
												needsUpdateFromForTables, 
												batchSize, fastReindexingLimit )

	-- fetchChangedDocsByTypeAndBringUpToDate is used for schema-version-list
	-- forced reindexing, and relies on changedDocs to fetch the documents. It 
	-- has to fetch all documents, but it only processes the documents of the 
	-- provided type/subtypes.

	self._forcedReindexingStats = { method = 'fetchChangedDocsByTypeAndBringUpToDate' }

	local forcedReindexViaSchemaVersionList = true
	local ignoreTablesThatDontExist = true
	
	return _fetchChangedDocsAndBringUpToDate( self, toBeReindexedTables, 
												currentChangeCounter, 
												needsUpdateFromForTables, {},
												batchSize, ignoreTablesThatDontExist,
												fastReindexingLimit, 
												toBeReindexedTypeSubtypeMap,
												forcedReindexViaSchemaVersionList )

end

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

local function setIndexStoreSchemaVersionInfo( db, sequence, inProgressReindexSequence )

	if sequence and type( sequence ) == 'number' then

		logger:debugf( "writing new schemaVersionSequence = %s", sequence )

		db:execOnce(
				[[INSERT OR REPLACE INTO localState values( ?, ? )]],
					kLocalState_SchemaVersionSequence, sequence )

	elseif not sequence and WFPlatformInfo.isUnitTest() then
	
		db:execOnce(
				[[DELETE FROM localState WHERE name == ? ]],
					kLocalState_SchemaVersionSequence )
	
	else
	
		logger:warnf( "Invalid schemaVersionSequence value: %s", sequence )
	
	end

	if inProgressReindexSequence and type( inProgressReindexSequence ) == 'number' then

		logger:debugf( "writing new inProgressReindexSequence = %s", inProgressReindexSequence )

		db:execOnce(
				[[INSERT OR REPLACE INTO localState values( ?, ? )]],
					kLocalState_InProgressReindexSchemaVersionSequence, inProgressReindexSequence )
					
	elseif not inProgressReindexSequence and WFPlatformInfo.isUnitTest() then
	
		db:execOnce(
				[[DELETE FROM localState WHERE name == ? ]],
					kLocalState_InProgressReindexSchemaVersionSequence )
	
	else
	
		logger:warnf( "Invalid inProgressReindexSequence value: %s", inProgressReindexSequence )
	
	end

end

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

local function getIndexStoreSchemaVersionInfo( db )

	local existingRow = db:execOnce(
			[[SELECT value FROM localState WHERE name == ?]], 
				kLocalState_SchemaVersionSequence )
		
	local schemaVersionSequence = existingRow and existingRow.value
			
	existingRow = db:execOnce(
			[[SELECT value FROM localState WHERE name == ?]],
				kLocalState_InProgressReindexSchemaVersionSequence )

	local inProgressReindexSchemaVersionSequence = existingRow and existingRow.value

	logger:debugf( "Fetched schemaVersionSequence = %s and inProgressReindexSchemaVersionSequence = %s from the Index Store.", 
			schemaVersionSequence, inProgressReindexSchemaVersionSequence )

	return schemaVersionSequence, inProgressReindexSchemaVersionSequence

end

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

local function addToChunkedArray( chunks, val, chunkSize )

	local currChunk = chunks[ #chunks ]

	local currChunkSize = #currChunk
	
	if currChunkSize > 0 and currChunkSize  % chunkSize == 0 then

		currChunk = {}

		chunks[ #chunks + 1 ] = currChunk

	end
		
	currChunk[ #currChunk + 1 ] = val

	return chunks

end

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

local function chunkedArray( arr, chunkSize )

	return WFFunctionUtils_reduce( { {} }, addToChunkedArray, arr, chunkSize )

end

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

local function prefetchNextBulkDocValue( docStore, cache, chunkedDocIds, currentChunkNumber, offset )
		
	local skipDocIdsValidation = true
			
	local prefetchChunkNumber = currentChunkNumber + offset

	if not cache[ prefetchChunkNumber ] and prefetchChunkNumber <= #chunkedDocIds then

		local docIdsChunk = chunkedDocIds[ prefetchChunkNumber ]

		if docIdsChunk then

			local bulkDocValuePrefetch = docStore:bulkDocValue( docIdsChunk, nil, nil, skipDocIdsValidation )
				
				:replay( nil, 1 )
					-- Cache the result; we'll subscribe right away
					-- (as a consequence of calling :connect()) and
					-- later if this Observable is pulled from the 
					-- cache, this :selectMany() will also subscribe.

			bulkDocValuePrefetch:connect()
				-- Activate it now so it starts the prefetch

			bulkDocValuePrefetch = bulkDocValuePrefetch
				:refCount()
					-- Finish enabling the caching and multiple subscriptions.
					-- If we don't end up using this from the cache then
					-- this was just wasted effort.

			cache[ prefetchChunkNumber ] = bulkDocValuePrefetch
		
		end

	end

end

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

local function getChunkOfDocValues( docStore, readyForDocValuesStream, chunkedDocIds, cache )
	
	local chunkNumber = 0
	
	return readyForDocValuesStream
	
		:flatMap( function()
	
			chunkNumber = chunkNumber + 1
	
			local bulkDocValueObs
			
			if cache[ chunkNumber ] then
				
				bulkDocValueObs = cache[ chunkNumber ]
				
			else
			
				local docIdsChunk = chunkedDocIds[ chunkNumber ]

				if not docIdsChunk then
				
					readyForDocValuesStream:onCompleted()
					return Observable.just( {} )
					
				end

				local skipDocIdsValidation = true
				
				bulkDocValueObs = docStore:bulkDocValue( docIdsChunk, nil, nil, skipDocIdsValidation )
					
			end

			cache[ chunkNumber ] = nil

			return bulkDocValueObs
			
				:tap( function()
				
					-- Prefetch the next two batches of doc values
					
					prefetchNextBulkDocValue( docStore, cache, chunkedDocIds, chunkNumber, 1 )
					prefetchNextBulkDocValue( docStore, cache, chunkedDocIds, chunkNumber, 2 )
					
				end )
				
	end )

end

--------------------------------------------------------------------------------
	
local function fetchDocsByTypeAndBringUpToDate( self, toBeReindexedTables, toBeReindexedTypeSubtypeMap,
												currentChangeCounter, needsUpdateFromForTables, 
												batchSize, fastReindexingLimit )

	-- fetchDocsByTypeAndBringUpToDate is used for schema-version-list forced
	-- reindexing. It uses bulkDocValue to fetch the documents from the doc store.
	-- It limits what it fetches based on type/subtype, and is therefore quite
	-- performant when we're only reindexing a fraction of the total documents.

	self._forcedReindexingStats = { method = 'fetchDocsByTypeAndBringUpToDate' }

	fastReindexingLimit = fastReindexingLimit or 0

	local combinedIndexingTable, tableHasNewIndexData, minNeedsUpdateFrom, canSkipRowLookup
			= parseTablesNeedingIndexing( self, toBeReindexedTables,
											currentChangeCounter, {}, false,
											needsUpdateFromForTables, fastReindexingLimit )

	-- Using the known list of document types that need reindexing, fetch a list of all document
	-- IDs of said types.
	
	local docIdsThatNeedReindexingObs = self._docStore:getDocsOfType( toBeReindexedTypeSubtypeMap, minNeedsUpdateFrom )

	-- Create a stream that fires with batches of doc values for all documents that need to
	-- be updated.
	
	local readyForDocValuesStream = BehaviorSubject( true )

	local cache = {}
		-- used to prefetch the doc values.

	local forcedReindexViaSchemaVersionList = true
	
	return docIdsThatNeedReindexingObs
	
		:flatMap( function( docIdsThatNeedReindexing )

			if not next( docIdsThatNeedReindexing ) or not minNeedsUpdateFrom then
			
				return Observable.just( {} )
					-- Nothing needs to be reindexed. Woot!
					
			end

			logger:debugf( "fetchDocsByTypeAndBringUpToDate: Found %d documents to reindex starting from change counter %d",
					#docIdsThatNeedReindexing, minNeedsUpdateFrom )
		
			if self._isUnitTest then
				self._forcedReindexingStats.numberOfDocsToReindex = #docIdsThatNeedReindexing
				self._forcedReindexingStats.minNeedsUpdateFrom = minNeedsUpdateFrom
				self._forcedReindexingStats.numberOfTablesToReindex = #toBeReindexedTables
			end
			
			local chunkedDocIds = chunkedArray( docIdsThatNeedReindexing, batchSize )
		
			return getChunkOfDocValues( self._docStore, readyForDocValuesStream, chunkedDocIds, cache )

		end )
		
		:flatMap( function( chunkOfDocValues )
		
			local numberOfDocs = #chunkOfDocValues
		
			if numberOfDocs > 0 then
			
				-- Extract the change counter of the last document in the batch,
				-- and store this is a top-level continueFrom field. This is to
				-- match the behavior of changeDocs with is used in standard
				-- reindexing. 
			
				local mostRecentDocInChunk = chunkOfDocValues[ numberOfDocs ]
				chunkOfDocValues.continueFrom = mostRecentDocInChunk._changeCounter + 1
				
			else
			
				readyForDocValuesStream:onCompleted()
				return Observable.just( needsUpdateFromForTables, tableHasNewIndexData )
			
			end
		
			if canSkipRowLookup 
				and chunkOfDocValues.continueFrom 
				and chunkOfDocValues.continueFrom > fastReindexingLimit + 1 then
			
				-- We are still in the (forced) re-indexing phase but batch contains changes
				-- beyond the fastReindexingLimit. End this cycle to restore SQL indices.
				
				readyForDocValuesStream:onCompleted()
				return Observable.just( needsUpdateFromForTables, tableHasNewIndexData )
				
			else
			
				return Observable.just( self, chunkOfDocValues, combinedIndexingTable,
						minNeedsUpdateFrom, needsUpdateFromForTables, currentChangeCounter, 
						tableHasNewIndexData, canSkipRowLookup, forcedReindexViaSchemaVersionList )
					
					:map( bringTableUpToDateWithChangedDocs )
					
					:map( function( needsUpdateForTable, _, tableHasNewIndexData )
					
						if self._isUnitTest then
							local reindexedDocCount = self._forcedReindexingStats.reindexedDocCount or 0
							self._forcedReindexingStats.reindexedDocCount = reindexedDocCount + numberOfDocs
						end
						
						readyForDocValuesStream:onNext( true )

						return needsUpdateForTable, tableHasNewIndexData
						
					end )
					
			end

		end )
		
end

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

local function updateIndexForOneDoc_inTransaction( self, db, db_exec,
												   tableName, docValue, mapFn,
												   existingRowQuery,
												   deleteExistingRowsStmt,
												   insertRowStmt,
												   nColumns, niceToHaveIndices,
												   oneRowPerDocument, useLocalIds,
												   canSkipRowLookup )

	local docId
	if useLocalIds then
		docId = docValue._localId
	else
		docId = docValue._id
	end

	-- Gather existing rows for this document from table.
	-- During a forced re-index of this table we might skip the lookup
	-- because the rows have been deleted.

	local existingRows = {}

	if not ( canSkipRowLookup and self._tableUpdatesAtInitialization[ 'automatic' ][ tableName ] ) then
		existingRows = db:gatherArrayOfArray( existingRowQuery, docId )
	end

	-- See what map functions says is the correct value for this doc now.

	local newRows = {}
	local nNewRows = 0

	local function emit( ... )
		nNewRows = nNewRows + 1
		local newRow = { _n = select( '#', ... ), ... }
		newRows[ nNewRows ] = newRow
	end

	mapFn( docValue, emit )

	if oneRowPerDocument and nNewRows > 1 then
	
		assert( false, string.format( [[map function for table '%s' emitted multiple values 
			despite hasOneRowPerDocument attribute]], tableName ) )
			
	end
		
	local nExistingRows = existingRows and #existingRows or 0

	-- Short-circuit out if old and new rows are identical.

	if nExistingRows == 0 and nNewRows == 0 then
		return false
	end

	local didMatch = true

	if existingRows and #existingRows == nNewRows then

		-- compare the rows emitted even if there are multiple rows...
		-- For simplicity, this comparison is assuming that rows will 
		-- always emit in the same order	

		for i, newRow in ipairs( newRows ) do
		
			local oldRow = existingRows[ i ]
			
			for j = 1, nColumns do
			
				local oldRowIndex = j
				
				if #oldRow > 1 then
				
					-- The first column is the SQLite rowid, so we ignore it.
					
					oldRowIndex = oldRowIndex + 1
				
				end
				
				local columnHasChanged = false
	
				if type( newRow[ j ] ) == 'boolean' then
					-- we might emit booleans, but SQLite stores them as 1/0 so convert to 1/0 to compare
					columnHasChanged = ( newRow[ j ] and 1 or 0 ) ~= oldRow[ oldRowIndex ]
				else 
					columnHasChanged = oldRow[ oldRowIndex ] ~= newRow[ j ]
				end
				
				if columnHasChanged then
					if niceToHaveIndices[ j ] then
						didMatch = 'niceToHave'
					else
						didMatch = false
						break
					end
				end
			end

			if didMatch == false then
				break
			end
			
		end
		
		if didMatch == true then
			return false 
		end
			
		-- Don't return early if 'niceToHave'; we must update the table			

	end

	-- No match: Replace all old rows with new rows.

	if ( nExistingRows > 0 ) and not ( oneRowPerDocument and nNewRows == 1 ) then
		db_exec( db, deleteExistingRowsStmt, docId )
	end

	if nNewRows == 0 then return true end

	for i = 1, nNewRows do

		local newRow = newRows[ i ]
		db_exec( db, insertRowStmt, docId,
				 unpack( newRow, 1, nColumns ) )

	end

	if didMatch == 'niceToHave' then
		return didMatch
	else
		return true
	end

end

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

local function updateIndexTable_inTransaction( self, db, tableName, changedDocs, 
												canSkipRowLookup )

	local hasNewData = false
	local hasNiceToHaveNewData = false
	local db_exec = self._db_exec

	-- Precompute some queries so we can assume they're present in the tight loop.
	local isFts = self._fullTextSearchTables[ tableName ]
	local docIdName

	if isFts then
		docIdName = 'docid'
			-- how unfortunate that SQLite's built-in docId differs from ours
			-- only in case. :-(
	else
		docIdName = 'docId'
	end

	local existingRowQuery = self._existingRowQueryForTable[ tableName ]
	if not existingRowQuery then
		existingRowQuery = string.format( [[SELECT * FROM %s WHERE %s == ?]], tableName, docIdName )
		self._existingRowQueryForTable[ tableName ] = existingRowQuery
	end

	local deleteExistingRowsStmt = self._deleteExistingRowsStmtForTable[ tableName ]
	if not deleteExistingRowsStmt then
		deleteExistingRowsStmt = string.format( [[DELETE FROM %s WHERE %s == ?]],
												tableName, docIdName )
		self._deleteExistingRowsStmtForTable[ tableName ] = deleteExistingRowsStmt
	end

	local nColumns = self._nColumnsPerTable[ tableName ]

	local oneRowPerDocument = self._tablesThatHaveOneRowPerDocument [ tableName ]
	local useLocalIds = self._tablesThatUseLocalIds [ tableName ]
	local insertRowStmt = self._insertRowStmtForTable[ tableName ]
	if not insertRowStmt then
		local maybeOrReplace = ""
		if oneRowPerDocument then
			maybeOrReplace = " OR REPLACE"
		end
		
		-- SPAM [jandrew 2019-12-19]: We build the following SQL statements using `nColumns + 1`.
		-- `nColumns` represents the number of USER DEFINED columns within the table. The docId column
		-- is *not* user defined, and so we must increment the count by 1 since our insertions 
		-- include the docId field.
		
		if isFts then
			insertRowStmt = string.format( [[INSERT OR REPLACE INTO %s( docid, search_text ) VALUES( %s )]],
									   tableName, 
									   WFSQLiteUtils.createVariableArgList( nColumns + 1 ) )
		else
			insertRowStmt = string.format( [[INSERT%s INTO %s VALUES( %s )]],
					maybeOrReplace, 
					tableName,
					WFSQLiteUtils.createVariableArgList( nColumns + 1 ) )
		end
		
		self._insertRowStmtForTable[ tableName ] = insertRowStmt
	end

	local filterFn = self._docValueFilters[ tableName ]
	local relevantTypeAndSubtypeMap = self._relevantTypeAndSubtypesByTable[ tableName ] or {}
	local mapFn = assert( self._mapFunctions[ tableName ] )
	local niceToHaveIndices = assert( self._niceToHaveIndices [ tableName ] )

	local shouldIncludeDocumentsWithSpaceIds = self._tablesThatShouldIncludeDocumentsWithSpaceIds[ tableName ]
						
	local shouldIncludeDocumentsWithoutSpaceIds = self._tablesThatShouldIncludeDocumentsWithoutSpaceIds[ tableName ]

	local shouldIncludeAllTypesAndSubtypes = not next( relevantTypeAndSubtypeMap ) 
			or KSTableUtils_derefOpt( relevantTypeAndSubtypeMap, '*', '*' )

	for i = 1, #changedDocs do

		local docValue = changedDocs[ i ]
		
		local spaceId = docValue._spaceId
		local docType = docValue.type
		local subType = docValue.subtype or 'NULL'
		
		if ( spaceId and shouldIncludeDocumentsWithSpaceIds )
			or ( not spaceId and shouldIncludeDocumentsWithoutSpaceIds ) then
			
			local isRelevantDocument = shouldIncludeAllTypesAndSubtypes
					or KSTableUtils_derefOpt( relevantTypeAndSubtypeMap, docType, subType )
					or KSTableUtils_derefOpt( relevantTypeAndSubtypeMap, docType, '*' )
			
			-- No matter the filter criteria, deletes should always be processed.
			-- This allows the table to clean up any entries that existed from before
			-- the deletion when the document may have matched the filter
			
			if docValue._deleted or ( isRelevantDocument and filterFn( docValue ) ) then

				local newData =
					updateIndexForOneDoc_inTransaction( self, db, db_exec,
													   tableName, docValue, mapFn,
													   existingRowQuery,
													   deleteExistingRowsStmt,
													   insertRowStmt,
													   nColumns, niceToHaveIndices,
													   oneRowPerDocument, useLocalIds,
													   canSkipRowLookup )
				if newData == 'niceToHave' then
					hasNiceToHaveNewData = true
				elseif newData then
					hasNewData = true
				end
				
			end
			
		end
		
	end

	db_exec( db, [[UPDATE indexStatus SET needsUpdateFrom = ? WHERE indexTableName = ?]],
			 changedDocs.continueFrom, tableName )

	return hasNewData, hasNiceToHaveNewData

end


--------------------------------------------------------------------------------
function transactInDB( self, changedDocs, tablesNeedingIndexing, 
					minNeedsUpdateFrom, needsUpdateForTable, changeCounter,  continueFrom, didReviewIndexForTable, tableHasNewIndexData,
					canSkipRowLookup )
					
	self._db:withTransactionDo( function()
			
			local indexingPerformanceByTableName = {}
			
			-- Apply updates to all potentially relevant tables.
			for tableName in pairs( tablesNeedingIndexing ) do

				-- Does table need updating? Skip if not.

				local needsUpdateFrom = needsUpdateForTable[ tableName ]
				local needsUpdate = type( needsUpdateFrom ) == 'number' and needsUpdateFrom < continueFrom
				
				tableHasNewIndexData[ tableName ] = needsUpdate
				
				if needsUpdate then

					-- Make sure the range of documents provided by this update is
					-- valid for this table's state.
				
					if minNeedsUpdateFrom <= needsUpdateFrom	-- changedDocs includes at least first doc of interest
						and continueFrom > needsUpdateFrom		-- changedDocs includes *any* docs of interest
					then
						-- There are relevant documents. Go index them.

						local startTime = KSDate.timeNowForAnimation()

						local success, hasNewDataOrMessage, hasNiceToHaveNewData = pcall(
							updateIndexTable_inTransaction, self, self._db, tableName, changedDocs, canSkipRowLookup )

						-- Update profiling stats for this table.

						local timeConsumed = KSDate.timeNowForAnimation() - startTime

						local stat = self._updateTimePerTable[ tableName ]

						if stat then

							stat.count = stat.count + 1
							stat.min = math.min( stat.min, timeConsumed )
							stat.max = math.max( stat.max, timeConsumed )
							stat.niceToHaveUpdatesIgnored = ( stat.niceToHaveUpdatesIgnored or 0 )
									+ ( ( hasNiceToHaveNewData and not hasNewDataOrMessage ) and 1 or 0 )
							stat.total = stat.total + timeConsumed

						else

							stat = {
								tableName = tableName,
								count = 1,
								min = timeConsumed,
								max = timeConsumed,
								total = timeConsumed,
							}

							self._updateTimePerTable[ tableName ] = stat

						end
						
						-- for stamina
						indexingPerformanceByTableName[ tableName ] = timeConsumed

						if success then
						
							didReviewIndexForTable[ tableName ] = true
							
							tableHasNewIndexData[ tableName ] = hasNewDataOrMessage
							
						else
							-- If an error occurs in this process, set index status to error
							-- and walk away. This table will not update again during this
							-- app session.

							tableHasNewIndexData[ tableName ] = true
							
							needsUpdateForTable[ tableName ] = 'error: ' .. tostring( hasNewDataOrMessage )
						end

					end

				end

			end
			
			if next( indexingPerformanceByTableName ) then
				indexingPerformanceByTableName.docsIndexed = #changedDocs
				indexingPerformanceByTableName.minNeedsUpdateFrom = minNeedsUpdateFrom
				indexingPerformanceByTableName.continueFrom = continueFrom
				WFPerformanceReport.recordSingleValue( 'WFRxIndexStoreMapServer:Indexing', indexingPerformanceByTableName )
			end		
		end )

end

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


function bringTableUpToDateWithChangedDocs( self, changedDocs, tablesNeedingIndexing, 
							minNeedsUpdateFrom, needsUpdateForTable, changeCounter, tableHasNewIndexData,
							canSkipRowLookup, forcedReindexViaSchemaVersionList )

	if WFCloseSupport.hasBeenOrWillBeClosed( self ) then return {} end

	logger:debugf( "received a bunch of changed docs: now what? (n = %d)",
				   #changedDocs )

	-- Apply all updates within one transaction.

	--local db = self._db
	local didReviewIndexForTable = {}

	local continueFrom = changedDocs.continueFrom or 0

	KSLuaUtils.dpcall( 'processChangedDocs', function()

		transactInDB( self, changedDocs, tablesNeedingIndexing, 
			minNeedsUpdateFrom, needsUpdateForTable, changeCounter, continueFrom, didReviewIndexForTable, tableHasNewIndexData,
			canSkipRowLookup )
		
	end )

	-- Update state cache.

	for tableName in pairs( didReviewIndexForTable ) do
		needsUpdateForTable[ tableName ] = continueFrom
	end

	if not forcedReindexViaSchemaVersionList then
		self._nDocumentsToIndex:onNext( continueFrom - 1, changeCounter )
	end
	
	return needsUpdateForTable, continueFrom, tableHasNewIndexData

end

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

-- Create the replay observable that tells how up to date all the index tables
-- are.
local function makeNeedsUpdateFromForTable( self )

	local tablesNeedingIndexing = Observable.create( function( observer )
		assert( self._changeTableIndexingStatus == false, '_changeTableIndexingStatus must have only one subscription' )

		self._changeTableIndexingStatus = function( addTable, removeTable )
			observer:onNext( addTable, removeTable )
		end
	end )
		:map( accumulateAndRemoveTables() )

	local changeCounter = self._docStore:changeCounter()
		:debounce( 0.125, WFRxScheduler.sharedScheduler() )

	local processing = tablesNeedingIndexing
		:combineLatest( changeCounter, self._isReady )
		:pausableBuffered( )

	self._processing = processing
	
	local needsUpdateFromForTable = processing
		:map( function( tablesNeedingIndexing, changeCounter, initialNeedsUpdateForTable )

			processing:pause( )

			return fetchChangedDocsAndBringUpToDate(
					self,
					tablesNeedingIndexing,
					changeCounter,
					initialNeedsUpdateForTable )
				:tapOnCompleted( function()
					processing:resume()
				end )
		end )
		:switch()
		:replay( nil, 1 )
		:refCount()

	return needsUpdateFromForTable, processing

end

--============================================================================--
--[[!
	\brief A \c WFRxIndexStoreMapServer is used by the index store implementation
		to keep the index store tables up to date.

	It should only be accessed by \c WFRxIndexStoreImp.
]]--

local WFRxIndexStoreMapServer = {

	_docStore = false,
		-- (WFRxDocStore or WFDocStore or OzDocStore) doc store we're getting our data from

	_db = false,
		-- (WFSQLiteDatabase)

	_db_exec = false,
		-- (function) cached reference to self._database:exec

	_tableNames = false,
		-- (table) names of the index tables declared in this schema

	_changeTableIndexingStatus = false,
		-- function used to start and stop indexing
		-- particular tables.

	_processing = false,
		-- Rx.BehaviorSubject oscilates between true / false when
		-- doing updating.  Used in conjunction with pausableBuffered to
		-- avoid starting and then stopping and ignoring work.

	_mapFunctions = false,
		-- (table) map function for each table in this schema
			-- key: (string) name of index table
			-- value: (function) map function declared for this table
			
	_sqlIndexInfos = false,
		-- (table) SQL indices information for each table in this schema
			-- key: (string) name of table
			-- value: (table) SQL indices
				-- key: SQL index name
				-- vale: creation SQL string
	
	_docValueFilters = false,
		-- (table) filter function for each table in this schema
			-- key: (string) name of index table
			-- value: (function) filter function that restricts the doc values 
			-- sent to the map function for this table
		
	_relevantTypeAndSubtypesByTable = false,
		-- (table) Nested tables containing the relevant type/subtypes for each table in this schema
		-- key: (string) name of index table
		-- value: (table)
			-- key: (string) relevant type
			-- value: (table)
				-- key (string) relevant subtype
				-- value: (boolean) true
		
	_nColumnsPerTable = false,
		-- (table) how many user-defined columns are in each index table?
			-- key: (string) name of index table
			-- value: (number) how many user-defined columns?

	_updateTimePerTable = false,
		-- (table)
			-- key: (string) name of index table
			-- value: (table) info on statement (count, min, max, total)

	_isReady = false,
		-- (Replay Observable< true | "out of sync" >) becomes true when we
			-- have verified that index store and doc store are properly
			-- related; resolves to true in most normal circumstances or
			-- "out of sync" if index and doc store were unrelated to each other

	_needsUpdateFromForTable = false,
		-- (Replay Observable< table >) how up to date is each index table?
			-- key: (string) name of index table
			-- value: (number) change counter for first document
				-- *not* yet indexed

	_existingRowQueryForTable = false,
		-- (table) query to gather existing rows
			-- key: (string) name of index table
			-- value: (string) SELECT * FROM (table name) WHERE docId == ?

	_deleteExistingRowsStmtForTable = false,
		-- (table) query to delete existing rows
			-- key: (string) name of index table
			-- value: (string) DELETE FROM (table name) WHERE docId == ?

	_insertRowStmtForTable = false,
		-- (table) sql statment to insert new rows
			-- key: (string) name of index table
			-- value: (string) INSERT INTO (table name) VALUES( ... )

	_nDocumentsToIndex = false,
		-- (Observable< number >) how many documents are remaining to index?

	_lastTempIds = false,

	_niceToHaveIndices = false,
		-- (table)
			-- key (string) : table name as declared in SQL or fields
			-- value (table): a set of columns that are nice to have (often the
			-- empty set but never nil)

	_tablesThatHaveOneRowPerDocument = false,
		-- (table)
			-- key (string) : table name as declared in SQL or fields
			-- value (boolean): true if this table only emits at most one row
			-- per document

	_tablesThatUseLocalIds = false,
		-- (table)
			-- key (string) : table name as declared in SQL or fields
			-- value (boolean): true if this table uses local ids

	_fullTextSearchTables = false,
		-- (table)
			-- key (string) : table name as declared in SQL or fields
			-- value (boolean): true if this table is a fts4 virtual table

	_keepTableUpToDate = false,
		-- (fully weak table)
			-- key (string) : table name as declared in SQL or fields
			-- value (Observable): refcount replay observable for the
			-- keepFullyUpToDate result
			
	_tablesThatNeedAggressiveUpdates = false,
		-- (table)
			-- key (string) : table name as declared in SQL or fields
			-- value (boolean): true if this table is requested to be kept updated by the client 
			
	_reindexingProgress = false,
		-- (WFProgressReporter)
	
	_tableUpdatesAtInitialization = false,
		-- (table) data about the index store tables that were reset/reindexed at initialization
		-- for the diagnostic log
		
	_tablesThatShouldIncludeDocumentsWithoutSpaceIds = false,
		-- (table)
			-- key (string): A table name as declared in SQL or via tableName
			-- value (boolean): true if this table should process catalog documents.
			
	_tablesThatShouldIncludeDocumentsWithSpaceIds = false,
		-- (table)
			-- key (string): A table name as declared in SQL or via tableName
			-- value (boolean): true if this table should process space-feed documents.
			
	_forcedReindexingStats = false,
		-- (table) Of table of forced reindexing metrics, primarily used by unit tests.
			
	_isUnitTest = false,
		-- (boolean) True if the code is running as part of a unit test.
			
	WFCloseSupport.addCloseSupportMethods,

}

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

local function populateNeedsUpdateFromForTables( self )

	-- Called once by docStoreCrossCheckB. Preload change counters for all
	-- tables from database.

	local db = self._db

	local indexStatusRows = db:gatherArrayOfArray
		[[SELECT indexTableName, needsUpdateFrom FROM indexStatus]]

	local needsUpdateFromForTable = {}

	for i = 1, #indexStatusRows do

		local row = indexStatusRows[ i ]
		local tableName = row[ 1 ]
		local needsUpdateFrom = row[ 2 ]

		needsUpdateFromForTable[ tableName ] = needsUpdateFrom

	end

	return needsUpdateFromForTable

end

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

local function logReindexingEvent( eventType, maybeTablesToReindex, maybeDuration )

	local durationEventThreshold = WFConfiguration.Reindex.durationEventThreshold or math.huge

	if ( eventType ~= kStandardReindexingEvent and
			eventType ~= kCompletedReindexingEvent ) or
			( maybeDuration and maybeDuration > durationEventThreshold ) or
			WFConfiguration.Reindex.verboseAnalytics then
	
		-- Log Analytics events for the non-standard reindexing workflows. This 
		-- includes resuming reindexing (indicating the user quit the app while
		-- reindexing), and reindexing due to a corrupt/missing index store.
		
		-- When WFConfiguration.Reindex.verboseAnalytics is enabled, we log
		-- analytics events for standard reindexing events as well.
		
		-- For reindexing completion events, if reindexing took longer than the
		-- specified threshold (configured via WFConfiguration.Reindex.durationEventThreshold)
		-- then the event will be logged.
	
		local event = {}

		event['event.type'] = ( eventType == kInvalidIndexStoreReindexingEvent ) and 'warning' or 'info'
		event['event.subtype'] = eventType
		event['event.appWorkflow'] = 'LAUNCH'
		event['event.subcategory'] = 'Reindex'
		
		if maybeTablesToReindex then
			event['event.tablesToReindex'] = WFTableArchiving.tableToString( maybeTablesToReindex )
		end
		
		if eventType == kCompletedReindexingEvent and maybeDuration then
			event['event.reindexDuration'] = maybeDuration
		end
		
		WFAnalytics.trackEvent( event )
	
	end

end

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

local function logPercentageOfDocsToReindex( percentage, strategy )

	
	local event = {}
	event['event.type'] = 'info'
	event['event.subtype'] = kPercentageToReindexingEvent
	event['event.appWorkflow'] = 'LAUNCH'
	event['event.subcategory'] = 'Reindex'
	event['event.percentageOfDocsToReindex'] = percentage
	event['event.reindexingStrategy'] = strategy
	
	WFAnalytics.trackEvent( event )
	
end


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

local function reportReindexingProgressEstimate( self, changeCounter, tablesRequiringForceUpdate, 
		needsUpdateFromForTables, fractionSqlIndicesRestored, initialEstimatedWork, scaleForSqlIndices )
			
	local remainingWork = 0
			
	local calculateWork = function()
	
		local work = 0
	
		for tableName, _ in pairs( tablesRequiringForceUpdate ) do
		
			local needsUpdateFrom = needsUpdateFromForTables[ tableName ] or changeCounter
		
			work = work + ( math.max( 0, changeCounter - needsUpdateFrom ) )
		
		end

		-- first 80% of the progressbar for the mapping, last 20% for recreating SQL indices
		local scale = scaleForSqlIndices or ( 0.2/(1.0-0.2) * work )
		work = work + scale * (1 - fractionSqlIndicesRestored )

		if fractionSqlIndicesRestored >= 1 then
			work = 0	-- doing this as we need to set the correct progress, fractionSqlIndecesRestored = 1 is the last call made to this method and somehow progress is incorrect.
		end

		return work, scale
	
	end
			
	if not initialEstimatedWork then
		initialEstimatedWork, scaleForSqlIndices = calculateWork()
		remainingWork = initialEstimatedWork
		self._reindexingProgress:setTotal( initialEstimatedWork, true )
	else
		remainingWork = calculateWork()	
	end

	local completedWork = initialEstimatedWork - remainingWork 
	
	logger:debugf( "reportReindexingProgressEstimate: Reindexing progress %d of %d.", completedWork, initialEstimatedWork )

	local mustDeliver = WFPlatformInfo.isUnitTest()
		-- Tests against the reindexing logic verify progress via _reindexingProgress.
		-- As designed, some progress events may be dropped from the stream. Dropping
		-- these events can cause test instability, so we force all progress events
		-- to be sent when running within unit tests.
	
	self._reindexingProgress:setProcessedCount( completedWork, mustDeliver )

	return initialEstimatedWork, scaleForSqlIndices

end

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

-- Recreate SQL indices which were dropped during a forced re-index.
-- Returns an observable wich fires if a indices has been recreated 
-- and carries the fraction recreated/totalDroppedIndices for progressbar updates.
local function recreateSqlIndices( self )

	local indices = {}

	for tableName, _ in pairs( self._tableUpdatesAtInitialization[ 'automatic' ] ) do
		for indexName, sql in pairs( self._sqlIndexInfos[ tableName ] ) do
			indices[ #indices + 1 ] = sql
		end
	end

	local nIndicesRecreated = 0

	if next( indices ) then
		return Observable.from( indices )
			:map( function ( sql )
				self._db_exec( self._db, sql )
				nIndicesRecreated = nIndicesRecreated + 1
				return nIndicesRecreated / #indices

			end )
	else
		return Observable.just( 1 )
	end

end

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

-- Drop SQL indices for tables which have been reset at initialization
local function dropSqlIndices( self )

	local nIndicesDropped = 0

	for tableName, _ in pairs( self._tableUpdatesAtInitialization[ 'automatic' ] ) do
		for indexName, _ in pairs( self._sqlIndexInfos[ tableName ] ) do
			self._db_exec( self._db, string.format([[DROP INDEX %s]], indexName ) )
			nIndicesDropped = nIndicesDropped + 1
		end
	end

	return nIndicesDropped

end

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

local function setDocStoreCrossCheck( self, crossCheck )

	return self._docStore:setLocalDocValue
		{
			_id = 'indexStoreCrossCheck',
			crossCheck = crossCheck,
		}

end

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

local function forceReindexUsingVersionList( self, reindexInfo, needsUpdateFromForTables )

	local currentVersionSequenceNumber = reindexInfo.currentSequence
	local newVersionSequenceNumber = reindexInfo.newSequence
	local toBeReindexedTables = reindexInfo.toBeReindexedTables
	local toBeReindexedTypeSubtypeMap = reindexInfo.toBeReindexedTypeSubtypeMap
	local maybeForceReindex = reindexInfo.maybeForceReindex
	
	local tablesRequiringForceUpdate 
			= maybeForceReindex and toBeReindexedTables or {}

	local tablesRequiringForceUpdateSet = KSTableUtils.invert( tablesRequiringForceUpdate )

	local db = self._db

	if next( tablesRequiringForceUpdate ) then
	
		logger:debugf( "forceReindexUsingVersionList: Reindexing is necessary for %d tables.", #tablesRequiringForceUpdate )
	
		local reindexingStartTime = KSDate.timeNow()
	
		setIndexStoreSchemaVersionInfo( db, currentVersionSequenceNumber, newVersionSequenceNumber )
		
		return Observable.combineLatest(
				self._docStore:changeCounter(),
				self._docStore:countAllDocs(),
				self._docStore:countDocsOfType( toBeReindexedTypeSubtypeMap )
			)
			
			:take( 1 )

			:flatMap( function( currentChangeCounter, totalDocCount, numberOfDocsNeedingReindexing )
			
				-- Make note of the current changeCounter. This value represents the highest changeCounter
				-- upto which we safely can map docs without looking up possible existing rows
				-- for the docId (and thus can drop the SQL indices).

				local fastReindexingLimit = currentChangeCounter

				local initialEstimatedWork, scaleForSqlIndices = reportReindexingProgressEstimate( self, currentChangeCounter, 
																	tablesRequiringForceUpdateSet,	needsUpdateFromForTables, 0 )

				-- drop the indices
				dropSqlIndices( self )

				local fetchDocsAndBringUpToDateFn
				
				local isUnitTest = WFPlatformInfo.isUnitTest()
				
				if isUnitTest and WFCapabilities.getCapability( 'WFRxIndexStoreTest.testReindexViaChangedDocs' ) == 'true' then
				
					fetchDocsAndBringUpToDateFn = fetchChangedDocsByTypeAndBringUpToDate
				
				elseif isUnitTest and WFCapabilities.getCapability( 'WFRxIndexStoreTest.testReindexViaBulkDocValue' ) == 'true' then
				
					fetchDocsAndBringUpToDateFn = fetchDocsByTypeAndBringUpToDate
			
				else
					
					local percentOfDocsNeedingReindexing = numberOfDocsNeedingReindexing / totalDocCount
					
					local thresholdPercentage = 0.80 
						-- https://wiki.corp.adobe.com/display/lightroom/WF+Schema-Version-List+Reindexing+Performance+Exploration
				
					local strategyString 
				
					if percentOfDocsNeedingReindexing < thresholdPercentage then
						strategyString = 'fetchDocsByTypeAndBringUpToDate'
						fetchDocsAndBringUpToDateFn = fetchDocsByTypeAndBringUpToDate
					else 
						strategyString = 'fetchChangedDocsByTypeAndBringUpToDate'
						fetchDocsAndBringUpToDateFn = fetchChangedDocsByTypeAndBringUpToDate
					end
					
					logPercentageOfDocsToReindex( percentOfDocsNeedingReindexing, strategyString )		
					
				end
			
				return fetchDocsAndBringUpToDateFn(
						self,
						tablesRequiringForceUpdate,
						toBeReindexedTypeSubtypeMap,
						currentChangeCounter,
						needsUpdateFromForTables,
						kMaxBatchSizeForDocMapping, 
						fastReindexingLimit )
			
					:map( function()
						return currentChangeCounter, initialEstimatedWork, scaleForSqlIndices
					end )
			
			end )
			
			:map( function( changeCounter, initialEstimatedWork, scaleForSqlIndices )
				
				-- fetchChangedDocsAndBringUpToDate fires with periodic updates as the
				-- tables are brought up to date. We can use this to measure reinde xing
				-- progress.
				
				local needsUpdateFromForTables = populateNeedsUpdateFromForTables( self )
				
				reportReindexingProgressEstimate( self, changeCounter, tablesRequiringForceUpdateSet,
					needsUpdateFromForTables, 0, initialEstimatedWork, scaleForSqlIndices )
				
				return needsUpdateFromForTables, changeCounter, initialEstimatedWork, scaleForSqlIndices
				
			end )

			:takeLast( 1 )

			:flatMap( function( needsUpdateFromForTables, changeCounter, initialEstimatedWork, scaleForSqlIndices )

				return recreateSqlIndices( self )
					:map( function( value )
						reportReindexingProgressEstimate( self, changeCounter, tablesRequiringForceUpdateSet,
							needsUpdateFromForTables, value, initialEstimatedWork, scaleForSqlIndices )
						return value
					end )
					:takeLast( 1 )
					:map( function () 
						return needsUpdateFromForTables
					end )
				
			end )

			:map( function( needsUpdateFromForTables )
			
				setIndexStoreSchemaVersionInfo( db, newVersionSequenceNumber, 0 )
			
				local duration = KSDate.timeNow() - reindexingStartTime
								
				logReindexingEvent( kCompletedReindexingEvent, nil, duration )
			
				self._reindexingProgress:done()
				return needsUpdateFromForTables
			
			end )
		                                   
	else
	
		logger:debugf( "forceReindexUsingVersionList: Reindexing is not necessary." )
	
		setIndexStoreSchemaVersionInfo( db, newVersionSequenceNumber, 0 )
	
		self._reindexingProgress:done()
			-- no forced reindexing required, close progress stream anyway
			
		return Observable.just( needsUpdateFromForTables )
		
	end

end

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

local docStoreCrossCheck = function( self, indexStoreSchema )

	if WFCloseSupport.hasBeenOrWillBeClosed( self ) then
		return Observable.throw( "docStoreCrossCheck() : Map server closed" )
	end

	local db = self._db

 	-- Make sure doc store is the same one we thought it was before.

	local existingRow = db:execOnce
		[[SELECT value FROM localState WHERE name == 'docStoreCrossCheck']]

	-- If we have no check code in the index store, then we create one.

	local expectedCheckCode = existingRow and existingRow.value

	local expectedCheckCodeIsNew = not expectedCheckCode

	if expectedCheckCodeIsNew then

		expectedCheckCode = KSIdGenerator.uuidInLowercaseHex()

		logger:debugf( "writing new docStoreCrossCheck = %s",
					   expectedCheckCode )

		db:execOnce(
			[[INSERT OR REPLACE INTO localState values( 'docStoreCrossCheck', ? )]],
			expectedCheckCode )

	end

	logger:debugf( "expectedCheckCode is %s is new = %s",
				   expectedCheckCode,
				   expectedCheckCodeIsNew )

	-- Does the index store cross-check value match the doc store's cross-check?

	local crossCheckStream = self._docStore:localDocValue( 'indexStoreCrossCheck' )
			
		:map( function( docValue )

			local maybeForceReindex = false

			if WFCloseSupport.hasBeenOrWillBeClosed( self ) then
				return "doc store was closed"
			end

			local docStoreCheckCode = docValue.crossCheck

			logger:debugf( "docStoreCheckCode resolved with %s, expected %s",
						   docStoreCheckCode, expectedCheckCode )

			local indexStoreSupportsSchemaVersionList = indexStoreSchema:useSchemaVersionList()

			local useSchemaVersionList = indexStoreSupportsSchemaVersionList

			local newSequence, toBeReindexedTables, resumeSchemaVersionReindex, currentSequence,
					toBeReindexedTypeSubtypeMap

			if useSchemaVersionList then
				
				local schemaVersionSequence, inProgressReindexSchemaVersionSequence = getIndexStoreSchemaVersionInfo( db )
				
				currentSequence = schemaVersionSequence
				local inProgressReindexSequence = inProgressReindexSchemaVersionSequence or 0
				
				local validTablesSet = indexStoreSchema:tableNames()
				local validTables = KSTableUtils.extractKeys( validTablesSet )
				
				newSequence, toBeReindexedTables 
						= WFRxIndexStoreSchemaVersionList.getTablesToReindex( currentSequence, validTables )

				if toBeReindexedTables == "hashBasedIndexing" then
				
					logger:debugf( "crossCheckStream: Falling back to hash-based indexing." )
				
					useSchemaVersionList = false
					
				else
				
					toBeReindexedTypeSubtypeMap = indexStoreSchema:getTypeSubtypeMapForTables( toBeReindexedTables )

					if newSequence == inProgressReindexSequence and inProgressReindexSequence > 0 then
					
						resumeSchemaVersionReindex = true
					
					end
				
				end
			
			end
		
			-- SPAM [jandrew 2021-06-15]: The following code may drop existing SQLite
			-- tables if the tables need to be reindexed. Note that this happens before
			-- we drop all SQLite indices. There's no performance reason to drop all indices
			-- before dropping the tables because, according to SQLite documention, associated
			-- indices and triggers are dropped when a table is dropped:
			--
			-- "All indices and triggers associated with the table are also deleted."
			-- From https://sqlite.org/lang_droptable.html
			
			-- SPAM [jandrew 2021-09-14]: WF-1741: If we fall back to hash-based reindexing,
			-- we force reindex *all* index store tables regardless of if there was a 
			-- sql schema or map function hash change. We do this to guarantee proper
			-- reindexing when index store tables contain changes which are invisible
			-- to the older hash-based reindexing logic (such as changing the docValueFilter
			-- function, the relevantTypeSubtypeTuples array, or any function called
			-- by an existing map function).
			-- The only downside to this is that hash-based reindexing always has to
			-- reindex every table and not just the tables that have changed. This should
			-- actually be okay considering our performance metrics have shown that
			-- the cost of reindexing one table via hash-based reindexing is only 
			-- slightly cheaper than reindexing all tables. The real cost comes from the 
			-- processing of the documents, which is the same regardless of table count.
			
			if expectedCheckCode == docStoreCheckCode then

				-- Nothing to do: Doc store and index store are already related.

				-- Depending on useSchemaVersionList, we'll either reset the index status
				-- of tables via the Schema Version list, or reset the index status and
				-- wipe out the existing tables via the table hash approach. 
				-- Please note that we should never remove support for the table hash
				-- approach because hash based indexing is necessary in the event of an 
				-- unknown current sequence number.

				if not resumeSchemaVersionReindex then
					-- If we're resuming reindexing, no tables need to be reset.

					if useSchemaVersionList then
						indexStoreSchema:resetTablesBySchemaVersionList( db, toBeReindexedTables )
					else
						indexStoreSchema:resetTablesIfNecessary( db, indexStoreSupportsSchemaVersionList )
					end

				end
				
			elseif expectedCheckCodeIsNew and docStoreCheckCode == nil then

				-- Doc store and index store are not related, but it's OK because
				-- index store is new. Make them related.

				logger:debug( "write check code to doc store" )

				useSchemaVersionList = false -- Fall back to hash-based reindexing
				
				indexStoreSchema:resetTablesIfNecessary( db, true )
				
				setDocStoreCrossCheck( self, expectedCheckCode )

			else
	
				-- Index store appears to describe some other doc store.
				-- Blow it away and start over!

				useSchemaVersionList = false -- Fall back to hash-based reindexing.
				
				maybeForceReindex = true
				
				indexStoreSchema:resetTablesIfNecessary( db, true )

				logReindexingEvent( kInvalidIndexStoreReindexingEvent )

				-- Generate new check code and apply it to both doc store
				-- and index store.

				expectedCheckCode = KSIdGenerator.uuidInLowercaseHex()

				logger:warn(
					"index store will blow away everything :: out of sync!!!" )

				logger:debugf( "docStoreCheckCode generated new check code %s",
							   expectedCheckCode )

				db:execOnce(
					[[INSERT OR REPLACE INTO localState values( 'docStoreCrossCheck', ? )]],
					expectedCheckCode )

				setDocStoreCrossCheck( self, expectedCheckCode )

			end
			
			local tablesResetAtInitialization = indexStoreSchema:tablesResetAtInitialization()
			self._tableUpdatesAtInitialization = {
				automatic = tablesResetAtInitialization,
			}
			
			if next( tablesResetAtInitialization ) then
				
				maybeForceReindex = true
				
				logReindexingEvent( kStandardReindexingEvent, tablesResetAtInitialization )
				
			elseif resumeSchemaVersionReindex then
					
				-- If no tables were reset, it could mean that forced reindexing was triggered
				-- on a past launch, but it never completed. This code detects that, and makes
				-- it so forced reindexing resumes if needed.
				
				logger:debugf( "Incomplete Reindex detected. Will resume reindexing." )
				maybeForceReindex = true
	
				logReindexingEvent( kResumeReindexingEvent, tablesResetAtInitialization )
	
			end

			-- Signal that DB has desired schema and we can start updating
			-- the index tables.

			local reindexInfo = {
				maybeForceReindex = maybeForceReindex,
				useSchemaVersionList = useSchemaVersionList,
				currentSequence = currentSequence,
				newSequence = newSequence,
				toBeReindexedTables = toBeReindexedTables,
				toBeReindexedTypeSubtypeMap = toBeReindexedTypeSubtypeMap
			}

			return reindexInfo, populateNeedsUpdateFromForTables( self )

		end )

		return crossCheckStream
		
			:flatMap( function( reindexInfo, needsUpdateFromForTables )

				if reindexInfo.useSchemaVersionList then

					return forceReindexUsingVersionList( self, reindexInfo, needsUpdateFromForTables )

				else

					local newVersionSequenceNumber = reindexInfo.newSequence

					local tablesRequiringForceUpdate 
							= reindexInfo.maybeForceReindex and needsUpdateFromForTables or {}
					
					if next( tablesRequiringForceUpdate ) then
					
						local reindexingStartTime = KSDate.timeNow()
					
						return self._docStore:changeCounter()
							:take( 1 )
							:flatMap( function( changeCounter )
							
								-- Make note of the current changeCounter. This value represents the highest changeCounter
								-- upto which we safely can map docs without looking up possible existing rows
								-- for the docId (and thus can drop the SQL indices).

								local fastReindexingLimit = changeCounter

								local initialEstimatedWork, scaleForSqlIndices = reportReindexingProgressEstimate( self, changeCounter, 
																					tablesRequiringForceUpdate,	needsUpdateFromForTables, 0)

								local ignoreTablesThatDontExist = true
								
								-- drop the indices
								dropSqlIndices( self )

								return fetchChangedDocsAndBringUpToDate(
										self,
										tablesRequiringForceUpdate,
										changeCounter,
										needsUpdateFromForTables,
										kMaxBatchSizeForDocMapping,
										ignoreTablesThatDontExist,
										fastReindexingLimit )
										
									:map( function()
										return changeCounter, initialEstimatedWork, scaleForSqlIndices
									end )
							
							end )
							
							:map( function( changeCounter, initialEstimatedWork, scaleForSqlIndices )
								
								-- fetchChangedDocsAndBringUpToDate fires with periodic updates as the
								-- tables are brought up to date. We can use this to measure reindxing
								-- progress.
								
								local needsUpdateFromForTables = populateNeedsUpdateFromForTables( self )
								
								reportReindexingProgressEstimate( self, changeCounter, tablesRequiringForceUpdate,
									needsUpdateFromForTables, 0, initialEstimatedWork, scaleForSqlIndices )
								
								return needsUpdateFromForTables, changeCounter, initialEstimatedWork, scaleForSqlIndices
								
							end )

							:takeLast( 1 )

							:flatMap( function( needsUpdateFromForTables, changeCounter, initialEstimatedWork, scaleForSqlIndices )

								return recreateSqlIndices( self )
									:map( function( value )
										reportReindexingProgressEstimate( self, changeCounter, tablesRequiringForceUpdate,
											needsUpdateFromForTables, value, initialEstimatedWork, scaleForSqlIndices )
										return value
									end )
									:takeLast( 1 )
									:map( function () 
										return needsUpdateFromForTables
									end )
								
							end )
							
							:map( function( needsUpdateFromForTables )
								
								if newVersionSequenceNumber then
									
									-- If had to fall back to hash-based reindexing because of an
									-- unknown sequence number, or for a full index store rebuild,
									-- we still want to update the sequence number after the reindex.
								
									setIndexStoreSchemaVersionInfo( db, newVersionSequenceNumber, 0 )
								end
								
								local duration = KSDate.timeNow() - reindexingStartTime
								
								logReindexingEvent( kCompletedReindexingEvent, nil, duration )
								
								self._reindexingProgress:done()
								
								return needsUpdateFromForTables
							
							end )
						
					else
					
						if newVersionSequenceNumber then
						
							-- If had to fall back to hash-based reindexing because of an
							-- unknown sequence number, or for a full index store rebuild,
							-- we still want to update the sequence number after the reindex.
						
							setIndexStoreSchemaVersionInfo( db, newVersionSequenceNumber, 0 )
							
						end
					
						-- no forced reindexing required, close progress stream anyway
						self._reindexingProgress:done()

						return Observable.just( needsUpdateFromForTables )
						
					end
				
				end

			end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Create a map server to support an index store implementation.

	Note that the map server assumes ownership (and close targeting) of
	\c docStore and \c indexStoreDb.

	\param docStore (\c WFDocStore or \c OzDocStore) doc store providing the data
	\param indexStoreDb (\c WFSQLiteDatabase) SQLite database for index store
	\param indexStoreSchema (\c WFRxIndexStoreSchema) index store schema description
]]--

function WFRxIndexStoreMapServer:__init( docStore, indexStoreDb, indexStoreSchema )

	check( 'WFRxIndexStoreMapServer', 'WFRxDocStore', 'WFSQLiteDatabase',
		   'WFRxIndexStoreSchema', '?table' )

	logger:infof( "Spinning up map server for index store %s", tostring( docStore ) )

	self:setOwned {
		_docStore = docStore,
		_db = indexStoreDb
	 }

	self._db_exec = indexStoreDb.exec

	WFCloseSupport.mustCloseAtThreadTermination( self )

	self._updateTimePerTable = {}

	self._tableNames = assert( indexStoreSchema:tableNames() )
	self._mapFunctions = assert( indexStoreSchema:mapFunctions() )
	self._sqlIndexInfos = assert( indexStoreSchema:indexInfos() )
	self._docValueFilters = assert( indexStoreSchema:docValueFilters() )
	self._relevantTypeAndSubtypesByTable = assert( indexStoreSchema:relevantTypeAndSubtypesByTable() )
	self._niceToHaveIndices = assert( indexStoreSchema:niceToHaveIndices() )
	self._tablesThatHaveOneRowPerDocument = assert( indexStoreSchema:tablesThatHaveOneRowPerDocument() )
	self._tablesThatUseLocalIds = assert( indexStoreSchema:tablesThatUseLocalIds() )
	self._fullTextSearchTables = assert( indexStoreSchema:tablesThatUseFTS4() )
	self._tablesThatNeedAggressiveUpdates = assert( indexStoreSchema:tablesThatNeedAggressiveUpdates() )
	self._nColumnsPerTable = assert( indexStoreSchema:nColumnsPerTable() )
	self._tablesThatShouldIncludeDocumentsWithoutSpaceIds = assert( indexStoreSchema:tablesThatShouldIncludeDocumentsWithoutSpaceIds() )
	self._tablesThatShouldIncludeDocumentsWithSpaceIds = assert( indexStoreSchema:tablesThatShouldIncludeDocumentsWithSpaceIds() )
	
	self._keepTableUpToDate = setmetatable( {}, { __weak = 'kv' } )

	self._existingRowQueryForTable = {}
	self._deleteExistingRowsStmtForTable = {}
	self._insertRowStmtForTable = {}
	self._forcedReindexingStats = {}
	
	self._isUnitTest = WFPlatformInfo.isUnitTest()
	
	self._nDocumentsToIndex = Rx.BehaviorSubject( 0, 0 )

	self._reindexingProgress = WFProgressReporter( "reindexing tables" )

 	-- Make sure doc store is the same one we thought it was before.

	self._isReady = docStoreCrossCheck( self, indexStoreSchema )
		:replay( nil, 1 )
	self._isReady:connect()

	local needsUpdateFromForTable, processing =
		makeNeedsUpdateFromForTable( self )

	local disposable = needsUpdateFromForTable:subscribe(function()end)

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

	self._needsUpdateFromForTable = needsUpdateFromForTable
	processing:resume()
	
	WFDiagnosticLog.addLogProvider( self )

end

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

local function isNativeDbClosed( self )
	
	return WFCloseSupport.hasBeenOrWillBeClosed( self ) or
			not self._db or
			WFCloseSupport.hasBeenOrWillBeClosed( self._db ) 

end

--------------------------------------------------------------------------------
-- runWithNativeDb throws if the db isn't alive
--------------------------------------------------------------------------------

function WFRxIndexStoreMapServer:runWithNativeDb( f, ... )

	assert( not isNativeDbClosed( self ), "can't use db, object is closed" )

	return f( self._db, ... )

end

--------------------------------------------------------------------------------
-- maybeRunWithNativeDb is a no-op if the db isn't alive
--------------------------------------------------------------------------------

function WFRxIndexStoreMapServer:maybeRunWithNativeDb( f, ... )

	if isNativeDbClosed( self ) then

		return

	end

	return self:runWithNativeDb( f, ... )

end

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

function WFRxIndexStoreMapServer:close()

	if self._isReady then
		self._isReady = Observable.throw( "_isReady: Map server closed" )
	end

	self._needsUpdateFromForTable = Observable.throw( "_needsUpdateFromForTable: Map server closed" )

end

--------------------------------------------------------------------------------
--[[!
	\brief Resolve when the index store database is ready for use.

	This is primarily used for testing.

	\return <code>P< true | "out of sync" ></code> resolves when we have
		verified that index store and doc store are properly related; resolves
		to true in most normal circumstances or "out of sync" if index and
		doc store were unrelated to each other
]]--

function WFRxIndexStoreMapServer:isReady()

	return self._isReady
		:map( function()
			return true
		end )

end

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

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

function WFRxIndexStoreMapServer:getReindexingProgress()

	return self._reindexingProgress:status()

end

--------------------------------------------------------------------------------
--[[!
	\brief What is the current doc store revision change counter?

	\return <code>Observable< number ></code> fires with the change number
	that the doc store currently has.
]]--

function WFRxIndexStoreMapServer:currentChangeCounter( tableName )
	return self._docStore:changeCounter():take(1)
end

--------------------------------------------------------------------------------
--[[!
	\brief Request that a table be kept up to date.

	This returns an observable that indicates the change number for which
	the table is up do date.  Subscribing to the observable triggers the
	system to start causing the table to be updated.

	Note: Unlike WFIndexStoreMapServer (the previous implementation) there
	is no way to find out if a table is up to date without causing it to
	start getting up to date.  We found that there was no need for that,
	and supporting it was actually causing the logic to be more complicated.

	\param tableName (string) name of index table to be brought up to date

	\return <code>Observable< number ></code> fires with the change number
	that the table is now up to date with.
]]--

function WFRxIndexStoreMapServer:keepTableUpToDate( tableName )

	check( 'WFRxIndexStoreMapServer', 'string' )

	if not self._tableNames[ tableName ] then
		error( "WFRxIndexStoreMapServer:bringTableUpToDate: no such table " .. tableName, 2 )
	end

	local result = self._keepTableUpToDate[ tableName ]
	
	if not result then
		result = self._needsUpdateFromForTable
			:map( function( tablesUpToDate, tableHasNewIndexData )
				
				local result = tablesUpToDate[ tableName ]
				if type( result ) ~= "number" then
					error( string.format( "error indexing table '%s': %s", tableName, result ) )
				end
				
				return result, tableHasNewIndexData[ tableName ]
				
			end )
			:distinctUntilChanged()
			:map( function( tableUpToDateCounter, tableHasNewIndexData )
			
				return {
					tableUpToDateCounter = tableUpToDateCounter,
					tableHasNewIndexData = tableHasNewIndexData,
				}
				
			end )
			:tapOnCompleted( function()
				self._changeTableIndexingStatus( nil, tableName )
				self._keepTableUpToDate[ tableName ] = nil
			end )
			:replay( nil, 1 )

		local watchForDispose = Rx.Disposable( function()
				self._changeTableIndexingStatus( nil, tableName )
				self._keepTableUpToDate[ tableName ] = nil
			end )

		watchForDispose:disposeWith( result:connect() )
			-- watchForDispose will be retained by result's subscription

		result = result
			:refCount()

		self._changeTableIndexingStatus( tableName, nil )
		self._keepTableUpToDate[ tableName ] = result
	end

	return result

end

--------------------------------------------------------------------------------
--[[
	Helper function to update the TEMP_ids table (if the next query to be run
	needs it).
]]--

local BATCH_SIZE = 32

local kBatchSql = [[INSERT INTO TEMP_ids ( docId )
		SELECT ?
		]] .. string.rep( "UNION SELECT ? ", BATCH_SIZE - 1 )

local function updateTempIdsImp( self, ids )

	local n = #ids
	local db = self._db
	local exec = db.exec
	self._lastTempIds = ids
	exec( db, [[DELETE FROM TEMP_ids]] )

	-- Do inserts in batches, which is more efficient. This cuts off
	-- around 20-30% of the total time.

	for i = 1, BATCH_SIZE * math.floor( n / BATCH_SIZE ), BATCH_SIZE do
		exec( db, kBatchSql, unpack( ids, i, i + BATCH_SIZE - 1 ) )
	end

	-- Any remaining are inserted one at a time.

	for i = BATCH_SIZE * math.floor( n / BATCH_SIZE ) + 1, n do
		exec( db, [[INSERT INTO TEMP_ids VALUES ( ? )]], ids[ i ] )
	end

end

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

local function updateTempIds( self, ids )

	local canSkip = false

	if self._lastTempIds and #self._lastTempIds == #ids then
		canSkip = true
		for i, id in ipairs( self._lastTempIds ) do
			if ids[ i ] ~= id then
				canSkip = false
				break
			end
		end
	end

	if canSkip then
		
		return

	end

	-- make sure we seed the TEMP_ids table under a transaction without nesting
	-- transactions
	
	if self._db:hasActiveTransaction() then

		return updateTempIdsImp( self, ids )

	end

	return self._db:withTransactionDo( updateTempIdsImp, self, ids )

end

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

local function makeTempTableReturningQuery( self, tempTableId, sql, ... )

	local tempTableName = 'temp_table_' .. tempTableId
	local createSql = string.format(
			'CREATE TEMPORARY TABLE IF NOT EXISTS %s ( docId INTEGER PRIMARY KEY )',
			tempTableName )
	self._db:exec( createSql )
	local clearTableSql = string.format(
			'DELETE FROM %s',
			tempTableName )
	self._db:exec( clearTableSql )
	sql = string.format( 'INSERT OR REPLACE INTO %s %s', tempTableName, sql )
	self._db:exec( sql, ... )
	local changes = self._db:exec [[SELECT changes() as N]]

	return tempTableName, changes.N

end

--------------------------------------------------------------------------------
--[[!
	\brief Run a query against the index database.

	No guarantee is made by this API that the tables are up to date.
	Use <code>:bringTableUpToDate()</code> to trigger an update.

	This query runs immediately in this thread.

	\param sql (string) SQL query string to run
	\param singleColumnQuery (optional boolean) If set, the query will run
		using gatherArray instead of gatherArrayOfArray.
	\param ... (optional) arguments to SQL query

	\return (table) array of arrays (outer arrays are result rows, inner arrays
		are columns)
]]--

function WFRxIndexStoreMapServer:query( sql, singleColumnQuery, encoder, ids, ... )

	check( 'WFRxIndexStoreMapServer', 'string' )

	if ids then
		updateTempIds( self, ids )
	end

	if encoder then

		local _, _, maybeTempTableId = string.find( encoder, '^< temp table >(.+)$' )
		if maybeTempTableId then
			return makeTempTableReturningQuery( self, maybeTempTableId, sql, ... )
		end
		return self._db:encodeResults( sql, encoderFn( encoder ), ... )

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

end

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

local kLongQueryLogThresholdInSecs = 0.5

--------------------------------------------------------------------------------
-- describeQuery is called when a sql query takes longer than 
-- kLongQueryLogThresholdInSecs

local function describeQuery( desc, _, sql, encoder, ids )

	local extraQueryStepDesc = ""

	if encoder then

		extraQueryStepDesc = " with encoder step"

	elseif type( ids ) == "table" then

		extraQueryStepDesc = string.format( " with %s ids in temp table", #ids )

	end

	local oneLineSQL = sql:gsub( "\n", " " ):gsub( "%s%s*", " " )

	return string.format( "%s%s: %s", desc, extraQueryStepDesc, oneLineSQL )

end

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

WFRxIndexStoreMapServer.query =
	WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithDynamicDescription(
		"IndexStore", kLongQueryLogThresholdInSecs, "query", describeQuery,
		WFRxIndexStoreMapServer.query )

--------------------------------------------------------------------------------
--[[!
	\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 WFRxIndexStoreMapServer:sizeOfIndexQueue()

	return self._nDocumentsToIndex

end

--------------------------------------------------------------------------------
--[[!
	\brief Return information about index update performance.

	Intended primarily for use in performance instrumentation (i.e. diagnostic log).

	\return (table) array of the following statistics, per index table that has been updated:
		- \c tableName: (string) table name as declared in SQL
		- \c max: (number) longest time it has taken to update this table
		- \c min: (number) shortest time it has taken to update this table
		- \c total: (number) total time spent for all updates to this table
		- \c count: (number) how many times has this table been updated?

	All times are in seconds.
]]--

function WFRxIndexStoreMapServer:indexUpdatePerformanceStats()

	local sortedTableUpdateTimes = {}
	local needsUpdateFromForTable = {}
	
	-- If self._needsUpdateFromForTable has a value now this will
	-- fetch it synchronously.
	self._needsUpdateFromForTable:subscribe( function( t )
		needsUpdateFromForTable = t
	end ):dispose()

	local nSortedTableEntries = 0
	local totalIndexingTime = 0
	for _, tableUpdatePerf in pairs( self._updateTimePerTable ) do
		nSortedTableEntries = nSortedTableEntries + 1
		totalIndexingTime = totalIndexingTime + tableUpdatePerf.total
		sortedTableUpdateTimes[ nSortedTableEntries ] = {
			max = tableUpdatePerf.max,
			min = tableUpdatePerf.min,
			total = tableUpdatePerf.total,
			count = tableUpdatePerf.count,
			niceToHaveUpdatesIgnored = tableUpdatePerf.niceToHaveUpdatesIgnored,
			tableName = tableUpdatePerf.tableName,
			tableNeedsUpdateFrom = needsUpdateFromForTable[ tableUpdatePerf.tableName ],
		}
	end
	
	table.insert( sortedTableUpdateTimes, {
		total = totalIndexingTime,
		tableName = 'TOTAL FOR ALL TABLES'
		} )
		
	table.sort( sortedTableUpdateTimes, function( a, b )
		return a.total > b.total
	end )

	return sortedTableUpdateTimes

end

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

function WFRxIndexStoreMapServer:sendDiagnosticLogData()
	
	return {
		id = 'WFRxIndexStoreMapServer-' .. WFCurrentThread.DEBUG_getThreadName():replace( ' :: ', '+' ),
		name = string.format( 'Map Server :: %s', WFCurrentThread.DEBUG_getThreadName() ),
		weight = 1700,
		parentSection = WFDiagnosticLog.Sections.INDEX_STORE,
		body = {
			["Index Store Initialization"] = self._tableUpdatesAtInitialization,
			["Index Store Table Indexing Performance"] = self:indexUpdatePerformanceStats(),
		}
	}

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 WFRxIndexStoreMapServer:getSQLiteFatalErrorCodeStream()

	check( 'WFRxIndexStoreMapServer' )

	assert( not isNativeDbClosed( self ), "can't use db, object is closed" )

	return self._db:getSQLiteFatalErrorCodeStream()

end

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

function WFRxIndexStoreMapServer:DEBUG_getMapFunctionHashes()

	check( 'WFRxIndexStoreMapServer' )

	local sql = [[SELECT indexTableName, mapFunctionHash FROM indexStatus ORDER BY indexTableName ASC]]

	return self._isReady
		:map( function()
			return self._db:gatherArrayOfArray( sql )
		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 WFRxIndexStoreMapServer:DEBUG_getForcedReindexingStats()

	check( 'WFRxIndexStoreMapServer' )

	return Observable.just( self._forcedReindexingStats )

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 WFRxIndexStoreMapServer:DEBUG_setIndexStoreSchemaVersionInfo( currentVersionSequenceNumber, inProgressVersionSequenceNumber )

	check( 'WFRxIndexStoreMapServer', '?number', '?number' )

	logger:debugf( " DEBUG_setIndexStoreSchemaVersionInfo CALLED %d %d ", currentVersionSequenceNumber, inProgressVersionSequenceNumber )

	setIndexStoreSchemaVersionInfo( self._db, currentVersionSequenceNumber, inProgressVersionSequenceNumber )

	return Observable.empty()

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 WFRxIndexStoreMapServer:DEBUG_getIndexStoreSchemaVersionInfo()

	check( 'WFRxIndexStoreMapServer' )

	local currentSequence, inProgressSequence = getIndexStoreSchemaVersionInfo( self._db )
	
	return Observable.just( currentSequence, inProgressSequence )

end

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

return Rx.Class 'WFRxIndexStoreMapServer'( WFRxIndexStoreMapServer )
