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

NmbFileMover.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2017 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 ReplaySubject = Rx.ReplaySubject
local Subject = Rx.Subject
local Observable = Rx.Observable

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

	-- WF.core.utils
local KSPathUtils = require 'KSPathUtils'
local KSTableUtils = require 'KSTableUtils'
local KSDate = require 'KSDate'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
local WFFileUtils = require 'WFFileUtils'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxLowFidelityTimer = require 'WFRxLowFidelityTimer'
local WFProgressReporter = require 'WFProgressReporter'
local WFRxScheduler = require 'WFRxScheduler'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxBatchedBulkApplyOverlay = require 'WFRxBatchedBulkApplyOverlay'
local WFRxThrottledBatchedApplyOverlayToIds = require 'WFRxThrottledBatchedApplyOverlayToIds'
local WFCrypto = require 'WFCrypto'

	-- WF.oz_client.managed_catalog
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatDeniedListHelper = require 'WFCatDeniedListHelper'
local WFCatDelete = require 'WFCatDelete'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatPixelPathGenerator = require 'WFCatPixelPathGenerator'
local WFCatVolumeAgent = require 'WFCatVolumeAgent'

local NmbPathRestorer = require 'NmbPathRestorer'
local NmbFileManagementSettings = require 'NmbFileManagementSettings'

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

local logger = WFLogger 'NmbFileMover'

local check = assert( WFArguments.check )

local isValidAssetDocument = assert( WFCatIndexUtils.isValidAssetDocument )
local isVideoDocument = assert( WFCatIndexUtils.isVideoDocument )

local derefOpt = assert( KSTableUtils.derefOpt )

local parent = assert( KSPathUtils.parent )

local leafName = assert( KSPathUtils.leafName )

local strformat = assert( string.format )

local runUntilDone = assert( WFRxRunUntilDone )

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

local maxFileMovesInParallel = 1
	-- Maximum number of file moves allowed at once.
	
local numberOfTriesBeforeDenyingFile = 3
	-- Number of attempts at moving the file before we temporarily block it.
	
local failedMigrationBlackoutWindowInSeconds = 180
	-- Numbers of seconds we block a file after we fail to move it.

local volumeStatusCheckPeriod = 2
	-- Every N seconds, check the status of each volume involved in the migration.
	
local bulkApplyOverlayPeriod = 0.5
	-- Every N seconds, the doc store is refreshed with updated file annotations.

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

local kRemapping = "REMAPPING"
local kMoving = "MOVING"
local kIndexing = "INDEXING"

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

local function toBool( zeroOrOne )

	return zeroOrOne > 0

end

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

local function distinctTableCompare( _, tbl1, _, tbl2 )

	return KSTableUtils.deepEquals( tbl1, tbl2 )

end

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

local function currentStepOfProgress( self )

	return { currentStep = self._currentStep }

end 

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

local function buildGuidBasePath( guid, relativePath )

	return KSPathUtils.child( guid, relativePath )
	
end 

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

local function checkIfMaybeDenied( docValue )

	local deniedListKey = 'NmbFileMover'

	local isDenied = WFCatDeniedListHelper.isDeniedForKey( docValue, deniedListKey )
	
	return isDenied and 1 or 0
	
end

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

local function getCaptureDate( docValue )

	return derefOpt( docValue, 'captureDate' ) or '0000-00-00T00:00:00.000'

end

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

local function getOriginalFileName( docValue )

	local originalRelativePath = WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue )

	if not originalRelativePath or originalRelativePath == '' then

		-- Only use importSource if originalRelativePath is null as it won't
		-- be hyphenated properly for duplicates or conflicting files.

		return derefOpt( docValue, 'importSource', 'fileName' )

	end

	return leafName( originalRelativePath )

end

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

local function calculateBaseBinaryPath( binaryPath )

	-- Original path: 	<base path>/originals/year/date/<file>
	-- Proxy path: 		<base path>/proxies/<proxy size>/<3-character subfolder>/<file>
	
	if not binaryPath or binaryPath == '' then
		return ''
	end
	
	return parent( parent( parent( parent( binaryPath ) ) ) )
	
end

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

local function calculateBaseRootPath( binaryPath )

	-- Original path: 	<root path>/Nimbus/<guid>/originals/year/date/<file>
	-- Proxy path: 		<root path>/Nimbus/<guid>/proxies/<proxy size>/<3-character subfolder>/<file>
	
	if not binaryPath or binaryPath == '' then
		return ''
	end
	
	local basePath = calculateBaseBinaryPath( binaryPath )
	
	return parent( parent( basePath ) )
	
end

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

local function generateIndexSchema( _, indexStoreSchema )

	check( '_', 'WFRxIndexStoreSchema' )
	
	indexStoreSchema:addTable {

		sql = [[CREATE TABLE assetPathsAndInfo(
				localId,
				originalVolumeGuid,
				originalRelativePath,
				originalGuidBasePath,
				ozHasOriginal,
				originalSize,
				sizeOfUnpurgableOriginals,
				fileName,
				captureDate,
				sha256,
				proxyVolumeGuid,
				proxyRelativePath,
				proxyGuidBasePath,
				proxyType,
				proxySize,
				isVideo,
				isMarkedForOffline,
				isMaybeDenied )]],

		hasOneRowPerDocument = true,
		
		relevantTypeSubtypeTuples = {
			{ 'asset', 'image' },
			{ 'asset', 'video' },
			{ 'asset', 'deleted_image' },
			{ 'asset', 'deleted_video' }
		},
		
		map = function( docValue, emit )

			if isValidAssetDocument( docValue ) then

				local localId = docValue._localId

				-- Local Original Info
				
				local originalVolumeGuid = WFCatFileSpec.getOriginalVolumeGuidFromDocValue( docValue )
				
				local originalRelativePath = WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue )

				local originalRelativeBasePath = originalRelativePath and calculateBaseBinaryPath( originalRelativePath )

				local originalGuidBasePath = originalRelativeBasePath and buildGuidBasePath( originalVolumeGuid, originalRelativeBasePath )
				
				-- Local Proxy Info

				local proxyVolumeGuid, proxyType = WFCatFileSpec.getProxyVolumeGuidFromDocValue( docValue )
				
				local proxyRelativePath = WFCatFileSpec.getProxyRelativePathFromDocValue( docValue )

				local proxyRelativeBasePath = proxyRelativePath and calculateBaseBinaryPath( proxyRelativePath )

				local proxyGuidBasePath = proxyRelativeBasePath and buildGuidBasePath( proxyVolumeGuid, proxyRelativeBasePath )

				-- Other

				local ozHasOriginal = WFCatFileSpec.getOriginalLinkFromDocValue( docValue ) and 1 or 0
				
				local fileName = getOriginalFileName( docValue )

				local captureDate = getCaptureDate( docValue )
				
				local sha256 = derefOpt( docValue, 'importSource', 'sha256' )
				
				local isVideo = isVideoDocument( docValue ) and 1 or 0
				
				-- File Migration Priority
				
				local isMarkedForOffline = WFCatFileSpec.getMarkedForOfflineUseFromDocValue( docValue ) and 1 or 0

				local isMaybeDenied = checkIfMaybeDenied( docValue )
				
				-- Size

				local willNotUploadMaster = KSTableUtils.derefOpt( docValue, 'importSource', 'willNotUploadMaster' )

				local originalSize = not willNotUploadMaster and derefOpt( docValue, 'importSource', 'fileSize' ) or 0

				local proxySize = WFCatFileSpec.getProxySizeFromDocValue( docValue )
				
				local sizeOfUnpurgableOriginals = 0
				
				if isMarkedForOffline == 1 or ozHasOriginal == 0 then
				
					-- The file _has_ to be migrated and can not be purged.
					
					sizeOfUnpurgableOriginals = originalSize
				
				end
				
				emit( localId, originalVolumeGuid, originalRelativePath, originalGuidBasePath, ozHasOriginal, 
						originalSize, sizeOfUnpurgableOriginals, fileName, captureDate, sha256, 
						proxyVolumeGuid, proxyRelativePath, proxyGuidBasePath, proxyType, proxySize, isVideo,
						isMarkedForOffline, isMaybeDenied )
				
			end

		end,

	}
	
	indexStoreSchema:addIndex
		[[CREATE INDEX assetPathsAndInfo_sortOrder
			ON assetPathsAndInfo( isMaybeDenied ASC, ozHasOriginal ASC, isMarkedForOffline DESC, localId )]]
	
	indexStoreSchema:addIndex
		[[CREATE INDEX assetPathsAndInfo_localIdOriginalPath
			ON assetPathsAndInfo( localId, originalVolumeGuid, originalRelativePath )]]
	
	indexStoreSchema:addIndex
		[[CREATE INDEX assetPathsAndInfo_localIdProxyPath
			ON assetPathsAndInfo( localId, proxyVolumeGuid, proxyRelativePath )]]
			
	indexStoreSchema:addIndex
		[[CREATE INDEX assetPathsAndInfo_SizeMarkedForOffline
			ON assetPathsAndInfo( originalSize, proxySize, isMarkedForOffline )]]
		
	indexStoreSchema:addIndex
		[[CREATE INDEX assetPathsAndInfo_originalGuidBasePath
			ON assetPathsAndInfo( originalGuidBasePath, localId )]]
	
	indexStoreSchema:addIndex
		[[CREATE INDEX assetPathsAndInfo_proxyGuidBasePath
			ON assetPathsAndInfo( proxyGuidBasePath, localId )]]
		
end

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

local function buildWhereClause( column, values, useOrInsteadOfWhere )

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

	local whereClause

	local operator = useOrInsteadOfWhere and 'OR' or 'WHERE'

	for i, value in ipairs( values ) do
	
		operator = i == 1 and strformat( " %s ( ", operator ) or "OR "
		whereClause = strformat( "%s%s%s = '%s' ", whereClause or '', operator, column, value )
		
	end

	return whereClause and strformat( "%s)", whereClause ) or ''

end

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

-- Builds the index store query we use to migrate assets from the source drives
-- to the destination drives. The query is built to filter out assets that are
-- coming from, or going to, offline drives.

local function buildMigrationQuery( indexStore, isOriginalDestinationDriveReady,
		isProxyDestinationDriveReady, onlineOriginalGuidBasePaths, onlineProxyGuidBasePaths )

	check( 'WFRxIndexStore', 'boolean', 'boolean', 'table', 'table' )

	local canMigrateOriginals = isOriginalDestinationDriveReady and #onlineOriginalGuidBasePaths > 0
	local canMigrateProxies = isProxyDestinationDriveReady and #onlineProxyGuidBasePaths > 0
	
	if not canMigrateOriginals and not canMigrateProxies then
		return Observable.just( {} )
	end

	local originalWhereClause = isOriginalDestinationDriveReady and 
			buildWhereClause( 'originalGuidBasePath', onlineOriginalGuidBasePaths ) or ''

	local proxyWhereClause = isProxyDestinationDriveReady and 
			buildWhereClause( 'proxyGuidBasePath', onlineProxyGuidBasePaths, 
					isOriginalDestinationDriveReady and #onlineOriginalGuidBasePaths > 0 ) or ''
	
	local queryString = 
			[[SELECT docId, localId, originalGuidBasePath, originalVolumeGuid, originalRelativePath, fileName, captureDate,
					proxyGuidBasePath, proxyVolumeGuid, proxyRelativePath, proxyType, isMaybeDenied, sha256,
					isMarkedForOffline, ozHasOriginal, isVideo
				FROM assetPathsAndInfo ]]
				.. originalWhereClause .. proxyWhereClause ..
				[[ ORDER BY isMaybeDenied ASC, ozHasOriginal ASC, isMarkedForOffline DESC
				LIMIT 250]]
	
	logger:infof( "Migration Query: %s", queryString )
	
	return indexStore:query {
			sql = queryString,
			quiesceUntilUpToDate = true,
				-- It's necessary that we only process up to date data because
				-- the query is automatically stopped when there are no results.
		}

end

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

local function logFileMoverMessage( self, logLevel, msgFmtStr , ... )

	check( 'NmbFileMover', 'string', 'string' )

	-- Creates peristent logging of file migration events.
	-- Used to capture errors and issues of assets being migrated
	-- Log is fileMover.log located in
	-- Pictures/Lightroom Library.lrlibrary/migration (MAC) and AppData\Local\Lightroom CC\Data\migration (WIN)

	-- format message,

	local message = string.format( msgFmtStr, ... )

	logger[ logLevel ]( logger, message )

	-- write to log file
	if message ~= "" then

		-- format message for persistent logging.
		message = string.format("%s\n", message)

		if not WFFileUtils.exists(self._fileMoveLogPath) then
			logger:infof( "Creating migration folder at %s", self._fileMoveLogPath)
			WFFileUtils.mkdir(self._fileMoveLogPath)
		end

		-- open or create file in 'append' mode
		local file = io.open( KSPathUtils.child(self._fileMoveLogPath, "fileMover.log"), "a" )
		if file then
			file:write(message)
			file:close()
		else
			logger:warnf("FileMover log file could not be opened at %s", self._fileMoverLogPath)
		end

	else
		logger:warnf("FileMover log message could not be interpreted.")
	end

end

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

--[[
	Returns an array of source volumes that includes the volumes online status 
	and the base paths associated with the volume:
	
	{
		'<volume guid>' = {
			online = <boolean>
			basePaths = {
				<base path>,
				<base path>,
				<base path>
			}
		},
		'<volume guid>' = {
			online = <boolean>
			basePaths = {
				<base path>,
				<base path>,
				<base path>
			}
		},
		...
	}
	
]]--

local function getSourceVolumeStatusFromBasePaths( baseGuidPathInfos )

	check( 'table' )

	local sourceVolumeStatuses = {}
	local numberOfAssetsToMigrate = 0
	local sizeOfAllAssets = 0
	local sizeOfRequiredAssets = 0
	
	for i, row in ipairs( baseGuidPathInfos ) do
	
		local volumeGuid = row[ 1 ]
		local baseGuidPath = row[ 2 ]
		local assetsWithBasePath = row[ 3 ]
		local assetSpace = row[ 4 ]
		local requiredAssetSpace = row[ 5 ] or 0
		
		logger:infof( "%d assets on volume %s need to be migrated.", assetsWithBasePath, volumeGuid )
	
		numberOfAssetsToMigrate = numberOfAssetsToMigrate + assetsWithBasePath
	
		sizeOfAllAssets = sizeOfAllAssets + assetSpace
		
		sizeOfRequiredAssets = sizeOfRequiredAssets + requiredAssetSpace
	
		local status = sourceVolumeStatuses[ volumeGuid ] or {}
		
		status.basePaths = status.basePaths or {}
		table.insert( status.basePaths, baseGuidPath )
		
		status.numberOfAssets = status.numberOfAssets and 
				status.numberOfAssets + assetsWithBasePath or 
				assetsWithBasePath
		
		if not WFFileUtils.isVolumeGuidOnline( volumeGuid ) then
		
			logger:infof( "Volume %s cannot be reached. Migration of %d assets will be skipped.", volumeGuid, assetsWithBasePath )
			status.online = false
			
		else
		
			status.online = true
			
		end
		
		sourceVolumeStatuses[ volumeGuid ] = status
		
	end

	return sourceVolumeStatuses, numberOfAssetsToMigrate, sizeOfAllAssets, sizeOfRequiredAssets
	
end

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

local function getSourceBasePathsAndSizes( indexStore, destBasePathForOriginals, destBasePathForProxies )

	check( 'WFRxIndexStore', 'string', 'string' )

	-- These queries returns the set of unique base paths with the count of assets
	-- for those base paths.
	
	local sourceBasePathsForOriginalsQuery =
			indexStore:queryOnce {
				sql = [[SELECT originalVolumeGuid, originalGuidBasePath, COUNT( localId ), SUM( originalSize ), SUM( sizeOfUnpurgableOriginals )
						FROM assetPathsAndInfo
						WHERE originalGuidBasePath != ? AND originalGuidBasePath IS NOT NULL
						GROUP BY originalGuidBasePath ]],
				destBasePathForOriginals,
				quiesceUntilUpToDate = true,
			}
	
	local sourceBasePathsForProxiesQuery =
			indexStore:queryOnce {
				sql = [[SELECT proxyVolumeGuid, proxyGuidBasePath, COUNT( localId ), proxyRelativePath
						FROM assetPathsAndInfo
						WHERE proxyGuidBasePath != ? AND proxyGuidBasePath IS NOT NULL
						GROUP BY proxyGuidBasePath]],
				destBasePathForProxies,
				quiesceUntilUpToDate = true,
			}
			:map( function( rows )
			
				for _, row in ipairs( rows ) do
				
					local proxyVolumeGuid = row[ 1 ]
					local proxyRelativePath = row[ 4 ]
					
					local proxyRelativeBasePath = calculateBaseBinaryPath( proxyRelativePath )
					
					local pathToProxySourceDir = WFFileUtils.createFullPathFromRelativePathAndGuid( proxyRelativeBasePath, proxyVolumeGuid )
	
					local success, totalSpaceUsedByNimbusProxies = 0
	
					if pathToProxySourceDir then
						success, totalSpaceUsedByNimbusProxies = pcall( WFFileUtils.calculatePathSize, pathToProxySourceDir )
					end
						
					row[ 4 ] = success and totalSpaceUsedByNimbusProxies or 0
				
				end
			
				return rows
			
			end )
	
	return Rx.Observable.combineLatest(
	
			function( originalSourceBasePathsAndSizes, proxySourceBasePathsAndSizes )
				return originalSourceBasePathsAndSizes, proxySourceBasePathsAndSizes
			end,

			sourceBasePathsForOriginalsQuery,
			sourceBasePathsForProxiesQuery
			
		)

end 

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

--[[
	Returns arrays of source volumes, for both originals and proxies, that include
	the volumes' online status and the base paths associated with the volume. See
	getSourceVolumeStatusFromBasePaths for more info.
]]--

local function monitorSourceVolumes( indexStore, volumeGuidForOriginals, relativeDestPathForOriginals,
									volumeGuidForProxies, relativeDestPathForProxies )

	check( 'WFRxIndexStore', '+string', 'string', '+string', 'string' )

	local destBasePathForOriginals = buildGuidBasePath( volumeGuidForOriginals, relativeDestPathForOriginals )
	local destBasePathForProxies = buildGuidBasePath( volumeGuidForProxies, relativeDestPathForProxies )

	return getSourceBasePathsAndSizes( indexStore, destBasePathForOriginals, destBasePathForProxies )
	
		:take( 1 )
	
		:flatMapLatest( function( originalSourceBasePathsAndSizes, proxySourceBasePathsAndSizes )

			return Observable.timer( 0, volumeStatusCheckPeriod, WFRxScheduler.sharedScheduler() )
				:map( function()
					return originalSourceBasePathsAndSizes, proxySourceBasePathsAndSizes
				end )
				
		end )
		
		:map( function( originalSourceBasePathsAndSizes, proxySourceBasePathsAndSizes )
		
			local originalSourceVolumeStatuses, originalsToMigrate, sizeOfAllOriginals, sizeOfRequiredOriginals
					= getSourceVolumeStatusFromBasePaths( originalSourceBasePathsAndSizes )
					
			local proxySourceVolumeStatuses, proxiesToMigrate, sizeOfAllProxies, sizeOfRequiredProxies
					= getSourceVolumeStatusFromBasePaths( proxySourceBasePathsAndSizes )
			
			local sourceVolumeStatuses = {
			
				originalSourceVolumeStatuses = originalSourceVolumeStatuses,
				proxySourceVolumeStatuses = proxySourceVolumeStatuses,
				originalsToMigrate = originalsToMigrate,
				proxiesToMigrate = proxiesToMigrate,
				sizeOfAllOriginals = sizeOfAllOriginals,
				sizeOfRequiredOriginals = sizeOfRequiredOriginals,
				sizeOfAllProxies = sizeOfAllProxies,
				sizeOfRequiredProxies = sizeOfRequiredProxies
				
			}
			
			return sourceVolumeStatuses
			
		end )

		:distinctUntilChanged( nil, distinctTableCompare )

end

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

local function getSpaceConstrainedVolumes( self, destinationVolumeStatuses, sourceVolumeStatuses ) 

	local volumeGuidForOriginals = self._volumeGuidForOriginals
	local volumeGuidForProxies = self._volumeGuidForProxies

	local spaceNeededForOriginals = sourceVolumeStatuses.sizeOfRequiredOriginals
	local spaceNeededForProxies = sourceVolumeStatuses.sizeOfRequiredProxies

	local spaceConstrainedVolumes = {}
	local count = 0
	
	if volumeGuidForOriginals == volumeGuidForProxies then
	
		-- Originals And Proxies share a destination drive
		
		local volume = destinationVolumeStatuses[ volumeGuidForOriginals ]
		
		if volume
			and volume.online
			and ( sourceVolumeStatuses.originalsToMigrate > 0 or sourceVolumeStatuses.proxiesToMigrate > 0 )
			and ( volume.outOfSpace or spaceNeededForOriginals + spaceNeededForProxies > volume.freeSpace ) then
		
			spaceConstrainedVolumes[ volumeGuidForOriginals ] = {

				freeSpace = volume.freeSpace,
				outOfSpace = volume.outOfSpace,
				spaceNeeded = spaceNeededForOriginals + spaceNeededForProxies,
				
			}

			count = count + 1

		end
		
	else
	
		-- Originals
		
		local volume = destinationVolumeStatuses[ volumeGuidForOriginals ]
		
		if volume
			and volume.online
			and sourceVolumeStatuses.originalsToMigrate > 0
			and ( volume.outOfSpace or spaceNeededForOriginals > volume.freeSpace ) then
		
			spaceConstrainedVolumes[ volumeGuidForOriginals ] = {

				freeSpace = volume.freeSpace,
				outOfSpace = volume.outOfSpace,
				spaceNeeded = spaceNeededForOriginals,
				
			}
		
			count = count + 1
		
		end

		-- Proxies
	
		volume = destinationVolumeStatuses[ volumeGuidForProxies ]
		
		if volume
			and volume.online
			and sourceVolumeStatuses.proxiesToMigrate > 0
			and ( volume.outOfSpace or spaceNeededForProxies > volume.freeSpace ) then
		
			spaceConstrainedVolumes[ volumeGuidForProxies ] = {

				freeSpace = volume.freeSpace,
				outOfSpace = volume.outOfSpace,
				spaceNeeded = spaceNeededForProxies,
				
			}
		
			count = count + 1
		
		end
	
	end

	local noSpaceConstraints = count == 0

	return noSpaceConstraints, spaceConstrainedVolumes

end

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

--[[
	Returns an array of destination volumes that includes their online status:
	{
		'<volume for originals>' = {
			online = <boolean>
			freeSpace = <number>
			outOfSpace = <boolean>
		},
		'<volume for proxies>' = {
			online = <boolean>
			freeSpace = <number>
			outOfSpace = <boolean>
		}
	}
]]--

local function monitorDestinationVolumes( volumeGuidForOriginals, volumeGuidForProxies, statusOfVolumesStream )

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

	return statusOfVolumesStream

		:map( function( statusOfVolumes )
		
			local offlineVolumes = statusOfVolumes.offlineVolumes
			local volumesOutOfSpace = statusOfVolumes.volumesOutOfSpace
		
			local destinationVolumeStatuses = {}
			
			destinationVolumeStatuses[ volumeGuidForOriginals ] = {
				online = not offlineVolumes[ volumeGuidForOriginals ],
				outOfSpace = volumesOutOfSpace[ volumeGuidForOriginals ]
			}

			destinationVolumeStatuses[ volumeGuidForProxies ] = destinationVolumeStatuses[ volumeGuidForProxies ] or {
				online = not offlineVolumes[ volumeGuidForProxies ],
				outOfSpace = volumesOutOfSpace[ volumeGuidForProxies ]
			}

			return destinationVolumeStatuses
			
		end )
	
		:distinctUntilChanged( nil, distinctTableCompare )
	
		:map( function( destinationVolumeStatuses )
			
			-- When the volume statuses change, update the freeSpace.
			
			local volume = WFFileUtils.getVolumeFromGuid( volumeGuidForOriginals )
			
			destinationVolumeStatuses[ volumeGuidForOriginals ].freeSpace 
					= volume and WFCatVolumeAgent.checkAvailableSpaceOnVolume( volume ) or 0
					
			volume = WFFileUtils.getVolumeFromGuid( volumeGuidForProxies )
			
			destinationVolumeStatuses[ volumeGuidForProxies ].freeSpace 
					= volume and WFCatVolumeAgent.checkAvailableSpaceOnVolume( volume ) or 0
		
			return destinationVolumeStatuses
		
		end )
	
end

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

-- Return an array of base paths associated with the online volumes.

local function getOnlineGuidBasePathsFromVolumes( volumes )

	check( 'table' )

	local basePaths = {}

	for _, volume in pairs( volumes ) do

		if volume.online then
		
			for _, basePath in ipairs( volume.basePaths or {} ) do
				basePaths[ #basePaths + 1 ] = basePath
			end
			
		end

	end

	return basePaths

end

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

-- Returns true if all the volumes are offline or out of space.

local function allVolumesOfflineOrOutOfSpace( volumes )

	check( 'table' )

	for _, volume in pairs( volumes ) do
	
		if volume.online and not volume.outofSpace then
			return false
		end
	
	end
	
	return true
	
end

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

local function reportCompletionAndCleanup( self, results )

	logger:infof( "Migration Complete: %s", results )
	
	self._migrationProgress:done( { summary = results } )
	
	if self._synchronizer then
		self._synchronizer:startSyncing()
	end
	
	self._running = false
	
	self._filesInFlightCount = 0

	self._migrationCount = 0

	self._recentlyMigratedFiles = {}
	
	self._filePathOverlays = {}
	
	self._assetsToDelete = {}
	
	-- Dispose of any in-flight file moves.
	
	for k, _ in pairs( self._fileMoveSubscriptions ) do
		logger:infof( "Cancelling in-flight file move for asset %s", k )
		self._fileMoveSubscriptions[ k ]:dispose()
	end
	
	self._fileMoveSubscriptions = {}
	
	if self._migrationSubscription then
		self._migrationSubscription:dispose()
		self._migrationSubscription = nil
	end

end

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

local function migrationComplete( self, destinationVolumeStatuses, sourceVolumeStatuses, 
		allFilesAreDenied, numberOfDeniedFiles )

	check( 'NmbFileMover', 'table', 'table', '?boolean', '?number' )

	local volumeGuidForOriginals = self._volumeGuidForOriginals
	local volumeGuidForProxies = self._volumeGuidForProxies

	local results = {
		success = true,
		filesMigrated = self._migrationCount or 0,
		missingFiles = self._missingFiles or {},
		offlineSourceDrives = {},
		offlineDestinationDrives = {},
		spaceConstrainedDestinationDrives = {},
		volumeNames = {}
	}
	
	-- Annotate the result with denied file information.
	
	if allFilesAreDenied then
		results.success = false
		results.numberOfDeniedFiles = numberOfDeniedFiles
	end

	-- Check to see if the migration failed due to space constraints.
	
	local enoughSpace, spaceConstrainedVolumes
			= getSpaceConstrainedVolumes( self, destinationVolumeStatuses, sourceVolumeStatuses )
	
	if not enoughSpace then
	
		results.success = false
		results.spaceConstrainedDestinationDrives = spaceConstrainedVolumes
	
	end

	-- Check to see if there are offline source volumes with assets
	-- that need to be migrated.
	
	local function checkOfflineSourceVolumes( volumeStatuses, isProxyVolume )
	
		for path, volume in pairs( volumeStatuses ) do
	
			if not volume.online and volume.numberOfAssets > 0 then

				results.success = false
				
				local assetCount = results.offlineSourceDrives[ path ] or 0
				results.offlineSourceDrives[ path ] = assetCount + volume.numberOfAssets 
				
			end
		
		end
		
	end
	
	checkOfflineSourceVolumes( sourceVolumeStatuses.originalSourceVolumeStatuses )
	checkOfflineSourceVolumes( sourceVolumeStatuses.proxySourceVolumeStatuses, true )
	
	-- Check if we have offline original destination drive that caused
	-- the migration to fail.
	
	local originalVolume = destinationVolumeStatuses[ volumeGuidForOriginals ]
	
	if originalVolume and
		not originalVolume.online and
		sourceVolumeStatuses.originalsToMigrate > 0 then

		results.success = false
		local assetCount = results.offlineDestinationDrives[ volumeGuidForOriginals ] or 0
		results.offlineDestinationDrives[ volumeGuidForOriginals ] = assetCount + sourceVolumeStatuses.originalsToMigrate
		
	end

	-- Check if we have offline proxy destination drive that caused
	-- the migration to fail.
	
	local proxyVolume = destinationVolumeStatuses[ volumeGuidForProxies ]
	
	if proxyVolume and
		not proxyVolume.online and
		sourceVolumeStatuses.proxiesToMigrate > 0 then

		results.success = false
		local assetCount = results.offlineDestinationDrives[ volumeGuidForProxies ] or 0
		results.offlineDestinationDrives[ volumeGuidForProxies ] = assetCount + sourceVolumeStatuses.proxiesToMigrate
		
	end
	
	-- Create a volume name lookup table for use in the UI
	
	----------------------------------------------------------------------------
	
	local function createNameLookupForVolumes( volumes )
	
		local lookupTable = results.volumeNames
	
		for guid, _ in pairs( volumes ) do
	
			if not lookupTable[ guid ] then

				local volumeName = WFFileUtils.getVolumeNameFromGuid( guid )

				if not volumeName then
			
					volumeName = NmbFileManagementSettings.getReadableIdentifierForVolumeGuid( guid )
							or guid
					
				end
				
				lookupTable[ guid ] =  volumeName
				
			end
		
		end
		
		results.volumeNames = lookupTable
		
	end
	
	createNameLookupForVolumes( results.offlineSourceDrives )
	createNameLookupForVolumes( results.offlineDestinationDrives )
	createNameLookupForVolumes( results.spaceConstrainedDestinationDrives )
	
	reportCompletionAndCleanup( self, results )

end

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

local function reportMigrationError( self, errorMessage )

	local results = {
		success = false,
		errorMessage = errorMessage,
		filesMigrated = self._migrationCount or 0,
	}

	reportCompletionAndCleanup( self, results )

end

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

local function createKey( localId, binaryType )

	if localId and binaryType then
		return localId .. "::" .. binaryType
	end
	
end

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

local function clearRecentlyMigrated( self, localId, binaryType )

	local key = createKey( localId, binaryType )
	self._recentlyMigratedFiles[ key ] = nil

end

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

local function markAsRecentlyMigrated( self, localId, binaryType )

	local key = createKey( localId, binaryType )
	self._recentlyMigratedFiles[ key ] = true

end

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

local function wasRecentlyMigrated( self, localId, binaryType )

	local key = createKey( localId, binaryType )
	return self._recentlyMigratedFiles[ key ]

end

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

local function doClearBlackoutWindow( self, localId, binaryType )

	-- Because the index store takes a few seconds to update, and the doc store
	-- could have potentially just updated, we will give the index store some time
	-- to bring itself up to date.

	local indexStoreUpdateBuffer = 30
	
	WFRxLowFidelityTimer.timer( indexStoreUpdateBuffer )
		:tap( function()
			clearRecentlyMigrated( self, localId, binaryType )
		end )
		:call( runUntilDone )
		
end

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

local function migrationCompletedOrDenied( self, localId, binaryType )

	self._filesInFlightCount = self._filesInFlightCount - 1
	
	self._fileMoveComplete:onNext( localId )

	doClearBlackoutWindow( self, localId, binaryType )
	
end

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

local function updateFileAnnotationsAndDeleteFiles( self, assetOverlays, assetsToDelete )

	local docStore = self._docStore
	local deletionBacklog = self._deletionBacklog
	
	local localFilePaths = {}
	
	for _, asset in ipairs( assetsToDelete ) do
		
		local pathToDelete = asset.pathToDelete
		
		local volumeGuid = pathToDelete and WFFileUtils.getVolumeGuidForPath( pathToDelete )
		local relativePath = pathToDelete and WFFileUtils.removeVolumeFromPath( pathToDelete )
		
		table.insert( localFilePaths, {
			binaryType = asset.binaryType,
			localId = asset.localId,
			volumeGuid = volumeGuid,
			relativePath = relativePath,
		} )
		
	end
	
	if #assetOverlays > 0 then
		
		logger:infof( "Caching %d file deletions in the backlog.", #localFilePaths )
		
		return deletionBacklog:bulkAddToDeletionBacklog( localFilePaths )
	
			:map( function()
			
				logger:info( "Ensuring deletion backlog is up to date." )
				
				return deletionBacklog:bringDeletionBacklogUpToDate()
				
			end )
			
			:flatMapLatest( function()
			
				logger:infof( "Updating the file path annotation for %d assets.", #assetOverlays )
			
				return WFRxBatchedBulkApplyOverlay( "bulkAnnotationUpdater", docStore, assetOverlays )
				
			end )
			
			:map( function()
			
				for _, localFile in ipairs( localFilePaths ) do
					
					local volumeGuid = localFile.volumeGuid
					local relativePath = localFile.relativePath
					local deleteSuccessful
					
					-- File successfully moved, so delete the old file.
					
					logFileMoverMessage( self, "info", "Now deleting old asset at path %s:%s", volumeGuid, relativePath )
				
					if volumeGuid and relativePath then
						deleteSuccessful = WFCatDelete.deleteAssetAndMaybeParentFolders( volumeGuid, relativePath )
					end
					
					logger:infof( "Deleting %s:%s from the deletionBacklog", volumeGuid, relativePath )
					
					if deleteSuccessful then
						deletionBacklog:removeFromDeletionBacklog( volumeGuid, relativePath )
					end
					
				end

				for i = 1, #assetOverlays do
				
					self._migrationProgress:reportSuccess( currentStepOfProgress( self ) )
					self._migrationCount = self._migrationCount + 1
					
				end
				
				return true
				
			end )
	
	else
	
		return Observable.just( true )
		
	end

end


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

--[[!
	\class NmbFileMover
]]--

local NmbFileMover = {
	
	_docStore = false,
		-- (WFRxDocStore)
	
	_indexStore = false,
		-- (WFRxIndexStore)
	
	_synchronizer = false,
		-- (WFCatSynchronizer)
	
	_fileSpec = false,
		-- (WFCatFileSpec)
	
	_deletionBacklog = false,
		-- (NmbDeletionBacklog)
		
	_isFreemiumMigration = false,
		-- (boolean) True if file move was trigger via a change in catalog ID.
	
	_keepAllOriginals = false,
		-- (boolean) True if the user has chosen to keep all originals locally.
	
	_keepAllProxies = false,
		-- (boolean) True if the user has chosen to keep all proxies locally.
	
	_volumeGuidForOriginals = false,
		-- (string) The GUID of the volume where originals are stored.
		
	_relativeBasePathForOriginals = false,
		-- (string) The relative base path for where originals are stored.
		
	_volumeGuidForProxies = false,
		-- (string) The GUID of the volume where proxies are stored.
		
	_relativeBasePathForProxies = false,
		-- (string) The relative base path for where proxies are stored.
		
	_disableHardLinks = false,
		-- (string) If true, don't use hard links when moving files across same volume.
		
	_filesInFlightCount = false,
		-- (number) The current number of file migrations in progress.
		
	_filePathOverlays = false,
		-- (table) A table of doc store overlays containing updated file paths, which
		-- are periodically written to the doc store.
		
	_assetsToDelete = false,
		-- (table) A table containing file paths to delete once the file path overlay
		-- for the asset has been applied.
		
	_bulkAnnotationUpdaterStream = false,
		-- (Rx Subscription) A subscription to a stream that periodically updates the
		-- doc store with file path overlays and deletes old assets.
		
	_fileMoveComplete = false,
		-- (Rx ReplaySubject) A local stream that fires when a file is moved.
		
	_migrationSubscription = false,
		-- (Rx Subscription) 
		
	_migrationCount = false,
		-- (number) THe number of successful file migrations.
		
	_missingFiles = false,
		-- (table) The paths of original files skipped because the were not found at expected location
		
	_recentlyMigratedFiles = false,
		-- (table) key: localId, value: true. Contains entries for recently migrated files.
		
	_failedFileDeniedListHelper = false,
		-- (WFCatDeniedListHelper) A helper classed used to temporarily deny files from being moved.
		
	_running = false,
		-- (boolean) Indicates if the migration is running.
	
	_migrationProgress = false,
		-- (WFProgressReporter) used for migration progress updates
	
	_statusOfVolumes = false,
		-- (table) A table of volumes statuses including connection status, and disk
		-- space status.
	
	_statusOfVolumesStream = false,
		-- (Rx ReplaySubject) A local stream that fires when a volumes' status changes.
	
	_forgettingVolumes = false,
		-- (boolean) Indicates if we're currently in the process of forgetting a volume.
		
	_forgottenVolumeStream = false,
		-- (Rx Subject) A local stream that fires once a volume has been forgetten.
	
	_currentStep = false,
		-- (string) A string that represents the current phase, or step, of the file mover.
	
	_fileMoveSubscriptions = false,
		-- (table) A table of Rx subscriptions for each individual file move.
	_fileMoveLogPath = false,
		-- (string) A string path to the migration folder inside the catalog where fileMover.log is stored.
	class = {
		generateIndexSchema = generateIndexSchema,
	},

	WFCloseSupport.addCloseSupportMethods,

	private = {
		
		-- The following methods are internal, and are only exposed for testing.
		
		_migrationComplete = migrationComplete,
		_allVolumesOfflineOrOutOfSpace = allVolumesOfflineOrOutOfSpace,
		_getOnlineGuidBasePathsFromVolumes = getOnlineGuidBasePathsFromVolumes,
		_monitorDestinationVolumes = monitorDestinationVolumes,
		_monitorSourceVolumes = monitorSourceVolumes,
		_getSourceBasePathsAndSizes = getSourceBasePathsAndSizes,
		_getSourceVolumeStatusFromBasePaths = getSourceVolumeStatusFromBasePaths,
		_buildMigrationQuery = buildMigrationQuery,
		_buildWhereClause = buildWhereClause,
	}

}

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

--[[!

	\brief Creates a new NmbFileMover instance.

	\param docStore (WFRxDocStore)
	\param indexStore (WFRxIndexStore)
	\param fileSpec (WFCatFileSpec)
	\param deletionBacklog (NmbDeletionBacklog)
	\param keepAllOriginals (boolean)
	\param keepAllProxies (boolean)
	\param optArgs (?table) A table of optional arguments only used by unit tests.
	
	\return NmbFileMover object
	
--]]

function NmbFileMover:__init( docStore, indexStore, fileSpec, deletionBacklog, keepAllOriginals, keepAllProxies, optArgs )

	check( 'NmbFileMover', 'WFRxDocStore', 'WFRxIndexStore', 'WFCatFileSpec', 'NmbDeletionBacklog', 'boolean', 'boolean', '?table' )
	
	logger:info( "Constructing NmbFileMover" )
	
	optArgs = optArgs or {}
	
	self._docStore = docStore
	
	self._indexStore = indexStore
	
	self._fileSpec = fileSpec
	
	self._deletionBacklog = deletionBacklog
	
	self._isFreemiumMigration = false
	
	self._keepAllOriginals = keepAllOriginals
	
	self._keepAllProxies = keepAllProxies
	
	self:setOwned {
	
		_failedFileDeniedListHelper = WFCatDeniedListHelper( 
				self._docStore, 
				'NmbFileMover',
				numberOfTriesBeforeDenyingFile,
				failedMigrationBlackoutWindowInSeconds,
				true )

	}
	
	local originalsPathInfo = optArgs.originalsPathInfo or self._fileSpec:originalsPathInfo()
	
	self._volumeGuidForOriginals = originalsPathInfo.volumeGuid
	self._relativeBasePathForOriginals = originalsPathInfo.relativePath

	local proxiesPathInfo = optArgs.proxiesPathInfo or self._fileSpec:proxiesPathInfo()

	self._volumeGuidForProxies = proxiesPathInfo.volumeGuid
	self._relativeBasePathForProxies = proxiesPathInfo.relativePath

	self._disableHardLinks = optArgs.disableHardLinks or false

	self._migrationProgress = WFProgressReporter( "Migrating Files" )

	self._fileMoveComplete = ReplaySubject( 1 )
	self._fileMoveComplete:onNext( true )
	
	self._statusOfVolumesStream = ReplaySubject( 1 )
	
	self._forgottenVolumeStream = Subject()
	
	self._filesInFlightCount = 0

	self._migrationCount = 0
	
	self._missingFiles = {}

	self._recentlyMigratedFiles = {}
	
	self._filePathOverlays = {}
	
	self._assetsToDelete = {}
	
	self._statusOfVolumes = {}

	self._fileMoveSubscriptions = {}

	self._bulkAnnotationUpdaterStream = 
			Observable.timer( 0, bulkApplyOverlayPeriod, WFRxScheduler.sharedScheduler() )
					:flatMap( function()
					
						local assetOverlays = self._filePathOverlays
						self._filePathOverlays = {}

						local assetsToDelete = self._assetsToDelete
						self._assetsToDelete = {}

						return updateFileAnnotationsAndDeleteFiles( self, assetOverlays, assetsToDelete )
						
					end )
					:subscribe()

	self._fileMoveLogPath = KSPathUtils.child( self._fileSpec:rootPath() , "migration")
	
end

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

local function createOverlayAndGatherFileInfo( self, localId, ozHasOriginal,
		binaryType, binarySize, newPath, oldPath )

	logger:infof( "Planning to update the file path annotation for asset %d to %s", 
			localId, newPath )

	local newAnnotation

	if newPath == 'delete' then

		if not ozHasOriginal then

			-- Never remove an overlay for an asset that doesn't have an original on oz.
			-- I don't think the rest of the code should be hitting this if it is
			-- it should probably be fixed, at a minimum this is here to prevent future mistakes.
			-- Note that the only place that should be removing the annoation is when the user
			-- selects forget original volume.  forgetVolume below doesn't call into here.

			logFileMoverMessage( self, "error", "Attempted to remove annotation for file not on Oz for binary type %s size %s", binaryType, binarySize )

			-- If we are failing to act on it we just deny it otherwise the
			-- migrate will not complete.
			self._failedFileDeniedListHelper:addToDeniedList( localId )

			return nil, nil

		end

		logFileMoverMessage( self, "warn", "Removing annotation for binary type %s size %s", binaryType, binarySize )

		newAnnotation = { _delete = true }
		
	else
	
		local volumeGuid = newPath and WFFileUtils.getVolumeGuidForPath( newPath )
		local relativeFilePath = newPath and WFFileUtils.removeVolumeFromPath( newPath )

		if volumeGuid and relativeFilePath then

			newAnnotation = { 
				relativePath = relativeFilePath,
				volumeGuid = volumeGuid
			}
		
			self._deletionBacklog:removeFromDeletionBacklog( volumeGuid, relativeFilePath )
				-- If the file has been previously added to the deletion backlog, clear it out
				-- so we don't delete it.
		
		else
			
			logFileMoverMessage(self, "warn","Unable to update file path annotation for asset %s to %s", localId, newPath)	

			-- If we are failing to act on it we just deny it otherwise the
			-- migrate will not complete.
			self._failedFileDeniedListHelper:addToDeniedList( localId )

			return nil, nil
			
		end
	
	end
	
	local sizeArray = {}
	sizeArray[ binarySize ] = newAnnotation
	
	local updatedFiles = {}
	updatedFiles[ binaryType ] = sizeArray 
	
	local overlay = {
		_id = localId,
		_localOnly = {
			files = updatedFiles,
		},
	}
	
	local assetToDeleteInfo

	-- Don't delete the asset if it's source and destination are the same
	if oldPath and oldPath ~= newPath then
	
		logger:infof( "Planning to delete the file at path %s", oldPath )
		
		assetToDeleteInfo = {
			localId = localId,
			binaryType = binaryType,
			pathToDelete = oldPath,
		}
		
	end
	
	return overlay, assetToDeleteInfo

end

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


local function deleteCorruptedPathElements(self, path )

	-- If a target drive is disconnected during a file move, it can result in corrupted 
	-- folders that appear as files. If this happens, subsequent file moves will fail
	-- unless those corrupted files are removed.
	
	local previousPathToCheck
	local pathToCheck = path
	
	while pathToCheck and pathToCheck ~= '' and pathToCheck ~= previousPathToCheck do
	
		-- The path elements should always be folders. If the path element is a file,
		-- it's a corrupted folder and should be removed.
	
		if WFFileUtils.exists( pathToCheck ) == 'file' then
			logFileMoverMessage( self, "info", "Deleting corrupted file %s", pathToCheck )
			WFFileUtils.deleteFile( pathToCheck )
		end
	
		previousPathToCheck = pathToCheck
		pathToCheck = KSPathUtils.parent( pathToCheck )
	
	end
	
end

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

local function getVolumeName( guid )

	local volumeName = WFFileUtils.getVolumeNameFromGuid( guid )

	if not volumeName then

		volumeName = NmbFileManagementSettings.getReadableIdentifierForVolumeGuid( guid )
				or guid
		
	end
	
	return volumeName

end

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

local function migrateOrPurge( self, currentPath, newPath, newPathVolumeGuid, migrationRequired, useHardLink, isVideo )
	
	if not migrationRequired and not useHardLink then
	
		if derefOpt( self._statusOfVolumes, 'volumesOverDiskSpaceTarget', newPathVolumeGuid ) or
			derefOpt( self._statusOfVolumes, 'volumesLowOnSpace', newPathVolumeGuid ) or
			derefOpt( self._statusOfVolumes, 'volumesOutOfSpace', newPathVolumeGuid ) then
	
			-- If we're above target disk space usage, only copy the required assets
			-- or assets that can be migrated via hard links.
			logFileMoverMessage( self, "info", "Disk space target exceeded - Deleting asset %s instead of copying it.", currentPath)	
			
			return Observable.just( 'delete' )
	
		elseif isVideo then
	
			-- Only copy over required video files, or videos that can be migrated via hard links,
			-- purge the rest. This is entirely for performance.

			logFileMoverMessage( self, "info" ,"Deleting video %s instead of copying it.", currentPath )	
			
			return Observable.just( 'delete' )
		
		end
	
	end
	
	logger:infof( "Now copying file from %s to %s!", currentPath, newPath )
	
	-- Double-check that the target volume is still connected.
	
	if not WFFileUtils.isVolumeGuidOnline( newPathVolumeGuid ) then
		local message = "Target volume disconnected!" 
		logFileMoverMessage( self, "debug", message)	
		return Observable.throw( message )
	end
	
	-- Since we're copying the file, we need to create the parent directory.

	local parentDir = parent( newPath )

	deleteCorruptedPathElements(self, parentDir )
		-- Delete any corrupted path elements that might cause the following mkdir to fail.
		
	WFFileUtils.mkdir( parentDir )
	
	-- SPAM [jandrew 2017-02-07]: We intentionally use `copyToUniquelyNamedFile` instead of
	-- `moveToUniquelyNamedFile` to maintain a very specific order of operations during the
	-- migration process.

	return WFAsyncFileUtils.copyToUniquelyNamedFile( currentPath, newPath, useHardLink )
		:tap( function( copiedToPath )
			if copiedToPath == newPath then
				logFileMoverMessage( self, "info", "Finished copying file from %s to %s", currentPath, newPath )
			else
				logFileMoverMessage( self, "warn", "Finished copying file WARNING Unique path used %s to %s", currentPath, copiedToPath )
			end
		end )
end

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

local function maybeMigrateFile( self, localId, ozHasOriginal, currentPath, currentPathVolumeGuid,
							newPath, newPathVolumeGuid, binaryType, binarySize, 
							migrationRequired, isVideo, maybeSHA256 )
	
	logFileMoverMessage( self, "info", "Attempting to migrate file from %s to %s",  currentPath, newPath )
	markAsRecentlyMigrated( self, localId, binaryType )
	
	local newParentFolder = leafName( calculateBaseRootPath( newPath ) )
	local currentParentFolder = leafName( calculateBaseRootPath( currentPath ) )
	
	self._migrationProgress:reportInfo( {
		oldVolumeGuid = currentPathVolumeGuid,
		oldVolume = getVolumeName( currentPathVolumeGuid ),
		oldParentFolder = currentParentFolder,
		newVolumeGuid = newPathVolumeGuid,
		newVolume = getVolumeName( newPathVolumeGuid ),
		newParentFolder = newParentFolder,
		currentStep = self._currentStep
	} )
	
	local useHardLink = 
		not self._disableHardLinks 
		and WFPlatformInfo.isCocoa
		and currentPathVolumeGuid == newPathVolumeGuid
		and WFFileUtils.volumeForPath( newPath ) == '/'
		-- On Mac and iOS, we use a hard link during the file copy when the new path is on the same
		-- volume as the old path. This greatly speeds up the copy operation.
		-- TODO: Hardlinks don't currently work on Windows, nor do they work on external Mac volumes.

	local obs
	if WFFileUtils.exists( newPath ) then
		
		logFileMoverMessage( self, "info", "File %s already exists in destination %s. Checking SHA256..", currentPath, newPath )
		-- If a file already exists at the newPath, check if it has the expected SHA256. If it matches,
		-- we assume that we're confused and recovering from a partial move scenario.
	
		local expectedSHA256Stream = maybeSHA256 and Observable.just( maybeSHA256 ) 
				or WFFileUtils.exists( currentPath ) and WFCrypto.asyncDigestFile( currentPath, 'SHA256' )
				or Observable.just( false )
				
		local actualSHA256Stream = WFCrypto.asyncDigestFile( newPath, 'SHA256' )

		obs = Observable.combineLatest( 
		
				function( expectedSHA256, actualSHA256 )
			
					return expectedSHA256 == actualSHA256

				end, expectedSHA256Stream, actualSHA256Stream )
				
				:flatMapLatest( function( match )
				
					if match then
					
						logFileMoverMessage( self, "info", "SHA matches, no need to perform copy." )
						return Observable.just( newPath )
					
					else

						if binaryType == 'proxy' then

							-- Proxy files have unique names based on asset id so don't let copy
							-- hyphenate the file name to make a copy, just delete the file.

							-- An unknown file exists in the target location. Delete the file, as it's
							-- likely a corrupt image created by a previous partial copy. This can occur
							-- if a user disconnects the target drive mid-copy.

							local ok, err = pcall( WFFileUtils.deleteFile, newPath )

							if not ok then
								logFileMoverMessage( self, "warn", "Unable to delete unknown file at %s, error %s.", newPath, err )
							end

						end

						-- Different original files can be imported with the same capture date and file name.
						-- When this happens we need to leave the existing file there and let the
						-- copy operation copy it to a new name hiphenated with a -2 or higher suffix.

						logger:info( "The SHA does not match. Maybe performing file copy.." )

						return migrateOrPurge( self, currentPath, newPath, newPathVolumeGuid, 
								migrationRequired, useHardLink, isVideo )

					end
				
				end )

	elseif not WFFileUtils.exists( currentPath ) then

		-- If nothing exists at the current path ( or the new path ), simply delete the existing annotation.

		logFileMoverMessage( self, "warn", "No file exists at %s", currentPath )

		local fileIsMissing = true

		table.insert( self._missingFiles, currentPath )

		if ozHasOriginal then

			obs = Observable.just( 'delete', fileIsMissing )

		else

			logFileMoverMessage( self, "warn", "updating annotation because ozHasOriginal is false but no file exists at %s", currentPath )

			obs = Observable.just( newPath, fileIsMissing )

		end

	else

		obs = migrateOrPurge( self, currentPath, newPath, newPathVolumeGuid,
			migrationRequired, useHardLink, isVideo )

	end
	
	return obs
	
		:tap( function()
		
			migrationCompletedOrDenied( self, localId, binaryType )
				-- Mark the file move as complete so the next file move can start.
				
		end )
	
		:delay( 1, WFRxScheduler.sharedScheduler() )
			-- SPAM [jandrew 2017-04-19]: This delay was added to deal with with a strange
			-- bug. If you disconnect the target volume just as a file copy completes,
			-- there is a chance the file will become corrupt on the external volume even
			-- though the copy succeeded and we were able to verify the file attributes of 
			-- the copied image.
			--
			-- Bkruse suggested that this is due to fsync and write caching. With write caching, 
			-- even if whatever you're transferring is "done," there may still be data hanging
			-- out in RAM that hasn't yet been written to disk. "Eject"ing a volume forces
			-- flushing these chached bits to disk.
			--
			-- This delay, coupled with the file attribute check below, guards against this. A
			-- cleaner solution would be to force flush the cache as part of the file copy
			-- operation. I've looked into this, but I don't see any mechanism to do this
			-- via NSFileManager for OSX. Perhaps someone with more Obj-C knowledge has
			-- some clues. This is really only an issue on OSX, as write caching is disabled
			-- by default on Windows for thumb drives.
			
		:tap( function( filePath, fileIsMissing )
		
			if not filePath then

				logFileMoverMessage( self, "error" , "File path missing for docId: %s", localId  )

			else

				-- Note WFFileUtils.fileAttributes will return an empty table if the file is missing
				-- so missing files and 'delete' will not set fileAttributesInvalid to true.
				-- Only a 0 length file is considered invalid here.

				local fileAttributes = WFFileUtils.fileAttributes( filePath )
				local fileAttributesInvalid = not fileAttributes or fileAttributes.fileSize == 0

				if fileAttributesInvalid then

					logFileMoverMessage( self, "error" , "File attributes of copied file are wrong: %s", filePath  )

				else

					local overlay, assetInfo = createOverlayAndGatherFileInfo( self, localId, ozHasOriginal,
							binaryType, binarySize, filePath, currentPath )

					if overlay and not fileIsMissing then
						-- Clear localOriginalError to retry uploading the original if a file exists.
						KSTableUtils.setNested( overlay, { _delete = true }, '_localOnly', 'localOriginalError' )
					end

					table.insert( self._filePathOverlays, overlay )
					table.insert( self._assetsToDelete, assetInfo )

				end

			end

		end )
		
		:catch( function( errorMessage )

			logFileMoverMessage( self, "error", "Error copying file from %s to %s: %s", currentPath, newPath, errorMessage)
		
			if not self._failedFileDeniedListHelper:maybeDeny( localId ) then
			
				-- File move failed, but it's not yet denied. Retry the file move.
				logFileMoverMessage( self, "warn", "File move failed for asset %d. Now retrying file move..", localId )
				
				local retryDelay = 3
					-- This slight delay allows for the system to properly detect and propegate any
					-- disconnected volume signals _before_ we actually perform the retry.
				
				return Observable.timer( retryDelay, WFRxScheduler.sharedScheduler() )
				
					:take( 1 )
				
					:flatMapLatest( function()
					
						return maybeMigrateFile( self, localId, ozHasOriginal, currentPath, currentPathVolumeGuid,
								newPath, newPathVolumeGuid, binaryType, binarySize, 
								migrationRequired, isVideo, maybeSHA256 )
				
					end )
				
			else
				
				logFileMoverMessage( self, "warn", "File has been denied from migration %s.", currentPath )
				
				migrationCompletedOrDenied( self, localId, binaryType )
				
				return Observable.just( false )
				
			end
		
		end )
		
end

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

--[[!

	\brief Removes the file annotations on a specified volume for all documents.
	
	In other words, if there are any assets that live on the specified volume, we
	simply delete those file paths from the doc store. We essentially forget all assets
	on the specified volume. 
	
	\param volumeGuid (string) - The guid of the volume to forget.
	
	\return (Rx Observable) - Completes when all originals and proxies found on the
		target volume have been forgotten.
	
--]]

function NmbFileMover:forgetVolume( volumeGuid )

	check( 'NmbFileMover', '+string' )
	
	local forgetOriginalsOnVolume = self._indexStore:queryOnce {
			sql = [[SELECT localId
				FROM assetPathsAndInfo
				WHERE originalVolumeGuid = ?]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true,
			volumeGuid,
		}
		:flatMap( function( originalDocIds )
		
			local overlay = {
				_localOnly = {  
					files = {
						original = { _delete = true }
					}
				}
			}

			return WFRxThrottledBatchedApplyOverlayToIds( "forget originals on volume", self._docStore, overlay, originalDocIds, 100, 10 )
		
		end )
		
	local forgetProxiesOnVolume = self._indexStore:queryOnce {
			sql = [[SELECT localId
				FROM assetPathsAndInfo
				WHERE proxyVolumeGuid = ?]],
			singleColumnQuery = true,
			quiesceUntilUpToDate = true,
			volumeGuid,
		}
		:flatMap( function( proxyDocIds )
		
			local overlay = {
				_localOnly = {  
					files = {
						proxy = { _delete = true }
					}
				}
			}
		
			return WFRxThrottledBatchedApplyOverlayToIds( "forget proxies on volume", self._docStore, overlay, proxyDocIds, 100, 10 )
		
		end )

	self._forgettingVolumes = true
	
	return forgetOriginalsOnVolume
		:merge( forgetProxiesOnVolume )
		:takeLast( 0 )
		:tap( function()
		
			self._forgettingVolumes = false
			self._forgottenVolumeStream:onNext( volumeGuid )
		
		end )
		:call( runUntilDone )
		
end

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

local function doesSHAMatchAssetAtPath( self, sha256, newPath )

	check( 'NmbFileMover', '+string', '+string' )

	if WFFileUtils.exists( newPath ) then

		return sha256 == WFCrypto.digestFile( newPath, 'SHA256' )

	end

	return false

end

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

local function remapOriginals( self, originalGuidBasePath, stream, throttleSize, 
								offset, assetsToRemap )

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

	throttleSize = throttleSize or 50
	offset = offset or 0

	assetsToRemap = assetsToRemap or {
		overlays = {},
		assetsToDelete = {}
	}

	if WFCloseSupport.hasBeenOrWillBeClosed( self._indexStore ) then
		
		stream:onNext( true )
		stream:onCompleted()
		
		return Observable.just( true )
		
	end

	-- Fetch a batch of originals that might need to be remapped. 

	return self._indexStore:queryOnce {
		sql = [[SELECT localId, originalVolumeGuid, fileName, captureDate, sha256, ozHasOriginal
				FROM assetPathsAndInfo
				WHERE localId > ? AND originalGuidBasePath != ? AND originalGuidBasePath IS NOT NULL
				ORDER BY localId ASC
				LIMIT ? ]], 
			offset,
			originalGuidBasePath,
			throttleSize,
		quiesceUntilUpToDate = true
	}
	
	:flatMap( function( docs )
	
		-- For each original, check to see if the original is on an offline volume.

		for _, doc in ipairs( docs ) do
		
			local localId = doc[ 1 ]
			local originalVolumeGuid = doc[ 2 ]
			local fileName = doc[ 3 ]
			local captureDate = doc[ 4 ] 
			local sha256 = doc[ 5 ]
			local ozHasOriginal = toBool( doc[ 6 ] )
			
			if sha256 and not WFFileUtils.isVolumeGuidOnline( originalVolumeGuid ) then
			
				-- If the original an offline volume, check if the original exists in the expected location.
			
				local originalBasePath = WFFileUtils.createFullPathFromRelativePathAndGuid(
						self._relativeBasePathForOriginals, self._volumeGuidForOriginals )
		
				local expectedOriginalLocation = originalBasePath and WFCatPixelPathGenerator.pathForOriginal( 
						originalBasePath, fileName, captureDate, nil, true )

				if originalBasePath and doesSHAMatchAssetAtPath( self, sha256, expectedOriginalLocation ) then
								
					local overlay, assetToDelete = createOverlayAndGatherFileInfo( self, localId, ozHasOriginal,
							'original', 'master', expectedOriginalLocation )

					-- Clear localOriginalError to retry uploading the original.
					if overlay then
						KSTableUtils.setNested( overlay, { _delete = true }, '_localOnly', 'localOriginalError' )
					end

					table.insert( assetsToRemap.overlays, overlay )
					table.insert( assetsToRemap.assetsToDelete, assetToDelete )
					
				else
				
					self._migrationProgress:reportSuccessMustDeliver( currentStepOfProgress( self ) )
				
				end

			else
			
				self._migrationProgress:reportSuccessMustDeliver( currentStepOfProgress( self ) )
				
			end

			offset = localId
					
		end
		
		if #docs > 0 then
		
			return Rx.Observable.defer( function()

				return remapOriginals( self, originalGuidBasePath, stream, throttleSize, offset, assetsToRemap )
					-- Process the next batch of originals.

			end )
			
		else
		
			return updateFileAnnotationsAndDeleteFiles( self, assetsToRemap.overlays, assetsToRemap.assetsToDelete )
			
				:tap( function()
				
					stream:onNext( true )
					stream:onCompleted()
					
				end )
				
		end
		
	end )

	:call( runUntilDone )

end

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

-- SPAM [jandrew 2017-03-01]: The following function was created to handle the
-- hard drive upgrade scenario: a user manually moves their files from one volume
-- ( Volume A ) to a new volume ( Volume B ). They they disconnect Volume A, launch
-- Nimbus, and point the original storage path to the new location on Volume B.
-- In this case, the originals already exist in the correct location, the paths
-- just need to be updated in the doc store.

local function maybeRemapOriginals( self )

	check( 'NmbFileMover' )

	local originalGuidBasePath = buildGuidBasePath( self._volumeGuidForOriginals, self._relativeBasePathForOriginals )

	-- Calculate how many originals need to be remapped or moved.

	return self._indexStore:queryOnce {
			sql = [[SELECT count( localId )
					FROM assetPathsAndInfo
					WHERE originalGuidBasePath != ? AND originalGuidBasePath IS NOT NULL]], 
				originalGuidBasePath,
			quiesceUntilUpToDate = true
		}
		
		:map( function( rows )
	
			-- Report the total via the progress reporter observable.
	
			local totalToRemapOrMove = rows[ 1 ][ 1 ] or 0
			self._migrationProgress:setTotal( totalToRemapOrMove, true, currentStepOfProgress( self ) )
			return totalToRemapOrMove
			
		end )
		
		:flatMap( function( totalToRemapOrMove )
		
			-- If we have assets to remap, attempt to remap them.
		
			if totalToRemapOrMove > 0 then
			
				if self._synchronizer then
					self._synchronizer:stopSyncing()
						-- Don't sync while moving files.
				end
			
				local remapOriginalsCompletionStream = Rx.Subject()
			
				remapOriginals( self, originalGuidBasePath, remapOriginalsCompletionStream )
				
				return remapOriginalsCompletionStream
				
			else
				return Observable.just( true )
			end
		
		end )

end

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

local function remapProxies( self, proxyGuidBasePath, stream, throttleSize, 
								offset, assetsToRemap )

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

	throttleSize = throttleSize or 50
	offset = offset or 0

	assetsToRemap = assetsToRemap or {
		overlays = {},
		assetsToDelete = {}
	}

	if WFCloseSupport.hasBeenOrWillBeClosed( self._indexStore ) then
		
		stream:onNext( true )
		stream:onCompleted()
		
		return Observable.just( true )
		
	end

	-- Fetch a batch of proxies that might need to be remapped. 

	return self._indexStore:queryOnce {
		sql = [[SELECT docId, localId, proxyVolumeGuid, proxyType, ozHasOriginal
				FROM assetPathsAndInfo
				WHERE localId > ? AND proxyGuidBasePath != ? AND proxyGuidBasePath IS NOT NULL
				ORDER BY localId ASC
				LIMIT ? ]], 
			offset,
			proxyGuidBasePath,
			throttleSize,
		quiesceUntilUpToDate = true
	}
	
	:flatMap( function( docs )
	
		-- For each proxy, check to see if the proxy is on an offline volume.

		for _, doc in ipairs( docs ) do
		
			local docId = doc [ 1 ]
			local localId = doc[ 2 ]
			local proxyVolumeGuid = doc[ 3 ]
			local proxyType = doc[ 4 ]
			local ozHasOriginal = toBool( doc[ 5 ] )
			
			if not WFFileUtils.isVolumeGuidOnline( proxyVolumeGuid ) then
			
				-- If the proxy in on an offline volume, check if a proxy exists in the expected location.
				
				local proxyBasePath = WFFileUtils.createFullPathFromRelativePathAndGuid(
										self._relativeBasePathForProxies, self._volumeGuidForProxies )
										
				local expectedProxyLocation = proxyBasePath and WFCatPixelPathGenerator.pathForProxyTypeAndUuid( 
										proxyBasePath, proxyType, docId )

				local overlay, assetToDelete

				if proxyBasePath and WFFileUtils.exists( expectedProxyLocation ) then
				
					-- Since we can only ever download the proxy from Oz, we don't need to perform a SHA256
					-- check.
									
					overlay, assetToDelete = createOverlayAndGatherFileInfo( self, localId, ozHasOriginal,
							'proxy', proxyType, expectedProxyLocation )
					
				else
				
					-- If the proxy is not in the expected location, just delete the proxy annotation from asset.
					-- We can redownload it from Oz if needed.

					overlay, assetToDelete = createOverlayAndGatherFileInfo( self, localId, ozHasOriginal,
							'proxy', proxyType, 'delete' )
				
				end

				table.insert( assetsToRemap.overlays, overlay )
				table.insert( assetsToRemap.assetsToDelete, assetToDelete )

			else
			
				self._migrationProgress:reportSuccessMustDeliver( currentStepOfProgress( self ) )
				
			end

			offset = localId
					
		end
		
		if #docs > 0 then
		
			return Rx.Observable.defer( function()

				return remapProxies( self, proxyGuidBasePath, stream, throttleSize, offset, assetsToRemap )
					-- Process the next batch of originals.

			end )
			
		else
		
			return updateFileAnnotationsAndDeleteFiles( self, assetsToRemap.overlays, assetsToRemap.assetsToDelete )
			
				:tap( function()
				
					stream:onNext( true )
					stream:onCompleted()
					
				end )
				
		end
		
	end )

	:call( runUntilDone )

end

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

local function maybeRemapProxies( self )

	check( 'NmbFileMover' )

	local proxyGuidBasePath = buildGuidBasePath( self._volumeGuidForProxies, self._relativeBasePathForProxies )

	-- Calculate how many proxies need to be remapped or moved.

	return self._indexStore:queryOnce {
			sql = [[SELECT count( localId )
					FROM assetPathsAndInfo
					WHERE proxyGuidBasePath != ? AND proxyGuidBasePath IS NOT NULL]], 
				proxyGuidBasePath,
			quiesceUntilUpToDate = true
		}
		
		:map( function( rows )
	
			-- Report the total via the progress reporter observable.
	
			local totalToRemapOrMove = rows[ 1 ][ 1 ] or 0
			self._migrationProgress:setTotal( totalToRemapOrMove, true, currentStepOfProgress( self ) )
			return totalToRemapOrMove
			
		end )
		
		:flatMap( function( totalToRemapOrMove )
		
			-- If we have assets to remap, attempt to remap them.
		
			if totalToRemapOrMove > 0 then
			
				if self._synchronizer then
					self._synchronizer:stopSyncing()
						-- Don't sync while moving files.
				end
			
				local remapProxiesCompletionStream = Rx.Subject()
			
				remapProxies( self, proxyGuidBasePath, remapProxiesCompletionStream )
				
				return remapProxiesCompletionStream
				
			else
				return Observable.just( true )
			end
		
		end )

end

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

--[[!

	\brief Migrates files to the new hard drive location.
	
	This component is responsible for migrating assets from their old locations to
	the new user-defined locations. Code coverage of NmbFileMover is close to 100%.

	NmbFileMover's key method is maybePerformMigration. When maybePerformMigration is called,
	we first setup some streams that periodically monitor the source and destination 
	drives and report their online status.

	The source drives are determined by looking at the existing file paths within the
	index store. From those paths, we can easily determine the unique source drives
	that we need to migrate assets from.

	Once we know which drives are online, we fire off an index store query that looks
	for assets that need to be migrated. The query filters out assets that are going
	to, or coming from, offline drives. If any new drives come online, or if a
	connected drive drops offline, the query is automatically refreshed. This allows
	someone to connect a drive mid-migration and for Nimbus to handle it smoothly.

	Before we perform a file copy on a given asset, we first check to see if that asset
	already exists in the destination path. If it does exist, and has a matching SHA,
	we simply update the doc store annotation and delete the file from the source drive.

	When we've migrated everything that we can ( to and from all drives that are
	connected ), we fire off a result that contains the following info:

		- Did the migration succeed.
		- A list of offline source drives that we were unable to migrate from, 
			including the number of assets on that drive.
		- A list of offline destination drives that we were unable to migrate to, 
			including the number of assets needed to migrate.
		- Number of assets migrated.
		- Denied assets
	
	\return (Rx Observable) - Completes when the migration completes.
	
--]]

function NmbFileMover:maybePerformMigration( migrationRequired )
	
	check( 'NmbFileMover', '?boolean' )

	if self._running then
		return self._migrationProgress:status()
	end
	
	self._running = true
	
	if self._migrationProgress:isDone() then
		self._migrationProgress:reset()
	end
	
	local hasSetTotal = false
	local complete = false
	
	-- If NmbFileMover is in the process of forgetting a volume ( See forgetVolume API )
	-- then we should wait until it finishes before trying to migrate any assets.
	
	local maybeForgettingVolumes = Observable.just( true )
	
	if self._forgettingVolumes then
		maybeForgettingVolumes = self._forgottenVolumeStream
	end
	
	if migrationRequired then
		self._currentStep = kIndexing
		self._migrationProgress:setTotal( 1, true, currentStepOfProgress( self ) )
	end
	
	-- Report if the move is caused by freemium catalog ID change
	-- to display correct UI.
	
	self._migrationProgress:reportInfo( {
		isFreemiumMigration = self._isFreemiumMigration
	} ,
	
	true)
	
	self._migrationSubscription = maybeForgettingVolumes
	
		:take( 1 )
	
		:flatMapLatest( function()

			-- Restore any archived files.

			self._currentStep = kRemapping
			
			return NmbPathRestorer.restoreArchivedPaths( self._docStore, self._indexStore, self._deletionBacklog, self._migrationProgress )

		end )
	
		:flatMapLatest( function()

			-- Remap originals.

			self._currentStep = kRemapping
			
			return maybeRemapOriginals( self )

		end )
		
		:flatMapLatest( function()

    		-- Remap proxies.
			
			self._currentStep = kRemapping
			
    		return maybeRemapProxies( self )

		end )

		:flatMapLatest( function()
		
			self._currentStep = kMoving
			
			return Rx.Observable.combineLatest(
	
				function( destinationVolumeStatuses, sourceVolumeStatuses )
					
					if not hasSetTotal then
					
						hasSetTotal = true
						
						local totalAssetsToMigrate = sourceVolumeStatuses.originalsToMigrate + 
							sourceVolumeStatuses.proxiesToMigrate
						
						logger:infof( "Need to migrate %d assets!", totalAssetsToMigrate )
						
						if totalAssetsToMigrate ~= 0 then
							-- Record migration session.
							logFileMoverMessage( self, "debug", "\n--------------- Migration Session %s ---------------" , KSDate.timeInIso8601Format() )
							logFileMoverMessage( self, "debug", "Migrating %s assets: %s originals and %s proxies \n", totalAssetsToMigrate, sourceVolumeStatuses.originalsToMigrate, sourceVolumeStatuses.proxiesToMigrate )
						end
						
						self._migrationCount = 0
						self._migrationProgress:setProcessedCount( 0, true )
						self._migrationProgress:setTotal( totalAssetsToMigrate, true, currentStepOfProgress( self ) )
							-- Track the total assets to migrate in order to track progress.
							
					end
					
					return destinationVolumeStatuses, sourceVolumeStatuses
					
				end,

				monitorDestinationVolumes( self._volumeGuidForOriginals, 
											self._volumeGuidForProxies, 
											self._statusOfVolumesStream ),
					-- Periodically monitor if the destination volumes are online.
					
				monitorSourceVolumes( self._indexStore, 
										self._volumeGuidForOriginals, 
										self._relativeBasePathForOriginals,
										self._volumeGuidForProxies, 
										self._relativeBasePathForProxies )
					-- Periodically monitor if the source volumes are online.
			) 
			
		end )
		
		:distinctUntilChanged( nil, distinctTableCompare )
			-- When the status changes for the multiple volumes, we need to refresh our query.
			
		:flatMapLatest( function( destinationVolumeStatuses, sourceVolumeStatuses )
		
			local originalSourceVolumeStatuses = sourceVolumeStatuses.originalSourceVolumeStatuses
			local proxySourceVolumeStatuses = sourceVolumeStatuses.proxySourceVolumeStatuses
		
			if allVolumesOfflineOrOutOfSpace( destinationVolumeStatuses ) then
			
				-- All destination volumes are offline, or out of space, so we can't migrate anything.
				
				return Observable.just( {},
						destinationVolumeStatuses,
						sourceVolumeStatuses )
				
			end
		
			if allVolumesOfflineOrOutOfSpace( originalSourceVolumeStatuses ) and
				allVolumesOfflineOrOutOfSpace( proxySourceVolumeStatuses ) then
				
				-- All source volumes are offline, so we can't migrate anything.
				
			 	return Observable.just( {},
						destinationVolumeStatuses,
						sourceVolumeStatuses )
				
			end
		
			-- Perform the initial check to see if there's enough space for the require assets
			-- on the destination drive(s).
			
			local enoughSpace = getSpaceConstrainedVolumes( self, destinationVolumeStatuses, sourceVolumeStatuses )
		
			if not enoughSpace then
			
				-- Not enough space on destination drive(s) for required assets, 
				-- so don't migrate anything.
			
				return Observable.just( {}, destinationVolumeStatuses, sourceVolumeStatuses )
			
			end
		
			local onlineOriginalGuidBasePaths = getOnlineGuidBasePathsFromVolumes( originalSourceVolumeStatuses )
			local onlineProxyGuidBasePaths = getOnlineGuidBasePathsFromVolumes( proxySourceVolumeStatuses )
		
			-- Build the index store query that returns the assets that are currently capable
			-- of being migrated ( source and destination drives are online ).
			
			local volumeGuidForOriginals = self._volumeGuidForOriginals
			local volumeGuidForProxies = self._volumeGuidForProxies
			
			local isOriginalDriveReady = destinationVolumeStatuses[ volumeGuidForOriginals ].online
					and not destinationVolumeStatuses[ volumeGuidForOriginals ].outOfSpace
			
			local isProxyDriveReady = destinationVolumeStatuses[ volumeGuidForProxies ].online
					and not destinationVolumeStatuses[ volumeGuidForProxies ].outOfSpace
			
			local assetPathsAndInfoQuery = buildMigrationQuery( 
					self._indexStore, 
					isOriginalDriveReady,
					isProxyDriveReady,
					onlineOriginalGuidBasePaths, 
					onlineProxyGuidBasePaths )
		
			return Rx.Observable.combineLatest(
			
				function( filesToMigrate )
				
					return filesToMigrate, destinationVolumeStatuses, sourceVolumeStatuses
					
				end,
				
				assetPathsAndInfoQuery,
				self._fileMoveComplete
			)

		end ) 

		:tap( function( filesToMigrate, destinationVolumeStatuses, sourceVolumeStatuses )

			if complete then
				return
			end

			local numberOfFilesToMigrate = #filesToMigrate

			logger:infof( "Processing batch of %d files to migrate.", numberOfFilesToMigrate )

			-- If the first row in the result is denied, _all_ files are denied.
			
			local firstFile = filesToMigrate[ 1 ]
			local allFilesAreDenied = firstFile and toBool( firstFile[ 12 ] )
			
			if allFilesAreDenied then
				allFilesAreDenied = self._failedFileDeniedListHelper:maybeResetDeniedListStatus( firstFile[ 2 ] )
			end
			
			if numberOfFilesToMigrate == 0 
				or allFilesAreDenied then
				
				-- If the current instance was due to a catalog ID update for freemium,
				-- reset _isFreemiumMigration so that any file move after the freemium 
				-- migration displays the proper UI.
				
				self._isFreemiumMigration = false
				
				complete = true
		
				return migrationComplete( self, destinationVolumeStatuses, sourceVolumeStatuses,
						allFilesAreDenied, numberOfFilesToMigrate )

			end

			for _, row in ipairs( filesToMigrate ) do

				local docId = row[ 1 ]
				local localId = row[ 2 ]
				local originalGuidBasePath = row[ 3 ]
				local originalVolumeGuid = row[ 4 ]
				local originalRelativePath = row[ 5 ]
				local fileName = row[ 6 ]
				local captureDate = row[ 7 ]
				local proxyGuidBasePath = row[ 8 ]
				local proxyVolumeGuid = row[ 9 ]
				local proxyRelativePath = row[ 10 ]
				local proxyType = row[ 11 ]
				local isMaybeDenied = toBool( row[ 12 ] )
				local sha256 = row[ 13 ]
				local isMarkedForOfflineUse = toBool( row[ 14 ] )
				local ozHasOriginal = toBool( row[ 15 ] )
				local isVideo = toBool( row[ 16 ] )
				
				local canMigrateOriginal = false
				local originalMigrationRequired = false
				
				local currentOriginalLocation
				local expectedOriginalLocation
				
				if originalVolumeGuid and originalRelativePath then
				
					local expectedOriginalGuidBasePath = buildGuidBasePath( self._volumeGuidForOriginals, self._relativeBasePathForOriginals )
				
					if originalGuidBasePath ~= expectedOriginalGuidBasePath and
						WFFileUtils.isVolumeGuidOnline( self._volumeGuidForOriginals ) and
						WFFileUtils.isVolumeGuidOnline( originalVolumeGuid ) then
							-- If either the source or destination path is offline, then our index store
							-- query hasn't yet updated to filter out these assets. We can ignore this
							-- asset until the query updates.
					
						currentOriginalLocation = WFFileUtils.createFullPathFromRelativePathAndGuid(
								originalRelativePath, originalVolumeGuid )
					
						if currentOriginalLocation then
						
							local originalBasePath
						
							originalBasePath = WFFileUtils.createFullPathFromRelativePathAndGuid(
									self._relativeBasePathForOriginals, self._volumeGuidForOriginals )
					
							expectedOriginalLocation = originalBasePath and WFCatPixelPathGenerator.pathForOriginal( 
									originalBasePath, fileName, captureDate, nil, true )
						
							canMigrateOriginal = not not expectedOriginalLocation
						
						end
						
					end
					
				end
				
				if canMigrateOriginal and 
					( not ozHasOriginal or isMarkedForOfflineUse or self._keepAllOriginals ) then
					originalMigrationRequired = true
				end
				
				local canMigrateProxy = false
				local proxyMigrationRequired = false
				
				local currentProxyLocation
				local expectedProxyLocation
				
				if proxyVolumeGuid and proxyRelativePath then
				
					local expectedProxyGuidBasePath = buildGuidBasePath( self._volumeGuidForProxies, self._relativeBasePathForProxies )
				
					if proxyGuidBasePath ~= expectedProxyGuidBasePath and
						WFFileUtils.isVolumeGuidOnline( self._volumeGuidForProxies ) and
						WFFileUtils.isVolumeGuidOnline( proxyVolumeGuid ) then
							-- If either the source or destination path is offline, then our index store
							-- query hasn't yet updated to filter out these assets. We can ignore this
							-- asset until the query updates.
		
						currentProxyLocation = WFFileUtils.createFullPathFromRelativePathAndGuid(
								proxyRelativePath, proxyVolumeGuid )
								
						if currentProxyLocation then
								
							local proxyBasePath
								
							proxyBasePath = WFFileUtils.createFullPathFromRelativePathAndGuid(
									self._relativeBasePathForProxies, self._volumeGuidForProxies )
									
							expectedProxyLocation = proxyBasePath and WFCatPixelPathGenerator.pathForProxyTypeAndUuid( 
									proxyBasePath, proxyType, docId )
								
							canMigrateProxy = not not expectedProxyLocation
						
						end
						
					end
					
				end

				-- To avoid a reindex consider a proxy as not being on oz if the original is not on oz
				-- the only cases where this may not be true is if an asset was downgraded to a proxy only
				-- asset marked with 'willNotUploadMaster' via a copy or trash operation where the original
				-- wasn't available locally and the original hadn't been uploaded yet.  But avoiding deleting
				-- the asset in this case would be rare and not harmful.
				local ozHasProxy = ozHasOriginal
				
				if canMigrateProxy and ( not ozHasProxy or isMarkedForOfflineUse or self._keepAllProxies ) then
					proxyMigrationRequired = true
				end
				
				if isMaybeDenied then
					isMaybeDenied = self._failedFileDeniedListHelper:maybeResetDeniedListStatus( localId )
				end
				
				local function clearSubscription( key )
					return function()
						self._fileMoveSubscriptions[ key ] = nil
					end
				end
				
				if not isMaybeDenied then
				
					if canMigrateProxy and
						not wasRecentlyMigrated( self, localId, 'proxy' ) and
						self._filesInFlightCount < maxFileMovesInParallel then
					
						self._filesInFlightCount = self._filesInFlightCount + 1
						
						local key = buildGuidBasePath( proxyVolumeGuid, proxyRelativePath )
						
						self._fileMoveSubscriptions[ key ] =
							maybeMigrateFile( self, localId, ozHasOriginal, currentProxyLocation, proxyVolumeGuid,
								expectedProxyLocation, self._volumeGuidForProxies, 'proxy', proxyType, 
								proxyMigrationRequired, isVideo )
								
							:subscribe( nil, clearSubscription( key ), clearSubscription( key ) )
						
						if self._filesInFlightCount == maxFileMovesInParallel then
							break
						end
						
					end

					if canMigrateOriginal and 
						not wasRecentlyMigrated( self, localId, 'original' ) and
						self._filesInFlightCount < maxFileMovesInParallel then
					
						self._filesInFlightCount = self._filesInFlightCount + 1

						local key = buildGuidBasePath( originalVolumeGuid, originalRelativePath )

						self._fileMoveSubscriptions[ key ] =
							maybeMigrateFile( self, localId, ozHasOriginal, currentOriginalLocation, originalVolumeGuid,
								expectedOriginalLocation, self._volumeGuidForOriginals, 'original', 'master', 
								originalMigrationRequired, isVideo, sha256 )
								
							:subscribe( nil, clearSubscription( key ), clearSubscription( key ) )
							
						if self._filesInFlightCount == maxFileMovesInParallel then
							break
						end
						
					end

				end
				
			end

		end )
		
		:tapOnError( function( errorMessage )
		
			logFileMoverMessage( self, "warn", "An unknown error occurred: %s", errorMessage )
			reportMigrationError( self, errorMessage )
		
		end )
		
		:subscribe()

	return self._migrationProgress:status()

end

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

--[[!

	\brief Report the status of the Nimbus Volumes

	\return (void)
	
--]]

function NmbFileMover:reportStatusOfVolumes( statusOfVolumes )

	check( 'NmbFileMover', 'table' )
	self._statusOfVolumes = statusOfVolumes
	self._statusOfVolumesStream:onNext( statusOfVolumes )
	
end

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

--[[!

	\brief Stops the NmbFileMover instance.
	
	\return (void)
	
--]]

function NmbFileMover:stop()

	check( 'NmbFileMover' )
	
	if self._synchronizer then
		self._synchronizer:startSyncing()
	end
	
	local migrationSubscription = self._migrationSubscription
	self._migrationSubscription = nil
	
	if migrationSubscription then
		migrationSubscription:dispose()
	end
	
end

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

--[[!

	\brief Sets the synchronizer instance.
	
	\return (void)
	
--]]

function NmbFileMover:setSynchronizer( synchronizer )

	check( 'NmbFileMover', 'WFCatSynchronizer' )
	self._synchronizer = synchronizer
	
end

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

--[[!

	\brief Sets if the instance is due to freemium upgrade
	
	\return (void)
	
--]]

function NmbFileMover:setFreemiumMigration( isMigrated )

	check( 'NmbFileMover', 'boolean' )
	self._isFreemiumMigration = isMigrated
	
end


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

--[[!

	\brief Close and stop the NmbFileMover instance.
	
	\return (void)
	
--]]

function NmbFileMover:close()

	check( 'NmbFileMover' )

	NmbFileMover.stop( self )

	local forgottenVolumeStream = self._forgottenVolumeStream
	self._forgottenVolumeStream = nil

	if forgottenVolumeStream then
		forgottenVolumeStream:dispose()
	end

	local fileMoveComplete = self._fileMoveComplete
	self._fileMoveComplete = nil
	
	if fileMoveComplete then
		fileMoveComplete:dispose()
	end

	local bulkAnnotationUpdaterStream = self._bulkAnnotationUpdaterStream
	self._bulkAnnotationUpdaterStream = nil
	
	if bulkAnnotationUpdaterStream then
		bulkAnnotationUpdaterStream:dispose()
	end
	
end

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

return Rx.Class 'NmbFileMover'( NmbFileMover )
