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

WFCatErrorHandler.lua

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

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

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

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

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

local KSDate = require 'KSDate'
local KSJsonDecoder = require 'KSJsonDecoder'
local KSIdGenerator = require 'KSIdGenerator'
local KSStringUtils = require 'KSStringUtils'
local KSTableUtils = require 'KSTableUtils'
local WFAnalytics = require 'WFAnalytics'
local WFArguments = require 'WFArguments'
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatTrash = require 'WFCatTrash'
local WFCatTrashUtils = require "WFCatTrashUtils"
local WFCloseSupport = require 'WFCloseSupport'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFLogger = require 'WFLogger'
local WFOzDocUtils = require 'WFOzDocUtils'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxRetryObservable = require 'WFRxRetryObservable'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxThrottledBatchedBulkApplyOverlay = require 'WFRxThrottledBatchedBulkApplyOverlay'
local WFRxThrottledBatchedBulkDocValue = require "WFRxThrottledBatchedBulkDocValue"
local WFRxScheduler = require 'WFRxScheduler'
local WFTableArchiving = require 'WFTableArchiving'
local WFURI = require 'WFURI'

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

local logger = WFLogger 'WFCatErrorHandler'

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

local derefOpt = KSTableUtils.derefOpt
local check = assert( WFArguments.check )
local setNested = KSTableUtils.setNested

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

local function errors_table( )

	return {

		sql = [[CREATE TABLE errors( type )]],
		
		relevantTypeSubtypeTuples = {
			{ 'error', '*' }
		},
	
		map = function( docValue, emit )
			
			if docValue.type == 'error' and docValue._deleted ~= true then
				
				local isUnknownError = derefOpt( docValue, '_localOnly', 'unknownError' )
				
				if not isUnknownError then
					emit( docValue.subtype )
				else
					
					if docValue.subtype == WFCatOzErrors.Errors.kStorageExceededError then
						emit( docValue.subtype )
					end
					
					-- Here is where we try to look at previously unknown errors and see if it's a type
					-- that we now know about.  If so, we should be able to emit to the errors
					-- table and handle it.
					if docValue.subtype == WFCatOzErrors.Errors.kInputValidationError then
					
						local previouslyFailedHandling = derefOpt( docValue, '_localOnly', 'localError', WFCatOzErrors.Errors.kInputValidationError, 'handlingFailed' )
						
						-- missing 'download' field? there was a short time where Lightroom CC wasn't sending 
						-- this correctly in some cases.
						local wasMissingDownloadValue = derefOpt( docValue, '_serverProvidedMetadata', 'errors', 'download', 1 )
						
						if wasMissingDownloadValue and not previouslyFailedHandling then
							emit( docValue.subtype )
							return
						end
						
						-- missing 'if_match'? there appear to be some cases where we are missing the server provided metadata
						-- and the revision_ids and so while we send an update, we don't send an if_match revision.  Not yet
						-- sure why this occurs, but it appears to happen mainly to deleted assets. See NMB-6190
						local wasIfMatchMissing = derefOpt( docValue, '_serverProvidedMetadata', 'errors', 'if_match', 1 )
						if wasIfMatchMissing == 'is required' and not previouslyFailedHandling then
							emit( docValue.subtype )
							return
						end
					end
					
					-- NMB-6352: Any errors previously marked unknown that have a status code of 5XX should be automatically 
					-- retried by the default handler which now should handle these appropriately
					local maybeStatusCode = tonumber( derefOpt( docValue, '_serverProvidedMetadata', 'status' ) )
					if maybeStatusCode and ( 500 <= maybeStatusCode and maybeStatusCode < 600 ) then
						emit( docValue.subtype )
						return
					end
					
				end
			end
		end,

		hasOneRowPerDocument = true,
	}
	
end

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

local blockedUploadSubtypes = {
	[ WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError ] = true, [ WFCatOzErrors.Errors.kEntitlementExpiredError ] = true,
	[ WFCatOzErrors.Errors.kResourceNotFoundError ] = true,
	[ WFCatOzErrors.Errors.kInsufficientPermissionError ] = true,
}

local function blocked_uploads_table( )
	
	return {
	
		sql = [[CREATE TABLE blocked_uploads( spaceId, reason, type )]],
		
		useLocalId = WFPlatformInfo.isDesktop and true or false,
		
		hasOneRowPerDocument = true,
		
		relevantTypeSubtypeTuples = {
			{ '*', '*' }
		},
		
		map = function( docValue, emit )
			
			local errorSubtype = derefOpt( docValue, '_localOnly', 'ozError', 'subtype' )
			local docType = derefOpt( docValue, 'type' )
			local spaceId = docValue._spaceId
			
			if errorSubtype and blockedUploadSubtypes[ errorSubtype ] and not docValue._deleted then
				emit( spaceId, errorSubtype, docType )				
			end
			
		end,
	}

end

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

local function generateIndexSchema( indexStoreSchema )

	check( 'WFRxIndexStoreSchema' )

	indexStoreSchema:addTable( errors_table( ) )

	indexStoreSchema:addIndex [[CREATE INDEX error_type ON errors( type )]]
	
	indexStoreSchema:addTable( blocked_uploads_table( ) )
	
	indexStoreSchema:addIndex [[CREATE INDEX blocked_uploads_reason ON blocked_uploads( reason )]]

end

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

local function getErrorHandler( self, errSubtype )

	return self._errorDocumentHandlerFns[ errSubtype ]
	
end

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

local function addErrorDocToDiagnosticLog( self, errorDoc )
	
	-- for each error subtype record
	-- subtype
	--    * subtype count
	--    * status
	--    * source type (e.g. asset)
	--    * source action (e.g. create, update)
	--    * errors section (parse differently for most errors and PVE)
	--    * source ids (for easier troubleshooting)???

	local spm = derefOpt( errorDoc, '_serverProvidedMetadata' )
	local sourceDocType = derefOpt( spm, 'source', 'type' )
	local action = derefOpt( spm, 'source', 'action' )
	local status = derefOpt( spm, 'status' )
	local sourceDocId = derefOpt( spm, 'source', 'id' )
	
	local subtype = errorDoc.subtype
	local summaryForSubtype = self._errorDocSummary[ subtype ] or {}
	
	-- increment the count for all documents which hit this error subtype
	summaryForSubtype.count = ( summaryForSubtype.count or 0 ) + 1
	
	-- get the errors that this error doc represents (may be multiple)
	local errorDetails = {}
	local errors = derefOpt( spm, 'errors' )
	if type( errors ) == 'table' then
		for err, detailTable in pairs( errors ) do
			errorDetails[ #errorDetails + 1 ] = string.format( '%s:%s', err, detailTable[ 1 ] ) 
		end
	elseif type( errors ) == 'string' then
		errorDetails[ #errorDetails + 1 ] = errors	
	end
	
	if not next( errorDetails ) then
		-- if, for some reason, there's no errors table then provide a default message so that
		-- it's still logged
		errorDetails[ #errorDetails + 1 ] = 'missing error info' 
	end
	
	for _, errorDetail in ipairs( errorDetails ) do
		
		local errorKey = string.format( '%s:%s:%s:%s', sourceDocType, action, status, errorDetail )
	
		local errorSummary = summaryForSubtype[ errorKey ] or {}
		
		errorSummary.count = ( errorSummary.count or 0 ) + 1
		
		errorSummary.sourceDocIds = errorSummary.sourceDocIds or {}
		
		errorSummary.sourceDocIds[ #errorSummary.sourceDocIds + 1 ] = sourceDocId
		
		summaryForSubtype[ errorKey ] = errorSummary
	
	end
	
	self._errorDocSummary[ subtype ] = summaryForSubtype
	
end

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

local function _addErrorResolutionToDiagnosticLog( self, errorDoc, resolutionMethod )

	-- for each error subtype record
	-- subtype
	--    * count of number of resolutions recorded for this subtype
	--    * table indexed by resolution
	--         * count of number of errors docs resolved by this resolution method
	--         * source ids resolved by this resolution method

	local spm = derefOpt( errorDoc, '_serverProvidedMetadata' )
	local sourceDocId = derefOpt( spm, 'source', 'id' )

	local subtype = errorDoc.subtype

	local resolutionSummaryForSubtype = self._errorDocResolution[ subtype ] or {}
	resolutionSummaryForSubtype.count = ( resolutionSummaryForSubtype.count or 0 ) + 1

	resolutionSummaryForSubtype.resolution = resolutionSummaryForSubtype.resolution or {}
	local resolutionInfo = resolutionSummaryForSubtype.resolution[ resolutionMethod ] or {}
	resolutionInfo.count = ( resolutionInfo.count or 0 ) + 1

	local sourceDocIds = resolutionInfo.sourceDocIds or {}
	sourceDocIds[ #sourceDocIds + 1 ] = sourceDocId
	resolutionInfo.sourceDocIds = sourceDocIds

	resolutionSummaryForSubtype.resolution[ resolutionMethod ] = resolutionInfo
	self._errorDocResolution[ subtype ] = resolutionSummaryForSubtype

end

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

local function addErrorResolutionToDiagnosticLog( self, errorDoc, resolutionMethod )
	pcall( _addErrorResolutionToDiagnosticLog, self, errorDoc, resolutionMethod )
end

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

local function deleteErrorDoc( docStore, errorDocId )
	return docStore:deleteDoc( errorDocId )
end

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

local function markErrorAsUnknown( docStore, errorDocId )
	return docStore:applyOverlay {
			_id = errorDocId,
			_localOnly = {
				unknownError = true,
			}
		}
end

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

local function resolveErrorWithResult( self, errorDoc, resolutionCaller, observableReturningResolution )

	return observableReturningResolution
		:tap( function( resolution )
			resolutionCaller = resolutionCaller and ( "_" .. resolutionCaller ) or ""
			addErrorResolutionToDiagnosticLog( self, errorDoc, resolution .. resolutionCaller )
		end )
		:catch( Observable.just() )
		:flatMap( function()
			-- delete the error doc once we've handled the error document
			return deleteErrorDoc( self._catalog:docStore(), errorDoc._localId )
		end )

end

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

local function resolveError( self, errorDoc, resolution, resolvingObservable )

	return resolvingObservable
		:tap( function()
			addErrorResolutionToDiagnosticLog( self, errorDoc, resolution )
		end )
		:catch( Observable.just() )
		:flatMap( function()
			-- delete the error doc once we've handled the error document
			return deleteErrorDoc( self._catalog:docStore(), errorDoc._localId )
		end )
		:map( function() 
			return resolution
		end ) 

end

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

local function resolveErrorByIgnoring( self, errorDoc, resolution )

	resolution = resolution or ""

	return resolveError( self, errorDoc, "Ignored" .. resolution, Observable.just( true ) )

end

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

local function resolveErrorByApplyingOverlay( self, errorDoc, resolution, overlay )

	resolution = resolution or ""

	return resolveError( self, errorDoc, resolution, self._catalog:docStore():applyOverlay( overlay ) )

end

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

local function resolveErrorByMarkingWithOzError( self, errorDoc, resourceId )

	return resolveError( self, errorDoc, "MarkedWithOzError", self._catalog:docStore():applyOverlay(
		WFCatOzErrors.createOverlayForOzErrorSubtype( resourceId, errorDoc.subtype )
	) )

end

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

local function resolveByRestoringAssetFromOz( self, errorDoc, resourceId )

	-- The following code discards the current, unsynced revision of a document
	-- and instead makes sure the local revision matches what's on Oz. It does 
	-- this by fetching the latest revision from Oz, figuring out the delta 
	-- against our local revision, then applying that as an overlay while 
	-- forcing the new revision Id to be that of the Oz revision.

	local refreshAssetFromOz = self._catalog:docStore():docValue( resourceId )
	
		:flatMap( function( localDocValue )
		
			local catalogSpec = self._catalog:ozCatalogSpec()
	
			local documentUrl = KSTableUtils.derefOpt( localDocValue, '_serverProvidedMetadata', 'links', "self", 'href' )
				
			return WFRxRetryObservable {

				maxAttempts = 3,
				retryDelay = 10,

				factory = function()
				
					local options = {
						bypassRateLimiter = true,
					}
				
					return catalogSpec:get( documentUrl, options )
					
						:flatMap( function( result, headers )

							if headers.status_code == '200' then
							
								local decodedDoc = KSJsonDecoder.decodeWebSafeJson_stripNull( result )

								local ozDocValue = WFOzDocUtils.remapBasicOzDocument( decodedDoc, decodedDoc.base )

								-- Calculate the overlay that we need to apply to the existing revision so that
								-- after applying the overlay, the document will exactly match the revision
								-- on Oz.

								local overlay = require 'WFTableMerge'.createOverlay( localDocValue, ozDocValue )
							
								-- Remove all immutable or local fields from the overlay. Mostly this removes things
								-- like the _changeCounter, _localOnly, etc.
								
								for k in pairs( overlay ) do
									if k:match( '^_' ) then
										overlay[ k ] = nil
									end
								end
								
								-- Add back in the doc Id, as well as a the 'oz' label which indicates we got this
								-- data from Oz and will avoid an unnecessary sync.
								
								overlay._id = ozDocValue._id
								overlay._newLabel = 'oz'
								
								-- Make sure that the new revision that we are creating has the same revision Id as
								-- the latest revision on Oz. Forcing a specific revision Id can cause there to
								-- be multiple revisions with the same rev id, but this is fine. More than one revision
								-- with the same rev id happens all the time for _localOnly only changes made by 
								-- client local processes
								
								local ozRevisions = derefOpt( ozDocValue, '_serverProvidedMetadata', 'revision_ids' )
								overlay._newRevId = ozRevisions and ozRevisions[ #ozRevisions ]
								
								return self._catalog:docStore():applyOverlay( overlay )
								
							end

							return Observable.throw( "Unable to refresh asset from Oz" )

						end )
						
				end
			}
			
		end )

	return resolveError( self, errorDoc, "RefreshAssetFromOz", refreshAssetFromOz )
	
		:catch( function()
		
			-- If we are not able to refresh the document from Oz. Just mark it with an Oz error.
			
			return resolveError( self, errorDoc, "MarkedWithOzError", self._catalog:docStore():applyOverlay(
					WFCatOzErrors.createOverlayForOzErrorSubtype( resourceId, errorDoc.subtype )
				) )
		
		end )

end

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

local function resolveErrorAsUnknown( self, errorDoc, resolution )

	return markErrorAsUnknown( self._catalog:docStore(), errorDoc._id )
		:tap( function()
			addErrorResolutionToDiagnosticLog( self, errorDoc, resolution )
		end )

end

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

local dependentDocTypes = {
	[ 'album_asset' ] = true,
	[ 'stack_asset' ] = true,
	[ 'comment' ] = true,
	[ 'favorite' ] = true,
	[ 'version' ] = true
}

local function isDependentDocumentType( docType )

	return docType and dependentDocTypes[ docType ]
	
end

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

local function mimicOzDeletionEvent( docStore, docId, type, subtype )
	
	-- Rather than just 'deleteDoc' we create a deletion document that looks like the 
	-- deletion we would get from Oz in the sync feed.  This ensures that we clean up the
	-- document in the same way we would if we saw the deletion via the sync feed.
	local deletionDoc = {
			_id = docId,
			_deleted = true,
			_revIds = {
				KSIdGenerator.uuidInLowercaseHex(), -- Oz does not send revision ids on deletions
				force = true,
			},
			_serverProvidedMetadata = {
				deleted_by = 'WFCatErrorHandler'
			},
			type = type,
			subtype = subtype
		}
	
	return docStore:bulkSetDocValue( { label = 'oz', deletionDoc } )

end

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

local function getOrCreateSubtablesForPath( overlay, path )

	local sub = overlay
	
	for i, v in ipairs( path ) do
	
		local d = sub[ v ]
		if not d then
			d = {}
			sub[ v ] = d
		end
		
		sub = d
		
	end
	
	return overlay, sub
	
end

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

local function createValueAtPath( overlay, path, value )

	local key = path[ #path ]

	table.remove( path, #path )

	local sub
	overlay, sub = getOrCreateSubtablesForPath( overlay, path )
	
	sub[ key ] = value
	
	
	return overlay

end

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

local function removeOffendingKey( overlay, docPath )
					
						
	-- INFO [pkamin 2016-07-25]: validate errorCode for correct handling
	-- ( disallowed, invalid, required )
	-- Right now we delete the value to unblock sync which is very aggressive.
	-- We might be able to act smarter if the logic can be driven by the Data Model 
	-- Schema JSON - escpecially for the `required` error code
		
	createValueAtPath( overlay, docPath, { _delete = true } )
	
end

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

local function fixTimestamp( overlay, docPath, value, errorCode )
	
	local date = KSDate.recoverDateForOz( value )
	if not date then
		date = KSDate.timeInIso8601Format( nil, false )
	end
	
	createValueAtPath( overlay, docPath, date )

end

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

local function fixCaptureDate( overlay, docPath, value, errorCode )
	
	local date = KSDate.recoverDateForOz( value )
	if not date then
		date = "0000-00-00T00:00:00"
	end
	
	createValueAtPath( overlay, docPath, date )

end

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

local function fixTruncate( overlay, docPath, value, errorCode )

	-- if it's a number, truncate it, otherwise, remove the field
	if type( value ) == 'number' then
		createValueAtPath( overlay, docPath, math.floor( value ) )		
	else
		createValueAtPath( overlay, docPath, { _delete = true } )
	end

end

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

local function fixFaceDimension( overlay, docPath, value, errorCode )
	
	if not value or value < 0 then
		createValueAtPath( overlay, docPath, 0 )
	elseif value > 1 then
		createValueAtPath( overlay, docPath, 1 )
	end
	
end

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

local function fixProcessingModel( overlay, docPath, value, errorCode )
	
	if not value and errorCode == 'required' then
		createValueAtPath( overlay, docPath, 'lightroom' )
	end
	
end

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

local function isUUID( str )
	
	return ( type( str ) == 'string' and
		#str == 32 and 
		string.match( str, '^[0-9a-f]+$' ) )
		
end

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

local function fixRating( overlay, docPath, value, errorCode )
	
	if not value and errorCode == 'required' then
		-- with no rating it's invalid so we should just delete the 
		-- full ratings section. So if the last section of the docPath is
		-- 'rating', then remove it. This check is mostly redundant because
		-- we shouldn't otherwise get here, but I don't want to blindly remove
		-- a path and cause way too much to be deleted.
		if #docPath >= 2 and 
			docPath[ #docPath ] == 'rating' and
			isUUID( docPath[ #docPath - 1 ] ) then
			
			-- strip off 'rating'
			table.remove( docPath, #docPath )
			
			-- remove ratings.uuid section
			createValueAtPath( overlay, docPath, { _delete = true } )
		
		end
	end
	
end

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

local function fixSpaceDocument( catalog, docValue, errorKey, errorValue )
	
	local overlay = {}
	
	if errorKey == 'download' and docValue.download == nil then
		overlay.download = false
	end
	
	if next( overlay ) then
		overlay._id = docValue._id
		return catalog:docStore():applyOverlay( overlay )
	else
		return Observable.just( false )
	end
	
end

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

local function fixPayloadPrimary( catalog, docValue, errorKey, errorValue )

	if errorKey == 'payload.primary'
		and docValue.type == 'asset' 
		and docValue.subtype == 'person'
		and docValue.primary then
		
		--NMB-10093 -> delete primary due to cyclic relationships.
		local overlay = {
			primary = { _delete = true },
			_id = docValue._id
		}
		return catalog:docStore():applyOverlay( overlay )
	end
	
	return Observable.just( false )
end

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

local pathComponentBySubtype = {
	image = 'assets',
	video = 'assets',
}

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

local function fixIfMatchIssue( catalog, docValue, errorKey, errorValue )
	
	-- happy path for fixing this error assumes internet access
	local docId = docValue._id
	
	-- get link from docValue's self href if it's there
	-- otherwise construct the url manually
	local documentUrl = KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'links', 'self', 'href' )
	if not documentUrl then
		local href = catalog:ozCatalogSpec():catalogHref()
		
		local pathComponent = pathComponentBySubtype[ docValue.subtype ] 
		if pathComponent then
			href = WFURI.stringByAppendingPathComponents( href, pathComponent )
			href = WFURI.stringByAppendingPathComponents( href, docId )
			documentUrl = href
		end
		
	end
	
	if not documentUrl then
		logger:warnf( "No URL, unable to handle InputValidationError for %s", docId )
		return Observable.just() 
	end
	
	return catalog:ozCatalogSpec():get( documentUrl, { returnHttpErrorsAsData = true } )
		:flatMap( function( result, headers )
			if headers.status_code == '200' then
				-- convert JSON document and insert from Oz
				local decodedDoc = KSJsonDecoder.decodeWebSafeJson_stripNull( result )
				
				local docValue = WFOzDocUtils.remapBasicOzDocument( decodedDoc, decodedDoc.base )
							
				if docValue and not docValue._deleted then
					return catalog:docStore():bulkSetDocValue( { label = 'oz',
																docValue } )
				end
				
			elseif headers.status_code == '404' then
				
				-- consider this a deletion from Oz
				return mimicOzDeletionEvent( catalog:docStore(), docId, docValue.type, docValue.subtype )
							
			end
			
			return Observable.just( )
			
		end )
		:catch( Observable.just( ) )
		
	
end

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

-- these functions just add to an overlay
local payloadValidationFixups = {
	-- fixups for asset documents
	[ 'payload.importSource.importTimestamp' ] = fixTimestamp,
	[ 'payload.captureDate' ] = fixCaptureDate,
	[ 'payload.develop.userUpdated' ] = fixTimestamp,
	[ 'payload.develop.processingModel' ] = fixProcessingModel,
	[ 'payload.importSource.originalWidth' ] = fixTruncate,
	[ 'payload.importSource.originalHeight' ] = fixTruncate,
	[ 'payload.xmp.mwg-rs.Regions.RegionList.uuid.Area.x' ] = fixFaceDimension,
	[ 'payload.xmp.mwg-rs.Regions.RegionList.uuid.Area.y' ] = fixFaceDimension,
	[ 'payload.ratings.uuid.rating' ] = fixRating,
}

local exifPayloadValidationFields = {
	[ 'FNumber' ] = true,
	[ 'ApertureValue' ] = true,
	[ 'ExposureTime' ] = true,
	[ 'FocalLength' ] = true,
}

-- these functions must return an Observable that fixes the document
local inputValidationFixups = {
	-- fixups for space documents
	[ 'download' ] = fixSpaceDocument,
	[ 'if_match' ] = fixIfMatchIssue,
	[ "payload.primary" ] = fixPayloadPrimary,
}

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

local function maybeGetFixFunctionForKey( key )
	
	-- some keys are at variable places in the data model 
	-- e.g. payload.xmp.mwg-rs.Regions.RegionList.0185a47f32354e3d9cf56634a8b20109.Area.x
	-- so we need to possible replace it before looking up the fixFn
	local canonicalKey = string.gsub( key, 'RegionList%..*%.Area', 'RegionList.uuid.Area' )
	canonicalKey = string.gsub( key, 'ratings%..*%.rating', 'ratings.uuid.rating' )
	
	canonicalKey = canonicalKey or key
		
	return payloadValidationFixups[ canonicalKey ]
	
end

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

local function parseDocPathFromKey( key, docValueThatIncurredError )
	
	local docPath = KSStringUtils.split( key, '%.' )
						
	if #docPath < 1 then
		return nil
	end

	if #docPath == 1 then
		-- Something below payload level...this generally makes no sense and is not fixable, but it may make sense 
		-- for some doc types (spaces) so if we *know* it's fixable, use it, otherwise don't consider the docPath 
		-- fixable.  Add future cases to this if
		if docValueThatIncurredError.type == 'space' then
			return docPath
		else 
			return nil
		end
	end
	
	-- remove 'payload' level
	table.remove( docPath, 1 )
	
	local actualKey = docPath[ #docPath ]
	
	-- remove final key from path (we'll add it back)
	table.remove( docPath, #docPath )
	
	-- If key is a number, then we should probably do the following:
	-- Check to make sure it being considered a string isn't correct by seeing
	-- whether the docValue has something at that key. 
	-- If so, assume it should be treated as a string for the delete
	-- If not, treat it as a number index into an array but we need to add 1 
	-- since Oz indexes are 0 based and lua is 1 based
	local maybeNumber = tonumber( actualKey )
	if maybeNumber then
		
		local valueAtStringPath 
		if #docPath > 0 then 
			valueAtStringPath = derefOpt( docValueThatIncurredError, unpack( docPath ), actualKey )
		else 
			valueAtStringPath = derefOpt( docValueThatIncurredError, actualKey )
		end
									
		if valueAtStringPath then
			-- it looks like a number, but it quacks like a string so treat it as such
			table.insert( docPath, actualKey )
		
		else
			-- it looks like a number and it doesn't quack like a string so treat it as a number
			-- (but add 1 because our arrays are 1 based)
			local parentOfActualKey = docPath[ #docPath ] -- parent key is same as #docPath because last value was removed previously.

			if not exifPayloadValidationFields [ parentOfActualKey ] then
				table.insert( docPath, maybeNumber + 1 )
			end

		end
		
	else
		-- not even close to a number
		table.insert( docPath, actualKey )
		
	end
	
	return docPath
	
end

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

local function lastRetryIsOlderThanXHours( maybeDate, hours )
	
	if not maybeDate then
		return true
	end
	
	local olderThanInSeconds = hours * 60 * 60
	local now = KSDate.timeInIso8601Format()
	
	local differenceInSeconds = KSDate.differenceInSeconds( maybeDate, now )
	
	return differenceInSeconds > olderThanInSeconds
end

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

local function maybeTrashOperationNewId( errorDoc )
	
	return derefOpt( errorDoc, '_serverProvidedMetadata', 'new_id' )

end

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

local function isErrorFromALocalDelete( errorDoc, missingResourceDocValue, docThatIncurredError )

	local missingResourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'missing_resource_id' )

	return missingResourceId
				and missingResourceId == derefOpt( docThatIncurredError, '_id' )
				and derefOpt( docThatIncurredError, '_deleted' )
				and missingResourceDocValue._newId

end

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

local function isErrorFromALocalRestore( errorDoc, missingResourceDocValue, docThatIncurredError )

	local missingResourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'missing_resource_id' )

	return missingResourceId
				and missingResourceId ~= derefOpt( docThatIncurredError, '_id' )
				and WFCatTrashUtils.getCreateFromId( docThatIncurredError ) == missingResourceId
				and not derefOpt( docThatIncurredError, '_deleted' )

end

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

local function isDeletedByAnotherUserWithLocalNewId( self, errorDoc, missingResourceDocValue, docThatIncurredError )
	
	local maybeDeletedBy = derefOpt( errorDoc, '_serverProvidedMetadata', 'deleted_by' )
	local maybeNewId = derefOpt( docThatIncurredError, '_newId' )
	
	return not not ( maybeDeletedBy and maybeDeletedBy ~= self._catalog:ozUserId() and
		maybeNewId )

end

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

local function markDocumentForRetry( docStore, docValue, errorDoc )
	
	local spm =  derefOpt( errorDoc, '_serverProvidedMetadata' )
	local statusCode = tonumber( derefOpt( spm, 'status' ) )
	local existingRetries = tonumber( derefOpt( docValue, '_localOnly', 'ozError', 'retries' ) ) or 0
	local maybeLastRetry = derefOpt( docValue, '_localOnly', 'ozError', 'lastRetry' )
	
	local maybeOverlayObservable
	-- if we've already retried 3 times in the last hour, don't do it again as it seems likely 
	-- to not really be a retryable error and instead something more persistent. a manual change
	-- may be required.
	if existingRetries < 3 or lastRetryIsOlderThanXHours( maybeLastRetry, 1 ) then
		if existingRetries >= 3 then
			-- if we'd already hit 3 retries from a previous series of errors then we got
			-- here because the last retry was old, so reset the retry count and try again
			existingRetries = 0
		end
		
		maybeOverlayObservable = docStore:applyOverlay {
				_id = docValue._id,
				_localOnly = {
					ozError = {
						subtype = errorDoc.subtype,
						status = statusCode,
						retries = existingRetries + 1,
						lastRetry = KSDate.timeInIso8601Format()
					}
				}
			}
		:map( function()
			return "MarkedForRetry"
		end )

	else
		logger:warnf( "Exceeded retries for error document %s", WFTableArchiving.tableToString( errorDoc ) )
		maybeOverlayObservable = Observable.just( "Ignored_ExceededRetries" )
	end
	
	return maybeOverlayObservable
		:map( function( resolution )
			return resolution
		end )

end

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

local function defaultHandler( self, errorDoc, docValue )
	
	local docStore = self._catalog:docStore()
	
	local stringOfErrorDoc = WFTableArchiving.tableToString( errorDoc )
	
 	local spm =  derefOpt( errorDoc, '_serverProvidedMetadata' )
	local statusCode = tonumber( derefOpt( spm, 'status' ) )
 	local idOfDocThatIncurredError = derefOpt( spm, 'source', 'id' )
 	local actionOnDocThatIncurredError = derefOpt( spm, 'source', 'action' )
 
 	local errorEvent = {}
 	errorEvent['event.type'] = 'error'
 	errorEvent['event.workflow'] = 'SYNC'
 	errorEvent['event.subcategory'] = 'error_handler'
 	errorEvent['event.error_type'] = tostring( errorDoc.subtype )
 	errorEvent['event.error_desc'] = stringOfErrorDoc
	errorEvent['event.error_code'] = statusCode
 	errorEvent['content.id'] = idOfDocThatIncurredError
	errorEvent['content.subtype'] = derefOpt( docValue, 'subtype' )
	errorEvent['content.action'] = actionOnDocThatIncurredError
	
 	WFAnalytics.trackEvent( errorEvent )

	-- on 5XX errors - regardless of the error type, we should be able to retry the document
	-- update on Oz. So annotate the document with some error info which will also update the
	-- sequence and cause it to try to be resent	
	if statusCode and ( 500 <= statusCode and statusCode < 600 ) then

		return resolveErrorWithResult( self, errorDoc, "MarkedForRetryByDefaultHandler",
			markDocumentForRetry( docStore, docValue, errorDoc ) )

	else
		-- Log a warning that we received an error document from Oz that we are not sure how to handle.
		logger:warnf( "Unhandled error document type %s: %s", tostring( errorDoc.subtype ), WFTableArchiving.tableToString( errorDoc ) )

		-- mark it so we're not stuck on it.
		return resolveErrorAsUnknown( self, errorDoc, "MarkedAsUnknownByDefaultHandler" )

	end	
end

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

local function payloadValidationHandler( self, errorDoc, docValueThatIncurredError )
	
	local spm =  derefOpt( errorDoc, '_serverProvidedMetadata' )
	local resourceId = derefOpt( spm, 'source', 'id' )
	local resourceRevId = derefOpt( spm, 'source', 'revision_id' )
	local typeOfDocThatIncurredError = derefOpt( spm, 'source', 'type' )
	local actionOnDocThatIncurredError = derefOpt( spm, 'source', 'action' )
	
	local errors = derefOpt( spm, 'errors' )
	
	logger:warnf( "payload validation error for %s revId: %s", resourceId, resourceRevId )
	
	local overlay = {}
	
	local hasUnknownError = false
	
	if errors and type( errors ) == 'table' then
		
		for key, errorCode in pairs( errors ) do
			
			-- errorCode comes from Oz as an array of a single item - the actual errorCode
			errorCode = type( errorCode ) == 'table' and derefOpt( errorCode, 1 ) or errorCode
			
			logger:warnf( "payload validation error on '%s' for %s revId: %s", key, resourceId, resourceRevId )
			
			local docPath = parseDocPathFromKey( key, docValueThatIncurredError )
			
			if docPath then
				
				local existingValue = derefOpt( docValueThatIncurredError, unpack( docPath ) )
				
				local fixFn = maybeGetFixFunctionForKey( key )
				if fixFn then
				
					fixFn( overlay, docPath, existingValue, errorCode )
					
				else
					
					if errorCode ~= 'required' and existingValue then 
						-- If it's required and we remove it again, we've created a looper, so don't
						-- Also, if we don't have it...don't remove it, that may add cruft back in (NMB-11045) and so 
						-- we shouldn't just trust Oz saying we sent something at this path
						removeOffendingKey( overlay, docPath )
					end
					
				end
			
			else
				
				hasUnknownError = true
				
			end
			
		end
		
	end
	
	local errorEvent = {}
			
	local interestingSourceDocSections = KSTableUtils.filter( 
		function( key, value )
			return key == 'importSource'
		end, docValueThatIncurredError )
	
	errorEvent['event.type'] = 'error'
	errorEvent['event.workflow'] = 'SYNC'
	errorEvent['event.subcategory'] = 'error_handler'
	errorEvent['event.error_type'] = 'PayloadValidationError'
	errorEvent['event.error_desc'] = WFTableArchiving.tableToString( errorDoc )
	errorEvent['event.error_code' ] = hasUnknownError and 'UnknownError'
	errorEvent['event.value'] = WFTableArchiving.tableToString( overlay )
	
	errorEvent['content.id'] = resourceId
	errorEvent['content.revision_id'] = resourceRevId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.subtype'] = derefOpt( docValueThatIncurredError, 'subtype' )
	errorEvent['content.action'] = actionOnDocThatIncurredError
	errorEvent['content.value'] = WFTableArchiving.tableToString( interestingSourceDocSections )
	
	WFAnalytics.trackEvent( errorEvent )
	
	if hasUnknownError and not next( overlay ) then 
		-- there's nothing we fixed and at least one error seen was unknown
		return resolveErrorAsUnknown( self, errorDoc, "MarkedAsUnknown" )
	end
	
	if next( overlay ) then
		overlay._id = resourceId
		return resolveErrorByApplyingOverlay( self, errorDoc, "AppliedOverlay", overlay )
	end

	return resolveErrorByIgnoring( self, errorDoc, "NoOverlayToApply", overlay )

end

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

local function inputValidationHandler( self, errorDoc, docValue ) 

	local catalog = self._catalog
	
	local spm =  derefOpt( errorDoc, '_serverProvidedMetadata' )
	local resourceId = derefOpt( spm, 'source', 'id' )
	local resourceRevId = derefOpt( spm, 'source', 'revision_id' )
	local typeOfDocThatIncurredError = derefOpt( spm, 'source', 'type' )
	local actionOnDocThatIncurredError = derefOpt( spm, 'source', 'action' )
	
	local errors = derefOpt( spm, 'errors' )
	local errorKey, errorValue
	if errors and type( errors ) == 'table' then
		
		-- the input validation table is in an odd form like 
		-- errors = {
		--     download = {
		--         "must be true or false" 
		--     }
		-- } 
		-- so parse the last level like an array
		errorKey = next( errors )
		if errorKey then
			errorValue = derefOpt( errors, errorKey, 1 )
		end
		
	end
	
	logger:warnf( "InputValidationError for %s revId: %s", resourceId, resourceRevId )
	
	return Observable.just( docValue )
		
		:flatMap( function( docValueThatIncurredError )
			
			if errorKey and inputValidationFixups[ errorKey ] then
				
				local fixFn = inputValidationFixups[ errorKey ]
						
				return fixFn( catalog, docValueThatIncurredError, errorKey, errorValue )
						:map( function( wasHandled )
							return not not wasHandled
						end )
			end
			
			return Observable.just( false )
		end )
		:flatMap( function ( wasHandled )

			 if wasHandled then
				return Observable.just( true )
			else
				-- this error is for us, but we don't know what to do with it
				-- mark the error document as unable to be handled right now so 
				-- we don't try to process it anymore
				return self._catalog:docStore():applyOverlay( {
							_id = errorDoc._id,
							_localOnly = {
								localError = {
									[ errorDoc.subtype ] = {
										handlingFailed = true
									}
								},
							}
						} )
						:map( function()
							-- don't delete it, we may be able to process in the future
							return false 
						end )
			end
		end )
		:flatMap( function( wasHandled )
			if wasHandled then
			
				-- fire an analytics event that we should be able to fix the error in the document
				local errorEvent = {}
				errorEvent['event.type'] = 'error'
				errorEvent['event.workflow'] = 'SYNC'
				errorEvent['event.subcategory'] = 'error_handler'
				errorEvent['event.error_type'] = WFCatOzErrors.Errors.kInputValidationError
				errorEvent['event.error_desc'] = errorValue
				errorEvent['event.error_code'] = 'fixed'
				
				errorEvent['content.id'] = resourceId
				errorEvent['content.revision_id'] = resourceRevId
				errorEvent['content.type'] = typeOfDocThatIncurredError
				errorEvent['content.subtype'] = derefOpt( docValue, 'subtype' )
				errorEvent['content.action'] = actionOnDocThatIncurredError
				errorEvent['content.url'] = errorKey
				
				-- track the error event
				WFAnalytics.trackEvent( errorEvent )

				-- delete error doc once we've marked the resource as storage exceeded
				return resolveError( self, errorDoc, "Fixed", Observable.just( true ) )
			else
				-- if we don't know how to handle a specific InputValidationError, use the default handler to mark it as 
				-- unhandled rather than deleting it as a future update might be able to help here.
				return defaultHandler( self, errorDoc, docValue )
			end
		end )
		
end

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

local function createBaseErrorEvent()

	local errorEvent = {}
	errorEvent['event.type'] = 'error'
	errorEvent['event.workflow'] = 'SYNC'
	errorEvent['event.subcategory'] = 'error_handler'

	return errorEvent

end

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

local function resourceExistsHandler( self, errorDoc, docThatIncurredError )

	local catalog = self._catalog
	local docStore = catalog:docStore()
	
	local spmSource =  derefOpt( errorDoc, '_serverProvidedMetadata', 'source' )
	
	local idOfDocThatIncurredError = derefOpt( spmSource, 'id' )
	local typeOfDocThatIncurredError = derefOpt( spmSource, 'type' )
	local actionOnDocThatIncurredError = derefOpt( spmSource, 'action' )
	
	local documentUrl = derefOpt( spmSource, 'links', 'self', 'href' )

	local eventErrorCode

	local handleErrorDoc
	if not documentUrl then
									
		logger:warnf( "No document url for ResourceExistsError for resource id of %s, ignoring error doc", idOfDocThatIncurredError )
		handleErrorDoc = Observable.just()
		
	else 
	
		-- let's handle the happy path...we're online...so check whether Oz has the asset by
		-- directly asking for it. If it does, we can ignore the ResourceExistsError - sync
		-- should resolve this eventually. If it does not, then it's a deleted asset and so 
		-- we should delete it locally
		handleErrorDoc = catalog:ozCatalogSpec():get( documentUrl )
			:flatMap( function( result, headers )
									
				if headers.status_code == '200' then
					
					local ok, resultJson = pcall( KSJsonDecoder.decodeWebSafeJson, result )
					resultJson = ok and resultJson or {}
					
					local isDeleted = derefOpt( resultJson, 'deleted' ) == true
					
					if docThatIncurredError.type == 'asset' and isDeleted then
						
						if not self._deleteAssetsFn then
							logger:errorf( "Setup error: no deleteAssets implementation" )
							
							return Observable.just()
						end

						-- if it's an asset document, we need to clean up any associated local files
						-- NOTE: We use concat here because deleteAssets impls are provided by the client and we
						-- can't assume that it will give us only a single event. So take only the completion with 
						-- takeLast( 0 ) and then concat with Observable.just so we do get a single value for our
						-- purposes
						eventErrorCode = 'deleted'
						return Observable.concat( 
								self._deleteAssetsFn( catalog, { idOfDocThatIncurredError } ):takeLast( 0 ),
								Observable.just() )
					
					elseif docThatIncurredError.type == 'album' and isDeleted then
						
						if not self._deleteIndividualAlbumFn then
							logger:errorf( "Setup error: no deleteIndividualAlbum implementation" )
						
							return Observable.just()
						end
						
						-- if it's an album we need to clean up and associated album_assets)
						-- NOTE: We use concat here because deleteIndividualAlbum impls are provided by the 
						-- client and we can't assume that it will give us only a single event. So take only 
						-- the completion with takeLast( 0 ) and then concat with Observable.just so we do 
						-- get a single value for our purposes
						eventErrorCode = 'deleted'
						return Observable.concat( 
								self._deleteIndividualAlbumFn(  catalog, idOfDocThatIncurredError ):takeLast( 0 ),
								Observable.just() )
					
					elseif docThatIncurredError.type == 'album_asset' and ( isDeleted or resultJson.id ~= idOfDocThatIncurredError ) then
					
						
						-- album_assets may get a ResourceExistsError if the same album_asset association is
						-- made with a different album_asset_id than is already being used. so also delete the 
						-- album_asset doc in that case
						if resultJson.id ~= idOfDocThatIncurredError then 
							eventErrorCode = 'duplicate'
						else
							eventErrorCode = 'deleted'
						end
						
						return docStore:deleteDoc( idOfDocThatIncurredError )
						
					elseif docThatIncurredError.type == 'favorite' and ( isDeleted or resultJson.id ~= idOfDocThatIncurredError ) then
					
						
						-- 'favorite' may get a ResourceExistsError if a user marked the same shared asset document as 
						-- favorite from two different clients and let them sync to Oz. In this case, we should delete
						-- the favorite doc that triggered the error
						if resultJson.id ~= idOfDocThatIncurredError then 
							eventErrorCode = 'duplicate'
						else
							eventErrorCode = 'deleted'
						end
						
						return docStore:deleteDoc( idOfDocThatIncurredError )
						
					elseif isDeleted then
						
						-- other doc types just delete the doc itself
						eventErrorCode = 'deleted'
						return docStore:deleteDoc( idOfDocThatIncurredError )
						
					else
						
						eventErrorCode = 'missing confirmation'
						
						-- don't alter the existing document based on the error doc received, trust the sync process
						-- to resolve the issue that caused the ResourceExistsError
						return Observable.just( KSJsonDecoder.decodeWebSafeJson_stripNull( result ) )
							:map( function( decodedDocumentJson )
								
								local docValue = WFOzDocUtils.remapBasicOzDocument( decodedDocumentJson, decodedDocumentJson.base )
								
								return docValue						
							end )
							:flatMap( function( docValue )
								
								if not docValue or docValue._deleted then
									return Observable.just()
								end
								
								return self._catalog:docStore():bulkSetDocValue( { label = 'oz',
                                                                                     docValue } )
							end )
							
					end
				end
				
				return Observable.just( false )
				
			end )
			:catch( function( responseData, headers )
				logger:warnf( "Error retrieving doc in ResourceExistsError for resource id of %s status_code %s, ignoring error doc", idOfDocThatIncurredError, derefOpt( headers, 'status_code' ) )
				return resolveError( self, errorDoc, "ignoredNotFound", Observable.just( false ) )
			end )
	end
	
	return handleErrorDoc:flatMap( function( result )

		local errorEvent = createBaseErrorEvent()
		errorEvent['event.error_type'] = WFCatOzErrors.Errors.kResourceExistsError
		errorEvent['event.error_code'] = eventErrorCode
		errorEvent['content.id'] = idOfDocThatIncurredError
		errorEvent['content.type'] = typeOfDocThatIncurredError
		errorEvent['content.subtype'] = derefOpt( docThatIncurredError, 'subtype' )
		errorEvent['content.action'] = actionOnDocThatIncurredError

		-- track the error event
		WFAnalytics.trackEvent( errorEvent )

		return resolveError( self, errorDoc, eventErrorCode or "ignored", Observable.just( result ) )

	end )

end

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

local function inlineSettingsHandler( self, errorDoc, docValue )
	
	local catalog = self._catalog
	
	local maybeLimit = derefOpt( errorDoc, '_serverProvidedMetadata', 'limit' )
	
	local spmSource =  derefOpt( errorDoc, '_serverProvidedMetadata', 'source' )	
	local idOfDocThatIncurredError = derefOpt( spmSource, 'id' )
	local typeOfDocThatIncurredError = derefOpt( spmSource, 'type' )
	local actionOnDocThatIncurredError = derefOpt( spmSource, 'action' )

	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kInlineXmpDevelopSettingTooBigError
	
	errorEvent['content.id'] = idOfDocThatIncurredError
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.subtype'] = derefOpt( docValue, 'subtype' )
	errorEvent['content.action'] = actionOnDocThatIncurredError
	
	WFAnalytics.trackEvent( errorEvent )
	
	return resolveError( self, errorDoc, "ExternalizedXMP",
		self._externalizeSettingsFn( catalog, idOfDocThatIncurredError, tonumber( maybeLimit ) ) )

end

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

-- Generate an overlay for the trash asset from Oz from the local trash asset
-- before deleting the local trash asset. (Scenario: An asset was modified
-- and deleted from a paused sync client, while it was also deleted from a
-- different online/syncing client. The other client's trash asset gets kept,
-- after grabbing applicable data from the local trash asset and merging it in.)

local function mergeDeleteOrUpdateCreateFromOfLocalDocWithOzDoc( catalog, newAssetIdFromOz, localAssetId )

	local docStore = catalog:docStore()

	return WFCatTrashUtils.getTrashOrRestoredDocFromNewId( docStore, localAssetId )
	:combineLatest( docStore:docValue( newAssetIdFromOz ) )
	:flatMap( function( localTrashDoc, ozTrashDoc )

		if derefOpt( localTrashDoc, '_deleted' ) then

			-- Asset was permanently deleted locally, delete the oz doc.
			return docStore:applyOverlay ( {
				_id = newAssetIdFromOz,
				_deleted = true,
			} )
			:map( function()
				return "DeletedOzAssetLocalWasPermanentlyDeleted"
			end )

		elseif localTrashDoc._id ~= localAssetId or ozTrashDoc._deleted or localTrashDoc.subtype ~= ozTrashDoc.subtype then

			-- Asset was deleted and restored locally, update the createFrom to
			-- create it off of the other client's doc instead. Or, if the newId
			-- provided by oz was also deleted just update the createFrom.

			return WFCatTrash.updateLocalCreateFromTargetingOzDoc( docStore, localTrashDoc, ozTrashDoc )
			:map( function()
				return "UpdatedLocalCreateFromTargetingOzDoc"
			end )

		elseif derefOpt( localTrashDoc, '_revId' ) ~= WFCatTrashUtils.getCreateWithRevisionId( localTrashDoc ) then

			return WFCatTrash.createOverlaysToMergeRestoredAssets( catalog, localTrashDoc, ozTrashDoc )
			:flatMap( function( overlays )
				return docStore:bulkApplyOverlay( overlays )
			end )
			:map( function()
				return "CreatedOverlaysToMergeTheRestoredAssetsDeletingLocal"
			end )

		else

			-- just delete the local trash asset
 			return docStore:applyOverlay ( {
				_id = localAssetId,
				_deleted = true,
			} )
			:map( function()
				return "DeletedLocalAsset"
			end )

		end

	end )

end

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

local function mergeOrUpdateCreateFromOfRestoredLocalDocWithOzDoc( catalog, ozRestoredAssetId, localRestoredDoc )
	
	local docStore = catalog:docStore()
	
	return docStore:docValue( ozRestoredAssetId )
		:flatMap( function( ozRestoredDoc )
			
			if ozRestoredDoc._deleted then

				return WFCatTrash.updateLocalCreateFromTargetingOzDoc( docStore, localRestoredDoc, ozRestoredDoc )
					:map( function()
						return "UpdatedLocalCreateFromTargetingOzDoc"
					end )

			end
			
			return WFCatTrash.createOverlaysToMergeRestoredAssets( catalog, localRestoredDoc, ozRestoredDoc )
				:flatMap( function( overlays )
					
					if #overlays >= 2 and overlays[1]._id == overlays[2]._id then
						logger:errorf( "mergeOrUpdateCreateFromOfRestoredLocalDocWithOzDoc updating and deleting the same asset id %s", overlays[1]._id )
					end

					return docStore:bulkApplyOverlay( overlays )
						:map( function()
							return "CreatedOverlaysToMergeTheRestoredAssetsDeletingLocal"
						end )

				end )
		
		end )
end

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

local function createAndApplyOverlaysToCreateAssetDirectly( catalog, docThatIncurredError )

	local overlay = WFCatTrash.overlayForRemoveTrashInfo( docThatIncurredError )

	-- NMB-9395/NMB-9398: The document that incurred the error was a cloud trash document (e.g. restore from trash)
	-- where the source document was no longer on Oz. In this case, just turn this into a regular doc
	-- create by removing the createFrom so that we push a regular create.

	-- When we turn this into a proper create, we need to add the userCreated field
	local createFromInfo = WFCatTrashUtils.createFromInfo( docThatIncurredError )
	overlay.userCreated = createFromInfo.lastTrashUserCreated

	-- also, check the local binaries - if we don't have a master, but *do* have a proxy, we need to turn this into a
	-- Lightroom style asset (willNotUploadMaster = true)
	local hasMasterLocally = WFCatAssetUtils.hasLocalOriginal( docThatIncurredError )
	local hasProxyLocally = WFCatAssetUtils.hasLocalProxy( docThatIncurredError )

	if hasProxyLocally and not hasMasterLocally then
		overlay.importSource = overlay.importSource or {}
		overlay.importSource.willNotUploadMaster = true
	end

	-- clean uploaded info for a clean asset create
	if hasMasterLocally then
		local deletionModifier = { _delete = true }

		if WFPlatformInfo.isAndroid then
			setNested( overlay, deletionModifier, '_localOnly', 'original', 'ozUrl' )
			setNested( overlay, deletionModifier, '_localOnly', 'original', 'uploaded' )
		elseif WFPlatformInfo.isiOS then
			setNested( overlay, deletionModifier, '_localOnly', 'files', 'master', 'ozUrl' )
			setNested( overlay, deletionModifier, '_localOnly', 'files', 'master', 'uploaded' )
		end
	end

	return WFCatTrash.moveAssociatedDocsToNewDocs( catalog, overlay._id )
	:flatMap( function( overlays )

		-- Add the overlay for the asset
		overlays[ #overlays + 1 ] = overlay

		return catalog:docStore():bulkApplyOverlay( overlays )

	end )

end

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

local function isImageOrVideo( doc )

	return derefOpt( doc, 'type' ) == 'asset'
		and (
			derefOpt( doc, 'subtype' ) == 'image'
			or derefOpt( doc, 'subtype' ) == 'video'
		)

end

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

local function resourceNotFoundHandler( self, errorDoc, docThatIncurredError )
	local catalog = self._catalog
	local docStore = catalog:docStore()
	
	local spm =  derefOpt( errorDoc, '_serverProvidedMetadata' )
	local idOfDocThatIncurredError = derefOpt( spm, 'source', 'id' )
	local typeOfDocThatIncurredError = derefOpt( spm, 'source', 'type' )
	local subtypeOfDocThatIncurredError = derefOpt( spm, 'source', 'subtype' )
	local actionOnDocThatIncurredError = derefOpt( spm, 'source', 'action' )
	
	local missingResourceId = derefOpt( spm, 'missing_resource_id' ) or idOfDocThatIncurredError
	logger:infof( "received resource not found error for %s", missingResourceId )
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kResourceNotFoundError
	
	errorEvent['content.id'] = idOfDocThatIncurredError
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.subtype'] = derefOpt( docThatIncurredError, 'subtype' )
	errorEvent['content.action'] = actionOnDocThatIncurredError
	errorEvent['content.missing_resource_id'] = missingResourceId
	
	local maybeNewAssetIdFromOz = maybeTrashOperationNewId( errorDoc )
	
	return docStore:docValue( missingResourceId )
		:flatMap( function( missingResourceDocValue )
		
			local errors = derefOpt( spm, 'errors' )
			local deleteWasSentToOzBeforeAssetExisted = missingResourceDocValue._deleted and missingResourceDocValue._newId and not maybeTrashOperationNewId( errorDoc )
			local docThatIncurredErrorWasTrashAsset = missingResourceDocValue._empty and errors[ 'deleted_asset' ] and ( WFCatTrashUtils.getCreateFromId( docThatIncurredError ) == missingResourceId )

			-- If multiple trash operations take place on a different client for the same asset
			-- that also undergoes multiple trash operations on this client, we resort to the
			-- following to identify the scenario, rather than trying to follow the new_id trail
			-- to the end.

			local localRestoreOfAssetMoreThanOneTrashOperationBehindOz = isImageOrVideo( docThatIncurredError )
					and derefOpt( missingResourceDocValue, '_deleted' )
					and derefOpt( missingResourceDocValue, '_labels', '_dontsendtooz' )
					and actionOnDocThatIncurredError == 'create'

			if maybeNewAssetIdFromOz then
				
				if ( actionOnDocThatIncurredError == 'create' and missingResourceId == WFCatTrashUtils.getCreateFromId( docThatIncurredError ) and maybeNewAssetIdFromOz == idOfDocThatIncurredError )
						-- restore where the missing resource is the deleted trash asset and the doc that incurred error is the new restored asset doc
					or
					( actionOnDocThatIncurredError == 'delete' and missingResourceId == docThatIncurredError._id and maybeNewAssetIdFromOz == docThatIncurredError._newId )
						-- delete to trash where the missing resource is the asset being deleted and the doc that caused the error and the new asset id is the same as the deleted assets new id
					then
					-- NMB-12753/NMB-12848: If we get a ResourceNotFoundError for a doc that we're trying to delete and the
					-- new_id in the ResourceNotFoundError matches the _newId that we were trying to create from that
					-- then we want to ignore this ResourceNotFoundError because it's basically saying this operation
					-- has already happened

					return resolveErrorByIgnoring( self, errorDoc, "OperationAlreadyHappened" )

				elseif isErrorFromALocalDelete( errorDoc, missingResourceDocValue, docThatIncurredError ) then

					return resolveErrorWithResult( self, errorDoc, "LocalDelete",
						mergeDeleteOrUpdateCreateFromOfLocalDocWithOzDoc( catalog, maybeNewAssetIdFromOz, missingResourceDocValue._newId ) )

				elseif isErrorFromALocalRestore( errorDoc, missingResourceDocValue, docThatIncurredError ) then

					return resolveErrorWithResult( self, errorDoc, "LocalRestore",
						mergeOrUpdateCreateFromOfRestoredLocalDocWithOzDoc( catalog, maybeNewAssetIdFromOz, docThatIncurredError ) )

				elseif docThatIncurredError._deleted and docThatIncurredError._newId and actionOnDocThatIncurredError == 'create' then

					-- Handle cases where both a different client and this client restored the photo,
					-- and this client also deleted it again and maybe did other trash actions as well.
					return resolveErrorWithResult( self, errorDoc, "LocalRestoreDeleted",
						mergeDeleteOrUpdateCreateFromOfLocalDocWithOzDoc( catalog, maybeNewAssetIdFromOz, docThatIncurredError._id ) )

				else
					logger:warnf( "ignoring %s error from Oz for %s with new_id %s", errorDoc.subtype, idOfDocThatIncurredError, maybeNewAssetIdFromOz )
				end
			
			elseif isDeletedByAnotherUserWithLocalNewId( self, errorDoc, missingResourceDocValue,  docThatIncurredError ) then
				
				return WFCatTrashUtils.getTrashOrRestoredDocFromNewId( docStore, docThatIncurredError._newId )
						:flatMap( function( localDocValue )
							return resolveError( self, errorDoc, "RemovedLocalDocument_DeletedByContributor",
					 			mimicOzDeletionEvent( docStore, localDocValue._id, typeOfDocThatIncurredError, subtypeOfDocThatIncurredError ) )
						end )

			elseif localRestoreOfAssetMoreThanOneTrashOperationBehindOz then

				-- Handled via the Delete RNFE which updates the createFromId.
				return resolveErrorByIgnoring( self, errorDoc, "MoreThanOneOperationBehind" )

			elseif docThatIncurredErrorWasTrashAsset then

				return resolveError( self, errorDoc, "RemovedTrashInfoToCreateAssetDirectly_docThatIncurredErrorWasTrashAsset",
					createAndApplyOverlaysToCreateAssetDirectly( catalog, docThatIncurredError ) )

			elseif deleteWasSentToOzBeforeAssetExisted then

				if docThatIncurredError._deleted then

					-- Get the local asset that exists corresponding to docThatIncurredError
					-- by following the _newId change, and create an overlay to remove the
					-- trash info and update importSource.willNotUploadMaster if need be.

					return WFCatTrashUtils.getTrashOrRestoredDocFromNewId( docStore, docThatIncurredError._newId )
						:flatMap( function( localDocValue )

							if localDocValue._deleted then

								-- User permanently deleted the photo, ignore the error.
								return resolveErrorByIgnoring( self, errorDoc, "AssetWasPermanentlyDeleted_deleteWasSentToOzBeforeAssetExisted" )

							elseif WFCatTrashUtils.getCreateFromId( localDocValue ) == missingResourceId then

								-- Since thing createFromId matches the missingResourceId and there isn't
								-- a new_id provided by oz (as maybeNewAssetIdFromOz is nil) we can't do
								-- anything but remove the createFrom info and create the photo from scratch.

								return resolveError( self, errorDoc, "RemovedTrashInfoToCreateAssetDirectly_deleteWasSentToOzBeforeAssetExisted",
									createAndApplyOverlaysToCreateAssetDirectly( catalog, localDocValue ) )

							else

								-- The error was reported for the second part of a two
								-- part delete/restore or restore/delete which should just be
								-- ignored as the first part will update or remove the createFrom info.
								-- to retry the operation.
								return resolveErrorByIgnoring( self, errorDoc, "SecondPartOfTrashPair_deleteWasSentToOzBeforeAssetExisted" )

							end

						end )

				else

					-- This is hit by test_CT_MockRestore_OtherRestoreOtherDelete
					return resolveErrorByIgnoring( self, errorDoc, "_deleteWasSentToOzBeforeAssetExisted" )

				end

			elseif idOfDocThatIncurredError == missingResourceId then

				-- Is the missing resource id the same as the id of the document we were trying to send?
				-- if so, then we either have a create/update to a doc that doesn't exist on Oz yet
				-- (ignore the error, it should clear up eventually) or an update to a doc that doesn't
				-- exist on Oz any more (remove the document locally)
				
				local didPreviouslyExistOnOz = not not derefOpt( missingResourceDocValue, '_serverProvidedMetadata', 'created' )
								
				if missingResourceDocValue._empty or didPreviouslyExistOnOz then

					return resolveError( self, errorDoc, "RemovedLocalDocument_MatchingMissingResourceId",
						 mimicOzDeletionEvent( docStore, missingResourceId, typeOfDocThatIncurredError, subtypeOfDocThatIncurredError ) )

				else
					logger:warnf( "ignoring %s error from Oz for idOfDocThatIncurredError == missingResourceId == %s", errorDoc.subtype, idOfDocThatIncurredError )
				end
			
			else 
				
				-- If the missing resource id is not the same as the id of the document we were trying to send, 
				-- then we are looking at doc that incurred the error as a dependent document of the missing 
				-- resource. In that case we'll need to determine why the missing resource is missing and 
				-- mark the dependent document with the same error message. This will allow us to clear the
				-- dependent documents error at the same time as it's 'parent' resource
				
				
				if missingResourceDocValue._empty and isDependentDocumentType( typeOfDocThatIncurredError ) then

					-- the missing resource is missing becaue it's deleted. so we should delete the dependent
					-- document too

					return resolveError( self, errorDoc, "RemovedLocalDocument_NotMatchingMissingResourceId",
						docStore:deleteDoc( idOfDocThatIncurredError ) )

				else
				
					local ozErrorForMissingResource = WFCatOzErrors.maybeGetOzErrorHash( missingResourceDocValue )
					
					local errorOverlay
					if ozErrorForMissingResource then 
						
						errorOverlay = {
								_id = idOfDocThatIncurredError,
								_localOnly = { 
									ozError = ozErrorForMissingResource
								}
							}
						
						errorEvent[ 'content.missing_resource_error_subtype' ] = ozErrorForMissingResource.subtype
						
					else 
						
						errorOverlay = WFCatOzErrors.createOverlayForOzErrorSubtype( idOfDocThatIncurredError, errorDoc.subtype )
						
					end

					return resolveErrorByApplyingOverlay( self, errorDoc, "MarkedWithOzError_NotMatchingMissingResourceId", errorOverlay )

				end				
			
			end
			
			-- If we receive a resource not found error for other reasons than those handled explicitly we are 
			-- hopefully in a situation that should clear itself up. Log that we received this error, but there
			-- is not much more we should do to handle the ResourceNotFound error locally.
			logger:warnf( "ignoring %s error from Oz for %s", errorDoc.subtype, idOfDocThatIncurredError )

			return resolveErrorByIgnoring( self, errorDoc, "UnknownReason" )

		end )
		
		:map( function( resolution )
		
			errorEvent['event.resolution'] = resolution
			WFAnalytics.trackEvent( errorEvent )
		
		end )
		
end

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

local function assetGoneHandler( self, errorDoc, assetGoneDoc )

	local resourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'id' )
	local typeOfDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'type' )
	local actionOnDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'action' )
	local newAssetId = derefOpt( errorDoc, '_serverProvidedMetadata', 'new_id' )

	logger:infof( "received assetGoneError for %s id %s updating to asset.id = %", typeOfDocThatIncurredError, resourceId, newAssetId )

	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kAssetGoneError
	errorEvent['content.id'] = resourceId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.action'] = actionOnDocThatIncurredError
	errorEvent['content.new_id'] = newAssetId
	WFAnalytics.trackEvent( errorEvent )

	local isErrorForAssociatedDoc =
		typeOfDocThatIncurredError == 'album_asset' or
		typeOfDocThatIncurredError == 'comment' or
		typeOfDocThatIncurredError == 'favorite' or
		typeOfDocThatIncurredError == 'version'

	if isErrorForAssociatedDoc then

		-- Update the asset.id in the document that failed with error 410
		return resolveErrorByApplyingOverlay( self, errorDoc, "UpdatedAssetIdOfAssocDoc", {
			_id = resourceId,
			asset = {
				id = newAssetId
			}
		} )

	else

		-- Don't know what to do with this type of asset, just add an OzError overlay
		return resolveErrorByMarkingWithOzError( self, errorDoc, resourceId )

	end

end

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

local function storageExceededHandler( self, errorDoc, storageExceededDoc )
	 
	local resourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'id' )
	local typeOfDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'type' )
	local actionOnDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'action' )
	local spaceId = errorDoc._spaceId
	
	logger:infof( "received storage exceeded error for %s", resourceId )
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kStorageExceededError
	
	errorEvent['content.id'] = resourceId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.action'] = actionOnDocThatIncurredError
	errorEvent['lr.wf.spaceId'] = spaceId
	
	WFAnalytics.trackEvent( errorEvent )

	-- Note: if storageExceededDoc._deleted and storageExceededDoc._newId are set
	-- the applyOverlay call will forward the change to the destination.
	return resolveErrorByMarkingWithOzError( self, errorDoc, resourceId )

end

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

local function multipartCompleteHandler( self, errorDoc, multipartCompleteErrorDoc )
	
	local resourceId = KSTableUtils.derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'asset_id' )
	
	if not resourceId then
		resourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'id' )
	end
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kMultipartCompleteError
	
	errorEvent['content.id'] = resourceId

	WFAnalytics.trackEvent( errorEvent )
	
	if not multipartCompleteErrorDoc._empty then
	
		logger:infof( "received multipart complete error for %s", resourceId )
		return resolveErrorByMarkingWithOzError( self, errorDoc, resourceId )

	end
	
	return Observable.just()
		
end

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

local function trialOrSubscriptionEntitlementExpiredHandler( self, errorDoc, subscriptionExpiredDoc )

	local resourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'id' )
	local typeOfDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'type' )
	local actionOnDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'action' )
	local spaceId = errorDoc._spaceId
	
	logger:infof( "received trial/subscription expired error for %s", resourceId )
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError
	
	errorEvent['content.id'] = resourceId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.action'] = actionOnDocThatIncurredError
	errorEvent['lr.wf.spaceId'] = spaceId
	
	WFAnalytics.trackEvent( errorEvent )

	return resolveErrorByMarkingWithOzError( self, errorDoc, resourceId )

end

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

local function invalidAccessRequestHandler( self, errorDoc, docValue )

	local source = derefOpt( errorDoc, '_serverProvidedMetadata', 'source' )
	local spaceInviteId = derefOpt( source, 'id' )
	local typeOfDocThatIncurredError = derefOpt( source, 'type' )
	local actionOnDocThatIncurredError = derefOpt( source, 'action' )
	
	logger:infof( "invalid access-request error for %s", spaceInviteId )
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kInvalidAccessRequestError
	
	errorEvent['content.id'] = spaceInviteId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.action'] = actionOnDocThatIncurredError
	
	WFAnalytics.trackEvent( errorEvent )
	
	-- This error is sent if the client tries to approve a request - but that request has already
	-- been approved on Oz. Since the update that the client tried sending is the same as the 
	-- update that already happened on Oz, we don't need to do anything, other than delete the 
	-- error document.
	return resolveErrorByIgnoring( self, errorDoc, "AlreadyApproved" )

end

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

local function spaceUserExistsHandler( self, errorDoc, docValue )
	
	local source = derefOpt( errorDoc, '_serverProvidedMetadata', 'source' )
	local spaceInviteId = derefOpt( source, 'id' )
	local typeOfDocThatIncurredError = derefOpt( source, 'type' )
	local actionOnDocThatIncurredError = derefOpt( source, 'action' )
	
	logger:infof( "space user exists error for %s", spaceInviteId )
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kSpaceUserExistsError
	
	errorEvent['content.id'] = spaceInviteId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.action'] = actionOnDocThatIncurredError
	
	WFAnalytics.trackEvent( errorEvent )

	return resolveErrorByApplyingOverlay( self, errorDoc, "SpaceInviteDeleted", {
		_id = spaceInviteId,
		_deleted = true,
	} )

end

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

local function isRestoreAssetError( errorDoc )
	
	local spm = derefOpt( errorDoc, '_serverProvidedMetadata' )
	local action = derefOpt( spm, 'source', 'action' )
	
	return errorDoc.subtype == WFCatOzErrors.Errors.kRestoreAssetError and
		action == 'create'

end

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

local function isRestoreAssetErrorForThisClient( docStore, errorDoc, docThatIncurredError )
	-- NMB-13281/LRMI-4193007/LRMA-4917671/NMB-14814
	-- In certain restore cases Oz may partially create the restored asset and
	-- send it in the sync feed but fail before it's finally created and so it 
	-- will send a RestoreAssetError. This means that all clients will get the
	-- partial asset (that doesn't really exist on Oz) as well as the error doc. 
	
	-- This function returns true if the error is a 502/RestoreAssetError and
	-- originated from this client (docThatIncurredError should have local cloud
	-- trash data related to how to create the restored asset.
	local spm = derefOpt( errorDoc, '_serverProvidedMetadata' )
	local sourceDocId = derefOpt( spm, 'source', 'id' )
	local sourceDocRevId = derefOpt( spm, 'source', 'revision_id' )

	local maybeCreateWithRevId = WFCatTrashUtils.getCreateWithRevisionId( docThatIncurredError )

	if not isRestoreAssetError( errorDoc ) then

		-- I don't think you can get here, preserving previous implementation's check.
		return Observable.just( false )

	elseif sourceDocId == docThatIncurredError._id and sourceDocRevId == maybeCreateWithRevId then

		-- The doc still exists and it's the correct revision, return true as
		-- we had to have issued the restore request.
		return Observable.just( true )

	else

		local restoreFromDocId = derefOpt( spm, 'source', 'restore_from', 'id' )

		if restoreFromDocId == nil then
			-- Probably can't get here either.
			return Observable.just( false )
		end

		-- NMB-14814 Check to see if the errorDoc's restore_from doc has a _newId that
		-- points to the docThatIncurredError._id (also known as sourceDocId). If it
		-- does then this client issued the restore and either the user modified or
		-- deleted the asset.

		return docStore:docValue( restoreFromDocId )
		:map( function( restoreFromDocValue )

			-- Note that docThatIncurredError may be deleted if so it
			-- will have a _newId and the overlay will be applied to
			-- what that _newId points at by the docStore.

			return restoreFromDocValue._newId == docThatIncurredError._id

		end )

	end

end

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

local function isDeletedByAnotherUser( errorDoc, currentUserId )
	
	local subtype = errorDoc.subtype
	
	local maybeDeletedBy = derefOpt( errorDoc, '_serverProvidedMetadata', 'deleted_by' )
	
	return not not ( subtype == WFCatOzErrors.Errors.kResourceNotFoundError
		and maybeDeletedBy and maybeDeletedBy ~= currentUserId )
end

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

local function restoreAssetErrorHandler( self, errorDoc, docValue )
	-- NMB-13281/LRMI-4193007/LRMA-4917671
	-- When we receive a RestoreAssetError/502 error from Oz we may have received a partial
	-- asset confirmation from Oz but Oz didn't end up creating the restored asset. 
	local catalog = self._catalog
	local docStore = catalog:docStore()

	return isRestoreAssetErrorForThisClient( docStore, errorDoc, docValue )
	:flatMap( function( wasRestoredFromThisClient )

		if wasRestoredFromThisClient then

			-- If we are the client that sent the restore, then we need to resend the attempt to
			-- restore the asset which should either result in the successful restoration of the asset
			-- or a non-502 error (e.g. 404 if the asset was restored simultaneously in another client).

			-- In order to ensure this happens we need to
			-- 1. Remove *any* 'oz' label from revisions of this document.
			-- 2. Bump the sequence so the pusher will look at it again and attempt the restore again

			local source = derefOpt( errorDoc, '_serverProvidedMetadata', 'source' )
			local resourceId = derefOpt( source, 'id' )
			local typeOfDocThatIncurredError = derefOpt( source, 'type' )
			local actionOnDocThatIncurredError = derefOpt( source, 'action' )

			local errorEvent = {}
			errorEvent['event.type'] = 'error'
			errorEvent['event.workflow'] = 'SYNC'
			errorEvent['event.subcategory'] = 'error_handler'
			errorEvent['event.error_type'] = WFCatOzErrors.Errors.kRestoreAssetError

			errorEvent['content.id'] = resourceId
			errorEvent['content.type'] = typeOfDocThatIncurredError
			errorEvent['content.action'] = actionOnDocThatIncurredError

			WFAnalytics.trackEvent( errorEvent )

			return docStore:getRevisionForLabel( docValue._id, 'oz' )
				:flatMap( function( revisionId )
					return docStore:removeLabel( docValue._id, 'oz', revisionId )
						:takeLast(0)
						:concat( Observable.just( nil ) )
				end )
				:catch( function()
					-- no revision with 'oz' label so return nil and we'll
					return Observable.just( nil )
				end )
				:flatMap( function()
					return resolveErrorWithResult( self, errorDoc, "MarkedForRetryByThisClient",
						markDocumentForRetry( docStore, docValue, errorDoc ) )
				end )

		else

			-- If we are not the client that sent the restore that generated the RestoreAssetError then
			-- we need to make sure the partial doc doesn't look full enough to appear on the client so
			-- we will make sure that the _serverProvidedMetadata is removed
			if isRestoreAssetError( errorDoc ) and not docValue._empty then

				return resolveErrorByApplyingOverlay( self, errorDoc, "RemovedServerProvidedMetadata", {
					_id = docValue._id,
					_serverProvidedMetadata = { _delete = true },
				} )

			end

			return resolveErrorByIgnoring( self, errorDoc, "ErrorFromOtherClient" )

		end

	end )

end

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

local function insufficientPermissionErrorHandler( self, errorDoc, docValue )

	local resourceId = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'id' )
	local typeOfDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'type' )
	local actionOnDocThatIncurredError = derefOpt( errorDoc, '_serverProvidedMetadata', 'source', 'action' )
	local spaceId = errorDoc._spaceId
	
	logger:infof( "received insufficient permission error for %s", resourceId )
	
	local errorEvent = createBaseErrorEvent() 
	errorEvent['event.error_type'] = WFCatOzErrors.Errors.kInsufficientPermissionError
	errorEvent['content.id'] = resourceId
	errorEvent['content.type'] = typeOfDocThatIncurredError
	errorEvent['content.action'] = actionOnDocThatIncurredError
	errorEvent['lr.wf.spaceId'] = spaceId
	
	WFAnalytics.trackEvent( errorEvent )
	
	if typeOfDocThatIncurredError == 'asset' and 
			actionOnDocThatIncurredError == 'update' and
			errorDoc._spaceId then
			
		-- This handling addresses the following scenario:
		
		-- 1. Alice grants Bob editor role in a shared album containing SomeImage.
		-- 2. Bob syncs down SomeImage and goes offline where he makes edits.
		-- 3. In the meantime, Alice downgrades Bob's privileges SomeImage album to contributor while Bob is offline.
		-- 4. Bob comes back online
			
		-- After Step 4, Bobs edits are invalidated as he can no longer sync them. To resolve
		-- this, we completely throw away bobs local revision and resync the latest revision
		-- from Oz.
			
		return resolveByRestoringAssetFromOz( self, errorDoc, resourceId )
		
	else
	
		return resolveErrorByMarkingWithOzError( self, errorDoc, resourceId )
		
	end
	
end

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

local function isStorageExceededErrorForRestore( subtype, action, revIdThatIncurredError, docThatIncurredError )

	-- Restored assets may have a modification applied to them so check if the
	-- createWithRevId matches and allow the create to be processed as an error
	-- as well. This resolves issues with Stuck syncs when restoring modified
	-- assets for kStorageExceededError. NMB-10393

	-- Also we may get a storage exceeded error when trying to delete an asset
	-- that was restored from the trash while out of storage and then put back
	-- into the trash as we are required to send a restore/delete combo. NMB-11108

	return WFCatOzErrors.Errors.kStorageExceededError and action == 'create' and
		( WFCatTrashUtils.getCreateWithRevisionId( docThatIncurredError ) == revIdThatIncurredError -- restore modify
		or docThatIncurredError._deleted and docThatIncurredError._newId ) -- restore delete

end

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

local function processOneErrorDocument( self, errorDocId )

	if self._docsBeingProcessed[ errorDocId ] then

		-- Query can return a new result while error doc is being processed
		-- avoid processing the same error docs multiple times.

		return -- Already being processed ignore this doc.

	end

	self._docsBeingProcessed[ errorDocId ] = true

	local docStore = self._catalog:docStore()
	
	return self._catalog:docStore():docValue( errorDocId )
		:flatMap( function( errorDoc )

			if errorDoc._empty then
				return Observable.just()
			end
			
			-- all error documents should have the id of the document that caused the error
			-- make sure we have that document before sending it to a handler since clients
			-- can receive error documents generated by other ecosystem clients
			local spm_source =  derefOpt( errorDoc, '_serverProvidedMetadata', 'source' )
			local idOfDocThatIncurredError = derefOpt( spm_source, 'asset_id' ) or derefOpt( spm_source, 'id' )
			local revIdThatIncurredError = derefOpt( spm_source, 'revision_id' )
			local action = derefOpt( spm_source, 'action' )

			if not idOfDocThatIncurredError then
				-- this error doesn't match the form we expect, mark it unknown for now
				logger:errorf( "Unexpected format in error document, missing source document id : %s", WFTableArchiving.tableToString( errorDoc ) )
				return markErrorAsUnknown( docStore, errorDocId )
			end
			
			return docStore:docValue( idOfDocThatIncurredError )
				:flatMap( function( docThatIncurredError )

					-- don't let any unexpected error formats that fail to get to the diagnostic loc cause any issues
					-- with processing more error documents
					pcall( addErrorDocToDiagnosticLog, self, errorDoc )

					local subtype = errorDoc.subtype
					
					-- Check that the document is not empty/deleted and that the revision matches the latest revision.
					--
					-- This is intended to filter out the error documents that may have been generated by changes from
					-- other clients. However, it also means that it filters out any error documents generated from this
					-- client, but at an earlier revision than what is current.
					-- 
					-- In these cases we expect that the later revision will be sent to Oz and, if that again results
					-- in an error, the error will be handled with an error document pertaining to that revision.
					if not maybeTrashOperationNewId( errorDoc ) and
						not isRestoreAssetError( errorDoc ) and -- NMB-13281/LRMI-4193007/LRMA-4917671
						not isStorageExceededErrorForRestore( subtype, action, revIdThatIncurredError, docThatIncurredError ) and
						not isDeletedByAnotherUser( errorDoc, self._catalog:ozUserId() ) and
						( docThatIncurredError._revId ~= revIdThatIncurredError or
						( docThatIncurredError._empty and 
							-- This is an edge case in NMB-9050, but if we sent a delete to Oz and Oz responds with a
							-- ResourceNotFoundError on the delete we'll never get a tombstone, but we should consider 
							-- it deleted on Oz and so we need to let the ResourceNotFoundError handler handle it.
							not ( ( subtype == WFCatOzErrors.Errors.kResourceNotFoundError ) and 
									( action == 'delete' ) ) ) )  then
						return resolveErrorByIgnoring( self, errorDoc, "RevisionDoesNotMatch" )
					end

					local handler = getErrorHandler( self, subtype ) or defaultHandler
				
					return handler( self, errorDoc, docThatIncurredError )
			
			end )
				
		end )
		:finally( function()
			self._docsBeingProcessed[ errorDocId ] = nil
		end )
		:call( WFRxRunUntilDone )

end

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

local function clearTrialOrSubscriptionExpiredErrors( self )

	self._clearTrialOrSubscriptionExpiredErrorsSubscription = Observable.combineLatest(
			self._blockedDocsStream, self._catalog:ozUser():isSubscriptionExpired() )
	
		:filter( function( blockedDocs, isExpired ) 
			return not isExpired
		end )
	
		:map( function( blockedDocs )
		
			local bulkOverlays = {}
			
			-- we should only be here if we have ids blocked because of subscription expiry and 
			-- we are not in an expired state so we can go ahead and clear all of these errors
				
			for _, blockedDoc in ipairs( blockedDocs ) do
				
				if blockedDoc.errorSubtype == WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError then
					bulkOverlays[ #bulkOverlays + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( blockedDoc.docId )
				end
				
			end
			
			return bulkOverlays
		
		end )
		
		:flatMap( function( bulkOverlays )
		
			logger:info( [[Cleared %d errors on documents previously blocked from upload due to
							an expired trial or subscription.]], #bulkOverlays )
							
			if #bulkOverlays > 0 then
			
				return self._catalog:docStore():bulkApplyOverlay( bulkOverlays )
		
			else
			
				return Observable.just( true )
				
			end
		
		end )
		
		:subscribe()

end

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

local function checkAlbumAssets( self, blockedAlbumAssetIds )

	if #blockedAlbumAssetIds == 0 then
		return Observable.just( {} )
	end

	-- Check Oz status of associated assets and albums.

	local blockedDocValueStream = 
			WFRxThrottledBatchedBulkDocValue( "WFCatErrorHandler_checkAlbumAssets", self._catalog:docStore(), blockedAlbumAssetIds, 100, 10 )
			
				:scan( function( docValues, batchOfDocValues )
				
					docValues = KSTableUtils.appendArray( docValues, batchOfDocValues )
					return docValues
					
				end, {} )

				:takeLast( 1 )	

	return Observable.combineLatest( 
			blockedDocValueStream,
			self._getSyncStatusOfAlbumsOfAlbumAssetsFn( self._catalog, blockedAlbumAssetIds ),
			self._getSyncStatusOfAssetsOfAlbumAssetsFn( self._catalog, blockedAlbumAssetIds ) )
	
		:map( function( blockedAlbumAssetDocValues, syncedAlbumsById, syncedAssetsById )
		
			local albumAssetIdsToClear = {}
		
			for _, docValue in ipairs( blockedAlbumAssetDocValues ) do
			
				local albumId = derefOpt( docValue, 'album', 'id' )
				local assetId = derefOpt( docValue, 'asset', 'id' )
		
				if albumId and syncedAlbumsById[ albumId ] and
					assetId and syncedAssetsById[ assetId ] then
				
					logger:infof( "The associated album %s and asset %s are now synced for album_asset %s.",
							albumId, assetId, docValue._id )
					
					albumAssetIdsToClear[ #albumAssetIdsToClear + 1 ] = docValue._id
				
				else
				
					logger:infof( "The album or asset for album_asset %s is not yet synced.", docValue._id )
				
				end
			
			end
			
			return albumAssetIdsToClear
			
		end )

end

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

local function checkAlbums( self, blockedAlbumIds )

	if #blockedAlbumIds == 0 then
		return Observable.just( {} )
	end

	-- Check Oz status of associated collection_set parent

	local blockedDocValueStream = 
			WFRxThrottledBatchedBulkDocValue( "WFCatErrorHandler_checkAlbums", self._catalog:docStore(), blockedAlbumIds, 100, 10 )
			
				:scan( function( docValues, batchOfDocValues )
				
					docValues = KSTableUtils.appendArray( docValues, batchOfDocValues )
					return docValues
					
				end, {} )

				:takeLast( 1 )		

	return Observable.combineLatest( 
			blockedDocValueStream,
			self._getSyncStatusOfParentCollectionOfAlbumsFn( self._catalog, blockedAlbumIds ) )
	
		:map( function( blockedAlbumDocValues, syncedAlbumsById )
		
			local albumIdsToClear = {}
		
			for _, docValue in ipairs( blockedAlbumDocValues ) do
			
				local parentId = derefOpt( docValue, 'parent', 'id' )
		
				if parentId and parentId ~= 'root' then
				
					if syncedAlbumsById[ parentId ] then
					
						logger:infof( "The parent collection_set for album %s is now synced.", docValue._id )
						
						albumIdsToClear[ #albumIdsToClear + 1 ] = docValue._id
					
					else
					
						logger:infof( "The parent collection_set for album %s is not yet synced.", docValue._id )
					
					end
				
				else
				
					logger:warnf( "This album is blocked from upload, and we don't know why." )
				
				end
			
			end
			
			return albumIdsToClear
			
		end )
	
end

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

local function checkAssetDependentDocs( self, blockedDocIds, forceUseFullBlockedDocIds, getSyncStatusOfAssetsOfDependentDocFn )

	if #blockedDocIds == 0 then
		return Observable.just( {} )
	end

	-- Check the Oz status of associated assets.

	local blockedDocValueStream = 
			WFRxThrottledBatchedBulkDocValue( "WFCatErrorHandler_checkAssetDependentDocs", self._catalog:docStore(), blockedDocIds, 100, 10 )
			
				:scan( function( docValues, blockedFullDocIds, batchOfDocValues )
				
					for _, docValue in ipairs( batchOfDocValues ) do
					
						docValues[ #docValues + 1 ] = docValue
						blockedFullDocIds[ #blockedFullDocIds + 1 ] = docValue._id
						
					end

					return docValues, blockedFullDocIds
					
				end, {}, {} )

				:takeLast( 1 )	

	return blockedDocValueStream
	
		:flatMap( function( blockedDependentDocValues, blockedFullDocIds )
		
			local blockedIds = blockedDocIds
		
			if forceUseFullBlockedDocIds then
				blockedIds = blockedFullDocIds
			end
		
			return getSyncStatusOfAssetsOfDependentDocFn( self._catalog, blockedIds )
				
				:map( function( ... )
					return blockedDependentDocValues, ...
				end )
		
		end )
	
		:map( function( blockedDependentDocValues, syncedAssetsById )
		
			local dependentDocIdsToClear = {}
		
			for _, docValue in ipairs( blockedDependentDocValues ) do
			
				local assetId = derefOpt( docValue, 'asset', 'id' )
		
				if assetId and syncedAssetsById[ assetId ] then
				
					logger:infof( "The associated asset %s is now synced for dependent doc %s.",
							assetId, docValue._id )
					
					dependentDocIdsToClear[ #dependentDocIdsToClear + 1 ] = docValue._id
				
				else
				
					logger:infof( "The asset for dependent doc %s is not yet synced.", docValue._id )
				
				end
			
			end
			
			return dependentDocIdsToClear
			
		end )

end

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

local function checkVersions( self, blockedVersionIds )

	return checkAssetDependentDocs( self, blockedVersionIds, false, self._getSyncStatusOfAssetsOfVersionsFn )

end

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

local function checkFavorites( self, blockedFavoriteIds )

	return checkAssetDependentDocs( self, blockedFavoriteIds, true, self._getSyncStatusOfAssetsOfFavoritesFn )

end

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

local function checkComments( self, blockedCommentIds )

	return checkAssetDependentDocs( self, blockedCommentIds, true, self._getSyncStatusOfAssetsOfCommentsFn )

end

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

local function clearResourceNotFoundErrors( self )

	self._clearResourceNotFoundErrorsSubscription = self._blockedDocsStream
	
		:flatMapLatest( function( blockedDocs )
		
			local blockedAlbumAssets = {}
			local blockedAlbums = {}
			local blockedVersions = {}
			local blockedFavorites = {}
			local blockedComments = {}
			
			for _, blockedDoc in ipairs( blockedDocs ) do
					
				if blockedDoc.errorSubtype == WFCatOzErrors.Errors.kResourceNotFoundError then
				
					if blockedDoc.type == 'album_asset' then
						blockedAlbumAssets[ #blockedAlbumAssets + 1 ] = blockedDoc.docId
					elseif blockedDoc.type == 'album' then
						blockedAlbums[ #blockedAlbums + 1 ] = blockedDoc.docId
					elseif blockedDoc.type == 'version' then
						blockedVersions[ #blockedVersions + 1 ] = blockedDoc.docId
					elseif blockedDoc.type == 'favorite' then
						blockedFavorites[ #blockedFavorites + 1 ] = blockedDoc.docId
					elseif blockedDoc.type == 'comment' then
						blockedComments[ #blockedComments + 1 ] = blockedDoc.docId
					end
					
				end
				
			end
		
			return Observable.combineLatest(
					checkAlbumAssets( self, blockedAlbumAssets ),
					checkAlbums( self, blockedAlbums ),
					checkVersions( self, blockedVersions ),
					checkFavorites( self, blockedFavorites ),
					checkComments( self, blockedComments ) )
		
		end )
		
		:map( function( albumAssetIdsToClear, albumIdsToClear, versionIdsToClear, favoriteIdsToClear, commentIdsToClear )
		
			local bulkOverlays = {}
			
			for _, docId in ipairs( albumAssetIdsToClear ) do
				
				bulkOverlays[ #bulkOverlays + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( docId )
				
			end
			
			for _, docId in ipairs( albumIdsToClear ) do
				
				bulkOverlays[ #bulkOverlays + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( docId )
				
			end
			
			for _, docId in ipairs( versionIdsToClear ) do
				
				bulkOverlays[ #bulkOverlays + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( docId )
				
			end
			
			for _, docId in ipairs( favoriteIdsToClear ) do
				
				bulkOverlays[ #bulkOverlays + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( docId )
				
			end
			
			for _, docId in ipairs( commentIdsToClear ) do
				
				bulkOverlays[ #bulkOverlays + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( docId )
				
			end
			
			return bulkOverlays
		
		end )
		
		:flatMapLatest( function( bulkOverlays )
		
			logger:info( [[Cleared %d errors on documents previously blocked from upload due to
							resource not found errors.]], #bulkOverlays )
		
			if #bulkOverlays > 0 then
			
				return WFRxThrottledBatchedBulkApplyOverlay( "WFCatErrorHandler_clearResourceNotFoundErrors", self._catalog:docStore(), bulkOverlays, 100, 10 )
				
			else
			
				return Observable.just( true )
				
			end
			
		end )
		
		:subscribe()

end

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

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

	return KSTableUtils.deepEquals( tbl1, tbl2 )

end

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

local function monitorBlockedUploads( self )
	
	-- This function only monitors blocked uploads that belong to the catalog.
	-- Space-based blocked uploads are handled WFCatSpaceFeedsManager
	
	local docIdRowsQuery = self._indexStore:query {
			sql = [[ SELECT docId, reason, type 
						FROM blocked_uploads
						WHERE type IN ( "asset", "album_asset", "album", "collection", 
										"version", "comment", "favorite" )
							AND spaceId IS NULL ]],
			calm = self._debugMode and nil or 5,
			enqueue = self._debugMode and 'fast' or 'slow',
		}
	
	local startupDelay = self._debugMode and 0.01 or 10
	
	return Observable.combineLatest( docIdRowsQuery, Observable.timer( startupDelay, WFRxScheduler.sharedScheduler() ) )
							
		:filter( function( docIdRows ) 
			return #docIdRows > 0
		end )
		
		:distinctUntilChanged( nil, distinctTableCompare )
		
		:map( function( docIdRows )
		
			local blockedDocs = {}
			
			for _, docIdRow in ipairs( docIdRows ) do
				
				local blockedDoc = {}
				
				blockedDoc.docId = docIdRow[ 1 ]
				blockedDoc.errorSubtype = docIdRow[ 2 ]
				blockedDoc.type = docIdRow[ 3 ]
				
				blockedDocs[ #blockedDocs + 1 ] = blockedDoc
				
			end
			
			return blockedDocs
		
		end )
		
		:subscribe( function( blockedDocs )
		
			self._blockedDocsStream:onNext( blockedDocs )
			
		end )
		
end

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

local function wrapFnForClass( fn )
	return function( _, ... )
		return fn( ... )
	end
end

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

-- @mark WFCatErrorHandler namespace

local WFCatErrorHandler = {

	_catalog = false,
		-- (WFCatalog)
		
	_indexStore = false,
		-- (WFRxIndexStore)
	
	_errorHandlerSubscription = false,
		-- (Subscription) Subscription to the Observable that will watch for process error documents
	
	_errorDocumentHandlerFns = false,
		-- (table) table mapping error document subtypes to error document handler functions
		
	_externalizeSettingsFn = false,
		-- (function) function that takes a catalog, document id, and optional enforced limit and returns 
		-- an Observable that will externalize the develop settings and return true when complete
	
	_deleteAssetsFn = false,
		-- (function) function that takes a catalog and list of document ids, and returns 
		-- an Observable that will delete the assets when complete

	_deleteIndividualAlbumFn = false,
		-- (function) function that takes a catalog and document id, and returns
		-- an Observable that will delete the individual album when complete

	_getSyncStatusOfAssetsOfAlbumAssetsFn = false,
		-- (function) function that takes a catalog and a list of album asset document ids, and returns
		-- an Observable which yields an array of isComplete status of said assets

	_getSyncStatusOfAlbumsOfAlbumAssetsFn = false,
		-- (function) function that takes a catalog and a list of album asset document ids, and returns
		-- an Observable which yields an array of isComplete status of the associated albums

	_getSyncStatusOfParentCollectionOfAlbumsFn = false,
		-- (function) function that takes a catalog and a list of albums document ids, and returns
		-- an Observable which yields an array of isComplete status of the albums

	_getSyncStatusOfAssetsOfVersionsFn = false,
		-- (function) function that takes a catalog and a list of version document ids, and returns
		-- an Observable which yields an array of isComplete status of the assets
		
	_getSyncStatusOfAssetsOfCommentsFn = false,
		-- (function) function that takes a catalog and a list of comment document ids, and returns
		-- an Observable which yields an array of isComplete status of the assets
		
	_getSyncStatusOfAssetsOfFavoritesFn = false,
		-- (function) function that takes a catalog and a list of favorite document ids, and returns
		-- an Observable which yields an array of isComplete status of the assets

	_errorDocSummary = false,
		-- (table) summary of the error documents received by the client

	_errorDocResolution = false,
		-- (table) table describing resolutions used to handle the encountered errors

	_blockedUploadSubscription = false,
		-- (Subscription) watches uploads blocked by Oz
	
	_blockedDocsStream = false,
		-- (Rx.ReplaySubject) Fires with documents blocked from upload.
		
	_clearResourceNotFoundErrorsSubscription = false,
		-- (Subscription) Clears uploads blocked because of resource not found errors.
		
	_clearTrialOrSubscriptionExpiredErrorsSubscription = false,
		-- (Subscription) Clears uploads blocked because of a expired trial/subscription.

	_docsBeingProcessed = false,
		-- table indexed by errorDocId to indicate the document is currently being processed.
	
	_debugMode = false,
		-- (boolean) If set, reduces many of the built in delays/calm values. Useful for
		-- speeding up unit tests.
		
	class = {
		generateIndexSchema = wrapFnForClass( generateIndexSchema ),
	},
	
	WFCloseSupport.addCloseSupportMethods,

}

local function maybeLogMissingHandlerError( debugMode, maybeHandlerFn, handlerName )
	if not debugMode and not maybeHandlerFn then
		logger:errorf( "WFCatErrorHandler missing handler function for %s", handlerName )
	end
end

--------------------------------------------------------------------------------
--[[
	\class WFCatErrorHandler
	
	\brief Indexes and process documents of type \c error sent back from Oz in the metadata sync feed.
	
	Oz may send documents of type \c error and various subtypes in the updates feed that indicate errors in 
	either metadata sync or binary sync from the client. These documents will be indexed to the \c errors table
	and then processed by this class which will clear the documents from the \c errors table.
	
	Error documents generally need to explicitly register a new handler to properly resolve the error, however 
	there are some error documents (e.g. NotHeadRevisionPreconditionError) that do not need to be handled as they 
	will be resolved naturally as part of sync.
	
	Any error documents received without an explicit handler will be marked with an 'unknownError' flag which will
	remove it from the \c errors table. If a handler is added later, it can easily be reindexed back in to the \c
	errors table to be processed with the new handler.
	
	IMPORTANT: Each client receives all error documents generated by *ANY* ecosystem client so error documents are 
	immediately filtered out by revision id which is generally specific to a single client (this is true for metadata
	sync errors but not for error docs related to binary sync). If the revision id does not exist in this client's
	doc store then the document is immediately deleted.
	
	
	To add a handler for a new error document type:
	
	1. Add the error document subtype to the \c Errors array in \c WFCatOzErrors
	2. Create a handler function in \c WFCatErrorHandler to address the error and returns an Observable
		a. The handler function will be passed \c self (WFCatErrorHandler), the error document, and the document that
			caused the error to be sent.
		b. The handler function should delete the error doc if it handles the error successfully or mark it as an unknown 
			error if the type/subtype is correct but it does not know how to handle that specific error. Marking it unknown 
			allows a future client the ability to retry the processing of the error after handling for the specific case has 
			been handled.
	3. Register the error handler by calling \c self:registerErrorHandler in the \c WFCatErrorHandler:__init function
	4. Update the implementation of the handler function to properly resolve the error. Handler functions will commonly
		a. Request Oz's version of the document
		b. Apply an overlay to alter the document that caused the error (e.g. truncating an album name)
		c. Look up associated documents (e.g. assets for an album_asset error) either locally or on Oz to determine how
			to handle the error
	5. Some error handlers may need client specific handling. For these cases we usually require clients to pass in client-
		specific handling functions to the \c WFCatErrorHandler constructor that encapsulate the client-specific code. See the
		\c optClientHandlerFnTbl usage in \c __init
	6. Add tests in WFCatErrorHandlerTest to validate that the error handler works correctly
	
]]--
function WFCatErrorHandler:__init( catalog, clientHandlerFnTbl, optArgs )
	
	check( '_', 'WFCatalog', 'table', '?table' )
	
	self._errorDocumentHandlerFns = {}
	self._errorDocSummary = {}
	self._errorDocResolution = {}
	self._docsBeingProcessed = {}
	
	self:setOwned {
		_catalog = catalog:relayObject()
	}
	
	self._indexStore = catalog:indexStore()
	
	optArgs = optArgs or {}
	
	self._debugMode = optArgs.debugMode
	
	-- set up error document handlers

	self:registerErrorHandler( WFCatOzErrors.Errors.kResourceNotFoundError, resourceNotFoundHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kStorageExceededError, storageExceededHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError, trialOrSubscriptionEntitlementExpiredHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kEntitlementExpiredError, trialOrSubscriptionEntitlementExpiredHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kPayloadValidationError, payloadValidationHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kInputValidationError, inputValidationHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kResourceExistsError, resourceExistsHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kMultipartCompleteError, multipartCompleteHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kAssetGoneError, assetGoneHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kInvalidAccessRequestError, invalidAccessRequestHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kSpaceUserExistsError, spaceUserExistsHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kRestoreAssetError, restoreAssetErrorHandler )
	self:registerErrorHandler( WFCatOzErrors.Errors.kInsufficientPermissionError, insufficientPermissionErrorHandler )
	
	-- Some handlers need client specific implementations which can be passed in the fn
	-- table. If the table exists and the fns are provided, add them to the error handler

	self._externalizeSettingsFn = clientHandlerFnTbl.externalizeSettings
	maybeLogMissingHandlerError( self._debugMode, self._externalizeSettingsFn, 'externalizeSettings' )
	
	self._deleteAssetsFn = clientHandlerFnTbl.deleteAssets
	maybeLogMissingHandlerError( self._debugMode, self._deleteAssetsFn, 'deleteAssets' )
	
	self._deleteIndividualAlbumFn = clientHandlerFnTbl.deleteIndividualAlbum
	maybeLogMissingHandlerError( self._debugMode, self._deleteIndividualAlbumFn, 'deleteIndividualAlbum' )
	
	self._getSyncStatusOfAssetsOfAlbumAssetsFn = clientHandlerFnTbl.getSyncStatusOfAssetsOfAlbumAssets
	maybeLogMissingHandlerError( self._debugMode, self._getSyncStatusOfAssetsOfAlbumAssetsFn, 'getSyncStatusOfAssetsOfAlbumAssets' )
	
	self._getSyncStatusOfAlbumsOfAlbumAssetsFn = clientHandlerFnTbl.getSyncStatusOfAlbumsOfAlbumAssets
	maybeLogMissingHandlerError( self._debugMode, self._getSyncStatusOfAlbumsOfAlbumAssetsFn, 'getSyncStatusOfAlbumsOfAlbumAssets' )
	
	self._getSyncStatusOfParentCollectionOfAlbumsFn = clientHandlerFnTbl.getSyncStatusOfParentCollectionOfAlbums
	maybeLogMissingHandlerError( self._debugMode, self._getSyncStatusOfParentCollectionOfAlbumsFn, 'getSyncStatusOfParentCollectionOfAlbums' )
	
	self._getSyncStatusOfAssetsOfVersionsFn = clientHandlerFnTbl.getSyncStatusOfAssetsOfVersions
	maybeLogMissingHandlerError( self._debugMode, self._getSyncStatusOfAssetsOfVersionsFn, 'getSyncStatusOfAssetsOfVersions' )
	
	self._getSyncStatusOfAssetsOfCommentsFn = clientHandlerFnTbl.getSyncStatusOfAssetsOfComments
	maybeLogMissingHandlerError( self._debugMode, self._getSyncStatusOfAssetsOfCommentsFn, 'getSyncStatusOfAssetsOfComments' )
	
	self._getSyncStatusOfAssetsOfFavoritesFn = clientHandlerFnTbl.getSyncStatusOfAssetsOfFavorites
	maybeLogMissingHandlerError( self._debugMode, self._getSyncStatusOfAssetsOfFavoritesFn, 'getSyncStatusOfAssetsOfFavorites' )
	
	-- IMPORTANT: If you're adding additional, client-specific handlers, make sure those handlers
	-- are passed into the WFCatErrorHandler constructors within NmbErrorHandler & LrmSyncClientImp.
	
	if self._externalizeSettingsFn then
		self:registerErrorHandler( WFCatOzErrors.Errors.kInlineXmpDevelopSettingTooBigError, inlineSettingsHandler )
	end
	
	--------------------------------------------------------------------------------

	local errorQuery = self._indexStore:query {
		sql = [[SELECT docId FROM errors ORDER BY docId LIMIT 10]], 
		calm = self._debugMode and 0.1 or 1,
		enqueue = 'slow',
		singleColumnQuery = true
	}
	
	self._errorHandlerSubscription = errorQuery:subscribe(
			function( docIds )
				if #docIds > 0 then
					logger:infof( "Error document handler identified %d error documents", #docIds )
				end
				for _, docId in ipairs( docIds ) do
					processOneErrorDocument( self, docId )
				end
			end,
			function( message )
				logger:errorf( "Error document handler query failed with message: %s", message )
			end,
			function()
				logger:debugf( "Error document handler query notified as done" )
			end )
	
	self._blockedUploadSubscription = monitorBlockedUploads( self )
	
	self._blockedDocsStream = Rx.ReplaySubject( 1 )
	
	clearTrialOrSubscriptionExpiredErrors( self )
	
	clearResourceNotFoundErrors( self )
	
	WFDiagnosticLog.addLogProvider( self )

end

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

function WFCatErrorHandler:registerErrorHandler( errSubtype, handlerFn )
	
	self._errorDocumentHandlerFns[ errSubtype ] = handlerFn
	
end

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

function WFCatErrorHandler:getDefaultErrorHandlerFn()
	return defaultHandler
end

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

function WFCatErrorHandler:sendDiagnosticLogData()
	
	WFDiagnosticLog.addSectionToLog
		{

			id = string.format( 'WFCatErrorHandler' ),

			name = string.format( "Sync Error Handler" ),
			weight = 510,
			parentSection = WFDiagnosticLog.Sections.SYNC,
			
			body = {
				errors = self._errorDocSummary,
				resolutions = self._errorDocResolution,
			}

		}
			
end

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

function WFCatErrorHandler:DEBUG_errorsProcessed()

	-- Returns error summary table, and boolean that is true if currently processing an error

	return self._errorDocSummary, next( self._docsBeingProcessed ) ~= nil

end

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

function WFCatErrorHandler:DEBUG_errorsResolved()

	-- Returns error summary table, and boolean that is true if currently processing an error

	return self._errorDocResolution

end

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

function WFCatErrorHandler:close()
	
	self._errorHandlerSubscription:dispose()
	self._blockedUploadSubscription:dispose()
	
	local clearResourceNotFoundErrorsSubscription = self._clearResourceNotFoundErrorsSubscription
	self._clearResourceNotFoundErrorsSubscription = nil
	
	if clearResourceNotFoundErrorsSubscription then
		clearResourceNotFoundErrorsSubscription:dispose()
	end
	
	local clearTrialOrSubscriptionExpiredErrorsSubscription = self._clearTrialOrSubscriptionExpiredErrorsSubscription
	self._clearTrialOrSubscriptionExpiredErrorsSubscription = nil
	
	if clearTrialOrSubscriptionExpiredErrorsSubscription then
		clearTrialOrSubscriptionExpiredErrorsSubscription:dispose()
	end
	
	local blockedDocsStream = self._blockedDocsStream
	self._blockedDocsStream = nil
	
	if blockedDocsStream then
		blockedDocsStream:dispose()
	end
	
end

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

return Rx.Class 'WFCatErrorHandler'( WFCatErrorHandler )
