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

WFCatTrash.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2018 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains 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 are protected by all applicable
intellectual property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

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

	-- WF.core.logger
local WFLogger = require 'WFLogger'
local WFPerformanceUtils = require "WFPerformanceUtils"
local WFAnalytics = require "WFAnalytics"

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

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

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local LrAssetUtils = require 'LrAssetUtils'
local WFArguments = require 'WFArguments'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxBatchedBulkOperation = require "WFRxBatchedBulkOperation"
local WFRxThrottledBatchedApplyOverlayToIds = require 'WFRxThrottledBatchedApplyOverlayToIds'
local WFTableMerge = require 'WFTableMerge'

	-- WF.oz_client.managed_catalog
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatIdUtils = require 'WFCatIdUtils'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatTrashUtils = require 'WFCatTrashUtils'

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

local WFCatTrash = {}

local logger = WFLogger( 'WFCatTrash' )

local kLogTrashOperationsLongerThanSecs = 1
local kLogTrashQueryLongerThanSecs = .2

local underscoreByte = string.byte( '_' )
local kEventWorkflow = 'Cloud Trash'

local appendArray = KSTableUtils.appendArray
local deepCopy = KSTableUtils.deepCopy
local derefOpt = KSTableUtils.derefOpt
local setNested = KSTableUtils.setNested

local check = assert( WFArguments.check )

local KSDate_timeNowForAnimation = assert( KSDate.timeNowForAnimation )

local WFCatOzErrors_hasTrialOrSubscriptionExpiredError = assert( WFCatOzErrors.hasTrialOrSubscriptionExpiredError )
local WFCatOzErrors_hasStorageExceededError = assert( WFCatOzErrors.hasStorageExceededError )

local WFCatIndexUtils_isAssetOnOz = assert( WFCatIndexUtils.isAssetOnOz )
local WFCatIndexUtils_isAssetRevisionOnOz = assert( WFCatIndexUtils.isAssetRevisionOnOz )

local createFromInfo = assert( WFCatTrashUtils.createFromInfo )
local getCreateWithRevisionId = assert( WFCatTrashUtils.getCreateWithRevisionId )
local originalExistsOnOz = assert( WFCatTrashUtils.originalExistsOnOz )
local getCreateFromId = assert( WFCatTrashUtils.getCreateFromId )


local albumAssetTable = "album_asset"
if WFPlatformInfo.isDesktop then
	albumAssetTable = "album_asset_v2"
end

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

local function trashLogf( logger, message, ... )

	local maybeNum = select( 1, ... )
	local maybeTable = select( 2, ... )
	if type( maybeNum ) == 'number' and type( maybeTable ) == 'table' and select( '#', ... ) == 2 then

		local startOfTable = {}
		for i = 1, math.min( #maybeTable, 100 ) do
			startOfTable[ i ] = maybeTable[ i ]
		end
		logger:warnf( message, maybeNum, startOfTable )

	else
		logger:warnf( message, ... )
	end

end

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

local function isAssetOnOzOrWasPushed( docValue, pusherChangeCounter )

	-- We currently look at the pusher's change counter and compare it to a
	-- document's _changeCounter in order to handle the case where an asset
	-- has been pushed to oz but we have not yet recieved a response with
	-- the _serverProvidedMetadata to know it is on oz by inspecting the
	-- local doc value.

	-- Without doing this we would end up assuming the asset is not yet on oz
	-- and issue a create operation without referring to the source asset, this
	-- can result in the loss of binaries, and loss of album membership,
	-- comments and favorites.

	-- One example of this is described in NMB-10383.

	-- NMB-12791 demonstrated the reason why we need to use _initialChangeCounter.
	-- if the document gets a modification causing the change counter to be updated
	-- higher than where the metadatapusher counter is at we no longer considered
	-- the asset as being pushed to oz (which would cause a duplicate asset).

	local docValueChangeCounter = docValue._initialChangeCounter or docValue._changeCounter or 0

	return WFCatIndexUtils_isAssetOnOz( docValue )
		or ( docValueChangeCounter < pusherChangeCounter
			and not ( WFCatOzErrors_hasTrialOrSubscriptionExpiredError( docValue )
				or WFCatOzErrors_hasStorageExceededError( docValue ) ) )

end

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

local function isVideoSubtype( subtype )
	return subtype == 'video' or subtype == 'deleted_video'
end

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

function WFCatTrash.overlayForRemoveTrashInfo( docValue )
	return {
		_id = docValue._id,
		_localOnly = {
			universal = {
				trash = { _delete = true }
			}
		}
	}
end

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

local function makePermanentlyDeleteOverlaysAndFileDeletionList( catalog, isDeleteTrash, batchOfDocValuesToPermanentlyDelete, pusherChangeCounter, optAddFilesToDeleteFromDocValue )

	local trashAssetOverlays = {}
	local localFiles = {}

	local deletionFlag = isDeleteTrash and "isDeleteTrash" or "isDirectlyDelete"

	for i, docValue in ipairs( batchOfDocValuesToPermanentlyDelete ) do

		local trashAssetOverlay =  {
			_id = docValue._id,
			_deleted = true,
			_localOnly = {
				universal = {
					trash = {
						[ deletionFlag ] = true, -- aka manual purge
					}
				}
			}
		}

		if isAssetOnOzOrWasPushed( docValue, pusherChangeCounter ) and createFromInfo( docValue ) then
			setNested( trashAssetOverlay, { _delete = true }, '_localOnly', 'universal', 'trash', 'createFrom' )
		end

		trashAssetOverlays[ i ] = trashAssetOverlay

		if optAddFilesToDeleteFromDocValue then
			optAddFilesToDeleteFromDocValue( localFiles, docValue )
		end

	end

	return trashAssetOverlays, localFiles

end

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

local function deleteDocOverlaysFromQueryResult( rows )

	local overlays = {}

	-- Oz will take care of deleting the associated documents so apply
	-- the '_dontsendtooz' label, these overlays will be grouped with
	-- other overlays that don't have this set so apply it to each entry.

	for i, row in ipairs( rows ) do

		overlays[ i ] = {
			_id = row[ 1 ],
			_deleted = true,
			_newLabel = '_dontsendtooz',
		}

	end

	return overlays

end

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

local function deleteDocOverlaysFromSingleColumnQueryResult( rows )

	local overlays = {}

	-- Oz will take care of deleting the associated documents so apply
	-- the '_dontsendtooz' label, these overlays will be grouped with
	-- other overlays that don't have this set so apply it to each entry.

	for i, docId in ipairs( rows ) do

		overlays[ i ] = {
			_id = docId,
			_deleted = true,
			_newLabel = '_dontsendtooz',
		}

	end

	return overlays

end

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

local function updateDocOverlaysFromQueryResult( docStore, rows, assetIdToNewAssetIdMap )

	local overlays = {}

	for i, row in ipairs( rows ) do

		local docId = row[ 1 ]
		local assetId = row[ 2 ]

		local newAssetId = assetIdToNewAssetIdMap[ assetId ]

		if newAssetId then

			overlays[ #overlays + 1 ] = {
				_id = docId,
				_preserveLabels = true,
				asset = {
					id = newAssetId
				}
			}

		else
			logger:errorf( "Trash assetId doesn't exist for assetId %s", assetId )
		end

	end

	return overlays

end

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

local function updateAlbumAssetIdOverlays( catalog, assetIds, assetIdToNewAssetIdMap )

	-- Note this function is used to update the asset ids in album assets,
	-- it is used for trash assets to handle the case where a user could be
	-- offline and we temporarily save off the album assets. Oz will delete these
	-- once it sends down new album asset ids that it generates.
	-- And this is also used for when assets are restored, in this case the
	-- restored album assets will just update the trash album_assets.

	return catalog:indexStore():queryOnce {
		-- get all the album_assets
		sql = [[SELECT aa.docId, aa.assetId FROM ]] .. albumAssetTable .. [[ AS aa JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = aa.assetId]],
		quiesceUntilUpToDate = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return updateDocOverlaysFromQueryResult( catalog:docStore(), rows, assetIdToNewAssetIdMap )
	end )

end

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

local function updateCommentAssetsIdOverlays( catalog, assetIds, assetIdToNewAssetIdMap )

	return catalog:indexStore():queryOnce {
		-- get all the comments related to this asset
		sql = [[SELECT comment.docId, comment.assetId FROM comment JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = comment.assetId]],
		quiesceUntilUpToDate = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return updateDocOverlaysFromQueryResult( catalog:docStore(), rows, assetIdToNewAssetIdMap )
	end )

end

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

local function updateFavoriteAssetIdOverlays( catalog, assetIds, assetIdToNewAssetIdMap )

	return catalog:indexStore():queryOnce {
		-- get all the favorites related to this asset
		sql = [[SELECT spacefavorite.docId, spacefavorite.assetId FROM spacefavorite JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = spacefavorite.assetId]],
		quiesceUntilUpToDate = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return updateDocOverlaysFromQueryResult( catalog:docStore(), rows, assetIdToNewAssetIdMap )
	end )

end

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

local function updateVersionIdOverlays( catalog, assetIds, assetIdToNewAssetIdMap )

	return catalog:indexStore():queryOnce {
		-- get all the versions
		sql = [[SELECT versions.docId, assetId FROM versions JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = versions.assetId]],
		quiesceUntilUpToDate = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return updateDocOverlaysFromQueryResult( catalog:docStore(), rows, assetIdToNewAssetIdMap )
	end )

end

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

function WFCatTrash.updateVersionIdOverlays( catalog, assetIds, assetIdToNewAssetIdMap )

	return updateVersionIdOverlays( catalog, assetIds, assetIdToNewAssetIdMap )

end

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

local function deleteAlbumAssetOverlays( catalog, assetIds )

	return catalog:indexStore():queryOnce {
		-- get all the album_assets
		sql = [[SELECT aa.docId FROM ]] .. albumAssetTable .. [[ AS aa JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = aa.assetId]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return deleteDocOverlaysFromSingleColumnQueryResult( rows )
	end )

end

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

local function deleteCommentAssetOverlays( catalog, assetIds )

	return catalog:indexStore():queryOnce {
		-- get all the comments
		sql = [[SELECT comment.docId FROM comment JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = comment.assetId]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return deleteDocOverlaysFromSingleColumnQueryResult( rows )
	end )

end

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

local function deleteFavoriteAssetOverlays( catalog, assetIds )

	return catalog:indexStore():queryOnce {
		-- get all the favorites related to this asset
		sql = [[SELECT spacefavorite.docId, spacefavorite.assetId FROM spacefavorite JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = spacefavorite.assetId]],
		quiesceUntilUpToDate = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
		return deleteDocOverlaysFromQueryResult( rows )
	end )

end

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

local function deleteVersionAssetOverlays( catalog, assetIds, deleteLocalSettingsFiles )

	return catalog:indexStore():queryOnce {
		-- get all the versions
		sql = [[SELECT versions.docId, assetId, externalXmpRelativePath FROM versions JOIN CARRAY( ?, ?, 'char*' ) AS ai ON ai.value = versions.assetId]],
		quiesceUntilUpToDate = true,
		assetIds,
		#assetIds,
	}
	:map( function( rows )
	
		if deleteLocalSettingsFiles then
		
			local catalogId = catalog:ozCatalogId()
		
			for _, row in ipairs( rows ) do
			
				local externalXmpRelativePath = row[ 3 ]
			
				local externalXmpAbsolutePath = WFCatAssetUtils.absolutePathToDevelopSettings( catalogId, externalXmpRelativePath )

				-- On Desktop, this cleanup is handled by the PixelProviderFileCachePurgeWorker.

				if WFPlatformInfo.isMobile() and externalXmpAbsolutePath then

					WFFileUtils.deleteFile( externalXmpAbsolutePath )

				end
			
			end
		
		end
	
		return deleteDocOverlaysFromQueryResult( rows )
	
	end )

end

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

local function deleteVersionAssetOverlaysAndDeleteLocalSettingsFiles( catalog, assetIds )

	return deleteVersionAssetOverlays( catalog, assetIds, true )

end

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

local function createNewDocOverlaysFromQueryResult( docStore, assetIds )

	return docStore:bulkDocValue( assetIds )
	:map( function( batchOfDocValues )

		for i, copyOfDocValue in ipairs( batchOfDocValues ) do

			for k, _ in pairs( copyOfDocValue ) do

				-- album_assets, comments and favorites probably don't have any
				-- _localOnly data but if they do in the future, preserve it.

				if string.byte( k, 1 ) == underscoreByte and k ~= '_localOnly' then
					copyOfDocValue[ k ] = nil
				end

			end

			copyOfDocValue._id = KSIdGenerator.uuidInLowercaseHex()

		end

		return batchOfDocValues

	end )

end

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

local function copyAlbumAssetOverlays( catalog, assetId )

	return catalog:indexStore():queryOnce {
		-- get all the album_assets
		sql = [[SELECT docId FROM ]] .. albumAssetTable .. [[ WHERE assetId = ?]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
		assetId
	}
	:flatMap( function( docIds )
		return createNewDocOverlaysFromQueryResult( catalog:docStore(), docIds )
	end )

end

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

local function copyCommentOverlays( catalog, assetId )

	return catalog:indexStore():queryOnce {
		-- get all the comments related to this asset
		sql = [[SELECT docId FROM comment WHERE assetId = ?]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
		assetId
	}
	:flatMap( function( docIds )
		return createNewDocOverlaysFromQueryResult( catalog:docStore(), docIds )
	end )

end

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

local function copyFavoriteOverlays( catalog, assetId )

	return catalog:indexStore():queryOnce {
		-- get all the favorites related to this asset
		sql = [[SELECT docId FROM spacefavorite WHERE assetId = ?]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
		assetId
	}
	:flatMap( function( docIds )
		return createNewDocOverlaysFromQueryResult( catalog:docStore(), docIds )
	end )

end

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

local function copyVersionOverlays( catalog, assetId )

	return catalog:indexStore():queryOnce {
		-- get all the versions related to this asset
		sql = [[SELECT docId FROM versions WHERE assetId = ?]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
		assetId
	}
	:flatMap( function( docIds )
	
		-- SPAM [jandrew 2020-01-22]: 
		-- If the Versions' develop settings are externalized and no longer local or on Oz, there's
		-- no way we can copy this document successfully. However, it looks like we'll copy the
		-- asset even if the local/oz original/proxy are not available either.
	
		return createNewDocOverlaysFromQueryResult( catalog:docStore(), docIds )
		
	end )

end

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

function WFCatTrash.moveAssociatedDocsToNewDocs( catalog, assetId )

	-- Create a copy of the associated docs as they probably have been deleted
	-- on other clients.  Don't just send an update as other clients
	-- may issue a subsequent delete via ConflictResolver if they saw an update
	-- for the deleted doc (album_assets, comments favorites).

	local batchOfOverlays = {}

	return Observable.combineLatest(
		function( albumAssetOverlays, commentOverlays, favoriteOverlays, versionOverlays,
			albumAssetOverlaysToDelete, commentOverlaysToDelete, favoriteOverlaysToDelete,
			versionOverlaysToDelete )

			appendArray( batchOfOverlays, albumAssetOverlays )
			appendArray( batchOfOverlays, commentOverlays )
			appendArray( batchOfOverlays, favoriteOverlays )
			appendArray( batchOfOverlays, versionOverlays )
			appendArray( batchOfOverlays, albumAssetOverlaysToDelete )
			appendArray( batchOfOverlays, commentOverlaysToDelete )
			appendArray( batchOfOverlays, favoriteOverlaysToDelete )
			appendArray( batchOfOverlays, versionOverlaysToDelete )

			return batchOfOverlays

		end,
		copyAlbumAssetOverlays( catalog, assetId ),
		copyCommentOverlays( catalog, assetId ),
		copyFavoriteOverlays( catalog, assetId ),
		copyVersionOverlays( catalog, assetId ),
		deleteCommentAssetOverlays( catalog, { assetId } ),
		deleteAlbumAssetOverlays( catalog, { assetId } ),
		deleteFavoriteAssetOverlays( catalog, { assetId } ),
		deleteVersionAssetOverlays( catalog, { assetId } )
	)

end

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

local function createCopyOfAssetDocValue( docValue )

	-- Note this copies the _localOnly section to the result to preserve
	-- the local files and other fields.

	local copyOfDocValue = deepCopy( docValue )

	for k, _ in pairs( copyOfDocValue ) do
		if string.byte( k, 1 ) == underscoreByte
			and k ~= '_localOnly' then
				copyOfDocValue[ k ] = nil
		end
	end

	copyOfDocValue._id = KSIdGenerator.uuidInLowercaseHex()

	return copyOfDocValue

end

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

local function setCreateWithRevId( origDocValue, trashDocOrRestoredDoc, assetIsOnOz )

	-- Specify _newRevId on the docValue and save it createWithRevId below. This
	-- is done in case another revision is added to the asset before the
	-- restore is issued. This is done to make sure a subsequent update
	-- will be sent with the modifications.

	-- It is also needed for trash asset to handle the case where you
	-- you do a modify delete sync.

	-- This revision id will be applied to the newly created asset (trash or restored).

	-- Note: createWithRevId does not belong in the createFrom section because
	-- it refers to information about the asset that the new asset is created from.
	-- createWithRevId is about the asset being created.

	local createWithRevId = KSIdGenerator.uuidInLowercaseHex()
	trashDocOrRestoredDoc._newRevId = createWithRevId
	setNested( trashDocOrRestoredDoc, createWithRevId, '_localOnly', 'universal', 'trash', 'createWithRevId')

	-- To handle docs that have been modified and then subsequently deleted and
	-- restored while offline (possibly multiple times)
	-- Generate a new revId to indicate that we should split the document
	-- into two revisions in addOverlaysForPreviouslyModifiedDocValues.

	-- The first revision won't ever go to oz but it's id will be used as
	-- the initial restore or trash asset revision id and conflict resolution
	-- will cause the second revision to be sent as an update.

	if not assetIsOnOz then

		local origDocCreateWithRevId = getCreateWithRevisionId( origDocValue )
		if origDocCreateWithRevId and origDocValue._revId ~= origDocCreateWithRevId then
			trashDocOrRestoredDoc._newRevId = KSIdGenerator.uuidInLowercaseHex()
		end

	end

end

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

local function setOrUpdateCreateFrom( docValue, docIsOnOz, docValueRevId, trashDocOrRestoredDoc )

	local originalId = docValue._id

	-- If creating a trash or restored asset from an asset that isn't on oz use
	-- it's prior createFrom info so that this asset will be created from the
	-- originally deleted asset that existed on oz (if there was one).

	if docIsOnOz then

		-- This is information that is used for the asset that the trash
		-- or restored asset will be created from.

		local createFrom = {
			id = originalId,
			revision_id = docValueRevId or docValue._revId,
			subtype = docValue.subtype
		}

		setNested( trashDocOrRestoredDoc, createFrom, '_localOnly', 'universal', 'trash', 'createFrom' )

	else

		-- createFrom was already copied from the original docValue in
		-- createCopyOfAssetDocValue (if it had one).
		-- So no need to set it here and we don't want to clear it as we want
		-- the trash document to refer to the original asset on oz.
		-- This is done to eliminate intermediate offline delete restore pairs
		-- for pattern [delete restore]+ delete

		docIsOnOz = originalExistsOnOz( docValue )
		if docIsOnOz then
			originalId = getCreateFromId( docValue )			
		end

	end

	if docIsOnOz then

		-- keep track of the last asset that this was generated from in the event
		-- that we need to create an intermediate asset to satisfy oz's requirement
		-- that a restore must be from a trash asset, and a delete generating a trash
		-- asset must be from a non-trash asset.
		setNested( trashDocOrRestoredDoc, docValue._id, '_localOnly', 'universal', 'trash', 'createFrom', 'lastAssetId' )
		setNested( trashDocOrRestoredDoc, docValue._revId, '_localOnly', 'universal', 'trash', 'createFrom', 'lastRevisionId' )
		setNested( trashDocOrRestoredDoc, docValue.userCreated, '_localOnly', 'universal', 'trash', 'createFrom', 'lastTrashUserCreated' )
		setNested( trashDocOrRestoredDoc, docValue.purgeDate, '_localOnly', 'universal', 'trash', 'createFrom', 'lastTrashPurgeDate' )

		-- Copy the links over so we know whether we have already uploaded them, and in case
		-- we need to download from them if the user is out of storage or the account is expired.

		local binaryLinksToCopy = {
			'/rels/master',
			'/rels/proxy_type/720p',
			'/rels/proxy_type/panorama',
			'/rels/proxy_type/2560',
			'/rels/xmp/develop',
			'/rels/xmp',
		}

		for _, linkName in ipairs( binaryLinksToCopy ) do

			local linkToBinaryOnOz = derefOpt( docValue, '_serverProvidedMetadata', 'links', linkName )
			if linkToBinaryOnOz then
				setNested( trashDocOrRestoredDoc, linkToBinaryOnOz, '_localOnly', 'universal', 'trash', 'createFrom', 'links', linkName )
				setNested( trashDocOrRestoredDoc, true, '_localOnly', 'universal', 'trash', 'uploaded', linkName)
			end

		end

	end

	if trashDocOrRestoredDoc._id == getCreateFromId( trashDocOrRestoredDoc ) then
		logger:errorf( "doc createFrom id == doc._id at %s\n\nDoc= %s", ':traceback:', trashDocOrRestoredDoc )
	end

	return originalId

end

WFCatTrash.setOrUpdateCreateFrom = setOrUpdateCreateFrom

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

-- There are some _localOnly values that are specific to the particular docValue/asset id
-- and when we're copying to another docValue/asset id shouldn't apply. These are things like
-- in progress binary upload state, etc.  This should remove those fields from the passed in 
-- docValue
local function removeDocSpecificLocalOnlyData( docValue, removeViaOverlay )
	
	local deletionModifier = removeViaOverlay and { _delete = true } or nil
	
-- For mobile, remove chunked upload info, so trash could restart upload from the beginning
	if WFPlatformInfo.isAndroid then

		if derefOpt( docValue, '_localOnly', 'originalFilePath' ) ~= nil then
			setNested( docValue, deletionModifier, '_localOnly', 'chunkedUpload' )
			setNested( docValue, deletionModifier, '_localOnly', 'original', 'uploaded' )
		end

	elseif WFPlatformInfo.isiOS then

		if derefOpt( docValue, '_localOnly', 'files', 'master' ) ~= nil then
			setNested( docValue, deletionModifier, '_localOnly', 'files', 'master', 'chunkedUpload' )
			setNested( docValue, deletionModifier, '_localOnly', 'files', 'master', 'uploaded' )
		end

		if derefOpt( docValue, '_localOnly', 'files', 'thumbnail2x' ) ~= nil then
			setNested( docValue, deletionModifier, '_localOnly', 'files', 'thumbnail2x' )
		end
		
		if derefOpt( docValue, '_localOnly', 'files', 'preview' ) ~= nil then
			setNested( docValue, deletionModifier, '_localOnly', 'files', 'preview' )
		end
		
		if derefOpt( docValue, '_localOnly', 'files', 'xmp_develop' ) ~= nil then
			setNested( docValue, deletionModifier, '_localOnly', 'files', 'xmp_develop' )
		end

	end
	
	return docValue
	
end

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

function WFCatTrash.copyLocalFilesOverlayForUnmodifiedAndFullyUploadedDoc( docValue, newId )

	-- This is only called when we don't have any binaries or modifications
	-- pending so set the uploaded status as true to prevent uploading a binary
	-- while we wait for the _serverProvidedMetadata.links from oz.

	local localFilesOverlay = {
		_id = newId,
		_localOnly = {
			files = deepCopy( derefOpt( docValue, '_localOnly', 'files' ) ),
		}
	}

	if WFPlatformInfo.isAndroid then

		localFilesOverlay = {
			_id = newId,
			_localOnly = {
				originalFilePath = deepCopy( derefOpt( docValue, '_localOnly', 'originalFilePath' ) ),
				latestOriginal = deepCopy( derefOpt( docValue, '_localOnly', 'latestOriginal' ) ),
				original = deepCopy( derefOpt( docValue, '_localOnly', 'original' ) ),
				proxy = deepCopy( derefOpt( docValue, '_localOnly', 'proxy' ) ),
			}
		}

	end

	local binaryLinksToNotUpload = {
		'/rels/master',
		'/rels/proxy_type/2560',
		'/rels/xmp/develop',
		'/rels/xmp',
	}

	for _, linkName in ipairs( binaryLinksToNotUpload ) do
		setNested( localFilesOverlay, true, '_localOnly', 'universal', 'trash', 'uploaded', linkName)
	end

	removeDocSpecificLocalOnlyData( localFilesOverlay )

	return localFilesOverlay

end

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

local function convertAssetToTrashAsset( docValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate, createAsOverlay )
	
	-- if we're creating a docValue the deletion modifier is just nil, if we're creating an overlay it
	-- should be { _delete = true }
	local deletionModifier = createAsOverlay and { _delete = true } or nil
	
	local trashDocOrOverlay = createAsOverlay and { _id = KSIdGenerator.uuidInLowercaseHex() } or createCopyOfAssetDocValue( docValue )
	
	-- deleted doc should be same as original except for
	-- id, subtype, and userCreated.
	-- Use passed in timeOfDelete so deleted assets could be batched together.

	trashDocOrOverlay.type = 'asset'
	trashDocOrOverlay.subtype = isVideoSubtype( docValue.subtype ) and 'deleted_video' or 'deleted_image'
	trashDocOrOverlay.userCreated = timeOfDelete
	trashDocOrOverlay.userUpdated = timeOfDelete
	trashDocOrOverlay.purgeDate = purgeDate

	local userCreated = docValue.userCreated
		or derefOpt( docValue, 'importSource', 'importTimestamp' )
		or derefOpt( docValue, '_serverProvidedMetadata', 'created' )

	setCreateWithRevId( docValue, trashDocOrOverlay, docIsOnOz )

	local originalId = setOrUpdateCreateFrom( docValue, docIsOnOz, originalNewRevId, trashDocOrOverlay )

	trashDocOrOverlay.original = {
		id = originalId,
		userCreated = userCreated,
	}

	trashDocOrOverlay = removeDocSpecificLocalOnlyData( trashDocOrOverlay, createAsOverlay )

	if WFPlatformInfo.isDesktop then

		-- Trash assets are considered removed from the album so clear out
		-- isMarkedForOfflineUse even though we have kept track of their album
		-- membership.
		setNested( trashDocOrOverlay, deletionModifier, '_localOnly', 'isMarkedForOfflineUse' )
		
		-- Remove the ozError from assets when moving them to trash if
		-- it's an out of storage and expired account error as trash doesn't
		-- count against cloud storage.
		if WFCatOzErrors_hasStorageExceededError( docValue )
			or WFCatOzErrors_hasTrialOrSubscriptionExpiredError( docValue ) then

			setNested( trashDocOrOverlay, deletionModifier, '_localOnly', 'ozError' )

		end

	end

	return trashDocOrOverlay
	
end

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

function WFCatTrash.createTrashAssetFromDocValue( docValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate )
	
	return convertAssetToTrashAsset( docValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate, false )

end

local createTrashAssetFromDocValue = WFCatTrash.createTrashAssetFromDocValue

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

local function createTrashAssetOverlayFromDocValue( docValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate )
	
	return convertAssetToTrashAsset( docValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate, true )

end

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

function WFCatTrash.updateLocalCreateFromTargetingOzDoc( docStore, localTrashDoc, ozTrashDoc )

	-- Update the createFrom section of the local trash asset so we will attempt
	-- to create it from the new oz asset instead. This is used by
	-- WFCatErrorHandler when running into resourceNotFoundError.

	local overlay = {
		_id = localTrashDoc._id,
	}

	local oppositeTrashSubtype = { image = 'deleted_image', video = 'deleted_video', deleted_image = 'image', deleted_video = 'video' }
	if ozTrashDoc._deleted then
		-- Make the createFrom good enough for the pusher to attempt the
		-- action even though we know this one will fail as well but we
		-- will get a newer new_id on the next error that will hopefully work.

		-- Swap the subtype of createFrom since we are moving one asset
		-- forward based on the new asset id provided by oz.
		local createFrom = WFCatTrashUtils.createFromInfo( localTrashDoc )
		ozTrashDoc.subtype = oppositeTrashSubtype[ createFrom.subtype ]

	end

	local ozTrashDocNewIdToAdd = localTrashDoc._id
	local intermediateDoc

	if ozTrashDoc.subtype == localTrashDoc.subtype then

		-- Need to create an intermediate asset as Oz doesn't allow
		-- a trash asset to be created from another trash asset and
		-- doesn't allow you to restore an asset from another regular asset.
		-- Currently don't need type and subtype but added them in case something
		-- looks at the deleted asset.
		local intermediateDocRevId = KSIdGenerator.uuidInLowercaseHex()
		intermediateDoc = {
			_id = KSIdGenerator.uuidInLowercaseHex(),
			_revId = intermediateDocRevId,
			_deleted = true,
			_newId = localTrashDoc._id,
			type = 'asset',
			subtype = oppositeTrashSubtype[ ozTrashDoc.subtype ],
			_newLabel = '_dontsendtooz'
		}

		-- Locally connect the oz trash doc to the intermediate asset
		-- via the setDocValue below.
		ozTrashDocNewIdToAdd = intermediateDoc._id

		-- Normally we copy documents during trash operations but since we are
		-- just interested in making a minimal overlay to update the createFrom info
		-- we can call this on the two assets to get the same result. We need lastAssetId to
		-- be different from create from id, as the pusher will use lastAssetId for
		-- the intermediate update.
		WFCatTrash.setOrUpdateCreateFrom( ozTrashDoc, true, ozTrashDoc._revId, overlay )
		intermediateDoc._localOnly = KSTableUtils.deepCopy( overlay._localOnly )
		setNested( intermediateDoc, intermediateDocRevId, '_localOnly', 'universal', 'trash', 'createWithRevId')

		WFCatTrash.setOrUpdateCreateFrom( intermediateDoc, false, intermediateDocRevId, overlay )

		-- WFCatTrash.setOrUpdateCreateFrom is expecting to see a _revId above, but the
		-- docStore wants us to supply it in _newRevId when creating it.
		intermediateDoc._newRevId = intermediateDocRevId
		intermediateDoc._revId = nil

	else

		WFCatTrash.setOrUpdateCreateFrom( ozTrashDoc, true, ozTrashDoc._revId, overlay )

	end

	if WFCatTrashUtils.getCreateWithRevisionId( localTrashDoc ) == nil then
		setNested( overlay, localTrashDoc._revId, '_localOnly', 'universal', 'trash', 'createWithRevId')
	end

	return docStore:applyOverlay( overlay )
	:flatMap( function()

		local docValuesToSet = {}

		if intermediateDoc then
			docValuesToSet = { intermediateDoc }
		end

		if ozTrashDoc._deleted and not ozTrashDoc._newId then
			-- This also needs to point the ozTrashDoc at the localTrashDoc._id (or the intermediate asset)
			-- so that the new RNFE will know what local document to update
			-- or delete.
			ozTrashDoc._newId = ozTrashDocNewIdToAdd
			ozTrashDoc._empty = nil
			ozTrashDoc._newLabel = '_dontsendtooz'
			docValuesToSet[ #docValuesToSet + 1 ] = ozTrashDoc
		end

		if #docValuesToSet == 0 then
			return Observable.just( true )
		end

		return docStore:bulkSetDocValue( docValuesToSet )

	end )

end

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

function WFCatTrash.createOverlaysToMergeRestoredAssets( catalog, localDoc, ozDoc )
	
	-- Initial implementation here is a really simple process
	-- of preferring the local document's payload. Given that we have two separate docs
	-- we don't have a known common history so a more nuanced merge is not necessarily 
	-- possible
	local copyOfLocalDocForOverlay = createCopyOfAssetDocValue( localDoc )
	
	local userCreated = localDoc.userCreated
		or derefOpt( localDoc, 'importSource', 'importTimestamp' )
		or derefOpt( localDoc, '_serverProvidedMetadata', 'created' )

	setCreateWithRevId( localDoc, copyOfLocalDocForOverlay, true )

	local originalId = setOrUpdateCreateFrom( localDoc, false, nil, copyOfLocalDocForOverlay )

	copyOfLocalDocForOverlay.original = {
		id = originalId,
		userCreated = userCreated,
	}
	
	-- remove _spm
	copyOfLocalDocForOverlay._spm = nil
	
	-- clean up some _localOnly data
	copyOfLocalDocForOverlay = removeDocSpecificLocalOnlyData( copyOfLocalDocForOverlay )
	
	-- remove aesthetics and autoTags if present, those are not client populated
	copyOfLocalDocForOverlay.aesthetics = nil
	copyOfLocalDocForOverlay.autoTags = nil
	
	-- use type, subtype, id, userCreated of ozDoc
	-- It's possible that error from oz conains new_id not locally existed yet. Use localDoc as fallback
	copyOfLocalDocForOverlay._id = ozDoc._id
	copyOfLocalDocForOverlay.type = ozDoc.type or localDoc.type
	copyOfLocalDocForOverlay.subtype = ozDoc.subtype or localDoc.subtype
	copyOfLocalDocForOverlay.userCreated = ozDoc.userCreated or nil -- setting to nil to not overlay a value
	copyOfLocalDocForOverlay.userUpdated = KSDate.timeInIso8601Format()
	
	local overlayToDeleteLocalRestoredAsset = {
		_id = localDoc._id,
		_deleted = true
	}


	local batchOfAssetIds = { localDoc._id }
	local assetIdToTrashAssetIdMap = { [ localDoc._id ] = ozDoc._id }
	local batchOfOverlays = { copyOfLocalDocForOverlay, overlayToDeleteLocalRestoredAsset }
	
	return Observable.combineLatest(
				function( albumAssetOverlays, commentOverlays, favoriteOverlays, versionOverlays )

					appendArray( batchOfOverlays, albumAssetOverlays )
					appendArray( batchOfOverlays, commentOverlays )
					appendArray( batchOfOverlays, favoriteOverlays )
					appendArray( batchOfOverlays, versionOverlays )
					
					return batchOfOverlays

				end,
				updateAlbumAssetIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap ),
				updateCommentAssetsIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap ),
				updateFavoriteAssetIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap ),
				updateVersionIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap )
			)

end

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

local function stripIrrelevantFieldsFromDiff( diff )

	diff._changeCounter = nil
	diff._initialChangeCounter = nil
	diff._revId = nil
	diff._labels = nil

end

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

local function createOverlaysForAssetsMovedToTrash( catalog, pusherChangeCounter, timeOfDelete, purgeDate, batchOfDocValuesToDelete, gatherBinaryFilesToMoveFn )

	-- Take a buffer of original docValues and create overlays to create
	-- trash docValues which contain an entire copy of most of the original asset,
	-- an overlay for the tombstones of the original assets being deleted,
	-- and, if necessary, an overlay for any modifications that were made to 
	-- the asset being deleted that were not yet synced to oz.

	-- When we issued the operations separately it was possible for both the
	-- trash and original asset to exist after the app died resulting in
	-- many duplicated assets when batch sizes were greater than 100.

	local modifiedAssetIds = {}
	local positionOfModifiedAssetForId = {}
	local batchOfFilesToMoveArray = {}

	for i, docValue in ipairs( batchOfDocValuesToDelete ) do

		local originalExistsOnOz = docValue._serverProvidedMetadata ~= nil

		if originalExistsOnOz and not WFCatIndexUtils_isAssetRevisionOnOz( docValue, docValue._revId ) then

			-- We need to make two revisions of modified documents that exist on oz
			-- as the delete operation in WFCatMetadataPusher isn't uploading any
			-- document changes in the payload (it just refers to the asset and revId
			-- for both deleted and trash asset.

			-- So make a list of assets that need updates and use their 'oz' doc value
			-- as the starting value for the original trash asset update.

			local nextModifiedAssetId = #modifiedAssetIds + 1
			modifiedAssetIds[ nextModifiedAssetId ] = docValue._id

			-- Since all assets are not modified make a look up table we will
			-- check against later when we make the trash asset.
			positionOfModifiedAssetForId[ docValue._id ] = nextModifiedAssetId
		end

	end

	-- Get the oz version of assets that were modified before they were deleted
	-- and use that as the starting doc value instead of the current.
	modifiedAssetIds.label = 'oz'
	local modifiedDocValues = #modifiedAssetIds > 0 and
		catalog:docStore():bulkDocValue( modifiedAssetIds ) or
			Observable.just( {} )

	return modifiedDocValues:map( function( batchOfOzDocValues )

		local assetOverlays = {}
		local assetIdToNewAssetIdMap = {}

		for i, docValue in ipairs( batchOfDocValuesToDelete ) do

			local originalNewRevId = KSIdGenerator.uuidInLowercaseHex() -- revId of tombstone

			local docIsOnOz = isAssetOnOzOrWasPushed( docValue, pusherChangeCounter )

			local ozDocValue = batchOfOzDocValues[ positionOfModifiedAssetForId[ docValue._id ] ]

			local trashAssetOverlay, trashAssetModificationOverlay
			
			if ozDocValue then
				-- We have an Oz docValue that is different from our local docValue. This means that the
				-- initial trash asset that Oz creates will be based off the Oz docValue so we need to create 
				-- an overlay that will set our local trash asset to the same state and then a second overlay
				-- to apply our local modifications on top of that
				local baseTrashAssetOverlay = createTrashAssetFromDocValue( ozDocValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate )
				
				local docValueToOzDocDiff = WFTableMerge.createDiff( docValue, ozDocValue )
				stripIrrelevantFieldsFromDiff( docValueToOzDocDiff )
				
				trashAssetOverlay = WFTableMerge.threeWayMerge( {}, baseTrashAssetOverlay, docValueToOzDocDiff )
				
				-- recreate the diff that brings the Oz version of the doc up to the latest local version
				-- so that we can apply the overlay and allow the eventual conflict resolution to ensure
				-- that the local modifications get applied to the trash asset and 
				trashAssetModificationOverlay = WFTableMerge.createDiff( ozDocValue, docValue )
				trashAssetModificationOverlay._id = trashAssetOverlay._id
				stripIrrelevantFieldsFromDiff( trashAssetModificationOverlay )

			else
			
				trashAssetOverlay = createTrashAssetOverlayFromDocValue( docValue, docIsOnOz, originalNewRevId, timeOfDelete, purgeDate )

			end

			if gatherBinaryFilesToMoveFn then
				
				if WFPlatformInfo.isDesktop then
					gatherBinaryFilesToMoveFn( catalog, docValue, WFTableMerge.applyOverlay( deepCopy( docValue ), trashAssetModificationOverlay or trashAssetOverlay), batchOfFilesToMoveArray )
				else
					gatherBinaryFilesToMoveFn( catalog, docValue, trashAssetModificationOverlay or trashAssetOverlay, batchOfFilesToMoveArray )
				end
			end

			-- Map the original asset id to the new trash asset id so we
			-- know how to update the associated docs (album_assets, comments, favorites).
			assetIdToNewAssetIdMap[ docValue._id ] = trashAssetOverlay._id
	
			-- Delete the original asset, however don't send the delete's to oz, the trash
			-- assets create will be swapped out with a delete of the original asset that
			-- they refer to. This is done to eliminate the need of sending intermediate
			-- delete/restore/delete's when the user repeats the sequence while offline.
			assetOverlays[ #assetOverlays + 1 ] = {
				_id = docValue._id,
				_newRevId = originalNewRevId,
				_newLabel = '_dontsendtooz',
				_deleted = true,
				_newId = trashAssetOverlay._id,
				_newIdOverlay = trashAssetOverlay,
				_localOnly = {
					deletedLocally = true, -- signals that file deletions are handled
				}
			}

			-- If there are any, apply modifications that took place to the original
			-- asset that happened after it was last pushed up before it was deleted.
			if trashAssetModificationOverlay then
				assetOverlays[ #assetOverlays + 1 ] = trashAssetModificationOverlay
			end

		end

		return assetOverlays, assetIdToNewAssetIdMap, batchOfFilesToMoveArray

	end )

end

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

local function convertTrashDocValueToRestoredDoc( trashDocValue, trashDocIsOnOz, timeOfRestore, createAsOverlay )
	
	-- if we're creating a docValue the deletion modifier is just nil, if we're creating an overlay it
	-- should be { _delete = true }
	local deletionModifier = createAsOverlay and { _delete = true } or nil
	
	local baseRestoredDoc = createAsOverlay and { _id = KSIdGenerator.uuidInLowercaseHex() } or createCopyOfAssetDocValue( trashDocValue )
	
	baseRestoredDoc.type = 'asset'
	baseRestoredDoc.subtype = isVideoSubtype( trashDocValue.subtype ) and 'video' or 'image'
	baseRestoredDoc.userCreated = derefOpt( trashDocValue, 'original', 'userCreated' ) or baseRestoredDoc.userCreated
	baseRestoredDoc.userUpdated = timeOfRestore

	if baseRestoredDoc.subtype == nil then
		logger:errorf( "Restored Document subtype is nil, trashDocValue subtype was %s", trashDocValue.subtype )
	end

	baseRestoredDoc.original = deletionModifier
	baseRestoredDoc.purgeDate = deletionModifier

	setCreateWithRevId( trashDocValue, baseRestoredDoc, trashDocIsOnOz )

	setOrUpdateCreateFrom( trashDocValue, trashDocIsOnOz, trashDocValue._revId, baseRestoredDoc )

	baseRestoredDoc = removeDocSpecificLocalOnlyData( baseRestoredDoc, createAsOverlay )

	return baseRestoredDoc
end

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

function WFCatTrash.createRestoredAssetFromDocValue( trashDocValue, trashDocIsOnOz, timeOfRestore )

	return convertTrashDocValueToRestoredDoc( trashDocValue, trashDocIsOnOz, timeOfRestore, false --[[ createAsOverlay ]] )

end

local function createRestoredAssetOverlayFromDocValue( trashDocValue, trashDocIsOnOz, timeOfRestore )
	return convertTrashDocValueToRestoredDoc( trashDocValue, trashDocIsOnOz, timeOfRestore, true --[[ createAsOverlay ]] )
end

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

local function createOverlaysForRestoredTrashAssets( catalog, pusherChangeCounter, timeOfRestore, restoreOnlyAssetsWithLocalOriginal, batchOfTrashDocValues, gatherBinaryFilesToMoveFn )

	-- Given a batch of trash doc values to restore, generate overlays for
	-- the creation of the restored asset, deletion of the trash asset, and
	-- a subsequent update to the restored asset if changes were made prior
	-- to the asset's deletion.

	local batchOfRestoredTrashAssetIds = {}
	local batchOfOverlays = {}
	local assetIdToNewAssetIdMap = {}
	local batchOfFilesToMoveArray = {}

	for i, trashDocValue in ipairs( batchOfTrashDocValues ) do
		
		-- Check if we should continue to restore. 
		-- If restoring only assets with binary, check local Master or Proxy binary
		local continueToRestore = not restoreOnlyAssetsWithLocalOriginal or LrAssetUtils.hasMasterLocally( trashDocValue )
		
		if continueToRestore then
		
			local trashDocIsOnOz = isAssetOnOzOrWasPushed( trashDocValue, pusherChangeCounter )
			
			-- Create an overlay that will convert the trash asset into the restored asset when we 
			-- delete the trash asset
			local restoredAssetOverlay = createRestoredAssetOverlayFromDocValue( trashDocValue, trashDocIsOnOz, timeOfRestore )
			
			-- get the binary files we'll need to move from trash asset to restored asset
			if gatherBinaryFilesToMoveFn then
				if WFPlatformInfo.isDesktop then
					-- the digest calculation for finding rendtions needs two full doc values so we copy and apply 
					-- the overlay to get what we expect the resulting docValue to look like
					local restoredDocValue = WFTableMerge.applyOverlay( deepCopy( trashDocValue ), restoredAssetOverlay )
					gatherBinaryFilesToMoveFn( catalog, trashDocValue, restoredDocValue, batchOfFilesToMoveArray )
				else
					-- On mobile, it needs only _id and update restoredAssetOverlay with destination path
					gatherBinaryFilesToMoveFn( catalog, trashDocValue, restoredAssetOverlay, batchOfFilesToMoveArray )
				end
			end
			
			local restoredAssetModificationOverlay
			local createWithRevId = getCreateWithRevisionId( restoredAssetOverlay )
			if restoredAssetOverlay._newRevId ~= createWithRevId then
	
				-- Asset was modified before it was originally deleted and that modification
				-- was not synced up, create an extra overlay so that the restore or
				-- delete operation will result in a conflict.
	
				-- Since we don't know what fields were modified we will use the _newIdOverlay
				-- to modify the copied doc in a way that removes all of the modifiable fields
				-- and then have the modification overlay ensure that they get put back into the
				-- document immediately under a new revision. This will allow the conflict 
				-- resolution to ensure the changes get applied
				restoredAssetModificationOverlay = deepCopy( restoredAssetOverlay )
				
				-- Modification should create a new revId we don't need to know what it
				-- is so clear out _newRevId which was copied from the restored doc.
				restoredAssetModificationOverlay._newRevId = nil
				
				-- The modification to the restored doc value doesn't need to update any
				-- of the trash info and we want to preserve the createWithRevId so the
				-- modification will cause a conflict generating a subsequent update.
				setNested( restoredAssetModificationOverlay, nil, '_localOnly', 'universal', 'trash' )		
				
				-- restoredAssetModificationOverlay will apply all of the fields in it's overlay,
				-- clear them out on the restoredAssetOverlay's first revision so we will 
				-- create a conflict with what oz has and the 3 way merge will sort out 
				-- only what actually changed.
	
				-- Remove any fields via the restoredAssetOverlay that the user can change
				-- develop settings, Stars, Flags, Title, Caption, Copyright
				-- Capture Date, location, city, state, country, keywords.
				-- But make sure that the overlay for the modification has these so they get applied 
				-- afterwards and the conflict resolution can ensure that the local version + the oz 
				-- version contains the correct data
				restoredAssetOverlay.develop = { _delete = true }
				restoredAssetModificationOverlay.develop = derefOpt( trashDocValue, 'develop' )
				
				restoredAssetOverlay.xmp = { _delete = true } -- location, city, state, country, captureDate, keywords, Title, Caption, Copyright
				restoredAssetModificationOverlay.xmp = derefOpt( trashDocValue, 'xmp' )
				
				restoredAssetOverlay.ratings = { _delete = true } -- stars
				restoredAssetModificationOverlay.ratings = derefOpt( trashDocValue, 'ratings' )
				
				restoredAssetOverlay.reviews = { _delete = true } -- pick/reject flags
				restoredAssetModificationOverlay.reviews = derefOpt( trashDocValue, 'reviews' )
				
				restoredAssetOverlay.captureDate = { _delete = true } -- if nil presents an issue, change it to be a fraction of a second earlier
				restoredAssetModificationOverlay.captureDate = derefOpt( trashDocValue, 'captureDate' )
				
				restoredAssetOverlay.userUpdated = { _delete = true } -- the userUpdated date from the modification
				restoredAssetModificationOverlay.userUpdated = derefOpt( trashDocValue, 'userUpdated' )
				
				restoredAssetOverlay.copy = { _delete = true }
				restoredAssetModificationOverlay.copy = derefOpt( trashDocValue, 'copy' )
			
			end
			
			
			-- Map the trash asset id to the new restored asset id so we
			-- know how to update the associated docs (album_assets, comments, favorites).
			assetIdToNewAssetIdMap[ trashDocValue._id ] = restoredAssetOverlay._id
	
			-- don't need to send the actual delete to oz as oz will do that as part
			-- of the restore.
			batchOfOverlays[ #batchOfOverlays + 1 ] = {
				_id = trashDocValue._id,
				_deleted = true,
				_newLabel = '_dontsendtooz',
				_newId = restoredAssetOverlay._id,
				_newIdOverlay = restoredAssetOverlay,
				_localOnly = {
					deletedLocally = true, -- signals that file deletions are handled
				}
			}
			
			if restoredAssetModificationOverlay then
				batchOfOverlays[ #batchOfOverlays + 1 ] = restoredAssetModificationOverlay
			end

			batchOfRestoredTrashAssetIds[ #batchOfRestoredTrashAssetIds + 1 ] = trashDocValue._id
			
		end

	end

	return batchOfRestoredTrashAssetIds, batchOfOverlays, assetIdToNewAssetIdMap, batchOfFilesToMoveArray

end

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

local function getAssetUuidsFromImageIds( catalog, imageIds, assetIdFromLocalIdFn )

	-- return Observable which fires only once with the list of oz-asset-ids
	-- converting local ids to oz-asset-ids if needed.

	-- Also ensures imageIds are either all local or all oz-asset-ids.

	local assetUuids

	if type( imageIds[ 1 ] ) == 'number' then

		for _, imageId in ipairs( imageIds ) do
			assert( type( imageId ) == 'number', 'All asset ids must be either local or global ids but not a mix' )
		end

		assetUuids = assetIdFromLocalIdFn( catalog, imageIds )

	elseif type( imageIds[ 1 ] ) == 'string' then

		for _, imageId in ipairs( imageIds ) do
			assert( type( imageId ) == 'string', 'All asset ids must be either local or global ids but not a mix' )
		end

		assetUuids = Observable.just( imageIds )

	else

		assert( false, 'Asset ids must all be uuid strings, or all be non-zero positive integers' )

	end

	return assetUuids

end

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

function WFCatTrash.getAssetUuidsFromTrashImageIds( catalog, imageIds )

	-- Convert a local trashAsset localId to a UUID.  If imageIds
	-- contain UUIDs already do nothing.

	-- Note that his will only work for trash asset local Ids.
	-- Regular asset Ids should use WFCatIdUtils.getOzAssetIdsForLocalIds

	return getAssetUuidsFromImageIds( catalog, imageIds, WFCatIdUtils.getOzTrashAssetIdsForLocalIds )

end

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

local function getAssetIdsFromBatchOfDocValues( batchOfDocValues )

	local batchOfIds = {}

	for i, docValue in ipairs( batchOfDocValues ) do
		batchOfIds[ i ] = docValue._id
	end

	return batchOfIds

end

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

local function moveBinaryFiles( batchOfFilesToMoveArray )

	batchOfFilesToMoveArray = batchOfFilesToMoveArray or {}

	return Observable.from( batchOfFilesToMoveArray )
	:flatMap( function( fileToMove )

		if fileToMove.srcFilePath then
			
			local dstFolder = KSPathUtils.parent( fileToMove.dstFilePath )
			WFFileUtils.mkdir( dstFolder )
		
			return WFAsyncFileUtils.moveFile( fileToMove.srcFilePath, fileToMove.dstFilePath )
			:catch( function( errorMessageForMove, platformErrorMessageForMove )

				-- If we failed moving it try copying it instead, Windows may have the
				-- file open in another process making it unmovable (error 32).

				logger:debugf( "moveBinaryFiles move error %s, (%s) src: %s to dst: %s",
					errorMessageForMove, platformErrorMessageForMove,
					fileToMove.srcFilePath, fileToMove.dstFilePath )

				return WFAsyncFileUtils.copyToUniquelyNamedFile( fileToMove.srcFilePath, fileToMove.dstFilePath )
				:catch( function( errorMessageForCopy, platformErrorMessageForCopy )

					-- Ignore any error if a file was already there we will use that as the
					-- hash should make it unique.  If we failed for some other reason we
					-- would end up trying to re-download the rendition so it likely wouldn't
					-- be a new kind of permanent failure.  Not sure what else we could do
					-- in any event.
					logger:warnf( "moveBinaryFiles copy error %s, (%s) src: %s to dst: %s",
						errorMessageForCopy, platformErrorMessageForCopy,
						fileToMove.srcFilePath, fileToMove.dstFilePath )

					return Observable.just( true )

				end )

			end )
			:flatMap( function()
				
				if WFPlatformInfo.isiOS then
					local srcParentFolder = KSPathUtils.parent( fileToMove.srcFilePath )
					local optOnlyDeleteDirIfEmpty = true
					
					WFFileUtils.deleteDirectory( srcParentFolder, optOnlyDeleteDirIfEmpty )
				end
				
				return Observable.just( true )
				
			end )
			
		else
			return Observable.just( fileToMove )
		end

	end )
	:startWith( {} ) -- if batchOfFilesToMoveArray is empty make sure we get at least one value.
	:takeLast( 1 )
	:map( function()
		return true
	end )

end

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

local pusherControlDisabled = {
	pauseReturningCounter = function()
		return Observable.just( 0 )
	end,
	resume = function()
	end
}

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

local function waitForIndexStoreToGetUpToDate( catalog )

	return function()

		-- Make a simple query on the asset related tables with quiesceUntilUpToDate
		-- to make sure it is up to date so we don't show the completion bezel
		-- before the list is empty.
		--
		-- Having the table names in comments is good enough to trigger the
		-- quiesceUntilUpToDate code for the tables listeds.

		return catalog:indexStore():queryOnce {
			sql = [[SELECT 0 -- trashAssets, coreInfo, assets]],
			quiesceUntilUpToDate = true,
		}
		:map( function()
			return true
		end )

	end

end

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

function WFCatTrash.moveAssetsToTrash( catalog, imageIds, pusherControl, stackRelatedOverlays, gatherBinaryFilesToMoveFn, origin )

	-- This currently deletes stacks immediately so they aren't restored.

	check( 'table', 'table', '?table', '?function', '?function' )

	pusherControl = pusherControl or pusherControlDisabled

	stackRelatedOverlays = stackRelatedOverlays or function( catalog, assetIds )
		-- iOS and Android don't do anything with stacks and will return an empty table.
		return Observable.just( {} )
	end

	local startOfTrashOperation = KSDate_timeNowForAnimation()
	local isFirstBatch = true

	return Observable.combineLatest( function( pusherChangeCounter, purgeAssetsAfter, assetIds )

			trashLogf( logger, "moveAssetsToTrash (%d assets) assetIds = %s", #assetIds, assetIds )

			local timeOfDelete = KSDate.timeInIso8601Format()
			local purgeDate = KSDate.shiftIso8601DateTime( timeOfDelete, purgeAssetsAfter )

			return pusherChangeCounter, assetIds, timeOfDelete, purgeDate

		end,
		pusherControl.pauseReturningCounter( true, kEventWorkflow ),
		catalog:ozUser():purgeAssetsAfterValue(),
		getAssetUuidsFromImageIds( catalog, imageIds, WFCatIdUtils.getOzAssetIdsForLocalIds )
	)

	:flatMap( function( pusherChangeCounter, assetIds, timeOfDelete, purgeDate )

		local docStore = catalog:docStore()

		-- Limit to 2 batches at once as we can consume too much memory on mobile and the grid gets
		-- out of sync with what is actually happening.

		-- The idea behind of processing 2 batches is we can be building up the array for the next
		-- bulkApplyOverlay while one is being issued, although they both may be issued simultaneously
		-- if we build up the array faster than the other completes. Using 1 has the affect of keeping
		-- the grid more accurate (as we can end up showing assets for a second or two after the badge).

		-- While watching what is happening it looks like we are issuing 2 bulkApplyOverlays
		-- pretty close together, so the above may not be true, however LrDesktop needs to do this to
		-- be similar in performance when restoring when using large catalogs (to when we weren't
		-- limiting how many bulkApplyOverlays were occurring simulataneously.  In a debug build I saw
		-- we were running 131 at the same time.


		-- Fully process the deletes in batches in case the app dies or gets killed.
		assetIds.includeInitialChangeCounter = true

		return WFRxBatchedBulkOperation( "moveAssetsToTrash", docStore, assetIds, 100, 2, function( batchOfDocValuesToDelete )

			local batchOfAssetIds = getAssetIdsFromBatchOfDocValues( batchOfDocValuesToDelete )

			return Observable.combineLatest(
				function( stackOverlays, assetOverlays, assetIdToTrashAssetIdMap, batchOfFilesToMoveArray )

					local stackOverlaysOz = stackOverlays[ 'oz' ] or {}
					local stackOverlaysLocal = stackOverlays[ 'local' ] or {}

					-- Combine the stackOverlays and the assetOverlays, making sure that the
					-- stackOverlays are first, then the assets...
					local combinedStackAndAssetOverlays = {}
					for i, stackOverlay in ipairs( stackOverlaysOz ) do
						stackOverlay._newLabel = '_dontsendtooz'
						combinedStackAndAssetOverlays[ i ] = stackOverlay
					end

					appendArray( combinedStackAndAssetOverlays, stackOverlaysLocal )

					appendArray( combinedStackAndAssetOverlays, assetOverlays )

					return batchOfAssetIds, combinedStackAndAssetOverlays, assetIdToTrashAssetIdMap, batchOfFilesToMoveArray

				end,
				stackRelatedOverlays( catalog, batchOfAssetIds ),
				createOverlaysForAssetsMovedToTrash( catalog, pusherChangeCounter,
					timeOfDelete, purgeDate, batchOfDocValuesToDelete, gatherBinaryFilesToMoveFn ) )

			:flatMap( function( batchOfAssetIds, batchOfOverlays, assetIdToTrashAssetIdMap, batchOfFilesToMoveArray )

				-- Update all the associated documents (album_assets, comments, and favorites)
				-- to refer to the trash asset instead instead of the original asset.
				-- These all use the preserve 'oz' label enable to only send the associated
				-- documents that were created before the delete.
				local startTimeOfAssocAssetQuery = KSDate_timeNowForAnimation()

				return Observable.combineLatest(
					function( albumAssetOverlays, commentOverlays, favoriteOverlays, versionOverlays )

						appendArray( batchOfOverlays, albumAssetOverlays )
						appendArray( batchOfOverlays, commentOverlays )
						appendArray( batchOfOverlays, favoriteOverlays )
						appendArray( batchOfOverlays, versionOverlays )

						WFPerformanceUtils.maybeLogLongExecutionInDiagnosticLogWithDynamicDescription(
							"moveAssetsToTrash queryAssocDocs", startTimeOfAssocAssetQuery, kLogTrashQueryLongerThanSecs,
							function( isFirstBatch )
								return isFirstBatch and "FirstBatch" or "SecondaryBatch"
							end,
							isFirstBatch )

						isFirstBatch = false


						return batchOfOverlays, batchOfFilesToMoveArray, #batchOfAssetIds

					end,
					updateAlbumAssetIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap ),
					updateCommentAssetsIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap ),
					updateFavoriteAssetIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap ),
					updateVersionIdOverlays( catalog, batchOfAssetIds, assetIdToTrashAssetIdMap )
				)

			end )

			:flatMap( function( batchOfOverlays, batchOfFilesToMoveArray, numAssetsInBatch)

				-- Apply all the stack overlays, trash asset creates, modifications
				-- and original asset deletes, and assoc doc updates in the same
				-- bulkApplyOverlay to make it more atomic. As bulkApplyOverlay
				-- has a transaction wrapped around it.

				return Observable.combineLatest( Observable.just( batchOfFilesToMoveArray ),
					docStore:bulkApplyOverlay( batchOfOverlays, "moveAssetsToTrash" ), Observable.just(numAssetsInBatch) )

			end )

			:flatMap( function( batchOfFilesToMoveArray, bulkApplyResults, numAssetsInBatch)

				-- Regarding user killing app before this completes:
				-- If moveBinaryFiles doesn't happen or is partially complete it should still be ok
				-- as we would end up re-downloading or regenerating the renditions. The worst
				-- of it is that the asset may not immediately show up in trash until the user
				-- is online.  Renditions are still part of the _localOnly array on mobile
				-- so not sure they need to move renditions. (see WFOzDocIndexUtils.assetHasThumbnail)


				if gatherBinaryFilesToMoveFn then
					return Observable.combineLatest(
						Observable.just(numAssetsInBatch),
						moveBinaryFiles( batchOfFilesToMoveArray )
					)
				end

				return Observable.just(numAssetsInBatch)

			end )

		end )

		:scan( function( totalTrashAssetsDeleted, numAssetsInBatch)

			totalTrashAssetsDeleted = totalTrashAssetsDeleted + numAssetsInBatch

			trashLogf(logger, "moveAssetsToTrash, Total assets deleted till now %s", totalTrashAssetsDeleted)

			return totalTrashAssetsDeleted

		end, 0 )

		:takeLast( 1 )

		:tap( function( totalTrashAssetsDeleted )

			local subtype, error_desc
			if totalTrashAssetsDeleted == #imageIds then -- success case
				subtype = 'success'
			elseif totalTrashAssetsDeleted > 0 then
				subtype = 'partial-success'
			else
				subtype = 'error'
				error_desc = 'failed'
			end

			local timeInSec = math.ceil( KSDate_timeNowForAnimation() - startOfTrashOperation )
			if timeInSec < 1 then
				timeInSec = 1
			end

			WFAnalytics.trackEvent {
				['event.type'] = 'delete',
				['event.workflow'] = kEventWorkflow,
				['event.subtype'] = subtype,
				['event.subcategory'] = origin,
				['event.count'] = #assetIds,
				['event.value'] = totalTrashAssetsDeleted,
				['event.error_desc'] = error_desc,
				['lr.operation_time'] = timeInSec,
			}

			trashLogf( logger, "moveAssetsToTrash assets . Successfully deleted %d assets", totalTrashAssetsDeleted )

		end)

		:flatMap( waitForIndexStoreToGetUpToDate( catalog ) )

	end )
	:takeLast( 0 )
	:finally( pusherControl.resume )
	:tap( nil, function( errorMessage )

		local timeInSec = math.ceil( KSDate_timeNowForAnimation() - startOfTrashOperation )
		if timeInSec < 1 then
			timeInSec = 1
		end

		WFAnalytics.trackEvent {
			['event.type'] = 'delete',
			['event.workflow'] = kEventWorkflow,
			['event.subtype'] = 'error',
			['event.subcategory'] = origin,
			['event.count'] = #imageIds,
			['event.error_desc'] = errorMessage,
			['lr.operation_time'] = timeInSec,
		}

		logger:errorf( "moveAssetsToTrash failed with error: %s", errorMessage )
		end,
		function()
			WFPerformanceUtils.maybeLogLongExecutionInDiagnosticLogWithDynamicDescription(
				"moveAssetsToTrash", startOfTrashOperation, kLogTrashOperationsLongerThanSecs,
				function( numImages )
					return string.format( "%d photos", numImages )
				end,
				#imageIds )
		end )

end

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

function WFCatTrash.restoreTrashAssets( catalog, trashImageIds, pusherControl, restoreOnlyAssetsWithLocalOriginal, restoreCountObservable, gatherBinaryFilesToMoveFn )

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

	pusherControl = pusherControl or pusherControlDisabled
	restoreOnlyAssetsWithLocalOriginal = not not restoreOnlyAssetsWithLocalOriginal

	local docStore = catalog:docStore()

	local startOfTrashOperation = KSDate_timeNowForAnimation()
	local isFirstBatch = true

	return Observable.combineLatest(
		pusherControl.pauseReturningCounter( true, kEventWorkflow ),
		getAssetUuidsFromImageIds( catalog, trashImageIds, WFCatIdUtils.getOzTrashAssetIdsForLocalIds )
	)

	:flatMap( function( pusherChangeCounter, trashAssetIds )

		trashLogf( logger, "restoreTrashAssets (%d assets) assetIds = %s", #trashAssetIds, trashAssetIds )

		local timeOfRestore = KSDate.timeInIso8601Format()

		-- Fully process the restores in batches in case the app dies or gets killed.
		trashAssetIds.includeInitialChangeCounter = true

		return WFRxBatchedBulkOperation( "restoreTrashAssets", docStore, trashAssetIds, 100, 2, function( batchOfDocValuesToRestore )

			local batchOfRestoredTrashAssetIds, batchOfOverlays, trashAssetIdToRestoredAssetIdMap, batchOfFilesToMoveArray =
				createOverlaysForRestoredTrashAssets( catalog, pusherChangeCounter, timeOfRestore,
					restoreOnlyAssetsWithLocalOriginal, batchOfDocValuesToRestore, gatherBinaryFilesToMoveFn )

			return Observable.just( batchOfRestoredTrashAssetIds, batchOfOverlays, trashAssetIdToRestoredAssetIdMap, batchOfFilesToMoveArray )

			:where( function(  batchOfRestoredTrashAssetIds  )
			    return #batchOfRestoredTrashAssetIds > 0
			end )

			:flatMap( function( batchOfRestoredTrashAssetIds, batchOfOverlays, trashAssetIdToRestoredAssetIdMap, batchOfFilesToMoveArray )

				-- Generate overlays for all the associated documents (album_assets, comments, and favorites)
				-- to update them to refer to the restored asset instead instead of the trash asset.
				-- These all use the _preserveLabel flag on the overlay so we only send the associated
				-- documents that were newly created or deleted before the asset delete.

				local startTimeOfAssocAssetQuery = KSDate_timeNowForAnimation()

				return Observable.combineLatest(
					function( albumAssetOverlays, commentOverlays, favoriteOverlays, versionOverlays )

						appendArray( batchOfOverlays, albumAssetOverlays )
						appendArray( batchOfOverlays, commentOverlays )
						appendArray( batchOfOverlays, favoriteOverlays )
						appendArray( batchOfOverlays, versionOverlays )
						
						WFPerformanceUtils.maybeLogLongExecutionInDiagnosticLogWithDynamicDescription(
							"restoreTrashAssets queryAssocDocs", startTimeOfAssocAssetQuery, kLogTrashQueryLongerThanSecs,
							function( isFirstBatch )
								return isFirstBatch and "FirstBatch" or "SecondaryBatch"
							end,
							isFirstBatch )

						isFirstBatch = false

						return batchOfOverlays, #batchOfRestoredTrashAssetIds, batchOfFilesToMoveArray

					end,
					updateAlbumAssetIdOverlays( catalog, batchOfRestoredTrashAssetIds, trashAssetIdToRestoredAssetIdMap ),
					updateCommentAssetsIdOverlays( catalog, batchOfRestoredTrashAssetIds, trashAssetIdToRestoredAssetIdMap ),
					updateFavoriteAssetIdOverlays( catalog, batchOfRestoredTrashAssetIds, trashAssetIdToRestoredAssetIdMap ),
					updateVersionIdOverlays( catalog, batchOfRestoredTrashAssetIds, trashAssetIdToRestoredAssetIdMap )
				)

			end )

			:flatMap( function( batchOfOverlays, numTrashAssetsRestoredInBatch, batchOfFilesToMoveArray )

				-- Apply all restored asset creates, modifications
				-- and trash asset deletes, and assoc doc updates in the same
				-- bulkApplyOverlay to make it more atomic. As bulkApplyOverlay
				-- has a transaction wrapped around it.

				return docStore:bulkApplyOverlay( batchOfOverlays, "restoreTrashAssets" )
					:flatMap( function()

						if gatherBinaryFilesToMoveFn then
							return moveBinaryFiles( batchOfFilesToMoveArray )
						end

						return Observable.just( true )

					end )
					:map( function()

						return numTrashAssetsRestoredInBatch

					end )

			end )

		end )

		:scan( function( totalTrashAssetsRestored, numTrashAssetsRestoredInBatch )

			totalTrashAssetsRestored = totalTrashAssetsRestored + numTrashAssetsRestoredInBatch

			trashLogf(logger, "restoreTrashAssets, Total assets restored till now %s", totalTrashAssetsRestored)

			return totalTrashAssetsRestored

		end, 0 )

		:takeLast( 1 )

		:map( function( totalTrashAssetsRestored )

			if restoreOnlyAssetsWithLocalOriginal then
				trashLogf( logger, "restoreTrashAssets only restore assets with local binary. Successfully restored %d assets", totalTrashAssetsRestored )
			else
				trashLogf( logger, "restoreTrashAssets. Successfully restored %d assets", totalTrashAssetsRestored )
			end

			if restoreCountObservable then
				restoreCountObservable:onNext( totalTrashAssetsRestored )
			end


			local subtype, error_desc
			if totalTrashAssetsRestored == #trashImageIds then -- success case
				subtype = 'success'
			elseif totalTrashAssetsRestored > 0 then
				subtype = 'partial-success'
			else
				subtype = 'error'
				error_desc = 'failed'
			end

			local timeInSec = math.ceil( KSDate_timeNowForAnimation() - startOfTrashOperation )
			if timeInSec < 1 then
				timeInSec = 1
			end

			WFAnalytics.trackEvent {
				['event.type'] = 'restore',
				['event.workflow'] = kEventWorkflow,
				['event.subtype'] = subtype,
				['event.subcategory'] = 'deleteGrid',
				['event.count'] = #trashImageIds,
				['event.value'] = totalTrashAssetsRestored,
				['event.error_desc'] = error_desc,
				['lr.operation_time'] = timeInSec,
			}

		end)

		:flatMap( waitForIndexStoreToGetUpToDate( catalog ) )

	end )
	:takeLast( 0 )
	:finally( pusherControl.resume )
	:tap( nil, function( errorMessage )

		local timeInSec = math.ceil( KSDate_timeNowForAnimation() - startOfTrashOperation )
		if timeInSec < 1 then
			timeInSec = 1
		end

		WFAnalytics.trackEvent {
			['event.type'] = 'restore',
			['event.workflow'] = kEventWorkflow,
			['event.subtype'] = 'error',
			['event.subcategory'] = 'deleteGrid',
			['event.count'] = #trashImageIds,
			['event.error_desc'] = errorMessage,
			['lr.operation_time'] = timeInSec,
		}

		logger:errorf( "restoreTrashAssets failed with error: %s", errorMessage )
		end,
		function()
			WFPerformanceUtils.maybeLogLongExecutionInDiagnosticLogWithDynamicDescription(
				"restoreTrashAssets", startOfTrashOperation, kLogTrashOperationsLongerThanSecs,
				function( numImages )
					return string.format( "%d photos", numImages )
				end,
				#trashImageIds )
		end )

end

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

local function permanentlyDeleteAssets( catalog, imageIds, isDeleteTrash, startOfDeleteOperation, pusherControl, optAddFilesToDeleteFromDocValue, optFileDeletionHandler )

	check( 'table', 'table', 'boolean', '?number', '?table', '?function', '?function' )
	
	local docStore = catalog:docStore()
	
	local isFirstBatch = true
	local functionLogString = isDeleteTrash and 'deleteTrashAssets' or 'directlyDeleteAssets'

	return Observable.combineLatest(
		pusherControl.pauseReturningCounter( true, kEventWorkflow ),
		getAssetUuidsFromImageIds( catalog, imageIds, isDeleteTrash and WFCatIdUtils.getOzTrashAssetIdsForLocalIds or WFCatIdUtils.getOzAssetIdsForLocalIds )
	)

	:flatMap( function( pusherChangeCounter, assetIds )

		trashLogf( logger, functionLogString .. " (%d assets) assetIds = %s", #assetIds, assetIds )

		-- Fully process the deleteTrash in batches in case the app dies or gets killed.
		assetIds.includeInitialChangeCounter = true
		return WFRxBatchedBulkOperation( "permanentlyDeleteAssets", docStore, assetIds, 100, 2, function( batchOfDocValuesToDelete )

			local batchOfAssetIds = getAssetIdsFromBatchOfDocValues( batchOfDocValuesToDelete )

			local batchOfAssetOverlays, batchOfLocalFilesToDelete =
				makePermanentlyDeleteOverlaysAndFileDeletionList( catalog, isDeleteTrash, batchOfDocValuesToDelete, pusherChangeCounter, optAddFilesToDeleteFromDocValue )

			return Observable.just( batchOfAssetIds, batchOfAssetOverlays, batchOfLocalFilesToDelete )

			:flatMap( function( batchOfAssetIds, batchOfOverlays, batchOfLocalFilesToDelete )

				local startTimeOfAssocAssetQuery = KSDate_timeNowForAnimation()

				return Observable.combineLatest(
					function( albumAssetOverlays, commentOverlays, favoriteOverlays, versionOverlays )

						appendArray( batchOfOverlays, albumAssetOverlays )
						appendArray( batchOfOverlays, commentOverlays )
						appendArray( batchOfOverlays, favoriteOverlays )
						appendArray( batchOfOverlays, versionOverlays )
						
						WFPerformanceUtils.maybeLogLongExecutionInDiagnosticLogWithDynamicDescription(
							functionLogString .. " queryAssocDocs", startTimeOfAssocAssetQuery, kLogTrashQueryLongerThanSecs,
							function( isFirstBatch )
								return isFirstBatch and "FirstBatch" or "SecondaryBatch"
							end,
							isFirstBatch )

						isFirstBatch = false

						return batchOfAssetIds, batchOfOverlays, batchOfLocalFilesToDelete

					end,
					deleteAlbumAssetOverlays( catalog, batchOfAssetIds ),
					deleteCommentAssetOverlays( catalog, batchOfAssetIds ),
					deleteFavoriteAssetOverlays( catalog, batchOfAssetIds ),
					deleteVersionAssetOverlaysAndDeleteLocalSettingsFiles( catalog, batchOfAssetIds )
				)

			end )

			-- Delete all trash assets (subtype deleted_image, deleted_video) or directly deleted assets from 
			-- the doc store along with all the associated documents. This is done in the same bulkApplyOverlay
			-- to make it atomic for the batch as there is a transaction wrapped around bulkApplyOverlay.

			:flatMap( function( batchOfAssetIds, batchOfOverlays, batchOfLocalFilesToDelete )

				return docStore:bulkApplyOverlay( batchOfOverlays, "permanentlyDeleteAssets" )
				:map( function()

					-- Delete the local files related to these assets
					if optFileDeletionHandler then
						optFileDeletionHandler( batchOfAssetIds, batchOfLocalFilesToDelete )
					end

					return #batchOfAssetIds

				end )

			end )

		end )

		:scan( function( totalAssetsDeleted, numAssetsDeletedInBatch)

			totalAssetsDeleted = totalAssetsDeleted + numAssetsDeletedInBatch

			trashLogf(logger, functionLogString .. ", Total assets deleted till now %s", totalAssetsDeleted)

			return totalAssetsDeleted

		end, 0 )

		:takeLast( 1 )

		:tap( function( totalAssetsDeleted )

			if isDeleteTrash then
			
				local subtype, error_desc
				if totalAssetsDeleted == #imageIds then -- success case
					subtype = 'success'
				elseif totalAssetsDeleted > 0 then
					subtype = 'partial-success'
				else
					subtype = 'error'
					error_desc = 'failed'
				end
	
				local timeInSec = math.ceil( KSDate_timeNowForAnimation() - startOfDeleteOperation )
				if timeInSec < 1 then
					timeInSec = 1
				end
	
				WFAnalytics.trackEvent {
					['event.type'] = 'permanent-delete',
					['event.workflow'] = kEventWorkflow,
					['event.subtype'] = subtype,
					['event.count'] = #imageIds,
					['event.value'] = totalAssetsDeleted,
					['event.subcategory'] = 'deleteGrid',
					['event.error_desc'] = error_desc,
					['lr.operation_time'] = timeInSec,
				}
				
			end
			
			trashLogf( logger, functionLogString .. " assets . Successfully deleted %d assets", totalAssetsDeleted )
			
		end )

		:flatMap( waitForIndexStoreToGetUpToDate( catalog ) )
		
	end )
	
end

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

function WFCatTrash.deleteTrashAssets( catalog, imageIds, pusherControl, optAddFilesToDeleteFromDocValue, optFileDeletionHandler )

	check( 'table', 'table', '?table', '?function', '?function' )
	
	pusherControl = pusherControl or pusherControlDisabled

	local isDeleteTrash = true
	local startOfTrashOperation = KSDate_timeNowForAnimation()

	return permanentlyDeleteAssets( catalog, imageIds, isDeleteTrash, startOfTrashOperation, pusherControl, optAddFilesToDeleteFromDocValue, optFileDeletionHandler )
	:takeLast( 0 )
	:finally( pusherControl.resume )
	:tap( nil, function( errorMessage )

		local timeInSec = math.ceil( KSDate_timeNowForAnimation() - startOfTrashOperation )
		if timeInSec < 1 then
			timeInSec = 1
		end

		WFAnalytics.trackEvent {
			['event.type'] = 'permanent-delete',
			['event.workflow'] = kEventWorkflow,
			['event.subtype'] = 'error',
			['event.count'] = #imageIds,
			['event.subcategory'] = 'deleteGrid',
			['event.error_desc'] = errorMessage,
			['lr.operation_time'] = timeInSec,
		}

		logger:errorf( "deleteTrashAssets failed with error: %s", errorMessage )
		end,
		function()
			WFPerformanceUtils.maybeLogLongExecutionInDiagnosticLogWithDynamicDescription(
				"deleteTrashAssets", startOfTrashOperation, kLogTrashOperationsLongerThanSecs,
				function( numImages )
					return string.format( "%d photos", numImages )
				end,
				#imageIds )
		end )

end

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

function WFCatTrash.directlyDeleteAssets( catalog, imageIds, pusherControl, optAddFilesToDeleteFromDocValue, optFileDeletionHandler )

	check( 'table', 'table', '?table', '?function', '?function' )

	pusherControl = pusherControl or pusherControlDisabled

	local isDeleteTrash = false
	
	return permanentlyDeleteAssets( catalog, imageIds, isDeleteTrash, nil, pusherControl, optAddFilesToDeleteFromDocValue, optFileDeletionHandler )
	:takeLast( 0 )
	:finally( pusherControl.resume )

end

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

local sortOrderSubject = Rx.BehaviorSubject( 'ASC' )

function WFCatTrash.setTrashAssetListSortOrder( order )

	sortOrderSubject:onNext( order )

end

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

local function getNextRefreshForPurgeDate( mostImmediatePurgeDate )

	local now = KSDate.timeInIso8601Format()
	local nextCheckDelay = KSDate.differenceInSeconds( now, mostImmediatePurgeDate )
	if nextCheckDelay < 1 then
		nextCheckDelay = 1
	end
	
	return nextCheckDelay, now

end

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

local nextListCheckSubject

function WFCatTrash.trashAssetList( catalog, fieldsToReturn, getMostImmediatePurgeDateFunction, encoder )

	nextListCheckSubject = Rx.BehaviorSubject( 0 )

	return Observable.combineLatest( sortOrderSubject, nextListCheckSubject:flatMapLatest( function( refreshInSeconds )
			return Rx.Observable.timer( refreshInSeconds, WFRxScheduler.sharedScheduler() )
		end ))
		:flatMap( function( ascDesc )
		
			local currentTime = KSDate.timeInIso8601Format()

			local indexStore = catalog:indexStore()
			
			local secondSortOrder = ", captureDate %s"
			
			local thirdSortOrder, fourthSortOrder

			if WFPlatformInfo.isDesktop then
				thirdSortOrder = ", filenameLC COLLATE i18n_collate %s"
				fourthSortOrder = ", assetId %s"
			else
				thirdSortOrder = ", fileName COLLATE i18n_collate %s"
				fourthSortOrder = ", docId %s"
			end
			
			local sql = "SELECT %s FROM trashAssets WHERE purgeDate > ? ORDER BY purgeDate %s" .. secondSortOrder .. thirdSortOrder .. fourthSortOrder
			sql = string.format( sql, fieldsToReturn, ascDesc, ascDesc, ascDesc, ascDesc )
			
			return indexStore:query { 
				sql = sql,
				currentTime,
				encoder = encoder,
			}
			
			:map( function( rows )
				
				local mostImmediatePurgeDate = getMostImmediatePurgeDateFunction( rows, ascDesc )
				
				if mostImmediatePurgeDate then
					local nextRefreshInSeconds, now = getNextRefreshForPurgeDate( mostImmediatePurgeDate )
					
					logger:infof( "WFCatTrash.trashAssetList: next check in %s seconds (time now is %s)", nextRefreshInSeconds, now )
					nextListCheckSubject:onNext( nextRefreshInSeconds )
				end
				
				return rows

			end )

		end )

end

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

local nextCountCheckSubject

function WFCatTrash.countTrashAssets( catalog )

	nextCountCheckSubject = Rx.BehaviorSubject( 0 )
	
	return Observable.combineLatest( nextCountCheckSubject:flatMapLatest( function( refreshInSeconds )
			return Rx.Observable.timer( refreshInSeconds, WFRxScheduler.sharedScheduler() )
		end ))
		:flatMap( function()
		
			local currentTime = KSDate.timeInIso8601Format()
			
			return catalog:indexStore():query { 
				sql = "SELECT COUNT(1), MIN(purgeDate) FROM trashAssets WHERE purgeDate > ?",
				currentTime,
			}
			:tap( function( rows )
	
				local mostImmediatePurgeDate = rows[ 1 ] and rows[ 1 ][ 2 ]
				
				if mostImmediatePurgeDate then
					local nextRefreshInSeconds, now = getNextRefreshForPurgeDate( mostImmediatePurgeDate )
					
					logger:infof( "WFCatTrash.countTrashAssets: next check in %s seconds (time now is %s)", nextRefreshInSeconds, now )
					nextCountCheckSubject:onNext( nextRefreshInSeconds )
				end
	
			end )
			
		end )
		
end

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

if not WFPlatformInfo.isDesktop then

function WFCatTrash.trashAssetsFileSize( catalog, assetIds )

	local currentTime = KSDate.timeInIso8601Format()
	
	if #assetIds == 0 then
		return Observable.just( 0 )
	end
	
	return catalog:indexStore():queryOnce { 
		sql = [[SELECT ifnull(sum(originalFileSize), 0.0), count(1) FROM trashAssets 
				WHERE purgeDate > ? AND docId IN CARRAY( ?, ?, 'char*')]],
		currentTime, 
		assetIds,
		#assetIds,
	}
	
end

end

function WFCatTrash.trashAssetsTotalFileSize( catalog )

	local currentTime = KSDate.timeInIso8601Format()
	
	return catalog:indexStore():queryOnce { 
		sql = "SELECT ifnull(sum(originalFileSize), 0.0), count(1) FROM trashAssets WHERE purgeDate > ?",
		currentTime,
	}

end

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

local nextAutoPurgeCheckSubject

function WFCatTrash.autoPurge( catalog )

	nextAutoPurgeCheckSubject = Rx.BehaviorSubject( 0 )

	return Observable.combineLatest( nextAutoPurgeCheckSubject:flatMapLatest( function( refreshInSeconds )
			return Rx.Observable.timer( refreshInSeconds, WFRxScheduler.sharedScheduler() )
		end ))
		:flatMap( function()
		
			local currentTime = KSDate.timeInIso8601Format()
			
			return catalog:indexStore():query { 
				sql = "SELECT docId FROM trashAssets WHERE purgeDate <= ? ORDER BY purgeDate ASC",
				singleColumnQuery = true,
				currentTime
			}
			
			:flatMap( function( autoPurgedAssetIds )
				
				local overlay = {
					_newLabel = '_dontsendtooz',
					_deleted = true,
				}
				
				return (#autoPurgedAssetIds > 0 and
					WFRxThrottledBatchedApplyOverlayToIds( "WFCatTrash_autoPurge", catalog:docStore(), overlay, autoPurgedAssetIds, 100, 10 )
					or
					Observable.just())
				:toArray()
					
			end )
			
			:map( function( deletedAssets )

				local totalDeletedAssets = 0
				for _, currentDeleteBatch in ipairs ( deletedAssets ) do
					if currentDeleteBatch and type ( currentDeleteBatch ) == 'table' then
						totalDeletedAssets = totalDeletedAssets + #currentDeleteBatch
					end
				end

				if totalDeletedAssets > 0 then

					local timeInSec = math.ceil( KSDate.differenceInSeconds( currentTime, KSDate.timeInIso8601Format() ) )
					if timeInSec < 1 then
						timeInSec = 1
					end

					WFAnalytics.trackEvent {
						['event.type'] = 'auto-purge',
						['event.workflow'] = kEventWorkflow,
						['event.subcategory'] = 'implicit',
						['event.subtype'] = 'success',
						['event.count'] = totalDeletedAssets,
						['event.value'] = totalDeletedAssets,
						['lr.operation_time'] = timeInSec,
					}

				end

				return currentTime
			end )
		
		end )
		
		:flatMap( function( currentTime )
		
			return catalog:indexStore():query( {
				sql = "SELECT purgeDate FROM trashAssets WHERE purgeDate > ? ORDER BY purgeDate ASC LIMIT 1",
				singleColumnQuery = true,
				currentTime
			} )
			:tap( function( purgeDates )
				
				if purgeDates then
					
					local mostImmediatePurgeDate = purgeDates[ 1 ]
				
					if mostImmediatePurgeDate then
						local nextRefreshInSeconds, now = getNextRefreshForPurgeDate( mostImmediatePurgeDate )
						
						logger:infof( "WFCatTrash.autoPurge: next check in %s seconds (time now is %s)", nextRefreshInSeconds, now )
						nextAutoPurgeCheckSubject:onNext( nextRefreshInSeconds )
					end
					
				end

			end )

		end )

end

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

return WFCatTrash
