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

WFCatBinaryUploader.lua

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

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

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

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

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

	-- WF.oz_client.managed_catalog
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFCatAssetAux = require 'WFCatAssetAux'
local WFCatCollaboration = require 'WFCatCollaboration'
local WFCatContentRangeUploader = require 'WFCatContentRangeUploader'
local WFCatDeniedListHelper = require 'WFCatDeniedListHelper'
local WFCatDirectChunkedUploader = require 'WFCatDirectChunkedUploader'
local WFCatErrorHandler = require 'WFCatErrorHandler'
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatImportDuplicateFiles = require 'WFCatImportDuplicateFiles'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatPhotoImportance = require 'WFCatPhotoImportance'
local WFCatSharedWithYouUtils = require 'WFCatSharedWithYouUtils'
local WFCatTrashUtils = require 'WFCatTrashUtils'
local WFOzContentTypes = require 'WFOzContentTypes'
local WFOzDocUtils = require 'WFOzDocUtils'
local WFTableArchiving = require 'WFTableArchiving'

-- Lazy load to avoid cycles
local WFCatStuckSyncUtils

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

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

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

	-- WF.core.performance
local WFPerformanceReport = require 'WFPerformanceReport'

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

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSJsonDecoder = require 'KSJsonDecoder'
local KSJsonEncoder = require 'KSJsonEncoder'
local KSTableUtils = require 'KSTableUtils'
local WFAnalytics = require 'WFAnalytics'
local WFArguments = require 'WFArguments'
local WFCapabilitiesCache = require "WFCapabilitiesCache"
local WFCrypto = require 'WFCrypto'
local WFCloseSupport = require 'WFCloseSupport'
local WFDeviceInfo = require 'WFDeviceInfo'
local WFMsgPack = require 'WFMsgPack'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxLowFidelityTimer = require 'WFRxLowFidelityTimer'
local WFSQLiteUtils = require 'WFSQLiteUtils'

	-- WF.web.http_client
local WFRxHttpClient = require 'WFRxHttpClient'

	-- WF.web.utils
local WFURI = require 'WFURI'

	-- WF.oz_client.shared.utils
local LrAssetUtils = require 'LrAssetUtils'

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

local check = assert( WFArguments.check )
local derefOpt = assert( KSTableUtils.derefOpt )
local setNested = assert( KSTableUtils.setNested )
local doesAssetDocumentHaveLocalBinary = assert( WFCatAssetUtils.doesAssetDocumentHaveLocalBinary )
local getLocalExternalDevelopSettingsPath = assert( WFCatAssetUtils.getLocalExternalDevelopSettingsPath )
local getLocalWatermarkBinaryPath = assert( WFCatAssetUtils.getLocalWatermarkBinaryRelativePath )
local doesWatermarkNeedBinaryUploaded = assert( WFCatAssetUtils.doesWatermarkNeedBinaryUploaded )
local WFCatAssetUtils_getSourceXmpDevelopLinkAndSha256FromDocValue = assert( WFCatAssetUtils.getSourceXmpDevelopLinkAndSha256FromDocValue )
local isValidAssetDocument = assert( WFCatIndexUtils.isValidAssetDocument )
local isNonDeletedStyleDocument = assert( WFCatIndexUtils.isNonDeletedStyleDocument )

local isLatestMetadataUploaded = assert( LrAssetUtils.isLatestMetadataUploaded )

local WFCatTrashUtils_getMasterIsUploaded = assert( WFCatTrashUtils.getMasterIsUploaded )
local WFCatTrashUtils_getProxyIsUploaded = assert( WFCatTrashUtils.getProxyIsUploaded )
local WFCatTrashUtils_getXmpDevelopIsUploaded = assert( WFCatTrashUtils.getXmpDevelopIsUploaded )
local WFCatTrashUtils_getAuxIsUploaded = assert( WFCatTrashUtils.getAuxIsUploaded )
local WFCatAssetUtils_getLocalAuxObjectPath = assert( WFCatAssetUtils.getLocalAuxObjectPath )

local WFCatAssetAux_getSourceAssetLinkForAuxFromDocValue = assert( WFCatAssetAux.getSourceAssetLinkForAuxFromDocValue )
local WFCatAssetAux_getCopyUrl = assert( WFCatAssetAux.getCopyUrl )
local WFCatAssetAux_getAuxPartsRequestUrlTemplate = assert( WFCatAssetAux.getAuxPartsRequestUrlTemplate )
local WFCatAssetAux_isSourceAuxUploadedToOz = assert( WFCatAssetAux.isSourceAuxUploadedToOz )

local runUntilDone = WFRxRunUntilDone

local logger = WFLogger 'WFCatBinaryUploader'

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

local kUploadBlackoutWindowInSeconds = 60  -- Number of seconds we disable the re-uploading
local kUploadBlackoutWindowCount = 6 -- Blackout Window Count, scalar for multiplying kUploadBlackoutWindowInSeconds

local numberOfTriesBeforeDenyingUpload = 3
	-- Number of attempts to upload an original before it is denied.

local failedUploadBlackoutWindowInSeconds = 300
	-- Number of seconds an original remains denied after it has failed to upload n times.

local numberOfTriesBeforeRevertToLegacyUpload = 3
	-- Number of attempts to use direct upload before fall back to use legacy upload.

local binaryUploadQueryCalmValue = WFPlatformInfo.isUnitTest() and 0.01 or 3
local kClearFromRecentlyUploadedSeconds = 30
local kWaitForOtherUploadSeconds = 10
local kWaitForCopyFailedSeconds = 10
local kContentTypeXML = 'application/rdf+xml'
local kContentTypeDng = 'image/dng'
local kContentTypeOctetStream = 'application/octet-stream'
local kContentTypeC2pa = 'application/zip'
local kOzMaximumXmpDevelopSize = 512 * 1024 * 1024
	-- See http://apidocs.adbecrsl.com/v2/docs/multi_page/#method_createexternalxmpdevelopsettingfile, parameters section
	-- [jandrew 2021-10-26]: Oz increased the limit from 200mb to 512mb in October 2021.
local kDirectUploadFallBackToOzUploadWindowInSecond = 30 * 60
local uploadHttpRequestCost = WFRxHttpClient.requestCostUploadNegative

local kUploadMethodOz = 'oz_upload'
local kUploadMethodOzCopy   = 'oz_copy'
local kUploadMethodDirect = 'direct_upload'
local kUploadMethodDirectFallback = 'oz_upload_direct_upload_fallback'

local kS3Uploads = 'S3 Uploads'
local kOzCopies = 'Oz Copies'

local kUploadFailedMissingOriginal = "UploadFailedMissingOriginal"
local kUploadFailedMissingProxy = "UploadFailedMissingProxy"

-- Values for isReadyToUpload in needsBinaryUpload table. Some day when atomic develop settings
-- is on for all clients and the feature flag is removed, if a reindex is required for this
-- table then we could clean this up, but keeping all state in a single column to make need
-- for reindex clean up less urgent.

local kNotReadyToUpload = 0					-- Not ready to upload at all
local kReadyToUploadLegacy = 1				-- Ready to upload under pre Atomic Develop Settings rules
local kReadyToUploadAtomic = 2				-- Ready to upload under Atomic Develop Settings rules

local readyToUploadEnum = {
	kNotReadyToUpload = kNotReadyToUpload,
	kReadyToUploadLegacy = kReadyToUploadLegacy,
	kReadyToUploadAtomic = kReadyToUploadAtomic
}

-- Special value for mainPartsRequestUrl that indicates that even after refresh of links
-- from Oz, template link for direct upload is not available. This is in place because
-- watermarks do not support direct upload and not all subtypes of asset do either, so
-- want to be extra cautious that no bug is injected that causes an infinite loop of
-- trying to get a direct upload template link that will never come.

local kMainPartsRequestUrlNotAvailable = 'mainPartsRequestUrlNotAvailable'

-- Special value for mainPartsRequestUrl to cover an edge case that can happen when
-- metadata for asset or version synced with a client that had atomic develop settings
-- disabled and the xmp_develop binary did NOT sync from that client, then user upgrades
-- to a client that has ADS enabled. In these cases the ADS enabled client must detect
-- this scenario and upload the xmp_develop binary in the old way that predates ADS feature.

-- For a version document, this simply means that the document has been created on Oz,
-- since for versions xmp_develop binary should be uploaded before version is created.

-- For asset documents it means that metadata update with xmpCameraRaw sha256 has synced
-- to Oz and that asset already had some xmp_develop binary previously uploaded which
-- is now marked invalid. Only do this when asset is completely in sync with Oz to confirm
-- this truly is the case where the metadata has synced to Oz and definitely not a place
-- where ADS workflow should be applied. It is possible the invalid=true has been triggered
-- by another client, in which case should follow ADS workflow. Having all local metadata
-- synced to Oz, having an invalid=true for xmp_develop and having the xmp_develop with
-- the correct sha256 locally guarantees we are not in a valid ADS workflow and means it
-- is safe to do old style of binary upload.

-- Yes quite the edge case, but a QE managed to hit it and surely at least some customers
-- will. Without this special annotation, with ADS feature enabled, the uploader will keep
-- trying to do an ADS style upload which will never succeed. It will just create a temp
-- entry that will never be attached to the version or asset. See story WF-2429

local kMainPartsRequestUrlMetadataAlreadySynced = 'mainPartsRequestUrlMetadataAlreadySynced'

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

local function intToBool( zeroOrOne )
	return zeroOrOne ~= 0
end

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

local function boolToInt( isTrue )
	return isTrue and 1 or 0
end

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

local function bytesToMb( bytes )
	return bytes / 1024 / 1024
end

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

local binaryDetailMetaMethods = {

	__eq = function( bd1, bd2 )
		return bd1.type == bd2.type and bd1.sha256 == bd2.sha256
	end

}

--[[

	\brief Creates a basic container that is used internally to identify a specific upload.

	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values
	\param sha256		(string) - Sha256 of the uploaded binary


	\return string

]]
local function newBinaryDetail( binaryType, sha256 )

	local binaryDetail = {
		type = binaryType,
		sha256     = sha256
	}

	return setmetatable( binaryDetail, binaryDetailMetaMethods )

end

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

--[[

	\brief Returns correct binaryType for an original binary based on type and
		subtype of the document. The original binary of different documents need
		different handling in several circumstances, so this differentiation in
		the binaryType is useful.

	\param type (string) - The type of the document

	\param subtype (string) - The subtype of the document

	\return string. Possible values are master, preset_master, camera_profile and
		watermark_original

]]
local function getBinaryTypeForOriginal( type, subtype  )

	if type == 'watermark' then

		return 'watermark_original'

	elseif subtype == 'preset' or subtype == 'profile' then

		return 'preset_master'

	elseif subtype == 'camera_profile' then

		return 'camera_profile'

	end

	return 'master'

end

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

--[[

	\brief Checks if binary is a master

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isOriginalBinary( binaryType )

	return ( binaryType == 'master' or binaryType == 'preset_master' or
			binaryType == 'camera_profile' or binaryType == 'watermark_original' )

end

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

--[[

	\brief Checks if binary is an original binary for an image or a video, also
		includes deleted_image or deleted_video which are typically treated the
		same as image and video for upload purposes.

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isImageOrVideoOriginal( binaryType )

	return binaryType == 'master'

end

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

--[[

	\brief Checks if binary is an original binary for a preset

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isPresetOriginal( binaryType )

	return binaryType == 'preset_master'

end

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

--[[

	\brief Checks if binary is an original binary for a camera profile

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isCameraProfileOriginal( binaryType )

	return binaryType == 'camera_profile'

end

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

--[[

	\brief Checks if binary is an original binary for either a watermark.

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isWatermarkOriginal( binaryType )

	return binaryType == 'watermark_original'

end

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

--[[

	\brief Checks if binary is an XMP sidecar

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isXmpSidecar( binaryType )

	return ( binaryType == 'xmp_sidecar' )

end

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

--[[

	\brief Checks if binary is an XMP develop

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isXmpDevelop( binaryType )

	return ( binaryType == 'xmp_develop' )

end

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

--[[

	\brief Checks if binary is an aux

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isAux( binaryType )

	return ( binaryType == 'aux' )

end

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

--[[

	\brief Checks if binary is a develop related binary, i.e. either xmp_develop
		or aux

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isDevelopBinary( binaryType )

	return ( isXmpDevelop( binaryType) or isAux( binaryType ) )

end

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

--[[

	\brief Checks if binary is a proxy

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function isProxy( binaryType )

	return ( binaryType == 'proxy' or binaryType == 'proxy_pano' )

end

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

local function checkIfDenied( docValue, binaryDetails )

	local deniedListKey = 'WFCatBinaryUploader'

	local isDenied = WFCatDeniedListHelper.isDeniedForKey( docValue, deniedListKey )

	return isDenied and 1 or 0

end

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

local function getLocalOriginalErrorAnnotation( docValue )

	return derefOpt( docValue, '_localOnly', 'localOriginalError' )

end

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

local function getDatesFromDocValue( docValue )

	local importDate = derefOpt( docValue, 'importSource', 'importTimestamp' )

	local captureDate = derefOpt( docValue, 'captureDate' )

	return importDate or 1, captureDate or 1

end

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

--[[

	\brief Extract the master URL from a document if possible

	\param docValue (table) - A complete Document

	\return string  (URL) - The link to a master on oz

]]
local function getOriginalLinkFromDocValue_IgnoringShaError( docValue )

	-- The binary uploader uses the rels/master link only to indicate whether
	-- an asset has completed upload, so return the field's contents even if
	-- there was an error in the import source status field.

	return derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href' )

end

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

--[[

	\brief Determines if Original is on Oz

	\param docValue (table) - A complete Document

	\return bool


]]
local function getOriginalExistsFromDocValue( docValue )

	return derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'exists' )

end

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

--[[

	\brief Determine if there is an image proxy for this document

	\param docValue (table) - A complete Document

	\return bool

]]
local function getProxyExistsFromDocValue( docValue )

	return derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/proxy_type/2560', 'exists' ) or
		derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/proxy_type/panorama', 'exists' ) or
		derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/proxy_type/720p', 'exists' )

end

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

--[[

	\brief Determine if a document's original is invalid

	\param docValue (table) - A complete Document

	\return bool

]]
local function isOriginalLinkInvalid( docValue )

	return derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'invalid' )

end

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

--[[

	\brief Determine if a document is a copy originating from this device

	\param docValue (table) - A complete Document

	\return bool

]]
local function isAssetDocumentACopyWithSource( docValue )

	-- This check sourceAssetId to determine if the copy was made locally on
	-- this machine.  It's possible to create a new photo and then copy it
	-- while offline uploader will wait for the first one to upload and then
	-- create copies from that.

	return isValidAssetDocument( docValue ) and derefOpt( docValue, 'copy', 'copy_id' )
		and ( derefOpt( docValue, '_localOnly', 'sourceAssetId' ) )

end

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

--[[

	\brief Determine if a document is a space asset with links

	\param docValue (table) - A complete Document

	\return bool

]]
local function isSpaceAssetWithSourceLinks( docValue )

	local hasSourceOriginalLink = derefOpt( docValue, '_localOnly', 'sourceOriginalLink' )
	local hasSourceProxyLink = derefOpt( docValue, '_localOnly', 'sourceProxyLink' )

	return isValidAssetDocument( docValue ) and docValue._spaceId
		and ( hasSourceOriginalLink or hasSourceProxyLink )

end

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

--[[

	\brief Determine if this shared asset can be edited

	\param docValue (table) - A complete Document

	\return bool

]]
local function isSpaceAssetEditable( docValue )

	return isValidAssetDocument( docValue )
			and docValue._spaceId
			and WFCatSharedWithYouUtils.canEditAsset( docValue )

end

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

--[[

	\brief Determines what headers are required for a specific binaryType and document state

	\param binaryType (string) - The type of the binary, refer to kBinaryType... values
	\param docValue (table) - A complete Document

	\return string (MessagePacked Value) or nil

]]
local function getEncodedUploadHeadersForBinaryType( binaryType, docValue )

	local uploadHeaders = {}

	-- master/proxy upload X-Read-Meta headers
	if isImageOrVideoOriginal( binaryType ) or isProxy( binaryType ) then

		-- if we don't need to upload the sidecar, then we can ask Oz to extract it
		if not ( WFCatFileSpec.getXmpSidecarExistsFromDocValue( docValue )
			or WFCatAssetUtils.getSourceXmpLinkFromDocValue( docValue ) ) then
			uploadHeaders['X-Read-Meta'] = isImageOrVideoOriginal( binaryType ) and 'xmp,location' or 'xmp'
		else
			uploadHeaders['X-Read-Meta'] = 'none'
		end
	end

	-- In App Contribute Redactions headers
	if docValue._spaceId and ( isImageOrVideoOriginal( binaryType ) or isProxy( binaryType ) or isXmpSidecar( binaryType ) ) then
		-- for masters, proxies, and xmp sidecars in spaces we may need
		-- to send headers to instruct Oz to redact some of the associated
		-- metadata to protect the contributing users privacy
		local addRedaction = function( redactionsList, additionalRedaction )
			if #redactionsList > 0 then
				return redactionsList .. "," .. additionalRedaction
			else
				return additionalRedaction
			end
		end

		local redactions = derefOpt( docValue, '_localOnly', 'universal', 'sharedwithyou', 'contribution' ) or {}
		local redactionsList = ""

		if redactions.redactKeywords then
			redactionsList = addRedaction( redactionsList, 'keywords' )
		end

		if redactions.redactLocation then
			redactionsList = addRedaction( redactionsList, 'location' )
		end

		if redactions.redactRatingsReviews then
			redactionsList = addRedaction( redactionsList, 'ratings' )
			redactionsList = addRedaction( redactionsList, 'reviews' )
		end

		if #redactionsList > 0 then
			uploadHeaders['X-Redact-Meta'] = redactionsList
		end

	end

	return next( uploadHeaders ) and WFMsgPack.pack( uploadHeaders ) or nil
end

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

--[[

	\brief Determines if a document has an associated XMP Sidecar

	Note: If we have an XMP sidecar to upload for the asset we want to wait on uploading
	the other binaries so that we can ask Oz to extract metadata from the XMP sidecar
	instead of the master/proxy. In the event that the xmp sidecar upload or copy fails
	we'll stop trying to upload the xmp and proceed to the master/proxy but will then
	ask Oz to extract metadata from the master/proxy.


	\param docValue (table) -  A complete Document

	\return bool

]]
local function shouldWaitForXmpSidecar( docValue )
	-- If we have an XMP sidecar to upload for the asset we want to wait on uploading
	-- the other binaries so that we can ask Oz to extract metadata from the XMP sidecar
	-- instead of the master/proxy. In the event that the xmp sidecar upload or copy fails
	-- we'll stop trying to upload the xmp and proceed to the master/proxy but will then
	-- ask Oz to extract metadata from the master/proxy.
	local xmpSidecarExists = WFCatFileSpec.getXmpSidecarExistsFromDocValue( docValue )
	local xmpLocalPath = WFCatFileSpec.getXmpSidecarPathFromDocValue( docValue )

	local sourceXmpLinkForCopy = WFCatAssetUtils.getSourceXmpLinkFromDocValue( docValue )

	return ( ( xmpSidecarExists and xmpLocalPath ) or -- locally imported xmp sidecar
		( sourceXmpLinkForCopy ) ) -- copy
		and not ( derefOpt( docValue, '_serverProvidedMetadata',
											 'links',
											 '/rels/xmp',
											 'href' ) or
			derefOpt( docValue, '_serverProvidedMetadata',
											 'links',
											 '/rels/xmp',
											 'exists' ) )

end

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

--[[

	\brief Checks if a document has associated auxilliary binaries which need to be uploaded.

	\param docValue (table) - A complete Document

	\return bool

]]
local function shouldWaitForAux( docValue )

	if WFCatIndexUtils.isFileBasedAssetDocument( docValue ) then

		local onlyConsiderDevelopAndC2pa = true
		local auxUploadsRequired = WFCatAssetAux.uploadsRequired( docValue, onlyConsiderDevelopAndC2pa )

		return auxUploadsRequired

	else

		return false

	end

end

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

--[[

	\brief Checks if the document's current develop settings are on
	the current device OR in the case of an asset copy - checks
	that the source asset has a develop link

	\param docValue (table) - A complete Document

	\return string or bool

]]
local function doesAssetHaveXmpDevelopSettingsLocally( docValue )

	local ozXmpCameraRawSha256 = derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )

	local localXmpDevelopInfo = WFCatFileSpec.getXmpDevelopInfoFromDocValue( docValue )

	local localXmpCameraRawSha256
	if localXmpDevelopInfo then
		localXmpCameraRawSha256 = localXmpDevelopInfo.sha256 or localXmpDevelopInfo.contentId -- iOS stores under contentId
	end

	local sourceLinkForCopy = WFCatAssetUtils_getSourceXmpDevelopLinkAndSha256FromDocValue( docValue )

	return sourceLinkForCopy or ( ozXmpCameraRawSha256 and ozXmpCameraRawSha256 == localXmpCameraRawSha256 )

end

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

--[[

	\brief Determines if an original/proxy upload needs to wait for the upload of externalized settings to complete.

	Note: WF-870: If we're uploading the asset the first time and there are external
	develop settings we want to send those to Oz first and guarantee that they
	get there before sending the master/proxy. If we just let the uploader upload
	both simultaneously we can hit a race condition on Oz where they both get there
	at the same time and yet the rendition generation fails because the full set of
	files isn't visible yet.

	\param docValue (table) - A complete Document

	\return bool
]]
local function shouldMasterOrProxyWaitForExternalXmpSettings( docValue )
	-- WF-870: If we're uploading the asset the first time and there are external
	-- develop settings we want to send those to Oz first and guarantee that they
	-- get there before sending the master/proxy. If we just let the uploader upload
	-- both simultaneously we can hit a race condition on Oz where they both get there
	-- at the same time and yet the rendition generation fails because the full set of
	-- files isn't visible yet.

	return WFOzDocUtils.doesDocumentNeedXmpDevelopSettingsUploaded( docValue )
			and doesAssetHaveXmpDevelopSettingsLocally( docValue )
end

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

--[[

	\brief Determines if a docvalue should not attempt DCU because direct upload
		template link is unavailable even after refreshing from Oz.

	\param docValue		(table) - A complete Document
	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values

	\return bool

]]
local function shouldSkipDirectUpload( docValue, binaryType )

	return derefOpt( docValue, '_localOnly', 'binaryUpload', binaryType, 'isSkipDirectUpload' )

end

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

local function canUploadOriginalViaCopy( docValue )

	local spaceId = docValue._spaceId

	local isSpaceAsset = not not spaceId

	local copyId = derefOpt( docValue, 'copy', 'copy_id' )

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

	local sourceLinkForCopy = derefOpt( docValue, '_localOnly', 'sourceOriginalLink' )

	local canUploadViaCopy = ( copyId and not willNotUploadMaster )
			or ( isSpaceAsset and sourceLinkForCopy )

	return not not canUploadViaCopy

end

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

--[[

	\brief Conditionally inserts a row into the needsBinaryUpload
		table via emit. TODO: Add more detail about this mechanism

	\param docValue			(table) A complete Document
	\param emit			(function) - provided via WFRxIndexStoreMapServer
	\param importDate		(string - iso8601) - The date a Binary was imported to Lr
	\param captureDate		(string - iso8601) - The date a Binary was captured, as reported by the capture device
	\param isDenied			(bool) - Has this upload been denied by Oz?
	\param priorityForUpload	(int) - See WFCatPhotoImportance for details about this value
	\param skipAccountErrors	(bool) - Set to true if we want to ignore accounts erors. Only
		used outside of reindexing.

	\return nil - side effect function - db insert

]]
local function maybeEmitMasterUploadRecord(
		docValue, emit, importDate, captureDate, isDenied, priorityForUpload, skipAccountErrors )

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

	--when willNotUploadMaster is true then original should not be uploaded
	if WFPlatformInfo.isClassic() and willNotUploadMaster then
		return
	end

	local localPath = WFCatAssetUtils.getFullOriginalPathFromDocValue( docValue )
	local fileMaybeOnOfflineVolume = ( WFPlatformInfo.isDesktop and not localPath ) or false

	-- Applicable only for desktop
	local localVolumeGuid = WFCatAssetUtils.getOriginalVolumeGuidFromDocValue( docValue )
	-- Applicable only for desktop
	local localRelativePath

	if localVolumeGuid then
		localRelativePath = WFCatAssetUtils.getOriginalRelativePathFromDocValue( docValue )
	end

	local localOriginalSha256 = WFCatAssetUtils.getOriginalContentIdFromDocValue( docValue )

	local binaryType = getBinaryTypeForOriginal( docValue.type, docValue.subtype )

	-- extract /main/parts from oz
	local ozMainPartsRequestUrlTemplate = derefOpt( docValue,
													'_serverProvidedMetadata',
													'links',
													'/rels/main_create/parts',
													'href' )

	local ozMasterUrl = getOriginalLinkFromDocValue_IgnoringShaError( docValue )

	local ozMasterCreateTemplate = derefOpt( docValue,
											 '_serverProvidedMetadata',
											 'links',
											 '/rels/master_create',
											 'href' )

	local revisionIds = derefOpt( docValue,
								  '_serverProvidedMetadata',
								  'revision_ids' )

	local latestRevisionId = revisionIds and revisionIds[ #revisionIds ]

	local isOzOriginalInvalid = ozMasterUrl and isOriginalLinkInvalid( docValue )

	local imageSha256 = derefOpt( docValue, 'importSource', 'sha256' )

	local isLocalOriginalValid = localOriginalSha256 == imageSha256

	-- if our local original is not a valid original for the asset record, it's not even a candidate for
	-- uploading to Oz
	if not isLocalOriginalValid then
		-- NMB-10275: no local sha256 which means that we should *only* upload the binary if there's no ozMasterUrl which
		-- would happen if we are the only client with an original for the asset. So if either of these two exist, then the local
		-- original with no sha256 on this machine is not a valid replacement for what's on Oz. If we could guarantee that
		-- all local original annotations have a sha256, then we could skip this check and only consider the 'isLocalOriginalValid'
		-- value.
		if localOriginalSha256 or ozMasterUrl then
			return
		end
	end

	-- NMB-5779: If we have a master file then we should upload if there is no master/proxy at all. But we
	-- may have to wait until Oz marks it invalid so we don't upload it too soon. Import(migration) should
	-- flag this case in the localOnly section.

	local isInvalid
	local waitForInvalid = derefOpt( docValue, '_localOnly', 'binaryUpload', 'waitForInvalid' )
	if waitForInvalid then
		-- does master url exist and is invalid? if it doesn't exist, then we shouldn't consider it invalid
		isInvalid = isOzOriginalInvalid

		-- has no master, but proxy url exist and is invalid?
		local ozProxyUrl, _, isProxyInvalid = WFCatFileSpec.getProxyLinkFromDocValue( docValue )
		isInvalid = isInvalid or ( ozProxyUrl and isProxyInvalid )

		-- is there no master or proxy at all? this is an edge case we'll consider invalid even if it's
		-- technically not
		isInvalid = isInvalid or ( not ozMasterUrl and not ozProxyUrl )

		if not isInvalid then return end
	end

	-- SPAM [matthewj 2017-08-29]: re: the above 'invalid' handling - technically, if the image is marked
	-- as invalid, but we have the appropriate binary based on our local sha256 matching the server sha256
	-- then we *should* be able to upload it

	local ozSha256Error = derefOpt( docValue,
								  'importSource',
								  'status',
								  'error',
								  'all',
								  'reason',
								  'sha256' ) == "does not match"

	-- If there's a sha256 error that we can fix, there should be a binaryUploadError for the current ozMasterUrl
	-- If so, we can actually clear the sha256error and try to re-upload locally.
	local localSha256Ok = ozMasterUrl and derefOpt( docValue, '_localOnly', 'binaryUploadErrors', '/rels/master', ozMasterUrl )

	local blockedBySha256Error = not not ( ozSha256Error and ( not localSha256Ok ) )

	local localOriginalHasError = getLocalOriginalErrorAnnotation( docValue )
	localOriginalHasError = localOriginalHasError and localOriginalHasError ~= kUploadFailedMissingProxy
		-- Don't consider the error if it's related to the proxy.

	local trialOrSubscriptionExpiredError = not not WFCatOzErrors.hasTrialOrSubscriptionExpiredError( docValue )

	local insufficientPermissionError = not not WFCatOzErrors.hasInsufficientPermissionError( docValue )

	local hasStorageExceededError = not not WFCatOzErrors.hasStorageExceededError( docValue )

	local sourceLinkForCopy = derefOpt( docValue, '_localOnly', 'sourceOriginalLink' )
	local sourceAssetId = derefOpt( docValue, '_localOnly', 'sourceAssetId' ) or docValue._id

	local ozHasMaster = WFOzDocUtils.isOriginalAvailableOnOz( docValue )
	local shouldUpload = false

	if WFCatTrashUtils_getMasterIsUploaded( docValue ) then

		-- If this asset was copied via a delete or restore it might not have
		-- it's master link set yet, but oz may provide a master_create link
		-- in an update because it is still copying the asset asynchronously.
		-- A subsequent update will provide the link but prevent uploading
		-- the original by referring to the original asset's link.
		shouldUpload = false

	elseif not ozHasMaster or blockedBySha256Error then

		-- there's no master url on Oz, or there is but it has a sha256 error (which we need to see if we can clear)

		shouldUpload = true

	elseif ozHasMaster and ozSha256Error and not blockedBySha256Error then

		-- there's a master url and we already know we can clear it.

		shouldUpload = true

	elseif ozHasMaster and isOzOriginalInvalid and isLocalOriginalValid then

		-- the master on Oz is invalid, and our local original is valid.

		shouldUpload = true

	end

	-- For Android client, after importing assets and before all of them get fully uploaded to OZ for some reason
	-- if the local directory gets deleted, then "localpath" of those assets will be null.
	-- in those scenarios, it has the facility to support upload from camera gallery.
	-- so perform additional checks to see whether emit shall be allowed.
	-- the asset must belongs to the same device and localUrl or localAssetId field must prsent in _localOnly
	-- localUrl or localAssetId refers to local path of the asset in the camera gallery location.

	local validAndroidPath
	if not localPath and WFPlatformInfo.isAndroid then

		local uniqueDeviceId = derefOpt( docValue, 'importSource', 'uniqueDeviceId' )

		if( uniqueDeviceId == WFDeviceInfo.uniqueDeviceId() ) then

			local localUrl = derefOpt( docValue, '_localOnly', 'localUrl' )
			local localAssetId = derefOpt( docValue, 'importSource', 'localAssetId' )

			if( localUrl or localAssetId ) then
				validAndroidPath = true
			end

		end

	end

	local hasAccountError = trialOrSubscriptionExpiredError
			or insufficientPermissionError
			or hasStorageExceededError

	-- It is observed that the masterCreateUrl variable takes a while to get update from Oz.
	-- So masterCreateUrl and latestRevisionId checks here causes not to emit into index table( needsBinaryUpload ) until masterCreateUrl updated by Oz.
	-- This causes delay of 18 to 20sec in showing upload progress over cloud pannel for the mobile apps as it relies upon th pending binary upload counts
	-- Hence removing those checks so that emit shall be allowed without waiting for masterCreateUrl. Applying the same to desktop doesn't hurt.
	-- So keeping it common for all platforms.
	if ( ( localPath or validAndroidPath or ( localVolumeGuid and localRelativePath ) )
			or canUploadOriginalViaCopy( docValue ) )
		and shouldUpload
		and not localOriginalHasError
		and ( skipAccountErrors or not hasAccountError )
	then

		local contentType = WFCatFileSpec.getContentTypeFromDocValue( docValue )

		local fileSize = WFCatFileSpec.getOriginalSizeFromDocValue( docValue )

		local masterCreateUrl

		if ozMasterCreateTemplate and latestRevisionId then
			masterCreateUrl = ozMasterCreateTemplate:replace( '{revision_id}', latestRevisionId )
		end

		local mainPartsRequestUrl

		if ozMainPartsRequestUrlTemplate and latestRevisionId then
			mainPartsRequestUrl = ozMainPartsRequestUrlTemplate:replace( '{revision_id}', latestRevisionId )
		elseif shouldSkipDirectUpload( docValue, binaryType ) then
			mainPartsRequestUrl = kMainPartsRequestUrlNotAvailable
				-- indicate that template link still not availabe even after document refresh
		end

		local uploadHeaders = getEncodedUploadHeadersForBinaryType( binaryType, docValue )

		local isReadyToUpload = kNotReadyToUpload

		if masterCreateUrl or mainPartsRequestUrl then
			isReadyToUpload = isReadyToUpload + kReadyToUploadLegacy
		end

		-- Extra check for Atomic Develop Settings to not upload original until most recent
		-- revision is synced to Oz. (WF-2235) This guards against following case where
		-- asset could end up with an original uploaded, a sha256 for develop settings,
		-- but that xmp file was never uploaded to Oz.

		-- 1. WF creates asset first revision on Oz with payload.develop.xmpCameraRaw.sha256 value A
		-- 2. Second revision created locally with payload.develop.xmpCameraRaw.sha256 value B
		-- 3. WF uploads latest xmpCameraRaw binary via DCU, with sha256 B (no longer waits until all metadata synced to Oz)
		-- 4. WF uploads master since all xmpCameraRaw binary is now uploaded
		-- 5. Client goes offline before sending second revision to Oz.

		if isLatestMetadataUploaded( docValue ) then
			isReadyToUpload = isReadyToUpload + kReadyToUploadAtomic
		end

		-- Note: maybeEmitMasterUploadRecord can be called for both asset documents AND watermark documents.
		if shouldWaitForXmpSidecar( docValue ) or shouldMasterOrProxyWaitForExternalXmpSettings( docValue ) or shouldWaitForAux( docValue ) then
			isReadyToUpload = kNotReadyToUpload
		end

		logger:debugf( "emit original: %s, %s, %s, %s", docValue._id, localPath, masterCreateUrl, isReadyToUpload )

		local spaceId = docValue._spaceId

		emit( localPath, localVolumeGuid, localRelativePath, boolToInt( fileMaybeOnOfflineVolume ),
				masterCreateUrl, binaryType, contentType, uploadHeaders, importDate,
				captureDate, isDenied, imageSha256, fileSize, boolToInt( blockedBySha256Error ),
				priorityForUpload, sourceLinkForCopy, spaceId, sourceAssetId, isReadyToUpload,
				docValue.type, mainPartsRequestUrl )

	end

end

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

--[[

	\brief Conditionally inserts a row into the needsBinaryUpload
		table via emit. TODO: Add more detail about this mechanism

	\param docValue			(table) - A complete Document
	\param emit			(function) - provided via WFRxIndexStoreMapServer
	\param importDate		(string - iso8601) - The date a Binary was imported to Lr
	\param captureDate		(string - iso8601) - The date a Binary was captured, as reported by the capture device
	\param isDenied			(bool) - Has this upload been denied by Oz?
	\param priorityForUpload	(int) - See WFCatPhotoImportance for details about this value

	\return nil - side effect function: db insert

]]
local function maybeEmitAuxUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

	local uploadsRequired, auxToUpload = WFCatAssetAux.uploadsRequired( docValue )

	local insufficientPermissionError = not not WFCatOzErrors.hasInsufficientPermissionError( docValue )
	local hasStorageExceededError = not not WFCatOzErrors.hasStorageExceededError( docValue )
	local hasReferencedAuxMissingError = not not WFCatOzErrors.hasReferencedAuxMissingError( docValue )

	if not uploadsRequired
			or insufficientPermissionError
			or hasStorageExceededError
			or hasReferencedAuxMissingError then

		return

	end

	local auxPartsRequestUrlTemplate = WFCatAssetAux_getAuxPartsRequestUrlTemplate( docValue )

	local revisionIds = derefOpt( docValue,
								  '_serverProvidedMetadata',
								  'revision_ids' )

	local latestRevisionId = revisionIds and revisionIds[ #revisionIds ]

	local isRevisionOnOz = isLatestMetadataUploaded( docValue )

	-- For all required to uploads, emit a row for that aux
	for auxId, auxDetails in pairs( auxToUpload ) do

		local digest = auxDetails.digest

		local auxPartsRequestUrl

		local auxBinaryOnOz = WFCatAssetAux.isAuxObjectOnOz( docValue, auxId )

		-- If this asset was copied via a delete or restore it might not have
		-- it's aux link set yet, but oz may provide a link in an update because
		-- it is still copying the asset asynchronously.

		local wasUploadedPreviously = WFCatTrashUtils_getAuxIsUploaded( docValue, auxId )
		local localAuxPath = WFCatAssetUtils_getLocalAuxObjectPath( docValue, digest )

		local copyUrl = WFCatAssetAux_getCopyUrl( docValue )

		if copyUrl and latestRevisionId then
			copyUrl = copyUrl:replace( '{revision_id}', latestRevisionId )
		end

		local sourceAssetLinkForAuxCopy = WFCatAssetAux_getSourceAssetLinkForAuxFromDocValue( docValue )
		local isSourceAuxUploadedToOz = WFCatAssetAux_isSourceAuxUploadedToOz( docValue, digest )
		local sourceAssetId = derefOpt( docValue, '_localOnly', 'sourceAssetId' ) or docValue._id

		if ( localAuxPath or ( sourceAssetLinkForAuxCopy and isSourceAuxUploadedToOz and copyUrl ) ) -- Is able to upload from this device?
				and not WFCatAssetAux.hasLocalAuxError( docValue, auxId ) -- Has an error preventing upload?
				and not wasUploadedPreviously then

			if not auxBinaryOnOz and auxPartsRequestUrlTemplate then

				-- NOTE: this template is only partially
				-- populated at this step and the remaining
				-- fields will be updated by the chunked
				-- uploader

				local content_type = WFURI.escape( kContentTypeOctetStream, 'http_query_key' )

				if auxDetails.type == 'c2pa' then
					content_type = WFURI.escape( kContentTypeC2pa, 'http_query_key' )
				end

				local _, partialTemplate = WFURI.replaceFromOzTemplate( auxPartsRequestUrlTemplate, {
					asset_id      = docValue._id,
					sha256        = auxDetails.sha256,
					type          = auxDetails.type,
					access        = "put",
					upload_status = "all",
					content_type  = content_type
				} )

				auxPartsRequestUrl = partialTemplate

			end

			local isReadyToUpload = kNotReadyToUpload

			if not shouldWaitForXmpSidecar( docValue ) then

				if auxPartsRequestUrl and isRevisionOnOz then
					isReadyToUpload = isReadyToUpload + kReadyToUploadLegacy
				end

				local hasDocEverSyncedToOz = not not docValue._serverProvidedMetadata
					-- SPM is only present for docs that have previously synced.

				if hasDocEverSyncedToOz then

					local tempAuxLinkExists = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/temp/aux/' .. auxDetails.sha256, 'exists' )
						-- if temp link exists then binary has been uploaded, do not need to upload again.

					if not tempAuxLinkExists then
						isReadyToUpload = isReadyToUpload + kReadyToUploadAtomic
					end

				end

			end

			logger:debugf( "emit aux: %s, %s, %s, %s", docValue._id, localAuxPath, auxPartsRequestUrl, isReadyToUpload )

			-- Assumed to be nil for aux binaries
			local localVolumeGuid = nil
			local localRelativePath = nil

			local fileMaybeOnOfflineVolume = boolToInt( false )

			-- not needed here because of direct upload by parts
			local uploadUrl = nil

			local binaryType = "aux"
			local contentType = kContentTypeOctetStream

			if auxDetails.type == 'c2pa' then
				contentType = kContentTypeC2pa
			end

			local uploadHeaders = getEncodedUploadHeadersForBinaryType( "aux", docValue )

			local sha256 = auxDetails.sha256
			local fileSize = auxDetails.fileSize

			local sha256Error = boolToInt( false )

			local sourceLinkForCopy = isSourceAuxUploadedToOz and sourceAssetLinkForAuxCopy or nil

			-- not needed
			local parentDocumentId = nil

			copyUrl = sourceAssetLinkForAuxCopy and isSourceAuxUploadedToOz and copyUrl or nil

			local developUserUpdated = derefOpt( docValue, 'develop', 'userUpdated' )

			emit( localAuxPath, localVolumeGuid, localRelativePath, fileMaybeOnOfflineVolume, uploadUrl, binaryType,
					contentType, uploadHeaders, importDate, captureDate, isDenied, sha256, fileSize, sha256Error,
					priorityForUpload, sourceLinkForCopy, docValue._spaceId, sourceAssetId, isReadyToUpload, docValue.type,
					auxPartsRequestUrl, parentDocumentId, copyUrl, auxId, developUserUpdated )

		end

	end

end

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

--[[

	\brief Conditionally inserts a row into the needsBinaryUpload
		table via emit. TODO: Add more detail about this mechanism

	\param docValue			(table) A complete Document
	\param emit			(function) - provided via WFRxIndexStoreMapServer
	\param importDate		(string - iso8601) - The date a Binary was imported to Lr
	\param captureDate		(string - iso8601) - The date a Binary was captured, as reported by the capture device
	\param isDenied			(bool) - Has this upload been denied by Oz?
	\param priorityForUpload	(int) - See WFCatPhotoImportance for details about this value

	\return nil - side effect function: db insert

]]
local function maybeEmitXmpDevelopUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )
	-- Note: maybeEmitXmpDevelopUploadRecord can be called for both asset documents AND version documents.

	local needsXmpDevelopSettingsUploaded = WFOzDocUtils.doesDocumentNeedXmpDevelopSettingsUploaded( docValue )
	local hasReferencedAuxMissingError = not not WFCatOzErrors.hasReferencedAuxMissingError( docValue )

	if not needsXmpDevelopSettingsUploaded
			or hasReferencedAuxMissingError then

		return

	end

	local xmpCameraRawSha256 = derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )
	local isXmpDevelopInvalid = not not derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp/develop', 'invalid' )

	local localSha256 = WFCatAssetUtils.getLocalExternalDevelopSettingsContentId( docValue )
	local localPath = WFCatAssetUtils.getLocalExternalDevelopSettingsPath( docValue )
	local localFileSize = WFCatAssetUtils.getLocalExternalDevelopSettingsSize( docValue )

	-- construction of full path needed only for Android client
	if WFPlatformInfo.isAndroid then
		local KSPathUtils = require 'KSPathUtils'
		localPath = localPath and KSPathUtils.child( KSPathUtils.getStandardFilePath( 'carouselDocuments' ), localPath )
	end

	local ozXmpDevelopCreateTemplate = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp_develop_create', 'href' )
	local ozXmpDevelopPartsRequestUrlTemplate = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp_develop_create/parts', 'href' )

	-- If this asset was copied via a delete or restore it might not have
	-- it's xmp/develop link set yet, but oz may provide a xmp_develop_create link
	-- in an update because it is still copying the asset asynchronously.
	-- See comment for function in WFCatTrashUtils.lua for more info.

	-- also make additional check with maybeLocalXmpDevelopSupersedesOzXmpDevelop()
	-- see comment for that method for more info.

	local wasUploadedPreviously = WFCatTrashUtils_getXmpDevelopIsUploaded( docValue )

	local sourceLinkForCopy, sourceLinkSha256 =
		WFCatAssetUtils_getSourceXmpDevelopLinkAndSha256FromDocValue( docValue )

	-- if source link sha256 is not available and local binary with correct sha256
	-- is available, to be safe use the local binary
	if not sourceLinkSha256 and ( localPath and localSha256 and localSha256 == xmpCameraRawSha256 ) then
		sourceLinkForCopy = nil
	end

	-- only rely on async trash copy or copy value from original the first time.
	-- if there have maybe been any edits then should upload the file

	if ( wasUploadedPreviously or sourceLinkForCopy ) and
			WFOzDocUtils.maybeLocalXmpDevelopSupersedesOzXmpDevelop( docValue ) then

		wasUploadedPreviously = false
		sourceLinkForCopy = nil

	end

	local revisionIds = derefOpt( docValue, '_serverProvidedMetadata', 'revision_ids' )
	local sourceAssetId = derefOpt( docValue, '_localOnly', 'sourceAssetId' ) or docValue._id
	local insufficientPermissionError = not not WFCatOzErrors.hasInsufficientPermissionError( docValue )
	local hasStorageExceededError = not not WFCatOzErrors.hasStorageExceededError( docValue )

	if ( ( localPath and localSha256 and localSha256 == xmpCameraRawSha256 ) or sourceLinkForCopy ) and
		xmpCameraRawSha256
		and not wasUploadedPreviously
		and not insufficientPermissionError
		and not hasStorageExceededError
	then

		local hasDocEverSyncedToOz = not not docValue._serverProvidedMetadata
			-- SPM is only present for docs that have previously synced.

		local latestMetadataSynced = isLatestMetadataUploaded( docValue )

		local latestRevisionId = revisionIds and revisionIds[ #revisionIds ]

		local xmpDevelopCreateUrl, xmpDevelopPartsRequestUrl

		-- this template link will only be used if WFCatBinaryUploader.atomicDevelopSettings is on in WFCapabilities
		-- but grab it either way here to avoid enabling that capability requiring an re-index.

		if ozXmpDevelopPartsRequestUrlTemplate then

			local _, partialTemplate = WFURI.replaceFromOzTemplate( ozXmpDevelopPartsRequestUrlTemplate, {
				asset_id = docValue._id,
				sha256 = xmpCameraRawSha256,
				access = "put",
				upload_status = "all",
				content_type  = WFURI.escape( kContentTypeXML, 'http_query_key' )
			} )

			xmpDevelopPartsRequestUrl = partialTemplate

		end

		if ( docValue.type == 'version' and hasDocEverSyncedToOz ) or
				( docValue.type == 'asset' and latestMetadataSynced and
					derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp/develop', 'invalid' ) ) then

			xmpDevelopPartsRequestUrl = kMainPartsRequestUrlMetadataAlreadySynced
				-- see comment for kMainPartsRequestUrlMetadataAlreadySynced. WF-2429

		end

		if ozXmpDevelopCreateTemplate and latestRevisionId then

			xmpDevelopCreateUrl = ozXmpDevelopCreateTemplate:replace( '{revision_id}', latestRevisionId )

		end

		local noAuxToUpload = not shouldWaitForAux( docValue )

		local isReadyToUpload = kNotReadyToUpload

		if not shouldWaitForXmpSidecar( docValue ) then

			if latestMetadataSynced and xmpDevelopCreateUrl and noAuxToUpload then
				isReadyToUpload = isReadyToUpload + kReadyToUploadLegacy
			end

			if docValue.type == 'version' then

				if xmpDevelopPartsRequestUrl == kMainPartsRequestUrlMetadataAlreadySynced 
					and not xmpDevelopCreateUrl then

					-- WF-2590 If the version is created by Oz or other user,
					-- it might be possible that xmp_develop link doesn't exist yet

				else

					isReadyToUpload = isReadyToUpload + kReadyToUploadAtomic
						-- for versions, xmp develops is uploaded before the version metadata is synced
						-- to Oz, so it is always considered ready to upload. (Technically it is not
						-- ready to upload if the related asset has not synced to Oz, but we do not
						-- want to check that in a map function, so we will check that later.)

				end

			elseif hasDocEverSyncedToOz then

				-- For assets we require _serverProvidedMetadata to be present before any binaries can sync.

				local tempDevelopXmpLinkExists = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/temp/xmp/develop/' .. xmpCameraRawSha256, 'exists' )
					-- if temp link exists then binary has been uploaded, do not need to upload again UNLESS
					-- it needs to be uploaded via old pre-ADS method see WF-2429

				if not tempDevelopXmpLinkExists or
						xmpDevelopPartsRequestUrl == kMainPartsRequestUrlMetadataAlreadySynced then

					isReadyToUpload = isReadyToUpload + kReadyToUploadAtomic

				end

			end

		end

		logger:debugf( "emit xmp_develop: %s, %s, %s, %s, %s", docValue._id, localPath, xmpDevelopCreateUrl, xmpDevelopPartsRequestUrl, isReadyToUpload )

		local parentDocumentId
		if WFCatIndexUtils.isValidVersionDocument( docValue ) then
			parentDocumentId = derefOpt( docValue, 'asset', 'id' )
		end

		-- Assumed to be nil for xmp binaries
		local localVolumeGuid = nil
		local localRelativePath = nil

		local fileMaybeOnOfflineVolume = boolToInt( false )

		local binaryType = "xmp_develop"
		local contentType = kContentTypeXML

		local uploadHeaders = getEncodedUploadHeadersForBinaryType( binaryType, docValue )

		local sha256Error = boolToInt( false )

		-- not needed
		local auxId = nil
		local copyUrl = nil

		local developUserUpdated = derefOpt( docValue, 'develop', 'userUpdated' )

		emit( localPath, localVolumeGuid, localRelativePath, fileMaybeOnOfflineVolume, xmpDevelopCreateUrl,
				binaryType, contentType, uploadHeaders, importDate, captureDate, isDenied, localSha256, localFileSize,
				sha256Error, priorityForUpload, sourceLinkForCopy, docValue._spaceId, sourceAssetId, isReadyToUpload,
				docValue.type, xmpDevelopPartsRequestUrl, parentDocumentId, copyUrl, auxId, developUserUpdated )

	end

end

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

--[[

	\brief Conditionally inserts a row into the needsBinaryUpload
		table via emit. TODO: Add more detail about this mechanism

	\param docValue			(table) A complete Document
	\param emit			(function) - provided via WFRxIndexStoreMapServer
	\param importDate		(string - iso8601) - The date a Binary was imported to Lr
	\param captureDate		(string - iso8601) - The date a Binary was captured, as reported by the capture device
	\param isDenied			(bool) - Has this upload been denied by Oz?
	\param priorityForUpload	(int) - See WFCatPhotoImportance for details about this value

	\return nil - side effect function: db insert

]]
local function maybeEmitXmpSidecarUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

	local xmpLocalPath = WFCatFileSpec.getXmpSidecarPathFromDocValue( docValue )

	local sourceXmpLinkForCopy = WFCatAssetUtils.getSourceXmpLinkFromDocValue( docValue )

	local xmpCreateUrlTemplate = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp_create', 'href' )

	local revisionIds = derefOpt( docValue, '_serverProvidedMetadata', 'revision_ids' )
	local latestRevisionId = revisionIds and revisionIds[ #revisionIds ]

	local insufficientPermissionError = not not WFCatOzErrors.hasInsufficientPermissionError( docValue )

	local hasStorageExceededError = not not WFCatOzErrors.hasStorageExceededError( docValue )

	if WFOzDocUtils.doesDocumentNeedXmpSidecarUploaded( docValue )
		 and not insufficientPermissionError
		 and not hasStorageExceededError then

		local xmpSha256 = nil -- not needed for xmp sidecar

		local xmpCreateUrl
		if xmpCreateUrlTemplate and latestRevisionId then
			xmpCreateUrl = xmpCreateUrlTemplate:replace( '{revision_id}', latestRevisionId )
		end

		local sourceAssetId = derefOpt( docValue, '_localOnly', 'sourceAssetId' ) or docValue._id

		local xmpFileSize = WFCatFileSpec.getXmpSidecarFileSizeFromDocValue( docValue )

		local binaryType = 'xmp_sidecar'
		local uploadHeaders = getEncodedUploadHeadersForBinaryType( binaryType, docValue )

		local spaceId = docValue._spaceId

		local isReadyToUpload = kNotReadyToUpload
		if xmpCreateUrl then
			isReadyToUpload = kReadyToUploadLegacy + kReadyToUploadAtomic
		end

		logger:debugf( "emit sidecar: %s, %s, %s, %s", docValue._id, xmpLocalPath, xmpCreateUrl, isReadyToUpload )

		emit( xmpLocalPath, nil, nil, boolToInt( false ), xmpCreateUrl, binaryType,
				kContentTypeXML, uploadHeaders, importDate, captureDate, isDenied,
				xmpSha256, xmpFileSize, boolToInt( false ), priorityForUpload,
				sourceXmpLinkForCopy, spaceId, sourceAssetId, isReadyToUpload,
				docValue.type )

	end

end

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

--[[

	\brief Conditionally inserts a row into the needsBinaryUpload
		table via emit. TODO: Add more detail about this mechanism

	\param docValue			(table) A complete Document
	\param emit			(function) - provided via WFRxIndexStoreMapServer
	\param importDate		(string - iso8601) - The date a Binary was imported to Lr
	\param captureDate		(string - iso8601) - The date a Binary was captured, as reported by the capture device
	\param isDenied			(bool) - Has this upload been denied by Oz?
	\param priorityForUpload	(int) - See WFCatPhotoImportance for details about this value

	\return nil - side effect function: db insert

]]
local function maybeEmitProxyUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

	local localPath = WFCatAssetUtils.getFullProxyPathFromDocValue( docValue )

	local fileMaybeOnOfflineVolume = not localPath

	local localVolumeGuid, proxySize = WFCatFileSpec.getProxyVolumeGuidFromDocValue( docValue )

	local localRelativePath = WFCatFileSpec.getProxyRelativePathFromDocValue( docValue )
	local localProxySha256 = WFCatFileSpec.getLocalProxySha256( docValue )

	local ozProxyUrl, _, isOzProxyInvalid = WFCatFileSpec.getProxyLinkFromDocValue( docValue, proxySize )

	local ozProxyCreateTemplate = derefOpt( docValue,
											 '_serverProvidedMetadata',
											 'links',
											 '/rels/proxy_create',
											 'href' )

	local revisionIds = derefOpt( docValue,
								  '_serverProvidedMetadata',
								  'revision_ids' )

	local latestRevisionId = revisionIds and revisionIds[ #revisionIds ]

	local imageProxySha256 = derefOpt( docValue, 'importSource', 'proxySHA256' )
	local imageSha256 = derefOpt( docValue, 'importSource', 'sha256' )
	local localOriginalSha256 = WFCatAssetUtils.getOriginalContentIdFromDocValue( docValue )

	local isLocalProxyValid
	if imageProxySha256 then
		isLocalProxyValid = localProxySha256 == imageProxySha256
	else
		isLocalProxyValid = imageSha256 == localOriginalSha256
	end

	-- if our local proxy is not valid for the asset record, it's not even a candidate for
	-- uploading to Oz
	if WFPlatformInfo.isClassic() and not isLocalProxyValid then
		-- No local proxy sha256 which means that we should *only* upload the binary if there's no ozProxyUrl which
		-- would happen if we are the only client with a proxy for the asset. So if either of these two exist, then the local
		-- proxy with no sha256 on this machine is not a valid replacement for what's on Oz.
		if localProxySha256 or ozProxyUrl then
			return
		end
	end

	local copyId = derefOpt( docValue, 'copy', 'copy_id' )

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

	local masterPath = WFCatAssetUtils.getFullOriginalPathFromDocValue( docValue )

	local insufficientPermissionError = not not WFCatOzErrors.hasInsufficientPermissionError( docValue )

	local hasStorageExceededError = not not WFCatOzErrors.hasStorageExceededError( docValue )

	-- If this asset was copied via a delete or restore it might not have
	-- it's proxy link set yet, but oz may provide a proxy_create link
	-- in an update because it is still copying the asset asynchronously.
	-- A subsequent update will provide the link.
	local wasUploadedPreviously = WFCatTrashUtils_getProxyIsUploaded( docValue )

	local sourceLinkForCopy = derefOpt( docValue, '_localOnly', 'sourceProxyLink' )

	local sourceAssetId = derefOpt( docValue, '_localOnly', 'sourceAssetId' ) or docValue._id

	local isSpaceAsset = not not docValue._spaceId
	local isSpaceAssetWithImpliedProxyUpload = isSpaceAsset and getProxyExistsFromDocValue( docValue )

	-- If client is LrClassic and willNotUploadMaster flag is set, we'll also use proxy upload

	local shouldNotUploadMaster = false

	if WFPlatformInfo.isClassic() then
		shouldNotUploadMaster = willNotUploadMaster
	else -- Nimbus
		shouldNotUploadMaster = willNotUploadMaster and not masterPath
	end

	-- https://jira.corp.adobe.com/browse/WF-2611
	-- Similar to what we do for originals, check for local proxy errors. Note
	-- that local proxy errors are stored in the same place within the document
	-- as local original errors.

	local localOriginalErrorAnnotation = getLocalOriginalErrorAnnotation( docValue )
	local localProxyHasError = localOriginalErrorAnnotation == kUploadFailedMissingProxy

	-- Only try to upload a proxy if it doesn't have a proxy or master link.
	-- and the asset indicates it was imported from lightroom without a master.
	-- Also assume that asset has to have a proxy locally.

	if shouldNotUploadMaster
		and ( localPath or ( localVolumeGuid and localRelativePath ) or copyId
				or ( isSpaceAsset and sourceLinkForCopy ) )
		and ( not ozProxyUrl or ( isOzProxyInvalid and isLocalProxyValid ) )
		and not ( wasUploadedPreviously or isSpaceAssetWithImpliedProxyUpload )
		and not insufficientPermissionError
		and not hasStorageExceededError
		and not localProxyHasError
	then

		local contentType = kContentTypeDng

		if WFPlatformInfo.isNimbus() then
			localProxySha256 = derefOpt( docValue, 'importSource', 'sha256' )
		end

		local fileSize = WFCatFileSpec.getProxySizeFromDocValue( docValue )

		local proxyCreateUrl
		if ozProxyCreateTemplate and latestRevisionId then
			proxyCreateUrl = ozProxyCreateTemplate:replace( '{revision_id}', latestRevisionId )
			-- Make a new guid for the proxyId, Nimbus doesn't use this guid outside
			-- of the pre-built string in the proxy link (typically it's created by oz).
			local proxyId = KSIdGenerator.uuidInLowercaseHex()
			proxyCreateUrl = proxyCreateUrl:replace( '{proxy_id}', proxyId )
		end

		local binaryType = 'proxy'
		if proxySize == 'panorama' then
			binaryType = 'proxy_pano'
		end

		local uploadHeaders = getEncodedUploadHeadersForBinaryType( binaryType, docValue )

		local spaceId = docValue._spaceId

		local isReadyToUpload = kNotReadyToUpload

		if proxyCreateUrl then
			isReadyToUpload = isReadyToUpload + kReadyToUploadLegacy
		end

		-- See long comment in maybeEmitMasterUploadRecord for more context on why this check is needed (WF-2235)
		if isLatestMetadataUploaded( docValue) then
			isReadyToUpload = isReadyToUpload + kReadyToUploadAtomic
		end

		-- If proxy link is invalid, we don't need to wait for develop binaries upload
		if not isOzProxyInvalid and ( shouldWaitForXmpSidecar( docValue )
									or shouldMasterOrProxyWaitForExternalXmpSettings( docValue )
									or shouldWaitForAux( docValue ) ) then
			isReadyToUpload = kNotReadyToUpload
		end

		logger:debugf( "emit proxy: %s, %s, %s, %s", docValue._id, localPath, proxyCreateUrl, isReadyToUpload )

		emit( localPath, localVolumeGuid, localRelativePath, boolToInt( fileMaybeOnOfflineVolume ),
				proxyCreateUrl, binaryType, contentType, uploadHeaders, importDate,
				captureDate, isDenied, localProxySha256, fileSize, boolToInt( false ),
				priorityForUpload, sourceLinkForCopy, spaceId, sourceAssetId, isReadyToUpload,
				docValue.type )

	end

end

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

--[[

	\brief Conditionally inserts a row into the needsBinaryUpload
		table via emit. TODO: Add more detail about this mechanism

	\param docValue			(table) - A complete Document
	\param emit			(function) - provided via WFRxIndexStoreMapServer
	\param importDate		(string - iso8601) - The date a Binary was imported to Lr
	\param captureDate		(string - iso8601) - The date a Binary was captured, as reported by the capture device
	\param isDenied			(bool) - Has this upload been denied by Oz?
	\param priorityForUpload	(int) - See WFCatPhotoImportance for details about this value

	\return nil - side effect function: db insert

]]
local function maybeEmitWatermarkUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

	if docValue.type == "watermark" and doesWatermarkNeedBinaryUploaded( docValue ) then
		maybeEmitMasterUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )
	end

end

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

local function isPendingCommit( docValue )

	return docValue._labels and docValue._labels[ '_pendingCommit' ]

end

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

local function hasSomeAncillaryLocalBinary( docValue )

	return isValidAssetDocument( docValue )
		and ( WFCatAssetUtils.getLocalExternalDevelopSettingsPath( docValue )
			or WFCatAssetAux.hasSomeLocalAuxFiles( docValue ) )

end

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

--[[

	\brief Checks if binaries associated with an asset document can be uploaded

	\param docValue (table) - A complete Document

	\returns bool

]]
local function canAssetBeUploaded( docValue )

	if isPendingCommit( docValue ) then
		return false -- https://jira.corp.adobe.com/browse/LRMI-4210081
	end

	if doesAssetDocumentHaveLocalBinary( docValue )
			or hasSomeAncillaryLocalBinary( docValue )
			or isAssetDocumentACopyWithSource( docValue )
			or isSpaceAssetWithSourceLinks( docValue )
			or isSpaceAssetEditable( docValue )
			or ( WFPlatformInfo.isAndroid or WFPlatformInfo.isClassic() ) and WFCatIndexUtils.isValidAssetDocument( docValue ) then
				 -- Additional check for Android.
				 -- localPath may be nil.( This may happen when after meta data sync, if the asset gets deleted
				 -- in those cases, attemp will be made to look for resources at gallery)
				 -- So, for android, even though localPath is nil and if it is a valid asset document allow to
				 -- emit for further processing.

		return true

	end

	return false

end

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

--[[

	\brief Checks if binaries associated with a watermark document can be uploaded

	\param docValue (table) - A complete Document

	\returns bool

]]
local function canWatermarkBeUploaded( docValue )

	if WFCatIndexUtils.isValidWatermarkDocument( docValue ) then
		return not not getLocalWatermarkBinaryPath( docValue )
	end

	return false

end

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

--[[

	\brief Checks if binaries associated with a version document can be uploaded

	\param docValue (table) - A complete Document

	\returns bool

]]
local function canVersionBeUploaded( docValue )

	return not not getLocalExternalDevelopSettingsPath( docValue ) and
			WFCatIndexUtils.isValidVersionDocument( docValue )

end

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

--[[

	\brief implementation function for generating a new index
		store table. See WFCatIndexDefinitions.lua

	\param _		(ignore) - Required to match implementation arguments
	\param indexStoreSchema (WFRxIndexStoreSchema)

	\return nil - side effect function: db insert

]]
local function generateIndexSchema( _, indexStoreSchema )

	check( '_', 'WFRxIndexStoreSchema' )

	indexStoreSchema:addTable {

		sql = [[CREATE TABLE needsBinaryUpload(
				localFilePath,
				localVolumeGuid,
				localRelativePath,
				fileMaybeOnOfflineVolume,
				uploadUrl,
				binaryType,
				contentType,
				uploadHeaders,
				importDate,
				captureDate,
				isDenied,
				sha256,
				fileSize,
				sha256Error,
				priorityForUpload,
				sourceLinkForCopy,
				spaceId,
				sourceAssetId,
				isReadyToUpload,
				docType,
				mainPartsRequestUrl,
				parentDocumentId,
				copyUrl,
				auxId,
				developUserUpdated )]],

					-- priorityForUpload: This priority is used in conjuction with other data that's
					-- not accessible inside of an asset/version document (such as shared album membership)
					-- to determine the binary upload order.

					-- sourceAssetId: For assets that were contributed to a shared album,
					-- sourceAssetId is the document ID of the source asset. Otherwise,
					-- it's the doc ID of the current asset. This field is only necessary
					-- to fix WF-750.

					-- mainPartsRequestUrl: This url is used to store /rels/main_create/parts link and `uploadUrl`
					-- is used to store /rels/master_create link

					-- parentDocumentId: A column to keep track of a hierarchical reference in the case
					-- of version documents. If a version document is being emited, this
					-- column will be populated with its parent (asset) docId. This allows
					-- our candidates query to have a filter on it, effectively forcing
					-- versions to be uploaded last.

					-- copyUrl: Certain binaries, such as auxiliaries, have unique URLs used for copying
					-- that are *not* the same as the standard upload URLs. For all other binaries, the
					-- copy url is the same as the upload URL, and the only difference is the payload
					-- that we send (binary vs JSON).

					-- auxId: The Copy-Auxiliary workflow requires that we pass in the auxIds of the
					-- binaries be uploaded.

					-- developUserUpdated: Taken from the develop section. Required to be passed to POST call
					-- to complete direct binary upload for aux and xmp_develop.

		relevantTypeSubtypeTuples = {
			{ 'asset', '*' },
			{ 'version', '*' },
			{ 'watermark', '*' },
		},

		map = function( docValue, emit )

			--logger:debugf( "indexer reviewing %s", docValue )

			if docValue.type == 'asset' and canAssetBeUploaded( docValue ) then

				local isDenied = checkIfDenied( docValue )

				local importDate, captureDate = getDatesFromDocValue( docValue )

				local priorityForUpload = WFCatPhotoImportance.getPriorityForUpload( docValue )

				-- maybe add row for each auxiliary needing to be uploaded - asset local
				maybeEmitAuxUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

				-- maybe add row for xmp/develop
				maybeEmitXmpDevelopUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

				-- maybe add row for full xmp sidecar
				maybeEmitXmpSidecarUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

				-- maybe add row for proxy but only if no master exists (copied Lightroom asset)
				maybeEmitProxyUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

				-- maybe add row for master
				maybeEmitMasterUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

			elseif docValue.type == 'version' and canVersionBeUploaded( docValue ) then

				local isDenied = checkIfDenied( docValue )

				local importDate = derefOpt( docValue, 'userCreated' )
				local captureDate = importDate
					-- Version documents don't have import or capture dates, so instead use the
					-- creation date for both.

				local priorityForUpload = WFCatPhotoImportance.getPriorityForUpload( docValue )

				-- TODO: add support for auxiliary emission on version row execution

				-- maybe add row for xmp/develop
				maybeEmitXmpDevelopUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

			elseif docValue.type == 'watermark' and canWatermarkBeUploaded( docValue ) then

				local isDenied = checkIfDenied( docValue )

				local importDate = derefOpt( docValue, 'userCreated' )
				local captureDate = importDate
					-- Watermark documents don't have import or capture dates, so instead use the
					-- creation date for both.

				local priorityForUpload = WFCatPhotoImportance.getPriorityForUpload( docValue )

				-- maybe add row for watermarks
				maybeEmitWatermarkUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )

			end

		end,

	}

	indexStoreSchema:addIndex(
		[[CREATE INDEX needsBinaryUpload_isBlistedImportCapture
			ON needsBinaryUpload( fileMaybeOnOfflineVolume, isDenied, priorityForUpload DESC, importDate, captureDate )]] )

	indexStoreSchema:addIndex(
		[[CREATE INDEX needsBinaryUpload_binaryType
			ON needsBinaryUpload( binaryType )]] )

	indexStoreSchema:addIndex(
		[[CREATE INDEX needsBinaryUpload_docType
			ON needsBinaryUpload( docType )]] )

end

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

--[[

	\brief Extract the self OZ link for a given document

	\param docValue (table) A complete Document

	\return string

]]
local function getDocumentUrl( docValue )

	return derefOpt( docValue, '_serverProvidedMetadata', 'links', 'self', 'href' )

end

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

--[[

	\brief Adds a log for an invalid original to the error log

	\param errorLog  (table) - Collection of errors
	\param docId     (string) - The ID of the Document an upload is associated with
	\param filePath  (string) - Path to the binary on this device
	\param reason	 (string) - The reason for the error

	\return nil

]]
local function addInvalidOriginalErrorToLog( errorLog, docId, filePath, reason )

	errorLog[ #errorLog + 1 ] = {

		time = KSDate.timeInIso8601FormatWithTimeZone(),

		id = docId,
		binaryType = 'master',
		error = reason,
		filePath = filePath,
	}

	logger:warnf( "Marking original as invalid for docId %s path %s", docId, filePath )

end

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

--[[

	\brief Configure request options for a binary upload

	\param binaryType		(string) - The type of the binary, refer to kBinaryType... values
	\param contentType		(string) - The content type associated with the upload
	\param isCopy			(bool) - Can this upload be skipped due to copy-on-oz
	\param additionalUploadHeaders	(table) - Extra headers to be associated with the upload
	\param optSpecialRendition	(bool) - Does this upload need a special rendition - see docNeedsSpecialProxyOrRenditionRequest
	\param optSpecialProxy			(bool) - Does this upload need a special proxy - see docNeedsSpecialProxyOrRenditionRequest

	\return table

]]
local function getRequestOptionsFor(
		binaryType, contentType, isCopy, additionalUploadHeaders, isDirectUpload,
		atomicDevelopSettingsEnabled, optSpecialRendition, optSpecialProxy )

	local options = {
		dataFormat = 'filePath',
		requestPriority = WFRxHttpClient.requestPriorityMedium,
		requestCost = uploadHttpRequestCost
	}

	if isCopy then
		contentType = 'application/json'
	end

	if isImageOrVideoOriginal( binaryType ) then

		local renditionsValue = 'all'
		local proxiesValue = 'all'

		if WFPlatformInfo.isWindows then

			-- proxy headers expects ; as delimiter while the renditions expects , as delimiter
			if optSpecialRendition then
				renditionsValue = 'all,1080p709'
			end

			if optSpecialProxy then
				proxiesValue = 'all;1080p709'
			end

		end

		local requestHeaders = {
			{
				header = 'X-Generate-Renditions',
				value = renditionsValue,
			},
			{
				header = 'Content-Type',
				value = contentType,
			},
			{
				header = "Accept",
				value = "application/json",
			},
			{
				header = 'X-Generate-Proxies',
				value = proxiesValue,
			}
		}

		if isCopy then
			table.insert( requestHeaders, {
				header = 'X-Async-Copy',
				value = 'true'
			} )
		end

		options.requestHeaders = requestHeaders

	elseif isPresetOriginal( binaryType ) then
		-- master for preset/profile

		local requestHeaders = {
			{
				header = 'Accept',
				value = 'application/json',
			},
			{
				header = 'Content-Type',
				value = contentType or kContentTypeXML,
			},
		}

		options.requestHeaders = requestHeaders

	elseif isCameraProfileOriginal( binaryType ) then
		-- master for camera_profile
		local requestHeaders = {
			{
				header = 'Accept',
				value = 'application/json',
			},
			{
				header = 'Content-Type',
				value = contentType or kContentTypeOctetStream,
			},
		}
		options.requestHeaders = requestHeaders

	elseif isProxy( binaryType ) then

		options.requestHeaders = {
			{
				header = 'Content-Type',
				value = contentType or kContentTypeDng,
			},
			{
				header = 'Accept',
				value = 'application/json',
			},
			{
				header = 'X-Generate-Renditions',
				value = 'all',
			},
		}

		if isCopy then
			table.insert( options.requestHeaders, {
				header = 'X-Async-Copy',
				value = 'true'
			} )
		else
			local proxyType = '2560'
			if binaryType == 'proxy_pano' then
				proxyType = 'panorama'
			end
			table.insert( options.requestHeaders,
				{
					header = 'X-Proxy-Type',
					value = proxyType,
				} )
		end

	elseif isXmpDevelop( binaryType ) then

		options.requestHeaders = {
			{
				header = 'Content-Type',
				value = contentType or kContentTypeXML,
			},
			{
				header = 'Accept',
				value = 'application/json',
			},
		}

		-- do not have to check for atomicDevelopSettings capability because isDirectUpload set to true implies it is enabled
		if isDirectUpload then

				table.insert( options.requestHeaders, {
					header = 'Prefer',
					value = 'update-atomic'
				} )

		else

			table.insert( options.requestHeaders, {
				header = 'X-Generate-Renditions',
				value = 'all',
			} )
			table.insert( options.requestHeaders, {
				header = 'X-Generate-Proxies',
				value = 'all',
			} )

		end

	elseif isXmpSidecar( binaryType ) then

		options.requestHeaders = {
			{
				header = 'Content-Type',
				value = contentType or kContentTypeXML,
			},
			{
				header = 'Accept',
				value = 'application/json',
			},
		}

	elseif isWatermarkOriginal( binaryType ) then

		options.requestHeaders = {
			{
				header = 'Content-Type',
				value = contentType,
			},
			{
				header = 'Accept',
				value = 'application/json',
			},
		}

	elseif isAux( binaryType ) then

		options.requestHeaders = {
			{
				header = 'Content-Type',
				value = contentType or kContentTypeOctetStream,
			},
			{
				header = 'Accept',
				value = 'application/json',
			},
		}

		if isDirectUpload and atomicDevelopSettingsEnabled then

			table.insert( options.requestHeaders, {
				header = 'Prefer',
				value = 'update-atomic'
			} )

		end

	end

	for header, value in pairs( additionalUploadHeaders or {} ) do
		local headers = options.requestHeaders or {}
		headers[ #headers + 1 ] = {
			header = header,
			value = value
		}
		options.requestHeaders = headers
	end

	return options

end

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

--[[

	\brief Checks if Oz reports the sha for an upload does not match.

	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorShaDoesNotMatch( binaryType, responseData, headers )

	if isOriginalBinary( binaryType ) and headers and headers.status_code == '400' and headers[ 'X-Result-Code' ] == '1005' then

		local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

		return ok and derefOpt( response, 'errors', 'sha256', 1 ) == "does not match"

	end

	return false

end

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

--[[

	\brief Checks if Oz reports the sha for an aux upload does not match.

	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorAuxShaDoesNotMatch( responseData, headers )

	if  headers and headers.status_code == '404' and headers[ 'X-Result-Code' ] == '1000' then

		local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

		return ok and derefOpt( response, 'errors', 'aux', 1 ) == "does not exist with provided sha256"

	end

	return false

end

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


--[[

	\brief Checks if Oz reports the user's subscription has expired

	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorTrialOrSubscriptionExpired( responseData, headers )

	if headers and headers.status_code == '403' and headers[ 'X-Result-Code' ] == '4300' then

		local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

		return ok and derefOpt( response, 'subtype' ) == WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError

	end

	return false

end

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

--[[

	\brief Checks if Oz reports the upload is a master upload when not allowed

	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorUploadMasterWhenNotAllowed( binaryType, responseData, headers )

	if isOriginalBinary( binaryType ) and headers and headers.status_code == '403' and headers[ 'X-Result-Code' ] == 'UploadMasterWhenNotAllowedError' then

		return true

	end

	return false

end

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

--[[

	\brief Check if Oz reports that the user has insufficient permissions to upload the master

	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorInsufficientPermissionForMaster( binaryType, responseData, headers )

	if isOriginalBinary( binaryType ) and headers and headers.status_code == '403' and headers[ 'X-Result-Code' ] == '1016' then

		return true

	end

	return false

end

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

--[[

	\brief Check if Oz reports that the user has insufficient permissions

	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorInsufficientPermission( responseData, headers )

	if headers and headers.status_code == '403' and headers[ 'X-Result-Code' ] == 'InsufficientPermissionError' then

		return true

	end

	return false

end

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

--[[

	\brief Check if Oz reports that the submitted content type is incorrect

	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorContentTypeMismatch( responseData, headers )

	if headers and headers.status_code == '415' and headers[ 'X-Result-Code' ] == '1003' then

		local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

		return ok and derefOpt( response, 'errors', 'content_type', 1 ) == "should match the subtype of the asset"

	end

	return false

end

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

--[[

	\brief Check if Oz reports that the document in question is missing an auxilliary

	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorAuxMissing( responseData, headers )

	if headers and headers.status_code == '403' and headers[ 'X-Result-Code' ] == 'AuxMissingError' then

		return true

	end

	return false

end

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

--[[

	\brief Check if Oz reports that the document has no payload

	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool

]]
local function errorPayloadMustExist( binaryType, responseData, headers )

	if responseData == 'payload must exist' then
		-- Chunked uploader check file size up front as it can't be 0.
		return true
	end

	if headers and headers.status_code == '400' and headers[ 'X-Result-Code' ] == '1005' then

		local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

		return ok and derefOpt( response, 'errors', 'payload', 1 ) == "must exist"

	end

	return false

end

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

--[[

	\brief Assign a new ID if an upload encouters a "resource not found error" from Oz

	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return bool or {bool, string}

]]
local function errorAndMaybeNewIdIfMasterOrProxyResourceDoesNotExist( binaryType, responseData, headers )

	if ( isOriginalBinary( binaryType )
			or isProxy( binaryType )
			or isXmpSidecar( binaryType )
			or isDevelopBinary( binaryType ) )
			and headers and headers.status_code == '404' and headers[ 'X-Result-Code' ] == '1000' then

		local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

		if ok then

			local resourceDoesNotExist = derefOpt( response, 'subtype' ) == "ResourceNotFoundError"

			return resourceDoesNotExist, response.new_id

		end

	end

	return false

end

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

--[[

	\brief Determine if a video transfer function requires special handling

	\param videoColorTransferFunction (string) - the video color transfer function

	\return {bool, bool}

]]
local function transferFunctionNeedsSpecialProxyOrRenditionRequest( videoColorTransferFunction )

	local optSpecialRendition = false
	local optSpecialProxy = false

	if WFPlatformInfo.isWindows and videoColorTransferFunction == "bt2100_hlg" then
		optSpecialRendition = true
		optSpecialProxy = true
	end

	return optSpecialRendition, optSpecialProxy

end

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

--[[

	\brief Determine if a document needs special proxy handling or renditions.

	\param docValue (table) - Full document value

	\return {bool, bool}

]]
local function docNeedsSpecialProxyOrRenditionRequest( docValue )

	local videoColorTransferFunction = derefOpt( docValue, 'video', 'transferFunction' )
	-- check WIN_ENV here
	if videoColorTransferFunction then
		return transferFunctionNeedsSpecialProxyOrRenditionRequest( videoColorTransferFunction )
	end

	return false, false

end

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

--[[

	\brief Wraps a BinaryUploader instance in a performance reporter

	\param id		(string) the name to be associcated with this BinaryUploader in the performance report.
	\param observable	(Observable<BinaryUploader>) - the instance of a BinaryUploader

	\return Observable - same observable as what is passed in, all triggered actions are side effects.

]]
local function maybePerformanceWrap( id, observable )

	id = id or 'unknown'

	return WFPerformanceReport.wrapObservable( 'WFCatBinaryUploader: ' .. id, observable )

end

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

--[[

	\brief When catch an error, make sure we correctly handle response error to string

	\param responseData	responsed data caught from error

	\return string - convert responseData to string according to type

]]
local function caughtResponseDataToString( responseData )

	local responseDataStr = "unknown"
	if type( responseData ) == 'table' then
		responseDataStr = WFTableArchiving.tableToString( responseData )
	elseif responseData then
		responseDataStr = tostring( responseData )
	end

	return responseDataStr

end

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

--[[

	\brief Utility function determining what kind of join to use in a sql query

	\param onlyUploadSpaceAssets (bool)

	\return bool

]]
local function getTypeOfJoinForSpaceTable( onlyUploadSpaceAssets )

	local typeOfJoin = "LEFT"

	if onlyUploadSpaceAssets then
		typeOfJoin = "INNER"
	end

	return typeOfJoin

end

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

--[[

	\brief Consruct an overlay which updates a source asset in the
	case of a copy. This will either get the createFrom links, or
	the actual links for this new asset. We may end up failing on
	the createFrom links but will get another error and another
	new_id that we can try.

	\param docIdToUpdate	(string) - Document ID
	\param sourceDocValue	(table) - The full source document

	\return table - to be passed to applyOverlay, eg. A docstore compliant update table.

]]
local function makeOverlayToUpdateSourceAsset( docIdToUpdate, sourceDocValue )

	-- This will either get the createFrom links, or the
	-- actual links for this new asset. We may end up failing
	-- on the createFrom links but will get another error and
	-- another new_id that we can try.

	local localOnly = {}

	localOnly.sourceOriginalLink = WFCatFileSpec.getOriginalLinkFromDocValue( sourceDocValue )
	localOnly.sourceXmpLink = WFCatFileSpec.getXmpLinkFromDocValue( sourceDocValue )
	localOnly.sourceXmpDevelopLink, localOnly.sourceXmpDevelopSha256 =
		WFCatFileSpec.getXmpDevelopLinkAndSha256FromDocValue( sourceDocValue )
	localOnly.sourceProxyLink = WFCatFileSpec.getProxyLinkFromDocValue( sourceDocValue )
	localOnly.sourceAuxLinks = WFCatFileSpec.getTableOfAuxLinksFromDocValue( sourceDocValue )

	if localOnly.sourceAuxLinks then
		localOnly.sourceAssetLinkForAux = WFCatFileSpec.getSelfLinkFromDocValue( sourceDocValue )
	end

	localOnly.sourceAssetId = sourceDocValue._id

	return {
		_id = docIdToUpdate,
		_localOnly = localOnly,
	}

end

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

-- @mark Class Declaration

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

local WFCatBinaryUploader = {

	_catalog = false,
		-- (WFCatalog)

	_docStore = false,
		-- (WFRxDocStore)

	_catalogSpec = false,
		-- (WFCatSpec)

	_numberOfScheduledBinaryUploads = false,
		-- (number) Number of uploads that we have scheduled for upload.

	_recentlyUploaded = false,
		-- (table) set of document IDs / binary types for which we've recently scheduled for upload.

	_scheduledAndWaitingBinaryUpload = false,
		-- (table) A table containing the doc ID, binary type, and subscription of the most recently
		-- scheduled binary upload. As soon as the upload starts to process, _scheduledAndWaitingBinaryUpload
		-- is set back to nil.

	_scheduledUploadSubscriptions = false,
		-- (table) Table of binary info of each scheduled binary upload. This contains
		-- the subscription to the upload and the upload priority.

	_uploadingAssetWithSha256 = false,
		-- (table) Table of counts of the number of assets attempted to be uploaded indexed by sha256.

	_numAssetsWaitingForUploadWithSha256 = false,
		-- (table) Table of counts of the number of assets waiting for upload to finish to be uploaded indexed by sha256.
		-- used to limit the number of times uploadSomeBinaries is called when multiple copies are waiting for the first
		-- asset to finish uploading.

	_continuousUploadTask = false,
		-- (WFRxContinuousTask) A continuous task to handle continuous uploads

	_errorStatusStream = false,
		-- (Subject) A Rx.Subject used to relay upload error status

	_idleStatusStream = false,
		-- (Subject) A Rx.Subject used to indicate idle status.

	_uploadScheduledStream = false,
		-- (Rx BehaviorSubject) A local stream that fires when we schedule a new HTTP request.

	_uploadStatusChangeStream = false,
		-- (Rx BehaviorSubject) A local stream that fires when the HTTP request for an upload is issued,
		-- or when a scheduled upload's priority changes.

	_uploadFinishedStream = false,
		-- (Rx BehaviorSubject) A local stream that fires when an upload completes.

	_volumeStatusHasChangedStream = false,
		-- (Rx BehaviorSubject) A local stream that fires when a volumes status changes.

	_allowBackgroundUpload = false,
		-- ( Rx ReplaySubject ) A boolean to indicate whether we should allow background upload to
		-- schedule more tasks bypassing rate limiter

	_failedUploadDeniedListHelper = false,
		-- (WFCatDeniedListHelper) A helper classed used to temporarily deny failed uploads.

	_chunkedUploader = false,
		-- (WFCatContentRangeUploader) Used to upload large files in chunks.

	_directChunkedUploader = false,
		-- (WFCatDirectChunkedUploader) Used to direct upload files.

	_reloadLinksForDocumentStream = false,
		-- ( Rx ReplaySubject ) A stream of docIds which need to have links refreshed to get
		-- template link required for a direct upload

	_reloadLinksForDocumentSubscription = false,
		-- A subscription on _reloadLinksForDocumentStream to call reloadLinksForDocument()

	_shouldReprocessVersionsStream = false,
		-- ( Rx Subject ) Triggered to force a reprocessing of candidates stream via method
		-- maybeFixUpOrRemoveCandidates()

	_timerAborter = false,
		-- (ReplaySubject) Used to indicate the uploader is being closed by sending an error, used to clear out blackout timers.

	_uploadErrorLog = false,
		-- (table) record of recent upload errors for the diagnostic log

	_uploadSpeedStats = false,
		-- (table) record of recent upload speeds

	_clearFromRecentlyUploadedSeconds = false,
		-- (number) Delay in seconds before clearing the recently uploaded table.

	_waitForOtherUploadSeconds = false,
		-- (number) Delay in seconds before clearing the recently uploaded table waiting for another upload to finish that has the same sha256.

	_waitForCopyFailedSeconds = false,
		-- (number) Delay in seconds before clearing the recently uploaded table after marking an asset with copyFailedSourceMissing.

	_timer = false,
		-- (timer) Use by unit tests to avoid warnings generated by WFRxLowFidelityTimer.

	_uploadProgress = false,
		-- (table)
			-- (WFProgressReporter) progressReporter instance
			-- (number) totalBytesNeedToUpload
			-- (number) totalBytesUploadedSoFar

	_onlyUploadSpaceAssets = false,
		-- (boolean) If true, the uploader will ignore catalog-feed assets.

	_uploadBlackoutWindowInSeconds = false,
		-- (number) delay betweeen upload re-attempt

	_uploadBlackoutWindowCount = false,
		-- (number) scalar value for multiplying the _uploadBlackoutWindowInSeconds

	_deletedDocIdsCache = false,
		-- (table) A boundless cache of deleted doc Ids. Used to prevent us from attempting
		-- to sync a document that has been deleted.
		-- key: docId, value: true

	_disposeActiveUploadsWhenDeletedSubscription = false,
		-- A subscription that monitors deletions from the doc store.

	_binaryUploadCompletionStream = false,
		-- (Subject) Fires every time a binary succesfully uploads.

	_uploadsByType = false,
		-- (table) Table of the number of binaries uploaded, organized by type.

	_uploadsByTypeStream = false,
		-- (Rx BehaviorSubject) Fires whenever _uploadsByType is updated.

	_directUploadConsecutiveFailureCounter = false,
		-- (number) Count consecutive http request failure using direct upload

	_directUploadFallBackToOzUploadTimerSub = false,
		-- (Disposable) Used to blackout direct-upload when an asset has encountered 3 consecutive failures
		-- using direct-upload, we'll redirect all assets to use the older oz-upload mechanism for some time window(default 30 mins)

	_directUploadFallBackToOzUploadWindowInSecond = false,
		-- (number) the duration in blackouting direct-upload, default 30 mins

	_fallbackToOzUploadForOriginalsOnly = false,
		-- (boolean) Default to false. If enabled, directChunkedUploads will be used as a fallback
		-- for original binary upload only.

	_includeProgress = false,
		 -- (boolean) True by default, but defaults to false for unit tests.
		 -- If true, progress events are included for binary uploads.

	_inFlightUploadProgress = false,
		-- (table) A table of in-flight upload progress information.

	_debugInFlightUploadProgressUpdateStream = false,
		-- (Rx BehaviorSubject) Fires whenever _inFlightUploadProgress is updated. Only used be tests.

	_wf2220Counter = false,
		-- (number) Counter to avoid sending upload statistics too often

	_debugForceDocIdUploadOrdering = false,
		-- (boolean) If true, the SQLite queries will include an additional ORDER BY docId statement.

	_capabilitiesCache = false,
		-- (WFCapabilitiesCache)

	class = {

		generateIndexSchema = generateIndexSchema,

		readyToUploadEnum = readyToUploadEnum,

		maybeEmitMasterUploadRecord = function( _, ... )
			return maybeEmitMasterUploadRecord( ... )
		end,

		canUploadOriginalViaCopy = function( _, ... )
			return canUploadOriginalViaCopy( ... )
		end

	},

	WFCloseSupport.addCloseSupportMethods,

}

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

--[[

	\brief Initialize and bootstrap a BinaryUploader instance

	\param catalog			(WFCatalog) - The catalog associated with the current account
	\param chunkedUploader		(WFCatDirectedChunkedUploader) - Instance of a chunked uploader for multi-part uploads
	\param optTaskDelays		(table) - Delay and backoff configurations
	\param optCatSpec		(WFCatalogSpec) - WFCatalogSpec instance - usually not overridden outside of testing
	\param optUploaderOptions	(table) - configuration options for the uploader TODO: Document these
		- forceDocIdUploadOrdering (boolean) if true, the SQLite queries will include
			an additional ORDER BY docId statement. Useful in validating tests
			that require strict upload ordering. Only available in test environments.

	\return nil

]]
function WFCatBinaryUploader:__init( catalog, chunkedUploader, optTaskDelays, optCatSpec, optUploaderOptions )

	check( 'WFCatBinaryUploader', 'WFCatalog', '?table', '?table', '?WFCatSpec', '?table' )

	optUploaderOptions = optUploaderOptions or {}

	self:setOwned {
		_catalog = catalog:relayObject()
	}

	self._docStore = catalog:docStore()

	self._catalogSpec = optCatSpec or catalog:ozCatalogSpec()

	self._failedUploadDeniedListHelper = WFCatDeniedListHelper(
												self._docStore,
												'WFCatBinaryUploader',
												numberOfTriesBeforeDenyingUpload,
												failedUploadBlackoutWindowInSeconds,
												true )

	if optUploaderOptions.progressReporter then

		self._uploadProgress = {
			progressReporter = optUploaderOptions.progressReporter,
			totalBytesNeedToUpload = 0,
			totalBytesUploadedSoFar = 0,
		}

		local binaryUploadProgress = {
			totalBytesNeedToUpload = self._uploadProgress.totalBytesNeedToUpload,
			totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar,
		}
		-- Trigger once during the startup to fire the upload progress RX chain.
		-- This will help to display pending uploads if any during restart of the app
		self._uploadProgress.progressReporter:updateProgressWithInfo( binaryUploadProgress, true )

	end

	self._includeProgress = not WFPlatformInfo.isUnitTest() or optUploaderOptions.forceIncludeProgress
		-- By default, progress events are disabled for unit tests unless explicitly specified.
		-- This is only done so that we don't have to modify all of the existing test cases
		-- that were written before upload progress was introduced.

	self._debugForceDocIdUploadOrdering = WFPlatformInfo.isUnitTest() and optUploaderOptions.forceDocIdUploadOrdering

	self._inFlightUploadProgress = {}
	self._debugInFlightUploadProgressUpdateStream = Rx.BehaviorSubject( {} )

	self._chunkedUploader = chunkedUploader or
		WFCatContentRangeUploader( catalog:docStore(), self._catalogSpec, nil, nil, nil, self._uploadProgress )

	self._directChunkedUploader = WFCatDirectChunkedUploader( self._catalogSpec )

	self._fallbackToOzUploadForOriginalsOnly =
		optUploaderOptions.fallbackToOzUploadForOriginalsOnly or false

	self._capabilitiesCache = WFCapabilitiesCache( optUploaderOptions.capabilities )

	self._directUploadConsecutiveFailureCounter = 0

	self._directUploadFallBackToOzUploadTimerSub = nil

	self._directUploadFallBackToOzUploadWindowInSecond =
		optUploaderOptions.directUploadFallBackToOzUploadWindowInSecond or kDirectUploadFallBackToOzUploadWindowInSecond

	self._recentlyUploaded = {}

	self._scheduledUploadSubscriptions = {}

	self._numberOfScheduledBinaryUploads = 0

	self._wf2220Counter = 0

	self._uploadingAssetWithSha256 = {}
	self._numAssetsWaitingForUploadWithSha256 = {}

	self._onlyUploadSpaceAssets = false

	local continuousTaskDelays = optTaskDelays or {
		progressiveBackoff = true,
		errorDelay = 10,
		maxDelay = 60,
	}

	local continuousTask = function()
		return self:uploadSomeBinaries()
	end

	self._continuousUploadTask = WFRxContinuousTask( "WFCatBinaryUploader", continuousTask, continuousTaskDelays )

	self._errorStatusStream = Rx.Subject()

	self._idleStatusStream = Rx.Subject()

	self._binaryUploadCompletionStream = Rx.Subject()

	self._uploadStatusChangeStream = Rx.BehaviorSubject()

	self._uploadFinishedStream = Rx.BehaviorSubject()

	self._uploadScheduledStream = Rx.BehaviorSubject()

	self._volumeStatusHasChangedStream = Rx.BehaviorSubject()

	self._uploadsByType = {}

	self._allowBackgroundUpload = Rx.ReplaySubject( 1 )
	self._allowBackgroundUpload:onNext( false )

	self._reloadLinksForDocumentStream = Rx.ReplaySubject( 1 )

	self._reloadLinksForDocumentSubscription = self._reloadLinksForDocumentStream
		:flatMap( function(...)
			return self:reloadLinksForDocument(...)
		end )
		:catch( function()
			return Observable.just( true )
		end )
		:subscribe()

	self._shouldReprocessVersionsStream = Rx.Subject()

	self._timerAborter = Rx.ReplaySubject( 1 )
	self._timerAborter:onNext( true )

	self._uploadErrorLog = {}

	self._uploadSpeedStats = {
		buckets = {},
		totalSize = 0,
	}

	self._clearFromRecentlyUploadedSeconds = optUploaderOptions.clearFromRecentlyUploadedSeconds or kClearFromRecentlyUploadedSeconds
	self._waitForOtherUploadSeconds = optUploaderOptions.waitForOtherUploadSeconds or kWaitForOtherUploadSeconds
	self._waitForCopyFailedSeconds = optUploaderOptions.waitForCopyFailedSeconds or kWaitForCopyFailedSeconds

	self._timer = optUploaderOptions.timer or WFRxLowFidelityTimer.timer

	self._uploadBlackoutWindowInSeconds = optUploaderOptions.uploadBlackoutWindowInSeconds or kUploadBlackoutWindowInSeconds
	self._uploadBlackoutWindowCount = optUploaderOptions.uploadBlackoutWindowCount or kUploadBlackoutWindowCount

	self._deletedDocIdsCache = {}

	local kBinaryUploaderDeletionChangeCounterLocalDocId = 'WFCatBinaryUploader-deletionChangeCounter'

	self._disposeActiveUploadsWhenDeletedSubscription = self._docStore:deletedDocsSinceNamedChangeCounter(
			kBinaryUploaderDeletionChangeCounterLocalDocId,
			{
				_id = true,
			} )

		:subscribe( function( deletedDocValues )

			for _, docValue in ipairs( deletedDocValues ) do

				local docId = docValue._id

				self._deletedDocIdsCache[ docId ] = true

				self:disposeOfUploadForDeletedDocId( docId )
					-- When a document is deleted, make sure we abort any active uploads
					-- associated with that document.

			end

		end )

	WFDiagnosticLog.addLogProvider( self )

end

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

function WFCatBinaryUploader:isAtomicDevelopSettingsEnabled()

	return not not self._capabilitiesCache:get( 'WFCatBinaryUploader.atomicDevelopSettings' )

end

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


--[[!

	This code works by running an index store query to identify the highest priority
	binary that needs to be uploaded, and then issuing a put HTTP request for that
	single binary.

	Just because a put request is issued by the binary uploader, it does not mean
	that the HTTP request has started. This is due to the scheduling within the HTTP
	rate limiter (WFRxRateLimiter). The HTTP rate limiter restricts the number of HTTP
	requests that can run in parallel. This means that an HTTP request may end up
	sitting in the rate limiter's queue for some time. Please see WFRxHttpClientImp
	for what settings are used to initialize the HTTP rate limiter.

	Once we receive a notification that the binary upload HTTP request has started, we
	identify the next highest priority binary and issue a new put request. We do this
	repeatedly so that there will always be at least one upload request in the HTTP
	rate limiter queue. In other words, we always issue N+1 binary upload put requests,
	where N is the number of in-progress binary upload HTTP requests.

	If ever the binary upload in the HTTP rate limiter queue is no longer the top priority,
	the upload is cancelled by the WFCatBinaryUploader and a new request is issued in it's
	place.

	This approach has the same, measurable performance as the previous logic and it
	guarantees than the highest priority binary is always uploaded immeadiately after
	the current, in-progress binary upload(s) complete.

--]]
function WFCatBinaryUploader:uploadSomeBinaries()

	return self:getHighestPriorityBinaryToUploadStream()

		:map( function( binaryToUpload )

			local docId = binaryToUpload[ 1 ]
			local localFilePath = binaryToUpload[ 2 ]
			local binaryUrl = binaryToUpload[ 3 ]
			local binaryType = binaryToUpload[ 4 ]
			local contentType = binaryToUpload[ 5 ]
			local encodedUploadHeaders = binaryToUpload[ 6 ]
			local sha256 = binaryToUpload[ 8 ]

			-- NOTE: @sewillia 2022-07-20 - Because an
			-- asset document can have more than one
			-- binary associated with it, there is a need
			-- to keep track of auxiliary sha256 more
			-- granularly than the one by one mechanism of
			-- "getRecentlyUploadedSha256".

			local binaryDetails = newBinaryDetail( binaryType, sha256 )

			local localFileSize = binaryToUpload[ 9 ]
			local sha256Error = intToBool( binaryToUpload[ 10 ] )
			local sourceLinkForCopy = binaryToUpload[ 11 ]
			local uploadPriority = binaryToUpload[ 14 ]
			local spaceId = binaryToUpload[ 15 ]
			local mainPartsRequestUrl = binaryToUpload[ 16 ]
			local copyUrl = binaryToUpload[ 17 ]
			local auxId = binaryToUpload[ 18 ]
			local parentDocumentId = binaryToUpload[ 19 ]
			local developUserUpdated = binaryToUpload[ 20 ]

			local uploadHeaders
			if encodedUploadHeaders then
				uploadHeaders = WFMsgPack.unpack( encodedUploadHeaders )
			end

			if localFilePath and ( not localFileSize or localFileSize <= 0 ) then
				local fileAttributes = WFFileUtils.fileAttributes( localFilePath )
				localFileSize = fileAttributes.fileSize
			end

			local isFirstUploadForSha256 = self:getUploadingAssetsWithSha256( sha256 ) == 0

			self:addToRecentlyUploaded( docId, binaryDetails, sha256 or true )

			local subscription = SingleAssignmentDisposable()

			local isCopy = not not sourceLinkForCopy

			self:addToScheduledUploads( docId, binaryDetails, subscription, localFileSize, uploadPriority, isCopy )

			local isDirectUpload = false
			local copyBinaryUrl = copyUrl or binaryUrl

			local directUploadFailureCountOK = self:checkDirectUploadFailureCount()

			if isXmpDevelop( binaryType ) and mainPartsRequestUrl and
				 not self:isAtomicDevelopSettingsEnabled() then

				-- disable direct upload for xmp_develop unless capability is enabled
				mainPartsRequestUrl = nil

			end

			if ( self._fallbackToOzUploadForOriginalsOnly and isOriginalBinary( binaryType ) ) or
					mainPartsRequestUrl == kMainPartsRequestUrlNotAvailable or
					mainPartsRequestUrl == kMainPartsRequestUrlMetadataAlreadySynced then

				-- Note that the default is isDirectUpload == false, so this is a noop.

			elseif mainPartsRequestUrl and ( directUploadFailureCountOK or not binaryUrl ) then

				-- Use Direct Upload

				isDirectUpload = true
				binaryUrl = mainPartsRequestUrl

			end

			self:trackUploadForDiagnostics( spaceId, binaryDetails, not not sourceLinkForCopy, isDirectUpload )

			local function removeUploadSubscription()

				-- Don't call uploadFinished from finally as it could be the result of
				-- dispose being called from disposeActiveUploads and we don't want to
				-- start other uploads in this case.
				self:removeFromScheduledUploads( docId, binaryDetails )

			end

			local function uploadCompletedOrError()

				-- Call removeFromActiveUploads here (in addition to the finally) so that _numberOfScheduledBinaryUploads
				-- is reduced so the following uploadFinished call can trigger another upload as
				-- finally is called after the completed not before it.
				self:removeFromScheduledUploads( docId, binaryDetails )

				-- Trigger combineLatest above to start other uploads.
				self:uploadFinished( docId, binaryDetails )

				if self._numberOfScheduledBinaryUploads == 0 then
					self._idleStatusStream:onNext( true )
				end

			end

			local action

			if sha256Error then

				local function delayedClearFromRecentlyUploadedFn()
					self:delayedClearFromRecentlyUploaded( "waiting for resolveSha256ErrorInInputSource doc update", docId, binaryDetails )
				end

				-- Handle asynchronous error from a prior chunked upload that is sent via catalog metadata.
				action = self:resolveSha256ErrorInInputSource( binaryUrl, localFilePath, docId, sha256 )
					:finally( delayedClearFromRecentlyUploadedFn )

			elseif sourceLinkForCopy ~= nil then

				-- Asset was copied and the original is already up on oz. Using the copy rather
				-- than the local binary should be preferred because it will almost always be --
				-- faster than uploading the file directly.

				action = self:copyOneBinary(
						docId, copyBinaryUrl, binaryDetails, sha256,
						contentType, uploadHeaders, sourceLinkForCopy,
						nil, nil, auxId )

					:catch( function( responseData, headers )

						local resourceDoesNotExist, newId = errorAndMaybeNewIdIfMasterOrProxyResourceDoesNotExist( binaryDetails.type, responseData, headers )

						if resourceDoesNotExist then

							return self:applyOrRemoveNewSourceLinksOrErrorOverlayForCopy( docId, newId, binaryDetails, copyBinaryUrl )

								:map( function( appliedOverlay )
									if appliedOverlay then
										self:delayedClearFromRecentlyUploaded( "updated or removed sourceAsset for copy", docId, binaryDetails )
									else
										self:handleUploadFailure( docId, binaryDetails, copyBinaryUrl, parentDocumentId,
												responseData, headers, false, contentType, localFileSize, true, false )
									end
								end )

						end

						self:handleUploadFailure( docId, binaryDetails, copyBinaryUrl, parentDocumentId,
								responseData, headers, false, contentType, localFileSize, true, false )

						return Observable.throw( responseData, headers )

					end )

			elseif isOriginalBinary( binaryDetails.type ) or isProxy( binaryDetails.type ) then

				-- proxy addresses: lightroom asset is copied while paused, then the
				-- copy is copied (results in no sourceLinkForCopy)
				action = self:uploadOrCopyOneMasterOrProxy(
						docId, localFilePath, localFileSize, binaryUrl,
						copyBinaryUrl, binaryDetails, contentType, uploadHeaders,
						sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload, auxId )

			else

				action = self:uploadOneBinary(
						docId, localFilePath, localFileSize, binaryUrl,
						binaryDetails, contentType, uploadHeaders, sha256,
						uploadPriority, isDirectUpload, developUserUpdated, parentDocumentId )
			end

			subscription:setDisposable(
				action
					:finally( removeUploadSubscription )
					:subscribe( nil, uploadCompletedOrError, uploadCompletedOrError )
			)

			self._continuousUploadTask:taskDidWork()

			return true

		end )

		:catch( function( errorMessage )

			logger:debugf( "uploadSomeBinaries %s", errorMessage )

			if self._numberOfScheduledBinaryUploads > 0 then

				-- Wait until all the in-flight requests have finished.

				return self._idleStatusStream
					:take( 1 )
					:flatMap( function()

						self._continuousUploadTask:taskDidWork( false )
						return self._continuousUploadTask:throwIfNotStopped( errorMessage )

					end )

			else

				self._continuousUploadTask:taskDidWork( false )
				return self._continuousUploadTask:throwIfNotStopped( errorMessage )

			end

		end )

end

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

--[[

	Instance Function

	\brief Tears down the WFCatBinaryUploader instance

]]
function WFCatBinaryUploader:close()

	check( 'WFCatBinaryUploader' )

	self:disposeAllActiveUploads()

	local errorStatus = self._errorStatusStream
	self._errorStatusStream = nil
	errorStatus:dispose()

	local continuousUploadTask = self._continuousUploadTask
	self._continuousUploadTask = nil

	if continuousUploadTask then
		continuousUploadTask:close()
	end

	local failedUploadDeniedListHelper = self._failedUploadDeniedListHelper
	self._failedUploadDeniedListHelper = nil

	self._uploadBlackoutWindowInSeconds = false

	if failedUploadDeniedListHelper then
		failedUploadDeniedListHelper:close()
	end

	local disposeActiveUploadsWhenDeletedSubscription = self._disposeActiveUploadsWhenDeletedSubscription
	self._disposeActiveUploadsWhenDeletedSubscription = nil

	if disposeActiveUploadsWhenDeletedSubscription then
		disposeActiveUploadsWhenDeletedSubscription:dispose()
	end

	local uploadsByTypeStream = self._uploadsByTypeStream
	self._uploadsByTypeStream = nil

	if uploadsByTypeStream then
		uploadsByTypeStream:dispose()
	end

	local uploadScheduledStream = self._uploadScheduledStream
	self._uploadScheduledStream = nil

	if uploadScheduledStream then
		uploadScheduledStream:dispose()
	end

	local inFlightUploadProgressUpdateStream = self._debugInFlightUploadProgressUpdateStream
	self._debugInFlightUploadProgressUpdateStream = nil

	if inFlightUploadProgressUpdateStream then
		inFlightUploadProgressUpdateStream:dispose()
	end

	local reloadLinksForDocumentSubscription = self._reloadLinksForDocumentSubscription
	self._reloadLinksForDocumentSubscription = nil

	if reloadLinksForDocumentSubscription then
		reloadLinksForDocumentSubscription:dispose()
	end

	local reloadLinksForDocumentStream = self._reloadLinksForDocumentStream
	self._reloadLinksForDocumentStream = nil

	if reloadLinksForDocumentStream then
		reloadLinksForDocumentStream:dispose()
	end

	local shouldReprocessVersionsStream = self._shouldReprocessVersionsStream
	self._shouldReprocessVersionsStream = nil

	if shouldReprocessVersionsStream then
		shouldReprocessVersionsStream:dispose()
	end

end

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

--[[

	Instance Function

	\brief Halts upload loop

]]
function WFCatBinaryUploader:stopContinuousUploads()

	self._continuousUploadTask:stopContinuousTask()

	self:disposeAllActiveUploads()

	self:uploadFinished()
		-- trigger combineLatest in uploadSomeBinaries so it can see it is stopped.

end

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

--[[

	Instance Function

	\brief Starts the upload loop

]]

function WFCatBinaryUploader:startContinuousUploads()

	if self._onlyUploadSpaceAssets then
		self:stopContinuousUploads()
	end

	self._onlyUploadSpaceAssets = false

	self._continuousUploadTask:startContinuousTask()

end

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

--[[

	Instance Function

	\brief Starts the upload loop only for space assets

]]
function WFCatBinaryUploader:startContinuousUploadOfOnlySpaceAssets()

	if not self._onlyUploadSpaceAssets then
		self:stopContinuousUploads()
	end

	self._onlyUploadSpaceAssets = true

	self._continuousUploadTask:startContinuousTask()

end

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

--[[

	Instance Function

	\brief Enables background uploads

]]
function WFCatBinaryUploader:enableBackgroundUpload( enable )

	self._allowBackgroundUpload:onNext( enable )

end

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

--[[

	Instance Function

	\brief Yields a handle to the error status stream

	\return Observable<error status stream>

]]
function WFCatBinaryUploader:errorStatus()

	return self._errorStatusStream

end

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

--[[

	Instance Function

	\brief Quick check to see if the uploader is currently doing anything

	\retun bool

]]
function WFCatBinaryUploader:isDoingWork()

	return self._continuousUploadTask:isTaskDoingWork()

end


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

--[[

	Instance Function

	\brief Set the work status of the uploader - only used in test or debugging

	\return void

]]
function WFCatBinaryUploader:DEBUG_taskDidWork( optDidWork )

	self._continuousUploadTask:taskDidWork( optDidWork )

end

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

--[[

	Instance Function

	\brief Yields a handle to the upload finished stream - only used in test or debugging

	\return Observable<upload finished stream>

]]
function WFCatBinaryUploader:DEBUG_uploadFinishedStream()

	return self._uploadFinishedStream

end

--------------------------------------------------------------------------------
--[[!

	\brief Report the status of the Nimbus Volumes

	\return (void)

--]]
function WFCatBinaryUploader:reportStatusOfVolumes( statusOfVolumes )

	check( 'WFCatBinaryUploader', 'table' )

	-- When a volume's status changes, namely when a volume is reconnected
	-- to Nimbus, there's a chance that we'll be able to upload binaries
	-- that were previously not accessible.

	-- The query WFCatBinaryUploader uses does not filter out assets that
	-- are on offline volumes, unlike WFCatIdleDownloader, because we also
	-- support uploading "copies" that have no local file.

	-- Instead, we prioritize uploading the files that are known to be connected,
	-- as well as copies, and then check to see if the possibly-disconnected
	-- files can be accessed and uploaded.

	-- The following statement guarantees we process the disconnected -> connected
	-- files once a volume comes online. The value here is unimportant, but making
	-- it unique can help in debugging.

	self._volumeStatusHasChangedStream:onNext( KSDate.timeInIso8601Format() )

end

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

--[[!

	\brief Fires every time a binary upload sucessfully completes.

	\return <code>Observable< nil ></code> that fires every time a binary upload
			sucessfully completes.

--]]

function WFCatBinaryUploader:binaryUploadCompletionStream()

	check( 'WFCatBinaryUploader' )

	return self._binaryUploadCompletionStream

end

--------------------------------------------------------------------------------
--[[!

	\brief Fires every time a binary upload is scheduled

	\return <code>Observable< nil ></code> that fires every time a binary upload
			is in progress and also fires when the priority of a scheduled (in progress or not) upload changes

--]]

function WFCatBinaryUploader:uploadStatusChangeStream()

	check( 'WFCatBinaryUploader' )

	return self._uploadStatusChangeStream

end

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


--[[

	Instance Function

	\brief Marks a specific upload job as complete and emit a
	unique value on the uploadFinishedStream if it is.

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(binaryDetails - domain construct)

	\return nil

]]
function WFCatBinaryUploader:uploadFinished( docId, binaryDetails )

	if self._uploadFinishedStream then
		-- Schedule the call to _uploadFinishedStream to prevent stack overflow.
		WFRxScheduler.sharedScheduler():schedule( function()
			-- the value here is unimportant, but making it unique can help in debugging

			-- It is possible for uploadFinished to be called with no parameters
			self._uploadFinishedStream:onNext({
				completeAt = KSDate.timeInIso8601Format(),
				docId = docId or '<noDocId>',
				binaryDetails = binaryDetails or {}
			})
		end )

	end

end

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

--[[

	Instance Function

	\brief Resets the currently scheduled upload.

	\return nil

]]
function WFCatBinaryUploader:clearScheduledAndWaitingBinaryUpload()

	if self._scheduledAndWaitingBinaryUpload and
			self._scheduledAndWaitingBinaryUpload.priorityTracker then

		self._scheduledAndWaitingBinaryUpload.priorityTracker:dispose()

	end

	self._scheduledAndWaitingBinaryUpload = nil

end

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

--[[

	Instance Function

	\brief initializes the WFCatBinaryUploader's priority tracker.
		TODO: Add more detail about this mechanism

	\return nil

]]
function WFCatBinaryUploader:setupBinaryUploadPriorityTracker()

	if self._scheduledAndWaitingBinaryUpload and
			not self._scheduledAndWaitingBinaryUpload.priorityTracker then

		self._scheduledAndWaitingBinaryUpload.priorityTracker = self._catalog:indexStore():query {
				sql = [[SELECT priorityForUpload
							FROM needsBinaryUpload
							WHERE docId = ? AND binaryType = ?]],
				singleColumnQuery = true,
				self._scheduledAndWaitingBinaryUpload.docId,
				self._scheduledAndWaitingBinaryUpload.binaryDetails.type
			}

			:pluck( 1 )

			:subscribe( function( priorityForUpload )

				if self._scheduledAndWaitingBinaryUpload then

					self._scheduledAndWaitingBinaryUpload.priority = priorityForUpload

					WFRxScheduler.sharedScheduler():schedule( function()

						-- Schedule the call to _uploadStatusChangeStream to prevent stack overflow.
						-- the value here is unimportant, but making it unique can help in debugging
						self._uploadStatusChangeStream:onNext( KSDate.timeInIso8601Format() )

					end )

				end

			end )

	end

end


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

--[[

	Instance Function

	\brief Set an upload for a particular document as "in progress"

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file

	\return nil

]]
function WFCatBinaryUploader:markUploadAsInProgress( docId, binaryDetails )

	if not self._scheduledAndWaitingBinaryUpload or
		self._scheduledAndWaitingBinaryUpload.docId ~= docId or
		self._scheduledAndWaitingBinaryUpload.binaryDetails ~= binaryDetails then

		logger:errorf( "The latest scheduled-and-waiting binary is not what we expected." )

	end

	logger:infof( "Binary upload %s:%s:%s is now in-flight", docId, binaryDetails.type, binaryDetails.sha256 )

	self:clearScheduledAndWaitingBinaryUpload()

	WFRxScheduler.sharedScheduler():schedule( function()

		-- Schedule the call to _uploadStatusChangeStream to prevent stack overflow.
		-- the value here is unimportant, but making it unique can help in debugging
		self._uploadStatusChangeStream:onNext( KSDate.timeInIso8601Format() )

	end )

end

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

--[[

	Instance Function

	\brief checks if the uploader should fall back to legacy upload

	\return bool

]]
function WFCatBinaryUploader:checkDirectUploadFailureCount()

    return self._directUploadConsecutiveFailureCounter < numberOfTriesBeforeRevertToLegacyUpload

end

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

--[[

	Instance Function

	\brief Used by tests to check _directUploadConsecutiveFailureCounter. Note
	that if we ever remove the direct upload fallback code and remove this
	variable, we will have to update the test to have some other way of
	checking whether there was a direct upload error.

	\return number

]]
function WFCatBinaryUploader:DEBUG_getDirectUploadFailureCountCounter()

    return self._directUploadConsecutiveFailureCounter

end

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

--[[

	Instance Function

	\brief Used by tests to set _fallbackToOzUploadForOriginalsOnly to force into
		fallback mode and NOT do direct upload for original binaries only.

	\param boolean

]]
function WFCatBinaryUploader:DEBUG_setFallbackToOzUploadForOriginalsOnly( value )

    self._fallbackToOzUploadForOriginalsOnly = value

end

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

--[[

	Instance Function

	\brief Used by tests to check _fallbackToOzUploadForOriginalsOnly value to see
		if uploader has fallen back to direct oz upload for original binaries.

	\return boolean

]]
function WFCatBinaryUploader:DEBUG_getFallbackToOzUploadForOriginalsOnly()

    return self._fallbackToOzUploadForOriginalsOnly

end

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

--[[

	Instance Function

	\brief Initializes an upload and provides a mechanism to track the status of the upload

	\param uploadUrl	(string) - Destination for the upload
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param filePath		(string) - Path to the binary on this device
	\param fileSize		(int)    - The size of the binary
	\param options		(table)  - Extra configuration parameters
	\param docId		(string) - The ID of the Document an upload is associated with
	\param isDirectUpload	(bool)   - Is this a DCU upload or not?

	\return Observable containing the upload status

]]
function WFCatBinaryUploader:putAndTrackPerformance( uploadUrl, binaryDetails, filePath, fileSize, options,
		docId, isDirectUpload, optDevelopUserUpdated )

	local putFile

	options.notifyWhenHttpRequestIsInProgress = true
	options.includeProgress = self._includeProgress

	local requestInProgress = false

	local function progressEventHandler( _, _, progress )

		if not requestInProgress then

			self:markUploadAsInProgress( docId, binaryDetails )
			requestInProgress = true

		end

		if progress then

			setNested( self._inFlightUploadProgress, progress, docId, binaryDetails.type, binaryDetails.sha256 )
			self._debugInFlightUploadProgressUpdateStream:onNext( self._inFlightUploadProgress )

		end

	end

	local function putPartRequestSuccessHandler()
		self._directUploadConsecutiveFailureCounter = 0
	end

	if isDirectUpload then

		local ozExistenceCheckFn = nil
		local developUserUpdated

		if isDevelopBinary( binaryDetails.type ) then

			-- when dealing with aux and xmp_develop,  binaries, Oz changes its
			-- response depending on if they already have the binary or not.
			-- This callback function provides a way to short circut an attempt
			-- to call rels/complete_upload in the case Oz already has this file

			ozExistenceCheckFn = function( mainPartsResponse )

				if derefOpt( mainPartsResponse, 'file_status') == 'uploaded' then
					return true
				end

				return false
			end

			-- for direct upload of aux and xmp_develop developUserUpdated is
			-- required so set it to current date and time if missing

			developUserUpdated = optDevelopUserUpdated or KSDate.timeInIso8601FormatNearestSecond()

		end

		putFile = self._directChunkedUploader:doDirectChunkedUpload( uploadUrl, filePath, developUserUpdated, options,
				progressEventHandler, putPartRequestSuccessHandler, ozExistenceCheckFn )

			:tap( progressEventHandler )

			:takeLast( 1 )

			:catch( function( responseData, headers, url )

				-- Fallback to Oz upload is only for original binary upload so do not increase
				-- the count if other file types have errors.

				if isOriginalBinary( binaryDetails.type ) then

					-- WF-1753 we only track consecutive failures; if the consecutive http request fails more than 3 times, we set
					--  and it means falling back to oz-upload

					self._directUploadConsecutiveFailureCounter = self._directUploadConsecutiveFailureCounter + 1

					local headerDetail = headers and headers.status_code or 'unknown'

					local responseDataStr = caughtResponseDataToString( responseData )

					local responseDetail = string.replace( responseDataStr, '%', '%%' )

					-- sewillia 2023-08-18 TODO: I think that this reassignment is a type error. Check other instances of responseData
					responseData = string.format("Can't use direct upload with %s error and status codes is %s", responseDetail, headerDetail )
					logger:errorf( responseData )

				end

				local requestInfo = {
					errorUrl = url
				}

				return Observable.throw( responseData, headers, requestInfo )

			end )


	elseif isOriginalBinary( binaryDetails.type ) and fileSize >= self._chunkedUploader:chunkFileSize() then

		putFile = self._chunkedUploader:doChunkedUpload( uploadUrl, filePath, options, progressEventHandler )

			:tap( progressEventHandler )

			:takeLast( 1 )

	else

		-- WF-1716, should be temporary until we can do chunked/direct uploads for xmp_develop too
		if isXmpDevelop( binaryDetails.type ) and fileSize >= kOzMaximumXmpDevelopSize then

			putFile = Observable.throw( "WF-1716 prevent futile upload", { status_code = 413 } )

		else

			putFile = self._catalogSpec:putBinary( uploadUrl, filePath, options )

				:tap( progressEventHandler )

				:takeLast( 1 )

		end

	end

	return WFPerformanceReport.wrapObservable( string.format( 'WFCatBinaryUploader: %s', binaryDetails.type ),
			putFile, { fileSize = fileSize } )

end

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

--[[

	Instance Function

	\brief returns what sha256 has recently been uploaded for a given type

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryType	(string) - The type of the binary, refer to kBinaryType... values

	\return table

]]
function WFCatBinaryUploader:getRecentlyUploadedSha256List( docId, binaryType )

	return derefOpt( self._recentlyUploaded, docId, binaryType ) or {}

end

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

--[[

	Instance Function

	\brief Check if there is an in flight upload for a given sha256

	\param sha256        (string) - Sha256 of the uploaded binary

	\return bool

]]
function WFCatBinaryUploader:getUploadingAssetsWithSha256( sha256 )

	return sha256 and self._uploadingAssetWithSha256[ sha256 ] or 0

end

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

--[[

	Instance Function

	\brief Return the number of remaining assets that we are attempting to upload
		with the specified sha256. Only called with masters.

	\param self		(WFCatBinaryuploader)
	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file

	\return int


]]
function WFCatBinaryUploader:decrementUploadingAssetWithSha256( docId, binaryDetails )

	-- Return the number of remaining assets that we are attempting to upload
	-- with the specified sha256. Only called with masters.

	if isOriginalBinary( binaryDetails.type ) then

		local sha256 = binaryDetails.sha256

		if self._uploadingAssetWithSha256[ sha256 ] then

			-- _uploadingAssetWithSha256 may get cleared if an error occurs.
			-- If the count goes negative it shouldn't hurt as the most timers
			-- we will queue up for a sha256 is 100 as limited by the query.
			local newCountOfAssets = self._uploadingAssetWithSha256[ sha256 ] or 0
			newCountOfAssets = newCountOfAssets - 1

			if newCountOfAssets > 0 then
				self._uploadingAssetWithSha256[ sha256 ] = newCountOfAssets
			else
				self._uploadingAssetWithSha256[ sha256 ] = nil
				newCountOfAssets = 0
			end

			return newCountOfAssets

		end

	end

	return -1

end

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

--[[

	Instance Function

	\brief Updates internal tracking variables for diagnostic purposes

	\param spaceId           (string) - Space ID
	\param binaryDetails (table) - A BinaryDetails domain object - see top of file
	\param wasCopied         (bool) - Was this upload caused by a copy?
	\param wasDirectS3Upload (bool) - Was this upload a DCU upload?

	\return table or nil
		Returns either the current _uploadsByType table OR causes the _uploadsByType stream to be updated

]]
function WFCatBinaryUploader:trackUploadForDiagnostics( spaceId, binaryDetails, wasCopied, wasDirectS3Upload )

	local function updateUploadCounts( typeOfBinary )

		local spaceIdOrCatalog = spaceId or 'catalog'

		local total, ozCopies, s3Uploads

		-- Catalog/Space Level

		total = derefOpt( self._uploadsByType, spaceIdOrCatalog, 'total' ) or 0
		setNested( self._uploadsByType, total + 1, spaceIdOrCatalog, 'total' )

		if wasCopied then
			ozCopies = derefOpt( self._uploadsByType, spaceIdOrCatalog, kOzCopies ) or 0
			setNested( self._uploadsByType, ozCopies + 1, spaceIdOrCatalog, kOzCopies )
		end

		if wasDirectS3Upload and not wasCopied then
			s3Uploads = derefOpt( self._uploadsByType, spaceIdOrCatalog, kS3Uploads ) or 0
			setNested( self._uploadsByType, s3Uploads + 1, spaceIdOrCatalog, kS3Uploads )
		end

		-- Binary Type Level

		total = derefOpt( self._uploadsByType, spaceIdOrCatalog, binaryDetails.type, 'total' ) or 0
		setNested( self._uploadsByType, total + 1, spaceIdOrCatalog, binaryDetails.type, 'total' )

		if wasCopied then
			ozCopies = derefOpt( self._uploadsByType, spaceIdOrCatalog, binaryDetails.type, kOzCopies ) or 0
			setNested( self._uploadsByType, ozCopies + 1, spaceIdOrCatalog, binaryDetails.type, kOzCopies )
		end

		if wasDirectS3Upload and not wasCopied then
			s3Uploads = derefOpt( self._uploadsByType, spaceIdOrCatalog, binaryDetails.type, kS3Uploads ) or 0
			setNested( self._uploadsByType, s3Uploads + 1, spaceIdOrCatalog, binaryDetails.type, kS3Uploads )
		end

		return self._uploadsByType

	end

	if self._uploadsByTypeStream then

		self._uploadsByType = updateUploadCounts( binaryDetails )
		self._uploadsByTypeStream:onNext( self._uploadsByType )

	end

end

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

--[[

	Instance Function

	\brief Adds an upload to the _recentlyUploaded tracking variable

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param sha256		(string) - Sha256 of the uploaded binary

	\return nil

]]
function WFCatBinaryUploader:addToRecentlyUploaded( docId, binaryDetails, sha256 )

	local recentlyUploadedForDocId = self._recentlyUploaded[ docId ] or {}

	if derefOpt( recentlyUploadedForDocId, binaryDetails.type, sha256 or binaryDetails.sha256 ) then

		WFAnalytics.trackEvent {
			['event.type'] = 'info',
			['event.subtype'] = binaryDetails.type .. "-uploaded",
			['event.appWorkflow'] = 'SYNC',
			['event.subcategory'] = 'Binary_upload_duplicate_schedule',
			['content.id'] = docId,
			['sha256'] = binaryDetails.sha256
		}

		logger:errorf( "addToRecentlyUploaded recentlyUploadedForDocId ALREADY SET for %s %s", binaryDetails.type, docId )

	end

	setNested( recentlyUploadedForDocId, true, binaryDetails.type, binaryDetails.sha256 or sha256 )

	self._recentlyUploaded[ docId ] = recentlyUploadedForDocId

	if isOriginalBinary( binaryDetails.type ) then
		local countOfAssets = self._uploadingAssetWithSha256[ sha256 ] or 0
		self._uploadingAssetWithSha256[ sha256 ] = countOfAssets + 1
	end

end

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

--[[

	Instance Function

	\brief Removes a specified binary upload from the recent uploads table

	\param reason				(string) - Why is this Binary being cleared?
	\param docId				(string) - The ID of the Document an upload is associated with
	\param binaryDetails			(table) - A BinaryDetails domain object - see top of file
	\param uploadAttemptedOrCancelled	(bool) - Was the upload attempted or cancelled in flight?

	\return nil

]]
function WFCatBinaryUploader:clearFromRecentlyUploaded( reason, docId, binaryDetails, uploadAttemptedOrCancelled )

	logger:debugf( "clearFromRecentlyUploaded %s %s %s because %s", binaryDetails.type, binaryDetails.sha256, docId, reason )

	local recentlyUploadedForDocId = self._recentlyUploaded[ docId ] or {}

	-- Note that the call to decrementUploadingAssetWithSha256 is placed inside here and
	-- controlled by a parameter because it is dependent on _recentlyUploaded
	-- In other words the ordering of when it is called must take place before
	-- _recentlyUploaded is cleared.
	if uploadAttemptedOrCancelled then
		self:decrementUploadingAssetWithSha256( docId, binaryDetails )
	end

	setNested( recentlyUploadedForDocId, nil, binaryDetails.type, binaryDetails.sha256 )

	if next( recentlyUploadedForDocId ) then
		self._recentlyUploaded[ docId ] = recentlyUploadedForDocId
	else
		self._recentlyUploaded[ docId ] = nil
	end

end

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

--[[

	Instance Function

	\brief If there are no more Binaries to upload associated with
		a document - make sure its priority is reset

	\param docId (string) - The ID of the Document an upload is associated with

	\return nil

]]
function WFCatBinaryUploader:maybeClearUploadPriorityModifier( docId )

	return self._catalog:indexStore():queryOnce {
			sql = [[SELECT count( docId )
					FROM needsBinaryUpload
					WHERE docId = ? ]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true,
			docId
		}

		:flatMap( function( row )

			local count = row[ 1 ]

			if count == 0 then

				logger:debugf( "Clearing upload priority modifier from document %s.", docId )

				return WFCatPhotoImportance.removeUploadPriorityModifier( self._catalog, docId )

			else

				logger:debugf( "Not clearing upload priority modifier from document %s because there are more binaries to upload.", docId )
				return Observable.just( true )

			end

		end )

		:call( runUntilDone )

end

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

--[[

	Instance Function

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

	TODO: Is there a more deterministic way to do this? An index
	store API that fires when it's up to date?

	\param reason		(string) - Why is this Binary being cleared?
	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param optDelay		(int) - Optional override for the class default

	\return nil

]]
function WFCatBinaryUploader:delayedClearFromRecentlyUploaded( reason, docId, binaryDetails, optDelay )

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

	-- TODO: Is there a more deterministic way to do this? An index store API that fires
	-- when it's up to date?

	local indexStoreUpdateBuffer = optDelay or self._clearFromRecentlyUploadedSeconds

	logger:debugf( "Calling clearFromRecentlyUploaded in %d seconds for doc %s type %s because %s.", indexStoreUpdateBuffer, docId, binaryDetails.type, reason )

	return self._timerAborter
		:flatMapLatest( function()
			return self._timer( indexStoreUpdateBuffer )
		end )
		:tap( function()

			self:clearFromRecentlyUploaded( reason, docId, binaryDetails, true )

			-- Trigger combineLatest in uploadSomeBinaries as we may need to try uploading docId again
			self:uploadFinished( docId, binaryDetails )

		end )
		:call( runUntilDone )

end

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

--[[

	Instance Function

	\brief Clears asset from recently uploaded without issuing a
	uploadFinished call unless it is the last asset related to the
	sha256. This is used to prevent spamming the uploader with
	requests to upload something new when we didn't even try to
	upload anything and didn't mark anything as invalid. It is
	used when we want to have assets wait for the first asset with
	the sha to upload so we can issue a copy on the subsequent
	assets.

	\param reason		(string) - Why is this Binary being cleared?
	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param optDelay		(int) - Optional override for the class default

	\return nil

]]
function WFCatBinaryUploader:delayedClearFromRecentlyUploaded_NoUpload( reason, docId, binaryDetails, optDelay )

	-- Clears asset from recently uploaded without issuing a uploadFinished call
	-- unless it is the last asset related to the sha256.

	-- This is used to prevent spamming the uploader with requests to upload
	-- something new when we didn't even try to upload anything and didn't
	-- mark anything as invalid.

	-- It is used when we want to have assets wait for the first asset with
	-- the sha to upload so we can issue a copy on the subsequent assets.

	-- Decrement this immediately as we are not uploading, just delaying
	-- the next attempt as we are waiting for the first asset with the sha256
	-- to complete it's upload attempt.

	self:decrementUploadingAssetWithSha256( docId, binaryDetails )

	local secondsToDelay = optDelay or self._waitForOtherUploadSeconds

	logger:debugf( "Calling clearFromRecentlyUploaded in %d seconds for doc %s type %s because %s.", secondsToDelay, docId, binaryDetails.type, reason )

	local sha256 = binaryDetails.sha256

	return self._timerAborter
		:flatMapLatest( function()

			if sha256 then
				local numAssetsWaitingForUpload = self._numAssetsWaitingForUploadWithSha256[ sha256 ] or 0
				self._numAssetsWaitingForUploadWithSha256[ sha256 ] = numAssetsWaitingForUpload + 1
			end

			return self._timer( secondsToDelay )

		end )
		:tap( function()

			self:clearFromRecentlyUploaded( reason, docId, binaryDetails, false )

			if sha256 then

				local numAssetsWaitingForUpload = self._numAssetsWaitingForUploadWithSha256[ sha256 ] or 0
				numAssetsWaitingForUpload = numAssetsWaitingForUpload - 1
				self._numAssetsWaitingForUploadWithSha256[ sha256 ] = numAssetsWaitingForUpload > 0 and numAssetsWaitingForUpload or nil

				if numAssetsWaitingForUpload == 0 then
					-- Trigger combineLatest in uploadSomeBinaries as we will need
					-- to upload the copies that were waiting.
					self:uploadFinished( docId, binaryDetails )
				end

			end

		end, function( err )
			-- clear when aborted
			self._numAssetsWaitingForUploadWithSha256[ sha256 ] = nil
		end )
		:call( runUntilDone )

end

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

--[[

	Instance Function

	\brief Issue an update to the instance's progressReporter

	\param operation (string) - The name of the action to be amended
	\param fileSize  (int) - The size of the binary being uploaded

	\return nil

]]
function WFCatBinaryUploader:uploadProgressStatusUpdate( operation, fileSize )

	check( 'WFCatBinaryUploader', '?string' )

	if not self._uploadProgress or not self._uploadProgress.progressReporter or not fileSize then
		return
	end

	if operation == "increaseBytesNeedToUpload" then

		-- Addition of asset to Active uploads
		self._uploadProgress.totalBytesNeedToUpload = self._uploadProgress.totalBytesNeedToUpload + fileSize

	elseif operation == "increaseBytesUploaded" then

		self._uploadProgress.totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar + fileSize

	else
		return
	end

	if( self._uploadProgress.totalBytesNeedToUpload - self._uploadProgress.totalBytesUploadedSoFar  > 0 ) then

		local binaryUploadProgress = {
			totalBytesNeedToUpload = self._uploadProgress.totalBytesNeedToUpload,
			totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar,
		}

		self._uploadProgress.progressReporter:updateProgressWithInfo( binaryUploadProgress, true )

	end

end

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

--[[

	Instance Function

	\brief Adds a binary to the scheduled uploads
	TODO: expand on what the difference between a scheduled upload and regular upload are

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param subscription	(observable) - The subscriber to notify once the upload is complete
	\param localFileSize	(int) - The size of the Binary to upload
	\param priority		(int) - The priority of this upload

	\return nil

]]
function WFCatBinaryUploader:addToScheduledUploads( docId, binaryDetails, subscription, localFileSize, priority )

	self._numberOfScheduledBinaryUploads = self._numberOfScheduledBinaryUploads + 1

	local uploadSubscriptionsForDocId = self._scheduledUploadSubscriptions[ docId ] or {}

	if uploadSubscriptionsForDocId[ binaryDetails ] then
		logger:errorf( "Already have subscription for uploading %s type = %s sha = %s", docId, binaryDetails.type, binaryDetails.sha256 )
	else
		self:uploadProgressStatusUpdate( "increaseBytesNeedToUpload", localFileSize )
	end

	if self._scheduledAndWaitingBinaryUpload then
		logger:errorf( "There is already a scheduled-and-waiting binary. Something went wrong." )
	end

	logger:infof( "Now scheduling binary upload %s:%s:%s", docId, binaryDetails.type, binaryDetails.sha256 )

	self._scheduledAndWaitingBinaryUpload = {
		docId = docId,
		binaryDetails = binaryDetails,
		subscription = subscription,
		priority = priority
	}

	self:setupBinaryUploadPriorityTracker()

	uploadSubscriptionsForDocId[ binaryDetails ] = subscription

	self._scheduledUploadSubscriptions[ docId ] = uploadSubscriptionsForDocId

	self._uploadScheduledStream:onNext( KSDate.timeInIso8601Format() )

end

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

--[[

	Instance Function

	\brief Remove an upload from the schedule

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file

	\return nil

]]
function WFCatBinaryUploader:removeFromScheduledUploads( docId, binaryDetails )

	if derefOpt( self._scheduledUploadSubscriptions, docId, binaryDetails ) == nil then
		-- Already removed by completed or error, finally will call a second
		-- time but is needed in case the subscription is disposed before
		-- completed or error.
		return
	end

	if self._scheduledAndWaitingBinaryUpload and
			self._scheduledAndWaitingBinaryUpload.docId == docId and
			self._scheduledAndWaitingBinaryUpload.binaryDetails == binaryDetails then

		-- In the event the scheduled binary upload encountered an error and was
		-- never processed, we must clear it so that we can schedule our next
		-- binary upload.

		self:clearScheduledAndWaitingBinaryUpload()

	end

	local uploadSubscriptionsForDocId = self._scheduledUploadSubscriptions[ docId ] or {}

	uploadSubscriptionsForDocId[ binaryDetails ] = nil

	if next( uploadSubscriptionsForDocId ) then
		self._scheduledUploadSubscriptions[ docId ] = uploadSubscriptionsForDocId
	else
		self._scheduledUploadSubscriptions[ docId ] = nil
	end

	self._numberOfScheduledBinaryUploads = self._numberOfScheduledBinaryUploads - 1

	if self._numberOfScheduledBinaryUploads < 0 then
		logger:errorf( "_numberOfScheduledBinaryUploads went negative == %d", self._numberOfScheduledBinaryUploads )
		self._numberOfScheduledBinaryUploads = 0
	end
	if self._numberOfScheduledBinaryUploads > 0 and next( self._scheduledUploadSubscriptions ) == nil then
		logger:errorf( "_numberOfScheduledBinaryUploads == %d and _scheduledUploadSubscriptions is empty", self._numberOfScheduledBinaryUploads )
	end

end

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

--[[

	Instance Function

	\brief Check if a given document has been deleted on Oz - and if it has, deletes it locally

	\param docId (string) - The ID of the Document an upload is associated with

	\return nil

]]
function WFCatBinaryUploader:checkAndUpdateDocIfDeleted( docId )

	check( 'WFCatBinaryUploader', '+string' )

	return self._docStore:docValue( docId )

		:flatMap( function( docValue )

			local documentUrl = getDocumentUrl( docValue )

			local requestOptions = {
				returnHttpErrorsAsData = true
			}

			return self._catalog:ozCatalogSpec():get( documentUrl, requestOptions )

				:flatMapLatest( function( result, headers )

					if headers and headers.status_code == '404' then

						return WFCatErrorHandler.mimicOzDeletionEvent(
								self._docStore, docId, docValue.type, docValue.subtype )

					else

						return Observable.just( false )

					end

				end )

		end )

		:catch( function( errorMessage )

			logger:errorf( "checkAndUpdateDocIfDeleted, Oz reports error : %s", errorMessage )

			return Observable.just( false )

		end)

		:call( runUntilDone )

end

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

--[[

	Instance Function

	\brief Check if a document URL is sourced from Oz - if it is, transform the document into a WF compatible document

	\param documentUrl (string) - A URL pointing to an Oz Document

	\return Observable which is either the return value from bulkSetDocValue (which emits a table)
			or an observable which simply emits false if bulkSetDocValue was not called.

]]
function WFCatBinaryUploader:checkAndUpdateDocIfOzhasMaster( documentUrl )

	check( 'WFCatBinaryUploader' )

	if ( documentUrl == nil ) then
		return Observable.just( false )
	end

	return self._catalog:ozCatalogSpec():get( documentUrl )
		:flatMapLatest( function( result, headers )

			if headers and 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

					local master_href = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href' )
					if master_href then
						return self._docStore:bulkSetDocValue( { label = 'oz', docValue } )
					else
						return Observable.just( false )
					end

				else
					return Observable.just( false )
				end
			end

		end )
		:catch(function ( errorMessage )
			logger:errorf( "checkAndUpdateDocIfOzhasMaster, Oz reports error : %s", errorMessage )
			return Observable.just( false )
		end)

end

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

--[[

	Instance Function

	\brief Sanity check for edge cases to reconfirm should we really be
		trying to upload this binary to Oz. In some cases might avoid
		upload and in some cases triggers another attempt to upload.

		This function calls checkAndUpdateDocIfOzhasMaster for original binary.
		While we *should* get an update via the metadata puller confirming
		with the link to the original binary, in some edge cases we might
		not so double check to avoid uploading same file to Oz over
		and over.

		For develop binary uploads with atomic develop settings enabled
		this function adds an error annotation to the document to ensure that
		retries are being properly attempted to send the METADATA after the
		binaries upload. (Not needed to ensure that binary retries continue,
		they will as long as confirmation of binary upload success never syncs.)
		While there should be a BinaryMissingError annotation added via the
		WFCatErrorHandler handling of that error, this addiitonal error
		annotation is a second line of defense to try ensure that sufficient
		retries occur after binaries upload. We did see an issue from a
		prerelease user (WF-2410) where it seemed as though a
		BinaryMissingError for some reason did not force retries as expected
		for a version create request, so better safe than sorry.

	\param docValue (table) a complete document
	\param binaryType (string) type of the binary being uploaded

	\return string reason to be send to delayedClearFromRecentlyUploaded

]]
function WFCatBinaryUploader:applyOverlayForBinaryUploadBlackoutWindowEdgeCases( docValue, binaryType )

	local docId = docValue._id

	if isOriginalBinary( binaryType ) then

		return self:checkAndUpdateDocIfOzhasMaster( getDocumentUrl( docValue ) )
			:map( function( result )
				if result then
					logger:infof( "Upload confirmed by manual check for %s", docId )
					return "upload confirmed by manual check"
				else
					logger:warnf( "Assuming upload failed for document %s. Retrying...", docId )
					return "assuming upload failed"
				end
			end )

	elseif isDevelopBinary( binaryType ) and self:isAtomicDevelopSettingsEnabled() then

		-- While it might seem odd to force a retry without a check for errors this code gets hit
		-- when we think we had a successful binary upload but waited around 10 minutes to get a
		-- confirmation from Oz that it was complete and we did not get one, so safer to force the
		-- metadata pusher to retry. Also since maybeMarkDocumentForRetry() function is used,
		-- it is limited to three retries, but then after an hour retries are unlocked as long as
		-- binary upload keeps on being attempted, which it should.

		return WFCatErrorHandler.maybeMarkDocumentForRetry( self._docStore, docValue, 'DirectUploadComplete', 202 )
			:map( function( )
				logger:warnf( "Applied DirectUploadComplete annotation to force metadata push retry on document %s", docId )
				return "Applied DirectUploadComplete annotation to force metadata push retry on document"
			end )

	else

		logger:warnf( "Assuming upload failed for document %s. Retrying...", docId )
		return Observable.just( "assuming upload failed" )

	end

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

--[[

	Instance Function

	\brief RECURSIVE - ensures that that a binary upload
	eventually comlpetes, by Oz confirmation OR via some error
	condition. Details change by binary type

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\waitCount		(int) - How many times this function has been called for this binary. Used in the recursive case.

	\return nil

]]
function WFCatBinaryUploader:eventuallyClearBlackoutWindow( docId, binaryDetails, waitCount )

	waitCount = waitCount or 0

	self._timerAborter
		:flatMapLatest( function()
			return WFRxLowFidelityTimer.docValueTimer( self._uploadBlackoutWindowInSeconds, self._docStore, docId )
		end )
		:tap( function( docValue )

			-- Verify that a binary was uploaded by checking document metadata and comparing sha256

			local uploadedSha256, currentSha256
			local uploadConfirmed

			if not isValidAssetDocument( docValue )
					and not isNonDeletedStyleDocument( docValue )
					and not WFCatIndexUtils.isNonDeletedDCPDocument( docValue )
					and not WFCatIndexUtils.isValidVersionDocument( docValue )
					and not WFCatIndexUtils.isValidWatermarkDocument( docValue ) then

				-- Clear out RecentlyUploaded for deleted documents.
				logger:warnf( "Asset deleted prior to upload completion for %s asset %s.", binaryDetails.type, docId )

				uploadConfirmed = true

			elseif isOriginalBinary( binaryDetails.type ) then

				uploadConfirmed = ( getOriginalLinkFromDocValue_IgnoringShaError( docValue ) ~= nil ) or
									( docValue._spaceId and getOriginalExistsFromDocValue( docValue ) )
				uploadedSha256 = binaryDetails.sha256
				currentSha256 = derefOpt( docValue, 'importSource', 'sha256' )

			elseif isProxy( binaryDetails.type ) then

				uploadConfirmed = ( WFCatFileSpec.getProxyLinkFromDocValue( docValue ) ~= nil ) or
									( docValue._spaceId and getProxyExistsFromDocValue( docValue ) )
				uploadedSha256 = binaryDetails.sha256
				currentSha256 = derefOpt( docValue, 'importSource', 'sha256' )

			elseif isXmpDevelop( binaryDetails.type ) then

				uploadConfirmed = WFCatFileSpec.getXmpDevelopLinkAndSha256FromDocValue( docValue ) ~= nil
				uploadedSha256 = binaryDetails.sha256
				currentSha256 = derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )

			elseif isXmpSidecar( binaryDetails.type ) then

				-- Verify legacy field _localyOnly.xmpSidecarNeedsUpload entry is removed from the docValue.
				uploadConfirmed = derefOpt( docValue, '_localOnly', 'xmpSidecarNeedsUpload' ) == nil
				uploadedSha256 = nil
				currentSha256 = nil

			elseif isAux( binaryDetails.type ) then

				-- We will then need to iterate over the aux list and extract the digest,
				-- and use the digest to check against the serverprovided metadata field

				local recentlyUploaded = binaryDetails.sha256

				-- check if an aux reference exists by sha256
				local auxId, auxDetails = KSTableUtils.findInTable( docValue.aux or {}, function( index, auxDoc )

						return auxDoc.sha256 == recentlyUploaded

					end )

				uploadConfirmed = false
				uploadedSha256 = nil

				if auxId and auxDetails then
					-- If there is an aux link in the server provided metadata, we know the binary has been registered on oz
					uploadConfirmed = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/aux/' .. auxId, 'href' ) ~= nil or
						-- The "shared with you case"
						derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/aux/' .. auxId, 'exists' ) ~= nil

					uploadedSha256 = auxDetails.sha256
				end

				currentSha256 = recentlyUploaded

			end

			if uploadedSha256 ~= currentSha256 then

				-- If there was a destructive edit that caused the original to change,
				-- resulting in a new sha256, we clear the asset from the list so that
				-- we may upload the new original. It should be noted that we don't yet
				-- support destructive edits in many other parts of sync.

				-- This probably happens when xmp_develop is modified during upload.

				logger:warnf( "sha no longer matches for %s type %s", docId, binaryDetails.type )

				self:delayedClearFromRecentlyUploaded( "sha no longer matches", docId, binaryDetails )

			elseif uploadConfirmed then

				-- Once the docValue contains an Oz master href, we can safely remove
				-- the asset from the recently uploaded table since we won't try to reupload
				-- assets that have oz master links.

				self:delayedClearFromRecentlyUploaded( "upload confirmed", docId, binaryDetails )

					:tap( function()

						-- Once an upload has completed, we can also clear any upload priority modifier
						-- set by client code. This flag can only be cleared once all binaries associated
						-- with a document have finished uploading. To accomplish this, we wait for
						-- a binary to upload and then run a delayed index store query which checks
						-- the needsBinaryUpload for more entries associated with the document; if there
						-- exist none, then we know we can clear the flag.

						if WFCatPhotoImportance.getUploadPriorityModifier( docValue ) then

							self:maybeClearUploadPriorityModifier( docId )

						end

						-- It's very important that we clear the progress events _after_ the upload has been
						-- confirmed so that progress doesn't appear to reset after an upload has finished,
						-- but before it's been confirmed.

						setNested( self._inFlightUploadProgress, nil, docId, binaryDetails.type, binaryDetails.sha256 )
						self._debugInFlightUploadProgressUpdateStream:onNext( self._inFlightUploadProgress )

					end )

					:call( runUntilDone )

			else

				waitCount = waitCount + 1

				logger:warnf( "Doc store hasn't updated with the presence of uploaded %s for %s after %d seconds",
						binaryDetails.type, docId, self._uploadBlackoutWindowInSeconds * waitCount )

				if waitCount > self._uploadBlackoutWindowCount then

					-- WF-921: Before assuming upload failed, check manually whether Oz has the master.
					-- if so, update the document with the master link and remove the doc from recently uploaded list
					-- Otherwise also just remove from the recently uploaded list, which will be tried later.

					-- WF-2410 with atomic develop settings enabled for version xmp_develop upload, if the
					-- version does not exist on Oz force another attempt to create version on Oz.

					return self:applyOverlayForBinaryUploadBlackoutWindowEdgeCases( docValue, binaryDetails.type )
						:map( function ( reason )
							self:delayedClearFromRecentlyUploaded( reason, docId, binaryDetails )
						end )
						:call( runUntilDone )

				else

					-- Keep waiting

					self:eventuallyClearBlackoutWindow( docId, binaryDetails, waitCount )

				end

			end

		end )

		:call( runUntilDone )

end

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

--[[

	Instance Function

	\brief Marks an upload as "complete" - meaning that is has
	been confirmed as being present in OZ, or an error has been
	encountered and the upload can be destroyed.

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\uploadSha256		(string) - The sha256 of the uploaded binary

	\return nil

]]
function WFCatBinaryUploader:uploadCompleted( docId, binaryDetails, uploadSha256 )

	self:eventuallyClearBlackoutWindow( docId, binaryDetails )

	local docStore = self._docStore

	if isOriginalBinary( binaryDetails.type ) then

		-- When upload complete, clear isSkipDirectUpload flag
		docStore:applyOverlay  {
			_id = docId,
			_localOnly = {
				binaryUpload = {
					[ binaryDetails.type ] = {
						isSkipDirectUpload = { _delete = true },
					}
				}
			},
		}
		:call( runUntilDone )

		-- NMB-5779: We may have received the master via migration or another
		-- change which required us to wait until Oz marked the previous binaries
		-- as invalid.  If this is the case, remove the annotation that tells us
		-- we must wait
		docStore:docValue( docId )
			:flatMap( function( docValue )

				local waitForInvalid = derefOpt( docValue, '_localOnly', 'binaryUpload', 'waitForInvalid' )

				-- NMB-7626: Previously wait for invalid was simply a boolean true. With the introduction of presets
				-- which are both small and the file is changeable, we
				if waitForInvalid == true -- old case (< 1.4) which may exist in doc store before upgrade
				    or ( waitForInvalid == uploadSha256 ) then -- 1.4+
					return docStore:applyOverlay {
						_id = docId,
						_localOnly = {
							binaryUpload = {
								waitForInvalid = { _delete = true },
							},
						},
					}
				else
					return Observable.just()
				end
			end )
			:call( runUntilDone )

	elseif isXmpSidecar( binaryDetails.type ) then

		-- Originals and develop settings have links set in the _serverProvidedMetadata
		-- for the binary's location on oz, whereas xmp is generated when
		-- an original is received.  So there is no way to detect the upload has
		-- completed by watching the sync feed, so update the legacy _localyOnly.xmpSidecarNeedsUpload
		-- when completed so that eventuallyClearBlackoutWindow can clear the upload
		-- from the blackout window when it's timer fires.

		docStore:applyOverlay  {
			_id = docId,
			_localOnly = {
				xmpSidecarNeedsUpload = { _delete = true },
			},
		}
		:call( runUntilDone )

	end

end

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

--[[

	Instance Function

	\brief Add point in time details about the upload to the diagnostic log
	TODO: Add some more color about how this is used and what it does

	\param bucket	(string) - Which bucket to add details about
	\param optTime	(num) - An optional time window over which to report on
	\param size	(num) - The total size of the binary being uploaded

	\return nil

]]
function WFCatBinaryUploader:addTransferStatsToDiagnosticLog( bucket, optTime, size )

	local sizeInMB = bytesToMb( size )

	local bucketStats = self._uploadSpeedStats.buckets[ bucket ]
	local uploadSpeed = optTime and sizeInMB / optTime

	if not bucketStats then

		bucketStats = {
			count = 0,
			timings = WFDiagnosticLog.makeRingBuffer { "Time (sec)" },
			sizes = WFDiagnosticLog.makeRingBuffer { "Size (MB)" },
			speed = WFDiagnosticLog.makeRingBuffer { "Upload speed (MBps)" },
		}

		self._uploadSpeedStats.buckets[ bucket ] = bucketStats

	end

	bucketStats.count = bucketStats.count + 1
	bucketStats.sizes:add( sizeInMB )

	if optTime then
		bucketStats.timings:add( optTime )
		bucketStats.speed:add( uploadSpeed )
	end

	local totalBinarySize = self._uploadSpeedStats.totalBinarySize or 0
	self._uploadSpeedStats.totalBinarySize = totalBinarySize + sizeInMB

end

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

--[[

	Instance Function

	\brief Conditionally emit an analytics event for a successful upload

	\param docId		(string) - The ID of the Document an upload is associated with
	\param uploadUrl	(string) - Destination for the upload
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param contentType	(string) - The content type of the binary
	\param isCopy		(bool) - Can this upload be skipped due to copy-on-oz
	\param isDirectUpload	(bool)   - Is this a DCU upload or not?

	\return nil

]]
function WFCatBinaryUploader:WFAnalyticsMaybeTrackEventNotifySuccess( docId, uploadUrl, binaryDetails, contentType, isCopy, isDirectUpload )

	-- this event is not expecially interesting and is very frequent, so let's conditionaly
	-- log so as send less to dunamis. Was previously limited to being logged for mobile
	-- and mac app store so keep logging it in those scenarios (for now at least) and also
	-- log proxy/master upload when asset has revisions not synced to Oz (see comment on
	-- story WF-2235 in maybeEmitWatermarkUploadRecord for more info).

	local function checkRevisionNotSyncedToOzForProxyOrMaster()

		if self:isAtomicDevelopSettingsEnabled() and
				( isOriginalBinary( binaryDetails.type ) or isProxy( binaryDetails.type ) ) then

			return self._docStore:docValue( docId )
				:map( function( docValue )

					if docValue.type == "asset" and not isLatestMetadataUploaded( docValue ) then
						return docValue._revId
					end

					return nil

				end )

		else

			return Observable.just( nil )

		end

	end

	checkRevisionNotSyncedToOzForProxyOrMaster()

		:map( function( revisionNotSyncedToOz )

			if not ( revisionNotSyncedToOz or WFPlatformInfo.isMobile() or WFPlatformInfo.isMacAppStore ) then
				-- do not send analytics event in this case
				return
			end

			local uploadMethod = kUploadMethodOz

			if isCopy then

				uploadMethod = kUploadMethodOzCopy

			elseif isDirectUpload then

				uploadMethod = kUploadMethodDirect

			elseif not self:checkDirectUploadFailureCount() then

				uploadMethod = kUploadMethodDirectFallback

			end

			local event = {
				['event.type'] = 'info',
				['event.subtype'] = binaryDetails.type .. "-uploaded",
				['event.appWorkflow'] = 'SYNC',
				['event.subcategory'] = 'Binary_upload_success',
				['event.uploadMethod'] = uploadMethod,
				['content.id'] = docId,
				['content.type'] = contentType,
				['content.url'] = WFAnalytics.sanitizeForAnalytics( uploadUrl ),
			}
			if revisionNotSyncedToOz then
				event.revisionNotSyncedToOz = revisionNotSyncedToOz
			end

			WFAnalytics.trackEvent(event)

		end )

		:call( runUntilDone )

end

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

-- Uppercase the first character in a string.

local function firstToUpper( str )
    return ( str:gsub( "^%l", string.upper ) )
end

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

--[[

	Instance Function

	\brief Emit an analytics event for a failed upload

	\param failureInfo (table) - details about the failed upload

	\return nil

]]
function WFCatBinaryUploader:WFAnalyticsTrackEventNotifyFailure( failureInfo )

	local eventSubcategory

	if failureInfo.reason then
		eventSubcategory = string.gsub( failureInfo.reason, "^%l", string.upper )
		eventSubcategory = string.gsub( eventSubcategory, "[%s\n]+", "_" )
	elseif failureInfo.fileSize and failureInfo.fileSize >= self._chunkedUploader:chunkFileSize() then
		eventSubcategory = 'Binary_upload_chunk'
	else
		eventSubcategory = 'Binary_upload_data'
	end

	local uploadMethod = kUploadMethodOz

	if failureInfo.isCopy then

		uploadMethod = kUploadMethodOzCopy

	elseif failureInfo.isDirectUpload then

		uploadMethod = kUploadMethodDirect

	elseif not self:checkDirectUploadFailureCount() then

		uploadMethod = kUploadMethodDirectFallback

	end

	local subtype -- subtype, content type, filesize shall be empty for the InvalidLocalOriginal case, so do not assign default value

	if isXmpDevelop( failureInfo.binaryDetails.type ) and failureInfo.parentDocumentId then

		subtype = 'version'

	elseif failureInfo.contentType then

		if WFOzContentTypes.isVideo( failureInfo.contentType ) then
			subtype = 'video'
		else
			subtype = 'image'
		end

	end

	local errorEvent = {}

	local binaryType = derefOpt( failureInfo, 'binaryDetails', 'type' ) or "UnknownType"
	local severity = failureInfo.warning and 'warning' or 'error'
	local errorType = string.format( "%sUpload.%s", binaryType, firstToUpper( severity ) )
		-- Upper case the severity to align with historical analytic entries.

	errorEvent['event.uploadMethod'] = uploadMethod
	errorEvent['event.type'] = severity
	errorEvent['event.appWorkflow'] = 'SYNC'
	errorEvent['event.subcategory'] = eventSubcategory
	errorEvent['event.url'] = WFAnalytics.sanitizeForAnalytics( failureInfo.url )
	errorEvent['event.error_type'] = errorType
	errorEvent['content.id'] = failureInfo.docId
	errorEvent['content.type'] = subtype
	errorEvent['content.mimetype'] = failureInfo.contentType
	errorEvent['content.size'] = failureInfo.fileSize

	if not failureInfo.httpError then

		WFAnalytics.trackEvent( errorEvent )
		return

	end

	errorEvent['content.source'] = (subtype == 'image') and 'file' or 'data'

	local errorCode = failureInfo.headers and failureInfo.headers.status_code or 'nil'

	if not failureInfo.headers then

		-- Modern native HTTP implementations put "wf_err_timeout" in the error JSON on timeout
		if (string.match( failureInfo.responseData, "wf_err_timeout" ) or string.match( failureInfo.responseData, "timed out" )) then
			errorCode = "timeout"
		end

		local annotatedReason = caughtResponseDataToString( failureInfo.responseData )

		errorEvent['event.subtype'] = 'os'
		errorEvent['event.error_code'] = errorCode
		errorEvent['event.error_desc'] = annotatedReason

	else

		local ozErrorResponse = caughtResponseDataToString( failureInfo.responseData )

		local ozRequestId = failureInfo.headers['X-Request-Id']

		errorEvent['event.subtype'] = 'api'
		errorEvent['event.error_code'] = errorCode
		errorEvent['event.error_desc'] = ozErrorResponse
		errorEvent['event.context_guid'] = ozRequestId

	end

	if failureInfo.extra then
		-- NOTE: DANGER - may cause long logs. Use this sparingly, and only when required.
		errorEvent['content.extra'] = KSJsonEncoder.encodeOzUploadJson( failureInfo.extra )
	end

	WFAnalytics.trackEvent( errorEvent )

end

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

--[[

	Instance Function

	\brief Add an HTTP error to the log and track the failure via dunamis

	\param docId (string) - The ID of the Document an upload is associated with
	\param masterPutUrl	(string) - The destination of the upload
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload
	\param binaryDetails    (table) - A BinaryDetails domain object - see top of file
	\param contentType	(string) - The content type associated with the upload
	\param localSize	(int) - The size of the binary on disk in bytes
	\param isCopy		(bool) - Can this upload be skipped due to copy-on-oz
	\param isDirectUpload	(bool) - Is this a DCU upload or not?
	\param isWarning	(bool) - Is this a warning or a legitimate error that should be reported?

	\return nil

]]
function WFCatBinaryUploader:addHttpErrorToLog( docId, masterPutUrl, responseData, headers,
		binaryDetails, parentDocumentId, contentType, localSize, isCopy, isDirectUpload, isWarning )

	-- Special case for HTTP transit failure.

	local errorLog = self._uploadErrorLog

	if not headers then

		local annotatedReason = caughtResponseDataToString( responseData )

		local logFmtStr = "Unable to communicate with Oz URL %s\n\nreason = %s"

		logger:errorf( logFmtStr, masterPutUrl, annotatedReason )

	elseif not isWarning then

		-- We were able to communicate with Oz, but Oz didn't like the response. Log this.

		if not headers then
			headers = {}
		end

		local logFmtStr = "Got status %s back from Oz\n\n"
		local logParams = { headers.status_code or 'nil' }

		if masterPutUrl then
			logFmtStr = logFmtStr .. "url we sent data to was:\n\n%s\n\n"
			logParams[ #logParams + 1 ] = masterPutUrl
		end

		logFmtStr = logFmtStr .. "HTML response body was:\n\n%s\n\n"
		logParams[ #logParams + 1 ] = responseData

		errorLog[ #errorLog + 1 ] = {

			time = KSDate.timeInIso8601FormatWithTimeZone(),

			id = docId,
			binaryType = binaryDetails.type or 'master',
			error = string.format( logFmtStr, unpack( logParams ) ),

		}

		if #errorLog > 10 then
			table.remove( errorLog, 1 )
		end

		logger:errorf( logFmtStr, unpack( logParams ) )

	end

	local failureInfo = {
		docId = docId,
		url = masterPutUrl,
		responseData = responseData,
		headers = headers,
		binaryDetails = binaryDetails,
		parentDocumentId = parentDocumentId,
		contentType = contentType,
		fileSize = localSize,
		httpError = true,
		isCopy = isCopy,
		isDirectUpload = isDirectUpload,
		warning = not not isWarning
	}

	self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

end

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

--[[

	Instance Method

	\brief Wrap up a successful upload - Updates upload status, dunamis events, and internal streams

	\param docId		(string) - The ID of the Document an upload is associated with
	\param uploadUrl	(string) - Destination for the upload
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param localSize        (int) - The size of the binary on disk in bytes
	\param response         NOTE: UNUSED
	\param contentType	(string) - The content type associated with the upload
	\param isCopy		(bool) - Can this upload be skipped due to copy-on-oz
	\param isDirectUpload	(bool)   - Is this a DCU upload or not?

]]
function WFCatBinaryUploader:recordSuccess( docId, uploadUrl, binaryDetails, localSize, response, contentType, isCopy, isDirectUpload )

	localSize = localSize or 0

	local bucket = binaryDetails.type

	if isDirectUpload then

		bucket = bucket .. "-direct"

	elseif localSize >= self._chunkedUploader:chunkFileSize() then

		bucket = bucket .. "-chunked"

	end

	if isCopy then

		bucket = bucket .. "-copy"

	end

	if not string.match( bucket, "-chunked" ) then

		-- FIXME [jandrew 2021-08-31]: This block and uploadProgressStatusUpdate should
		-- be removed as part of the work for https://jira.corp.adobe.com/browse/WF-1710

		-- for chunked uploads audit happens in WFCatContentRangeUploader
		self:uploadProgressStatusUpdate( "increaseBytesUploaded", localSize )

	end

	local progress = derefOpt( self._inFlightUploadProgress, docId, binaryDetails.type, binaryDetails.sha256 )

	local progressOfChunks = derefOpt( progress, 'extraInfo', 'progressOfChunks' )

	local function trackProgressEventStats( progressEvent )

		local bytesUploaded = progressEvent.totalCount
		local startTime = derefOpt( progressEvent, 'extraInfo', 'startTime' )
		local endTime = derefOpt( progressEvent, 'extraInfo', 'endTime' )

		if startTime and endTime then
			self:addTransferStatsToDiagnosticLog( bucket, endTime - startTime, bytesUploaded )
		end

	end

	if progressOfChunks then

		for _, chunkProgress in pairs( progressOfChunks ) do

			trackProgressEventStats( chunkProgress )

		end

	elseif progress then

		trackProgressEventStats( progress )

	else

		self:addTransferStatsToDiagnosticLog( bucket, nil, localSize )

	end

	self:WFAnalyticsMaybeTrackEventNotifySuccess( docId, uploadUrl, binaryDetails, contentType, isCopy, isDirectUpload )

	self._binaryUploadCompletionStream:onNext( docId )

end

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

--[[

	Instance Function

	\brief Remove the XMP sidecar annotation for a given document

	\param docId (string) - The ID of the Document an upload is associated with

	\return Observable - resolves with the new value's
		revision ID when the write completes. You may safely ignore this promise.

]]
function WFCatBinaryUploader:removeSidecarFileAnnotation( docId )

	local binarySizes = {}
	binarySizes[ 'xmp' ] = { _delete = true }

	local files = {}
	files[ 'original' ] = binarySizes

	return self._docStore:applyOverlay {
			_id = docId,
			_localOnly = {
				files = files,
				xmpSidecarNeedsUpload = { _delete = true },
				xmpSidecarExists = false,
			}
		}

end

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

--[[

	Instance Function

	\brief Mark the original as invalid since the sha256 no longer
	matches what we imported. This can happen because of disk
	corruption or from an invalid copy during import. (since sha
	may be calculated prior to importing by the Import UI). It
	could also happen if the user copied a new file into the
	~/Pictures/Nimbus folder manually (which they should not
	do). Adding localOriginalError will cause this file to no
	longer be considered for uploaded.

	\param docId (string) - The ID of the Document an upload is associated with
	\param filePath       (string) - Path to the binary on this device
	\param binaryUrl (string)
	\param reason (string)
	\param info NOTE: unused downstream

	\return Observable<revisionID or bool> - resolves with the new value's
		revision ID when the write completes.

]]
function WFCatBinaryUploader:applyInvalidOriginalOverlay( docId, filePath, binaryUrl, reason, info )

	check( 'WFCatBinaryUploader', '+string', '?string', '+string', '?string', '?string' )

	reason = reason or 'InvalidLocalOriginal'

	local shouldApplyOverlay = true

	local volumeIsOnline = filePath and WFFileUtils.getVolumeGuidForPath( filePath )

	if filePath and not volumeIsOnline then

		-- If the file is on a volume which was disconnected
		-- in the middle of the upload, then we shouldn't apply the
		-- error overlay.

		shouldApplyOverlay = false

	end

	if shouldApplyOverlay then

		addInvalidOriginalErrorToLog( self._uploadErrorLog, docId, filePath, reason )

		local failureInfo = {
			docId = docId,
			url = binaryUrl,
			binaryDetails = { type = 'master' },
			httpError = false,
			reason = reason,
			info = info
		}

		self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

		-- Mark the original as invalid since the sha256 no longer matches what we imported.

		-- This can happen because of disk corruption or from an invalid copy during import.
		-- (since sha may be calculated prior to importing by the Import UI).

		-- It could also happen if the user copied a new file into the ~/Pictures/Nimbus folder
		-- manually (which they should not do).

		-- Adding localOriginalError will cause this file to no longer be considered for uploaded.

		return self._docStore:applyOverlay {
				_id = docId,
				_localOnly = {
					localOriginalError = reason,
				}
			}

	else

		return Observable.just( false )

	end

end

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

--[[

	Instance Function

	\brief Check if a given document has any missing or invalid
	auxiliary references. If there are, mark the document with an
	error

	\param docId (string) - The ID of the Document an upload is associated with

	\return Observable<revisionID or bool> - resolves with the new value's
		revision ID when the write completes.

]]
function WFCatBinaryUploader:maybeApplyInvalidAuxOverlay( docId )

	check( 'WFCatBinaryUploader', '+string' )

	local docStore = self._docStore

	return docStore:docValue( docId )

		:flatMap( function( docValue )

			-- The following code is very similar to the handling inside of
			-- referencedAuxMissingErrorHandler within the WFCatErrorHandler.
			-- The primary difference is that the following code cares about
			-- the status of the binaries -- whether they exist on Oz or exist
			-- locally -- while referencedAuxMissingErrorHandler only cares
			-- about the metadata.

			if WFCatIndexUtils.isValidVersionDocument( docValue ) then

				local versionAuxInfoList = WFCatAssetAux.getAuxInfoList( docValue )

				local assetId = derefOpt( docValue, 'asset', 'id' )

				return docStore:docValue( assetId )

					:flatMap( function( assetDocValue )

						local assetAuxInfoList = WFCatAssetAux.getAuxInfoList( assetDocValue )

						local confirmedMissingAuxIds = {}

						for auxId in pairs( versionAuxInfoList ) do

							local auxInfo = assetAuxInfoList[ auxId ]

							if not auxInfo then

								logger:warnf( "Parent asset %s is missing auxiliary: %s", assetId, auxId )

								confirmedMissingAuxIds[ #confirmedMissingAuxIds + 1 ] = auxId

							elseif not WFCatAssetAux.isAuxiliaryAccessible( self._catalog, assetDocValue, auxId ) then

								logger:warnf( "Parent asset %s is contains a reference to auxiliary: %s, but it's not accessible.", assetId, auxId )

								confirmedMissingAuxIds[ #confirmedMissingAuxIds + 1 ] = auxId

							else

								logger:infof( "Parent asset %s is not missing auxiliary: %s", assetId, auxId )

							end

						end

						if #confirmedMissingAuxIds > 0 then

							return self._docStore:applyOverlay {
								_id = docId,
								_localOnly = {
									ozError = {
										subtype = WFCatOzErrors.Errors.kReferencedAuxMissingError
									}
								}
							}

						end

						return Observable.just( false )

					end )

			end

			return Observable.just( false )

		end )

end

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

--[[

	Instance Function

	\brief Introduced with WF-2343. If there is a case where an uploading
	aux binary fails, force a resync from Oz and treat the parent document
	as a "stuck sync"

	\param url           (string) - The url which first encountered the http error from Oz
	\param binaryDetails (binaryDetails - domain construct) -
	\param docId         (string) - The ID of the Document an upload is associated with
	\param responseData  (table) - The parsed HTTP response
	\param headers       (table) - The headers included in the request

	\return nil - SIDE EFFECT

]]
function WFCatBinaryUploader:handleAuxDoesNotMatch( url, binaryDetails, docId, responseData, headers, options)

	if not WFCatStuckSyncUtils then
	    WFCatStuckSyncUtils = require 'WFCatStuckSyncUtils'
	end

	return self._docStore:docValue( docId )
		:flatMap( function( docValue )

			-- In the case of an oz reported aux mismatch - We have very little
			-- data that can be used to debug. Extract some of the existing aux
			-- tables associated with the asset and log them in hopes that we can
			-- debug the issue

			local failureInfo = {
				docId = docId,
				url = url,
				responseData = responseData,
				headers = headers,
				binaryDetails = binaryDetails,
				isDirectUpload = true,
				warning = true,
				reason = 'binary_upload_aux_sha256_mismatch',
				extra = {
					assetAux = {},
					revisionIds = derefOpt(docValue, '_serverProvidedMetadata', 'revision_ids') or 'unknown',
					labels = derefOpt(docValue, '_labels') or 'unknown'
				}
			}

			local maxAux = 10
			local auxSection = derefOpt( docValue, 'aux')
			for digest, auxDoc in pairs( auxSection ) do
				if maxAux < 1 then
					break
				end

				failureInfo.extra.assetAux[ digest ] = auxDoc
				maxAux = maxAux - 1

			end

			if options.TEST_OVERRIDE then
				-- Provide a hook which allows a test
				-- to check the format of the
				-- analytics error message
				options.TEST_OVERRIDE = failureInfo
			end

			self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

			return WFCatStuckSyncUtils.resolveStuckSync( self._catalog, { docId } )

	end)
	:call( runUntilDone )

end

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

--[[

	Instance Function

	\brief MAYBE resolve a sha256 error if possible. The most
	comment case for this to happen is when an async upload
	operation has a sha mismatch error.  This happens when either
	WFCatContentRangeUploader or WFCatContentRangeUploader
	complete (they get success) but oz assembles the pieces
	together and updates importSource.status with this error if
	the sha doesn't match.

	\param masterUploadUrl	(string) - URL for the upload
	\param filePath		(string) - Path to the binary on this device
	\param docId		(string) - The ID of the Document an upload is associated with
	\param imageSha256	(string) - unique sha of the binary

	\return Observable<revisionId> - resolves with the new value's
		revision ID when the write completes.

]]
function WFCatBinaryUploader:resolveSha256ErrorInInputSource( masterUploadUrl, filePath, docId, imageSha256 )

	-- The most comment case for this to happen is when an async upload operation
	-- has a sha mismatch error.  This happens when either WFCatContentRangeUploader
	-- or WFCatContentRangeUploader complete (they get success) but oz assembles
	-- the pieces together and updates importSource.status with this error if
	-- the sha doesn't match.

	-- Copied assets can also get marked with the sha256error so check to see if
	-- we have a filePath before attempting the digest, if there is no filePath
	-- then just mark the asset as invalid.
	if filePath == nil then
		return self:applyInvalidOriginalOverlay( docId, filePath, masterUploadUrl , nil, "file path is nil" )
	end

	-- Must have been from an async upload, see if we can fix it.

	-- Recalculate sha256 to verify it matches the sha256 in the importSource to
	-- verify we have the correct file to upload.

	return WFCrypto.asyncDigestFile( filePath, 'SHA256' )

	:flatMap( function( localSha256 )

		if localSha256 == imageSha256 then

			-- Update the doc so the binary uploader will try to upload this file again.

			logger:debugf( "re-uploading corrupted original for docId %s path %s", docId, filePath )

			-- https://jira.corp.adobe.com/browse/WF-1902: This is considered
			-- a warning and not an error.

			local failureInfo = {
				docId = docId,
				url = masterUploadUrl,
				binaryDetails = { type = 'master', sha256 = localSha256 },
				httpError = false,
				reason = "re-uploading corrupted original",
				warning = true
			}

			self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

			return self._docStore:docValue( docId )
				:flatMap( function( docValue )

					local currentMasterUrl = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href' )

					return self._docStore:applyOverlay {
						_id = docId,
						_localOnly = {
							-- Add a local annotation for the sha error since we're going to
							-- otherwise remove all traces that it happened.  This is necessary
							-- to prevent us from accidentally deleting the local master because
							-- we think the master is uploaded and fine (NMB-4862)
							binaryUploadErrors = {
								[ '/rels/master' ] = {
									[ currentMasterUrl ]  = "corrupt upload",
								}
							}
						},
					}
			end )

		else

			return self:applyInvalidOriginalOverlay( docId, filePath, masterUploadUrl, nil, "sha256 mismatch" )

		end

	end )

end

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

--[[

	Instance Function

	\brief Remove or cancel uploads for a given document

	\param docId (string) - The ID of the Document an upload is associated with

	\return nil - side efffect - update internal tracking variables

]]
function WFCatBinaryUploader:disposeOfUploadForDeletedDocId( docId )

	local uploadSubscriptionsForDocId = self._scheduledUploadSubscriptions[ docId ] or {}

	for binaryDetails, subscription in pairs( uploadSubscriptionsForDocId ) do

		logger:debugf( "Aborting %s upload for docId %s", binaryDetails.type, docId )

		subscription:dispose()

		self:clearFromRecentlyUploaded( "disposed", docId, binaryDetails, true )

		self:removeFromScheduledUploads( docId, binaryDetails )

	end

end

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

--[[

	Instance Function

	\brief Terminate all active uploads.

	\return nil - side effect: reset all internal tracking variables related to uploads

]]
function WFCatBinaryUploader:disposeAllActiveUploads()

	logger:debugf( "disposeActiveUploads started" )

	-- Abort the timers we have for clearing the recently uploaded fields
	-- since we are clearing them out immediately below.  This prevents
	-- a timer from a prior upload request from being issued which can cause
	-- a file to be uploaded multiple times simultaneously.
	self._timerAborter:onError( 'Abort' )

	-- Create a new _timerAborter that isn't in the terminal state for the
	-- next set of timers.
	self._timerAborter = Rx.ReplaySubject( 1 )
	self._timerAborter:onNext( true )

	for docId, uploadSubscriptionsForDocId in pairs( self._scheduledUploadSubscriptions ) do

		for binaryDetails, subscription in pairs( uploadSubscriptionsForDocId ) do

			logger:debugf( "Aborting upload for docId %s", docId )

			subscription:dispose()

			self:clearFromRecentlyUploaded( "disposed", docId, binaryDetails, true )

		end

	end

	if self._numberOfScheduledBinaryUploads > 0 then
		logger:errorf( "_numberOfScheduledBinaryUploads == %d should be 0", self._numberOfScheduledBinaryUploads )
	end

	self._numberOfScheduledBinaryUploads = 0

	self._recentlyUploaded = {}

	self._scheduledUploadSubscriptions = {}

	self._uploadingAssetWithSha256 = {}

end

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

--[[

	 Instance Function

	\brief Determine how a failed upload should be handled. This
	could be a simple delay + retry or a complete blocking of this
	asset for future uploads.

	\param docId (string) - The ID of the Document an upload is associated with
	\param binaryDetails (table) - A BinaryDetails domain object - see top of file
	\param uploadUrl	(string) - Destination for the upload
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload
	\param invalidOriginal  (bool) - was the original marked as invalid?
	\param contentType	(string) - The content type associated with the upload
	\param localSize	(int) - The size of the binary on disk in bytes
	\param isCopy		(bool) - Can this upload be skipped due to copy-on-oz
	\param isDirectUpload	(bool)   - Is this a DCU upload or not?
	\param isWarning	(bool) - Did Oz report a sha256 difference for this binary?

	\return nil - side effect: clear upload from recent uploads

]]
function WFCatBinaryUploader:handleUploadFailure( docId, binaryDetails, uploadUrl, parentDocumentId,
		responseData, headers, invalidOriginal, contentType, localSize, isCopy, isDirectUpload, isWarning )

	self._errorStatusStream:onNext( docId, responseData )

	self:addHttpErrorToLog( docId, uploadUrl, responseData, headers, binaryDetails,
		parentDocumentId, contentType, localSize, isCopy, isDirectUpload, isWarning )

	if invalidOriginal then
		self:delayedClearFromRecentlyUploaded( "wait for docstore update as invalidOriginal", docId, binaryDetails )
		return
	end

	if not isDirectUpload and self._failedUploadDeniedListHelper:maybeDeny( docId ) then

		-- The asset is now denied. It will be automatically removed from the
		-- deniedList after failedUploadBlackoutWindowInSeconds seconds.

		WFPerformanceReport.recordSingleValue( 'WFCatBinaryUploader:DeniedList', {
			docId = docId,
			numberOfFailedAttempts = numberOfTriesBeforeDenyingUpload,
			deniedListDuration = failedUploadBlackoutWindowInSeconds
		} )

		self:delayedClearFromRecentlyUploaded( "wait for docstore update as denied", docId, binaryDetails )

	else

		-- Upload wasn't denied so try to upload the original again.

		local uploadErrorDelay = 10
		self:delayedClearFromRecentlyUploaded( "upload failed, delay to let other assets upload first", docId, binaryDetails, uploadErrorDelay )

	end

end

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

--[[

	Instance Function

	\brief Allows a master to be uploaded if it has previously been blocked

	\param docId (string) - The ID of the Document an upload is associated with

	\return Observable<revisionID> - resolves with the new documents revision ID when the write completes

]]
function WFCatBinaryUploader:applyRemoveWillNotUploadMasterOverlay( docId )

	return self._docStore:applyOverlay {
			_id = docId,
			importSource = {
				willNotUploadMaster = { _delete = true },
			},
		}

end


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

--[[

	Instance Function

	\brief Removes an imporoperly attached _localOnly.files.original.master file

	\param docId (string) - The ID of the Document an upload is associated with

	\return Observable<revisionID> - resolves with the new documents revision ID
		when the write completes or false if the overlay wasn't applied.

]]

function WFCatBinaryUploader:applyRemoveLocalOnlyMasterFileOverlay( catalog, docStore, docId )

	return docStore:docValue( docId )
	:flatMap( function( docValue )

		-- Don't allow a different user to upload the original if in a shared space
		-- rejects with 403 {"code":1016,"description":"Requested operation not permitted","subtype":"PermissionError"}
		-- This can happen if they have the same photo as another user that uploaded it via LrClassic
		-- and the original is available locally.
		
		-- This removes incorrectly applied overlays from checkExpectedPathForOriginal in
		-- PixelProviderLocalBinaryPathCheckTask.lua. Couldn'[t put this in SimpleDocStoreUpdater because
		-- of lack of access to the ozUserId which is obtained through the catalog

		-- The permission issue only would happen with photos imported from Classic
		-- as they didn't have a master link / binary provided.

		local ozUserId = catalog:ozUserId()
		local importedBy = derefOpt( docValue, 'importSource', 'importedBy' )
		local spaceId = derefOpt( docValue, '_spaceId' )
		local wasImportedByThisUser = not spaceId or importedBy == ozUserId

		-- Also check willNotUploadMaster to limit to Classic style assets only to reduce
		-- chance of removing _localOnly master file
		if not wasImportedByThisUser and derefOpt( docValue, 'importSource', 'willNotUploadMaster' ) then
			return self._docStore:applyOverlay( {
				_id = docId,
				_localOnly = {
					files = {
						original = {
							master = { _delete = true }
						}
					}
				}
			} )
		else
			return Observable.just( false )
		end

	end )

end

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

--[[

	Instance Function

	\brief Apply a specific Oz Error to a document

	\param docId (string) - The ID of the Document an upload is associated with
	\param errorType (string) - An OZ provided error

	\return Observable<revisionID> - resolves with the new documents revision ID when the write completes

]]
function WFCatBinaryUploader:applySpecificErrorOverlay( docId, errorType )

	local overlay = WFCatOzErrors.createOverlayForOzErrorSubtype( docId, errorType )
	return self._docStore:applyOverlay( overlay )

end

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

--[[

	Instance Function

	\brief Checks to see if an Account has any current errors associated with it.

	\param docId (string) - The ID of the Document an upload is associated with
	\param binaryType (string) - The type of the binary, refer to kBinaryType... values
	\param responseData	(table) - The parsed HTTP response
	\param headers		(string - msgpack) - The headers used in the upload

	\return Observable Error

]]
function WFCatBinaryUploader:checkForAccountError( docId, binaryType, responseData, headers )

	if errorTrialOrSubscriptionExpired( responseData, headers ) then

		return self:applySpecificErrorOverlay( docId, WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError )
			:flatMap( function()
				return Observable.throw( responseData, headers, true )
			end )

	end

	if errorInsufficientPermission( responseData, headers ) then

		return self:applySpecificErrorOverlay( docId, WFCatOzErrors.Errors.kInsufficientPermissionError )
			:flatMap( function()
				return Observable.throw( responseData, headers, true )
			end )

	end

end

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

--[[

	Instance Function

	\brief Remove references to a source asset from a child asset.

	\param docStore		(table) - docstore instance
	\param docValue		(table) - A complete Document
	\param copyBinaryUrl	(string) - the URL to issue a binary copy

	\return Observable<revisionID>  - resolves with the new documents revision ID when the write completes

]]
function WFCatBinaryUploader:applyOverlayToRemoveSourceAsset( docStore, docValue, copyBinaryUrl )

	local localOnly = {}

	localOnly.sourceOriginalLink = { _delete = true }
	localOnly.sourceXmpLink = { _delete = true }
	localOnly.sourceXmpDevelopLink = { _delete = true }
	localOnly.sourceProxyLink = { _delete = true }
	localOnly.sourceAuxLinks = { _delete = true }
	localOnly.sourceAssetLinkForAux = { _delete = true }

	localOnly.sourceAssetId = { _delete = true }

	local hasMasterLocally = WFCatAssetUtils.hasLocalOriginal( docValue )
	local hasProxyLocally = WFCatAssetUtils.hasLocalProxy( docValue )

	local importSource
	if hasProxyLocally and not hasMasterLocally then
		importSource = {
			willNotUploadMaster = true
		}
	end

	local maybeApplyInvalidOriginal = Observable.empty()
	if not hasProxyLocally and not hasMasterLocally then
		maybeApplyInvalidOriginal =
			self:applyInvalidOriginalOverlay( docValue._id, nil, copyBinaryUrl, "CopyFailedSourceMissing" )
			:takeLast(0)
	end

	return Observable.concat(
		maybeApplyInvalidOriginal,
		docStore:applyOverlay( {
			_id = docValue._id,
			_localOnly = localOnly,
			importSource = importSource
		} )
	)
end

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

--[[

	Instance Function

	\brief Bring a document into a good state as a copy by adding
	or removing source asset links in the document. Used in the
	case that a master or proxy resource does not exist.

	TODO: This needs more detail - it is not a good comment to describe why this function exists

	\param docId		(string) - The ID of the Document an upload is associated with
	\param newSourceId	(string) - The ID of the new source document
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param copyBinaryUrl	(string) - The URL to issue a binary copy request

	\return Observable<bool> - Used to allow the upload to retry or fail

]]
function WFCatBinaryUploader:applyOrRemoveNewSourceLinksOrErrorOverlayForCopy( docId, newSourceId, binaryDetails, copyBinaryUrl )

	-- Update source asset and links in docId with id and links retrieved from
	-- the document identified by newId.

	local docStore = self._docStore

	return docStore:docValue( docId )
	:flatMap( function( docValue )

		if newSourceId then

			-- Oz provided a new_id, so source must have been deleted.
			return Observable.just( true, newSourceId, docValue )

		else

			-- Oz didn't provide a new_id in the result headers, check to see
			-- if the source was deleted and get the _newId from the deleted doc.
			local sourceAssetId = derefOpt( docValue, '_localOnly', 'sourceAssetId' )

			return docStore:docValue( sourceAssetId )
			:map( function( existingSourceDocValue )

				local existingSourceDeleted = existingSourceDocValue._deleted

				newSourceId = existingSourceDocValue._newId

				if newSourceId then
					return existingSourceDeleted, newSourceId, docValue
				else
					return existingSourceDeleted, nil, docValue
				end

			end )

		end

	end )
	:flatMap( function( existingSourceDeleted, newSourceId, docValue )

		if newSourceId then

			return WFCatTrashUtils.getTrashOrRestoredDocFromNewId( docStore, newSourceId )
			:flatMap( function( newSourceDocValue )

				if newSourceDocValue._deleted then

					-- Remove the source links as the asset was permanently deleted.
					return self:applyOverlayToRemoveSourceAsset( docStore, docValue, copyBinaryUrl )
						:map( function()
							 -- Don't count as failed yet, see if we can upload a binary.
							return true
						end )

				elseif newSourceDocValue._empty then

					-- Doesn't exist yet, newId likely came from oz for asset
					-- from which we haven't received metadata.

					-- Count this as a failed attempt, it will retry with existing
					-- invalid link and get error again resulting in redoing all this
					-- in hopes that it will get an update with the links to the
					-- binary.

					return Observable.just( false )

				else

					-- Exists

					if WFCatFileSpec.getOriginalLinkFromDocValue( newSourceDocValue ) == nil
					and WFCatFileSpec.getProxyLinkFromDocValue( newSourceDocValue ) == nil then

						-- Links aren't there yet

						-- Count this as a failed attempt, it will retry with existing
						-- invalid link and get error again resulting in redoing all this
						-- in hopes that it will get an update with the links to the
						-- binary.

						return Observable.just( false )

					else

						local overlay = makeOverlayToUpdateSourceAsset( docValue._id, newSourceDocValue )
						return docStore:applyOverlay( overlay )
							:map( function()
								-- Don't count as failed as we applied a new link.
								return true
							end )

					end

				end

			end )

		else

			if existingSourceDeleted then
				-- Remove sourceLink as it no longer exists.
				return self:applyOverlayToRemoveSourceAsset( docStore, docValue, copyBinaryUrl )
					:map( function()
						return true -- Try again, don't count as see if we can upload a binary.
					end )
			else
				 -- Count this as a failed attempt, but retry.
				return Observable.just( false )
			end

		end

	end )

end

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

--[[

	Instance Function

	\brief Main entry function to upload a binary. Constructs
	request paramters, initiates an upload, and keeps track of
	progress, success, and failure.

	\param docId			(string) - The ID of the Document an upload is associated with
	\param localPath		(string) - The path to the binary-to-upload on disk
	\param localSize		(int) - The size of the binary-to-upload on disk in bytes
	\param binaryUrl		(string) - The OZ provided destination URL for the binary
	\param binaryDetails		(table) - A BinaryDetails domain object - see top of file
	\param contentType		(string) - The content type associated with the upload
	\param addlUploadHeaders	(table) - Extra headers to be associated with the upload
	\param sha256			(string) - Sha256 of the binary-to-upload
	\param uploadPriority		(int) - The priority of this upload
	\param isDirectUpload		(bool)   - Is this a DCU upload or not?
	\param optDevelopUserUpdated  (string) - used for aux and xmp_develop direct upload
	\param optParentDocumentId (string) - used for error handling to determine if upload was for a version
	\param optSpecialRendition	(bool) - Does this upload need a special rendition - see docNeedsSpecialProxyOrRenditionRequest
	\param optSpecialProxy			(bool) - Does this upload need a special proxy - see docNeedsSpecialProxyOrRenditionRequest

	\returns nil or Observable<bool> - side effect function: drive upload requests

]]
function WFCatBinaryUploader:uploadOneBinary(
		docId, localPath, localSize, binaryUrl, binaryDetails, contentType,
		addlUploadHeaders, sha256, uploadPriority, isDirectUpload, optDevelopUserUpdated,
		optParentDocumentId, optSpecialRendition, optSpecialProxy )

	logger:debugf( "Uploading %s for docId %s from %s to %s", binaryDetails.type, docId, localPath, binaryUrl )

	local options = getRequestOptionsFor( binaryDetails.type, contentType, false, addlUploadHeaders, isDirectUpload,
		self:isAtomicDevelopSettingsEnabled(), optSpecialRendition, optSpecialProxy )
	options.requestDescription = string.format( '%s : Uploading %s', docId, binaryDetails.type )

	return self._allowBackgroundUpload

	:take( 1 )

	:flatMap( function( allowBackgroundUpload )

		if uploadPriority >= 1000 or allowBackgroundUpload then

			-- One of the ways for the priority to be greater than 1000 is if the client code has marked this
			-- upload as being time sensitive. When we detect this, we allow the request to bypass the rate
			-- limiter and so that it starts to upload immediately.
			-- Another is when iOS enter background, we would allow bypass the rate limiter in order to schedule
			-- more tasks into background upload session

			options.bypassRateLimiter = true
		end

		return self:putAndTrackPerformance( binaryUrl, binaryDetails, localPath, localSize, options,
				docId, isDirectUpload, optDevelopUserUpdated )

	end )

	:flatMap( function( responseData, responseHeader, progress )

		-- after completing xmp_develop or aux direct upload, if the asset is annotated with a BinaryMissingError,
		-- then just to be super safe that the revision will eventually sync to Oz, replace that error annotation.
		-- with another one. This does a few things:
		-- 1. It triggers WFCatMetadataPusher to try to push the revision to Oz again (whether or this is
		-- currently necessary, Oz might already have the revision and be retrying it.)
		-- 2. The error annotation in this case has a limited number of retries to Oz and will eventually stop
		-- retrying, which could also cause revision to never be pushed to Oz if binary upload takes too long,
		-- but setting to a different error resets the error mechanism.
		-- 3. We previously cleared the error, but that could in theory introduce some race conditions that could
		-- cause shouldSendRevisionToOz check in WFCatMetadataPusher to compare _localOnly before the error
		-- annotation had been added at all and after it has been removed giving a false positive that the
		-- _localOnly is unchanged, so better to change the error rather than remove.

		-- (Note that this is NOT intended to handle an error that occurred in the binary upload workflow; this
		-- error would have been triggered by pushing revision metadata for a version or asset to Oz and it would
		-- have been an error synced via metadata puller and handled by WFCatErrorHandler.)

		local result = Observable.just( true )

		if isDirectUpload and isDevelopBinary( binaryDetails.type ) then

			result = self._docStore:docValue( docId )

				:flatMap( function( docValue )

					if KSTableUtils.derefOpt( docValue, '_localOnly', 'ozError', 'subtype' ) == 'BinaryMissingError' then

						local result = WFCatErrorHandler.maybeMarkDocumentForRetry( self._docStore, docValue, 'DirectUploadComplete', 202 )
							:map( function( )
								logger:warnf( "Switched BinaryMissingError annotation to DirectUploadComplete annotation to force metadata push retry on document %s", docId )
								return true
							end )

						local overlay = WFCatErrorHandler.recoverMissingAuxOverlay( self._catalog, docValue )

						if next( overlay ) then

							result = self._docStore:applyOverlay( overlay )
								:map( function( )
									logger:warnf( "Recovered missing auxiliaries on document %s", docId )
									return true
								end )

						end

						return result

					else

						return Observable.just( true )

					end

				end )

		end

		return result:map( function()
			return responseData, responseHeader, progress
		end )

	end )

	:catch( function( responseData, headers, requestInfo )

		local isWarning = false

		if requestInfo then

			local errorUrl = requestInfo.errorUrl

			local failureInfo = {
				docId = docId,
				url = errorUrl,
				responseData = responseData,
				headers = headers,
				binaryDetails = binaryDetails,
				parentDocumentId = optParentDocumentId,
				contentType = contentType,
				fileSize = localSize,
				httpError = true,
				isCopy = false,
				isDirectUpload = isDirectUpload
			}

			-- WF-1752 If we detect any asset has encountered 3 consecutive DU
			-- failures, then we should temporarily prioritize Oz-based upload
			-- for 30 mins before switching back to DU.

			if not self._fallbackToOzUploadForOriginalsOnly
					and not self._directUploadFallBackToOzUploadTimerSub
					and not self:checkDirectUploadFailureCount() then

				self._fallbackToOzUploadForOriginalsOnly = true
					-- Temporarily fallback to oz upload for originals only

				self._directUploadFallBackToOzUploadTimerSub =
						WFRxLowFidelityTimer.timer( self._directUploadFallBackToOzUploadWindowInSecond )

					:subscribe( function()
						self._fallbackToOzUploadForOriginalsOnly = false
						self._directUploadConsecutiveFailureCounter = 0
						self. _directUploadFallBackToOzUploadTimerSub:dispose()
						self. _directUploadFallBackToOzUploadTimerSub = nil
					end )

			end

			self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

			return Observable.throw( responseData, headers, false, isWarning )

		end

		if errorShaDoesNotMatch( binaryDetails.type, responseData, headers ) then

			-- Recalculate sha256 to verify it matches the sha256 in the importSource to
			-- verify we have the correct file to upload.

			return WFCrypto.asyncDigestFile( localPath, 'SHA256' )

				:flatMap( function( localSha256 )

					if localSha256 == sha256 then

						-- https://jira.corp.adobe.com/browse/WF-1902: This is considered
						-- a warning and not an error, since the SHA256 mismatch reported
						-- by Oz does not reflect what we see locally. When this occurs, we
						-- will eventually attempt the upload again.

						isWarning = true

						local failureInfo = {
							docId = docId,
							url = binaryUrl,
							responseData = responseData,
							headers = headers,
							binaryDetails = binaryDetails,
							contentType = contentType,
							fileSize = localSize,
							httpError = true,
							reason = "upon reception of sha256 mismatch from oz recalculation of sha matched",
							isCopy = false,
							isDirectUpload = isDirectUpload,
							warning = isWarning
						}

						self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

						return Observable.throw( responseData, headers, false, isWarning )

					else

						return self:applyInvalidOriginalOverlay( docId, localPath, binaryUrl, nil, "sha256 mismatch from Oz confirmed on recalculation" )
							:flatMap( function()
								return Observable.throw( responseData, headers, true, isWarning )
							end )

					end

				end )

		end


		if errorUploadMasterWhenNotAllowed( binaryDetails.type, responseData, headers ) then

			return self:applyRemoveWillNotUploadMasterOverlay( docId )

				:flatMap( function()

					-- Throw an error so we retry the upload after the doc store
					-- has had time to push the removal of the original through
					-- the changes feed.

					return Observable.throw( responseData, headers, true, isWarning )

				end )

		end

		if errorInsufficientPermissionForMaster( binaryDetails.type, responseData, headers ) then

			return self:applyRemoveLocalOnlyMasterFileOverlay( self._catalog, self._docStore, docId )

			:flatMap(function()

				-- Throw an error so we retry the upload after the doc store
				-- has had time to push the removal of the file path through
				-- the changes feed.

				return Observable.throw( responseData, headers, true, isWarning )

			end)

		end

		if errorPayloadMustExist( binaryDetails.type, responseData, headers ) then

			if isOriginalBinary( binaryDetails.type ) then

				return self:applyInvalidOriginalOverlay( docId, localPath, binaryUrl, nil, "payload must exist" )
				:flatMap( function()
					return Observable.throw( responseData, headers, true, isWarning )
				end )

			elseif binaryDetails.type == 'xmp_develop' then

				local fileAttributes = WFFileUtils.fileAttributes( localPath )
				local localFileSize = fileAttributes.fileSize

				if localFileSize and localFileSize == 0 then
					WFFileUtils.delete( localPath )
					logger:errorf( "resolving xmp_develop payload must exist by deleting empty file" )
					localSize = localFileSize
				end

			end

		end

		if errorContentTypeMismatch( responseData, headers ) then

			return self:applyInvalidOriginalOverlay( docId, localPath, binaryUrl, "UploadFailedContentTypeMismatch" )
				:flatMap( function()
					return Observable.throw( responseData, headers, true, isWarning )
				end )

		end

		if errorAuxShaDoesNotMatch(responseData, headers) then

			-- This should be added to requestInfo by the
			-- "thrower", but its best to be safe
			local errorURL = derefOpt(requestInfo, 'errorUrl') or 'unknown'

			-- Force a resync of the parent document and log an error
			self:handleAuxDoesNotMatch( errorURL, binaryDetails, docId, responseData, headers )

		end

		if errorAuxMissing( responseData, headers ) then

			return self:maybeApplyInvalidAuxOverlay( docId )

				:flatMap( function()

					-- Treat AuxMissingErrror as benign, as there are ADS workflows where
					-- we expect to get this error and recover from it.

					-- This can happen when uploading develop settings AFTER syncing metadata
					-- to Oz with ADS enabled. This occurs when the first revision synced to
					-- Oz has external develop settings. And in this case, Oz requires that
					-- aux be uploaded before the develop settings, but we don't enforce that.

					-- https://lightroom.slack.com/archives/GFLLW2SSC/p1718213327487519

					isWarning = true

					return Observable.throw( responseData, headers, true, isWarning )

				end )

		end

		local maybeAccountError = self:checkForAccountError( docId, binaryDetails.type, responseData, headers )

		if maybeAccountError then
			return maybeAccountError
		end

		return Observable.throw( responseData, headers, false, isWarning )

	end )

	:tap(
		function( response )

			self:recordSuccess( docId, binaryUrl, binaryDetails, localSize, response, contentType, false, isDirectUpload )

		end,
		nil,
		function()

			self:uploadCompleted( docId, binaryDetails, sha256 )

		end )

	:catch( function( responseData, headers, invalidOriginal, isWarning )

			-- On binary uploads of the xmp_develop where we request rendition generation we
			-- may get a 404 with a SourceDoesNotExist error from Oz if a master/proxy does
			-- not yet exist. In these cases, the upload succeeds, but the rendition generation
			-- does not. This failure can be treated as a success as we will request renditions
			-- again when the master/proxy is uploaded

			local uploadFailed = true

			if binaryDetails.type == 'xmp_develop' and headers and headers.status_code == '404' then

				local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, responseData )

				if ok and type( response ) == 'table' then

					if response.code == "SourceDoesNotExist" then

						self:recordSuccess( docId, binaryUrl, binaryDetails, localSize, responseData, contentType, false, isDirectUpload )
						self:uploadCompleted( docId, binaryDetails, sha256 )
						uploadFailed = false

					elseif response.subtype == 'ResourceNotFoundError' and response.code == 1000 then

						-- [jandrew 2022-03-16]: The following code was put into place to address
						-- https://jira.corp.adobe.com/browse/WF-1639. We were seeing repeated attempts
						-- to upload xmp_develop files for version documents that had been deleted
						-- on Oz. Now when we detect this scenario, we delete the local version.

						local maybeVersionError = derefOpt( response, 'errors', 'version', 1 )

						if maybeVersionError == 'does not exist' then

							self:checkAndUpdateDocIfDeleted( docId )

						end

					end

				end

			end

			if uploadFailed then

				self:handleUploadFailure( docId, binaryDetails, binaryUrl, optParentDocumentId, responseData,
						headers, invalidOriginal, contentType, localSize, false, isDirectUpload, isWarning )

			end

			return Observable.just( false )

	end )

	:catch( function()
		-- In case the processing in the above catch throws an unexpected error,
		-- this will catch it.
		return Observable.just( false )
	end )

end

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

--[[

	Instance Function

	\brief Copy a binary in place on Oz and record
	failure/success. see copyBinaryOnOz for details

	\param docId			(string) - The ID of the Document an upload is associated with
	\param uploadUrl		(string) - Destination for the upload
	\param binaryDetails		(table) - A BinaryDetails domain object - see top of file
	\param sha256			(string) - Sha256 of the uploaded binary
	\param contentType		(string) - The content type associated with the upload
	\param addlUploadHeaders	(table) - Extra headers to be associated with the upload
	\param sourceLinkForCopy	(string) - The URL associated with the source asset document
	\param optSpecialRendition	(bool) - Does this upload need a special rendition - see docNeedsSpecialProxyOrRenditionRequest
	\param optSpecialProxy			(bool) - Does this upload need a special proxy - see docNeedsSpecialProxyOrRenditionRequest
	\param optAuxId			(string) - The ID of the auxillary to be copied (if the copy command is copying an auxillary)

	\return Observable<> - observable responsible for issueing the copy command

]]
function WFCatBinaryUploader:copyOneBinary(
		docId, binaryUrl, binaryDetails, sha256, contentType,
		addlUploadHeaders, sourceLinkForCopy, optSpecialRendition,
		optSpecialProxy, optAuxId )

	local copyBinaryObservable = self:copyBinaryOnOz(
			docId, binaryUrl, binaryDetails, contentType,
			addlUploadHeaders, sourceLinkForCopy, optSpecialRendition,
			optSpecialProxy, optAuxId )

	copyBinaryObservable = copyBinaryObservable:tap(

		function( dataLength, response, headers )

			self:recordSuccess(docId, binaryUrl, binaryDetails,
					dataLength, response, contentType, true, false )

		end,
		nil, -- caller must issue call to handleUploadFailure
		function()
			self:uploadCompleted( docId, binaryDetails, sha256 )
		end )

	return copyBinaryObservable

end

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

local function isMalformedSourceLinkResponse( responseData, headers )

	return not not ( headers and headers.status_code == '400' and responseData and responseData.code == 1005 )

end

WFCatBinaryUploader.class._TEST_isMalformedSourceLinkResponse = function ( _, ... ) -- Exposed for testing
	return isMalformedSourceLinkResponse( ... )
end

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

local function handleMalformedSourceLinkInCopyRequest( self, docId, binaryDetails, auxId )

	local kDelete = { _delete = true }

	local maybeGetAuxDigest = Observable.just( false )

	if isAux( binaryDetails.type ) and auxId then

		maybeGetAuxDigest = self._docStore:docValue( docId )

				:map( function( docValue )
					return WFCatAssetAux.getDigestForAuxId( docValue, auxId )
				end )

	end

	return maybeGetAuxDigest

		:flatMap( function( maybeAuxDigest )

			local localOnly = {}

			if isXmpDevelop( binaryDetails.type ) then

				setNested( localOnly, kDelete, 'sourceXmpDevelopLink' )
				setNested( localOnly, kDelete, 'sourceXmpDevelopSha256' )

			elseif isXmpSidecar( binaryDetails.type ) then

				setNested( localOnly, kDelete, 'sourceXmpLink' )

			elseif isProxy( binaryDetails.type ) then

				setNested( localOnly, kDelete, 'sourceProxyLink' )

			elseif isAux( binaryDetails.type ) and maybeAuxDigest then

				setNested( localOnly, kDelete, 'sourceAuxLinks', maybeAuxDigest )

			elseif isOriginalBinary( binaryDetails.type ) then

				setNested( localOnly, kDelete, 'sourceOriginalLink' )

			else

				return Observable.just( false )

			end

			local overlay = {
				_id = docId,
				_localOnly = localOnly
			}

			return self._docStore:applyOverlay( overlay )

		end )

end

WFCatBinaryUploader.class._TEST_handleMalformedSourceLinkInCopyRequest = function ( _, ... ) -- Exposed for testing
	return handleMalformedSourceLinkInCopyRequest( ... )
end

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

--[[

	Instance Function

	\brief Copy a binary in place using OZ APIs rather than re-uploading an already present binary.

	\param docId			(string) - The ID of the Document an upload is associated with
	\param uploadUrl		(string) - Destination for the upload
	\param binaryDetails		(table) - A BinaryDetails domain object - see top of file
	\param contentType		(string) - The content type associated with the upload
	\param additionalUploadHeaders	(table) - Extra headers to be associated with the upload
	\param sourceAssetLink		(string) - The URL associated with the source asset document
	\param optSpecialRendition	(bool) - Does this upload need a special rendition - see docNeedsSpecialProxyOrRenditionRequest
	\param optSpecialProxy			(bool) - Does this upload need a special proxy - see docNeedsSpecialProxyOrRenditionRequest
	\param auxId			(string) - The ID of the auxillary to be copied (if the copy command is copying an auxillary)

	\return {int, string, table} or Observable<error> - {count of bytes, response data, response headers}

]]
function WFCatBinaryUploader:copyBinaryOnOz( docId, uploadUrl, binaryDetails, contentType,
		additionalUploadHeaders, sourceAssetLink, optSpecialRendition,
		optSpecialProxy, auxId )

	logger:debugf( "Copying %s for docId %s from %s to %s",
			binaryDetails.type, docId, sourceAssetLink, uploadUrl )

	local data = '{ "file_href": "' .. sourceAssetLink .. '" }'

	if isAux( binaryDetails.type ) then

		-- SPAM [jandrew 2022-08-11]: The following is not as efficient as
		-- as it could be. Because of the way the binary upload is structured,
		-- we process the uploading of one binary at a time. However, the
		-- Aux copy API supports copying over N auxiliaries at a time; this is
		-- an optimization we may want to consider in the future.

		data = string.format( '{ "source_href": "%s", "aux_ids": [ "%s" ] }', sourceAssetLink, auxId )

	end

	local isDirectUpload = false
	local options = getRequestOptionsFor( binaryDetails.type, contentType, true, additionalUploadHeaders,
		isDirectUpload, self:isAtomicDevelopSettingsEnabled(), optSpecialRendition, optSpecialProxy )

	options.requestCost = WFRxHttpClient.requestCostLow
	options.dataFormat = nil
	options.notifyWhenHttpRequestIsInProgress = true
	options.includeProgress = self._includeProgress

	local requestInProgress = false

	local function progressEventHandler( _, _, progress )

		if not requestInProgress then

			self:markUploadAsInProgress( docId, binaryDetails )
			requestInProgress = true

		end

		setNested( self._inFlightUploadProgress, progress, docId, binaryDetails.type, binaryDetails.sha256 )
		self._debugInFlightUploadProgressUpdateStream:onNext( self._inFlightUploadProgress )

	end

	return self._catalogSpec:putBinary( uploadUrl, data, options )

		:tap( progressEventHandler )

		:takeLast( 1 )

		:map( function( response, headers )
			return #data, response, headers
		end )

		:catch( function( responseData, headers )

			if isMalformedSourceLinkResponse( responseData, headers ) then

				-- Check for 400 error with code 1005, indicating a malformed source link.

				return handleMalformedSourceLinkInCopyRequest( self, docId, binaryDetails, auxId )

					:flatMap( function()

						-- Throw an error so we retry the upload after the doc store
						-- has had time to push the removal of the source link through
						-- the changes feed.

						return Observable.throw( responseData, headers, true )

					end )

			elseif errorUploadMasterWhenNotAllowed( binaryDetails.type, responseData, headers ) then

				return self:applyRemoveWillNotUploadMasterOverlay( docId )

					:flatMap( function()

						-- Throw an error so we retry the upload after the doc store
						-- has had time to push the removal of the original through
						-- the changes feed.

						return Observable.throw( responseData, headers, true )

					end )

			end

			if errorInsufficientPermissionForMaster( binaryDetails.type, responseData, headers ) then

				return self:applyRemoveLocalOnlyMasterFileOverlay( self._catalog, self._docStore, docId )

					:flatMap( function()

						-- Throw an error so we retry the upload after the doc store
						-- has had time to push the removal of the file path through
						-- the changes feed.

						return Observable.throw( responseData, headers, true )

					end)

			end

			local maybeAccountError = self:checkForAccountError( docId, binaryDetails.type, responseData, headers )

			if maybeAccountError then
				return maybeAccountError
			end

			return Observable.throw( responseData, headers, false )

		end )

end

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

--[[

	Instance Function

	\brief Issues an upload for a proxy or master in a particular
	set of circumstances. TODO: More detail unknown, ask @jandrew when possible

	\param docId (string) - The ID of the Document an upload is associated with
	\param localFilePath (string) - The path to the binary to upload
	\param localFileSize   (int) - The size of the binary being uploaded
	\param binaryUrl		(string) - The OZ provided destination URL for the binary
	\param binaryDetails (table) - A BinaryDetails domain object - see top of file
	\param contentType	(string) - The content type associated with the upload
	\param addlUploadHeaders	(table) - Extra headers to be associated with the upload
	\param sha256        (string) - Sha256 of the uploaded binary
	\param isFirstUploadForSha256 (bool) - is this the first time this sha256 has been uploaded?
	\param uploadPriority   (int) - The priority of this upload
	\param isDirectUpload	(bool)   - Is this a DCU upload or not?
	\param optSpecialRendition	(bool) - Does this upload need a special rendition - see docNeedsSpecialProxyOrRenditionRequest
	\param optSpecialProxy			(bool) - Does this upload need a special proxy - see docNeedsSpecialProxyOrRenditionRequest


	\return Observable<uploadOneBinary> or nil or Observable<bool>

]]
function WFCatBinaryUploader:uploadOneMasterOrProxy( docId, localFilePath, localFileSize, binaryUrl, binaryDetails, contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload, optSpecialRendition, optSpecialProxy )

	-- Upload the master.

	if ( isFirstUploadForSha256
			or isWatermarkOriginal( binaryDetails.type )
			or not not self._capabilitiesCache:get( 'WFCatBinaryUploaderTest.debug_allowSameSha256Upload' ) ) then

		if localFilePath then

			local optDevelopUserUpdated = nil
			local optParentDocumentId = nil
				-- never needed for proxy or master upload

			return self:uploadOneBinary(
					docId, localFilePath, localFileSize, binaryUrl, binaryDetails, contentType, addlUploadHeaders,
					sha256, uploadPriority, isDirectUpload, optDevelopUserUpdated, optParentDocumentId,
					optSpecialRendition, optSpecialProxy )

		else

			-- Copied an binary that didn't have a master and source was deleted before we could upload it.

			return self:applyInvalidOriginalOverlay( docId, nil, binaryUrl, "CopyFailedSourceMissing" )
				:finally( function()
					-- delay can be shorter as we won't re-issue any http request if we
					-- didn't wait long enough, we will just end up back here.
					self:delayedClearFromRecentlyUploaded( "upload failed, wait for docstore update as CopyFailedSourceMissing", docId, binaryDetails, self._waitForCopyFailedSeconds )
				end )

		end

	else

		self:delayedClearFromRecentlyUploaded_NoUpload( "Uploading first asset with same sha256, delay so we can copy later", docId, binaryDetails )
		return Observable.just( true )

	end

end

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

--[[

	Instance Function

	\brief Upload or Copy a Master or Proxy, keep track of any internal state needed. TODO: More detail needed - this function is super opaque

	\param docId			(string) - The ID of the Document an upload is associated with
	\param localPath		(string) - The path to the binary-to-upload on disk
	\param localSize		(int) - The size of the binary-to-upload on disk in bytes
	\param binaryUrl		(string) - The OZ provided destination URL for the binary
	\param copyBinaryUrl		(string) - The OZ provided copy URL for the binary
	\param binaryDetails		(table) - A BinaryDetails domain object - see top of file
	\param contentType		(string) - The content type associated with the upload
	\param addlUploadHeaders	(table) - Extra headers to be associated with the upload
	\param sha256			(string) - Sha256 of the uploaded binary
	\param isFirstUploadForSha256	(bool) - is this the first upload for this sha?
	\param uploadPriority		(int) - The priority of this upload
	\param isDirectUpload		(bool)   - Is this a DCU upload or not?
	\param optAuxId			(string) - The id of the aux to copy - if the upload type is aux

	\return Observable<>

]]
function WFCatBinaryUploader:uploadOrCopyOneMasterOrProxy(
		docId, localFilePath,localFileSize, binaryUrl, copyBinaryUrl,
		binaryDetails, contentType, addlUploadHeaders, sha256,
		isFirstUploadForSha256, uploadPriority, isDirectUpload, optAuxId )

	if isWatermarkOriginal( binaryDetails.type ) then
		return self:uploadOneMasterOrProxy( docId, localFilePath, localFileSize, binaryUrl, binaryDetails,
						contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload )
	end

	return WFCatImportDuplicateFiles.findUploadedAssetWithSha256HintNotDocId( self._catalog, sha256, docId )
		:flatMap( function( sourceAssetId, sourceHasNoMaster )

			if sourceAssetId == nil then

				local optSpecialRendition, optSpecialProxy = false, false

				if WFPlatformInfo.isWindows and WFOzContentTypes.isVideo( contentType ) then

					return self._docStore:docValue( docId )
					:flatMap( function( docValue )
						-- For windows, check if
						optSpecialRendition, optSpecialProxy = docNeedsSpecialProxyOrRenditionRequest( docValue )
						return self:uploadOneMasterOrProxy( docId, localFilePath, localFileSize, binaryUrl, binaryDetails,
								contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload, optSpecialRendition, optSpecialProxy )

					end )

				else

					return self:uploadOneMasterOrProxy( docId, localFilePath, localFileSize, binaryUrl, binaryDetails,
						contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload, optSpecialRendition, optSpecialProxy )

				end
			end

			-- Copy the asset.
			return self._docStore:docValue( sourceAssetId )
			:flatMap( function( sourceDocValue )

				-- SourceAssetLink may be nil if the source asset has an invalid original.
				local sourceAssetLink
				if isOriginalBinary( binaryDetails.type ) then
					sourceAssetLink = WFCatFileSpec.getOriginalLinkFromDocValue( sourceDocValue )
				else
					sourceAssetLink = WFCatFileSpec.getProxyLinkFromDocValue( sourceDocValue )
				end

				local optSpecialRendition, optSpecialProxy = docNeedsSpecialProxyOrRenditionRequest( sourceDocValue )

				if sourceAssetLink == nil then

					if localFilePath then

						return self:uploadOneBinary( docId, localFilePath, localFileSize, binaryUrl, binaryDetails, contentType, addlUploadHeaders, sha256, uploadPriority, isDirectUpload )

					else

						-- Copied an asset that didn't have a master that was deleted before we could upload it.
						return self:applyInvalidOriginalOverlay( docId, nil, binaryUrl, "CopyFailedSourceMissing" )
							:finally( function()
								self:delayedClearFromRecentlyUploaded( "copy failed, wait for docstore update CopyFailedSourceMissing", docId, binaryDetails )
							end )

					end

				else

					return self:copyOneBinary(
							docId, copyBinaryUrl, binaryDetails,
							sha256, contentType, addlUploadHeaders, sourceAssetLink,
							optSpecialRendition, optSpecialProxy, optAuxId )

						:tapOnError( function( responseData, headers )

							local parentDocumentId = nil
								-- never needed for proxy or master upload

							self:handleUploadFailure( docId, binaryDetails, copyBinaryUrl, parentDocumentId,
									responseData, headers, false, contentType, localFileSize, true, false )

						end )

				end

			end )

		end )

end

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

--[[

	Instance Function

	\param Desktop specific (?) - checks the status of a maybe
	offline volume and determines if a binary path is accessible
	or not. TODO: follow up with @jandrew on this when possible.

	\param localFileVolumeGuid	(string) - The GUID of the volume where a binary is stored
	\param localFileRelativePath	(string) - The path to a binary on an external volume
	\param localFilePath		(string) - The path to a binary on the local disk
	\param binaryDetails		(table) - A BinaryDetails domain object - see top of file

	\return {string, bool} - {path to binary, is the asset accessible?}

]]
function WFCatBinaryUploader:getBinaryFullPathAndAssetOfflineStatus( localFileVolumeGuid, localFileRelativePath, localFilePath, binaryDetails )

	local binaryFullPath
	-- assetIsOffline status applicable only to Desktop. for other platforms simply return false.
	-- assetIsOffline status is always false for watermark
	local assetIsOffline = false

	if isAux( binaryDetails.type ) then

		binaryFullPath = localFilePath and WFCatAssetAux.getAbsolutePath( self._catalog, localFilePath )

	elseif localFileVolumeGuid and localFileRelativePath then

		if isWatermarkOriginal( binaryDetails.type ) then

			binaryFullPath = WFFileUtils.createFullPathFromRelativePathAndCatalogPath( self._catalog, localFileRelativePath, binaryDetails.type )

		else

			local success, filePath = pcall(
				WFFileUtils.createFullPathFromRelativePathAndGuid,
				localFileRelativePath, localFileVolumeGuid )

			if success and filePath then
				binaryFullPath = localFilePath or filePath
			else
				assetIsOffline = true
			end

		end

	elseif WFPlatformInfo.isMobile() then

		-- SPAM [matthewj 2020-01-30]: Mobile does not normally use full file paths. However, there are some
		-- binary uploader tests which currently do put full file paths in so this check will only get the
		-- full binary path if the path is not a relative path. Ideally though, the tests are updated to
		-- properly populate the paths on mobile as catalog relative paths. See WF-880 for a story that
		-- will help remove the first half of this if statement that checks if localFilePath is an absolute
		-- path. UPDATE [jkamerer] 2023-01-13: As part of investigation into WF-2129 saw scenarios where
		-- first upload attempt failed with error, then on retry the path had already been converted to
		-- a full path, meaning there are workflows where this check is helpful outside of tests.
		if localFilePath and localFilePath:sub(1,1) == '/' then
			binaryFullPath = localFilePath
		else
			local OzFilePaths = require 'OzFilePaths'
			local LrmAssetUtils = require 'LrmAssetUtils'

			local catalogBasePath = OzFilePaths.getCatalogBasePath( self._catalog:ozCatalogId() )
			local fileFullPath = LrmAssetUtils.createFullPath( catalogBasePath, localFilePath, binaryDetails.type )
			if fileFullPath then
				binaryFullPath = fileFullPath
			end
		end

		if WFPlatformInfo.isAndroid then
			local OzCatalogUploadStatus = require 'OzCatalogUploadStatus'

			-- This code corrects the local path based on which storage is being used (Device/SD Card) before it is being picked up for upload
			binaryFullPath = OzCatalogUploadStatus.checkAndCorrectLocalPath(binaryFullPath)
		end

	end

	return binaryFullPath, assetIsOffline

end

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

--[[

	Instance Function

	\brief Deletes a document from the docstore

	\param docId (string) - The ID of a Document

	\return Observable<revisionID> - resolves with the new
	documents revision ID when the write completes

	TODO: Follow up that this is actually correct.

]]
function WFCatBinaryUploader:deleteAsset( docId )

	logger:debugf( "deleteAsset: delete the docId = %s", docId )

	return self._docStore:applyOverlay {
		_id = docId,
		_deleted = true
	}

end

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

--[[
n
	Instance Function

	\brief Mobile Only: handle failures to upload when the master
	is not present on the local system. For iOS this is a hard
	upload failure, for android it is potentially recoverable
	depending on user settings. See implementation for details.

	\param docId		(string) - The ID of the Document an upload is associated with
	\param binaryUrl	(string) - The OZ provided destination URL for the binary
	\param binaryDetails	(table) - A BinaryDetails domain object - see top of file
	\param contentType	(string) - The content type associated with the upload

	\return Observable<revisionID> - deletion observable - might be a tombstone TODO: follow up with @jandrew

]]
function WFCatBinaryUploader:handleMasterNotExistsLocally( docId, binaryUrl, binaryDetails, contentType )

	logger:warnf( "Trying to upload %s for %s but original is missing.", binaryDetails.type, docId )

	return self._docStore:docValue( docId )
		:flatMap( function( docValue )

			local reason = 'binary_upload_missing_original'

			if WFPlatformInfo.isiOS then

				local failureInfo = {
					docId = docId,
					url = binaryUrl,
					binaryDetails = binaryDetails,
					contentType = contentType,
					httpError = false,
					reason = reason
				}

				self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

				return self:deleteAsset( docId )

			elseif WFPlatformInfo.isAndroid then

				local localUrl = derefOpt( docValue, '_localOnly', 'localUrl' )
				local localAssetId = derefOpt( docValue, 'importSource', 'localAssetId' )
				local uniqueDeviceId = derefOpt( docValue, 'importSource', 'uniqueDeviceId' )

				if localAssetId and ( uniqueDeviceId == WFDeviceInfo.uniqueDeviceId() ) then
					localUrl = localAssetId
				end

				if localUrl == nil then
					return Observable.just( true );
				end

				if not WFFileUtils.exists( localUrl ) then -- marking asset as deleted as file locally was deleted

					local failureInfo = {
						docId = docId,
						url = binaryUrl,
						binaryDetails = binaryDetails,
						contentType = contentType,
						httpError = false,
						reason = reason
					}

					self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

					return self:deleteAsset( docId )

				end

				-- In Android user has to enable permission for storage, camera, location etc.
				-- It may so happen that these permissions may not be enabled.
				-- And also WFCrypto.digestFile() assumes it will be invoked on files that has read permission.
				-- So better check before invoking
				local isFileReadable = WFFileUtils.isFileReadable( localUrl )

				if not isFileReadable then
					logger:debugf( "File exist but no read permission. leave it in queue" )
					return Observable.just( true )
				end

				local sha256 = docValue.importSource and docValue.importSource.sha256
				sha256 = sha256 and string.upper( sha256 )

				local sha256Digest = WFCrypto.digestFile(localUrl, 'SHA256')
				local expectedSHA256 = sha256Digest and string.upper( sha256Digest )

				if expectedSHA256 and sha256 ~= expectedSHA256 then -- marking asset as deleted as file locally was modified

					local failureInfo = {
						docId = docId,
						url = binaryUrl,
						binaryDetails = binaryDetails,
						contentType = contentType,
						httpError = false,
						reason = reason
					}

					self:WFAnalyticsTrackEventNotifyFailure( failureInfo )

					return self:deleteAsset( docId )

				end

				return self._docStore:applyOverlay {
					_id = docId,
					_localOnly = {
						originalFilePath = localUrl,
					}
				}

			end

		end )

end

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

--[[

	Instance Function

	\brief Resets all internal variables associated with upload progress.

	\returns nil - side effect: reset internal variables.

]]
function WFCatBinaryUploader:resetDetailedProgressReportCounters()

	if not self._uploadProgress then
		return
	end

	if( self._uploadProgress.totalBytesNeedToUpload ~= 0 and
		self._uploadProgress.totalBytesUploadedSoFar ~= 0 ) then

		-- reset the upload progress counters
		self._uploadProgress.totalBytesUploadedSoFar = 0
		self._uploadProgress.totalBytesNeedToUpload = 0

		local binaryUploadProgress = {
			totalBytesNeedToUpload = self._uploadProgress.totalBytesNeedToUpload,
			totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar,
		}
		self._uploadProgress.progressReporter:updateProgressWithInfo( binaryUploadProgress, true )

	end

end

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

--[[

	Instance Function

	\brief Finalizes progress report for uploads and then resets.

	\returns nil - side effect: reset internal variables.

]]
function WFCatBinaryUploader:sendFinalDetailedProgressUpdate()

	if not self._uploadProgress then
		return
	end

	if( self._uploadProgress.totalBytesNeedToUpload ~= 0 and
		self._uploadProgress.totalBytesUploadedSoFar ~= 0 and
		self._uploadProgress.totalBytesNeedToUpload - self._uploadProgress.totalBytesUploadedSoFar >= 0 ) then

		local binaryUploadProgress = {
			totalBytesNeedToUpload = self._uploadProgress.totalBytesNeedToUpload,
			totalBytesUploadedSoFar = self._uploadProgress.totalBytesUploadedSoFar,
		}
		self._uploadProgress.progressReporter:updateProgressWithInfo( binaryUploadProgress, true )
		self:resetDetailedProgressReportCounters()

	end

end

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

--[[

	\brief Reload document from Oz to refresh the links section of the
	document. This is necessary when attempting to leverage a newer feature
	like direct upload or atomic develop settings when the asset document
	has not synced any update from Oz since before Oz implemented the
	template. In this case, refreshing the document from Oz will contain
	all the latest templates and unblock uploading using the new feature.

	\param docId (string, uuid): Document ID
	\param shouldReprocessVersions (boolean) In the case of an asset being
		reloaded to get the template link for versions, if the asset does
		not have any binaries to upload, updating the asset will NOT trigger
		new data from the candidates observables, so trigger a reprocess of
		the list using the _shouldReprocessVersionsStream

	\returns Observable

]]
function WFCatBinaryUploader:reloadLinksForDocument( docId, shouldReprocessVersions )

	return self._docStore:docValue( docId )

		:flatMap( function( localDocValue )

			return WFRxRetryObservable { maxAttempts = 3, retryDelay = WFPlatformInfo.isUnitTest() and 0.1 or 10, factory = function()

				local options = { bypassRateLimiter = true, }

				local documentUrl = KSTableUtils.derefOpt( localDocValue, '_serverProvidedMetadata', 'links', 'self', 'href' )

				if not documentUrl then
					return Observable.throw( "reloadLinksForDocument: No document URL available." )
				end

				return self._catalogSpec:get( documentUrl, options )

					:flatMap( function( result, headers )

						local decodedDoc = KSJsonDecoder.decodeWebSafeJson_stripNull( result )
						local ozDocValue = WFOzDocUtils.remapBasicOzDocument( decodedDoc, decodedDoc.base )

						-- just want the overlay to update the links section, not updating anything else

						local wholeOverlay = require 'WFTableMerge'.createOverlay( localDocValue, ozDocValue )
						local overlay = {

							_id = docId,

							_serverProvidedMetadata = {
								links = derefOpt( wholeOverlay, '_serverProvidedMetadata', 'links' )
							}

						}

						local ozMainPartsRequestUrlTemplate = derefOpt( overlay, '_serverProvidedMetadata', 'links',
							'/rels/main_create/parts', 'href' )

						-- if original binary direct upload link is STILL not available, might be that this
						-- document type or subtype does not support direct upload. Mark the document
						-- isSkipDirectUpload so we do not infinite loop on reloading the links.

						if not ozMainPartsRequestUrlTemplate then
							overlay._localOnly = {
								binaryUpload = {
									master = {
										isSkipDirectUpload = true
									}
								}
							}
						end

						return self._catalog:docStore():applyOverlay( overlay )

							:map( function()

								if shouldReprocessVersions then
									self._shouldReprocessVersionsStream:onNext( true )
								end

							end )

					end )

					:catch( function( responseData, headers )

						WFAnalytics.trackEvent {
							['event.type'] = 'error',
							['event.appWorkflow'] = 'SYNC',
							['event.subcategory'] = 'Document_link_refresh',
							['event.error_code'] = headers and headers.statusCode,
							['content.id'] = docId,
						}

						logger:errorf( "reloadLinksForDocument error %s %s", headers.statusCode, responseData )

						return Observable.throw( responseData, headers )

					end )

				end }

			end )

end

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

local function maybeFixUpOrRemoveCandidates( docStore, candidates,
		atomicDevelopSettingsEnabled, reloadLinksForDocumentStream )

	-- WF-2131: When uploading the develop setting for a Version, the associated
	-- auxiliaries within its parents Asset document must be fully synced to Oz,
	-- or else Oz will reject the develop settings and return an AuxMissingError.
	-- One mechanism we have in place to enforce this is the parentDocumentId
	-- field within the needsBinaryUpload table; we never upload binaries from
	-- a document if it's parent document has binaries within the upload queue.
	-- However, there exist certain edge cases where the parent document does not
	-- have queued uploads, nor are its binaries fully synced. To deal with this,
	-- we must check the state of the auxiliaries of the parent document before
	-- attempting the Version develop settings upload.

	-- It is important to note that we are not able to perform this filtering
	-- within the index store map function, since it requires knowledge about
	-- more than one document at a time. It's also not easy to do this filtering
	-- at the SQLite level without adding additional columns and clauses to
	-- all of the existing needsBinaryUpload queries.

	-- IMPORTANT: The parentDocumentId field is ONLY populated for entries in
	-- the needsBinaryUpload table when the document is a Version, and the binary
	-- needing upload is the develop settings.

	-- WF-2229: As part of atomic develop settings, need to grab template link
	-- to upload xmp develop for versions from the asset document, so extending
	-- this method which is already looking up all of these documents to also
	-- set up the template link.

	-- WF-2234: When template link is unavailable, asynchronously reload asset
	-- document to get the template link. This is required for atomic develop
	-- settings upload for versions and also for assets, where xmp develop
	-- must be uploaded directly before revision can sync. While not strictly
	-- necessary for ADS, also extending this to original binary upload case
	-- so will not automatically fallback to Oz upload when template link
	-- is not available but will instead refresh and get template link.

	-- First create a list of Version document IDs for which we need to fetch
	-- the parent Asset documents, while at the same time checking for original
	-- and xmp develop uploads for assets that are missing direct upload links.

	local assetDocsToReload = {}
	local candidateIndicesByParentDocumentId = {}

	for index, uploadCandidate in ipairs( candidates ) do

		-- uploadCandidate can be false if it's stripped out in a prior
		-- run of maybeFixUpOrRemoveCandidates.

		local parentDocumentId = uploadCandidate and uploadCandidate[ 19 ]

		if parentDocumentId then

			local candidateIndices = candidateIndicesByParentDocumentId[ parentDocumentId ] or {}
			candidateIndices[ #candidateIndices + 1 ] = index

			candidateIndicesByParentDocumentId[ parentDocumentId ] = candidateIndices

		elseif uploadCandidate then

			local assetId = uploadCandidate[ 1 ]
			local binaryType = uploadCandidate[ 4 ]
			local spaceId = uploadCandidate[ 15 ]
				-- direct upload of original binary for space contributors has not been implemented
			local directUploadLink = uploadCandidate[ 16 ]

			if ( ( isDevelopBinary( binaryType ) and atomicDevelopSettingsEnabled ) or
					( isImageOrVideoOriginal( binaryType ) and not spaceId ) )
				and directUploadLink == nil and not uploadCandidate.skipForNow then
						-- explitly checking for isImageOrVideoOriginal() instead of just isOriginalBinary()
						-- here as do not want to apply the check for watermark, profile, preset etc

				uploadCandidate.skipForNow = true
				assetDocsToReload[ assetId ] = false
					-- setting to false indicates that shouldReprocessVersions should be set to false
					-- when trigging call to reloadLinksForDocument()

			end

		end

	end

	local function asyncReloadLinks(...)

		if reloadLinksForDocumentStream then

			for docId, shouldReprocessVersions in pairs( assetDocsToReload ) do
				reloadLinksForDocumentStream:onNext( docId, shouldReprocessVersions )
			end

		end

		return ...

	end

	if not next( candidateIndicesByParentDocumentId ) then
		-- done processing versions, just reload links as needed
		return Observable.just( candidates )
			:map( asyncReloadLinks )
	end

	local parentDocumentIds = {}
	for parentDocumentId, _ in pairs( candidateIndicesByParentDocumentId ) do
		parentDocumentIds[ #parentDocumentIds + 1 ] = parentDocumentId
	end

	-- Fetch the parent Asset documents, then remove the Version upload candidates
	-- if the parent Asset document contains unsynced auxiliaries.

	return docStore:bulkDocValue( parentDocumentIds )

		:map( function( parentDocValues )

			for _, parentDocValue in ipairs( parentDocValues ) do

				local uploadsRequired = WFCatAssetAux.uploadsRequired( parentDocValue )

				local ozXmpDevelopPartsRequestUrlTemplate, shouldFlagCandidates

				local shouldRemoveCandidates = uploadsRequired or parentDocValue._deleted

				if not shouldRemoveCandidates and atomicDevelopSettingsEnabled then

					local parentHasSelfLink =
							derefOpt( parentDocValue, '_serverProvidedMetadata', 'links', 'self', 'href' )

					if assetDocsToReload[ parentDocValue._id ] == false or not parentHasSelfLink then

						shouldRemoveCandidates = true
							-- Case 1: if we are already reloading the links for upload of an asset binary,
							-- then just remove version from the candidates array as it will be
							-- added back when the refresh of the asset document updates the list
							-- of upload candidates
							-- Case 2: Parent has no self link, meaning it's metadata has not been uploaded.
							-- In this case, we cannot upload this candidate nor can we refresh the
							-- parents links. We have to wait until the parent Asset metadata
							-- syncs.

					else

						ozXmpDevelopPartsRequestUrlTemplate =
							derefOpt( parentDocValue, '_serverProvidedMetadata', 'links', '/rels/versions/xmp_develop_create/parts', 'href' )

						shouldFlagCandidates = not ozXmpDevelopPartsRequestUrlTemplate

						if shouldFlagCandidates then
							assetDocsToReload[ parentDocValue._id ] = true
								-- setting to true indicates that shouldReprocessVersions should be set to true
								-- when trigging call to reloadLinksForDocument()
						end

					end

				end

				if shouldRemoveCandidates or ozXmpDevelopPartsRequestUrlTemplate or shouldFlagCandidates then

					local candidateIndices = candidateIndicesByParentDocumentId[ parentDocValue._id ] or {}

					local flaggedAtLeastOneCandidate = false

					for _, candidateIndex in ipairs( candidateIndices ) do

						if shouldRemoveCandidates then

							candidates[ candidateIndex ] = false
								-- We intentionally leave entries in the table and set them to false
								-- to maintain our ability to itereate through the table using ipairs.
								-- This is necessary, as the order we process the entries is critical.

						elseif ozXmpDevelopPartsRequestUrlTemplate then

							local versionCandidate = candidates[ candidateIndex ]

							-- link may have already been populated by a previous run of
							-- maybeFixUpOrRemoveCandidates() in which case do not repopulate

							if versionCandidate and versionCandidate[ 16 ] == nil then

								local versionId = versionCandidate[ 1 ]
								local sha256 = versionCandidate[ 8 ]

								local _, partialTemplate = WFURI.replaceFromOzTemplate( ozXmpDevelopPartsRequestUrlTemplate, {
									version_id = versionId,
									sha256 = sha256,
									upload_status = "all",
									content_type  = WFURI.escape( kContentTypeXML, 'http_query_key' )
								} )

								versionCandidate[ 16 ] = partialTemplate
									-- replace the mainPartsRequestUrl stored in needsbinaryupload table which
									-- will be nil with the partially completed template

								versionCandidate.skipForNow = nil
									-- clear skipForNow flag in case it was set in a previous run of this function

							end

						elseif shouldFlagCandidates then

							if candidates[ candidateIndex ].skipForNow then

								assetDocsToReload[ parentDocValue._id ] = nil
									-- In this case a reload of the links for the parent asset document
									-- will have already been triggered in a previous call to this method,
									-- so avoid triggering it again and wait for it to complete.

							elseif candidates[ candidateIndex ][ 16 ] == nil then
								-- do not set skipForNow if candidate already has something written to
								-- parts template link slot (would be special value to avoid direct upload,
								-- mainPartsRequestUrlNotAvailable, WF-2429

								flaggedAtLeastOneCandidate = true

								candidates[ candidateIndex ].skipForNow = true
									-- When reloading the asset to make the template link available, we
									-- need to keep the candidates in the list but mark them such that
									-- we will skip them until the template link is available. Note that
									-- this data is only on the Lua table held in memory and never saved
									-- to the SQL table, which is unnecessary because when new data is
									-- available in SQL, it will be reprocessed through this function
									-- again.

							end

						end

					end

					if shouldFlagCandidates and not flaggedAtLeastOneCandidate then

						assetDocsToReload[ parentDocValue._id ] = nil
							-- This is the case in which the asset did not have the version direct
							-- upload template link, however every version was marked with
							-- mainPartsRequestUrlVersionAlreadySynced. While we could safely
							-- reload the asset, it is not strictly necessary and reloading it
							-- without marking any versions skipForNow makes running this method
							-- on the same candidates list a second time non-idempotent. In fact
							-- running a second time would trigger a second refresh of the asset
							-- which is not desirable, so do this to keep the idempotence. WF-2429

					end

				end

			end

			return candidates

		end )
		:map( asyncReloadLinks )

end

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

local function hasCandidateInCandidateList( candidates )

	for _, uploadCandidate in ipairs( candidates ) do

		-- There can exist entries in this table that have keys, but are set to false.
		-- These entries are not candidates and must be ignored.

		if uploadCandidate and not uploadCandidate.skipForNow then
			return true
		end

	end

	return false

end

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

--[[

	Instance Function

	\brief Get a stream of binaries which need to be uploaded

	\return Observable<table of rows> - a stream where each member
	is an array based table corresponding to columns referenced in
	the select statement

]]
function WFCatBinaryUploader:getStandardBinaryUploadCandidatesStream()

	local typeOfJoin = getTypeOfJoinForSpaceTable( self._onlyUploadSpaceAssets )

	local isReadyToUpload
	if self:isAtomicDevelopSettingsEnabled() then
		isReadyToUpload = kReadyToUploadAtomic
	else
		isReadyToUpload = kReadyToUploadLegacy
	end

	local maybeForceDocIdUploadOrdering = self._debugForceDocIdUploadOrdering and "needsBinaryUpload.docId" or "RANDOM()"

	local needsUploadQueryStr = string.format(
			[[SELECT needsBinaryUpload.docId, localFilePath, uploadUrl, binaryType,
					contentType, uploadHeaders, isDenied, sha256, fileSize,
					sha256Error, sourceLinkForCopy, localVolumeGuid, localRelativePath,
					priorityForUpload, spaceId, mainPartsRequestUrl, copyUrl, auxId,
					parentDocumentId, developUserUpdated
				FROM needsBinaryUpload
				%s JOIN space ON space.docId == needsBinaryUpload.spaceId
				WHERE isReadyToUpload & ?
					AND ( parentDocumentId IS NULL OR parentDocumentId NOT IN ( SELECT docId FROM needsBinaryUpload ) )
				ORDER BY fileMaybeOnOfflineVolume ASC, isDenied ASC,
						priorityForUpload DESC, importDate ASC, %s
				LIMIT 500]], typeOfJoin,
				maybeForceDocIdUploadOrdering )

	local binaryUploadCandidatesStream = self._catalog:indexStore():query {
			sql = needsUploadQueryStr,
			quiesceUntilUpToDate = true,
			calm = binaryUploadQueryCalmValue,
			isReadyToUpload
		}

	return Observable.combineLatest(
		maybePerformanceWrap( 'query', binaryUploadCandidatesStream ),
		self._shouldReprocessVersionsStream:startWith( true ) )

		:flatMap( function( uploadCandidates )
			return maybeFixUpOrRemoveCandidates( self._docStore, uploadCandidates,
					self:isAtomicDevelopSettingsEnabled(), self._reloadLinksForDocumentStream ) -- WF-2131
		end )

end

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

--[[

	Instance Function

	\brief Get a stream of shared binaries which need to be
	uploaded TODO: Follow up on the definition of "shared"

	\return Observable<table of rows> - a stream where each member
	is an array based table corresponding to columns referenced in
	the select statement

]]
function WFCatBinaryUploader:getOwnerSharedBinaryUploadCandidatesStream()

	if self._onlyUploadSpaceAssets then

		-- If _onlyUploadSpaceAssets is true, then we are only allowed to upload
		-- assets into shared albums that the user is a member of, not the ones
		-- they they own.

		return Observable.just( {} )

	end

	local albumAssetTableName = 'album_asset'

	if WFPlatformInfo.isDesktop then
		albumAssetTableName = 'album_asset_v2'
	end

	local candidatesStream = WFCatCollaboration.getSharingStatusByAlbumId( self._catalog )

		:flatMapLatest( function( sharingStatusByAlbumId )

			local sharedAlbumIds = {}

			for albumId, sharingInfo in pairs( sharingStatusByAlbumId ) do

				local isShared = sharingInfo and sharingInfo.isShared

				if isShared and #sharedAlbumIds < 999 then -- 999 = SQLITE_MAX_VARIABLE_NUMBER
					sharedAlbumIds[ #sharedAlbumIds + 1 ] = albumId
				end

			end

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

			local isReadyToUpload
			if self:isAtomicDevelopSettingsEnabled() then
				isReadyToUpload = kReadyToUploadAtomic
			else
				isReadyToUpload = kReadyToUploadLegacy
			end

			local maybeForceDocIdUploadOrdering = self._debugForceDocIdUploadOrdering and "needsBinaryUpload.docId" or "RANDOM()"

			local ownerSharedBinaryUploadCandidatesQueryStr = string.format(
				[[SELECT needsBinaryUpload.docId, localFilePath, uploadUrl, binaryType,
						contentType, uploadHeaders, isDenied, sha256, fileSize,
						sha256Error, sourceLinkForCopy, localVolumeGuid, localRelativePath,
						priorityForUpload, spaceId, mainPartsRequestUrl, copyUrl, auxId,
						parentDocumentId, developUserUpdated
					FROM needsBinaryUpload
					INNER JOIN %s ON %s.assetId == needsBinaryUpload.docId
					WHERE isReadyToUpload & ?
						AND albumId IN ( %s )
						AND ( parentDocumentId IS NULL OR parentDocumentId NOT IN ( SELECT docId FROM needsBinaryUpload ) )
					ORDER BY fileMaybeOnOfflineVolume ASC, isDenied ASC,
							priorityForUpload DESC, importDate ASC, %s
					LIMIT 500]],
				albumAssetTableName, albumAssetTableName,
				WFSQLiteUtils.createVariableArgList( #sharedAlbumIds ),
				maybeForceDocIdUploadOrdering )

			local queryParams = sharedAlbumIds
			table.insert( queryParams, 1, isReadyToUpload )
			queryParams.sql = ownerSharedBinaryUploadCandidatesQueryStr
			queryParams.quiesceUntilUpToDate = true
			queryParams.calm = binaryUploadQueryCalmValue

			local ownerSharedBinaryUploadCandidatesStream = self._catalog:indexStore():query( queryParams )

			ownerSharedBinaryUploadCandidatesStream = maybePerformanceWrap( 'query', ownerSharedBinaryUploadCandidatesStream )

			return ownerSharedBinaryUploadCandidatesStream

		end )

	return Observable.combineLatest(
		candidatesStream,
		self._shouldReprocessVersionsStream:startWith( true ) )

		:flatMap( function( uploadCandidates )
			return maybeFixUpOrRemoveCandidates( self._docStore, uploadCandidates,
					self:isAtomicDevelopSettingsEnabled(), self._reloadLinksForDocumentStream ) -- WF-2131
		end )

end

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

--[[

	Instance Function

	\brief Cancel the next currently scheduled binary upload. Used
	in case the priority of a different upload is higher than the
	currently scheduled.

	\return nil - side effect: modify internal variables

]]
function WFCatBinaryUploader:maybeCancelScheduledAndWaitingBinaryUpload()

	if self._scheduledAndWaitingBinaryUpload then

		local docId = self._scheduledAndWaitingBinaryUpload.docId

		local binaryDetails = self._scheduledAndWaitingBinaryUpload.binaryDetails

		local subscription = self._scheduledAndWaitingBinaryUpload.subscription

		logger:infof( "Now cancelling binary upload because it's no longer the top priority %s:%s", docId, binaryDetails.type )

		self:clearScheduledAndWaitingBinaryUpload()

		if subscription then
			subscription:dispose()
		end

		self:clearFromRecentlyUploaded( "higher priority upload rescheduled", docId, binaryDetails, true )

		self:removeFromScheduledUploads( docId, binaryDetails )

	end

end

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

--[[

	Instance Function

	\brief Iterate through the upload candidates, ignoring
	binaries that have been denied or otherwise lack the
	appropriate binaries/links necessary for a succesful
	upload. Once a single upload candidate has been identified,
	this function returns.

	\param candidates (table) - index based table of tables, where
	the subtable corresponds to the columns in the candidates
	query.

	\returns nil - side effect: An observable with is handed to RunUntilDone.

]]
function WFCatBinaryUploader:getHighestPriorityBinaryToUpload( candidates )

	-- Iterate through the upload candidates, ignoring binaries that have been
	-- denied or otherwise lack the appropriate binaries/links necessary
	-- for a succesful upload. Once a single upload candidate has been identified,
	-- this function returns.

	-- IMPORTANT: The candidates array is already in priority order. However,
	-- the order may not be completely consistent when candidates have the
	-- same priority.

	for _, uploadCandidate in ipairs( candidates ) do

		-- uploadCandidate can be false if it's stripped out via
		-- maybeFixUpOrRemoveCandidates.

		if uploadCandidate and not uploadCandidate.skipForNow then

			local docId = uploadCandidate[ 1 ]
			local binaryType = uploadCandidate[ 4 ]

			-- The following block of code detects if we encounter the binary that
			-- we've most recently scheduled for upload. If we *do* encounter it,
			-- it means that it's still the highest prioirity binary that needs to
			-- be uploaded. If we do not encounter it before we find another
			-- binary that needs uploading, it means that it is no longer the top
			-- priority and should therefore be rescheduled. That rescheduling
			-- happens later on in this function.

			if self._scheduledAndWaitingBinaryUpload and
					self._scheduledAndWaitingBinaryUpload.docId == docId and
					self._scheduledAndWaitingBinaryUpload.binaryDetails.type == binaryType then

				-- The currently scheduled-and-waiting binary is still the top priority.

				return false

			end

			-- https://jira.corp.adobe.com/browse/WF-1066
			-- Because the query used to fetch these binaries within getBinariesToUploadStream
			-- does not use `quiesceUntilUpToDate = true`, there's a chance that we could
			-- fetch documents from the table that have already been deleted; such is the
			-- case in WF-1066. By tracking the deleted doc Ids in real time within _deletedDocIdsCache,
			-- we can prevent attempting to upload an already-deleted doc.

			local sha256 = uploadCandidate[ 8 ] or true

			local recentlyUploadedBySha256 = self:getRecentlyUploadedSha256List( docId, binaryType )

			if not self._deletedDocIdsCache[ docId ]
					and not recentlyUploadedBySha256[ sha256 ] then

				local localFilePath = uploadCandidate[ 2 ]
				local binaryUrl = uploadCandidate[ 3 ]
				local contentType = uploadCandidate[ 5 ]
				local isDenied = intToBool( uploadCandidate[ 7 ] )
				local sourceLinkForCopy = uploadCandidate[ 11 ]
				local localFileVolumeGuid = uploadCandidate[ 12 ]
				local localFileRelativePath = uploadCandidate[ 13 ]
				local uploadPriority = uploadCandidate[ 14 ]
				local auxId = uploadCandidate[ 18 ]

				local binaryDetails = newBinaryDetail( binaryType, sha256 )

				if isDenied then

					-- Check if the asset is actually denied.
					isDenied = self._failedUploadDeniedListHelper:maybeResetDeniedListStatus( docId )

				end

				-- ensure localFilePath is a string and not bad data
				localFilePath = type( localFilePath ) == 'string' and localFilePath or nil

				-- Note that it's possible that localFilePath, localFilePath and localFileRelativePath are nil.
				-- This can occur due to copy-related workflows (trash, SWY contributions, etc).

				local binaryFullPath, assetIsOffline = self:getBinaryFullPathAndAssetOfflineStatus( localFileVolumeGuid, localFileRelativePath, localFilePath, binaryDetails )

				if binaryFullPath then
					localFilePath = binaryFullPath
				end

				uploadCandidate[ 2 ] = localFilePath

				local binaryExistsLocally = localFilePath and WFFileUtils.exists( localFilePath )
					-- Though we could check that the original exists in the indexer, it's
					-- more efficient to check for it here since it won't run all items,
					-- but only the items before we've reached maxOriginalUploadsInFlight.

				-- Check that the priority of this binary upload is higher than the currently
				-- scheduled binary upload. Only then will we cancel the existing binary
				-- upload.

				local impossiblyLowPriority = -1000

				local scheduledUploadPriority = self._scheduledAndWaitingBinaryUpload
						and self._scheduledAndWaitingBinaryUpload.priority or impossiblyLowPriority

				if not isDenied
						and ( uploadPriority > scheduledUploadPriority )
						and ( binaryExistsLocally or -- Standard Uploads
						sourceLinkForCopy or -- Copies + InAppContribute assets
						( WFPlatformInfo.isDesktop and localFilePath == nil and not assetIsOffline ) or  -- SPAM [matthewj 2019-09-25]: This case covers edge cases like testUploadCopy_sourceDeleted_noMasterLocal
																											-- but these if cases should be restructured significantly to each handle fewer cases and this one should
																											-- be moved out of this if block. This is related to story WF-685
						( WFPlatformInfo.isMobile( )  and localFilePath == nil  and  sourceLinkForCopy) ) -- SPAM [matthewj 2019-09-27]: this allows copies/IAC through but prevents a special Android case (WF-501) from
																											-- falling in and instead those should end up getting caught in the not ( binaryExistsLocally or sourceLinkForCopy )
																											-- case. This is related to story WF-686/WF-689 and to a lesser extent WF-685
						then

					self:maybeCancelScheduledAndWaitingBinaryUpload()

					return uploadCandidate

				elseif not ( binaryExistsLocally or sourceLinkForCopy ) then

					-- Trying to upload a file but don't have one to upload and it's not a copy.

					if isXmpSidecar( binaryDetails.type ) and not sourceLinkForCopy then

						-- Xmp sidecar files are always stored on the main volume. If the
						-- xmp sidecar file doesn't exist, then we need to delete the
						-- file annotation.

						self:addToRecentlyUploaded( docId, binaryDetails, sha256 or true )

						self:removeSidecarFileAnnotation( docId )
							:finally( function()
								self:delayedClearFromRecentlyUploaded( "removing sidecar annotation", docId, binaryDetails )
							end )
							:call( runUntilDone )

					elseif isImageOrVideoOriginal( binaryDetails.type ) or isWatermarkOriginal( binaryDetails.type )
							or ( WFPlatformInfo.isClassic() and isProxy( binaryType ) ) then

						if WFPlatformInfo.isMobile() then

							self:handleMasterNotExistsLocally( docId, binaryUrl, binaryDetails, contentType )
								:call( runUntilDone )

						elseif localFilePath then -- localFilePath can be nil doing a copy for desktop.

							logger:warnf( "Uh oh. Trying to upload %s for %s but it doesn't exist at path %s.", binaryDetails.type, docId, localFilePath )

							if ( isOriginalBinary( binaryDetails.type ) or ( WFPlatformInfo.isClassic() and isProxy( binaryType ) ) )
									and not assetIsOffline then

								local reason = isProxy( binaryType ) and kUploadFailedMissingProxy or kUploadFailedMissingOriginal

								self:applyInvalidOriginalOverlay( docId, localFilePath, binaryUrl, reason )

									:finally( function()
										self:delayedClearFromRecentlyUploaded( "binary missing, marked invalid if volume not offline", docId, binaryDetails )
									end )

									:call( runUntilDone )

							end

						end

					elseif isXmpDevelop( binaryDetails.type ) then

						-- Set flag that we are processing this upload entry to prevent multiple applyOverlay calls.
						self:addToRecentlyUploaded( docId, binaryDetails, sha256 or true )

						self._docStore:docValue( docId )

							:flatMap( function( docValue )

								local deleteLocalOnlyXmpDevelop = false
								local sha256InLocalOnly = WFCatAssetUtils.getLocalExternalDevelopSettingsContentId( docValue )

								if sha256InLocalOnly ~= sha256 then

									return Observable.just( deleteLocalOnlyXmpDevelop, sha256InLocalOnly )

								else

									-- Use error to make this visible to analytics as somehow we lost the users edits.
									logger:errorf( "xmp_develop file missing removed _localOnly files entry %s", docId )

									return self._docStore:applyOverlay( {
											_id = docId,
											_localOnly = {
												files = {
													xmp_develop = { _delete = true },
												}
											}
										} )
										:map( function()
											deleteLocalOnlyXmpDevelop = true
											return deleteLocalOnlyXmpDevelop
										end )
										:finally( function()
											-- Clear the flag we use to indicate we are processing an upload.
											self:delayedClearFromRecentlyUploaded( "removed xmp_develop annotation", docId, binaryDetails )
										end )

								end

							end )

							:map( function( deleteLocalOnlyXmpDevelop, maybeSha256InLocalOnly )

								local event = {
									['event.type'] = 'error',
									['event.subtype'] = binaryDetails.type .. "-uploaded",
									['event.appWorkflow'] = 'SYNC',
									['content.id'] = docId,
									['content.type'] = contentType,
								}

								if deleteLocalOnlyXmpDevelop then

									local desc = string.format( 'Removed xmp_develop annotation for docId %s', docId )
									event['event.error_desc'] = desc
									event['event.error_type'] = 'xmp_develop_annotation_remove'

								else

									local desc = string.format( 'docValue sha256 - %s does not match uploading sha256 - %s', maybeSha256InLocalOnly, sha256 )
									event['event.error_desc'] = desc
									event['event.error_type'] = 'xmp_develop_does_not_match'

								end

								WFAnalytics.trackEvent( event )

							end )

							:call( runUntilDone )

					elseif isAux( binaryDetails.type ) and auxId then

						-- Set flag that we are processing this upload entry to prevent multiple applyOverlay calls.

						self:addToRecentlyUploaded( docId, binaryDetails, sha256 or true )

						-- Delete the local aux annotation when we detect the file doesn't exist.

						self._docStore:docValue( docId )

							:map( function( docValue )

								return WFCatAssetAux.getDigestForAuxId( docValue, auxId )

							end )

							:flatMap( function( auxDigest )

								local overlay = {
									_id = docId,
									_localOnly = {
										files = {
											aux = {
												[ auxDigest ] = {
													relativePath = { _delete = true }
												}
											}
										}
									}
								}

								return self._docStore:applyOverlay( overlay )

							end )

							:map( function()

								local desc = string.format( 'Removed aux annotation for auxId', auxId )

								local event = {
									['event.type'] = 'error',
									['event.subtype'] = binaryDetails.type .. "-uploaded",
									['event.appWorkflow'] = 'SYNC',
									['content.id'] = docId,
									['content.type'] = contentType,
									['event.error_desc'] = desc,
									['event.error_type'] = 'aux_annotation_remove'
								}

								WFAnalytics.trackEvent( event )

							end )

							:finally( function()
								-- Clear the flag we use to indicate we are processing an upload.
								self:delayedClearFromRecentlyUploaded( "removed aux annotation", docId, binaryDetails )
							end )

							:call( runUntilDone )

					end

				end

			end

		end

	end

end

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

--[[

	Instance Function

	\brief Construct a stream of upload candidates where the highest priority candidate is yielded first.

	\return Observable<lots-of-stuff> - See implementation for details.

]]
function WFCatBinaryUploader:getHighestPriorityBinaryToUploadStream()

	-- There's potentially for a race condition where the query might be refreshed
	-- before we've determined on this end that the upload has finished. When this happens,
	-- the below rx chain will do no work since we still think we have the maximum
	-- number of uploads in flight. This can potentialy cause sync to freeze. In
	-- order to remedy this, we use combineLatest to cause the rx chain to fire on
	-- query refresh, as well as when an upload finishes.

	local previousBackgroundUploadStatus

	return Observable.combineLatest(
			self:getOwnerSharedBinaryUploadCandidatesStream(),
			self:getStandardBinaryUploadCandidatesStream(),
			self._allowBackgroundUpload:distinctUntilChanged(),
			self._uploadFinishedStream,
			self._uploadStatusChangeStream,
			self._volumeStatusHasChangedStream )

		:flatMapLatest( function( ownerSharedUploadCandidates, standardUploadCandidates, allowBackgroundUpload )

			return self._continuousUploadTask:throwIfNotRunning( ownerSharedUploadCandidates, standardUploadCandidates, allowBackgroundUpload )

		end )

		:flatMap( function( ownerSharedUploadCandidates, standardUploadCandidates, allowBackgroundUpload )

			if previousBackgroundUploadStatus and not allowBackgroundUpload then
				self:disposeAllActiveUploads()
			elseif not previousBackgroundUploadStatus and allowBackgroundUpload then
				self:maybeCancelScheduledAndWaitingBinaryUpload()
			end

			previousBackgroundUploadStatus = allowBackgroundUpload

			logger:debugf( "ownerSharedUploadCandidates raw query result = %s", #ownerSharedUploadCandidates )
			logger:debugf( "standardUploadCandidates raw query result = %s", #standardUploadCandidates )

			if not hasCandidateInCandidateList( ownerSharedUploadCandidates )
					and not hasCandidateInCandidateList( standardUploadCandidates ) then

				self:sendFinalDetailedProgressUpdate()
				self._continuousUploadTask:taskDidWork( false )
					-- Since we don't typically send onCompleted events we need to
					-- indicate when we don't do work.

			end

			-- Binaries that belong to assets that exist within a user-owned shared
			-- album get priority over other assets; note that we do not prioritize
			-- uploads to a shared album when the album is *not* owned by the user.

			local uploadCandidate = self:getHighestPriorityBinaryToUpload( ownerSharedUploadCandidates )

			if uploadCandidate == nil then -- Intentional nil check, see below.

				-- Specifically making sure `uploadCandidate` is not false above. It's false
				-- when getHighestPriorityBinaryToUpload detects the best candidate for upload
				-- is already the most recently scheduled-but-waiting upload. This is why we
				-- explicitly test against nil.

				uploadCandidate = self:getHighestPriorityBinaryToUpload( standardUploadCandidates )

			end

			return Observable.just( uploadCandidate )

		end )

		:filter( function( uploadCandidate )

			local hasUploadCandidate = not not uploadCandidate
			local shouldScheduleUpload = not self._scheduledAndWaitingBinaryUpload

			return hasUploadCandidate and shouldScheduleUpload

		end )

end

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

--[[

	\brief Test helper - exposes a way to construct a binaryDetail table.

]]
WFCatBinaryUploader.class.TEST_newBinaryDetail = function(self, ...)

	return newBinaryDetail(...)

end

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


--[[

	\brief Returns a text representation of the errors headed for the diagnostic log

]]
function WFCatBinaryUploader:DEBUG_getDiagnosticLogErrors()
	return logger.DEBUG_makeDiagnosticLogErrorsHtmlText()
end

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

--[[

	\brief resets the diagnostic log

]]
function WFCatBinaryUploader:DEBUG_resetDiagnosticLogErrors()
	return logger.DEBUG_resetDiagnosticLogErrors()
end

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

--[[

	\brief yields a handle on the debug inflight upload progress update stream for testing

]]
function WFCatBinaryUploader:DEBUG_getInFlightUploadProgressUpdateStream()
	return self._debugInFlightUploadProgressUpdateStream
end

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

--[[

	\brief Creates a summary string to be inserted into the diagnostic log

	\param uploadsByType (table) - the table of uploads by binary type

	\return string

]]
local function createDiagnosticSummaryStr( uploadsByType )

	local numberOfCatalogBinariesUploaded
			= derefOpt( uploadsByType, 'catalog', 'total' ) or 0

	local summary = string.format( "Catalog: Uploaded %d binaries to Oz.", numberOfCatalogBinariesUploaded )

	for catalogOrSpaceId, uploads in pairs( uploadsByType or {} ) do

		if catalogOrSpaceId ~= 'catalog' then

			local numberOfSpaceBinariesUploaded
					= derefOpt( uploads, 'total' ) or 0

			summary = string.format( "%s\nSpace %s: Uploaded %d binaries to Oz.",
					summary, catalogOrSpaceId, numberOfSpaceBinariesUploaded )

		end

	end

	return summary

end

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

--[[!

	\brief This API that can be used gather real-time diagnostic data from the
		binary uploader.

	\param optCalm (optional number) The maximum frequency at which the stream will
		fire with updates. Defaults to 1 second.

	\return Observable< table > An Observable whose value tracks a table of
		sync diagnostic data. An example data event may look like this:

	Please note that the uploadsByType section only shows binaries that have
		been uploaded _after_ calling the syncDiagnostics API.

	{
		documentsNeedingBinaryUpload = 1,
		inFlightUploads = {
			["37ef13baa82a44f0ae90a9533b112a3e"] = {
				master = true,
			},
		},
		isCurrentlySyncing = true,
		numberOfInFlightUploads = 1,
		numberOfScheduledBinaryUploads = 1,
		recentlyUploaded = {
			["37ef13baa82a44f0ae90a9533b112a3e"] = {
				master = "f8bc7af5e0266638030626fcb93d8d2f95c571ad8626ba5d0a12dff7d76c5ee2",
			},
		},
		summaryStr = "Catalog: Uploaded 2 binaries to Oz. Space 1cbf5a97cf9149be8d6d9ad5c8b075e5: Uploaded 3 binaries to Oz.",
		uploadsByType = {
			["1cbf5a97cf9149be8d6d9ad5c8b075e5"] = {
				["Oz Copies"] = 3,
				master = {
					["Oz Copies"] = 1,
					total = 1,
				},
				total = 3,
				xmp_sidecar = {
					["Oz Copies"] = 2,
					total = 2,
				},
			},
			catalog = {
				master = {
					total = 2,
				},
				total = 2,
			},
		},
	}

--]]

function WFCatBinaryUploader:syncDiagnostics( optCalm )

	check( 'WFCatBinaryUploader', '?number' )

	if not self._uploadsByTypeStream then
		self._uploadsByTypeStream = Rx.BehaviorSubject( {} )
	end

	local documentsNeedingBinaryUploadStream = self._catalog:indexStore():query(
			{
				sql = [[SELECT COUNT( * ) FROM needsBinaryUpload]],
				calm = optCalm or 1,
				singleColumnQuery = true,
			} )

			:map( function( rows )

				return derefOpt( rows, 1 ) or 0

			end )

	return Observable.combineLatest(
			self._uploadsByTypeStream:startWith( {} ),
			self:isDoingWork(),
			documentsNeedingBinaryUploadStream,
			self._uploadFinishedStream:startWith( {} ),
			self._uploadStatusChangeStream:startWith( true ),
			self._uploadScheduledStream:startWith( true ) )

		:calm( optCalm or 1, WFRxScheduler.sharedScheduler() )

		:map( function( uploadsByType, isCurrentlySyncing, documentsNeedingBinaryUpload )

			local numberOfInFlightUploads = #self._scheduledUploadSubscriptions

			local inFlightUploads = {}

			for docId, uploadSubscriptions in pairs( self._scheduledUploadSubscriptions ) do

				inFlightUploads[ docId ] = {}

				for binaryDetails, _ in pairs( uploadSubscriptions ) do

					inFlightUploads[ docId ][ binaryDetails.type ] = true

				end

			end

			local scheduledAndWaitingBinaryUpload

			if self._scheduledAndWaitingBinaryUpload then

				scheduledAndWaitingBinaryUpload = {
					docId = self._scheduledAndWaitingBinaryUpload.docId,
					binaryType = self._scheduledAndWaitingBinaryUpload.binaryDetails.type,
					priority = self._scheduledAndWaitingBinaryUpload.priority
				}

			end

			local diagnostics = {
				summaryStr = createDiagnosticSummaryStr( uploadsByType ),
				numberOfScheduledBinaryUploads = self._numberOfScheduledBinaryUploads,
				numberOfInFlightUploads = numberOfInFlightUploads,
				isCurrentlySyncing = not not isCurrentlySyncing,
				uploadsByType = uploadsByType,
				inFlightUploads = inFlightUploads,
				scheduledUpload = scheduledAndWaitingBinaryUpload,
				recentlyUploaded = self._recentlyUploaded,
				documentsNeedingBinaryUpload = documentsNeedingBinaryUpload,
			}

			logger:infof( "Binary Uploader syncDiagnostics update: %s", diagnostics )

			return diagnostics

		end )

end

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

--[[

	Instance Function

	\param needsBinaryUploadInfo (table) - table of binaries which need to be uploaded

	\return table - used to construct the diagnostic log

]]
function WFCatBinaryUploader:makeBinaryUploaderHTML( needsBinaryUploadCount, needsBinaryUploadInfo )

	local scheduledUploads = {}

	for docId, uploadSubscriptionsForDocId in pairs( self._scheduledUploadSubscriptions ) do
		for binaryDetails, _ in pairs( uploadSubscriptionsForDocId ) do
			table.insert( scheduledUploads, { id = docId, binaryType = binaryDetails.type } )
		end
	end

	local finalStats = {}

	for bucket, stats in pairs( self._uploadSpeedStats.buckets ) do
		finalStats[ bucket ] = {
			count = stats.count,
			timings = { stats.timings:visualize 'boxAndWhisker' },
			size = { stats.sizes:visualize 'boxAndWhisker' },
			speed = { stats.speed:visualize 'boxAndWhisker' },
		}
	end

	local scheduledAndWaitingBinaryUpload = {}

	if self._scheduledAndWaitingBinaryUpload then

		scheduledAndWaitingBinaryUpload = {
			docId = self._scheduledAndWaitingBinaryUpload.docId,
			binaryType = self._scheduledAndWaitingBinaryUpload.binaryDetails.type,
			priority = self._scheduledAndWaitingBinaryUpload.priority
		}

	end

	return {
			scheduledAndWaitingBinaryUpload = scheduledAndWaitingBinaryUpload,
			scheduledUploads = scheduledUploads,
			numberOfScheduledBinaryUploads = self._numberOfScheduledBinaryUploads,
			needsBinaryUploadCount = needsBinaryUploadCount,
			needsBinaryUploadInfo = needsBinaryUploadInfo,
			recentErrors = self._uploadErrorLog,
			uploadHttpRequestCost = uploadHttpRequestCost,
			uploadSpeedStats = finalStats,
			totalBinarySize = self._uploadSpeedStats.totalBinarySize,
			chunkFileAtSize = self._chunkedUploader:chunkFileSize(),
			chunkSize = self._chunkedUploader:chunkSize()
		}

end

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

--[[
	Instance Function

	\brief sends information from the uploader to the diagnostic log for debugging.

]]
function WFCatBinaryUploader:sendDiagnosticLogData()

	self._catalog:indexStore():queryOnce {
		sql = [[SELECT docId, binaryType, captureDate, fileSize, docType, isReadyToUpload FROM needsBinaryUpload
				ORDER BY fileMaybeOnOfflineVolume ASC, isDenied ASC,
						priorityForUpload DESC, importDate ASC, captureDate ASC,
						docId
				LIMIT 500]]
	}
	:flatMap( function( rows )

		if not rows or not next( rows ) then
			return Rx.Observable.just( {} )
		end

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

		return self._docStore:bulkDocValue( docIds )
			:map( function( docValues )

				local result = {}

				for i, row in ipairs( rows ) do

					local typeSubtype = row[ 5 ] or 'unknown'
					local subtype = derefOpt( docValues, i, 'subtype' )
					if subtype then
						typeSubtype = typeSubtype .. '/' .. subtype
					end

					result[ i ] = {
						_id  = row[ 1 ],
						typeSubtype = typeSubtype,
						binaryType = row[ 2 ],
						captureDate = row[ 3 ],
						fileSize = row[ 4 ],
						isReadyToUpload = row[ 6 ],
						fileName = derefOpt( docValues, i, 'importSource', 'fileName' )
					}

				end

				return result

			end )
		end )
		:combineLatest(
			function( needsBinaryUploadInfo, countNeedsBinaryUploadQueryResult )
				return derefOpt( countNeedsBinaryUploadQueryResult, 1 ) or 0, needsBinaryUploadInfo
			end,
			self._catalog:indexStore():queryOnce( {
				sql = [[SELECT COUNT( * ) FROM needsBinaryUpload]],
				singleColumnQuery = true,
			} )
		)
		:map( function( needsBinaryUploadCount, needsBinaryUploadInfo )

			WFDiagnosticLog.addSectionToLog
				{
					name = "Binary Uploader",
					id = 'WFCatBinaryUploader',
					weight = 510,
					parentSection = WFDiagnosticLog.Sections.SYNC,
					body = self:makeBinaryUploaderHTML( needsBinaryUploadCount, needsBinaryUploadInfo )
				}

		end )
		:call( WFRxRunUntilDone )


	return {}

end

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

-- Used by desktop to get ids of docs that need upload for inclusion in sync
-- issues report.

function WFCatBinaryUploader:getNeedsUploadDocIds()

	return self._catalog:indexStore():queryOnce {
		sql = "SELECT docId FROM needsBinaryUpload",
		singleColumnQuery = true,
	}

end

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

--[[

	Instance Function

	\brief Yields how many uploads have gone to a specific bucket - debug only

]]
function WFCatBinaryUploader:DEBUG_uploadCount( bucket )

	local bucketStats = self._uploadSpeedStats.buckets[ bucket ]

	return derefOpt( bucketStats, 'count' ) or 0

end

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

--[[

	Instance Function

	\brief Resets the first measurement. It's necessary to reset
	these values in the event new binaries need to be uploaded
	*after* we take our first measurement.

	return nil - side effect

]]
local function resetFirstMeasurement( firstMeasurement )

	-- Resets the first measurement. It's necessary to reset these values
	-- in the event new binaries need to be uploaded *after* we take
	-- our first measurement.

	if firstMeasurement then
		firstMeasurement.timestamp = nil
		firstMeasurement.size = nil
	end

end

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

--[[

	\brief Calculate the current upload speed based on previous completed uploads

	\param rows				(table) a table of sql rows - see getSizeOfRemainingUploadsStream for details
	\param firstMeasurement			(table)
	\param previousRemainingBinaryUploads	(table) - map of binaries which need to be uploaded
	\param inFlightUploadProgress		(table) - in flight upload information

	\return table, table

]]
local function calculateUploadSpeed( rows, firstMeasurement,
										previousRemainingBinaryUploads,
										inFlightUploadProgress )

	local currentSizeMeasurement = 0
	local ozUploadSpeed = 0
	local remainingUploads = {}

	local currentTime, hasReset

	if #rows == 0 then
		resetFirstMeasurement( firstMeasurement )
	end

	for _, row in ipairs( rows ) do

		local docId = row[ 1 ]
		local binaryType = row[ 2 ]
		local sizeOfRemainingUpload = row[ 3 ]
		local sha256 = row[ 4 ]

		local key = string.format( "%s::%s::%s", docId, binaryType, sha256 )

		if not previousRemainingBinaryUploads[ key ] and not hasReset then

			-- Previously unknown upload detected.

			resetFirstMeasurement( firstMeasurement )
			hasReset = true

		end

		remainingUploads[ key ] = true

		-- SPAM [jandrew 2021-10-07]: Android has reported some cases where the fileSize
		-- is not available in the index store table. It's currently not understood how
		-- this is possible, but if we detect it, we assume some imperfect file size
		-- defaults. Time estimation won't be accurate for these files, but at least
		-- it won't blow up. WF-1767

		if not sizeOfRemainingUpload then

			if isXmpSidecar( binaryType ) then
				sizeOfRemainingUpload = 1048576 -- 1MB
			else
				sizeOfRemainingUpload = 26214400 -- 25MB (imperfect!)
			end

		end

		-- Check to see if there's a progress event for the binary. If there is, that means
		-- that the upload has at least been scheduled.

		local progress = derefOpt( inFlightUploadProgress, docId, binaryType, sha256 )

		if progress then

			sizeOfRemainingUpload = math.max( 0, sizeOfRemainingUpload - progress.processedCount )
				-- Adjust the size to upload by the number of bytes that have already
				-- been reported as sent.

			local processProgressEvent = function( progressEvent )

				-- We ignore progress events for completed uploads, and for uploads that
				-- have been scheduled but not started.

				local uploadHasCompleted = derefOpt( progressEvent, 'extraInfo', 'endTime' )
				local uploadStartTime = derefOpt( progressEvent, 'extraInfo', 'startTime' )
				local timeSinceLastUpdate = derefOpt( progressEvent, 'extraInfo', 'timestamp' )

				if not uploadHasCompleted and uploadStartTime then

					if not firstMeasurement.timestamp or ( uploadStartTime < firstMeasurement.timestamp ) then

						-- Record our first timestamp measurement

						firstMeasurement.timestamp = uploadStartTime

					end

					if not currentTime or ( timeSinceLastUpdate and currentTime < timeSinceLastUpdate ) then

						-- Record our other timestamp measurement

						currentTime = timeSinceLastUpdate

					end

				end

			end

			-- In the case of direct/chunked upload, the progress event contains nested
			-- progress events. The parent event represents the progress of the entire upload,
			-- while the child events represent the progress of the individual parts.

			local progressOfChunks = derefOpt( progress, 'extraInfo', 'progressOfChunks' )

			if progressOfChunks then

				-- Direct/chunked upload, so process the progress of the the parts.

				for _, chunkProgress in pairs( progressOfChunks ) do

					processProgressEvent( chunkProgress )

				end

			else

				-- Non direct/chunked upload.

				processProgressEvent( progress )

			end

		end

		currentSizeMeasurement = currentSizeMeasurement + sizeOfRemainingUpload

	end

	previousRemainingBinaryUploads = remainingUploads

	if not firstMeasurement.size then

		-- Record our first size measurement.

		firstMeasurement.size = currentSizeMeasurement

	elseif firstMeasurement.timestamp and currentTime then

		-- Use the size/timestamp measurements to calculate the upload speed.

		local bytesElapsed = firstMeasurement.size - currentSizeMeasurement
		local timeElapsed = currentTime - firstMeasurement.timestamp

		ozUploadSpeed = bytesElapsed / timeElapsed

	end

	logger:debugf( "Size of Remaining Uploads %.2f mb, Oz Upload Speed: %.2f mb, In-Flight Count: %d",
			bytesToMb( currentSizeMeasurement ), bytesToMb( ozUploadSpeed ), #rows )

	return { currentSizeMeasurement, ozUploadSpeed, #rows }, previousRemainingBinaryUploads

end

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

--[[

	\brief Exposed for testing calculateUploadSpeed
]]
WFCatBinaryUploader.class._TEST_calculateUploadSpeed = function( self, ... )
	return calculateUploadSpeed( ... )

end

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

--[[

	Instance Function

	\brief In order to calculate the upload speed, we use the
	total number of remaining bytes to upload from two points in
	time. Previous attempts at calculating upload speed by
	measuring the throughput of individual requests proved
	unreliable and innacurate.

	\return Observable<number>

]]
function WFCatBinaryUploader:getSizeOfRemainingUploadsStream()

	-- In order to calculate the upload speed, we use the total number of remaining
	-- bytes to upload from two points in time. Previous attempts at calculating
	-- upload speed by measuring the throughput of individual requests proved
	-- unreliable and innacurate.

	-- The first measurement is recorded at the time this function is called.

	local firstMeasurement = {
		-- timestamp -- The timestamp associated with the first bytes-remaining-to-upload measurement.
		-- size -- The first bytes-remaining-to-upload measurement.
	}

	local previousRemainingBinaryUploads = {}
		-- A map of binaries that need to be uploaded. Updated every time the query fires.
		-- key: docId::binaryType::sha256
		-- value: true

	-- Query to calculate what binaries need to be uploaded (not copied), and their size.

	local queryCalm = WFPlatformInfo.isUnitTest() and 0.1 or 3

	local query = self._catalog:indexStore():query {
			sql = [[SELECT docId, binaryType, fileSize, sha256
						FROM needsBinaryUpload
						WHERE docType == 'asset' and sourceLinkForCopy IS NULL]],
			calm = queryCalm,
			quiesceUntilUpToDate = true,
		}

	-- We reprocess the query results periodically, since progress events are reported
	-- frequently from HTTP code. Note that we don't need to process every HTTP progress
	-- event in real time; processing them every second is fine since the time-remaining
	-- estimate is reported only every few seconds to the client.

	local timerInterval = WFPlatformInfo.isUnitTest() and 0.1 or 1

	local periodicTimer = Observable.timer( timerInterval, timerInterval, WFRxScheduler.sharedScheduler() )

	return Observable.combineLatest( query, periodicTimer )

		:skip( 5 )
			-- Skip the first few progress measurements. In testing, I found that this
			-- greatly improves the accuracy, especially in the case of resumed uploads.

		:map( function( rows )

			local result

			result, previousRemainingBinaryUploads =
					calculateUploadSpeed( rows, firstMeasurement, previousRemainingBinaryUploads, self._inFlightUploadProgress )

			return result

		end )

end

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

--[[

	Instance Function

	\brief calculate the byte size of copy binary actions which remain to be executed

]]
function WFCatBinaryUploader:getSizeOfRemainingCopiesStream()

	local queryCalm = WFPlatformInfo.isUnitTest() and 0.1 or 3
	local timerInterval = WFPlatformInfo.isUnitTest() and 0.1 or 1

	return Observable.combineLatest( self._catalog:indexStore():query {
				sql = [[SELECT docId, binaryType, fileSize, sha256
							FROM needsBinaryUpload
							WHERE docType == 'asset' and sourceLinkForCopy IS NOT NULL ]],
				calm = queryCalm,
			},
		Observable.timer( timerInterval, timerInterval, WFRxScheduler.sharedScheduler() ) )

		:map( function( rows )

			local numberOfInFlightCopies = 0
			local sizeOfRemainingCopies = 0
			local secondsBuffer = 0

			-- SPAM [jandrew 2021-06-25]: The following number is an estimate of the Oz
			-- copy speed based on the Oz splunk logs.

			local ozAverageCopySpeed = 26214400 -- 25MBps.
			local minOzProcessingTime = 5 -- seconds

			for _, row in ipairs( rows ) do

				local docId = row[ 1 ]
				local binaryType = row[ 2 ]
				local fileSize = row[ 3 ]
				local sha256 = row[ 4 ]

				if not fileSize then

					if isXmpSidecar( binaryType ) then
						fileSize = 1048576 -- 1MB
					else
						fileSize = 26214400 -- 25MB default. Imperfect, but copies are fast so it's okay.
					end

				end

				local progress = derefOpt( self._inFlightUploadProgress, docId, binaryType, sha256 )

				local startTime = derefOpt( progress, 'extraInfo', 'startTime' )

				if startTime then

					numberOfInFlightCopies = numberOfInFlightCopies + 1

					-- Oz is doing the heavy lifting for copies, so we have to estimate the
					-- speed of the copy request on S3 without using the progress of the
					-- PUT request, and without knowing the actual throughput of the copy
					-- on Oz. The speed here is also dependent on Oz load, so its imperfect.

					local estimatedBytesCopied = 0

					local endTime = derefOpt( progress, 'extraInfo', 'endTime' )

					if endTime then

						local timeSpentCopying = KSDate.timeNow() - endTime

						estimatedBytesCopied = timeSpentCopying * ozAverageCopySpeed

					end

					-- Never estimate that we've copied more than the total file size.

					estimatedBytesCopied = math.min( fileSize * 0.95, estimatedBytesCopied )

					local sizeOfRemainingCopy = fileSize - estimatedBytesCopied

					sizeOfRemainingCopies = sizeOfRemainingCopies + sizeOfRemainingCopy

					local timeElapsed = KSDate.timeNow() - startTime

					secondsBuffer = secondsBuffer + math.max( 0, minOzProcessingTime - timeElapsed )

				else

					-- For copies, the actual PUT request doesn't have much of an effect
					-- on the timings. The request size is only ~150 bytes. If a copy PUT
					-- request has not yet complete, we add 0.5 seconds to
					-- the time estimation.

					secondsBuffer = secondsBuffer + 0.5

					secondsBuffer = secondsBuffer + minOzProcessingTime

					sizeOfRemainingCopies = sizeOfRemainingCopies + fileSize

				end

			end

			return { sizeOfRemainingCopies, numberOfInFlightCopies, ozAverageCopySpeed, secondsBuffer, #rows }

		end )

end

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

--[[

	\brief Get the estimated time remaining for all uploads

	\param sizeOfRemainingUploadsStream	(observable) - Stream of remaining upload sizes
	\param sizeOfRemainingCopiesStream	(observable) - Stream of remaining copy sizes

	\return int, int - estimated time remaining in seconds, calculated oz uploaded speed from the device perspective

]]
local function _getEstimatedTimeRemainingImp(
		sizeOfRemainingUploadsStream, sizeOfRemainingCopiesStream )

	return Observable.combineLatest(
			sizeOfRemainingUploadsStream,
			sizeOfRemainingCopiesStream )

		:map( function( uploadInfo, copyInfo )

			local sizeOfRemainingUploads, ozUploadSpeed, remainingUploadCount = unpack( uploadInfo )

			local sizeOfRemainingCopies, numberOfInFlightCopies, ozAverageCopySpeed,
					secondsBuffer, remainingCopyCount = unpack( copyInfo )

			if remainingUploadCount == 0 and remainingCopyCount == 0 then
				return -1 -- Nothing to upload
			end

			-- Uploads

			local timeRemainingForUploads = 0 -- seconds

			if sizeOfRemainingUploads > 0 then

				if ozUploadSpeed > 0 then
					timeRemainingForUploads = sizeOfRemainingUploads / ozUploadSpeed
				else
					timeRemainingForUploads = math.huge
				end

			end

			-- Copies

			local timeRemainingForCopies = 0 -- seconds

			if ozAverageCopySpeed and ozAverageCopySpeed > 0 and sizeOfRemainingCopies > 0 then

				timeRemainingForCopies = ( sizeOfRemainingCopies / ozAverageCopySpeed ) + secondsBuffer

				if numberOfInFlightCopies > 0 then

					timeRemainingForCopies = timeRemainingForCopies / numberOfInFlightCopies

				end

			end

			-- Total

			local kOzUploadConfirmationDelay = 5
			local estimatedTimeRemaining = timeRemainingForUploads + timeRemainingForCopies + kOzUploadConfirmationDelay

			logger:debugf( "getEstimatedTimeRemaining:: Current Time : Size-Remaining : Upload Speed : Time-Remaining: %f, %.2f mb, %.2f mb, %.2f seconds",
					KSDate.timeNow(), bytesToMb( sizeOfRemainingUploads ), bytesToMb( ozUploadSpeed ), timeRemainingForUploads )

			return estimatedTimeRemaining, ozUploadSpeed

		end )

end

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

--[[

	\brief Exposed for testing

]]
WFCatBinaryUploader.class._TEST_getEstimatedTimeRemainingImp = function( _, ... )
	return _getEstimatedTimeRemainingImp( ... )
end

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

WFCatBinaryUploader.class._DEBUG_sanitizeDirectUploadUrlForAnalytics = function( _, url )
	return WFAnalytics.sanitizeForAnalytics( url )
end

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

WFCatBinaryUploader.class._DEBUG_maybeEmitAuxUploadRecord = function( _, ... )
	return maybeEmitAuxUploadRecord( ... )
end

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

WFCatBinaryUploader.class._DEBUG_maybeEmitXmpDevelopUploadRecord = function( _, ... )
	return maybeEmitXmpDevelopUploadRecord( ... )
end

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

WFCatBinaryUploader.class._TEST_maybeFixUpOrRemoveCandidates = function( _, ... )
	return maybeFixUpOrRemoveCandidates( ... )
end

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

function WFCatBinaryUploader:TEST_getCapabilitiesCache()
	-- Only used by unit tests.
	return self._capabilitiesCache
end

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

function WFCatBinaryUploader:_TEST_getReloadLinksForDocumentStream()
	-- Only used by unit tests.
	return self._reloadLinksForDocumentStream
end

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

--[[!

	\brief This API provides an estimation -- in seconds -- for the time remaining
		to upload any outstanding binaries. -1 is returned in the event there is
		nothing to upload.

	\return Observable< number > An Observable that fires with numbers representing
		the time remaining in seconds.

--]]
function WFCatBinaryUploader:getEstimatedTimeRemaining()

	-- Time remaining takes into account:
	-- Progress of in-flight uploads/copies
	-- Size of remaining binary uploads (Client -> Oz)
	-- Size of remaining binary copies (Oz -> Oz)

	local kCalmValue = WFPlatformInfo.isUnitTest() and 0.1 or 3

	return _getEstimatedTimeRemainingImp(
			self:getSizeOfRemainingUploadsStream(),
			self:getSizeOfRemainingCopiesStream() )

		:filter( function( estimatedTimeRemaining, averageUploadSpeed )

			return estimatedTimeRemaining ~= 0 and estimatedTimeRemaining ~= math.huge

		end )

		:calm( kCalmValue, WFRxScheduler.sharedScheduler() )

		:map( function( estimatedTimeRemaining, averageUploadSpeed )

			-- WF 2220 send bandwidth statistics (once at app start and then at most
			-- once per hour of active uploads)
			if averageUploadSpeed and averageUploadSpeed > 0 then

				self._wf2220Counter = self._wf2220Counter + 1

				if self._wf2220Counter == 1 then

					logger:infof( 'WF-2220 upload speed (bytes/second): %.2f', averageUploadSpeed )

					WFAnalytics.trackEvent {
						['event.type'] = 'info',
						['event.subtype'] = 'upload-statistics',
						['event.appWorkflow'] = 'SYNC',
						['averageBps'] = averageUploadSpeed, -- bytes/s
					}

				elseif self._wf2220Counter == 3600 / kCalmValue then
					-- reset after an hour of active uploads so it will be sent again
					self._wf2220Counter = 0
				end

			end

			return estimatedTimeRemaining, averageUploadSpeed

		end )

		:distinctUntilChanged()

end

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

return Rx.Class 'WFCatBinaryUploader'( WFCatBinaryUploader )
