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

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

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

	-- RxLua
local Rx = require 'Rx'
local BehaviorSubject = Rx.BehaviorSubject
local Observable = Rx.Observable
local ReplaySubject = Rx.ReplaySubject

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

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

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

	-- WF.core.rxlua
local WFRxWeakObservableTable = require 'WFRxWeakObservableTable'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.table
local WFTableArchiving = require 'WFTableArchiving'
local WFMsgPack = require 'WFMsgPack'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFConfiguration = require 'WFConfiguration'
local WFCurrentThread = require 'WFCurrentThread'
local WFTableMerge = require 'WFTableMerge'
local WFPerformanceUtils = require 'WFPerformanceUtils'
local WFPerformanceReport = require 'WFPerformanceReport'
local WFPlatformInfo = require 'WFPlatformInfo'

	-- WF.sqlite.database
local WFSQLiteDatabase = require 'WFSQLiteDatabase'
local WFSQLiteDatabaseSchema = require 'WFSQLiteDatabaseSchema'
local WFSQLiteUtils = require 'WFSQLiteUtils'

	-- WF.sqlite.doc_store
local WFRxDocStoreUtils = require "WFRxDocStoreUtils"

   -- docStoreMigration
local LrDocStoreTypeMigrator = require 'LrDocStoreTypeMigrator'

local WFCapabilities = require "WFCapabilities"

local WFAnalytics = require "WFAnalytics"

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

local docStoreImpLog = WFLogger 'WFRxDocStoreImp'

local changedDocsLog = WFLogger 'WFRxDocStoreImp :: changedDocs'

local compactionLog = WFLogger 'WFRxDocStore :: compact'

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

local check = assert( WFArguments.check )

local onlyFieldsInSpec = assert( WFRxDocStoreUtils.onlyFieldsInSpec )
local contentFromFlattenedValue = assert( WFRxDocStoreUtils.contentFromFlattenedValue )
local isMsgPacked = assert( WFRxDocStoreUtils.isMsgPacked )
local unpackDocValue = assert( WFRxDocStoreUtils.unpackDocValue )

local KSTableUtils_deepCopy = assert( KSTableUtils.deepCopy )
local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )

local string_byte = assert( string.byte )
local string_format = string.format

local math_floor = assert( math.floor )

local WFTableArchiving_tableToString = assert( WFTableArchiving.tableToString )

local WFMsgPack_pack = assert( WFMsgPack.pack )
local tableToString = assert( WFMsgPack_pack )

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

local emptyArchive = tableToString( {} )

local emptyTombstone = tableToString( { _deleted = true } )

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

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

local kAllDocuments = "all"
local kCatalogDocuments = "catalog"
local kSpaceFeedDocuments = "spaceFeed"

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

local kDocStoreTypeMigratorProgress = 'docStoreTypeMigratorProgress'

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

local kAllDocTypesAndSubtypes = {
	[ "*" ] = { "*" }
}

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

local function isAllDocTypesAndSubtypes( typeSubtypeMap )

	return KSTableUtils.deepEquals( typeSubtypeMap, kAllDocTypesAndSubtypes )

end

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

-- SPAM [ashj 2017-08-02]: This string is being referred elsewhere
-- to handle the specific error. If you are changing here, please
-- do handle the usages as well.
local ERROR_REVISION_WITHOUT_PARENT = "revision_without_parent"

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

local function errorf( ... )
	error( string_format( ... ), 2 )
end

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

local makeOngoingDocValueForDocId -- forward declaration

local makeOngoingLocalDocValueForDocName -- forward declaration

local bulkLocalIdToDocId_inTransaction -- forward declaration

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

-- @mark enable/disable LRDocStoreTypeMigrator.

local docStoreSchemaVersion, docsTableAlterStatements, docsTableCreationSql

docStoreSchemaVersion = '1.2'

docsTableAlterStatements = {
	[[ALTER TABLE docs ADD COLUMN spaceId TEXT]], -- WF-415
	[[ALTER TABLE docs ADD COLUMN type TEXT]],    -- WF-1128
	[[ALTER TABLE docs ADD COLUMN subtype TEXT]], -- WF-1128
}

docsTableCreationSql =
	[[CREATE TABLE docs( localDocId INTEGER PRIMARY KEY,
						 fullDocId TEXT UNIQUE NOT NULL,
						 winningRevSequence INTEGER REFERENCES revs( sequence ) ON DELETE SET NULL,
						 mostRecentRevSequence INTEGER REFERENCES revs( sequence ) ON DELETE SET NULL,
						 deleted BOOLEAN DEFAULT 0,
						 hasConflicts BOOLEAN DEFAULT 0,
						 annotation BLOB,
						 spaceId TEXT,
						 type TEXT,
						 subtype TEXT )]]

-- @mark database schema

local pageSize = WFConfiguration.DocStore.SQLITE_PAGE_SIZE or 8192

local schema = WFSQLiteDatabaseSchema( docStoreSchemaVersion ) -- Last incremented for WF-1128
	:useWAL()
	:useForeignKeys()
	:setSynchronous( 1 )
	:setPageSize( pageSize )

--------------------------------------------------------------------------------
--[[
	Map public document IDs to private (integer) document IDs for
	faster reference through the rest of the schema.
]]--

schema:addTable( docsTableCreationSql )

	:onConflict( 'alter', docsTableAlterStatements )

		-- WF-415, WF-1128: The spaceId, type and subtype columns were added in a later version
		-- of the doc store. We must include conflict resolution logic that creates the column
		-- in the event that the local table schema does not match what is described above.

schema:addIndex
	[[CREATE INDEX docs_mostRecentRevSequence ON docs( mostRecentRevSequence, spaceId )]]

schema:addIndex
	[[CREATE INDEX docs_winningRevSequence ON docs( winningRevSequence )]]

schema:addIndex
	[[CREATE INDEX docs_fullDocId ON docs( fullDocId )]]

schema:addIndex
	[[CREATE INDEX docs_spaceId ON docs( spaceId ) WHERE spaceId NOT NULL]]

schema:addIndex
	[[CREATE INDEX docs_typeSubtypeWinningRevSequence ON docs( type, subtype, winningRevSequence )]]

-------------------------------------------------------------------------------
-- Store actual revision content.

schema:addTable
	[[CREATE TABLE revs( sequence INTEGER PRIMARY KEY AUTOINCREMENT,
						 localDocId INTEGER NOT NULL REFERENCES docs( localDocId ) ON DELETE CASCADE,
						 revId TEXT NOT NULL COLLATE BINARY,
						 parentRevSequence INTEGER REFERENCES revs( sequence ) ON DELETE SET NULL,
						 current BOOLEAN,
						 deleted BOOLEAN DEFAULT 0,
						 content BLOB )]]

	:onConflict 'replaceAll'

	-- TODO [scouten 2015-08-10]: Remove this provision in a month or so. See also
	-- corresponding test in WFRxDocStoreImpTest.

	-- TODO [scouten 2015-05-13]: Consider moving content out to its own field?

schema:addIndex
	[[CREATE INDEX revs_localDocIdAndCurrent on revs(localDocId, current)]]

schema:addIndex
	[[CREATE INDEX revs_localDocIdAndRevIdAndSequence ON revs( localDocId, revId, sequence )]]

schema:addIndex
	[[CREATE INDEX revs_sequences ON revs( sequence )]]

schema:addIndex
	[[CREATE INDEX revs_parentRevSequence ON revs( parentRevSequence )]]

--------------------------------------------------------------------------------
-- Store labels.

schema:addTable
	[[CREATE TABLE labels( localDocId INTEGER REFERENCES docs( localDocID ) ON DELETE CASCADE,
						   revSequence INTEGER REFERENCES revs( sequence ) ON DELETE CASCADE,
						   name STRING,
						   UNIQUE( localDocId, name ) )]]

schema:addIndex
	[[CREATE INDEX labels_revSequence on labels(revSequence)]]

--------------------------------------------------------------------------------
-- Store local documents that do not sync.

schema:addTable
	[[CREATE TABLE localDocs( name TEXT UNIQUE NOT NULL,
							  content BLOB )]]

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

-- @mark WFRxDocStoreImp class

--------------------------------------------------------------------------------
--[[!
	\class WFRxDocStoreImp

	TO DO: Write new documentation for this.
]]--

local WFRxDocStoreImp = {

	_db = false,
		-- (WFSQLiteDatabase) database file

	_exec = false,
		-- (function) cached reference to db:exec

	_gatherArrayOfArray = false,
		-- (function) cached reference to db:gatherArrayOfArray

	_path = false,
		-- (string) path to doc store DB

	_changeCounter = false,
		-- (BehaviorSubject< number >) most recent rev change counter value

	_deletedDocStream = false,
		-- (ReplaySubject< table>) A stream that fires with deleted doc values
			-- (or a shell of the deleted documents for purged spaces)

	_ongoingDocValues = false,
		-- (WFRxWeakObservableTable) document value subjects that are currently active
			-- key: (string) doc ID
			-- value: (Rx.ReplaySubject) current document value

	_ongoingLocalDocValues = false,
		-- (WFRxWeakObservableTable) local document value subjects that are currently active
			-- key: (string) doc name
			-- value: (Rx.ReplaySubject) current document value

	_docIdToLocalIdCache = false,
		-- (table) simple cache of non-deleted full doc ids to local ids that have been fetched once already

	_localIdToDocIdCache = false,
		-- (table) simple cache of non-deleted local ids to full doc ids that have been fetched once already

	_bulkRequestInfo = false,
		-- (table) indexed by operation name, contains tuples of num calls and
			-- num docs requested, used by diagnostic log and debug panel.

	_typeAndSubtypeDiagnosticData = false,
		-- (table) Data gathered around type/subtype extraction metrics. Only used in the diagnostic log.

	_TEST_unpackDocValue = false,
		-- (boolean) set to true for WFRxDocSToreImpTest

	WFCloseSupport.addCloseSupportMethods,

}

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

local function openAndObtainDatabase( self, dbPath )

	self._path = dbPath
	docStoreImpLog:infof( 'dbPath = %s', dbPath )

	-- show housekeeping dialog if new docStore indexes are about to be created
	local dbOptions = {
		installProgressHandlerForDocStoreInit = true
	}

	-- NOTE [immisch 2021-11-24]: this is needed for the readOnly docStore
	schema._allowMultipleConnections = true

	local db = WFSQLiteDatabase( dbPath, schema, dbOptions )
	self:setOwned { _db = db }

	self._exec = assert( db.exec )
	self._gatherArrayOfArray = assert( db.gatherArrayOfArray )

	return db

end

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

local function maybeCloseAndClearDatabase( self )

	if not self._db then

		return

	end

	self._db:close()

	self._db = nil

end


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

function WFRxDocStoreImp:__init( path )

	check( 'WFRxDocStoreImp', 'string' )

	-- Map :memory:(random) and :temp:(random) paths back to SQLite pseudo-paths.

	local dbPath = path

	if dbPath:match( '^:memory:' ) then

		dbPath = ':memory:'

	elseif dbPath:match( '^:temp:' ) then

		dbPath = ':temp:'

	elseif #dbPath > 0 then

		-- An empty path is legal and means create a temporary path on disk.

		local parentDir = KSPathUtils.parent( dbPath )
		WFFileUtils.mkdir( parentDir )

	end

	local canBackFillTypeAndSubtype, maybeSpaceNeeded
			= LrDocStoreTypeMigrator.canBackFillTypeAndSubtype( dbPath )

	-- Track diagnostic data for use in the diagnostic log:

	self._typeAndSubtypeDiagnosticData = {
		canBackFillTypeAndSubtype = canBackFillTypeAndSubtype,
		maybeSpaceNeeded = maybeSpaceNeeded,
	}

	if not canBackFillTypeAndSubtype
			and maybeSpaceNeeded
			and maybeSpaceNeeded > 0 then

		WFCurrentThread.broadcastNotificationWithUserData( kDocStoreTypeMigratorProgress,
			{
				dbPath = path,
				hasSpaceError = true,
				spaceNeeded = maybeSpaceNeeded
			} )

		docStoreImpLog:warnf( "Unable to perform required doc store type and subtype extraction: not enough space. Need %d bytes.", maybeSpaceNeeded )

		local event = {}

		event['event.type'] = 'warning'
		event['event.subtype'] = 'type-and-subtype-extraction-not-enough-space'
		event['event.appWorkflow'] = 'LAUNCH'
		event['event.subcategory'] = 'Migration'
		event['event.spaceNeeded'] = maybeSpaceNeeded

		WFAnalytics.trackEvent( event )

	end

	-- Create the actual database. Note that we disable query performance logging
	-- because the query manifest is very simple for this database and we've done
	-- what can be done IMHO.

	local db = openAndObtainDatabase( self, dbPath )

	-- Check for need of type/subtype backfill ( one time migration step )
	local docsWithTypeBackFillRequired = LrDocStoreTypeMigrator.docsWithTypeBackFillRequired( self )

	self._typeAndSubtypeDiagnosticData[ 'docsWithTypeBackFillRequired' ] = docsWithTypeBackFillRequired

	if docsWithTypeBackFillRequired > 0 then

		local success, maybeErrorMessage, plausibleTypesAndSubtypes, notPlausibleTypesAndSubtypes =
				LrDocStoreTypeMigrator.backFillTypeAndSubtype( self )

		-- Track diagnostic data for use in the diagnostic log:

		self._typeAndSubtypeDiagnosticData[ 'extractionSucceeded' ] = success

		if maybeErrorMessage and maybeErrorMessage ~= '' then
			self._typeAndSubtypeDiagnosticData[ 'extractionError' ] = maybeErrorMessage
		end

		self._typeAndSubtypeDiagnosticData[ 'plausibleTypesAndSubtypes' ] = plausibleTypesAndSubtypes
		self._typeAndSubtypeDiagnosticData[ 'notPlausibleTypesAndSubtypes' ] = notPlausibleTypesAndSubtypes

	end

	-- this will stop the initialization progress
	WFCurrentThread.broadcastNotificationWithUserData( kDocStoreTypeMigratorProgress,
		{
			dbPath = path,
			done = true
		} )

	self._docIdToLocalIdCache = {}
	self._localIdToDocIdCache = {}

	-- Set up change notification.

	local changeCounterSeedRow = db:execOnce(
		"SELECT max( sequence ) AS changeCounter FROM revs" )

	local changeCounter = ( changeCounterSeedRow
		and changeCounterSeedRow.changeCounter or 0 )

	self._changeCounter = BehaviorSubject( changeCounter )

	self._deletedDocStream = ReplaySubject( 1 )

	self._ongoingDocValues = WFRxWeakObservableTable(
		function( docId )
			return makeOngoingDocValueForDocId( self, docId )
		end )

	self._ongoingLocalDocValues = WFRxWeakObservableTable(
		function( name )
			return makeOngoingLocalDocValueForDocName( self, name )
		end )

	self._bulkRequestInfo = {}

	WFDiagnosticLog.addLogProvider( self )

end

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

function WFRxDocStoreImp:close()

	maybeCloseAndClearDatabase( self )

	if self._changeCounter then
		self._changeCounter:onCompleted()
		self._changeCounter = nil
	end

	if self._ongoingDocValues then
		self._ongoingDocValues:dispose()
		self._ongoingDocValues = nil
	end

	if self._ongoingLocalDocValues then
		self._ongoingLocalDocValues:dispose()
		self._ongoingLocalDocValues = nil
	end

	if self._deletedDocStream then
		self._deletedDocStream:onCompleted()
		self._deletedDocStream = nil
	end

end

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

local function assertNotClosed( self, methName )

	if self._db then

		return

	end

	local errMsg = string.format(
					"WFRxDocStoreImp:%s: doc store has been closed",
					type( methName ) == "string" and
						methName or "<unknown method>" )

	docStoreImpLog:error( errMsg )

	error( errMsg )

end


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

-- @mark shared utilities

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

local underscoreByte = string_byte( '_' )

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

local function assertDocIdIsValid( docId )

	check( '+string|number' )

	if type( docId ) == 'number' then
		assert( docId > 0 and math_floor( docId ) == docId,
				"local id must be positive integer, was " .. tostring( docId ) )
	else
		assert( docId:byte( 1 ) ~= underscoreByte,
				"document ID must not start with _" )

		assert( docId:match( "^[0-9a-zA-Z_.%-]+$" ),
				"document ID must only contain letters, numbers, and _.-" )
	end

	return docId

end
--------------------------------------------------------------------------------

local function assertSpaceIdIsValid( spaceId )

	check( '+string' )

	assert( spaceId:byte( 1 ) ~= underscoreByte,
			"space ID must not start with _" )

	assert( spaceId:match( "^[0-9a-zA-Z_.%-]+$" ),
			"space ID must only contain letters, numbers, and _.-" )

	return spaceId

end

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

local function createLocalDocId_inTransaction( db, exec, fullDocId )

	exec( db, [[INSERT INTO docs( fullDocId ) VALUES( ? )]], fullDocId )

	return db:getLastInsertRowId()

end

--------------------------------------------------------------------------------
-- Returns four values:
	-- 1: winning revision ID for this document ID
	-- 2: what is its sequence number?
	-- 3: is this revision deleted?
	-- 4: are there any conflicting revisions?

local function winningRevIdForLocalDocId_inTransaction( self, localDocId, favorSequence )

	assert( localDocId > 0 )

	local revRows = self._db:gatherArrayOfArray(
		[[SELECT revId, sequence, deleted
			FROM revs
				WHERE localDocId = ? AND current = 1
				ORDER BY ( sequence == ? ) DESC, deleted ASC, revId DESC
				LIMIT 2]],
		localDocId, favorSequence )

	local winningRow = revRows[ 1 ]

	if winningRow then

		local deleted = winningRow[ 3 ] == 1
		local conflict = false

		if not deleted then

			local maybeConflictingRow = revRows[ 2 ]
			conflict = not not( maybeConflictingRow and maybeConflictingRow[ 3 ] == 0 )

		end

		return winningRow[ 1 ], winningRow[ 2 ], deleted, conflict

	end

end

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

local acceptableTypes = {
	[ 'boolean' ] = true,
	[ 'number' ] = true,
	[ 'string' ] = true,
	[ 'table' ] = true,
}

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

local approvedTopLevelKeys = {
	_id = true,
	_revId = true,
	_revIds = true,
	_localId = true,
	_deleted = true,
	_localOnly = true,
	_labels = true,
	_serverProvidedMetadata = true,
	_newRevId = true,
	_newLabel = true,
	_preserveLabels = true,
	_spaceId = true,
	_newId = true
}

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

local approvedTopLevelKeysOverlayOnly = {
	_newIdOverlay = true
}

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

local kInfinity = 1/0
local kNegativeInfinity = -kInfinity

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

local function assertDocValueIsValidImp( value, overlayOrDocValue, nestingLevel )

	local isOverlay = overlayOrDocValue == 'overlay'

	nestingLevel = nestingLevel or 1

	if nestingLevel > 20 then
		error "table nested too deeply"
	end

	if getmetatable( value ) then
		error "table contains metatable"
	end

	if nestingLevel == 1 and value._deleted ~= nil and value._deleted ~= true then
		errorf( "value contains illegal _deleted value: %s (must be nil or true)",
				tostring( value._deleted ) )
	end

	if nestingLevel == 1 and value._newId ~= nil and value._deleted ~= true then
		errorf( "value contains _newId value: %s, but not _deleted (must be true)",
				tostring( value._newId ) )
	end

	if nestingLevel == 1 and value._localOnly ~= nil and type( value._localOnly ) ~= 'table' then
		errorf( "_localOnly must be a table" )
	end

	-- labels starting with an underscore are allowed
	if nestingLevel == 1 and value._labels then

		local lv = value[ '_labels' ]

		for label, labelTrue in pairs( lv ) do

			if type( label ) ~= 'string' or #label == 0 then
				error( "label name must be non-empty string" )
			end

			if labelTrue ~= true then
				error( "label value must be true" )
			end

		end

	end

	local hasStringKeys, hasNumericKeys

	for k, v in pairs( value ) do

		local tk = type( k )

		if tk == 'string' then

			if nestingLevel == 1 and string_byte( k, 1 ) == underscoreByte and ( not approvedTopLevelKeys[ k ] and not ( isOverlay and approvedTopLevelKeysOverlayOnly[ k ] ) ) then
				errorf( "value contains key %s with leading underscore (%s) in level %d", k, k, nestingLevel )
			end

			hasStringKeys = true

		elseif tk == 'number' then

			if nestingLevel == 1 then
				error( "numeric keys not allowed at top level of doc value" )
			end

			if k <= 0 or math_floor( k ) ~= k then
				error( "numeric keys must be positive integers" )
			end

			if k == kInfinity then
				-- Note that Lua disallows NaN as key, so no need to test for that case here.

				errorf( "key %s is an invalid number", tostring( k ) )

			end

			hasNumericKeys = true

		else

			errorf( "key %s is of type %s, must be string%s",
					tostring( k ), tk,
					nestingLevel > 1 and " or number" or "" )

		end

		local tv = type( v )

		if v == kInfinity or v == kNegativeInfinity or v ~= v then
			errorf( "value %s is an invalid number", tostring( v ) )
		end

		if not acceptableTypes[ tv ] then
			errorf( "value at key %s is of type %s (must be scalar or table)", k, tv )
		end

		if nestingLevel == 1 and k == '_revIds' then

			if tv ~= 'table' then
				error( "_revIds must be a table if present" )
			end

			for i, revId in pairs( v ) do

				if i == 'force' and revId == true then

					-- OK: allow-listed

				else

					if type( i ) ~= 'number' then
						error( "_revIds must be an array of revision IDs" )
					end

					if type( revId ) ~= 'string' then
						error( "_revIds must be an array of (string) revision IDs" )
					end

				end

			end

		elseif tv == 'table' then

			assertDocValueIsValidImp( v, overlayOrDocValue, nestingLevel + 1 )

		end

	end

	if hasStringKeys and hasNumericKeys then
		error( "tables must not have both string and numeric keys" )
	end

end

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

local function assertDocOverlayIsValid( overlay )
	return assertDocValueIsValidImp( overlay, 'overlay' )
end

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

local function assertDocValueIsValid( docValue )
	return assertDocValueIsValidImp( docValue, 'docValue' )
end

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

local function documentRevisionExists_inTransaction( db, exec, localDocId, revId )

	local revisionRow = exec( db,
			[[SELECT sequence FROM revs WHERE localDocId = ? AND revId = ?
				ORDER BY sequence DESC LIMIT 1]],
			localDocId, revId )

	return not not revisionRow

end

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

local function getSequenceForDocumentAndRevision_inTransaction( db,
																exec,
																localDocId,
																revId,
																onlyCurrent )

	-- Choose the appropriate SQL query. The "easy" way would be to do
	-- string.format to assemble the query, but this is likely to be called
	-- frequently enough that the formatting overhead will be noticable.
	-- So we'll spell out the variations explicitly.

	local sql

	if onlyCurrent then

		sql = [[SELECT sequence, revId FROM revs
					WHERE localDocId = ? AND revId = ? AND current = 1
					ORDER BY sequence DESC LIMIT 1]]

	else

		sql = [[SELECT sequence, revId FROM revs
					WHERE localDocId = ? AND revId = ?
					ORDER BY sequence DESC LIMIT 1]]

	end

	local sequenceRow = exec( db, sql, localDocId, revId )

	if sequenceRow then
		return sequenceRow.sequence, sequenceRow.revId
	end

end

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

local function updateIdCaches( self, localId, fullDocId, isDeleted )

	if isDeleted then
		-- remove from caches
		self._docIdToLocalIdCache[ fullDocId ] = nil
		self._localIdToDocIdCache[ localId ] = nil
	else
		-- add to caches
		self._docIdToLocalIdCache[ fullDocId ] = localId
		self._localIdToDocIdCache[ localId ] = fullDocId

	end

end

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

local function addToBulkRequestInfo( self, bulkOperation, description, numDocs )

	description = description or 'unknown'

	local bulkOperationHashMap = self._bulkRequestInfo[ bulkOperation ] or {}
	self._bulkRequestInfo[ bulkOperation ] = bulkOperationHashMap

	local bulkCount = bulkOperationHashMap[ description ] or { 0, 0 }
	bulkCount[ 1 ] = bulkCount[ 1 ] + 1
	bulkCount[ 2 ] = bulkCount[ 2 ] + numDocs
	bulkOperationHashMap[ description ] = bulkCount

end

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

-- @mark whole document values

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


local function getLabels( db, sequence )

	local labels = nil

	local labelRows = db:gatherArray( [[SELECT labels.name FROM labels WHERE revSequence = ?]],
											 sequence )

	if #labelRows > 0 then

		labels = {}

		for _, labelName in ipairs( labelRows ) do
			labels[ labelName ] = true
		end

	end

	return labels

end

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

local function revisionRow_inTransaction( db, exec, docId, revId )

	local swapIds = type( docId ) == 'number'
		-- after querying, swap localId and docId
	local whereIdCol = swapIds and 'docs.localDocId' or 'fullDocId'
	local selectIdCol = swapIds and 'fullDocId' or 'docs.localDocId'

	local revisionRow
	if revId then

		revisionRow = exec( db,
			[[SELECT ]] .. selectIdCol .. [[ AS selectIdCol, revId, sequence, content,
					annotation, revs.deleted, hasConflicts, spaceId
				FROM revs
					INNER JOIN docs ON docs.localDocId = revs.localDocId
				WHERE
					]] .. whereIdCol .. [[ = ? AND revId = ?
				ORDER BY sequence DESC LIMIT 1]], docId, revId )

			-- EXPLAIN QUERY PLAN says this uses automatic indexes on both
			-- revs and docs tables.

		if not revisionRow then

			-- If we don't find a match on rev ID, try matching it as a label.

			local labelRow = exec( db,
				[[SELECT revSequence
					FROM labels
						INNER JOIN docs ON docs.localDocId = labels.localDocId
					WHERE
						]] .. whereIdCol .. [[ = ? AND name = ?]], docId, revId )

			if labelRow and labelRow.revSequence then
				revisionRow = exec( db,
					[[SELECT ]] .. selectIdCol .. [[ AS selectIdCol, revId, sequence, content,
							annotation, revs.deleted, hasConflicts, spaceId
						FROM revs
							INNER JOIN docs ON docs.localDocId = revs.localDocId
						WHERE
							]] .. whereIdCol .. [[ = ? AND sequence = ?]], docId, labelRow.revSequence )
			end

		end

	else

		revisionRow = exec( db,
			[[SELECT ]] .. selectIdCol .. [[ AS selectIdCol, revId, sequence, content, annotation, hasConflicts,
					docs.deleted AS deleted,
					docs.mostRecentRevSequence AS mostRecentRevSequence,
					spaceId
				FROM revs
					INNER JOIN docs ON docs.winningRevSequence = revs.sequence
				WHERE
					]] .. whereIdCol .. [[ = ?]], docId )

			-- EXPLAIN QUERY PLAN says this uses automatic index on docs and
			-- integer primary key on revs.

	end

	if revisionRow and revisionRow.content == nil then
		error "Revision has been purged"
	end

	return revisionRow
end

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

local function docValue_inTransaction( self, db, exec, docId, revId, args )

	args = args or {}
	local ignoreNonMatch = args.ignoreNonMatch
	local addDeletionMarker = args.addDeletionMarker
	local unpackDoc = args.unpackDocValue or self._TEST_unpackDocValue

	local simplifyDeletedDocs = true

	if args.simplifyDeletedDocs ~= nil then
		simplifyDeletedDocs = args.simplifyDeletedDocs
	end

	local revisionRow = revisionRow_inTransaction( db, exec, docId, revId )

	if revId then

		if not revisionRow then
			if ignoreNonMatch then
				return
			else
				errorf( "WFRxDocStoreImp:docValue: no such revision or label %s for %s", revId, docId ) end
			--end  -- funky formatting because coverage views this as an unreachable line :-(
		end

	else

		if not revisionRow or ( revisionRow.deleted == 1 and simplifyDeletedDocs ) then
			local content = revisionRow and contentFromFlattenedValue( revisionRow.content ) or {}

			local deletedDocValue = {
				_id = docId,
				_revId = ( revisionRow and revisionRow.revId ) or nil,
				_empty = true,
				_conflict = ( revisionRow and revisionRow.hasConflicts ~= 0 ) or nil,
				_deleted = ( addDeletionMarker and revisionRow and revisionRow.deleted == 1 ) or nil,
				_newId = content._newId,
			}

			if revisionRow then
				deletedDocValue._labels = getLabels( db, revisionRow.sequence )
			end

			if unpackDoc then
				return deletedDocValue
			end

			return { deletedDocValue, nil, nil }
		end

	end

	local docValue = {}

	local swapIds = type( docId ) == 'number'

	if swapIds then
		docValue._id = revisionRow.selectIdCol
		docValue._localId = docId
	else
		docValue._id = docId
		docValue._localId = revisionRow.selectIdCol
	end

	docValue._revId = revisionRow.revId
	docValue._changeCounter = revisionRow.mostRecentRevSequence or revisionRow.sequence
	docValue._spaceId = revisionRow.spaceId

	-- Special flag to request the initial change counter, not the counter related to this revision.
	if args.includeInitialChangeCounter then
		local firstSequenceRow = exec( db,
			[[SELECT sequence FROM revs WHERE localDocId == ? ORDER BY sequence ASC LIMIT 1]], docValue._localId )
		docValue._initialChangeCounter = KSTableUtils_derefOpt( firstSequenceRow, 'sequence' )
			or docValue._changeCounter
	end

	if revisionRow.hasConflicts == 1 then
		docValue._conflict = true
	end

	-- Merge in any labels that may be associated with this revision.

	local labels = getLabels( db, revisionRow.sequence )
	if labels then
		docValue._labels = labels
	end

	if unpackDoc then
		return unpackDocValue( { docValue, revisionRow.content, revisionRow.annotation } )
	end

	return { docValue, revisionRow.content, revisionRow.annotation }

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the value of a single document once.

	\param docId (string) the document's unique ID
	\param revId (optional, string) revision to retrieve (can also be a label name)
	\param optFieldSpec (optional table) This is an ignored param and is only necessary
		so that the interface to WFRxDocStoreImp:docValue exactly matches that
		of WFRxDocStore:docValue.
	\param simplifyDeletedDoc (optional boolean) Defaults to true. If set to false,
		then deleted docs will be returned in their full form. If true or nil, a
		simplified tombstone will be returned in place of a full deleted document.

	\return <code>Observable< table ></code> value for this document

	Note that the document content will be annotated with the following
	system-generated entries (at the top level of the value table):

		* \c _id (string): the document's unique ID
		* \c _empty (boolean): true if there isn't actually a value for this doc ID
		* \c _deleted (boolean): true if this revision is deleted (only used when
			\c revId is present
		* \c _changeCounter (number): a monotonically-increasing change counter
			that is increased whenever any branch of this document is revised
			(The functions \c changeCounter and \c changeCounterOnce will
			return the maximum value of \c _changeCounter across the entire
			database.)
		* \c _labels (table): if present, this table is the set of labels that
			have been applied to this revision.
]]--

function WFRxDocStoreImp:docValue( docId, revId, optFieldSpec, simplifyDeletedDoc )

	check( 'WFRxDocStoreImp', 'string|number', '?+string', '?table', '?boolean' )

	assertNotClosed( self, 'docValue' )

	docStoreImpLog:debugf( "docValue %s revId = %s", docId, revId )

	assertDocIdIsValid( docId )

	local db = self._db

	-- Wrap this in a transaction to ensure no other thread / process changes
	-- values between the various queries that are performed.

	local docArchive = db:withTransactionDo( function()

		local args = {
			addDeletionMarker = true,
			simplifyDeletedDocs = simplifyDeletedDoc
		}

		return docValue_inTransaction( self, db, self._exec, docId, revId, args )

	end )

	return Observable.just( docArchive )

end

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

local function bulkDocValue_mixedIds_inTransaction( self, docIds, description )

	local docArchives = {}

	local nDocs = 0

	local includeInitialChangeCounter = docIds.includeInitialChangeCounter
	local simplifyDeletedDocs = docIds.simplifyDeletedDocs
	local label = docIds.label
	if label ~= nil then
		assert( type( label ) == 'string' and #label > 0, "label, if present, must be a non-empty string" )
	end

	for i = 1, #docIds do

		local args = {
			ignoreNonMatch = true,
			addDeletionMarker = true,
			includeInitialChangeCounter = includeInitialChangeCounter,
			simplifyDeletedDocs = simplifyDeletedDocs
		}

		local docArchive = docValue_inTransaction( self, self._db, self._exec, docIds[ i ], label, args )

		if docArchive then
			nDocs = nDocs + 1
			docArchives[ nDocs ] = docArchive
		end

	end

	return docArchives

end

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

local function bulkDocValueImp_mixedIds( self, docIds, description )

	check( 'WFRxDocStoreImp', 'table', '?+string' )

	addToBulkRequestInfo( self, 'bulkDocValue_old', description, #docIds )

	assertNotClosed( self, 'bulkDocValue_old' )

	docStoreImpLog:debugf( "bulkDocValue docIds = %s", docIds )

	for i = 1, #docIds do
		assertDocIdIsValid( docIds[ i ] )
	end

	local db = self._db

	-- Wrap this in a transaction to ensure no other thread / process changes
	-- values between the various queries that are performed.

	return db:withTransactionDo( bulkDocValue_mixedIds_inTransaction, self, docIds, description )

end

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

local function bulkDocValue_inTransaction( self, docIds, description, skipDocIdsValidation )

	local db = self._db

	local includeInitialChangeCounter = docIds.includeInitialChangeCounter
	docIds.includeInitialChangeCounter = nil

	local simplifyDeletedDocs = true

	if docIds.simplifyDeletedDocs ~= nil then
		simplifyDeletedDocs = docIds.simplifyDeletedDocs
	end

	docIds.simplifyDeletedDocs = nil

	if not #docIds then
		return {}
	end

	local useLocalDocId = type( docIds[ 1 ] ) == 'number'

	local rawDocRows

	-- this is just a spot check. If the data is inconsistent, then the CARRAY etension will complain
	if useLocalDocId then

		if not skipDocIdsValidation then

			for _, docId in ipairs( docIds ) do

				if type( docId ) == 'string' then
					-- fall back to the old implementation
					docStoreImpLog:warnf( "bulkDocValue (%s): mixing localIds with fullDocIds is slower", description )
					return bulkDocValue_mixedIds_inTransaction( self, docIds, description )
				end

				assert( docId > 0 and math_floor( docId ) == docId,
						"local id must be positive integer, was " .. tostring( docId ) )
			end

		end

		rawDocRows = db:gatherArrayOfArray(
			[[SELECT fullDocId, revId, revs.sequence,
					content, revs.deleted, hasConflicts, docs.localDocId, annotation,
					docs.mostRecentRevSequence, spaceId
				FROM docs
					INNER JOIN revs ON docs.winningRevSequence = revs.sequence
				WHERE docs.localDocId in CARRAY(?, ?, 'int32')]],
			docIds,
			#docIds )
	else

		-- fullDocId

		if not skipDocIdsValidation then

			for _, docId in ipairs( docIds ) do

				if type( docId ) == 'number' then
					-- fall back to the old implementation
					docStoreImpLog:warnf( "bulkDocValue (%s): mixing localIds with fullDocIds is slower", description )
					return bulkDocValue_mixedIds_inTransaction( self, docIds, description )
				end

				assert( docId:byte( 1 ) ~= underscoreByte,
					"document ID must not start with _" )

				assert( docId:match( "^[0-9a-zA-Z_.%-]+$" ),
					"document ID must only contain letters, numbers, and _.-" )
			end

		end

		rawDocRows = db:gatherArrayOfArray(
			[[SELECT fullDocId, revId, revs.sequence,
					content, revs.deleted, hasConflicts, docs.localDocId, annotation,
					docs.mostRecentRevSequence, spaceId
				FROM docs
					INNER JOIN revs ON docs.winningRevSequence == revs.sequence
				WHERE fullDocId in CARRAY(?, ?, 'char*')]],
			docIds,
			#docIds )

	end

	local result = {}
	local revSequences = {}
	local resultIndexByRevSequence = {}

	for _, currentRevRow in ipairs( rawDocRows ) do

		local docValue = {
			_id = currentRevRow[ 1 ],
			_localId = currentRevRow[ 7 ],
			_revId = currentRevRow[ 2 ],
			_changeCounter = currentRevRow[ 9 ] or currentRevRow[ 3 ],
			_conflict = currentRevRow[ 6 ] == 1 or nil
		}

		local revSequence = currentRevRow[ 3 ]

		-- deleted

		local deleted = currentRevRow[ 5 ] == 1

		if deleted then

			docValue._empty = true
			docValue._deleted = true
			docValue._newId = docValue._newId

		else

			docValue._spaceId = currentRevRow[ 10 ]
			revSequences[ #revSequences + 1 ] = revSequence

		end

		if includeInitialChangeCounter then

			local firstSequenceRow = db:exec( [[SELECT sequence FROM revs WHERE localDocId == ? ORDER BY sequence ASC LIMIT 1]], docValue._localId )
			docValue._initialChangeCounter = KSTableUtils_derefOpt( firstSequenceRow, 'sequence' )
				or docValue._changeCounter

		end

		if useLocalDocId then

			if deleted and simplifyDeletedDocs then
				result[ docValue._localId ] = { docValue, nil, nil }
			else
				result[ docValue._localId ] = { docValue, currentRevRow[ 4 ], currentRevRow[ 8 ] }
				resultIndexByRevSequence[ revSequence ] = docValue._localId
			end
		else
			if deleted and simplifyDeletedDocs then
				result[ docValue._id ] = { docValue, nil, nil }
			else
				result[ docValue._id ] = { docValue, currentRevRow[ 4 ], currentRevRow[ 8 ] }
				resultIndexByRevSequence[ revSequence ] = docValue._id
			end
		end

	end

	local labelRows = db:gatherArrayOfArray( [[SELECT revSequence, labels.name
												FROM labels
												WHERE revSequence IN CARRAY( ?, ?, 'int32' )]],
											 revSequences,
											 #revSequences )

	for _, row in ipairs( labelRows ) do

		local revSequence = row[ 1 ]
		local labelName  = row[ 2 ]

		local resultIndex = resultIndexByRevSequence[ revSequence ]

		if resultIndex then

			-- result contains docArchives
			local doc = result[ resultIndex ][ 1 ]

			local labels = doc._labels or {}
			labels[ labelName ] = true
			doc._labels = labels

		end

	end

	local orderedResult = {}

	for i, docId in ipairs( docIds ) do

		if not result[ docId ] then

			local docValue = {
				_id = docId,
				_empty = true,
			}

			orderedResult[ i ] = { docValue, nil, nil }

		else
			orderedResult[ i ] = result[ docId ]
		end

	end

	addToBulkRequestInfo( self, 'bulkDocValue', description, #result )

	changedDocsLog:debugf( "bulkDocValue is returning %s", result )

	return orderedResult

end

--------------------------------------------------------------------------------
--[[!
	\brief Gets many document values at once.

	\param docIds (table) array of document IDs
		Optional arguments to this api are passed as named parameters in the
		docIds table they are:
		* \c label (optional, string): Specifies what label-ed revision to
			get for the documents.
		* \c includeInitialChangeCounter (optional, boolean): if true, return the
			first changeCounter (sequence number) related to the documents.
		* \c simplifyDeletedDocs (optional, boolean): Defaults to true. If set to
			false, then deleted docs will be returned in their full form. If true
			or nil, simplified tombstones will be returned in place of full deleted
			documents.

	\param optFieldSpec (optional, table of arrays of strings) paths of fields to include.
		For example, for the purpose of retrieving a thumbnails, this optFieldSpec might be used:
		{
			{ "_id" },
			{ "_revId" },
			{ "_serverProvidedMetadata", "base" },
			{ "_serverProvidedMetadata", "links", "/rels/rendition_type/thumbnail2x", "href" },
		}
	}

	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.

	\param skipDocIdsValidation (optional boolean) If true, the array of docIds is not
		validated. This should only be used in select circumstances where performance
		is critical, and docIds is already known to be valid.

	\return <code>Observable< table ></code> resolves to an array of document
		values when done. The items in the array will be the document values
		in the same sequence as requested on input. Each item will contain the
		same results as would be returned for an equivalent \c :docValue call.

	It is not possible to request specific revision IDs via this API.
]]--

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

function WFRxDocStoreImp:bulkDocValue( docIds, optFieldSpec, description, skipDocIdsValidation )

	if docIds.label then
		return Observable.just( bulkDocValueImp_mixedIds( self, docIds, description ) )
	end

	addToBulkRequestInfo( self, 'bulkDocValue', description, #docIds )

	assertNotClosed( self, 'bulkDocValue' )

	docStoreImpLog:debugf( "bulkDocValue docIds = %s", docIds )

	local result = self._db:withTransactionDo(
			bulkDocValue_inTransaction, self, docIds, description, skipDocIdsValidation )

	if self._TEST_unpackDocValue then
 		local unpacked = {}

		unpacked.continueFrom = result.continueFrom

		for i, docArchive in ipairs( result ) do
			unpacked[ i ] = unpackDocValue(  docArchive, optFieldSpec )
		end

		return Observable.just( unpacked )
	end

	return Observable.just( result )

end

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

makeOngoingDocValueForDocId = function( self, docId )

	local result = ReplaySubject( 1 )

	local db = self._db

	local docArchive = db:withTransactionDo( function()

		return docValue_inTransaction( self, db, self._exec, docId, nil, { unpackDocValue = true } )

	end )

	result:onNext( docArchive )

	return result

end

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

function WFRxDocStoreImp:ongoingDocValue( docId )

	check( 'WFRxDocStoreImp', '+string|number' )

	assertNotClosed( self, 'ongoingDocValue' )

	docStoreImpLog:debugf( "ongoingDocValue %s", docId )

	assertDocIdIsValid( docId )

	if type( docId ) == 'number' then

		local localId = docId
		local db = self._db

		docId = db:withTransactionDo( function( myDb )
			local localIdToDocIdMap = bulkLocalIdToDocId_inTransaction( self, myDb, { localId }, true )
			return localIdToDocIdMap[ localId ]
		end, db )

	end

	return self._ongoingDocValues:observableForKey( docId )

end

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

makeOngoingLocalDocValueForDocName = function( self, name )

	local result = ReplaySubject( 1 )

	local db = self._db

	local localDocValueRow = self._exec( db,
										 [[SELECT content FROM localDocs WHERE name == ?]],
										 name )

	local docValue = localDocValueRow and contentFromFlattenedValue( localDocValueRow.content )

	if docValue then
		docValue._id = name
	end

	result:onNext( docValue or { _id = name, _empty = true } )

	return result

end

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

function WFRxDocStoreImp:ongoingLocalDocValue( name )

	check( 'WFRxDocStoreImp', 'string' )

	assertNotClosed( self, 'ongoingLocalDocValue' )

	docStoreImpLog:debugf( "ongoingLocalDocValue %s", name )

	return self._ongoingLocalDocValues:observableForKey( name )

end

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

local function extractDocValue( docValue )

	local annotationContent = {}
	local bodyContent = {}

	for k, v in pairs( docValue ) do
		if string_byte( k, 1 ) == underscoreByte
			and k ~= '_deleted'
			and k ~= '_newId' then
			annotationContent[ k ] = v
		else
			bodyContent[ k ] = v
		end
	end

	return bodyContent, annotationContent

end

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

local function validateAndExtractDocValue( _, docValue )

	-- Verify that doc value is acceptable (does not contain illegal keys, etc.).

	assertDocValueIsValid( docValue )

	return extractDocValue( docValue )

end

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

local function validateAndExtractOverlay( _, overlay )

	assertDocOverlayIsValid( overlay )

	return extractDocValue( overlay )

end

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

local function validateAndArchiveDocValue( _, _, docValue, annotation )

	-- Extract doc ID from the value.

	local docId = assertDocIdIsValid( annotation._id )
	annotation._id = nil
	annotation._localId = nil

	-- Extract the new revId from the value

	local newRevId = annotation._newRevId
	annotation._newRevId = nil

	if newRevId ~= nil and type( newRevId ) ~= 'string' then
		error( "_newRevId must be a string if present" )
	end

	-- Extract parent rev ID from the value.

	local parentRevId = annotation._revId

	if parentRevId ~= nil and type( parentRevId ) ~= 'string' then
		error( "_revId must be a string if present" )
	end

	annotation._revId = nil

	-- Extract the spaceId from the value

	local spaceId = annotation._spaceId and assertSpaceIdIsValid( annotation._spaceId )
	annotation._spaceId = nil

	if spaceId ~= nil and type( spaceId ) ~= 'string' then
		error( "_spaceId must be a string if present" )
	end

	-- Or extract parent rev IDs array.

	local parentRevIds = annotation._revIds

	if parentRevIds ~= nil
		and ( type( parentRevIds ) ~= 'table' or #parentRevIds == 0 )
	then
		error( "_revIds must be a non-empty array if present" )
	end

	annotation._revIds = nil

	-- Can't have both.

	if parentRevId and parentRevIds then
		error( "Must not specify both _revId and _revIds" )
	end

	-- Extract labels from the value.

	local labels = annotation._labels
	annotation._labels = nil

	--Extract type and subtype from docValue

	local docType = docValue.type
	local docSubtype = docValue.subtype

	-- Convert Lua to text.

	local bodyContentArchive = tableToString( docValue )
	local annotationContentArchive = tableToString( annotation )

	return
		{
			docId = docId,
			parentRevId = parentRevId,
			parentRevIds = parentRevIds,
			newRevId = newRevId,
			bodyContentArchive = bodyContentArchive,
			annotationContentArchive = annotationContentArchive,
			docValue = docValue,
			docValueIsDeleted = docValue._deleted == true,
			annotation = annotation,
			labels = labels,
			spaceId = spaceId,
			docType = docType,
			docSubtype = docSubtype,
		}

end

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

local function extractAndValidateDocValue( self, funcName, docValue )

	local bodyContent, annotationContent = validateAndExtractDocValue( self, docValue )

	return validateAndArchiveDocValue( self, funcName, bodyContent, annotationContent )

end

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

local function validateDocValue_inTransaction() -- self, db )

	-- TODO [scouten 2014-08-19]: Implement validation.
	-- For now, everything validates.

end

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

local function setDocValueImp_inTransaction( self, db, exec, _, args )

	local docId = args.docId
	local parentRevId = args.parentRevId
	local parentRevIds = args.parentRevIds
	local bodyContentArchive = args.bodyContentArchive
	local annotationContentArchive = args.annotationContentArchive
	local annotation = args.annotation
	local docValue = args.docValue
	local annotationAlreadyUpdated = args.annotationAlreadyUpdated
	local docValueIsDeleted = args.docValueIsDeleted
	local spaceId = args.spaceId
	local docType = args.docType
	local docSubtype = args.docSubtype

	if docValue and docValue._deleted and docValue._newId then
		-- if we're deleting and setting a _newId then the _newId document
		-- needs to already exist as we do not create it as part of setDocValue
		-- (which will always pass docValue in the args) as we would in applyOverlay
		local newIdDocRow = exec( db, [[SELECT fullDocId FROM docs WHERE fullDocId = ?]], docValue._newId )
		if not newIdDocRow then
			errorf( "Cannot call setDocValue with _newId = %s if the _newId doc does not exist", docValue._newId )
		end
	end

	local labels = args.labels

	local swapIds = type( docId ) == 'number'
		-- after querying, swap localId and docId
	local whereIdCol = swapIds and 'docs.localDocId' or 'fullDocId'
	local selectIdCol = swapIds and 'fullDocId' or 'docs.localDocId'

	-- Get the doc's local numeric ID and its current winning revision.

	local docIdRow = exec( db,
						   [[SELECT ]] .. selectIdCol .. [[ AS selectId, winningRevSequence,
						   		deleted, annotation
								FROM docs WHERE ]] .. whereIdCol .. [[ = ?]],
						   docId )

		-- EXPLAIN QUERY PLAN says this uses auto index on docs table.

	local localDocId
	if swapIds then
		localDocId = docId
		docId = docIdRow and docIdRow.selectId
	else
		localDocId = docIdRow and docIdRow.selectId
	end

	if not docId then
		errorf( "Cannot create a new document with setDocValue and localId (%d).", localDocId )
	end

	local oldWinningRevSequence = docIdRow and docIdRow.winningRevSequence
	local oldRevIsDeleted = docIdRow and ( docIdRow.deleted == 1 )
	local oldDocExisted = not not docIdRow

	-- Special case: If no _localOnly value is provided, then carry over any
	-- previous value.

	if not annotationAlreadyUpdated and not( annotation and annotation._localOnly ) then

		local priorAnnotationArchive = docIdRow and docIdRow.annotation

		local priorAnnotation = priorAnnotationArchive
			and contentFromFlattenedValue( priorAnnotationArchive )
			or {}

		local priorLocalOnly = priorAnnotation._localOnly

		if priorLocalOnly then
			annotation._localOnly = priorLocalOnly
			annotationContentArchive = tableToString( annotation )
		end

	end

	-- Special case: If the old rev is deleted, we may need to carry across a _newId value
	if oldRevIsDeleted and docValueIsDeleted then
		local revIdRow = exec( db, [[SELECT content FROM revs WHERE localDocId = ? AND sequence = ?]], localDocId, oldWinningRevSequence )
		local content = contentFromFlattenedValue( revIdRow.content )
		if content._newId then
			local newBodyContent = contentFromFlattenedValue( bodyContentArchive )
			newBodyContent._newId = content._newId
			bodyContentArchive = tableToString( newBodyContent )
		end
	end

	-- Identify parent revision.

	local parentSequence, parentSequenceRevId

	if parentRevIds and localDocId then

		-- We think we're replacing some existing revision. In this case, we allow
		-- the revision to be non-current, but we always lose if a conflict would be
		-- created.

		for i = #parentRevIds, 1, -1 do

			parentSequence, parentSequenceRevId = getSequenceForDocumentAndRevision_inTransaction( db, exec,
											localDocId, parentRevIds[ i ], false )

			if parentSequence then
				break
			end

		end

		if oldRevIsDeleted and not parentSequence then

			-- If current rev is a deletion, then we'll ignore the fact that
			-- we didn't get a _revId parameter and plunk it in.

			local revIdRow = exec( db, [[SELECT revId FROM revs WHERE sequence = ?]],
								   oldWinningRevSequence )

			parentRevId = revIdRow and revIdRow.revId
			parentSequence = oldWinningRevSequence

		end

		-- The search for a parent sequence may fail to come up with the right parentSequence
		-- if Oz compressed away revisions. If we find a revision in the docStore with a higher
		-- sequence number that bears the same label as this new doc, use that sequence number
		-- as parentSequence. If the feed sets more then one label, use the highest sequence number
		-- for any of the labels.

		if not parentRevIds.force then
			-- Optimize away the common case, that is that there are no local edits.
			if labels and (not parentSequence or parentSequence ~= oldWinningRevSequence) then
				for label in pairs( labels ) do
					local  revSequenceRow = exec( db, [[SELECT revSequence FROM labels
													WHERE localDocId = ? AND name = ?
													ORDER BY revSequence DESC LIMIT 1]],
												localDocId, label )
					-- EXPLAIN QUERY PLAN says this uses auto index on labels table (created by UNIQUE constraint)

					local labeledSequence = revSequenceRow and revSequenceRow.revSequence

					if not parentSequence or (labeledSequence and labeledSequence > parentSequence) then
						parentSequence = labeledSequence
					end
				end
			end
		end

		-- Validate doc value.

		validateDocValue_inTransaction( self, db )

	elseif parentRevId then

		-- We're replacing an existing revision. Make sure it's current
		-- and find its sequence number.

		-- TODO [scouten 2014-08-25]: Optimize for the common case where we
		-- already got the parent sequence ID from previous query.

		parentSequence, parentSequenceRevId = getSequenceForDocumentAndRevision_inTransaction( db, exec,
											localDocId, parentRevId, true )

		if not parentSequence then

			if documentRevisionExists_inTransaction( db, exec, localDocId, parentRevId ) then

				errorf( "_rev %s is not current (would create conflict)", parentRevId )

			else

				errorf( "can not find _rev %s", parentRevId )

			end

		end

		-- Validate doc value.

		validateDocValue_inTransaction( self, db )

	else

		-- Validate doc value.

		validateDocValue_inTransaction( self, db )

		-- Does this doc exist?

		if localDocId then

			if oldRevIsDeleted then

				-- If current rev is a deletion, then we'll ignore the fact that
				-- we didn't get a _revId parameter and plunk it in.

				local revIdRow = exec( db, [[SELECT revId FROM revs WHERE sequence = ?]],
									   oldWinningRevSequence )

				parentRevId = revIdRow and revIdRow.revId
				parentSequence = oldWinningRevSequence

			elseif oldWinningRevSequence then

				parentSequence = oldWinningRevSequence

			end

		else

			-- Doc didn't previously exist: Create a local ID for it now.

			localDocId = createLocalDocId_inTransaction( db, exec, docId )

		end

	end

	local parentRevIsWinning = ( parentRevId or parentRevIds ) and ( parentSequence == oldWinningRevSequence )

	-- Generate new rev ID.

	local newRevId

	if parentRevIds then

		newRevId = parentRevIds[ #parentRevIds ]

		-- If we're force-deleting (i.e. deletion from Oz without revision_ids provided),
		-- graft this deletion revision on top of the previous parent.

		if docValueIsDeleted and not parentSequence and parentRevIds.force then
			parentSequence = oldWinningRevSequence
			parentSequenceRevId = nil
			parentRevIsWinning = true
		end

	else

		-- The change is originating locally (i.e. via typical :setDocValue API).

		-- Special case: If we're not changing the body content of the document, don't
		-- change the revision ID unless we've been specifically provided with a new
		-- revision ID to use

		local suppressNewRevId = false

		if args.newRevId then
			suppressNewRevId = false
		elseif args.didContentChange ~= nil then

			-- We already know whether or not the content changed.

			suppressNewRevId = not args.didContentChange

		elseif parentSequence and parentRevIsWinning and not oldRevIsDeleted
		  or ( parentSequence == oldWinningRevSequence )then

			local existingParentRow = exec( db,
							[[SELECT content FROM revs WHERE sequence == ?]],
							parentSequence )

			local convertToArchiveAndCompare = function( strToConvert, strToCompare )

				local contentTbl = contentFromFlattenedValue( strToConvert )
				local contentStr = WFTableArchiving_tableToString( contentTbl )

				return contentStr == strToCompare

			end

			if existingParentRow then

				local isExistingContentMsgPacked = isMsgPacked( existingParentRow.content )
				local isBodyContentMsgPacked = isMsgPacked( bodyContentArchive )

				if not isExistingContentMsgPacked and not isBodyContentMsgPacked then

					suppressNewRevId = bodyContentArchive == existingParentRow.content
						-- Tables serialized with WFTableArchiving can be compared with '=='

				elseif isExistingContentMsgPacked and isBodyContentMsgPacked then

					-- Because MsgPack does not guarantee a strict ordering of table keys, we
					-- must deserialize these values to compare them. This is expensive, but it
					-- rarely happens.

					docStoreImpLog:warnf( [[setDocValueImp_inTransaction must now compare two MsgPacked
							doc values using KSTableUtils.deepEquals.]] )

					local existingContentTbl = contentFromFlattenedValue( existingParentRow.content )
					local bodyContentArchiveTbl = contentFromFlattenedValue( bodyContentArchive )

					suppressNewRevId = KSTableUtils.deepEquals( existingContentTbl, bodyContentArchiveTbl )


				elseif isExistingContentMsgPacked then

					-- existingParentRow.content was serialized with MsgPack, but bodyContentArchive
					-- was not. To compare these values, convert the existingParentRow.content encoding.

					suppressNewRevId = convertToArchiveAndCompare( existingParentRow.content, bodyContentArchive )

				else

					-- bodyContentArchive was serialized with MsgPack, but existingParentRow.content
					-- was not. To compare these values, convert the bodyContentArchive encoding.

					suppressNewRevId = convertToArchiveAndCompare( bodyContentArchive, existingParentRow.content )

				end

			end

		end

		if oldDocExisted and parentRevIsWinning == nil and oldRevIsDeleted == false
		  and not suppressNewRevId and not labels then

			docStoreImpLog:errorf( "Creating new revision without specifying a parent ID: %s",
								   contentFromFlattenedValue( bodyContentArchive ) )

			errorf( "%s, can not create new revision (would create conflict)", ERROR_REVISION_WITHOUT_PARENT )

		end

		if suppressNewRevId and parentRevId then
			newRevId = parentRevId
		else
			newRevId = args.newRevId or KSIdGenerator.uuidInLowercaseHex()
		end

	end

	-- We need to know if parent rev is current. If not, we're creating a conflict.

	local parentRevWasCurrent = parentRevIsWinning

	if not parentRevIsWinning and parentSequence then
		-- A winning rev is inherently current. In this (common) case, we should
		-- optimize away the following query.

		local parentRevIsCurrentRow = exec( db,
							[[SELECT current FROM revs WHERE sequence == ?]],
							parentSequence )

		if parentRevIsCurrentRow.current == 1 then
			parentRevWasCurrent = true
		end

	end

	-- Actually insert the new revision.
	exec( db,
		  [[INSERT INTO revs( localDocId, revId, parentRevSequence, current, deleted, content )
				VALUES( ?, ?, ?, ?, ?, ? )]],
		  localDocId,
		  newRevId,
		  parentSequence,
		  args.markNonCurrent == true and 0 or 1,
		  docValueIsDeleted and 1 or 0,
		  bodyContentArchive )

	-- Get new change counter, but don't update yet.

	local newRevSequence = self._db:getLastInsertRowId()

	-- If we replaced a revision, make it no longer current.
	if parentSequence then
		exec( db, [[UPDATE revs SET current = 0 WHERE sequence = ?]], parentSequence )
	end

	-- if the revision id hasn't changed, moved labels to the new sequence and
	-- remove the old content to save space
	if parentSequenceRevId == newRevId then

		exec( db,
			[[UPDATE revs SET content = NULL WHERE
				current == 0 AND sequence = ?]],
			parentSequence )

		exec( db,
			[[UPDATE labels SET revSequence = ? WHERE revSequence = ?]],
			newRevSequence,
			parentSequence )

	end

	-- Update document summary information.

	local newWinningRevId, newWinningRevSequence, newRevIsDeleted, newRevIsInConflict

	if oldDocExisted then

		local favorRevSequence
		if parentRevIsWinning and parentSequence == oldWinningRevSequence then
			favorRevSequence = newRevSequence
		elseif parentRevIds then
			if parentRevIds.force then
				favorRevSequence = newRevSequence
			else
				favorRevSequence = oldWinningRevSequence
			end
		end

		newWinningRevId, newWinningRevSequence, newRevIsDeleted, newRevIsInConflict
			= winningRevIdForLocalDocId_inTransaction( self, localDocId, favorRevSequence )

		-- Any derivation from a non-head revision is a conflict, even if we didn't
		-- detect it above. (Reasoning: We need to know about and preserved a deletion
		-- from Oz that conflicts with a locally-originated edit. But we also rely on
		-- deleting a current revision as the means of resolving a conflict, so a rule
		-- that simply says creating a non-winning revision flags a conflict is
		-- incorrect.)

		if not parentRevWasCurrent then
			newRevIsInConflict = true
		end

	else

		newWinningRevId = newRevId
		newWinningRevSequence = newRevSequence
		newRevIsDeleted = docValueIsDeleted
		newRevIsInConflict = false

	end

	if newRevIsInConflict then
		docStoreImpLog:debugf( "Creating a conflicting revision for %s %s", docId, contentFromFlattenedValue( bodyContentArchive ) )
	end

	local function docUpdateSqlAndArgs()

		local sql = "UPDATE docs SET winningRevSequence = ?, mostRecentRevSequence = ?, deleted = ?, hasConflicts = ?, annotation = ?"
		local updateArgs =
		{
			newWinningRevSequence,
			newRevSequence,
			newRevIsDeleted and 1 or 0,
			newRevIsInConflict and 1 or 0,
			annotationContentArchive,
		}

		if spaceId then
			sql = sql .. ", spaceId = ?"
			updateArgs[#updateArgs+1] = spaceId
		end

		if docType then
			sql = sql .. ", type = ?"
			updateArgs[#updateArgs+1] = docType
		end

		if docSubtype then
			sql = sql .. ", subtype = ?"
			updateArgs[#updateArgs+1] = docSubtype
		end

		sql = sql .. " WHERE localDocId = ?"
		updateArgs[#updateArgs+1] = localDocId

		return sql, updateArgs

	end

	local updateSql, updateArgs = docUpdateSqlAndArgs()

	exec( db, updateSql, unpack( updateArgs ) )
	-- EXPLAIN QUERY PLAN says this uses integer primary key on docs.

	-- Add/remove ids from cache here
	updateIdCaches( self, localDocId, docId, newRevIsDeleted )

	-- Update labels.

	if args.preserveLabels then
		local newLabels = labels
		labels = getLabels( db, oldWinningRevSequence )

		-- If newLabels were provided while using preserveLabels
		-- add them onto the list of labels.
		if newLabels then
			for label, _ in pairs( newLabels ) do
				labels[ label ] = true
			end
		end

	end

	if labels then

		for label in pairs( labels ) do
			exec( db, [[INSERT OR REPLACE INTO labels VALUES( ?, ?, ? )]],
				  localDocId, newRevSequence, label )
		end

	end

	--[[ Clean up revisions that are:
		 - not in conflict
		 - older than any labelled revision not starting with an underscore

		 This implies that unlabelled revisions are never removed.

		 We only need to perform this check when the new revision has a label.
	]]--

	if not newRevIsInConflict and labels and
			not WFCapabilities.getCapability( 'WFRxDocStore.disableRevisionCleanupForCompact' ) then

		local hasLabelsAffectingCleanup = false

		for label, _ in pairs( labels ) do
			if string.sub( label, 1, 1 ) ~= '_' then
				hasLabelsAffectingCleanup = true
			end
		end

		if hasLabelsAffectingCleanup then

			-- we need to defer foreign keys
			exec( db, [[PRAGMA defer_foreign_keys = ON]] )

			local minSequenceRow = exec( db,
					[[SELECT MIN(labels.revSequence) FROM labels
						JOIN docs ON labels.localDocId = docs.localDocId
						WHERE labels.revSequence <= docs.winningRevSequence
							AND substr(labels.name, 1, 1) != '_'
							AND docs.localDocId = ?]], localDocId )

			if next( minSequenceRow ) then

				local minSequence = minSequenceRow[ 'MIN(labels.revSequence)' ]

				if newWinningRevSequence < minSequence then
					minSequence = newWinningRevSequence
				end


				exec( db, [[UPDATE revs SET parentRevSequence = NULL WHERE revs.sequence = ?]], minSequence )
				exec( db, [[DELETE FROM revs WHERE sequence < ? and localDocId = ?]], minSequence, localDocId )

			end

		end

	end

	if not newRevIsInConflict and newRevIsDeleted and spaceId then

		--[[
			If you are *not* the owner of document in a space and that document is deleted,
			then we need to clean up *everything* related to the document.
			For legal reasons, we must remove all history of the asset. This includes purging
			the content of any revisions associated with the document.
		]]--

		 exec( db, [[UPDATE revs SET content = ? WHERE sequence != ? AND localDocId = ? AND current = 0]], emptyArchive, newRevSequence, localDocId )

	end

	local buildFinalDocValue = function( )

		local body = args.body or ( bodyContentArchive and contentFromFlattenedValue( bodyContentArchive ) or {} )
		local annotation = args.annotation or ( annotationContentArchive and contentFromFlattenedValue( annotationContentArchive ) or {} )

		local dv = {}

		for k, v in pairs( body ) do
			dv[ k ] = v
		end

		for k, v in pairs( annotation ) do
			dv[ k ] = v
		end

		dv._localId = localDocId
		dv._id = docId
		dv._revId = newRevId
		dv._changeCounter = newRevSequence
		dv._spaceId = spaceId

		if newRevIsInConflict then
			dv._conflict = true
		end

		return dv

	end

	-- If the new revision is deleted, signal the deletion via the deleted doc stream

	local finalDocValue = nil

	if newRevIsDeleted and not oldRevIsDeleted then

		finalDocValue = buildFinalDocValue()
		self._deletedDocStream:onNext( finalDocValue, newRevSequence )

	else

		self._deletedDocStream:onNext( nil, newRevSequence )

	end

	-- Do we need to update an observable doc value?

	local ongoingDocValues = self._ongoingDocValues

	if ongoingDocValues:hasObservableForKey( docId ) then

		local ongoingDocValue = ongoingDocValues:observableForKey( docId )

		if newWinningRevId == newRevId then

			if not finalDocValue then
 				finalDocValue = buildFinalDocValue()
				finalDocValue._labels = labels
 			end

			if finalDocValue._deleted then
				local deletedDocValue = { _id = docId, _empty = true, _labels = labels }
				if finalDocValue._newId then
					deletedDocValue._newId = finalDocValue._newId
					deletedDocValue._deleted = true
				end
				ongoingDocValue:onNext( deletedDocValue )
			else
				ongoingDocValue:onNext( finalDocValue )
			end

		else

			local docValueNow = docValue_inTransaction( self, db, self._exec, docId, nil, { unpackDocValue = true } )

			ongoingDocValue:onNext( docValueNow )

		end

	end

	-- Return rev ID that we assigned and also the new change counter,
	-- which must be updated outside of the transaction.

	return newRevId, newRevSequence, localDocId

end


--------------------------------------------------------------------------------
--[[!
	\brief Sets the document value, replacing any existing value.

	\param value (table) new value for the entire document

	\c value \b must contain the following elements:
		* \c _id (string) the document's unique ID. If the document already
			exists and you know the _localId, you may supply an integer
			local ID instead of the unique ID.
		* \c _revId (string) the document's current revision ID
			(unless the document is new, in which case it may be omitted)
		* \c _revIds (array of strings) intended for use in sync; if provided:
			* the array must be non-empty
			* the new revision's ID will be the last rev ID in the array
			* the new revision's ID will be the last member of this array
			* the new revision's parent ID will be the \b LAST rev ID in the
				array that matches an existing ID
			* any conflict that might arise will be resolved by making the
				new revision the \b losing revision (separate code will resolve
				the conflict)
			* in this case, \c _revId must \b not be specified
		* \c _labels (table) revision labels to apply to this revision
			(Any existing labels named here will be moved to this revision.)

	\c value \b may contain the following elements:
		* \c _deleted (boolean): if present, must be \c true and no other
			values may be present

	\b Important: The top-level key \c _localOnly gets special handling. If
		no value is provided for this key, any previous value is left in place.
		This behavior does not apply to any other key. This does imply that you
		can not delete any existing \c _localOnly entry in the document.

	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.

	\return <code>Observable< string ></code> resolves with the new value's
		revision ID when the write completes. You may safely ignore this Observable.
]]--

function WFRxDocStoreImp:setDocValue( docValue, description )

	check( 'WFRxDocStoreImp', 'table', '?string' )

	assertNotClosed( self, 'setDocValue' )

	addToBulkRequestInfo( self, 'setDocValue', description, 1 )

	docStoreImpLog:debugf( "setDocValue %s", docValue )

	local args = extractAndValidateDocValue( self, 'setDocValue', docValue )

	local db = self._db

	local revId, newRevSequence = db:withTransactionDo( function()

		return setDocValueImp_inTransaction( self, db, self._exec,
											 'setDocValue', args )

	end )

	if newRevSequence then
		self._changeCounter:onNext( newRevSequence )
	end

	return Observable.just( revId )

end

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

local function setLocalDocValueImp_inTransaction( self, db, exec, _, docValue )

	local localDocId = assertDocIdIsValid( docValue._id )
	docValue._id = nil

	assertDocValueIsValid( docValue )

	if docValue._revId then
		error( "WFRxDocStoreImp:setLocalDocValue does not accept _revId", 2 )
	end

	if docValue._deleted then
		error( "WFRxDocStoreImp:setLocalDocValue does not accept _deleted", 2 )
	end

	local serializedDocValue = tableToString( docValue )

	self._exec( self._db,
				[[INSERT OR REPLACE INTO localDocs VALUES( ?, ? )]],
				localDocId,
				serializedDocValue )

	-- Do we need to update an observable doc value

	local ongoingLocalDocValues = self._ongoingLocalDocValues

	if ongoingLocalDocValues:hasObservableForKey( localDocId ) then

		local ongoingLocalDocValue = ongoingLocalDocValues:observableForKey( localDocId )

		if docValue then
			docValue._id = localDocId
		end

		ongoingLocalDocValue:onNext( docValue )

	end

end

--------------------------------------------------------------------------------
--[[!
	\brief Sets many document values at once, replacing any existing values.

	\param values (table) array of entire document values
	\param localDocValues (?table) optional array of local document values

	The array may contain zero or more document values. Each value entry
	looks like the \c value parameter to \c :setDocValue.

	The array may also contain a <code>label = "name"</code> entry. If this exists,
	that label will be applied to all documents in this batch.

	If an error occurs while setting any one of these document values, then the
	entire set of document values will be discarded. In other words, this set
	of documents succeeds or fails atomically.

	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.

	\return <code>Observable< table ></code> resolves to a single value containing
		an array of the following tables with one entry for each of the supplied doc
		values in order:
			id - the supplied doc _id.
			ok - success or failure, currently always true.
			revId - the revId associated with the document.
			localId - the integer based local doc id that was just allocated.
		You may safely ignore this result.
]]--

function WFRxDocStoreImp:bulkSetDocValue( docValues, localDocValues, description )

	check( 'WFRxDocStoreImp', 'table', '?table', '?string' )

	assertNotClosed( self, 'bulkSetDocValue' )

	addToBulkRequestInfo( self, 'bulkSetDocValue', description, #docValues )

	docStoreImpLog:debugf( "bulkSetDocValue %s", docValues )

	local nDocValues = #docValues

	if nDocValues == 0 then
		error( "WFRxDocStoreImp:bulkSetDocValue: no values in this array", 2 )
	end

	if nDocValues > 100 then
		docStoreImpLog:warnf( string_format( "WFRxDocStoreImp:bulkSetDocValue: "
											 .. "%d values provided in this array, "
											 .. "exceeds recommended maximum (100)",
							  nDocValues ), 2 )
	end

	local label = docValues.label

	if label and ( type( label ) ~= 'string' or #label == 0 ) then
		error( "WFRxDocStoreImp:bulkSetDocValue: label must be a non-empty string if provided", 2 )
	end

	local db = self._db

	local revResults, finalRevSequence = db:withTransactionDo( function()

		local results = {}
		local latestRevSequence

		for i = 1, nDocValues do

			local args = extractAndValidateDocValue( self, 'bulkSetDocValue', docValues[ i ] )

			if label then
				args.labels = args.labels or {}
				args.labels[ label ] = true
			end

			local revId, newRevSequence, localId = setDocValueImp_inTransaction(
								self, db, self._exec, 'bulkSetDocValue', args )

			if newRevSequence then
				latestRevSequence = newRevSequence
			end

			results[ i ] = {
				id = args.docId,
				ok = true,
				revId = revId,
				localId = localId,
			}

		end

		local nLocalDocValues = localDocValues and #localDocValues or 0
		for i = 1, nLocalDocValues do
			setLocalDocValueImp_inTransaction( self, db, self._exec, 'bulkSetDocValue', localDocValues[ i ] )
		end

		return results, latestRevSequence

	end )

	if finalRevSequence then
		self._changeCounter:onNext( finalRevSequence )
	end

	return Observable.just( revResults )

end

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

local function currentRevisionsForDoc_inTransaction( _, db, _, docId, includeDeleted )

	local revisionRows

	if includeDeleted then

		revisionRows = db:gatherArrayOfArray(
			[[SELECT revId, sequence, content, revs.deleted, hasConflicts, annotation, docs.localDocId, spaceId
				FROM revs
					INNER JOIN docs ON docs.localDocId = revs.localDocId
				WHERE
					fullDocId == ? AND current == 1
				ORDER BY ( sequence == docs.winningRevSequence ) DESC,
						 revs.deleted ASC, revId DESC]],
				docId )

	else

		revisionRows = db:gatherArrayOfArray(
			[[SELECT revId, sequence, content, revs.deleted, hasConflicts, annotation, docs.localDocId, spaceId
				FROM revs
					INNER JOIN docs ON docs.localDocId = revs.localDocId
				WHERE
					fullDocId == ? AND current == 1 AND revs.deleted == 0
				ORDER BY ( sequence == docs.winningRevSequence ) DESC,
						 revId DESC]],
				docId )

	end

	return revisionRows

end

--------------------------------------------------------------------------------
--[[!
	\brief Returns all current revisions of a document.

	\param docId (string) the document's unique ID
	\param includeDeleted (boolean) true to include revisions that are current
		but have been deleted

	\return <code>Observable< table ></code> array of revisions

	Each revision in the array will look like the response from \c :docValue.

	If there are no conflicts (no resolved conflicts, in the case of \c includeDeleted),
	then this function will look like the single response from \c :docValue.

	The winning revision will always be first in the array.
]]--

function WFRxDocStoreImp:currentRevisionsForDoc( docId, includeDeleted )

	check( 'WFRxDocStoreImp', 'string', '?boolean' )

	assertNotClosed( self, 'currentRevisionsForDoc' )

	docStoreImpLog:debugf( "currentRevisionsForDoc %s %s", docId, includeDeleted )

	assertDocIdIsValid( docId )

	local db = self._db

	local revisionRows = db:withTransactionDo( function()

		return currentRevisionsForDoc_inTransaction( self, db, self._exec, docId, includeDeleted )

	end )

	local result = {}

	for i = 1, #revisionRows do

		local revisionRow = revisionRows[ i ]

		local docValue = contentFromFlattenedValue( revisionRow[ 3 ] )
		local annotation = contentFromFlattenedValue( revisionRow[ 6 ] )

		if annotation then
			for k, v in pairs( annotation ) do
				docValue[ k ] = v
			end
		end

		docValue._id = docId
		docValue._localId = revisionRow[ 7 ]
		docValue._revId = revisionRow[ 1 ]
		docValue._changeCounter = revisionRow[ 2 ]
		docValue._spaceId = revisionRow[ 8 ]

		if revisionRow[ 5 ] == 1 then
			docValue._conflict = true
		end

		if revisionRow[ 4 ] == 1 then
			docValue._deleted = true
		end

		result[ i ] = docValue

	end

	return Observable.just( result )

end

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

local function deleteDoc_imp( self, docId, optRevId, markNonCurrent )

	-- Extract doc ID from the value.

	assertDocIdIsValid( docId )

	local db = self._db

	local result, newRevSequence = db:withTransactionDo( function()

		local exec = self._exec

		local docValue = docValue_inTransaction( self, db, exec, docId, optRevId, { unpackDocValue = true } )

		if docValue._empty then
			return true
		end

		if docValue._deleted then
			return docValue._revId
		end

		docValue._deleted = true
		docValue._conflict = nil
		docValue._changeCounter = nil
		docValue._labels = nil

		local args = extractAndValidateDocValue( self, 'deleteDoc', docValue )

		args.markNonCurrent = markNonCurrent
		args.didContentChange = true

		return setDocValueImp_inTransaction( self, db, exec,
											 'deleteDoc', args )

	end )

	if newRevSequence then
		self._changeCounter:onNext( newRevSequence )
	end

	return Observable.just( result )

end

--------------------------------------------------------------------------------
--[[!
	\brief Deletes a document or a specific revision of a document.

	\param docId (string|number) the document's unique or local ID
	\param revId (optional, string) revision to delete

	\return <code>Observable< string ></code> resolves with the new value's hash code
		when the write completes. You may safely ignore this Observable.

	This replaces the revision in question with a new value that contains
	the value <code>_deleted = true</code>. This signals that the previous
	revision is now deleted.

	Among other things, this can be used to resolve a conflict. A conflict
	exists only when multiple current revisions for a document exist. If you
	delete revisions until there is only one current revision, the conflict
	is resolved.

	The revision being deleted must be current.
]]--

function WFRxDocStoreImp:deleteDoc( docId, optRevId )

	check( 'WFRxDocStoreImp', 'string|number', '?+string' )

	assertNotClosed( self, 'deleteDoc' )

	docStoreImpLog:debugf( "deleteDoc %s %s", docId, optRevId )

	return deleteDoc_imp( self, docId, optRevId )

end

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

local function deleteSpace_imp( self, spaceId, excludedDocumentIds )

	assertSpaceIdIsValid( spaceId )

	local db = self._db

	local maybeExcludeDocumentIdsClause = ""

	excludedDocumentIds = excludedDocumentIds or {}

	if next( excludedDocumentIds ) then

		maybeExcludeDocumentIdsClause =
				string.format( "AND docs.fullDocId NOT IN ( %s )",
						WFSQLiteUtils.createVariableArgList( #excludedDocumentIds ) )

	end

	local exec = self._exec
	local gatherArrayOfArray = self._gatherArrayOfArray

	local deletedInfos = {}

	local newRevSequence = db:withTransactionAndDisabledForeignKeysDo( function()

		-- In the following code we manually delete the affected labels and update the affected docs,
		-- making the cascade deletion logic unnecessary.

		-- Step 1. Delete all revisions for all documents associated with the space.

		exec( db, [[DELETE FROM revs
						WHERE localDocId IN (
								SELECT docs.localDocId
									FROM docs
									WHERE spaceId = ? ]] .. maybeExcludeDocumentIdsClause .. [[ )]],
									spaceId,
									unpack( excludedDocumentIds ) )

		-- Step 2. Delete all labels for all documents associated with the space.

		exec( db, [[DELETE FROM labels
						WHERE localDocId IN (
								SELECT docs.localDocId
									FROM docs
									WHERE spaceId = ? ]] .. maybeExcludeDocumentIdsClause .. [[ )]],
									spaceId,
									unpack( excludedDocumentIds ) )

		-- Step 3. Insert a new deletion tombstone into the revs table for all documents
		-- associated with the space.

		local deletedSpaceTombstoneRevisionId = '00000000000000000000000000000000'

		exec( db, [[INSERT INTO revs( localDocId, revId, parentRevSequence, current, deleted, content )
						SELECT docs.localDocId, ?, NULL, 1, 1, ?
							FROM docs
							WHERE spaceId = ? ]] .. maybeExcludeDocumentIdsClause,
							deletedSpaceTombstoneRevisionId,
							emptyTombstone,
							spaceId,
							unpack( excludedDocumentIds ) )

		-- Step 4. Insert a new label for each of the new tombstone revisions to prevent
		-- them from syncing to Oz.

		exec( db, [[INSERT OR REPLACE INTO labels
						SELECT revs.localDocId, revs.sequence, '_dontsendtooz' as label
							FROM revs
							INNER JOIN docs USING( localDocId )
							WHERE spaceId = ? ]] .. maybeExcludeDocumentIdsClause,
							spaceId,
							unpack( excludedDocumentIds ) )

		-- Step 5. Update all documents associated with the space to point at the new
		-- revisions.

		exec( db, [[UPDATE docs
						SET winningRevSequence = (
								SELECT sequence
									FROM revs
									WHERE localDocId == docs.localDocId ),
							mostRecentRevSequence = (
								SELECT sequence
									FROM revs
									WHERE localDocId == docs.localDocId ),
							deleted = 1,
							hasConflicts = 0,
							annotation = ?
						WHERE localDocId in (
							SELECT docs.localDocId
								FROM docs
								WHERE spaceId = ? ]] .. maybeExcludeDocumentIdsClause .. [[ )]],
								emptyTombstone,
								spaceId,
								unpack( excludedDocumentIds ) )

		-- Step 6. Get the new change counter

		local latestRevSequenceRow =
			exec( db, [[SELECT sequence FROM revs
							ORDER by sequence DESC
							LIMIT 1]] )

		-- Step 7. Get the list of documents we just deleted so we can notify the _deletedDocStream
		deletedInfos = gatherArrayOfArray( db, [[SELECT fullDocId, winningRevSequence FROM docs
													WHERE spaceId = ? ]] .. maybeExcludeDocumentIdsClause,
													spaceId,
													unpack( excludedDocumentIds ) )

		return latestRevSequenceRow and latestRevSequenceRow.sequence

	end )


	if newRevSequence then
		self._changeCounter:onNext( newRevSequence )
	end

	for _, row in ipairs( deletedInfos ) do
		local deletedDoc = {
			_id = row[ 1 ],
			_spaceId = spaceId,
			_changeCounter = row[ 2 ],
			_deleted = true,
		}
		self._deletedDocStream:onNext( deletedDoc, newRevSequence )
	end

	return Observable.just( true )

end

--------------------------------------------------------------------------------
--[[!
	\brief Efficiently mass-delete all documents/revisions associated with a space.
		This API is meant to purge the local doc store of space-related content.

		The API does the following:

		- Deletes all revisions for all documents associated with the space.
		- Inserts a new deletion tombstone into the revs table for all documents
			associated with the space.
		- Inserts a new '_dontsendtooz' label for each of the new tombstone revisions
			to prevent them from syncing to Oz.
		- Updates all documents associated with the space to point at the new tombstone
			revisions.

	\param spaceId (string) the space Id
	\param excludedDocumentIds (?table) A list of document Ids, belonging to the space,
		that should be excluded from the actions of this function.

	\return <code>Observable< boolean ></code> Resolves to true.
]]--

function WFRxDocStoreImp:deleteSpace( spaceId, excludedDocumentIds )

	check( 'WFRxDocStoreImp', '+string', '?table' )

	assertNotClosed( self, 'deleteSpace' )

	docStoreImpLog:debugf( "deleteSpace %s", spaceId )

	return deleteSpace_imp( self, spaceId, excludedDocumentIds )

end

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

function WFRxDocStoreImp:resolveConflictWithDelete( docId, optSuppressNewRevision )

	check( 'WFRxDocStoreImp', 'string|number', '?boolean' )

	assertNotClosed( self, 'deleteDoc' )

	docStoreImpLog:debugf( "resolveConflictWithDelete %s", docId )

	-- Extract doc ID from the value.

	assertDocIdIsValid( docId )

	local db = self._db

	local result, newRevSequence = db:withTransactionDo( function()

		local exec = self._exec

		local dvArgs = {
			simplifyDeletedDocs = false,
			unpackDocValue = true
		}

		local docValue = docValue_inTransaction( self, db, exec, docId, nil, dvArgs )

		docValue._deleted = true
		docValue._conflict = nil
		docValue._changeCounter = nil
		docValue._labels = nil

		local args = extractAndValidateDocValue( self, 'resolveConflictWithDelete', docValue )

		-- Force a new revision id here on the delete this should
		-- cause the WFCatMetadataPusher to send up a 'delete'
		-- based on optSuppressNewRevision not being set.
		args.didContentChange = not optSuppressNewRevision

		return setDocValueImp_inTransaction( self, db, exec,
											 'resolveConflictWithDelete', args )

	end )

	if newRevSequence then
		self._changeCounter:onNext( newRevSequence )
	end

	return Observable.just( result )

end

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

--[[!
	\brief Deletes a specific revision of a document without creating the new revision as
		a current revision. This is necessary when resolving conflicts so that, once
		processed, the deletion of the conflicting revision does not create a current
		revision (see bug NMB-2952)

	\param docId (string|number) the document's unique or local ID
	\param revId (string) revision to delete

	\return <code>Observable< string ></code> resolves with the new value's hash code
		when the write completes. You may safely ignore this Observable.
]]--
function WFRxDocStoreImp:deleteRevisionAsNonCurrent( docId, revId )

	check( 'WFRxDocStoreImp', 'string|number', 'string' )

	assertNotClosed( self, 'deleteRevisionAsNonCurrent' )

	return deleteDoc_imp( self, docId, revId, true )

end

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

local function docIdToLocalIdImp_InTransaction( self, db, docId, includeDeleted )

	local localId, deleted

	localId = self._docIdToLocalIdCache[ docId ]

	if not localId then
		local localIdRow = includeDeleted and
			self._exec( db, [[SELECT localDocId, deleted FROM docs WHERE fullDocId = ?]], docId ) or
			self._exec( db, [[SELECT localDocId, deleted FROM docs WHERE fullDocId = ? AND deleted = 0]], docId )

		localId = localIdRow and localIdRow.localDocId or nil
		deleted = localIdRow and localIdRow.deleted == 1
	end

	if localId then
		updateIdCaches( self, localId, docId, deleted )
	end

	return localId

end

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

-- @mark revision labels

--------------------------------------------------------------------------------
--[[!
	\brief Attaches a label to a given revision of a document.

	\param docId (string) document ID
	\param label (string) label name (any non-empty string)
	\param revId (string) revision ID

	Think of a label as a movable tag that can be used later to search for
	particular document revisions. Like git tags, they are simply named aliases
	for another entity in the system (in this case document revisions).
	Unlike git tags, they apply only to a single document.

	Labeled revisions are not purged when the database is compacted.

	\return <code>Observable< empty ></code> resolves when done
]]--

function WFRxDocStoreImp:setLabel( docId, label, revId )

	check( 'WFRxDocStoreImp', '+string', '+string', '+string' )

	assertNotClosed( self, 'setLabel' )

	docStoreImpLog:debugf( "setLabel %s %s %s", docId, label, revId )

	local db = self._db

	db:withTransactionDo( function()

		local exec = self._exec

		-- Get the doc's local numeric ID.

		local localDocId = docIdToLocalIdImp_InTransaction( self, db, docId, true )

		if not localDocId then
			error( string.format( "WFRxDocStoreImp:setLabel: no such document %s", docId ), 4 )
		end

		-- Get the rev's local numeric ID (change counter).

		local revIdRow = exec( db, [[SELECT sequence FROM revs WHERE revId = ? ORDER BY sequence DESC LIMIT 1]], revId )

		local revSequence = revIdRow and revIdRow.sequence
		if not revSequence then
			error( string.format( "WFRxDocStoreImp:setLabel: no such revision %s", revId ), 4 )
		end

		-- Update the label.

		exec( db, [[INSERT OR REPLACE INTO labels VALUES( ?, ?, ? )]],
			  localDocId, revSequence, label )

		-- Do we need to update an observable doc value?

			-- TODO [scouten 2015-05-13]: Unimplemented.

	end )

	return Observable.empty()

end

--[[!
	\brief Removes the label (if present) from a given revision of a document.

	\param docId (string) document ID
	\param label (string) label name (any non-empty string)
	\param revId (string) revision ID

	\return <code>Observable< empty ></code> resolves when done
]]--

function WFRxDocStoreImp:removeLabel( docId, label, revId )

	check( 'WFRxDocStoreImp', '+string', '+string', '+string' )

	assertNotClosed( self, 'removeLabel' )

	docStoreImpLog:debugf( "removeLabel %s %s %s", docId, label, revId )

	local db = self._db

	db:withTransactionDo( function()

		local exec = self._exec

		-- Get the doc's local numeric ID.

		local localDocId = docIdToLocalIdImp_InTransaction( self, db, docId, true )

		if not localDocId then
			error( string.format( "WFRxDocStoreImp:removeLabel: no such document %s", docId ), 4 )
		end

		-- Get the rev's local numeric ID (change counter).

		local revIdRow = exec( db, [[SELECT sequence FROM revs WHERE revId = ? ORDER BY sequence DESC LIMIT 1]], revId )

		local revSequence = revIdRow and revIdRow.sequence
		if not revSequence then
			error( string.format( "WFRxDocStoreImp:removeLabel: no such revision %s", revId ), 4 )
		end

		-- Remove the label.

		exec( db, [[DELETE FROM labels WHERE localDocId = ? AND revSequence = ? AND name = ?]],
			  localDocId, revSequence, label )

	end )

	return Observable.empty()

end

--------------------------------------------------------------------------------
--[[!
	\brief Get the revision for a specific label of a document.

	\param docId (string) document ID
	\param label (string) label name (any non-empty string)

	\return <code>Observable< string ></code> tagged revision
]]--

function WFRxDocStoreImp:getRevisionForLabel( docId, label )

	check( 'WFRxDocStoreImp', '+string', '+string' )

	assertNotClosed( self, 'getRevisionForLabel' )

	docStoreImpLog:debugf( "getRevisionForLabell %s %s", docId, label )

	local db = self._db

	local result = db:withTransactionDo( function()

		local exec = self._exec

		-- Get the doc's local numeric ID.

		local localDocId = docIdToLocalIdImp_InTransaction( self, db, docId, true )

		if not localDocId then
			error( string.format( "WFRxDocStoreImp:getRevisionForLabel: no such document %s", docId ), 4 )
		end

		-- Get the tagged revs local sequenceId.

		local revSequenceRow = exec( db, [[SELECT revSequence FROM labels WHERE localDocId = ? AND name = ? ORDER BY revSequence DESC LIMIT 1]],
										localDocId, label )

		local revSequence = revSequenceRow and revSequenceRow.revSequence
		if not revSequence then
			error( string.format( "WFRxDocStoreImp:getRevisionForLabel: no such label %s", label ), 4 )
		end

		-- Retrieve the revision.

		local revIdRow = exec( db, [[SELECT revId FROM revs WHERE sequence = ? ORDER BY revId DESC LIMIT 1]], revSequence )

		local revId = revIdRow and revIdRow.revId

		if not revId then
			error( string.format( "WFRxDocStoreImp:getRevisionForLabel: no such local revision %s", revSequence ), 4 )
		end

		return revId

	end )

	return Observable.just( result )

end

--------------------------------------------------------------------------------
--[[!
	\brief Update a specific label for a doc from a revision list.

	Updates the specified label of the specified docId to the latest revId
	in the revIdList which also exists in the local docStore under the condition
	that the local sequencenumber is higher than the current tagged sequence number.

	\param docId (string) document ID
	\param label (string) label name (any non-empty string)
	\param revList (table) revision list.

	\return <code>Observable< empty ></code> resolves when done
]]--

function WFRxDocStoreImp:updateLabelFromRevisionList( docId, label, revIdList )

	check( 'WFRxDocStoreImp', '+string', '+string', 'table' )

	assertNotClosed( self, 'updateLabelFromRevisonList' )

	docStoreImpLog:debugf( "updateLabelFromRevisonList %s %s %s", docId, label, revIdList )

	local db = self._db

	db:withTransactionDo( function()

		local exec = self._exec

		-- Get the doc's local numeric ID.

		local localDocId = docIdToLocalIdImp_InTransaction( self, db, docId, true )

		if not localDocId then
			error( string.format( "WFRxDocStoreImp:updateLabelFromRevisonList: no such document %s", docId ), 4 )
		end

		-- Get the tagged revs local sequenceId.

		local revTaggedSequenceRow = exec( db, [[SELECT revSequence FROM labels WHERE localDocId = ? ORDER BY revSequence DESC LIMIT 1]], localDocId )

		local revTaggedSequence = revTaggedSequenceRow and revTaggedSequenceRow.revSequence
		if not revTaggedSequence then
			error( string.format( "WFRxDocStoreImp:updateLabelFromRevisonList: no such label %s", label ), 4 )
		end

		-- Get the local revisions which have a higher sequence number than the tagged one

		local revSequenceRows = db:gatherArrayOfArray(
				[[SELECT sequence, revId FROM revs WHERE localDocId = ? AND sequence > ? ORDER BY sequence DESC]],
				localDocId, revTaggedSequence )

		local found = false

		local toBeLabeledSequence
		local revId

		for i = 1, #revSequenceRows do

			local row = revSequenceRows[ i ]

			toBeLabeledSequence = row[1]
			revId = row[2]

			for j = #revIdList, 1, -1 do
        		if revIdList[j] == revId then
					found = true
					break
				end
			end

			if found == true then
				break
			end

		end

		-- If we found a local revision in the revIdList, update the label.
		if found then

			exec( db, [[INSERT OR REPLACE INTO labels VALUES( ?, ?, ? )]],
			  	localDocId, toBeLabeledSequence, label )

		end

	end )

	return Observable.empty()

end

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

-- @mark document type and subtype

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

local function buildTypeExpression( docTypes )

	local queryParams = {}
	local typeExpression = ""

	-- build the query

	for docType, docSubtypes in pairs( docTypes ) do

		if type( docSubtypes ) ~= 'table' or #docSubtypes == 0 then
			error( "WFRxDocStoreImp:getDocsOfType(): empty or invalid subtype list for type '%s'", docType )
		end

		for _, docSubtype in ipairs( docSubtypes ) do
			if typeExpression ~= "" then
				typeExpression = typeExpression .. " OR"
			end

			if docSubtype == "*" then

				typeExpression = typeExpression .. " type = ?"
				queryParams[ #queryParams + 1 ] = docType

			elseif docSubtype == "NULL" then

				typeExpression = typeExpression .. " type = ? AND subtype IS NULL"
				queryParams[ #queryParams + 1 ] = docType

			else

				typeExpression = typeExpression .. " type = ? AND subtype = ?"
				queryParams[ #queryParams + 1 ] = docType
				queryParams[ #queryParams + 1 ] = docSubtype

			end

		end

	end

	-- We'll never get there, but we can only have up to 999 bind parameters.
	if #queryParams > 990 then
		error( "WFRxDocStoreImp:getDocsOfType(): too many type/subtype tuples in query" )
	end

	return queryParams, typeExpression

end

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

--[[!
	\brief Get the document ids of all docs of the specified types and subtypes.
		Returns the document ids ordered by change counter, smallest to largest.

	\param docTypes (table) map with doc types as keys and a list of subtypes as values
	\param fromChangeCounter (optional number) A previous change-counter value. Documents
		whose change counter is less than this value will be ignored.
	\param maxDocuments (optional number) The maximum number of documents to return

	Spezial subtype values:
	- '*' returns all docs of specified type  (regardless of subtype)
	- 'NULL' returns only docs of the specified type where the subtype is exactly NULL

	No verification is done if the passed type and subtype arguments
	are compatible

	\return <code>Observable<table,changeCounter></code> list of fulldocIds and last changeCounter (for batching)
]]--

function WFRxDocStoreImp:getDocsOfType( docTypes, fromChangeCounter, maxDocuments )

	check( 'WFRxDocStoreImp', 'table', '?number', '?number' )

	assertNotClosed( self, 'getDocsOfType' )

	docStoreImpLog:debugf( "getDocsOfType" )

	local db = self._db

	if not fromChangeCounter or type( fromChangeCounter ) ~= 'number' or fromChangeCounter < 0 then
		fromChangeCounter = 0
	end

	local sql, queryParams

	if isAllDocTypesAndSubtypes( docTypes ) then

		queryParams = {}

		sql = [[SELECT fullDocId, winningRevSequence FROM docs
					WHERE winningRevSequence >= ?
					ORDER by winningRevSequence ASC]]

	else

		local maybeQueryParams, typeExpression = buildTypeExpression( docTypes )

		if #maybeQueryParams > 0 then

			queryParams = maybeQueryParams

			sql = string.format(
						[[SELECT fullDocId, winningRevSequence FROM docs
							WHERE ( %s ) AND ( winningRevSequence >= ? )
							ORDER by winningRevSequence ASC]], typeExpression )

		end

	end

	if sql and queryParams then

		if maxDocuments then
			sql = sql .. ' LIMIT ?'
		end

		-- In the below call to gatherArray, we need to unpack queryParams
		-- and also include the fromChangeCounter. However, the behavior of
		-- unpack changes depending on how it's used. The following is from
		-- http://www.lua.org/pil/5.1.html:

		-- "Lua always adjusts the number of results from a function to the
		-- circumstances of the call. When we call a function as a statement,
		-- Lua discards all of its results. When we use a call as an expression,
		-- Lua keeps only the first result. We get all results only when the
		-- call is the last (or the only) expression in a list of expressions."

		-- For example: db:gatherArray( sql, unpack( queryParams ), fromChangeCounter )
		-- In this example, unpack will only return the first result from queryParams.

		-- As a workaround, we can append fromChangeCounter to queryParams which
		-- gives us our desired result.

		queryParams[ #queryParams + 1 ] = fromChangeCounter

		if maxDocuments then
			queryParams[ #queryParams + 1 ] = maxDocuments
		end

		local rows = db:gatherArrayOfArray( sql, unpack( queryParams ) )

		if rows and next( rows ) then

			local docIds = {}
			for i, row in ipairs( rows ) do
				docIds[ i ] = row[ 1 ]
			end

			return Observable.just( docIds, rows[ #rows ][ 2 ] )

		end

	end

	return Observable.just( {}, nil )

end

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

--[[
	\brief Get the document ids of all docs which have the specified label

	\param label (string)

       	\return <code>Observable<table></code> list of fulldocIds
]]
function WFRxDocStoreImp:getIdsOfDocsWithLabel( label )

	check( 'WFRxDocStoreImp', '+string')

	assertNotClosed( self, 'getIdsOfDocsWithLabel' )

	docStoreImpLog:debugf( 'getIdsOfDocsWithLabel' )

	local db = self._db

	-- Select the doc Id where the lastest revision has a matching label name
	local sql = [[SELECT doc.fullDocId
			FROM docs AS doc
				INNER JOIN revs ON doc.winningRevSequence == revs.sequence
				LEFT JOIN labels AS label ON label.localDocId = doc.localDocId
			WHERE doc.winningRevSequence = label.revSequence
			AND doc.hasConflicts = 0
			AND doc.deleted = 0
			AND label.name = ?]]

	local results = db:gatherArray( sql, label )

	if results and next( results ) then

		return Observable.just( results )

	else

		return Observable.just( {} )

	end

end

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

--[[!
	\brief Count the documents in the doc store.

	\return <code>Observable<number></code> the number of documents.
]]--

function WFRxDocStoreImp:countAllDocs()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'countAllDocs' )

	docStoreImpLog:debugf( "countAllDocs" )

	local db = self._db
	local docIdRows = db:gatherArray( [[SELECT COUNT( * ) FROM docs]] )

	local count = docIdRows and docIdRows[ 1 ] or 0

	return Observable.just( count )

end

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

--[[!
	\brief Count the documents of the specified types and subtypes.

	\param docTypes (table) map with doc types as keys and a list of subtypes as values

	Spezial subtype values:
	- '*' returns all docs of specified type  (regardless of subtype)
	- 'NULL' returns only docs of the specified type where the subtype is exactly NULL

	No verification is done if the passed type and subtype arguments
	are compatible.

	\return <code>Observable<number></code> the number of docs matching the
		specified types and subtypes.
]]--

function WFRxDocStoreImp:countDocsOfType( docTypes )

	check( 'WFRxDocStoreImp', 'table' )

	assertNotClosed( self, 'countDocsOfType' )

	docStoreImpLog:debugf( "countDocsOfType" )

	if isAllDocTypesAndSubtypes( docTypes ) then
		return self:countAllDocs()
	end

	local db = self._db
	local docIdRows

	local queryParams, typeExpression = buildTypeExpression( docTypes )

	if #queryParams > 0 then

		local sql =
				string.format(
					[[SELECT COUNT( * ) FROM docs
						WHERE ( %s )]], typeExpression )

		docIdRows = db:gatherArray( sql, unpack( queryParams ) )

		if docIdRows then

			local count = docIdRows[ 1 ]

			return Observable.just( count )

		end

	end

	return Observable.just( 0 )

end

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

  --[[!
 	\brief Return a table with the number of documents grouped by type and subtype.
	This function is only used on testing at: LrDocStoreTypeMigratorVerificationTest.lua

 	Note: no verification is done if the passed type and subtype arguments
 	are compatible

 	\return <code>Observable<table></code> number of documents of a given type
 ]]--

 function WFRxDocStoreImp:DEBUG_getNumberOfDocsByTypeAndSubtype()

	check( 'WFRxDocStoreImp')

	assertNotClosed( self, 'DEBUG_getNumberOfDocsByTypeAndSubtype' )

	docStoreImpLog:debugf( "DEBUG_getNumberOfDocsByTypeAndSubtype")

	local db = self._db
    local numOfDocs

	-- Count the type and subtype
	numOfDocs = db:gatherArrayOfArray( [[SELECT type, subtype, COUNT(*) FROM docs WHERE deleted != 1 GROUP BY type, subtype  ]] )

   if numOfDocs then
		return Observable.just( numOfDocs )
	end

	return Observable.just(0)
end

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

  --[[!
 	\brief Tests the integrity of a docstore.
	This function is only used on testing at: LrDocStoreTypeMigratorVerificationTest.lua

 	\return <code>Observable<boolean></code>  Success if the docstore is corrupt,
			failure if docstore is not corrupt.
 ]]--

 function WFRxDocStoreImp:isCorrupt()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'isCorrupt' )

	docStoreImpLog:debugf( "isCorrupt")

	local db = self._db
	local isDocstoreCorrupt = db:gatherArray( [[PRAGMA integrity_check;]])

	if isDocstoreCorrupt[1] == "ok" then
		return Observable.just('false')
	else
		return Observable.just('true')
	end
end
--------------------------------------------------------------------------------

-- @mark partial document values

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

local function assertValidOverlay( overlay )

	-- It's not allowed to specify parent _revId or _revIds.

	if overlay._revId ~= nil then
		error( "WFRxDocStoreImp:applyOverlay: _revId is not permitted", 2 )
	end

	if overlay._revIds ~= nil then
		error( "WFRxDocStoreImp:applyOverlay: _revIds is not permitted", 2 )
	end

	-- It's not allowed to update labels.

	if overlay._labels ~= nil then
		error( "WFRxDocStoreImp:applyOverlay: _labels is not permitted", 2 )
	end

	-- Verify that doc value is acceptable (does not contain illegal keys, etc.).

	assertDocOverlayIsValid( overlay )

end

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

local function applyOverlayImp_inTransaction( self, db, docId, overlay, labels, preserveLabels )

	local exec = self._exec

	local revisionRow = revisionRow_inTransaction( db, exec, docId  )

	if not revisionRow then
		revisionRow = {
			annotation = emptyArchive,
			content = emptyArchive,
		}
	end

	-- if the revisionRow is marked deleted, keep it that way
	local docValueIsDeleted = revisionRow.deleted == 1

	-- record and remove any new revision id from the overlay so we can apply the overlay cleanly
	local newRevId = overlay._newRevId
	overlay._newRevId = nil

	-- same for space id
	local spaceId = overlay._spaceId and assertSpaceIdIsValid( overlay._spaceId ) or revisionRow.spaceId
	overlay._spaceId = nil

	local newIdOverlay = overlay._newIdOverlay
	overlay._newIdOverlay = nil

	-- applyOverlay() isn't allowed to change the type or subtype of a document if the doc
	-- is already typed (see contentOverlay verification below). But if the doc is not
	-- yet typed/subtyped, a type/subtype can be introduced via an applyOverlay().

	-- Record type/subtype, but different from spaceId, do not remove them from the overlay.
	local docType = overlay.type
	local docSubtype = overlay.subtype

	-- the content and annotation are stored and serialized separately so separate out the overlay
	-- in the same manner so that we can apply the overlay to either or both pieces separately
	-- and avoid changing those we don't have to
	local contentOverlay, annotationOverlay = validateAndExtractOverlay( self, overlay )

	-- if the overlay is a deletion and has a new id we are moving the current doc over
	-- to the new id, but we don't do this for already deleted documents
	if overlay._deleted == true and overlay._newId then

		if docValueIsDeleted then
			error( "Cannot apply an overlay with _newId to an already deleted document" )
		else

			local annotation = contentFromFlattenedValue( revisionRow.annotation )
			local content = contentFromFlattenedValue( revisionRow.content )

			-- any _serverProvidedMetadata that applied to the previous document does not apply as _spm
			-- to the new document, it is not yet on Oz (should this be a caller's decision?)
			annotation._serverProvidedMetadata = nil

			local newDocRevId
			if newIdOverlay then
				newDocRevId = newIdOverlay._newRevId
				newIdOverlay._newRevId = nil

				local newContentOverlay, newAnnotationOverlay = validateAndExtractOverlay( self, newIdOverlay )

				content = WFTableMerge.applyOverlay( content, newContentOverlay )
				annotation = WFTableMerge.applyOverlay( annotation, newAnnotationOverlay )

			end

			local args = {
				docId = overlay._newId,
				docType = content.type,
				docSubtype = content.subtype,
				newRevId = newDocRevId,
				bodyContentArchive = tableToString( content ),
				annotationContentArchive = tableToString( annotation ),
				didContentChange = true,
				didAnnotationChange = true,
				docValueIsDeleted = false,
			}
			setDocValueImp_inTransaction( self, self._db, exec, 'applyOverlay_newId', args )
		end
	end

	-- if the doc we want to apply this overlay to is already deleted and has a newId, we should apply
	-- the overlay to the document referenced by newId instead as long as the overlay is not a deletion
	if docValueIsDeleted and not overlay._deleted then
		local content = contentFromFlattenedValue( revisionRow.content )

		if content._newId then
			return applyOverlayImp_inTransaction( self, db, content._newId, overlay, labels, preserveLabels )
		end
	end

	-- apply the overlay to the content if necessary
	local didContentChange, contentArchive

	if next( contentOverlay ) then

		local oldContent = contentFromFlattenedValue( revisionRow.content )

		if oldContent.type and contentOverlay.type and oldContent.type ~= contentOverlay.type then
			errorf( "Cannot change type of (%s to %s) of document %s via applyOverlay once set before", oldContent.type, contentOverlay.type, docId )
		end

		if oldContent.subtype and contentOverlay.subtype and oldContent.subtype ~= contentOverlay.subtype then
			errorf( "Cannot change subtype (%s to %s) of document %s via applyOverlay once set before", oldContent.subtype, contentOverlay.subtype, docId )
		end

		local newContent
		newContent, didContentChange = WFTableMerge.applyOverlay( KSTableUtils.deepCopy( oldContent ), contentOverlay )

		if didContentChange then
			contentArchive = tableToString( newContent )
		end

		-- is, or was, the docValue deleted
		docValueIsDeleted = newContent._deleted == true

	end

	contentArchive = contentArchive or revisionRow.content

	-- apply the overlay to the annotation
	local didAnnotationChange, annotationArchive
	if next( annotationOverlay ) then

		local oldAnnotation = contentFromFlattenedValue( revisionRow.annotation )

		local newAnnotation
		newAnnotation, didAnnotationChange = WFTableMerge.applyOverlay( KSTableUtils.deepCopy( oldAnnotation ), annotationOverlay )

		if didAnnotationChange then
			annotationArchive = tableToString( newAnnotation )
		end

	end

	annotationArchive = annotationArchive or revisionRow.annotation

	-- if neither the annotation or the content changed, return the existing revision
	-- unless the caller explicitly asked for a new revision
	if not ( didContentChange or didAnnotationChange ) and not newRevId then
		return revisionRow.revId
	end

	local args =
		{
			docId = docId,
			parentRevId = revisionRow.revId,
			newRevId = newRevId,
			bodyContentArchive = contentArchive,
			didContentChange = didContentChange or false,
			didAnnotationChange = didAnnotationChange or false,
			annotationContentArchive = annotationArchive,
			annotationAlreadyUpdated = true,
			docValueIsDeleted = docValueIsDeleted,
			labels = labels,
			preserveLabels = preserveLabels,
			spaceId = spaceId,
			docType = docType,
			docSubtype = docSubtype,
		}

	return setDocValueImp_inTransaction( self, self._db, exec,
										 'applyOverlay', args )

end

--------------------------------------------------------------------------------
--[[!
	\brief Updates the document value, applying an overlay to any existing value.

	\param overlay (table) delta to apply to the existing document value

	\c overlay \b must contain the following elements:
		* \c _id (string|integer) the document's unique ID or the document's
			integer local ID.

	\c overlay \b may contain the following element:
		* \c _newRevId (string) if present, the doc store uses this uuid as the
			new revision id

	\c overlay \b may contain the following elements:
		* \c _deleted = true, if present, this marks the document as deleted similar
			to calling <code>deleteDoc( docId )</code>
		* \c _newId (string) only allowed when _deleted = true, _newId may be set
			to a new uuid which will cause the doc store to copy the current
			version of the doc into a new document with the _newId uuid
		* \c _newIdOverlay (table) when _newId is set this is a separate overlay
			that will be applied to copied document as it's being copied into the
			new uuid. This overlay *can* change the type/subtype as this is a new
			document.
	\c overlay <b>MUST NOT</b> contain a \c _rev entry. This function can only
		be applied to the current winning revision of a document.

	\c overlay <b>MUST NOT</b> contain a \c _labels entry. This function can not
		be used to create or update labels.

	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.

	\return <code>Observable< string ></code> resolves with the new value's
		revision ID when the write completes. You may safely ignore this promise.
]]--

function WFRxDocStoreImp:applyOverlay( overlay, description )

	check( 'WFRxDocStoreImp', 'table', '?string' )

	assertNotClosed( self, 'applyOverlay' )

	addToBulkRequestInfo( self, 'applyOverlay', description, 1 )

	docStoreImpLog:debugf( "applyOverlay %s", overlay )

	-- Extract doc ID from the value.

	local docId = assertDocIdIsValid( overlay._id )
	overlay._id = nil

	local docLabels = nil
	if overlay._newLabel then
		docLabels = {}
		docLabels[ overlay._newLabel ] = true
		overlay._newLabel = nil
	end

	local docPreserveLabels = overlay._preserveLabels
	overlay._preserveLabels = nil

	assertValidOverlay( overlay )

	local db = self._db

	local result, newRevSequence = db:withTransactionDo( function( myDb)
		return applyOverlayImp_inTransaction( self, myDb, docId, overlay, docLabels, docPreserveLabels )
	end, db )

	if newRevSequence then
		self._changeCounter:onNext( newRevSequence )
	end

	return Observable.just( result )

end

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

local inSqlStringCacheMap = {}

local function modifyIdSqlString( sql, idColumnName, numItems )
	local key = sql .. idColumnName .. tostring( numItems )
	local sqlString = inSqlStringCacheMap[ key ]

	if not sqlString then

		sqlString = string.format( sql, idColumnName, WFSQLiteUtils.createVariableArgList( numItems ) )
		inSqlStringCacheMap[ key ] = sqlString

	end
	return sqlString
end

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

local function bulkIdTranslation_inTransaction( self, db, ids, includeDeleted,
		idType, idColumnName, lookupCachedTranslatedIdFn, addTranslatedIdToResultAndCacheFn )

	local idToTranslatedIdMap = {}

	local idsToLookup = {}

	for _, id in ipairs( ids ) do
		assert( type( id ) == idType )

		-- check the cache, maybe we can save a db lookup
		local translatedId = lookupCachedTranslatedIdFn( id )
		if translatedId then
			idToTranslatedIdMap[ id ] = translatedId
		else
			idsToLookup[ #idsToLookup + 1 ] = id
		end
	end

	-- use the db to lookup any that weren't cached
	local numberOfDocsToProcess = #idsToLookup

	local offset = 1

	while offset <= numberOfDocsToProcess do
		local chunkSize = 0
		local remainingDocCount = numberOfDocsToProcess - offset
		if remainingDocCount >= 128 then
			chunkSize = 128
		elseif remainingDocCount >= 64 then
			chunkSize = 64
		elseif remainingDocCount >= 32 then
			chunkSize = 32
		elseif remainingDocCount >= 16 then
			chunkSize = 16
		elseif remainingDocCount >= 8 then
			chunkSize = 8
		elseif remainingDocCount >= 4 then
			chunkSize = 4
		elseif remainingDocCount >= 2 then
			chunkSize = 2
		else
			chunkSize = 1
		end

		local sql = includeDeleted and
			modifyIdSqlString( [[SELECT localDocId, fullDocId, deleted FROM docs WHERE %s IN ( %s ) ]], idColumnName, chunkSize ) or
			modifyIdSqlString( [[SELECT localDocId, fullDocId, deleted FROM docs WHERE %s IN ( %s ) AND deleted = 0]], idColumnName, chunkSize )

		db:reduceSQLQueryViaValueArrays( idToTranslatedIdMap, addTranslatedIdToResultAndCacheFn, sql, unpack( idsToLookup, offset, offset + chunkSize - 1 ) )

		offset = offset + chunkSize
	end

	return idToTranslatedIdMap
end


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

bulkLocalIdToDocId_inTransaction = function( self, db, ids, includeDeleted )

	local lookupCachedTranslatedIdFn = function( id ) return self._localIdToDocIdCache[ id ] end

	local addTranslatedIdToResultAndCacheFn = function( idToTranslatedIdMap, localId, docId, deleted )
		idToTranslatedIdMap[ localId ] = docId
		updateIdCaches( self, localId, docId, ( deleted == 1 ) )
		return idToTranslatedIdMap
	end

	return bulkIdTranslation_inTransaction( self, db, ids, includeDeleted,
		'number', 'localDocId', lookupCachedTranslatedIdFn, addTranslatedIdToResultAndCacheFn )

end

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

local function bulkDocIdToLocalId_inTransaction( self, db, ids, includeDeleted )

	local lookupCachedTranslatedIdFn = function( id ) return self._docIdToLocalIdCache[ id ] end

	local addTranslatedIdToResultAndCacheFn = function( idToTranslatedIdMap, localId, docId, deleted )
		idToTranslatedIdMap[ docId ] = localId
		updateIdCaches( self, localId, docId, ( deleted == 1 ) )
		return idToTranslatedIdMap
	end

	return bulkIdTranslation_inTransaction( self, db, ids, includeDeleted,
		'string', 'fullDocId', lookupCachedTranslatedIdFn, addTranslatedIdToResultAndCacheFn )

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the local Id of a document given the full doc id. If there is
	no document with the given full doc id, nil is returned.

	\param docId (string) the document's full doc id
	\param includeDeleted (boolean) optional parameter to indicate whether to return the
		id of a deleted document

	\return <code>Observable<number></code> local doc id for this document or nil
	if no document exists with the given full doc id
]]--

function WFRxDocStoreImp:docIdToLocalId( docId, includeDeleted )

	check( 'WFRxDocStoreImp', '+string', '?boolean' )

	assertNotClosed( self, 'docIdToLocalId' )

	assertDocIdIsValid( docId )

	local db = self._db

	includeDeleted = not not includeDeleted

	local localId = db:withTransactionDo( function( myDb)
		local docIdToLocalIdMap = bulkDocIdToLocalId_inTransaction( self, myDb, { docId }, includeDeleted )
		return docIdToLocalIdMap[ docId ]
	end, db )

	return Observable.just( localId )

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves a map of full doc ids to local ids for the given array of full doc ids. If
	a given doc id does not exist, it will not be present in the returned map.

	\param docIds (table) an array of docIds
	\param includeDeleted (boolean) optional parameter to indicate whether to return the
		id of a deleted document

	\return <code>Observable<table></code> map of full doc ids to local ids
]]--

function WFRxDocStoreImp:bulkDocIdToLocalIdAsMap( docIds, includeDeleted )

	check( 'WFRxDocStoreImp', 'table', '?boolean' )

	assertNotClosed( self, 'bulkDocIdToLocalIdAsMap' )

	local db = self._db

	includeDeleted = not not includeDeleted

	local map = db:withTransactionDo( function( myDb )
		return bulkDocIdToLocalId_inTransaction( self, myDb, docIds, includeDeleted )
	end, db )

	return Observable.just( map )
end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the full doc id of a document given the local id. If there is
	no document with the given local id, nil is returned.

	\param localId (number) the document's local doc id
	\param includeDeleted (boolean) optional parameter to indicate whether to return the
		id of a deleted document

	\return <code>Observable<string></code> full doc id for this document or nil
	if no document exists with the given local ids
]]--

function WFRxDocStoreImp:localIdToDocId( localId, includeDeleted )

	check( 'WFRxDocStoreImp', 'number', '?boolean' )

	assertNotClosed( self, 'localIdToDocId' )

	assertDocIdIsValid( localId )

	local db = self._db

	includeDeleted = not not includeDeleted

	local docId = db:withTransactionDo( function( myDb)
		local localIdToDocIdMap = bulkLocalIdToDocId_inTransaction( self, myDb, { localId }, includeDeleted )
		return localIdToDocIdMap[ localId ]
	end, db )

	return Observable.just( docId )

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves a map of local ids to full doc ids for the given array of local ids. If
	a given local id does not exist, it will not be present in the returned map.

	\param localIds (table) an array of local ids
	\param includeDeleted (boolean) optional parameter to indicate whether to return the
		id of a deleted document

	\return <code>Observable<table></code> map of local ids to full doc ids
]]--
function WFRxDocStoreImp:bulkLocalIdToDocIdAsMap( localIds, includeDeleted )

	check( 'WFRxDocStoreImp', 'table', '?boolean' )

	assertNotClosed( self, 'bulkLocalIdToDocIdAsMap' )

	local db = self._db

	includeDeleted = not not includeDeleted
	local map = db:withTransactionDo( function( myDb )
		return bulkLocalIdToDocId_inTransaction( self, myDb, localIds, includeDeleted )
	end, db )

	return Observable.just( map )
end

--------------------------------------------------------------------------------
--[[!
	\brief Updates multiple document values, applying a different overlay to
		each existing value, if any.

	\param overlayArray (array) an array of deltas, just like those supplied to
		applyOverlay() above.

		The array may also contain a <code>label = "name"</code> entry. If this exists,
		that label will be applied to all documents in this batch.

		The array may also contain a <code>preserveLabels = true</code> entry. If this exists,
		the labels will not be changed from their previous value.

	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.

	\return <code>Observable< string ></code> resolves with an array of new value's
		revision IDs when the write completes. You may safely ignore this.
]]--

function WFRxDocStoreImp:bulkApplyOverlay( overlayArray, description )

	check( 'WFRxDocStoreImp', 'table', '?string' )

	assertNotClosed( self, 'bulkApplyOverlay' )

	if #overlayArray == 0 then
		return Observable.throw( "Empty overlay array" )
	end

	addToBulkRequestInfo( self, 'bulkApplyOverlay', description, #overlayArray )

	local db = self._db

	local label = overlayArray.label

	if label and ( type( label ) ~= 'string' or #label == 0 ) then
		error( "WFRxDocStoreImp:bulkApplyOverlay: label must be a non-empty string if provided", 2 )
	end

	local labels
	if label then
		labels = {}
		labels[ label ] = true
	end

	local preserveLabels = overlayArray.preserveLabels

	local result, finalRevSequence = db:withTransactionDo( function( myDb )

		local result = {}
		local latestRevSequence

		for i, overlay in ipairs( overlayArray ) do

			-- Extract doc ID from the value.

			local docId = assertDocIdIsValid( overlay._id )
			overlay._id = nil

			assertValidOverlay( overlay )

			local newRevSequence

			local docLabels = KSTableUtils_deepCopy( labels )
			if overlay._newLabel then
				docLabels = docLabels or {}
				docLabels[ overlay._newLabel ] = true
				overlay._newLabel = nil
			end

			local docPreserveLabels = preserveLabels or overlay._preserveLabels
			overlay._preserveLabels = nil

			result[ i ], newRevSequence = applyOverlayImp_inTransaction( self, myDb, docId, overlay, docLabels, docPreserveLabels )

			if newRevSequence then
				latestRevSequence = newRevSequence
			end

		end

		return result, latestRevSequence

	end, db )

	if finalRevSequence then
		self._changeCounter:onNext( finalRevSequence )
	end

	return Observable.just( result )

end

--------------------------------------------------------------------------------
--[[!
	\brief Updates multiple document values, applying the same overlay to
		each existing value, if any.

	\param overlay (table) delta to apply to the specific document Ids, just
		like the delta supplied to applyOverlay() above with a few notable
		exceptions:

	\c overlay \b must NOT contain the following elements:

		* \c _id: Instead, the document's unique ID should be passed in via the ids
			parameter.

		* \c _newRevId, _newId, and _newIdOverlay: Since the overlay is applied to
			multiple ids, you cannot supply any field that needs to be unique to
			a specific document.

	\param ids (table) a table of either full document ids, or local ids, to
		which the overlay will be applied.

	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.

	\return <code>Observable< string ></code> resolves with an array of new value's
		revision IDs when the write completes. You may safely ignore this.
]]--

function WFRxDocStoreImp:applyOverlayToIds( overlay, ids, description )

	check( 'WFRxDocStoreImp', 'table', 'table', '?string' )

	assertNotClosed( self, 'applyOverlayToIds' )

	if #ids == 0 then
		return Observable.throw( "Empty ids array" )
	end

	if overlay._id ~= nil then
		error( "WFRxDocStoreImp:applyOverlayToIds: _id is not permitted", 2 )
	end

	if overlay._newRevId ~= nil then
		error( "WFRxDocStoreImp:applyOverlayToIds: _newRevId is not permitted", 2 )
	end

	if overlay._newId ~= nil then
		error( "WFRxDocStoreImp:applyOverlayToIds: _newId is not permitted", 2 )
	end

	if overlay._newIdOverlay ~= nil then
		error( "WFRxDocStoreImp:applyOverlayToIds: _newIdOverlay is not permitted", 2 )
	end

	assertValidOverlay( overlay )

	local labels = nil

	if overlay._newLabel then
		labels = {}
		labels[ overlay._newLabel ] = true
		overlay._newLabel = nil
	end

	local docPreserveLabels = overlay._preserveLabels
	overlay._preserveLabels = nil

	addToBulkRequestInfo( self, 'applyOverlayToIds', description, #ids )

	local db = self._db

	local result, finalRevSequence = db:withTransactionDo( function( myDb )

		local result = {}
		local latestRevSequence

		for i, docId in ipairs( ids ) do

			assertDocIdIsValid( docId )

			local newRevSequence

			local docLabels = KSTableUtils_deepCopy( labels )

			result[ i ], newRevSequence = applyOverlayImp_inTransaction( self, myDb, docId, overlay, docLabels, docPreserveLabels )

			if newRevSequence then
				latestRevSequence = newRevSequence
			end

		end

		return result, latestRevSequence

	end, db )

	if finalRevSequence then
		self._changeCounter:onNext( finalRevSequence )
	end

	return Observable.just( result )

end

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

--[[!
	\brief Deletes a specific revision of a document without creating the new revision as
		a current revision followed by an applyOverlay in one transaction.
		This is necessary occasionally when resolving conflicts.

	\param docId (string|number) the document's unique or local ID
	\param revId (string) revision to delete
	\param overlay (table) the overlay

	\return <code>Observable< string ></code> resolves with the new value's hash code
		when the write completes. You may safely ignore this Observable.
]]--
function WFRxDocStoreImp:deleteRevisionAsNonCurrentAndApplyOverlay( docId, revId, overlay)

	check( 'WFRxDocStoreImp', 'string|number', 'string', 'table' )

	docStoreImpLog:debugf( "deleteRevisionAsNonCurrentAndApplyOverlay %s", overlay )

	-- Sanity checks.
	assertNotClosed( self, 'deleteRevisionAsNonCurrentAndApplyOverlay' )

	assertDocIdIsValid( docId )

	-- check that the passed docId and the docId in the overlay match
	if overlay._id ~= docId then
		error("WFRxDocStoreImp::deleteRevisionAsNonCurrentAndApplyOverlay: passed docId and overlay docId do not match.")
	end
	overlay._id = nil

	-- no labels allowed
	if overlay._newLabel then
		error("WFRxDocStoreImp::deleteRevisionAsNonCurrentAndApplyOverlay: no labels allowed.")
	end
	local docLabels = nil

	-- no perserveLabels flag allowed
	if overlay._preserveLabels then
		error("WFRxDocStoreImp::deleteRevisionAsNonCurrentAndApplyOverlay: no preserveLabels flag allowed.")
	end
	local docPreserveLabels = nil

	assertValidOverlay( overlay )

	local db = self._db

	local result, newRevSequence = db:withTransactionDo( function()

		local exec = self._exec

		local docValue = docValue_inTransaction( self, db, exec, docId, revId, { unpackDocValue = true } )

		if not docValue._empty and not docValue._deleted then

			docValue._deleted = true
			docValue._conflict = nil
			docValue._changeCounter = nil
			docValue._labels = nil

			local args = extractAndValidateDocValue( self, 'deleteDoc', docValue )

			args.markNonCurrent = true
			args.didContentChange = true

			setDocValueImp_inTransaction( self, db, exec, 'deleteDoc', args )

		end

		return applyOverlayImp_inTransaction( self, db, docId, overlay, docLabels, docPreserveLabels )

	end )

	if newRevSequence then
		self._changeCounter:onNext( newRevSequence )
	end

	return Observable.just( result )

end

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

-- @mark change notification

--------------------------------------------------------------------------------
--[[!
	\brief Tracks changes in this database by incrementing a counter for each
		change. Be aware that this value can jump, especially if many revisions
		are changed at once.

	\return <code>Subject< number ></code> the current maximum value for
		\c _changeCounter across all revisions in this database
]]--

function WFRxDocStoreImp:changeCounter()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'changeCounter' )

	return self._changeCounter

end

--------------------------------------------------------------------------------
--[[!
	\brief Reports the current change count for this revision.
		This count includes every operation that has completed to this point.

	\return <code>Observable< number ></code> the current maximum value for
		\c _changeCounter across all revisions in this database
]]--

function WFRxDocStoreImp:changeCounterOnce()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'changeCounterOnce' )

	return self._changeCounter:take( 1 )

end

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

local function changedDocsPrep( _, args )

	local from = args.fromChangeCounter

	if not from then
		error( "WFRxDocStoreImp:changedDocs: fromChangeCounter argument missing", 3 )
	end

	if type( from ) ~= 'number'
		or from == kInfinity
		or from == kNegativeInfinity
		or from ~= from
		or from < 0
	then
		error( string_format( "WFRxDocStoreImp:changedDocs: fromChangeCounter illegal (%s)",
							  from ), 2 )
	end

	local defaultLimit = 100
	local maxLimit = 1000

	local limit = args.limit or defaultLimit
	if type( limit ) ~= 'number' then
		error( "WFRxDocStoreImp:changedDocs: limit must be a number or nil for default", 2 )
	end

	if limit < 1
		or limit == kInfinity
		or limit ~= limit
	then
		error( "WFRxDocStoreImp:changedDocs: limit must be a positive number", 2 )
	end

	if limit > maxLimit then

		docStoreImpLog:warnf( "changedDocs: limit is too high (%d), reducing to %d",
							  limit, maxLimit )

		limit = maxLimit

	end

	local includeDocumentsFromSource = args.includeDocumentsFromSource or kAllDocuments
		-- Defaults to only both catalog and space documents

	local filterByTypeSubtypeMap = args.filterByTypeSubtypeMap

	return {
		from = from,
		limit = limit,
		includeDocumentsFromSource = includeDocumentsFromSource,
		filterByTypeSubtypeMap = filterByTypeSubtypeMap
	}

end

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

local function changedDocsCore_inTransaction( self, parsedArgs, description )

	local db = self._db

	local limit = parsedArgs.limit
	local from = parsedArgs.from

	local includeDocumentsFromSource = parsedArgs.includeDocumentsFromSource

	local maybeSpaceIdCondition = " "

	if includeDocumentsFromSource == kAllDocuments then

		-- Include documents with and without spaceIds. This means that both
		-- catalog and space-feed documents will be considered.

	elseif includeDocumentsFromSource == kCatalogDocuments then

		-- Only process documents without spaceIds. These are standard
		-- catalog-based documents.

		maybeSpaceIdCondition = "AND spaceId IS NULL"

	elseif includeDocumentsFromSource == kSpaceFeedDocuments then

		-- Only process documents with spaceIds. These are space-feed documents.

		maybeSpaceIdCondition = "AND spaceId IS NOT NULL"

	else

		changedDocsLog:warn(
			[[changedDocs was specified with an invalid includeDocumentsFromSource argument:
				defaulting to kAllDocuments]] )

	end

	local rawDocRows

	local changeCounter = self._changeCounter:getValue()

	rawDocRows = db:gatherArrayOfArray(
		[[SELECT fullDocId, revId, mostRecentRevSequence,
				content, revs.deleted, hasConflicts, docs.localDocId, annotation,
				revs.sequence, spaceId, type, subtype, labels.name
			FROM docs
				INNER JOIN revs ON docs.winningRevSequence == revs.sequence
				LEFT JOIN labels ON labels.revSequence == revs.sequence
			WHERE mostRecentRevSequence >= ?
				]] .. maybeSpaceIdCondition .. [[
			ORDER BY mostRecentRevSequence
			LIMIT ? ]],
		from, limit + 1 )

	addToBulkRequestInfo( self, 'changedDocs', description, #rawDocRows )

	changedDocsLog:debugf( "changedDocs is returning %s", rawDocRows )

	return rawDocRows, changeCounter, limit

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the set of documents that have changed since a specified point.
	\param args (table) A table with these keys:
		* \c fromChangeCounter (number): A previous change-counter value. Documents
			whose change counter is equal to or greater than this value can match
			this query.
		* \c limit (optional, number): The maximum number of documents to return
		* \c includeDocumentsFromSource (optional, string): A string indicating
			which documents to consider: "all", "catalog" or "spaceFeed". The
			default is "all".
		* \c filterByTypeSubtypeMap (optional, table): map with doc types as
			keys and a list of subtypes as values. If present, only documents
			matching the types/subtypes in this map will be returned. If this
			filter is specified, this API may return fewer results than
			specified via the limit parameter.
	\param description (optional string) string used by debug panel and
		diagnostic log to identify what operations made calls.
	Note that regardless of any value specified in \c limit, a limit of 100 documents
	is always enforced. (The limit is raised to 1000 if \c idsOnly is true.)
	\return <code>Observable< table, number, number ></code> The first return value
		is the raw results of the SQLite query used to fetch document information.
		The second return value is the current doc store change-counter, and the third
		return value is the new limit which was used when running the SQLite query;
		limit may differ from the limit passed in to the args array if args.limit
		was invalid. The receiver thread is responsible for creating fully formed
		document values using these return values.
]]--

function WFRxDocStoreImp:changedDocs( args, description )

	check( 'WFRxDocStoreImp', 'table' )

	assertNotClosed( self, 'changedDocs' )

	local parsedArgs = changedDocsPrep( self, args )

	local rawDocRows, changeCounter, limit = self._db:withTransactionDo(
			changedDocsCore_inTransaction, self, parsedArgs, description )

	if self._TEST_unpackDocValue then

		local nestedTypeSubtypeMap, shouldIncludeAllTypesAndSubtypes
				= WFRxDocStoreUtils.processFilterByTypeSubtypeMap( args.filterByTypeSubtypeMap )

		local result = WFRxDocStoreUtils.processChangedDocsRawDocRows( rawDocRows,
						changeCounter, limit, shouldIncludeAllTypesAndSubtypes,
						nestedTypeSubtypeMap )

		return Observable.just( result )

	end

	return Observable.just( rawDocRows, changeCounter, limit )

end

--------------------------------------------------------------------------------
--[[!
	\brief Watches the database for document value changes and reports them
		as they happen.

	\param optFromChangeCounter (optional, number): A previous change-counter value.
		When provided, any changes >= optFromChangeCounter will be sent, even if
		they have already occurred. It is an error to specify optFromChangeCounter
		that is more than 100 behind the current change counter. If not provided,
		the observations will start from the current change counter.

	\return <code>Observable< table, number, number ></code> The first return value
		is the raw results of the SQLite query used to fetch document information.
		The second return value is the current doc store change-counter, and the third
		return value is the new limit which was used when running the SQLite query;
		limit may differ from the limit passed in to the args array if args.limit
		was invalid. The receiver thread is responsible for creating fully formed
		document values using these return values.
]]--

function WFRxDocStoreImp:_raw_changesFeed( optFromChangeCounter )

	check( 'WFRxDocStoreImp', '?number' )

	assertNotClosed( self, '_raw_changesFeed' )

	-- We use a local behavior subject to track what we've reported to date.
	-- This is either the initial starting point

	local continueFromValue = optFromChangeCounter or -1

	local continueFromB = Rx.BehaviorSubject( continueFromValue )

	-- This first part of the Observable chain should fire only if
	-- continueFromB is behind the doc store's change counter.

	return Observable.combineLatest( continueFromB, self._changeCounter )

		:map( function( _, changeCounter )

			-- There's a slight race condition that can occur if _changeCounter fires
			-- after continueFromB:onNext has been scheduled but before it executes.
			-- When this happens, this map function will trigger due to the
			-- _changeCounter update, but the continueFrom value will be stale. This
			-- then causes the code further down the chain to fetch duplicate updates
			-- via changedDocs because continueFrom hasn't yet updated with the next
			-- value.

			-- To workaround this race condition, we store the current continueFrom value
			-- inside of continueFromValue; continueFromValue is only different from continueFrom
			-- in that small window where the race condition can occur (which only occurs
			-- because continueFromB:onNext is scheduled).

			return continueFromValue, changeCounter

		end )

		:distinctUntilChanged()
			-- This avoids duplicate data events that can occur now because of the map
			-- function above.

		:filter( function( continueFrom, changeCounter )

			-- Special (ugly?) case: If we didn't get a starting point,
			-- swap in the current change counter. Obviously, that implies
			-- we're caught up now.

			if continueFrom == -1 then
				continueFromValue = changeCounter
				continueFromB:onNext( continueFromValue )
				return false
			end

			return continueFrom <= changeCounter
				-- or is it < ?

		end )

		-- We have new documents to report: Use :changedDocs to get that info now.

		:flatMapLatest( function( continueFrom )

			return self:changedDocs( { fromChangeCounter = continueFrom }, 'raw_changesFeed' )

				:flatMapLatest( function( rawDocRows, changeCounter, limit )

					local result, continueFrom

					if self._TEST_unpackDocValue then

						-- Convert the :changedDocs result array into a series of
						-- individual onNext notifications.

						local changedDocs = rawDocRows

						if #changedDocs > 0 then
							result = Observable.from( changedDocs )
						else
							result = Observable.empty()
						end

						continueFrom = changedDocs.continueFrom

					else

						-- #rawDocRows > 0 covers the (unlikely?) case that
						-- we get no documents back. Not sure what Observable.from
						-- does in that case.

						if #rawDocRows > 0 then

							-- SPAM [jandrew 2021-10-01]: Instead of _raw_changesFeed firing with
							-- one document at a time, it will fire with all the new documents at once.
							-- Then, in the caller thread, we process the documents and call
							-- Observable.from( ... ) on the resulting docValues array.
							-- Performance should be better since there are less messages between
							-- threads and the new changedDocs processing code is about 10% more efficient.

							result = Observable.just( rawDocRows, changeCounter, limit )

						else
							result = Observable.empty()
						end

						-- Once we've sent out all of those values, bump
						-- continueFromB so we can do this all again
						-- once there is more to report (which might well be
						-- right now).

						if rawDocRows[ limit + 1 ] then
							continueFrom = rawDocRows[ limit + 1 ][ 3 ]
						else
							continueFrom = changeCounter + 1
						end

					end

					return result:tapOnCompleted( function()

						continueFromValue = continueFrom

						-- The use the of the scheduler here is to prevent a stack overflow
						-- when there are many changed docs to process

						local scheduler = WFRxScheduler.sharedScheduler()

						scheduler:schedule( function()
							continueFromB:onNext( continueFromValue )
						end )

					end )

				end )

		end )

end

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

local function docRevisionHistory_inTransaction( self, docId, optRevId )

	local db = self._db

	local exec = self._exec

	local revSequenceRow

	if optRevId then

		revSequenceRow = exec( db,
			[[SELECT sequence
				FROM revs
					INNER JOIN docs ON docs.localDocId = revs.localDocId
				WHERE
					fullDocId = ? AND revId = ?
				ORDER BY sequence DESC LIMIT 1]], docId, optRevId )

	else

		revSequenceRow = exec( db,
			[[SELECT winningRevSequence AS sequence FROM docs WHERE fullDocId = ?]], docId )

	end

	local nextRevSequence = revSequenceRow and revSequenceRow.sequence

	local historyRows = db:gatherArrayOfArray(
		[[SELECT parentRevSequence, sequence, revId, revs.deleted, content ISNULL
			FROM revs
				INNER JOIN docs ON docs.localDocId == revs.localDocId
			WHERE
				fullDocId = ?
			ORDER BY sequence DESC]], docId )

	local result = {}
	local nResults = 0

	for i = 1, #historyRows do

		local row = historyRows[ i ]

		local rowSequence = row[ 2 ]

		if rowSequence == nextRevSequence then

			local rowRevId = row[ 3 ]

			local status = 'available'

			if row[ 4 ] == 1 then
				status = 'deleted'
			end

			-- NOTE: The sequence of deleted and missing here is based on
			-- an original goal of compatibility with Apache CouchDB. A rev
			-- that is a deletion and has since been purged will be reported
			-- as "missing", not "deleted".

			if row[ 5 ] == 1 then
				-- If there's no content, we may have purged it, but it *may*
				-- be available in another entry in the revs table.
				local revContentRow = exec( db,
					[[SELECT content FROM revs
						INNER JOIN docs ON docs.localDocId == revs.localDocId
						WHERE fullDocId = ? AND revId = ? AND content NOTNULL
						LIMIT 1]], docId, rowRevId )

				if not ( revContentRow and revContentRow.content ) then
					status = 'missing'
				end
			end

			nResults = nResults + 1
			result[ nResults ] = {
				revId = rowRevId,
				status = status,
			}

			nextRevSequence = row[ 1 ]

		end

	end

	return result

end

--------------------------------------------------------------------------------
--[[!
	\brief Returns additional history about the document revisions leading to
		a specific revision.

	\param docId (string) the document's unique ID

	\param optRevId (optional, string) the revision to trace (if not specified,
		defaults to the current winning revision)

	\param optStartingFromAnyOf (optional, table) if any of the revision IDs
		specified in this array occur, stop tracing history beyond the first
		matching rev ID

	\return <code>Observable< table ></code> array of matching revision IDs in history

	An example return might be:
		<code>
			{
				"rev": "8-6f5ad8db0f34af24a6e0984cd1a6cfb9",
				"status": "available"
			},
			{
				"rev": "7-77fba3a059497f51ec99b9b478b569d2",
				"status": "deleted"
			},
			{
				"rev": "6-136813b440a00a24834f5cb1ddf5b1f1",
				"status": "available"
			},
			{
				"rev": "5-fd96acb3256302bf0dd2f32713161f2a",
				"status": "missing"
			},
			{
				"rev": "4-874985bc28906155ba0e2e0538f67b05",
				"status": "missing"
			},
			{
				"rev": "3-0de77a37463bf391d14283e626831f2e",
				"status": "missing"
			},
			{
				"rev": "2-d795d1b924777732fdea76538c558b62",
				"status": "missing"
			},
			{
				"rev": "1-917fa2381192822767f010b95b45325b",
				"status": "missing"
			}
		</code>
]]--

function WFRxDocStoreImp:docRevisionHistory( docId, optRevId, optStartingFromAnyOf )

	check( 'WFRxDocStoreImp', 'string', '?+string', '?table' )

	assertNotClosed( self, 'docRevisionHistory' )

	local result = self._db:withTransactionDo( docRevisionHistory_inTransaction,
											   self, docId, optRevId, optStartingFromAnyOf )

	return Observable.just( result )

end

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

-- @mark local (non-syncing) documents

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the value of a single local (non-syncing) document.

	\param docId (string) the local document's unique ID

	\return <code>Observable< table ></code> value for this document

	Note that the document content will be annotated with the following
	system-generated entries (at the top level of the value table):

		* \c _empty (boolean): true if there isn't actually a value for this doc ID
]]--

function WFRxDocStoreImp:localDocValue( name )

	check( 'WFRxDocStoreImp', 'string' )

	assertNotClosed( self, 'localDocValue' )

	local localDocValueRow = self._exec( self._db,
										 [[SELECT content FROM localDocs WHERE name == ?]],
										 name )

	local docValue = localDocValueRow and contentFromFlattenedValue( localDocValueRow.content )

	if docValue then
		docValue._id = name
	end

	return Observable.just( docValue or { _id = name, _empty = true } )

end

--------------------------------------------------------------------------------
--[[!
	\brief Sets the value for a local (non-syncing) document, replacing any
		existing value.

	Local documents are intended for use as state bookmarks. They differ from
	traditional syncing documents in the following ways:

		* Local document IDs are stored in a separate table and a separate
		  namespace from traditional syncing documents. You may reuse the same
		  document ID among the document types and the values will not cross-pollute.
		* Local documents are not versioned. (For that reason, the \c _rev
		  parameter must not be specified in a local document value.)
		* Local documents can not have conflicts.
		* There is no change notification for local documents.
		* Local documents are not replicated to other document databases.

	\param value (table) new value for the entire document

	\c value \b must contain the following elements:
		* \c _id (string) the document's unique ID

	\return <code>Observable< empty ></code> Resolves when the write completes.
		(You may safely ignore this promise.)
]]--

function WFRxDocStoreImp:setLocalDocValue( docValue )

	check( 'WFRxDocStoreImp', 'table' )

	assertNotClosed( self, 'setLocalDocValue' )

	if self._db:hasActiveTransaction() then
		setLocalDocValueImp_inTransaction( self, self._db, self._exec, 'setLocalDocValue', docValue )
	else
		self._db:withTransactionDo( setLocalDocValueImp_inTransaction, self, self._db, self._exec, 'setLocalDocValue', docValue )
	end

	return Observable.just( true )

end

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

function WFRxDocStoreImp:applyOverlayLocalDocValue( overlay )

	check( 'WFRxDocStoreImp', 'table' )

	assertNotClosed( self, 'applyOverlayLocalDocValue' )

	local localDocId = assertDocIdIsValid( overlay._id )
	overlay._id = nil

	if overlay._revId then
		error( "WFRxDocStoreImp:applyOverlayLocalDocValue does not accept _revId", 2 )
	end

	if overlay._deleted then
		error( "WFRxDocStoreImp:applyOverlayLocalDocValue does not accept _deleted", 2 )
	end

	assertValidOverlay( overlay )

	local localDocValueRow = self._exec( self._db,
										 [[SELECT content FROM localDocs WHERE name == ?]],
										 localDocId )

	local docValue = localDocValueRow and contentFromFlattenedValue( localDocValueRow.content ) or {}

	WFTableMerge.applyOverlay( docValue, overlay )

	docValue._id = localDocId

	return self:setLocalDocValue( docValue )

end

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

-- @mark diagnostic log

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

local performIntegrityCheck -- forward declaration

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

local function getSingleResult( db, sql, ... )

	local rows = db:gatherArray( sql, ... )

	return rows and rows[ 1 ]

end

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

local function computeAveSize( db, sql )

	local rows = db:gatherArrayOfArray( sql )

	if #rows == 0 then

		return

	end

	return string.format( " %s (%s records)", unpack( rows[ 1 ] ) )

end

--------------------------------------------------------------------------------
--[[!
	\brief Sends diagnostic data on database performance to the log.
]]--

function WFRxDocStoreImp:sendDiagnosticLogData()

	check( 'WFRxDocStoreImp' )

	-- Send information about doc store performance.
	local aveAssetRevContentSize, aveAssetDocAnnotationSize, nDocs, nRevs
	if WFConfiguration.DocStore.DiagnosticLogDocStats then

		aveAssetRevContentSize = computeAveSize(
											self._db,
											[[SELECT  SUM( LENGTH ( content ) ) / COUNT( * ), COUNT(  * )
												FROM revs INNER JOIN docs
														ON docs.winningRevSequence = revs.sequence
												WHERE docs.deleted != 1 AND content LIKE '%type = "asset"%';]]
											)

		aveAssetDocAnnotationSize = computeAveSize(
											self._db,
											[[SELECT  SUM( LENGTH ( annotation ) ) / COUNT( * ), COUNT(  * )
												FROM revs INNER JOIN docs
														ON docs.winningRevSequence = revs.sequence
												WHERE docs.deleted != 1 AND content LIKE '%type = "asset"%';]]
											)

		nDocs = getSingleResult( self._db,
										 "SELECT count() FROM docs" )

		nRevs = getSingleResult( self._db,
										 "SELECT count() FROM revs" )


	end

	return {

		name = "WFRxDocStoreImp :: " .. self._path,
		id = 'WFRxDocStoreImp-' .. self._path,
		weight = 1600,
		parentSection = WFDiagnosticLog.Sections.DOC_STORE,

		body = {

			nDocs = nDocs,
			nRevs = nRevs,
			aveAssetRevContentSize = aveAssetRevContentSize,
			aveAssetDocAnnotationSize = aveAssetDocAnnotationSize,
			nTransactions = self._db:nTransactions(),
			bulkRequestInfo = self._bulkRequestInfo,
			integrityCheck = performIntegrityCheck( self ),
			typeAndSubtypeDiagnosticData = self._typeAndSubtypeDiagnosticData
		},

	}

end

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

-- @mark miscellaneous

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves the path to this document store.

	\return (string) The catalog's path.
]]--

function WFRxDocStoreImp:path()

	check( 'WFRxDocStoreImp' )

	return self._path

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves this document store as a string.

	\return (string) The string representation.
]]--

function WFRxDocStoreImp:__tostring()

	check( 'WFRxDocStoreImp' )

	return string_format( "WFRxDocStoreImp( %s )", self._path )

end

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

local function getDeletedDocs_inTransaction( self, from )

	local db = self._db

	local continueFrom = self._changeCounter:getValue() + 1

	local rawDocRows = db:gatherArrayOfArray(
		[[SELECT fullDocId, docs.localDocId, revId, mostRecentRevSequence,
				content, annotation, revs.sequence, spaceId
			FROM docs
				INNER JOIN revs ON docs.winningRevSequence == revs.sequence
			WHERE mostRecentRevSequence >= ?
				AND docs.deleted = 1
				AND current = 1
				AND hasConflicts = 0
			ORDER BY mostRecentRevSequence ]],
		from )

	local result = {}

	for i = 1, #rawDocRows do

		local currentRevRow = rawDocRows[ i ]

		local docValue = contentFromFlattenedValue( currentRevRow[ 5 ] )
		local annotation = contentFromFlattenedValue( currentRevRow[ 6 ] )

		for k, v in pairs( annotation ) do
			docValue[ k ] = v
		end

		docValue._id = currentRevRow[ 1 ]
		docValue._localId = currentRevRow[ 2 ]
		docValue._revId = currentRevRow[ 3 ]
		docValue._changeCounter = currentRevRow[ 4 ]
		docValue._spaceId = currentRevRow[ 8 ]
		docValue._deleted = true

		-- Merge in any labels that may be associated with this revision.
		-- (This is the same approach we use in docValue() and bulkDocValue())

		local labels = getLabels( db, currentRevRow[ 7 ] )

		if labels then
			docValue._labels = labels
		end

		result[ i ] = docValue

	end

	return result, continueFrom

end

--------------------------------------------------------------------------------
--[[!
	\brief Retrieves a stream of deleted doc values from a specific point of time,
		as provided by the change counter, onward.

	\param changeCounterName (string) The name of the changeCounter. If a change
			counter does not exist for the given name, the current changeCounter
			value will be used as a starting point.

	\param optFieldSpec (optional, table of arrays of strings) paths of fields to include.

	\return <code>Observable< table ></code> Fires with batches of doc values as documents are
		deleted.
]]--

function WFRxDocStoreImp:deletedDocsSinceNamedChangeCounter( changeCounterName, optFieldSpec )

	check( 'WFRxDocStoreImp', 'string', '?table' )

	-- Get all documents that have been deleted since the provided changeCounter.

	local bufferTime = WFPlatformInfo.isUnitTest() and 0.01 or 5
	local bufferSize = WFPlatformInfo.isUnitTest() and 1 or 100

	return self:localDocValue( changeCounterName )

		:flatMap( function( docValue )

			local changeCounter = KSTableUtils_derefOpt( docValue, "changeCounter" )

			return changeCounter
					and Rx.Observable.just( changeCounter )
					or self:changeCounterOnce()

		end )

		:flatMap( function( changeCounter )

			local previouslyDeletedDocs, continueFrom = self._db:withTransactionDo(
													getDeletedDocs_inTransaction,
													self, changeCounter )

			local numberOfPreviouslyDeletedDocsToProcess = #previouslyDeletedDocs

			if numberOfPreviouslyDeletedDocsToProcess > 100 then

				docStoreImpLog:warnf( "deletedDocsSinceNamedChangeCounter: The catch up stream contains %d records. This is a sign that we are lagging on deletion processing.", numberOfPreviouslyDeletedDocsToProcess )

			end

			local deletedDocsStream = Observable.from( previouslyDeletedDocs )

			-- New deletions will show up on the deletedDocStream from elsewhere in
			-- WFRxDocStoreImp
			-- Note that the the _deletedDocStream emits a tuple of < docValue, changeCounter >,
			-- while the query above just emits docValues, hence the optChangeCounter below

			:concat( self._deletedDocStream )

			:distinctUntilChanged()

			:map( function( docValue, optChangeCounter )

				if optFieldSpec then

					optFieldSpec._changeCounter = true
						-- This field is required.

					docValue = onlyFieldsInSpec( optFieldSpec, docValue )

				end

				if optChangeCounter then
					changeCounter = optChangeCounter
				end

				return docValue

			end )

			:bufferWithTimeOrCount( bufferTime, bufferSize, WFRxScheduler.sharedScheduler() )

			:map( function( docValues )

				local counter = changeCounter + 1

				if #docValues >= 1 then
					counter = docValues[ #docValues ]._changeCounter + 1
				end

				addToBulkRequestInfo( self, 'deletedDocsSinceNamedChangeCounter', changeCounterName, #docValues )

				return docValues, counter

			end )

			if numberOfPreviouslyDeletedDocsToProcess == 0 then

				return Rx.Observable.just( {}, continueFrom )

				:concat( deletedDocsStream )
			else

				return deletedDocsStream

		end

	end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Create or update a change counter associated with deleted documents.

	\param changeCounterName (string) The name of the changeCounter to update or
		create.

	\param newChangeCounter (number) The new change counter value.

	\return <code>Observable< true ></code> Completes with a true response if successful.
]]--

function WFRxDocStoreImp:updateDeletedDocsSinceNamedChangeCounter( changeCounterName, newChangeCounter )

	check( 'WFRxDocStoreImp', 'string', 'number' )

	return self:setLocalDocValue {
				_id = changeCounterName,
				changeCounter = newChangeCounter
			}

end

--------------------------------------------------------------------------------
--[[!
	\brief Remove synchronized revisions.

	Delete all revisions that:
    - have a label
    - but only those before both the earliest sequence of all labels and the winningRevSequence

	This will remove all revisions that have been successfully synced.

	\warning This function should be used cautiously as it may take significant
		time. All other doc store features will be queued up and will not
		complete until the compaction has completed.

	\return <code>Observable< affectedRows ></code> Resolves when compaction has completed.
		(You may safely ignore this Observable.)
]]--

function WFRxDocStoreImp:compact( progress, optVacuum )

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'compact' )

	local DOCS_PER_TRANSACTION = 200

	compactionLog:info( "Deleting synchronized revisions" )

	local exec = self._exec

	local foreign_keys = exec( self._db, [[ PRAGMA foreign_keys ]] )
	foreign_keys = foreign_keys and foreign_keys.foreign_keys == 1 or false

	if foreign_keys then
	 	exec( self._db, [[ PRAGMA foreign_keys = OFF ]] )
	end

	local nDocs = getSingleResult( self._db, [[ SELECT count() FROM docs ]] )

	local nRevs = getSingleResult( self._db, [[ SELECT count() FROM revs ]] )

	local docStmt = self._db:prepare( [[ SELECT localDocId, winningRevSequence FROM docs ]] )

	local count = 0
	local docsCleaned = 0
	local revisionsRemoved = 0
	-- NOTE [immisch 2019-02-18]: in the data collected so far, compact takes roughly as long
	-- as vacuum, so split 50:50
	local percentCompact = 50.0
	local scaleTo = optVacuum and percentCompact or 100.0

	if progress then
		progress( { percentComplete = 0.0, revisionsRemoved = 0 }  )
	end

	local doc = docStmt:step()

	exec( self._db, [[ BEGIN TRANSACTION ]] )

	-- we send up to 199 notifications, so we get at least one per percent completed
	local steps = math.max( math.floor(nDocs / 100), 1 )

	while doc and next( doc ) do

		local minRevSeq = getSingleResult( self._db,
			[[ SELECT MIN(revSequence) FROM labels
            		JOIN docs ON labels.localDocId = docs.localDocId
                WHERE docs.localDocId = ?
					AND substr(labels.name, 1, 1) != '_' ]],
			doc.localDocId )

		if minRevSeq then

			local minSequence = doc.winningRevSequence
			if minRevSeq < minSequence then
				minSequence = minRevSeq
			end

			-- unlink the parentRevSequence so we don't violate the foreign key constraints
			exec( self._db, [[ UPDATE revs SET parentRevSequence = NULL
				WHERE revs.sequence <= ? AND localDocId = ? ]],
				minSequence, doc.localDocId )

			exec(self._db, [[ DELETE FROM revs WHERE sequence < ? AND localDocId = ? ]], minSequence, doc.localDocId )

			revisionsRemoved = revisionsRemoved + self._db:getAffectedRowCount()
			docsCleaned = docsCleaned + 1

			-- update the parentRevSeq of the most recent surviving revision

			exec( self._db, [[ UPDATE revs SET parentRevSequence = ?
				WHERE sequence = ? ]],
				minSequence, minSequence )

			-- commit and open a new transaction every DOCS_PER_TRANSACTION
			if docsCleaned % DOCS_PER_TRANSACTION == 0 then
				exec( self._db, [[ COMMIT ]] )
				exec( self._db, [[ BEGIN TRANSACTION ]] )
			end

		end

        count = count + 1

		if count == nDocs or ( count > 0 and count % steps == 0 )  then

			local percent = count * scaleTo / nDocs

			if progress then
				progress( { percentComplete = percent, revisionsRemoved = revisionsRemoved } )
			end

			local logMsg = string.format( "Deleting synchronized revisions progress: %d%%", percent )
			compactionLog:info( logMsg )

		end

		doc = docStmt:step()

	end

	pcall( docStmt.finalize, docStmt )

	exec( self._db, [[ COMMIT ]] )

	if foreign_keys then
		exec( self._db, [[ PRAGMA foreign_keys = ON ]] )
	end

	local event = {
			['event.appWorkflow'] = 'DB',
			['event.type'] = 'info',
			['event.subtype'] = 'compact',
			['stats.revisionsRemoved'] = revisionsRemoved,
			['stats.nDocs'] = nDocs,
			['stats.nRevs'] = nRevs
		}

	if optVacuum then

		if progress then

			scaleTo = 100.0 - percentCompact

			local progressCount = 0

			-- this is the logic to combine two progress sequences into one
			local function vacuumProgress( value )

				local percent = value.percentComplete * (scaleTo / 100) + percentCompact

				progress( { percentComplete = percent, revisionsRemoved = revisionsRemoved } )

				progressCount = progressCount + 1

			end

			local pageCount = getSingleResult( self._db, 'PRAGMA page_count' )

			local freeList = getSingleResult( self._db, 'PRAGMA freelist_count' )

			self:vacuum( vacuumProgress, revisionsRemoved )

			event['stats.freelistCount'] = freeList
			event['stats.pageCount'] = pageCount
			event['stats.progressCount'] = progressCount

		else
			self:vacuum()
		end

	end

	compactionLog:info( 'Deleting synchronized revisions complete' )

	WFAnalytics.trackEvent( event )

end

--------------------------------------------------------------------------------
--[[!
	\brief Estimate the number of calls to the progress handler while vacuum is running

	The coefficients of this polynomial were determined using linear regression
	on a set of docStores collected from mobile (iOS) and desktop QE.

	For the calculations, see stats.ipynb in Thio/oz-shell (branch immisch/doctransform)

	\return The estimated number of progress calls while vacuum is running.
]]--

local function estimateVacuumProgressCalls( db, revsCleared )

	revsCleared = revsCleared or 0.0

	local nDocs = getSingleResult( db, 'SELECT count() FROM docs' )

	local nRevs = getSingleResult( db, 'SELECT count() FROM revs' )

	local pageCount = getSingleResult( db, 'PRAGMA page_count' )

	local freeList = getSingleResult( db, 'PRAGMA freelist_count' )

	local progressCalls =  -1.46615567e-04 * freeList
		+ 1.76830518e-10 * freeList * freeList
		- 1.89372000e-04 * pageCount - 1.79882802e-01 * revsCleared + 3.10098331e-01 * nDocs
		+ 1.80047194e-01 * nRevs

	if progressCalls < 0 then
		progressCalls = 2
	else
		-- pessimize the result a bit. The user experience is better if we are quicker than expected
		progressCalls = math.floor( progressCalls * 1.1 )
	end

	return progressCalls
end

--------------------------------------------------------------------------------
--[[!
	\brief Free disk space.

	\warning This function should be used cautiously as it may take significant
		time. All other doc store features will be queued up and will not
		complete until the vacuum has completed.

	\return <code>nil< affectedRows ></code> Resolves when vacuum has completed.
		(You may safely ignore this Observable.)
]]--

function WFRxDocStoreImp:vacuum( progress, revsCleared )

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'vacuum' )

	local path = self._path
	if path and path ~= ':temp:' and path ~= ':memory:' and path ~= '' then

		-- Close and reopen the database.

		maybeCloseAndClearDatabase( self )

		local db = openAndObtainDatabase( self, path )

		compactionLog:info( "Flushing SQLite WAL" )

		db:exec( [[PRAGMA wal_checkpoint( RESTART )]] )

		local expectedProgressCalls = estimateVacuumProgressCalls( db, revsCleared )

		local progressCalls = 0
		local lastPercent = 0

		local sqliteProgress = function()

			progressCalls = progressCalls + 1

			local percent = progressCalls * 100.0 / expectedProgressCalls
			if percent > 100 then
				percent = 100
			end

			-- Throttle the output
			if percent - lastPercent >= 0.5 then
				if progress then
					progress( { percentComplete = percent })
				end

				local logMsg = string.format( "Vacuum progress: %d%%", percent )
				compactionLog:info( logMsg )
			end

			lastPercent = percent

			return 0

		end

		db:setProgressCallback( sqliteProgress, 100 )

		if progress then
			progress( { percentComplete = 0.0 })
		end

		compactionLog:info( "Vacuuming SQLite database" )

		db:exec( [[VACUUM]] )

		if lastPercent < 100.0 then

			if progress then
				progress( { percentComplete = 100.0 } )
			end

			compactionLog:info( "Vacuum complete" )

		end

		db:setProgressCallback( nil )

	end

	return Observable.just()

end

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

local function dumpDocsImp_inTransaction( self, docIds)

	local db = self._db

	local exec = self._exec

	-- add docIds to a TEMP_ids table so we can join on it later
	exec( db, [[CREATE TEMPORARY TABLE TEMP_ids( docId PRIMARY KEY )]] )

	local n = #docIds
	for i = 1, n do
		-- If we are doing a lot of these, we may want to consider batching
	    exec( db, "INSERT INTO Temp_Ids SELECT ?", docIds[ i ] )
  	end

	-- Turn off foreign key check in this transaction so that we can
	-- batch copy the data from the revs/docs/labels tables without the
	-- bidirection fk constraints between the revs and docs tables causing
	-- issues when doing the bulk inserts.
	exec( db, [[PRAGMA defer_foreign_keys=ON]] )

	-- revs, docs tables tend to be  particularly large and so
	-- for those we'll select only those records in the table that
	-- match the docIds.
	exec( db, [[INSERT INTO dumpDb.revs
					SELECT revs.* FROM revs
					JOIN TEMP_ids ON localDocId = TEMP_ids.docId]] )
	exec( db, [[INSERT INTO dumpDb.docs
					SELECT docs.* FROM docs
					JOIN TEMP_ids ON localDocId = TEMP_ids.docId]] )

	-- labels isn't that large, but we only need to bring over those
	-- labels that relate to the revs/docs entries
	exec( db, [[INSERT INTO dumpDb.labels
					SELECT labels.* FROM labels
					JOIN TEMP_ids ON localDocId = TEMP_ids.docId]] )

	-- copy the full localDocs table over, it's small and there are no
	-- foreign key constraints that will be violated by taking the whole
	-- table
	exec( db, [[INSERT INTO dumpDb.localDocs SELECT * FROM localDocs]] )

	-- clean up the TEMP_ids table
	exec( db, [[DROP TABLE TEMP_ids]] )

end

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

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

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

	check( 'WFRxDocStoreImp', '+string' )

	assertNotClosed( self, 'onlineBackup' )

	local WFSQLiteDatabaseUtils = require 'WFSQLiteDatabaseUtils'

	return WFSQLiteDatabaseUtils.onlineBackupDbToPathWithProgressStream( self._db, path )

end


--------------------------------------------------------------------------------
--[[!
	\brief Dumps the doc store data for multiple documents to a new db at the
	specified path. This does not guarantee a fully consistent db is created at
	the resulting path and so it should only be used for troubleshooting/diagnostic
	purposes.

	\param docIds (table) array of local docIds

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

	\return <code>Observable<string> Resolves when complete with the path at which
		the db was created.
]]--
function WFRxDocStoreImp:dumpDocsToDbAtPath( docIds, path )

	check( 'WFRxDocStoreImp', 'table', '+string' )

	assertNotClosed( self, 'dumpDocsToDbAtPath' )

	-- create the new db with the doc store schema
	local newDb = WFSQLiteDatabase( path, schema )

	-- Set journal mode to DELETE to prevent -shm and -wal files
	newDb:exec( [[PRAGMA journal_mode = DELETE]] )

	newDb:close()

	-- now with the current db, attach to it
	local success, _ = pcall( self._exec, self._db, [[ATTACH DATABASE ? AS dumpDb]], path )
	if not success then
		return Observable.throw( "Failed to attach to database to dump documents" )
	end

	-- pcall this so that, even if it fails, we will turn the foreign_keys back on
	pcall( self._db.withTransactionDo, self._db, dumpDocsImp_inTransaction, self, docIds )

	self._exec( self._db, [[DETACH DATABASE 'dumpDb']] )

	-- Clean up any existing shm and wal files. This is likely a no-op, but
	-- it's possible that a file was left behind by a previous dump. Also,
	-- in the call to WFSQLiteDatabase init, the journal mode is set to WAL;
	-- afterwards we are setting it to DELETE right away. But it's possible
	-- that in the interem an empty file was left behind.

	pcall( WFFileUtils.deleteFile, path .. '-shm' )
	pcall( WFFileUtils.deleteFile, path .. '-wal' )

	return Observable.just( path )

end

--------------------------------------------------------------------------------
--[[!
	\brief Close the underlying sqlite db

	\warning This function should only be used at shutdown because all other
		docStore features that require an open db will fail. This applies to
		all threads that have a docStoreImp instance

	\return <code>Observable< empty ></code> Resolves when close has completed.
		Caller must subscribe to this Observable or closeDb call might not run.
]]--

function WFRxDocStoreImp:closeDb()

	check( 'WFRxDocStoreImp' )

	maybeCloseAndClearDatabase( self )

	return Observable.empty()

end

--------------------------------------------------------------------------------
--[[!
	\brief (Intended for debugging use only) Generates code coverage statistics
		for the worker thread that implemented this doc store.

	\warning This function should only be used by WFRxDocStoreTest.lua.

	\return <code>Observable< empty ></code> Resolves when compaction has completed.
		generated. (You may safely ignore this Observable.)
]]--

WFRxDocStoreImp.DEBUG_dumpCoverage = function( _, path )

	check( 'WFRxDocStoreImp', '+string' )

	require 'SetupCoverage'.dumpCoverage( path )

	return Observable.empty()

end

--------------------------------------------------------------------------------
--[[!
	\brief (Intended for performance testing only) Enables a mode where the doc
		store will collect performance statistics on its SQLite database.

	\return <code>Observable< empty ></code> Resolves when compaction has completed.
		been enabled. (You may safely ignore this Observable.)
]]--

function WFRxDocStoreImp:PERF_collectQueryPerformanceStats()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'PERF_collectQueryPerformanceStats' )

	self._db:PERF_collectQueryPerformanceStats()

	return Observable.empty()

end

--------------------------------------------------------------------------------
--[[!
	\brief (Intended for performance testing only) Generates a string suitable
		for logging to a text file or window describing performance results.

	\return <code>Observable< string ></code> query statistics report
]]--

function WFRxDocStoreImp:PERF_generateQueryStatsReport()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'PERF_generateQueryStatsReport' )

	return Observable.just( self._db:PERF_generateQueryStatsReport() )

end

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

--[[
local function makeIntegrityCheckFunction( expectedEmptyCountQuery, messageFormat )

	expectedEmptyCountQuery = "SELECT count(*) FROM " .. expectedEmptyCountQuery

	return function( db, exec )

		local _, count = next( exec( db, expectedEmptyCountQuery ) )

		if count == 0 then
			return true
		else
			return false, string.format( messageFormat, count )
		end

	end

end
]]--

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

local integrityCheckers = {

	-- TODO [scouten 2015-05-06]: Update with current integrity checks.
	-- We're not using hash any more, so these are not relevant.

--	makeIntegrityCheckFunction( [[bodyContent WHERE bodyContentHash NOT
--		IN ( SELECT bodyContentHash FROM documents )]], "orphaned bodyContent: %d" ),
--
--	makeIntegrityCheckFunction( [[annotationContent WHERE annotationContentHash NOT
--		IN (SELECT annotationContentHash FROM documents)]], "orphaned annotationContent: %d" ),
--
--	makeIntegrityCheckFunction( [[documents WHERE bodyContentHash NOT
--		IN (SELECT bodyContentHash FROM bodyContent)]], "dangling bodyContent: %d" ),
--
--	makeIntegrityCheckFunction( [[documents WHERE annotationContentHash NOT
--		IN (SELECT annotationContentHash FROM annotationContent)]], "dangling annotationContent: %d" ),

}

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

performIntegrityCheck = function( self )

	local failures = {}

	for _, checker in ipairs( integrityCheckers ) do

		local result, message = checker( self._db, self._exec )

		if not result then
			failures[ #failures + 1 ] = message
		end

	end

	return failures

end

--------------------------------------------------------------------------------
--[[!
	\brief (Intended for unit test use only) Perform an integrity check
		of the doc store and report errors.

	\warning This function should only be used by unit tests.

	\return <code>Observable</code> signals onCompleted if successful;
		signals onError with an array of failure messages if any problems are found
]]--

function WFRxDocStoreImp:performIntegrityCheck()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'performIntegrityCheck' )

	local result = performIntegrityCheck( self )

	if #result == 0 then
		return Observable.empty()
	else
		return Observable.throw( result )
	end

end

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

function WFRxDocStoreImp:halt()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'halt' )

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

end

--------------------------------------------------------------------------------
--[[!
	\brief Permanently shuts down this doc store and deletes all related files.

	\return <code>Observable< empty ></code> Resolves when done.
		(You may safely ignore this Observable.)
]]--

function WFRxDocStoreImp:haltAndDelete()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'haltAndDelete' )

	local path = self._path

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

	if path and #path > 0 then
		WFFileUtils.deleteFile( path )
		WFFileUtils.deleteFile( path .. '-shm' )
		WFFileUtils.deleteFile( path .. '-wal' )
	end

	return Observable.empty()

end

--------------------------------------------------------------------------------
--[[!
	\brief Get a stream that fires with SQLite error codes when a doc 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 WFRxDocStoreImp:getSQLiteFatalErrorCodeStream()

	check( 'WFRxDocStoreImp' )

	assertNotClosed( self, 'getSQLiteFatalErrorCodeStream' )

	return self._db:getSQLiteFatalErrorCodeStream()

end

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

function WFRxDocStoreImp:_DEBUG_simulateCorruption()

	check( 'WFRxDocStoreImp' )

	self._db:_DEBUG_simulateCorruption()

end

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

function WFRxDocStoreImp:_DEBUG_bulkRequestInfoStream( refreshSeconds )

	check( 'WFRxDocStoreImp', 'number' )

	return Observable.timer( 0, refreshSeconds, WFRxScheduler.sharedScheduler() )
		:map( function()
			return self._bulkRequestInfo
		end )

end

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

function WFRxDocStoreImp:_DEBUG_clearIdCache()

	check( 'WFRxDocStoreImp' )

	self._localIdToDocIdCache = {}
	self._docIdToLocalIdCache = {}

	return Observable.empty()
end

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

local _perfWarningThreshold = 1 / 10 -- seconds

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

-- SPAM [jandrew 2017-05-24]: The following metrics are hidden behind config flags.
-- They're usually too noisey and result in very large stamina logs.

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

local _apisThatRequireExplicitStaminaAllowlist = {
		'changedDocs',
		'changeCounter' }

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

local function isAPIBlockedFromStamina( apiName )

	if not _apisThatRequireExplicitStaminaAllowlist[ apiName ] then

		return false

	end

	-- the API must have a config.lua entry in order to be enabled

	if WFConfiguration.Stamina[ string.format( "%sMetrics", apiName ) ] then

		return false

	end

	return true

end

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

local function monitorPerformanceImp( apiName, descFunc )

	-- Logger Warnings

	WFRxDocStoreImp[ apiName ] =
		WFPerformanceUtils_wrapInLogLongExecutionInDiagnosticLogWithDynamicDescription(
			"DocStore", _perfWarningThreshold, apiName, descFunc,
				WFRxDocStoreImp[ apiName ] )

	if isAPIBlockedFromStamina( apiName ) then

		return WFRxDocStoreImp[ apiName ]

	end

	-- Stamina Monitoring

	return WFPerformanceReport.wrapObservableFunction(
				'WFRxDocStoreImp:' .. apiName,
				WFRxDocStoreImp[ apiName ] )

end

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

local function justReturn( v )

	return v

end

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

local function monitorPerformance1( apiName )

	return monitorPerformanceImp( apiName, justReturn )

end

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

local function makeArrayCountDesc( desc, _, array )

	return string.format( "%s( %s )",
							desc,
							type( array ) == "table" and
								#array or
								"not an array!" )

end

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

local function monitorPerformanceN( apiName )

	return monitorPerformanceImp( apiName, makeArrayCountDesc )

end

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

WFRxDocStoreImp.localDocValue = monitorPerformance1( 'localDocValue' )

WFRxDocStoreImp.applyOverlayLocalDocValue = monitorPerformance1( 'applyOverlayLocalDocValue' )

WFRxDocStoreImp.setLocalDocValue = monitorPerformance1( 'setLocalDocValue' )

WFRxDocStoreImp.changedDocs = monitorPerformance1( 'changedDocs' )

WFRxDocStoreImp.changeCounter = monitorPerformance1( 'changeCounter' )

WFRxDocStoreImp.bulkApplyOverlay = monitorPerformance1( 'bulkApplyOverlay' )

WFRxDocStoreImp.applyOverlay = monitorPerformance1( 'applyOverlay' )

WFRxDocStoreImp.deleteDoc = monitorPerformance1( 'deleteDoc' )

WFRxDocStoreImp.deleteRevisionAsNonCurrent = monitorPerformance1( 'deleteRevisionAsNonCurrent' )

WFRxDocStoreImp.deleteRevisionAsNonCurrentAndApplyOverlay = monitorPerformance1( 'deleteRevisionAsNonCurrentAndApplyOverlay' )

WFRxDocStoreImp.bulkSetDocValue = monitorPerformanceN( 'bulkSetDocValue' )

WFRxDocStoreImp.setDocValue = monitorPerformance1( 'setDocValue' )

WFRxDocStoreImp.bulkDocValue = monitorPerformanceN( 'bulkDocValue' )

WFRxDocStoreImp.docValue = monitorPerformance1( 'docValue' )

WFRxDocStoreImp.deletedDocsSinceNamedChangeCounter = monitorPerformance1( 'deletedDocsSinceNamedChangeCounter' )

WFRxDocStoreImp.updateDeletedDocsSinceNamedChangeCounter = monitorPerformance1( 'updateDeletedDocsSinceNamedChangeCounter' )

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

return Rx.Class 'WFRxDocStoreImp'( WFRxDocStoreImp )
