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

NmbFileAccessDatabase.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.
s
------------------------------------------------------------------------------]]

--[[
	AdobePatentID="P7167-US"
]]

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

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

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local KSDate = require 'KSDate'
local WFCloseSupport = require 'WFCloseSupport'
local WFRxScheduler = require 'WFRxScheduler'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

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

local check = assert( WFArguments.check )

local logger = WFLogger 'NmbFileAccessDatabase'

local KSDate_timeNow = assert( KSDate.timeNow, "KSDate.timeNow does not exist" )

local ACCESS_QUERY_DEFAULT_LIMIT = 1000

local TOUCH_CACHE_COMMIT_THRESHOLD = 1000

local runUntilDone = WFRxRunUntilDone

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

local kNullState = 'null'

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

local function fromBool( bool )

	if bool == nil then
		return nil
	end

	return bool and 1 or 0

end

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

-- @mark NmbFileAccessDatabase namespace

--[[!

	\class NmbFileAccessDatabase

	\brief The file access database is a traditional persistent SQLite database which
	can be used to track binary data access on disk. This database can be easily queried to
	identify candidates for purging when we need to make more disk space available.
	
--]]

local NmbFileAccessDatabase = {

	_docStore = false,
		-- (WFRxDocStore)

	_mruSQLiteDatabase = false,
		-- (WFSQLiteDatabase)
	
	_touchCache = false,
		-- (table) Used to track updates that haven't yet been written to the db.
		-- key: local id
		-- value: table {
		--    touchDate = <timestamp>
		--    hasProxy = <boolean>
		--    hasOriginal = <boolean>
		-- }
		
	_cacheSize = false,
		-- (number) The number of entries in the touch cache. If the cache grows too large,
		-- we commit the data to the db.
		
	_writeStreamSubscription = false,
		-- ( Rx.Subscription )
		
	_changeCounter = false,
		-- (number) The local change counter.
		
	_changeCounterStream = false,
		-- ( Rx.ReplaySubject ) Fires with the local change counter.
		
	_lastTempIds = false,
		
	_isDebugMode = false,
		-- (boolean) If true, updates to assets are immeadiately commited to db.
		
	WFCloseSupport.addCloseSupportMethods,	
	
}

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

local function commitTouchTimesToDb( self )

	if not next( self._touchCache ) then
		return
	end

	if WFCloseSupport.hasBeenOrWillBeClosed( self ) then
		return
	end

	local sql = 'SAVEPOINT updateLastAccess'

	self._mruSQLiteDatabase:exec( sql )

	local statement

	for localId, entry in pairs( self._touchCache ) do
	
		if type( entry ) == 'table' then
			
			if entry.delete then
		
				statement = [[DELETE FROM lastAccessTimestamps WHERE localId = ?]]
				self._mruSQLiteDatabase:exec( statement, localId )
							
				statement = [[DELETE FROM assetInfo WHERE localId = ?]]
				self._mruSQLiteDatabase:exec( statement, localId )
							
			else
		
				-- We're not using INSERT OR REPLACE in the below SQL because we want to 
				-- maintain the values of the other fields in the row; REPLACE would squash them.
		
				if entry.newEntryTouchDate then
				
					statement = [[INSERT OR IGNORE INTO lastAccessTimestamps( localId, accessTimestamp ) VALUES( ?, ? )]]
					self._mruSQLiteDatabase:exec( statement, localId, entry.newEntryTouchDate )
						-- Insert a new entry into the DB if it does't exist.
						
					statement = [[INSERT OR IGNORE INTO assetInfo( localId ) VALUES( ? )]]
					self._mruSQLiteDatabase:exec( statement, localId )
						-- If the asset doesn't exist in the assetInfo table, insert it.
						
				end
			
				if entry.touchDate then
				
					statement = [[INSERT OR IGNORE INTO lastAccessTimestamps( localId, accessTimestamp ) VALUES( ?, ? )]]
					self._mruSQLiteDatabase:exec( statement, localId, entry.touchDate )
						-- Insert a new entry into the DB if it does't exist.
						
					statement = [[INSERT OR IGNORE INTO assetInfo( localId ) VALUES( ? )]]
					self._mruSQLiteDatabase:exec( statement, localId )
						-- If the asset doesn't exist in the assetInfo table, insert it.
					
					statement = [[UPDATE lastAccessTimestamps SET accessTimestamp = ? WHERE localId = ?]]
					self._mruSQLiteDatabase:exec( statement, entry.touchDate, localId )
						-- If the entry exists, update the touch date. 
						
				end
				
				local currentTime = KSDate_timeNow()
				
				for field, value in pairs( entry.simple or {} ) do
					
					statement = [[INSERT OR IGNORE INTO lastAccessTimestamps( localId, accessTimestamp ) VALUES( ?, ? )]]
					self._mruSQLiteDatabase:exec( statement, localId, currentTime )
						-- If the asset doesn't exist in the lastAccessTimestamps table, insert it.
						
					statement = [[INSERT OR IGNORE INTO assetInfo( localId ) VALUES( ? )]]
					self._mruSQLiteDatabase:exec( statement, localId )
						-- If the asset doesn't exist in the assetInfo table, insert it.
						
					if value == kNullState then
						statement = string.format( [[UPDATE assetInfo SET %s = NULL WHERE localId = ?]], field )
						self._mruSQLiteDatabase:exec( statement, localId )
					else
						statement = string.format( [[UPDATE assetInfo SET %s = ? WHERE localId = ?]], field )
						self._mruSQLiteDatabase:exec( statement, value, localId )
					end
					
				end
			
			end
			
		end
		
	end
	
	local continueFromChangeCounter = self._touchCache[ 'changeCounter' ]
	
	if continueFromChangeCounter then
		self._mruSQLiteDatabase:exec( 
				[[UPDATE tableStatus SET needsUpdateFrom = ? WHERE tableName = 'assetInfo']], continueFromChangeCounter )
	end
	
	self._touchCache = {}
	self._cacheSize = 0
	
	sql = 'RELEASE SAVEPOINT updateLastAccess'
	
	self._mruSQLiteDatabase:exec( sql )

end

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

local function maybeResetTable( self, tableName, sqlSchemaHash )

	-- Similar to the index store, we track a hash of the sql schema used
	-- to create the tables in the MRU. If the hash of the sql schema changes,
	-- we dump the data and regather it.

	local db = self._mruSQLiteDatabase

	db:exec( [[ DROP TABLE IF EXISTS lastAccessV2 ]] )
	db:exec( [[ DROP TABLE IF EXISTS lastAccessV3 ]] )
		-- SPAM [jandrew 2017-10-02]: These tables were used prior to switching to
		-- the new sql schema hash approach for table versioning, so we should 
		-- manually delete them if they happen to exist.

	local doesLastAccessExist = self:gatherArray( 
			[[SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'lastAccess' ]] )
			
	if doesLastAccessExist and doesLastAccessExist[ 1 ] then
	
		-- SPAM [jandrew 2019-10-31]: On this date we switched to only storing the timestamp
		-- in the lastAccessTimestamps table. The other data stored within lastAccess
		-- is now stored within the assetInfo table. This allows us to add new fields to
		-- assetInfo without losing our timestamps when the data is repopulated.
		
		db:exec( [[ BEGIN TRANSACTION ]] )

		db:exec( [[ INSERT OR REPLACE INTO lastAccessTimestamps( localId, accessTimestamp )
					  SELECT localId, accessTimestamp
					  FROM lastAccess ]] )

		db:exec( [[ DROP TABLE IF EXISTS lastAccess ]] )

		db:exec( [[ COMMIT ]] )
	
	end

	local rows = self:gatherArray( 
			[[SELECT sqlSchemaHash 
				FROM tableStatus
				WHERE tableName = ?]], 
			tableName )

	local cachedSqlSchemaHash  = rows[ 1 ]
	
	if sqlSchemaHash ~= cachedSqlSchemaHash then

		local sql = 'SAVEPOINT resetTable'

		db:exec( sql )
		
		db:exec( [[DELETE FROM tableStatus WHERE tableName == ?]],
				tableName )
		
		db:exec( [[INSERT INTO tableStatus VALUES( ?, ?, ? )]],
				tableName,
				0,
				sqlSchemaHash )
		
		db:exec( string.format( [[DELETE FROM %s]], tableName ) )
		
		sql = 'RELEASE SAVEPOINT resetTable'
			
		db:exec( sql )
			
	end

	local changeCounter = self:gatherArray( 
			[[SELECT needsUpdateFrom FROM tableStatus WHERE tableName = 'assetInfo']] )[ 1 ] or 0
					
	self._changeCounterStream:onNext( changeCounter )

end

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

--[[!

	\brief Creates a persistent SQLite database for tracking binary data access.

	\param dbPath (string) The path to the db.

	\return NmbFileAccessDatabase instance object
	
--]]


function NmbFileAccessDatabase:__init( docStore, mruSQLiteDatabase, sqlSchemaHash, optOverrides )

	check( 'NmbFileAccessDatabase', 'WFRxDocStore', 'WFSQLiteDatabase', '+string', '?table' )

	logger:debugf( "Creating NmbFileAccessDatabase" )

	optOverrides = optOverrides or {}

	self._isDebugMode = optOverrides.isDebugMode
	
	self._mruSQLiteDatabase = mruSQLiteDatabase
	
	self._docStore = docStore
	
	self._changeCounter = 0
	
	self._changeCounterStream = Rx.ReplaySubject( 1 )
	
	self._touchCache = {}
	
	self._cacheSize = 0
	
	maybeResetTable( self, 'assetInfo', sqlSchemaHash )

	self._writeStreamSubscription = Observable.timer( 0, optOverrides.writeInterval or 60, WFRxScheduler.sharedScheduler() )
		:subscribe(
			function()
				commitTouchTimesToDb( self )
			end )
	
end

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

--[[!

	\brief Closes the database.
	
	\return void
	
--]]

function NmbFileAccessDatabase:close()
										
	check( 'NmbFileAccessDatabase' )
	
	local writeStreamSubscription = self._writeStreamSubscription
	self._writeStreamSubscription = nil
	
	if writeStreamSubscription then
		writeStreamSubscription:dispose()
	end
	
	-- Clear the cache
	commitTouchTimesToDb( self )
	
	local mruSQLiteDatabase = self._mruSQLiteDatabase
	self._mruSQLiteDatabase = nil
	
	if mruSQLiteDatabase then
		mruSQLiteDatabase:close()
	end
	
	local changeCounterStream = self._changeCounterStream
	self._changeCounterStream = nil
	
	if changeCounterStream then
		changeCounterStream:dispose()
	end
	
end

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

function NmbFileAccessDatabase:getContinueFromChangeCounter()

	check( 'NmbFileAccessDatabase' )

	return self._changeCounterStream
	
		:take( 1 )
	
end

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

local function maybeCommitTouchTimesToDb( self )

	if self._cacheSize > TOUCH_CACHE_COMMIT_THRESHOLD or self._isDebugMode then
		commitTouchTimesToDb( self )
	end
		
end

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

local function addToTouchCache( self, localIdOrDocId, key, value, maybeSimpleUpdate )

	logger:debugf( "Updating %s status of asset %s to %s.", key, localIdOrDocId, value )

	local function addLocalIdToTouchCache( localId )
	
		local entry = self._touchCache[ localId ]
				
		if not entry then
		
			self._cacheSize = self._cacheSize + 1
			entry = {}
		
		end
		
		if maybeSimpleUpdate then
		
			local simpleUpdates = entry[ 'simple' ] or {}
			simpleUpdates[ key ] = value or kNullState
			entry[ 'simple' ] = simpleUpdates
			
		else
		
			entry[ key ] = value	
		
		end
		
		self._touchCache[ localId ] = entry
		
		maybeCommitTouchTimesToDb( self )

	end
	
	if type( localIdOrDocId ) == 'number' then
		
		addLocalIdToTouchCache( localIdOrDocId )
		
	else

		self._docStore:docIdToLocalId( localIdOrDocId )
	
			:map( function( localId )
				
				addLocalIdToTouchCache( localId )
				
			end )
			
			:call( runUntilDone )
		
	end
	
	return Observable.empty()

end

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

--[[!

	\brief Add a new entry, or update an existing entry in the db.

	\param localId (number) Local ID of the asset on disk.
	\param optDate (timestamp) optionally override the last access date.
	
	\return <code>Observable< nil ></code> that completes on success. There's
  		no need to subscribe to this observable.
	
--]]

function NmbFileAccessDatabase:touch( localIdOrDocId, optDate )

	check( 'NmbFileAccessDatabase', 'number|string', '?number' )

	local touchDate = optDate or KSDate_timeNow()

	return addToTouchCache( self, localIdOrDocId, 'touchDate', touchDate )
	
end

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

--[[!

	\brief Add multiple new entries, or ignore the updates if they already exists.

	\param localIdsOrDocIds (table) table of local ids or doc ids
	\param optDate (timestamp) optionally override the last access date.
	
	\return <code>Observable< nil ></code> that completes on success. There's
  		no need to subscribe to this observable.
	
--]]

function NmbFileAccessDatabase:touchMultipleIfNew( localIdsOrDocIds, optDate )

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

	logger:debugf( "touchIfNew for %d assets.", #localIdsOrDocIds )
	
	if #localIdsOrDocIds == 0 then
		return Observable.empty()
	end
	
	local touchDate = optDate or KSDate_timeNow()

	local localIdStream = Observable.just( localIdsOrDocIds )
		
	if type( localIdsOrDocIds[ 1 ] ) == 'string' then
		
		localIdStream = self._docStore:bulkDocIdToLocalIdAsMap( localIdsOrDocIds )
			:tap( function( map )
				for _, localId in pairs( map ) do
					addToTouchCache( self, localId, 'newEntryTouchDate', touchDate )
				end
			end )
	
	else
	
		for _, localId in pairs( localIdsOrDocIds ) do
			addToTouchCache( self, localId, 'newEntryTouchDate', touchDate )
		end
	
	end
		
	return localIdStream
		:takeLast( 0 )
		:call( runUntilDone )
		
end

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

--[[!

	\brief Update the change counter for this DB.

	\param changerCounter (number) The change counter at the time of the update.
	
	\return <code>Observable< nil ></code> that completes on success. There's
  		no need to subscribe to this observable.
	
--]]

function NmbFileAccessDatabase:updateChangeCounter( changeCounter )

	check( 'NmbFileAccessDatabase', 'number' )

	local key = 'changeCounter'
	
	local newChangeCounter = math.max( self._changeCounter, changeCounter )
	
	self._changeCounter = newChangeCounter
	self._changeCounterStream:onNext( newChangeCounter )
	
	self._touchCache[ key ] = newChangeCounter

	return Observable.empty()
	
end

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

--[[!

	\brief Update an asset's info in the MRU db.

	\param assetInfo (table) A table consisting of fields to update in the MRU db:
	
		_localId: The local ID Of the asset to update. (required)
		docId: The doc ID of the asset to update.
		isImportant: True if the asset is important. False otherwise.
		ozHasOriginal: True if there's an original on Oz.
		ozHasProxy = True of there's a proxy on Oz.
		proxyFileSize = The size of the local proxy, if it exists locally.
		originalFileSize = The size of the original.
		xmpSidecarSize = The size of the xmp sidecar, if it exists locally.
		volumeGuidUsedByProxy = The volume GUID containing the local proxy.
		volumeGuidUsedByOriginal = aThe volue GUID containing the local original.
		isMarkedForOfflineUse = True if the asset is marked for offline use.
		sufficientRenditionFoundOnOz: True if Oz contains a sufficiently large rendition.
		
		NOTE: To set a field to NULL, you can pass in the string 'null' as the value.
	
	\return <code>Observable< nil ></code> that completes on success. There's
  		no need to subscribe to this observable.
	
--]]


function NmbFileAccessDatabase:updateAssetInfo( assetInfo )

	check( 'NmbFileAccessDatabase', 'table' )

	local localId = assetInfo._localId

	local entry = self._touchCache[ localId ]
			
	if not entry then
	
		self._cacheSize = self._cacheSize + 1
		entry = {}
	
	end
	
	local simpleUpdates = entry[ 'simple' ] or {}
	
	simpleUpdates.docId = assetInfo.docId
	simpleUpdates.isImportant = fromBool( assetInfo.isImportant )
	simpleUpdates.ozHasOriginal = fromBool( assetInfo.ozHasOriginal )
	simpleUpdates.ozHasProxy = fromBool( assetInfo.ozHasProxy )
	simpleUpdates.proxyFileSize = assetInfo.proxyFileSize
	simpleUpdates.originalFileSize = assetInfo.originalFileSize
	simpleUpdates.xmpSidecarSize = assetInfo.xmpSidecarSize
	simpleUpdates.volumeGuidUsedByProxy = assetInfo.volumeGuidUsedByProxy
	simpleUpdates.volumeGuidUsedByOriginal = assetInfo.volumeGuidUsedByOriginal
	simpleUpdates.isMarkedForOfflineUse = fromBool( assetInfo.isMarkedForOfflineUse )
	simpleUpdates.sufficientRenditionFoundOnOz = fromBool( assetInfo.sufficientRenditionFoundOnOz )
	
	entry[ 'simple' ] = simpleUpdates
	
	self._touchCache[ localId ] = entry
	
	maybeCommitTouchTimesToDb( self )

end

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

--[[!

	\brief Delete an asset from the database for a given local Id.

	\param localIdOrDocId (number|string) localId or docId of the asset.
	
	\return <code>Observable< nil ></code> that completes on success. There's
  		no need to subscribe to this observable.
	
--]]

function NmbFileAccessDatabase:delete( localIdOrDocId )

	check( 'NmbFileAccessDatabase', 'number|string' )
	
	logger:debugf( "Deleting asset %s.", localIdOrDocId )
	
	return addToTouchCache( self, localIdOrDocId, 'delete', true )
	
		:takeLast( 0 )
	
		:call( runUntilDone )
	
end

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

local function convertSqlRowsToUsefulLuaTables( arr )

	local usefulArr = {}

	for idx, row in pairs( arr ) do

		local localId = row[ 1 ]
		local lastAccessed = row[ 2 ]
		local docId = row[ 3 ]
		local volumeGuidUsedByProxy = row[ 4 ]
		local ozHasProxy = row[ 5 ]
		local proxyFileSize = row[ 6 ]
		local volumeGuidUsedByOriginal = row[ 7 ]
		local ozHasOriginal = row[ 8 ]
		local originalFileSize = row[ 9 ]
		local xmpSidecarSize = row[ 10 ]
		local isImportant = row[ 11 ]
		local isMarkedForOfflineUse = row[ 12 ]
		local sufficientRenditionFoundOnOz = row[ 13 ]

		usefulArr[ idx ] = {
			localId = localId,
			docId = docId,
			lastAccessed = lastAccessed,
			volumeGuidUsedByProxy = volumeGuidUsedByProxy,
			ozHasProxy = ozHasProxy,
			proxyFileSize = proxyFileSize,
			volumeGuidUsedByOriginal = volumeGuidUsedByOriginal,
			ozHasOriginal = ozHasOriginal,
			originalFileSize = originalFileSize,
			xmpSidecarSize = xmpSidecarSize,
			isImportant = isImportant,
			isMarkedForOfflineUse = isMarkedForOfflineUse,
			sufficientRenditionFoundOnOz = sufficientRenditionFoundOnOz,
		}

	end

	return usefulArr

end

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

function NmbFileAccessDatabase:gatherArrayOfArray( ... )

	if not WFCloseSupport.hasBeenOrWillBeClosed( self ) then
		commitTouchTimesToDb( self )
  		return self._mruSQLiteDatabase:gatherArrayOfArray( ... )
	else
		return {}
	end
	
end

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

function NmbFileAccessDatabase:gatherArray( ... )

	if not WFCloseSupport.hasBeenOrWillBeClosed( self ) then
		commitTouchTimesToDb( self )
  		return self._mruSQLiteDatabase:gatherArray( ... )
	else
		return {}
	end
end

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

--[[!

	\brief TEST USE ONLY - dump the contents of the db.

	\param void
	
	\return Contents of the db
	
--]]

function NmbFileAccessDatabase:DEBUG_dumpLastAccessContents()

	check( 'NmbFileAccessDatabase' )

	return convertSqlRowsToUsefulLuaTables( self:gatherArrayOfArray( 
			[[SELECT * FROM lastAccessTimestamps
				INNER JOIN assetInfo USING( localId )]] ) )

end

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

local function _getRecentlyAccessed( self, limit, sort, skip )

	sort = sort or 'ASC'

	skip = skip or 0

	local nextBatchOffset = skip + limit

	return self:gatherArrayOfArray( 
			[[SELECT localId, docId, proxyFileSize, originalFileSize,
					xmpSidecarSize, volumeGuidUsedByProxy
				FROM lastAccessTimestamps
				INNER JOIN assetInfo USING( localId )
				ORDER BY accessTimestamp ]] .. sort .. -- SPAM: '?' Doesn't work with sort order.
				[[ LIMIT ? OFFSET ?]],
			limit,
			skip ), nextBatchOffset
	
end

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

local daysToSeconds = function( days )
	return days * 24 * 60 * 60
end

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

local function _getLeastRecentlyAccessedIdsInWindow( self, volumeGuid, binaryTypeFilter, timeWindow, inactiveAssetsOnly, limit, skipNMostRecent, offset )

	offset = offset or 0

	local nextBatchOffset = offset + limit

	local currentTime = KSDate_timeNow()
	
	local beforeNDaysAgo = timeWindow.beforeNDaysAgo
	local beforeNDaysAgoTimestamp = currentTime - daysToSeconds( beforeNDaysAgo )

	local filterParams = {}

	-- Calculate the upper time bound, if there is one.
	
	local upperTimeBoundFilter = " "
	
	if timeWindow.afterNDaysAgo then
	
		local afterNDaysAgo = timeWindow.afterNDaysAgo
		local afterNDaysAgoTimestamp = currentTime - daysToSeconds( afterNDaysAgo )
	
		upperTimeBoundFilter = " AND accessTimestamp >= ? "
	
		filterParams[ #filterParams + 1 ] = afterNDaysAgoTimestamp 
	
	end

	-- Build the binary status query filter and the ORDER BY statement
	
	local binaryFilter = " "
	local orderByStatement = " "
	
	if binaryTypeFilter == 'proxy' then
	
		binaryFilter = " AND volumeGuidUsedByProxy == ? AND ozHasProxy = 1 AND sufficientRenditionFoundOnOz = 1 "
			-- SPAM [jandrew 2017-09-11]: https://jira.corp.adobe.com/browse/NMB-6008
			-- Oz must have renditions for the asset. If Oz does not have renditions, 
			-- we need the proxy locally to generate renditions. This logic could be
			-- improved to take into account what renditions are local, but we'd have to
			-- hook into the PP to get that information. Given that all assets on Oz
			-- should have renditions, the following is likely sufficient.

		orderByStatement = " ORDER BY isImportant ASC, proxyFileSize DESC, accessTimestamp ASC "
	
		filterParams[ #filterParams + 1 ] = volumeGuid 
	
	elseif binaryTypeFilter == 'original' then
	
		binaryFilter = [[ AND volumeGuidUsedByOriginal == ? AND ozHasOriginal = 1 
							AND ( sufficientRenditionFoundOnOz = 1 OR volumeGuidUsedByProxy IS NOT NULL ) ]]
			-- SPAM [jandrew 2017-09-11]: https://jira.corp.adobe.com/browse/NMB-6008
			-- If Oz does not have renditions and we have no local proxy, then we need to keep 
			-- the local original to generate renditions.
			
		orderByStatement = " ORDER BY isImportant ASC, originalFileSize DESC, accessTimestamp ASC "
		
		filterParams[ #filterParams + 1 ] = volumeGuid 
		
	else
	
		binaryFilter = [[ AND ( ( volumeGuidUsedByProxy == ? AND ozHasProxy = 1 ) 
							OR ( volumeGuidUsedByOriginal == ? AND ozHasOriginal = 1 ) ) ]]
				
		filterParams[ #filterParams + 1 ] = volumeGuid 
		filterParams[ #filterParams + 1 ] = volumeGuid 
				
		orderByStatement = " ORDER BY isImportant ASC, originalFileSize DESC, proxyFileSize DESC, accessTimestamp ASC "
		
	end

	-- Maybe add an inactive asset filter
	
	local inactiveFilter = " "
	
	if inactiveAssetsOnly then
		inactiveFilter = " AND isImportant = 0 "
	end

	-- Fetch the most recently accessed timestamps
	
	local mostRecentlyAccessed = self:gatherArray( 
			[[ SELECT accessTimestamp 
				FROM lastAccessTimestamps 
				ORDER BY accessTimestamp DESC LIMIT ? ]],
				skipNMostRecent )
 
 	-- Calculate the lower time bound
 
	local lowerTimeBound = mostRecentlyAccessed[ #mostRecentlyAccessed ] or math.huge
	
	lowerTimeBound = math.min( beforeNDaysAgoTimestamp, lowerTimeBound )
	
	-- Run the query
	
	filterParams[ #filterParams + 1 ] = limit
	filterParams[ #filterParams + 1 ] = offset
		
	return self:gatherArrayOfArray( string.format(
			[[ SELECT localId, docId, proxyFileSize, originalFileSize,
					xmpSidecarSize, volumeGuidUsedByProxy, ozHasOriginal
				FROM lastAccessTimestamps
				INNER JOIN assetInfo USING( localId )
				WHERE accessTimestamp < ?
				AND ( isMarkedForOfflineUse == 0 OR isMarkedForOfflineUse IS NULL )
				%s
				%s
				%s
				%s
				LIMIT ? OFFSET ? ]],
				upperTimeBoundFilter, inactiveFilter, binaryFilter, orderByStatement ), 
			lowerTimeBound, unpack( filterParams ) ), nextBatchOffset

end

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

function NmbFileAccessDatabase:waitUntilUpToDate()

	return Observable.combineLatest( self._docStore:changeCounter(), self._changeCounterStream )
		
		:filter( function( docStoreChangeCounter, localChangeCounter )
		
			-- Wait until the NmbFileAccessDatabase is up to date with the Doc Store. 
		
			return localChangeCounter >= docStoreChangeCounter
		
		end )
		
		:take( 1 )
		
		:map( function()
		
			commitTouchTimesToDb( self )
			
			return true
		
		end )

end

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

local function _getLeastRecentlyAccessedIdsInWindowStream( self, volumeGuid, binaryTypeFilter, timeWindow, limit, skipNMostRecent, offset )

	return self:waitUntilUpToDate()
		
		:map( function()
		
			limit = limit or ACCESS_QUERY_DEFAULT_LIMIT
	
			skipNMostRecent = skipNMostRecent or 0
		
			local inactiveAssetsOnly = false
		
			return _getLeastRecentlyAccessedIdsInWindow( self, volumeGuid, binaryTypeFilter, timeWindow, inactiveAssetsOnly, limit, skipNMostRecent, offset )
		
		end )

end

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

--[[!

	\brief Retrieve the current list of purge candidates from the db, sorted
	by time of last access, but only for assets that haven't been accessed within
	N days.

	\param volumeGuid (string) Only return assets that live on this specific volume.
	\param timeWindow (table)
	\param limit (optional number) Limit the number of assets to search for.
	\param skipNMostRecent (optional number) Skip the N most recent assets from the query.
	\param offset (optional number) Used to fetch the next batch of assets.
	
	\return <code>Observable< array ></code> An array of purge candidates, along with the 
			offset to fetch the next batch.
	
--]]

function NmbFileAccessDatabase:getLeastRecentlyAccessedInWindow( volumeGuid, timeWindow, limit, skipNMostRecent, offset )

	check( 'NmbFileAccessDatabase', 'string', 'table', '?number', '?number', '?number' )

	return _getLeastRecentlyAccessedIdsInWindowStream( self, volumeGuid, nil, timeWindow, limit, skipNMostRecent, offset )

end

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

function NmbFileAccessDatabase:getLeastRecentlyAccessedProxiesInWindow( volumeGuid, timeWindow, limit, skipNMostRecent, offset )

	check( 'NmbFileAccessDatabase', 'string', 'table', '?number', '?number', '?number' )
	
	return _getLeastRecentlyAccessedIdsInWindowStream( self, volumeGuid, 'proxy', timeWindow, limit, skipNMostRecent, offset )

end

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

function NmbFileAccessDatabase:getLeastRecentlyAccessedOriginalsInWindow( volumeGuid, timeWindow, limit, skipNMostRecent, offset )

	check( 'NmbFileAccessDatabase', 'string', 'table', '?number', '?number', '?number' )
	
	return _getLeastRecentlyAccessedIdsInWindowStream( self, volumeGuid, 'original', timeWindow, limit, skipNMostRecent, offset )

end

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

local function _getInactiveAssetsInWindowStream( self, volumeGuid, binaryTypeFilter, timeWindow, limit, skipNMostRecent, offset )

	return self:waitUntilUpToDate()
		
		:map( function()
		
			limit = limit or ACCESS_QUERY_DEFAULT_LIMIT
	
			skipNMostRecent = skipNMostRecent or 0
		
			local inactiveAssetsOnly = true
		
			return _getLeastRecentlyAccessedIdsInWindow( 
					self, volumeGuid, binaryTypeFilter, timeWindow, inactiveAssetsOnly, limit, skipNMostRecent, offset )
		
		end )

end

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

--[[!

	\brief Retrieve the current list of inactive, proactive purge candidates from the 
	db, sorted by time of last access, but only for assets that haven't been accessed 
	within N days.

	\param volumeGuid (string) Only return assets that live on this specific volume.
	\param timeWindow (table)
	\param limit (optional number) Limit the number of assets to search for.
	\param skipNMostRecent (optional number) Skip the N most recent assets from the query.
	\param offset (optional number) Used to fetch the next batch of assets.

	\return <code>Observable< array ></code> An array of purge candidates, along with the 
			offset to fetch the next batch.
	
--]]

function NmbFileAccessDatabase:getInactiveAssetsInWindow( volumeGuid, timeWindow, limit, skipNMostRecent, offset )

	check( 'NmbFileAccessDatabase', 'string', 'table', '?number', '?number', '?number' )
	
	return _getInactiveAssetsInWindowStream( self, volumeGuid, nil, timeWindow, limit, skipNMostRecent, offset )

end

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

function NmbFileAccessDatabase:getInactiveProxiesInWindow( volumeGuid, timeWindow, limit, skipNMostRecent, offset )

	check( 'NmbFileAccessDatabase', 'string', 'table', '?number', '?number', '?number' )
	
	return _getInactiveAssetsInWindowStream( self, volumeGuid, 'proxy', timeWindow, limit, skipNMostRecent, offset )

end

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

function NmbFileAccessDatabase:getInactiveOriginalsInWindow( volumeGuid, timeWindow, limit, skipNMostRecent, offset )

	check( 'NmbFileAccessDatabase', 'string', 'table', '?number', '?number', '?number' )
	
	return _getInactiveAssetsInWindowStream( self, volumeGuid, 'original', timeWindow, limit, skipNMostRecent, offset )

end

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

--[[!

	\brief Retrieve the current list of purge candidates from the db, sorted
	by time of last access.

	\param limit (optional number) Limit the number of assets to search for.
	\param offset (optional number) Used to fetch the next batch of assets.
	
	\return An array of purge candidates, along with the offset to fetch the next batch.
	
--]]

function NmbFileAccessDatabase:getLeastRecentlyAccessed( limit, offset )

	check( 'NmbFileAccessDatabase', '?number', '?number' )
	
	limit = limit or ACCESS_QUERY_DEFAULT_LIMIT
	
	return Observable.just( _getRecentlyAccessed( self, limit, nil, offset ) )

end

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

--[[!

	\brief Retrieve the N most recently accessed assets.

	\param limit (optional number) Limit the number of assets to search for.
	\param offset (optional number) Used to fetch the next batch of assets.
	
	\return An array of purge candidates, along with the offset to fetch the next batch.
	
--]]

function NmbFileAccessDatabase:getMostRecentlyAccessed( limit, offset )

	check( 'NmbFileAccessDatabase', '?number', '?number' )
	
	limit = limit or ACCESS_QUERY_DEFAULT_LIMIT
	
	return Observable.just( _getRecentlyAccessed( self, limit, 'DESC', offset ) )

end

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

function NmbFileAccessDatabase:_DEBUG_simulateCorruption()

	self._mruSQLiteDatabase:_DEBUG_simulateCorruption()

end

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

function NmbFileAccessDatabase:_DEBUG_numberOfLocalOriginals()

	local result = self:gatherArray( 
			[[SELECT COUNT( * )
				FROM assetInfo
				WHERE volumeGuidUsedByOriginal IS NOT NULL ]] )

	return Observable.just( result[ 1 ] )

end

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

function NmbFileAccessDatabase:_DEBUG_sizeOfLocalOriginals()

	local result = self:gatherArray( 
			[[SELECT SUM( originalFileSize )
				FROM assetInfo
				WHERE volumeGuidUsedByOriginal IS NOT NULL ]] )

	return Observable.just( result[ 1 ] )

end

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

function NmbFileAccessDatabase:_DEBUG_numberOfLocalProxies()

	local result = self:gatherArray( 
			[[SELECT COUNT( * )
				FROM assetInfo
				WHERE volumeGuidUsedByProxy IS NOT NULL ]] )

	return Observable.just( result[ 1 ] )

end

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

function NmbFileAccessDatabase:_DEBUG_sizeOfLocalProxies()

	local result = self:gatherArray( 
			[[SELECT SUM( proxyFileSize )
				FROM assetInfo
				WHERE volumeGuidUsedByProxy IS NOT NULL ]] )

	return Observable.just( result[ 1 ] )

end

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

return Rx.Class 'NmbFileAccessDatabase'( NmbFileAccessDatabase )
