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

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 WFCatDeniedListHelper = require 'WFCatDeniedListHelper'
local WFCatCollaboration = require 'WFCatCollaboration'
local WFCatSharedWithYouUtils = require 'WFCatSharedWithYouUtils'
local WFCatContentRangeUploader = require 'WFCatContentRangeUploader'
local WFCatImportDuplicateFiles = require 'WFCatImportDuplicateFiles'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatPhotoImportance = require 'WFCatPhotoImportance'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatTrashUtils = require 'WFCatTrashUtils'
local WFOzContentTypes = require 'WFOzContentTypes'
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFOzDocUtils = require 'WFOzDocUtils'
local WFCatDirectChunkedUploader = require 'WFCatDirectChunkedUploader'

	-- 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 WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSJsonDecoder = require 'KSJsonDecoder'
local KSTableUtils = require 'KSTableUtils'
local WFAnalytics = require 'WFAnalytics'
local WFArguments = require 'WFArguments'
local WFCapabilities = require "WFCapabilities"
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'

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

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 isValidAssetDocument = assert( WFCatIndexUtils.isValidAssetDocument )
local isNonDeletedStyleDocument = assert( WFCatIndexUtils.isNonDeletedStyleDocument )
local WFCatIndexUtils_isRevisionOnOz = assert( WFCatIndexUtils.isRevisionOnOz )
local WFCatTrashUtils_getMasterIsUploaded = assert( WFCatTrashUtils.getMasterIsUploaded )
local WFCatTrashUtils_getProxyIsUploaded = assert( WFCatTrashUtils.getProxyIsUploaded )
local WFCatTrashUtils_getXmpDevelopIsUploaded = assert( WFCatTrashUtils.getXmpDevelopIsUploaded )
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 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 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 function checkIfDenied( docValue, binaryType )

	local deniedListKey = 'WFCatBinaryUploader'

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

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

local function getDatesFromDocValue( docValue )

	local importDate = derefOpt( docValue, 'importSource', 'importTimestamp' )
	
	local captureDate = derefOpt( docValue, 'captureDate' )
	
	return importDate or 1, captureDate or 1

end

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

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

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

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

end

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

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

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

local function isOriginalLinkInvalid( docValue )

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

end

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

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

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

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

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

local function isSpaceAssetEditable( docValue )

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

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

local function getEncodedUploadHeadersForBinaryType( binaryType, docValue )
	
	local uploadHeaders = {}
	
	-- master/proxy upload X-Read-Meta headers
	if binaryType == 'master' or
		binaryType == 'proxy' or
		binaryType == 'proxy_pano' 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'] = binaryType == 'master' and 'xmp,location' or 'xmp'
		else
			uploadHeaders['X-Read-Meta'] = 'none'
		end
	end
	
	-- In App Contribute Redactions headers
	if docValue._spaceId and 
		( binaryType == 'master' or 
		binaryType == 'proxy' or
		binaryType == 'proxy_pano' or
		binaryType == 'xmp_sidecar' ) 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

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

local function shouldMasterOrProxyWaitForXmpSidecar( docValue )
	-- If we have an XMP sidecar to upload for the asset we want to wait on uploading
	-- the master/proxy 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

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

local function doesAssetHaveXmpDevelopSettingsLocally( docValue ) 
	
	local ozXmpCameraRawSha256 = derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )
	
	local localXmpDevelopInfo = WFCatFileSpec.getXmpDevelopInfoFromDocValue( docValue )
	
	if not localXmpDevelopInfo then
		return false
	end
	
	local localXmpCameraRawSha256 = localXmpDevelopInfo.sha256 or localXmpDevelopInfo.contentId -- iOS stores under contentId

	local sourceLinkForCopy = derefOpt( docValue, '_localOnly', 'sourceXmpDevelopLink' )
	
	return sourceLinkForCopy or ( ozXmpCameraRawSha256 and ozXmpCameraRawSha256 == localXmpCameraRawSha256 )

end

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

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

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

local function shouldSkipDirectUpload( docValue, binaryType )
    return derefOpt( docValue, '_localOnly', 'binaryUpload', binaryType, 'isSkipDirectUpload' )
end

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

local function maybeEmitMasterUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload )
										
	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 = 'master'
	if docValue.subtype == "preset" or docValue.subtype == "profile" then
		binaryType = 'preset_master'
	elseif docValue.subtype == "camera_profile" then
		binaryType = 'camera_profile'
	elseif docValue.type == "watermark" then
		binaryType = 'watermark_original'
	end
	
	local isSkipDirectUpload = shouldSkipDirectUpload( docValue, binaryType )
	
	-- 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 = derefOpt( docValue, '_localOnly', 'localOriginalError' )
	
	local trialOrSubscriptionExpiredError = not not WFCatOzErrors.hasTrialOrSubscriptionExpiredError( docValue )
	
	local insufficientPermissionError = not not WFCatOzErrors.hasInsufficientPermissionError( docValue )

	local hasStorageExceededError = not not WFCatOzErrors.hasStorageExceededError( docValue )

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

	local willNotUploadMaster = derefOpt( docValue,
										   'importSource',
										   'willNotUploadMaster' )
	
	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 spaceId = docValue._spaceId
	local isSpaceAsset = not not spaceId
	
	-- 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 ( copyId and not willNotUploadMaster )
		or ( isSpaceAsset and sourceLinkForCopy ) )
		and shouldUpload
		and not localOriginalHasError
		and not trialOrSubscriptionExpiredError
		and not insufficientPermissionError
		and not hasStorageExceededError
	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 isSkipDirectUpload is set to true, we will abort direct upload and use singleUpload/contentRange instead
		if not isSkipDirectUpload and ozMainPartsRequestUrlTemplate and latestRevisionId then
			mainPartsRequestUrl = ozMainPartsRequestUrlTemplate:replace( '{revision_id}', latestRevisionId )
		end
		
		local uploadHeaders = getEncodedUploadHeadersForBinaryType( binaryType, docValue )
		
		logger:debugf( "emit %s, %s", localPath, masterCreateUrl )
		
		local isReadyToUpload = boolToInt( ( not not masterCreateUrl ) or ( not not mainPartsRequestUrl ) )
		
		-- Note: maybeEmitMasterUploadRecord can be called for both asset documents AND watermark documents.
		if shouldMasterOrProxyWaitForXmpSidecar( docValue ) or shouldMasterOrProxyWaitForExternalXmpSettings( docValue ) then
			isReadyToUpload = false
		end
		
		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

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

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 )
	
	if not needsXmpDevelopSettingsUploaded then
		return
	end
	
	local xmpCameraRaw = derefOpt( docValue, 'develop', 'xmpCameraRaw' )
	
	local isXmpDevelopInvalid = not not derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp/develop', 'invalid' )
	
	local xmpCameraRawSha256 = xmpCameraRaw.sha256

	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' )
	
	-- 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.
	-- A subsequent update will provide the link.
	local wasUploadedPreviously = WFCatTrashUtils_getXmpDevelopIsUploaded( docValue ) and not isXmpDevelopInvalid
		
	local revisionIds = derefOpt( docValue, '_serverProvidedMetadata', 'revision_ids' )

	local sourceLinkForCopy = derefOpt( docValue, '_localOnly', 'sourceXmpDevelopLink' )
	if isXmpDevelopInvalid then -- WF-1323: sourceXmpDevelopLink is only for the first time we upload the settings
		sourceLinkForCopy = nil -- if we're uploading settings a second time they definitely didn't come from the source asset
	end

	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 latestRevisionId = revisionIds and revisionIds[ #revisionIds ]

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

		local binaryType = 'xmp_develop'
		local uploadHeaders = getEncodedUploadHeadersForBinaryType( binaryType, docValue )
		local spaceId = docValue._spaceId
		
		logger:debugf( "emit %s, %s", localPath, xmpDevelopCreateUrl )
		
		local isReadyToUpload = boolToInt( WFCatIndexUtils_isRevisionOnOz( docValue._revId, revisionIds  )
				and xmpDevelopCreateUrl )
		
		emit( localPath, nil, nil, boolToInt( false ), xmpDevelopCreateUrl, binaryType, 
				kContentTypeXML, uploadHeaders, importDate, captureDate, isDenied,
				localSha256, localFileSize, boolToInt( false ), priorityForUpload, 
				sourceLinkForCopy, spaceId, sourceAssetId, isReadyToUpload, docValue.type )

	end

end

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

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
		
		logger:debugf( "emit %s, %s", xmpLocalPath, xmpCreateUrl )
	
		local isReadyToUpload = boolToInt( not not xmpCreateUrl )
	
		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

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

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 ozProxyUrl = 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 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 )						
	
	-- 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 willNotUploadMaster
		and not masterPath
		and ( localPath or ( localVolumeGuid and localRelativePath ) or copyId
				or ( isSpaceAsset and sourceLinkForCopy ) )
		and not ozProxyUrl
		and not ( wasUploadedPreviously or isSpaceAssetWithImpliedProxyUpload )
		and not insufficientPermissionError
		and not hasStorageExceededError
	then

		local contentType = kContentTypeDng

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

		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
		
		logger:debugf( "emit %s, %s", localPath, proxyCreateUrl )
		
		local isReadyToUpload = boolToInt( not not proxyCreateUrl )
		
		if shouldMasterOrProxyWaitForXmpSidecar( docValue ) or shouldMasterOrProxyWaitForExternalXmpSettings( docValue ) then
			isReadyToUpload = false
		end
		
		emit( localPath, localVolumeGuid, localRelativePath, boolToInt( fileMaybeOnOfflineVolume ),
				proxyCreateUrl, binaryType, contentType, uploadHeaders, importDate,
				captureDate, isDenied, imageSha256, fileSize, boolToInt( false ), 
				priorityForUpload, sourceLinkForCopy, spaceId, sourceAssetId, isReadyToUpload,
				docValue.type )

	end

end

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

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 canAssetBeUploaded( docValue )
	if( doesAssetDocumentHaveLocalBinary( docValue ) or isAssetDocumentACopyWithSource( docValue )  or
		isSpaceAssetWithSourceLinks( docValue ) or isSpaceAssetEditable( docValue ) or
		 -- 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.
		 WFPlatformInfo.isAndroid and WFCatIndexUtils.isValidAssetDocument( docValue ) ) then
		return true
	end
		return false
end

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

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

	return false
	
end

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

local function canVersionBeUploaded( docValue )
	
	return not not getLocalExternalDevelopSettingsPath( docValue )

end

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

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 )]],
				
					-- 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

		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 xmp/develop
				maybeEmitXmpDevelopUploadRecord( docValue, emit, importDate, captureDate, isDenied, priorityForUpload, 2 )
					-- NMB-15173 / LRMI-4195062
					-- Add an extra version param set 2 to force re-index, this is needed as as a
					-- fix was made to maybeEmitXmpDevelopUploadRecord to upload xmp_develop settings
					-- to any restored asset that originally had an external develop setting but is
					-- marked as invalid. Previously we would never upload them again.

				-- 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, 35 )
					-- set to 35 for fixing WF-688/WF-808

			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 )
			
				-- 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

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

local function isMaster( binaryType )

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

end

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

local function isXmpDevelop( binaryType )

	return ( binaryType == 'xmp_develop' )
	
end

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

local function isProxy( binaryType )
	
	return binaryType == 'proxy' or binaryType == 'proxy_pano'
	
end

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

local function uploadFinished( self )

	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
			self._uploadFinishedStream:onNext( KSDate.timeInIso8601Format() )
		end )

	end

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

local function clearScheduledAndWaitingBinaryUpload( self )

	if self._scheduledAndWaitingBinaryUpload and
			self._scheduledAndWaitingBinaryUpload.priorityTracker then
	
		self._scheduledAndWaitingBinaryUpload.priorityTracker:dispose()
	
	end

	self._scheduledAndWaitingBinaryUpload = nil

end

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

local function setupBinaryUploadPriorityTracker( self )

	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.binaryType
			}
			
			: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


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

local function markUploadAsInProgress( self, docId, binaryType )

	if not self._scheduledAndWaitingBinaryUpload or
		self._scheduledAndWaitingBinaryUpload.docId ~= docId or
		self._scheduledAndWaitingBinaryUpload.binaryType ~= binaryType then
		
		logger:errorf( "The latest scheduled-and-waiting binary is not what we expected." )
		
	end

	logger:infof( "Binary upload %s:%s is now in-flight", docId, binaryType )

	clearScheduledAndWaitingBinaryUpload( self )
	
	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

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

local function checkDirectUploadFailureCount(self, docId, binaryType )
    
    return self._directUploadConsecutiveFailureCounter < numberOfTriesBeforeRevertToLegacyUpload

end

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

local function putAndTrackPerformance( self, uploadUrl, binaryType, filePath, fileSize, options, docId, isDirectUpload )

	local putFile

	options.notifyWhenHttpRequestIsInProgress = true
	options.includeProgress = self._includeProgress
	
	local requestInProgress = false
	
	local function progressEventHandler( _, _, progress )
		
		if not requestInProgress then
		
			markUploadAsInProgress( self, docId, binaryType )
			requestInProgress = true
			
		end
		
		if progress then
		
			KSTableUtils.setNested( self._inFlightUploadProgress, progress, docId, binaryType )
			self._debugInFlightUploadProgressUpdateStream:onNext( self._inFlightUploadProgress )
			
		end
		
	end	
	
	local function putPartRequestSuccessHandler()
		self._directUploadConsecutiveFailureCounter = 0
	end

	if isDirectUpload then
		
		putFile = self._directChunkedUploader:doDirectChunkedUpload( uploadUrl, filePath, options, progressEventHandler, putPartRequestSuccessHandler )
			
			:tap( progressEventHandler )
			
			:takeLast( 1 )
			
			:catch( function( responseData, headers, url )
				
				-- 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 reponseData = string.format("Can't use direct upload with %s error and status codes is %s", responseData, headers and headers.status_code or 'unknown' )
				logger:errorf( reponseData )
								
				local requestInfo = {
					errorUrl = url
				}
				
				return Observable.throw( responseData, headers, requestInfo )
				
			end )


	elseif isMaster( binaryType ) and fileSize >= self._chunkedUploader:chunkFileSize() then
		
		local hasAppliedSkipDirectUploadOverlay = false
		
		putFile = self._chunkedUploader:doChunkedUpload( uploadUrl, filePath, options, progressEventHandler )
		
			:tap( progressEventHandler )
			
			:map( function( ... )
			
				if not hasAppliedSkipDirectUploadOverlay then
					self._docStore:applyOverlay {
						_id = docId,
						_localOnly = {
							binaryUpload = {
								[ binaryType ] = {
									isSkipDirectUpload = true
								}
							}
						}
					}  
					:call( runUntilDone )
		
					hasAppliedSkipDirectUploadOverlay = true
			
				end
				
				return ...
				
			end )
			
			:takeLast( 1 )
		
	else
	
		-- WF-1716, should be temporary until we can do chunked/direct uploads for xmp_develop too
		if isXmpDevelop( binaryType ) 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', binaryType ), 
			putFile, { fileSize = fileSize } )
		
		
end

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

local function getRecentlyUploadedSha256( self, docId, binaryType )
	
	return derefOpt( self._recentlyUploaded, docId, binaryType )

end

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

local function getUploadingAssetsWithSha256( self, sha256 )

	return sha256 and self._uploadingAssetWithSha256[ sha256 ] or 0

end

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

local function decrementUploadingAssetWithSha256( self, docId, binaryType )

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

	if isMaster( binaryType ) then

		local sha256 = getRecentlyUploadedSha256( self, docId, binaryType )

		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

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

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

local function trackUploadForDiagnostics( self, spaceId, binaryType, 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 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, binaryType, 'total' ) or 0
		setNested( self._uploadsByType, total + 1, spaceIdOrCatalog, binaryType, 'total' )
		
		if wasCopied then
			ozCopies = derefOpt( self._uploadsByType, spaceIdOrCatalog, binaryType, kOzCopies ) or 0
			setNested( self._uploadsByType, ozCopies + 1, spaceIdOrCatalog, binaryType, kOzCopies )
		end
		
		if wasDirectS3Upload then
			s3Uploads = derefOpt( self._uploadsByType, spaceIdOrCatalog, binaryType, kS3Uploads ) or 0
			setNested( self._uploadsByType, s3Uploads + 1, spaceIdOrCatalog, binaryType, kS3Uploads )
		end
		
		return self._uploadsByType
		
	end

	if self._uploadsByTypeStream then
	
		self._uploadsByType = updateUploadCounts( binaryType )
		self._uploadsByTypeStream:onNext( self._uploadsByType )
		
	end

end

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

local function addToRecentlyUploaded( self, docId, binaryType, sha256 )

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

	if recentlyUploadedForDocId[ binaryType ] then
		
		WFAnalytics.trackEvent {
			['event.type'] = binaryType .. "_uploaded",
			['event.workflow'] = 'SYNC',
			['event.subcategory'] = 'binary_upload_duplicate_schedule',
			['content.id'] = docId
		}
		
		logger:errorf( "addToRecentlyUploaded recentlyUploadedForDocId ALREADY SET for %s %s", binaryType, docId )
	end

	recentlyUploadedForDocId[ binaryType ] = sha256 

	self._recentlyUploaded[ docId ] = recentlyUploadedForDocId

	if isMaster( binaryType ) then
		local countOfAssets = self._uploadingAssetWithSha256[ sha256 ] or 0
		self._uploadingAssetWithSha256[ sha256 ] = countOfAssets + 1
	end

end

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

local function clearFromRecentlyUploaded( self, reason, docId, binaryType, uploadAttemptedOrCancelled )

	logger:debugf( "clearFromRecentlyUploaded %s %s because %s", binaryType, 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
		decrementUploadingAssetWithSha256( self, docId, binaryType )
	end

	recentlyUploadedForDocId[ binaryType ] = nil

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

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

local function maybeClearUploadPriorityModifier( self, 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

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

local function delayedClearFromRecentlyUploaded( self, reason, docId, binaryType, 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, binaryType, reason )

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

			clearFromRecentlyUploaded( self, reason, docId, binaryType, true )

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

		end )
		:call( runUntilDone )
		
end

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

local function delayedClearFromRecentlyUploaded_NoUpload( self, reason, docId, binaryType, 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.

	decrementUploadingAssetWithSha256( self, docId, binaryType )

	local secondsToDelay = optDelay or self._waitForOtherUploadSeconds

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

	local sha256 = getRecentlyUploadedSha256( self, docId, binaryType )

	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()

			clearFromRecentlyUploaded( self, reason, docId, binaryType, 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.
					uploadFinished( self )
				end

			end

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

end

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

local function uploadProgressStatusUpdate(self, 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

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

local function addToScheduledUploads( self, docId, binaryType, subscription, localFileSize, priority )

	self._numberOfScheduledBinaryUploads = self._numberOfScheduledBinaryUploads + 1

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

	if uploadSubscriptionsForDocId[ binaryType ] then
		logger:errorf( "Already have subscription for uploading %s binaryType = %s", docId, binaryType )
	else
		uploadProgressStatusUpdate( self, "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", docId, binaryType )
	
	self._scheduledAndWaitingBinaryUpload = {
		docId = docId,
		binaryType = binaryType,
		subscription = subscription,
		priority = priority
	}

	setupBinaryUploadPriorityTracker( self )

	uploadSubscriptionsForDocId[ binaryType ] = subscription
	
	self._scheduledUploadSubscriptions[ docId ] = uploadSubscriptionsForDocId

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

end

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

local function removeFromScheduledUploads( self, docId, binaryType )

	if derefOpt( self._scheduledUploadSubscriptions, docId, binaryType ) == 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.binaryType == binaryType 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.
		
		clearScheduledAndWaitingBinaryUpload( self )
		
	end

	local uploadSubscriptionsForDocId = self._scheduledUploadSubscriptions[ docId ] or {}
	
	uploadSubscriptionsForDocId[ binaryType ] = 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

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

local function getDocumentUrl( self, docValue )

	return derefOpt( docValue, '_serverProvidedMetadata', 'links', "self", 'href' )
	
end
--------------------------------------------------------------------------

local function checkAndUpdateDocIfOzhasMaster( self, 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)
		:call( runUntilDone )

end

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

local function isDocUpdatedByManualCheckWithOz ( self, docValue )
	
	local documentUrl = getDocumentUrl( self, docValue )

	return checkAndUpdateDocIfOzhasMaster( self, documentUrl )
		:map( function( result )
			 return result ~= false
		end )

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

local function eventuallyClearBlackoutWindow( self, docId, binaryType, waitCount )

	-- Don't let this original be uploaded again

	waitCount = waitCount or 0

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

			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.", binaryType, docId )

				uploadConfirmed = true

			elseif isMaster( binaryType ) then
				
				uploadConfirmed = ( getOriginalLinkFromDocValue_IgnoringShaError( docValue ) ~= nil ) or
									( docValue._spaceId and getOriginalExistsFromDocValue( docValue ) )
				uploadedSha256 = getRecentlyUploadedSha256( self, docId, binaryType )
				currentSha256 = derefOpt( docValue, 'importSource', 'sha256' )
			
			elseif isProxy( binaryType ) then
				
				uploadConfirmed = ( WFCatFileSpec.getProxyLinkFromDocValue( docValue ) ~= nil ) or
									( docValue._spaceId and getProxyExistsFromDocValue( docValue ) )
				uploadedSha256 = getRecentlyUploadedSha256( self, docId, binaryType )
				currentSha256 = derefOpt( docValue, 'importSource', 'sha256' )
				
			elseif binaryType == 'xmp_develop' then
				
				uploadConfirmed = WFCatFileSpec.getXmpDevelopLinkFromDocValue( docValue ) ~= nil
				uploadedSha256 = getRecentlyUploadedSha256( self, docId, binaryType )
				currentSha256 = derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )

			elseif binaryType == 'xmp_sidecar' then

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

			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, binaryType )

				delayedClearFromRecentlyUploaded( self, "sha no longer matches", docId, binaryType )
				
			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.

				delayedClearFromRecentlyUploaded( self, "upload confirmed", docId, binaryType )

					: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
			
							maybeClearUploadPriorityModifier( self, 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.

						KSTableUtils.setNested( self._inFlightUploadProgress, nil, docId, binaryType )
						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",
						binaryType, 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.
					
					return isDocUpdatedByManualCheckWithOz( self, docValue )
						:tap( function ( status )
							if( status == true ) then
								logger:warnf( "Upload confirmed by manual check for %s", docId )
								delayedClearFromRecentlyUploaded( self, "upload confirmed by manual check", docId, binaryType )
							else
								logger:warnf( "Assuming upload failed for asset %s. Retrying...", docId )
								delayedClearFromRecentlyUploaded( self, "assuming upload failed", docId, binaryType )
							end
						end )
						:call( runUntilDone )
				else
				
					-- Keep waiting
					
					eventuallyClearBlackoutWindow( self, docId, binaryType, waitCount )
					
				end
				
			end
			
		end )
		
		:call( runUntilDone )

end

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

local function uploadCompleted( self, docId, binaryType, uploadSha256 )

	eventuallyClearBlackoutWindow( self, docId, binaryType )
	
	local docStore = self._docStore
	
	if isMaster( binaryType ) then
		
		-- When upload complete, clear isSkipDirectUpload flag 
		docStore:applyOverlay  {
			_id = docId,
			_localOnly = {
				binaryUpload = {
					[ binaryType ] = {
						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 binaryType == 'xmp_sidecar' 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

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

local function addTransferStatsToDiagnosticLog( self, 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

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

local function sanitizeDirectUploadUrlForAnalytics( url )

	url = url and url:gsub( "&X%-Amz%-Security%-Token=.*&", "&X-Amz-Security-Token=***&" )
	
	return url

end

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

local function WFAnalyticsTrackEventNotifySuccess( self, docId, uploadUrl, binaryType, contentType, isCopy, isDirectUpload )

	local uploadMethod = 'oz_upload'
	
	if isCopy then
		uploadMethod = 'oz_copy'
	elseif isDirectUpload then
		uploadMethod = 'direct_upload'
	elseif not checkDirectUploadFailureCount( self, docId, binaryType ) then
		uploadMethod = "direct-upload fall back to oz-upload"
	end

	WFAnalytics.trackEvent {
		['event.type'] = binaryType .. "_uploaded",
		['event.workflow'] = 'SYNC',
		['event.subcategory'] = 'binary_upload_success',
		['event.uploadMethod'] = uploadMethod,
		['content.id'] = docId,
		['content.type'] = contentType,
		['content.url'] = sanitizeDirectUploadUrlForAnalytics( uploadUrl ),
	}

end

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

local function WFAnalyticsTrackEventNotifyFailure( self, failureInfo )

	local eventSubcategory

	if failureInfo.reason then
		eventSubcategory = failureInfo.reason
	elseif failureInfo.fileSize and failureInfo.fileSize >= self._chunkedUploader:chunkFileSize() then
		eventSubcategory = 'binary_upload_chunk'
	else
		eventSubcategory = 'binary_upload_data'
	end

	local uploadMethod = 'oz_upload'
	
	if failureInfo.isCopy then
		uploadMethod = 'oz_copy'
	elseif failureInfo.isDirectUpload then
		uploadMethod = 'direct_upload'
	elseif not checkDirectUploadFailureCount( self, failureInfo.docId, failureInfo.binaryType ) then
		uploadMethod = "direct-upload fall back to oz-upload"
	end

	local subtype -- subtype, content type, filesize shall be empty for the InvalidLocalOriginal case, so do not assign default vale

	if failureInfo.contentType then

		if WFOzContentTypes.isVideo( failureInfo.contentType ) then 
			subtype = 'video'
		else
			subtype = 'image'
		end

	end

	local errorEvent = {}

	errorEvent['event.uploadMethod'] = uploadMethod
	errorEvent['event.type'] = 'error'
	errorEvent['event.workflow'] = 'SYNC'
	errorEvent['event.subcategory'] = eventSubcategory
	errorEvent['event.url'] = sanitizeDirectUploadUrlForAnalytics( failureInfo.url )
	errorEvent['event.error_type'] = failureInfo.binaryType .. "Upload.Error"
	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 = string.format( "%s", tostring( failureInfo.responseData ) )
		
		errorEvent['event.subtype'] = 'OS'
		errorEvent['event.error_code'] = errorCode
		errorEvent['event.error_desc'] = annotatedReason

	else

		local ozErrorResponse = string.format( "%s",
									   tostring( 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

	WFAnalytics.trackEvent( errorEvent )

end

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

local function addHttpErrorToLog( self, docId, masterPutUrl, responseData, headers, 
		binaryType, contentType, localSize, isCopy, isDirectUpload )

	-- Special case for HTTP transit failure.

	local errorLog = self._uploadErrorLog

	if not headers then

		local annotatedReason = string.format( "%s", tostring( responseData ) )

		local logFmtStr = "Unable to communicate with Oz URL %s\n\nreason = %s"

		logger:errorf( logFmtStr, masterPutUrl, annotatedReason )

	else

		-- 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 = binaryType 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,
		binaryType = binaryType,
		contentType = contentType,
		fileSize = localSize,
		httpError = true,
		isCopy = isCopy,
		isDirectUpload = isDirectUpload
	}

	WFAnalyticsTrackEventNotifyFailure( self, failureInfo )

end

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

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

	errorLog[ #errorLog + 1 ] = {

		time = KSDate.timeInIso8601FormatWithTimeZone(),

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

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

end

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

local function getRequestOptionsFor( binaryType, contentType, isCopy, additionalUploadHeaders, enableVideoProxies )
	
	local options = {
		dataFormat = 'filePath',
		requestPriority = WFRxHttpClient.requestPriorityMedium,
		requestCost = uploadHttpRequestCost
	}
	
	local isVideo = WFOzContentTypes.isVideo( contentType )

	if isCopy then
		contentType = 'application/json'
	end

	if binaryType == 'master' then
		
		local requestHeaders = {
			{
				header = 'X-Generate-Renditions',
				value = 'all',
			},
			{
				header = 'Content-Type',
				value = contentType,
			},
			{
				header = "Accept",
				value = "application/json",
			},
		}
		
		if isCopy then
			table.insert( requestHeaders, {
				header = 'X-Async-Copy',
				value = 'true'
			} )
		end
		
		if not isVideo then
			-- don't request proxy generation for videos, but 
			-- do for image/* and raw files (application/octet-stream)
			table.insert( requestHeaders, 
				{
					header = 'X-Generate-Proxies',
					value = 'all',
				} )

		elseif isVideo and enableVideoProxies then
			--FIXME [zhiqhuan 2021-11-23]:
			-- We would have feature flag to turn header on and off, but with "X-Generate-Proxies: standard" for now.
			-- Will change "standard" to "all" "X-Generate-Proxies: all" before 2022 June release.
			table.insert( requestHeaders,
				{
					header = 'X-Generate-Proxies',
					value = 'standard',
				} )
		end
		
		options.requestHeaders = requestHeaders

	elseif binaryType == 'preset_master' then
		-- master for preset/profile
		
		local requestHeaders = {
			{
				header = 'Accept',
				value = 'application/json',
			},
			{
				header = 'Content-Type',
				value = contentType or kContentTypeXML,
			},
		}
		
		options.requestHeaders = requestHeaders
		
	elseif binaryType == 'camera_profile' then
		-- master for camera_profile
		local requestHeaders = {
			{
				header = 'Accept',
				value = 'application/json',
			},
			{
				header = 'Content-Type',
				value = contentType or kContentTypeOctetStream,
			},
		}
		options.requestHeaders = requestHeaders
		
	elseif binaryType == 'proxy' or binaryType == 'proxy_pano' 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 binaryType == 'xmp_develop' then
		
		options.requestHeaders = {
				{
					header = 'X-Generate-Renditions',
					value = 'all',
				},
				{
					header = 'X-Generate-Proxies',
					value = 'all',
				},
				{
					header = 'Content-Type',
					value = contentType or kContentTypeXML,
				},
				{
					header = 'Accept',
					value = 'application/json',
				},
			}

	elseif binaryType == 'xmp_sidecar' then
		
		options.requestHeaders = {
				{
					header = 'Content-Type',
					value = contentType or kContentTypeXML,
				},
				{
					header = 'Accept',
					value = 'application/json',
				},
			}
			
	elseif binaryType == 'watermark_original' then
		
		options.requestHeaders = {
				{
					header = 'Content-Type',
					value = contentType,
				},
				{
					header = 'Accept',
					value = 'application/json',
				},
			}
			
	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

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

local function recordSuccess( self, docId, uploadUrl, binaryType, localSize, response, contentType, isCopy, isDirectUpload )

	localSize = localSize or 0

	local bucket = binaryType
	
	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
		uploadProgressStatusUpdate(self, "increaseBytesUploaded", localSize )
		
	end

	local progress = derefOpt( self._inFlightUploadProgress, docId, binaryType )

	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
			addTransferStatsToDiagnosticLog( self, bucket, endTime - startTime, bytesUploaded )
		end
	
	end
	
	if progressOfChunks then
	
		for _, chunkProgress in pairs( progressOfChunks ) do
			
			trackProgressEventStats( chunkProgress )
			
		end
	
	elseif progress then
	
		trackProgressEventStats( progress )
		
	else
	
		addTransferStatsToDiagnosticLog( self, bucket, nil, localSize )
	
	end

	-- This is limited to mobile and Mac App Store only because tracking successes for regular desktop 
	-- builds overwhelms highbeam. If desktop fully switches over to Dunamis, the if can be removed
	-- and we can always track the upload success.
	
	if WFPlatformInfo.isMobile() or WFPlatformInfo.isMacAppStore then
	
		WFAnalyticsTrackEventNotifySuccess( self, docId, uploadUrl, binaryType, contentType, isCopy, isDirectUpload )
	
	end
	
	self._binaryUploadCompletionStream:onNext( docId )

end

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

local function removeSidecarFileAnnotation( self, 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

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

local function applyInvalidOriginalOverlay( self, 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,
			binaryType = 'master',
			httpError = false,
			reason = info or reason,
		}

		WFAnalyticsTrackEventNotifyFailure( self, 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

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

local function resolveSha256ErrorInInputSource( self, 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 applyInvalidOriginalOverlay( self, 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 )
			
			local failureInfo = {
				docId = docId,
				url = masterUploadUrl,
				binaryType = 'master',
				httpError = false,
				reason = "re-uploading corrupted original"
			}
			
			WFAnalyticsTrackEventNotifyFailure( self, 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 applyInvalidOriginalOverlay( self, docId, filePath, masterUploadUrl, nil, "sha256 mismatch" )

		end

	end )

end

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

local function disposeOfUploadForDeletedDocId( self, docId )

	local uploadSubscriptionsForDocId = self._scheduledUploadSubscriptions[ docId ] or {}
		
	for binaryType, subscription in pairs( uploadSubscriptionsForDocId ) do 
	
		logger:debugf( "Aborting %s upload for docId %s", binaryType, docId )

		subscription:dispose()

		clearFromRecentlyUploaded( self, "disposed", docId, binaryType, true )

		removeFromScheduledUploads( self, docId, binaryType )
		
	end

end

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

local function disposeAllActiveUploads( self )

	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 binaryType, subscription in pairs( uploadSubscriptionsForDocId ) do 
		
			logger:debugf( "Aborting upload for docId %s", docId )
	
			subscription:dispose()
	
			clearFromRecentlyUploaded( self, "disposed", docId, binaryType, 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

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

local function handleUploadFailure( self, docId, binaryType, uploadUrl, responseData, 
		headers, invalidOriginal, contentType, localSize, isCopy, isDirectUpload )

	self._errorStatusStream:onNext( docId, responseData )

	addHttpErrorToLog( self, docId, uploadUrl, responseData, headers, binaryType, contentType, localSize, isCopy, isDirectUpload )

	if invalidOriginal then
		delayedClearFromRecentlyUploaded( self, "wait for docstore update as invalidOriginal", docId, binaryType )
		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
		} )

		delayedClearFromRecentlyUploaded( self, "wait for docstore update as denied", docId, binaryType )

	else

		-- Upload wasn't denied so try to upload the original again.

		local uploadErrorDelay = 10
		delayedClearFromRecentlyUploaded( self, "upload failed, delay to let other assets upload first", docId, binaryType, uploadErrorDelay )

	end

end

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

local function errorShaDoesNotMatch( binaryType, responseData, headers )

	if isMaster( 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

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

local function errorTrialOrSubscriptionExpired( binaryType, 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

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

local function errorUploadMasterWhenNotAllowed( binaryType, responseData, headers )

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

		return true

	end

	return false

end

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

local function errorInsufficientPermission( binaryType, responseData, headers )

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

		return true
		
	end

	return false

end

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

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

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

local function errorPayloadMustExist( binaryType, responseData, headers )

	if not isMaster( binaryType ) then
		return false
	end

	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

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

local function applyRemoveWillNotUploadMasterOverlay( self, docId )

	return self._docStore:applyOverlay {
			_id = docId,
			importSource = {
				willNotUploadMaster = { _delete = true },
			},
		}

end

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

local function applySpecificErrorOverlay( self, docId, errorType )
	
	local overlay = WFCatOzErrors.createOverlayForOzErrorSubtype( docId, errorType )
	return self._docStore:applyOverlay( overlay )
	
end

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

local function checkForAccountError( self, docId, binaryType, responseData, headers )

	if errorTrialOrSubscriptionExpired( binaryType, responseData, headers ) then
		
		return applySpecificErrorOverlay( self, docId, WFCatOzErrors.Errors.kTrialOrSubscriptionExpiredError )
			:flatMap( function()
				return Observable.throw( responseData, headers, true )
			end )
	
	end

	if errorInsufficientPermission( binaryType, responseData, headers ) then
		
		return applySpecificErrorOverlay( self, docId, WFCatOzErrors.Errors.kInsufficientPermissionError )
			:flatMap( function()
				return Observable.throw( responseData, headers, true )
			end )
	
	end
		
end

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

local function errorAndMaybeNewIdIfMasterOrProxyResourceDoesNotExist( binaryType, responseData, headers )

	if ( isMaster( binaryType ) or isProxy( binaryType ) or 
		binaryType == 'xmp_sidecar' or binaryType == 'xmp_develop' ) 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

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

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 = WFCatFileSpec.getXmpDevelopLinkFromDocValue( sourceDocValue )
	localOnly.sourceProxyLink = WFCatFileSpec.getProxyLinkFromDocValue( sourceDocValue )

	localOnly.sourceAssetId = sourceDocValue._id

	return {
		_id = docIdToUpdate,
		_localOnly = localOnly,
	}

end

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

local function applyOverlayToRemoveSourceAsset( self, docStore, docValue, copyBinaryUrl )

	local localOnly = {}

	localOnly.sourceOriginalLink = { _delete = true }
	localOnly.sourceXmpLink = { _delete = true }
	localOnly.sourceXmpDevelopLink = { _delete = true }
	localOnly.sourceProxyLink = { _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

	if not hasProxyLocally and not hasMasterLocally then
		return applyInvalidOriginalOverlay( self, docValue._id, nil, copyBinaryUrl, "CopyFailedSourceMissing" )
	end

	return docStore:applyOverlay( {
		_id = docValue._id,
		_localOnly = localOnly,
		importSource = importSource
	} )

end

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

local function applyOrRemoveNewSourceLinksOrErrorOverlayForCopy( self, docId, newSourceId, binaryType, 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 applyOverlayToRemoveSourceAsset( self, 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 applyOverlayToRemoveSourceAsset( self, 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

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

local function uploadOneBinary( self, docId, localPath, localSize, binaryUrl, binaryType, contentType, addlUploadHeaders, sha256, uploadPriority, isDirectUpload )

	logger:debugf( "Uploading %s for docId %s from %s to %s", binaryType, docId, localPath, binaryUrl )

	local enableVideoProxies = self._enableVideoProxies

	local options = getRequestOptionsFor( binaryType, contentType, false, addlUploadHeaders, enableVideoProxies )
	options.requestDescription = string.format( '%s : Uploading %s', docId, binaryType )
	
	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 putAndTrackPerformance( self, binaryUrl, binaryType, localPath, localSize, options, docId, isDirectUpload )
			
	end )
	
	:catch( function( responseData, headers, requestInfo )
	
		if requestInfo then
		
			local errorUrl = requestInfo.errorUrl
			
			local failureInfo = {
				docId = docId,
				url = errorUrl,
				responseData = responseData,
				headers = headers,
				binaryType = binaryType,
				contentType = contentType,
				fileSize = localSize,
				httpError = true,
				isCopy = false,
				isDirectUpload = isDirectUpload
			}
			
			-- WF-1752 If we detect any asset has encountered 3 consecutive failures, then we should redirect all 
			-- assets to use the older upload mechanism for 30 mins before switching back to DU.
			if self._useDirectUploadIfPossible and not self._directUploadFallBackToOzUploadTimerSub and not checkDirectUploadFailureCount( self, docId, binaryType ) then

				self._useDirectUploadIfPossible = false

				self._directUploadFallBackToOzUploadTimerSub = WFRxLowFidelityTimer.timer( self._directUploadFallBackToOzUploadWindowInSecond )

					:subscribe( function()
						-- Reset _useDirectUploadIfPossible
						self._useDirectUploadIfPossible = true
						self._directUploadConsecutiveFailureCounter = 0
						self. _directUploadFallBackToOzUploadTimerSub:dispose()
						self. _directUploadFallBackToOzUploadTimerSub = nil
					end )
					
			end
			
			WFAnalyticsTrackEventNotifyFailure( self, failureInfo )
					
			return Observable.throw( responseData, headers, false, isDirectUpload )
			
		end

		if errorShaDoesNotMatch( binaryType, 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
					
						local failureInfo = {
							docId = docId,
							url = binaryUrl,
							responseData = responseData,
							headers = headers,
							binaryType = binaryType,
							contentType = contentType,
							fileSize = localSize,
							httpError = true,
							reason = "upon reception of sha256 mismatch from oz recalculation of sha matched",
							isCopy = false,
							isDirectUpload = isDirectUpload
						}
					
						WFAnalyticsTrackEventNotifyFailure( self, failureInfo )
								
						return Observable.throw( responseData, headers, false, isDirectUpload )

					else

						return applyInvalidOriginalOverlay( self, docId, localPath, binaryUrl, nil, "sha256 mismatch from Oz confirmed on recalculation" )
							:flatMap( function()
								return Observable.throw( responseData, headers, true, isDirectUpload )
							end )

					end
					
				end )

		end


		if errorUploadMasterWhenNotAllowed( binaryType, responseData, headers ) then

			return applyRemoveWillNotUploadMasterOverlay( self, 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, isDirectUpload )

				end )

		end

		if errorPayloadMustExist( binaryType, responseData, headers ) then

			return applyInvalidOriginalOverlay( self, docId, localPath, binaryUrl, nil, "payload must exist" )
				:flatMap( function()
					return Observable.throw( responseData, headers, true, isDirectUpload )
				end )

		end
		
		if errorContentTypeMismatch( responseData, headers ) then

			return applyInvalidOriginalOverlay( self, docId, localPath, binaryUrl, "UploadFailedContentTypeMismatch" )
				:flatMap( function()
					return Observable.throw( responseData, headers, true, isDirectUpload )
				end )

		end
		
		local maybeAccountError = checkForAccountError( self, docId, binaryType, responseData, headers )
		
		if maybeAccountError then
			return maybeAccountError
		end

		return Observable.throw( responseData, headers, false )

	end )

	:tap(
		function( response )
			recordSuccess( self, docId, binaryUrl, binaryType, localSize, response, contentType, false, isDirectUpload )
		end,
		nil,
		function()
			uploadCompleted( self, docId, binaryType, sha256 )
		end )
		
	:catch( function( responseData, headers, invalidOriginal )

			-- 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 binaryType == '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' and response.code == "SourceDoesNotExist" then
				
					recordSuccess( self, docId, binaryUrl, binaryType, localSize, responseData, contentType, false, isDirectUpload )
					uploadCompleted( self, docId, binaryType, sha256 )
					
					uploadFailed = false
					
				end
				
			end

			if uploadFailed then
				
				handleUploadFailure( self, docId, binaryType, binaryUrl, responseData, headers, invalidOriginal, contentType, localSize, false, isDirectUpload )
			
			end
			
			return Observable.just( false )

	end )


end

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

local function copyBinaryOnOz( self, docId, uploadUrl, binaryType, contentType, additionalUploadHeaders, sourceAssetLink )

	logger:debugf( "Copying %s for docId %s from %s to %s", binaryType, docId, sourceAssetLink, uploadUrl )

	local data = '{ "file_href": "' .. sourceAssetLink .. '" }'

	local enableVideoProxies = self._enableVideoProxies

	local options = getRequestOptionsFor( binaryType, contentType, true, additionalUploadHeaders, enableVideoProxies )

	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
		
			markUploadAsInProgress( self, docId, binaryType )
			requestInProgress = true

		end
	
		KSTableUtils.setNested( self._inFlightUploadProgress, progress, docId, binaryType )
		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 errorUploadMasterWhenNotAllowed( binaryType, responseData, headers ) then

				return applyRemoveWillNotUploadMasterOverlay( self, 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
			
			local maybeAccountError = checkForAccountError( self, docId, binaryType, responseData, headers )
		
			if maybeAccountError then
				return maybeAccountError
			end
	
			return Observable.throw( responseData, headers, false )
			
		end )

end

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

local function copyOneBinary( self, docId, binaryUrl, binaryType, sha256, contentType, addlUploadHeaders, sourceLinkForCopy )
	
	local copyBinaryObservable = copyBinaryOnOz( self, docId, binaryUrl, binaryType, contentType, addlUploadHeaders, sourceLinkForCopy )
	
	copyBinaryObservable = copyBinaryObservable:tap( function( dataLength, response, headers )
			recordSuccess( self, docId, binaryUrl, binaryType, dataLength, response, contentType, true, false )
		end,
		nil, -- caller must issue call to handleUploadFailure
		function()
			uploadCompleted( self, docId, binaryType, sha256 )
		end )

	return copyBinaryObservable

end

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

local function uploadOneMasterOrProxy( self, docId, localFilePath, localFileSize, binaryUrl, binaryType, contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload )

	-- Upload the master.

	if ( isFirstUploadForSha256 
			or binaryType == "watermark_original" 
			or WFCapabilities.getCapability( 'WFCatBinaryUploaderTest.debug_allowSameSha256Upload' ) ) then

		if localFilePath then

			return uploadOneBinary( self, docId, localFilePath, localFileSize, binaryUrl, binaryType, contentType, addlUploadHeaders, sha256, uploadPriority, isDirectUpload )

		else

			-- Copied an binary that didn't have a master and source was deleted before we could upload it.

			return applyInvalidOriginalOverlay( self, 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.
					delayedClearFromRecentlyUploaded( self, "upload failed, wait for docstore update as CopyFailedSourceMissing", docId, binaryType, self._waitForCopyFailedSeconds )
				end )

		end

	else
	
		delayedClearFromRecentlyUploaded_NoUpload( self, "Uploading first asset with same sha256, delay so we can copy later", docId, binaryType )
		return Observable.just( true )

	end

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

local function uploadOrCopyOneMasterOrProxy( self, docId, localFilePath, localFileSize, binaryUrl, copyBinaryUrl, binaryType, contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload )
	
	if binaryType == "watermark_original" then
		return uploadOneMasterOrProxy( self, docId, localFilePath, localFileSize, binaryUrl, binaryType,
						contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload )
	end
	
	return WFCatImportDuplicateFiles.findUploadedAssetWithSha256HintNotDocId( self._catalog, sha256, docId )
		:flatMap( function( sourceAssetId, sourceHasNoMaster )

			if sourceAssetId == nil then
				return uploadOneMasterOrProxy( self, docId, localFilePath, localFileSize, binaryUrl, binaryType,
						contentType, addlUploadHeaders, sha256, isFirstUploadForSha256, uploadPriority, isDirectUpload )
			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 isMaster( binaryType ) then
					sourceAssetLink = WFCatFileSpec.getOriginalLinkFromDocValue( sourceDocValue )
				else
					sourceAssetLink = WFCatFileSpec.getProxyLinkFromDocValue( sourceDocValue )
				end
				
				if sourceAssetLink == nil then

					if localFilePath then

						return uploadOneBinary( self, docId, localFilePath, localFileSize, binaryUrl, binaryType, contentType, addlUploadHeaders, sha256, uploadPriority, isDirectUpload )

					else

						-- Copied an asset that didn't have a master that was deleted before we could upload it.
						return applyInvalidOriginalOverlay( self, docId, nil, binaryUrl, "CopyFailedSourceMissing" )
							:finally( function()
								delayedClearFromRecentlyUploaded( self, "copy failed, wait for docstore update CopyFailedSourceMissing", docId, binaryType )
							end )

					end

				else

					return copyOneBinary( self, docId, copyBinaryUrl, binaryType, sha256, contentType, addlUploadHeaders, sourceAssetLink )
					:tapOnError( function( responseData, headers )
						handleUploadFailure( self, docId, binaryType, copyBinaryUrl, responseData, headers, false, contentType, localFileSize, true, false )
					end )

				end

			end )

		end )

end

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

local function maybePerformanceWrap( id, observable )

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

end 

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

local function getBinaryFullPathAndAssetOfflineStatus( self, localFileVolumeGuid, localFileRelativePath, localFilePath, binaryType )

	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 localFileVolumeGuid and localFileRelativePath then
		
		if binaryType == "watermark_original" then
		
			binaryFullPath = WFFileUtils.createFullPathFromRelativePathAndCatalogPath( self._catalog, localFileRelativePath, binaryType )
			
		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
		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, binaryType )
			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

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

local function deleteAsset( self, docId )

	logger:debugf( "deleteAsset: delete the docId = %s", docId )
	
	return self._docStore:applyOverlay {
		_id = docId,
		_deleted = true
	}
	
end

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

local function handleMasterNotExistsLocally( self, docId, binaryUrl, binaryType, contentType )
	
	logger:warnf( "Trying to upload %s for %s but original is missing.", binaryType, 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,
					binaryType = binaryType,
					contentType = contentType,
					httpError = false,
					reason = reason
				}

				WFAnalyticsTrackEventNotifyFailure( self, failureInfo )
				
				return deleteAsset( self, 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,
						binaryType = binaryType,
						contentType = contentType,
						httpError = false,
						reason = reason
					}
					
					WFAnalyticsTrackEventNotifyFailure( self, failureInfo )
					
					return deleteAsset( self, 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,
						binaryType = binaryType,
						contentType = contentType,
						httpError = false,
						reason = reason
					}
				
					WFAnalyticsTrackEventNotifyFailure( self, failureInfo )
					
					return deleteAsset( self, docId )
					
				end

				return self._docStore:applyOverlay {
					_id = docId,
					_localOnly = {
						originalFilePath = localUrl,
					}
				}
				
			end
	
		end )

end

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

local function resetDetailedProgressReportCounters( self )

	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

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

local function sendFinalDetailedProgressUpdate( self )

	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 )
		resetDetailedProgressReportCounters( self )
		
	end
	
end

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

local function getTypeOfJoinForSpaceTable( onlyUploadSpaceAssets )

	local typeOfJoin = "LEFT"
	 
	if onlyUploadSpaceAssets then
		typeOfJoin = "INNER"
	end
	
	return typeOfJoin

end

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

local function getStandardBinaryUploadCandidatesStream( self )

	local typeOfJoin = getTypeOfJoinForSpaceTable( self._onlyUploadSpaceAssets )
	
	local needsUploadQueryStr = string.format( 
			[[SELECT needsBinaryUpload.docId, localFilePath, uploadUrl, binaryType, 
					contentType, uploadHeaders, isDenied, sha256, fileSize, 
					sha256Error, sourceLinkForCopy, localVolumeGuid, localRelativePath,
					priorityForUpload, spaceId, mainPartsRequestUrl
				FROM needsBinaryUpload
				%s JOIN space ON space.docId == needsBinaryUpload.spaceId
				WHERE isReadyToUpload == 1
				ORDER BY fileMaybeOnOfflineVolume ASC, isDenied ASC,
						priorityForUpload DESC, importDate ASC, captureDate ASC,
						needsBinaryUpload.docId
				LIMIT 100]], typeOfJoin )

	local binaryUploadCandidatesStream = self._catalog:indexStore():query {
			sql = needsUploadQueryStr,
			quiesceUntilUpToDate = true,
			calm = binaryUploadQueryCalmValue,
		}

	return maybePerformanceWrap( 'query', binaryUploadCandidatesStream )

end

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

local function getOwnerSharedBinaryUploadCandidatesStream( self )

	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

	return 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 ownerSharedBinaryUploadCandidatesQueryStr = string.format( 
				[[SELECT needsBinaryUpload.docId, localFilePath, uploadUrl, binaryType, 
						contentType, uploadHeaders, isDenied, sha256, fileSize, 
						sha256Error, sourceLinkForCopy, localVolumeGuid, localRelativePath,
						priorityForUpload, spaceId, mainPartsRequestUrl
					FROM needsBinaryUpload
					INNER JOIN %s ON %s.assetId == needsBinaryUpload.docId
					WHERE isReadyToUpload == 1
						AND albumId IN ( %s )
					ORDER BY fileMaybeOnOfflineVolume ASC, isDenied ASC,
							priorityForUpload DESC, importDate ASC, captureDate ASC,
							needsBinaryUpload.docId
					LIMIT 100]], 
				albumAssetTableName, albumAssetTableName, 
				WFSQLiteUtils.createVariableArgList( #sharedAlbumIds ) )
			
			local queryParams = sharedAlbumIds
			queryParams.sql = ownerSharedBinaryUploadCandidatesQueryStr
			queryParams.quiesceUntilUpToDate = true
			queryParams.calm = binaryUploadQueryCalmValue
			
			local ownerSharedBinaryUploadCandidatesStream = self._catalog:indexStore():query( queryParams )
			
			ownerSharedBinaryUploadCandidatesStream = maybePerformanceWrap( 'query', ownerSharedBinaryUploadCandidatesStream )
		
			return ownerSharedBinaryUploadCandidatesStream	
		
		end )

end

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

local function maybeCancelScheduledAndWaitingBinaryUpload( self )

	if self._scheduledAndWaitingBinaryUpload then
	
		local docId = self._scheduledAndWaitingBinaryUpload.docId
		local binaryType = self._scheduledAndWaitingBinaryUpload.binaryType
		local subscription = self._scheduledAndWaitingBinaryUpload.subscription
		
		logger:infof( "Now cancelling binary upload because it's no longer the top priority %s:%s", docId, binaryType )
		
		clearScheduledAndWaitingBinaryUpload( self )
		
		if subscription then
			subscription:dispose()
		end

		clearFromRecentlyUploaded( self, "higher priority upload rescheduled", docId, binaryType, true )

		removeFromScheduledUploads( self, docId, binaryType )
	
	end

end 

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

local function getHighestPriorityBinaryToUpload( self, 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

		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.binaryType == 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 wasRecentlyScheduled = not not getRecentlyUploadedSha256( self, docId, binaryType )
		
		if not self._deletedDocIdsCache[ docId ] 
				and not wasRecentlyScheduled then
					
			local localFilePath = uploadCandidate[ 2 ]
			local binaryUrl = uploadCandidate[ 3 ]
			local contentType = uploadCandidate[ 5 ]
			local isDenied = intToBool( uploadCandidate[ 7 ] )
			local sha256 = uploadCandidate[ 8 ]
			local sourceLinkForCopy = uploadCandidate[ 11 ]
			local localFileVolumeGuid = uploadCandidate[ 12 ]
			local localFileRelativePath = uploadCandidate[ 13 ]
			local uploadPriority = uploadCandidate[ 14 ]
			
			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

			local binaryFullPath, assetIsOffline = getBinaryFullPathAndAssetOfflineStatus( self, localFileVolumeGuid, localFileRelativePath, localFilePath, binaryType )

			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
				
				maybeCancelScheduledAndWaitingBinaryUpload( self )
				
				return uploadCandidate
				
			elseif not ( binaryExistsLocally or sourceLinkForCopy ) then
				
				if binaryType == 'xmp_sidecar' 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.

					addToRecentlyUploaded( self, docId, binaryType, sha256 or true )

					removeSidecarFileAnnotation( self, docId, binaryType )
						:finally( function()
							delayedClearFromRecentlyUploaded( self, "removing sidecar annotation", docId, binaryType )
						end )
						:call( runUntilDone )

			elseif ( binaryType == 'master' or binaryType == 'watermark_original' ) then
				
					if WFPlatformInfo.isMobile() then
					
						handleMasterNotExistsLocally( self, docId, binaryUrl, binaryType, 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.", binaryType, docId, localFilePath )

						if isMaster( binaryType ) and not assetIsOffline then

							applyInvalidOriginalOverlay( self, docId, localFilePath, binaryUrl, 'UploadFailedMissingOriginal' )
								:finally( function()
									delayedClearFromRecentlyUploaded( self, "binary missing, marked invalid if volume not offline", docId, binaryType )
								end )
								:call( runUntilDone )
						end

					end
				
				end
				
			end
		
		end
		
	end

end

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

local function getHighestPriorityBinaryToUploadStream( self )

	-- 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(
			getOwnerSharedBinaryUploadCandidatesStream( self ),
			getStandardBinaryUploadCandidatesStream( self ),
			self._allowBackgroundUpload:distinctUntilChanged(),
			self._uploadFinishedStream,
			self._uploadStatusChangeStream,
			self._volumeStatusHasChangedStream )
	
		:flatMapLatest( function( ownerSharedUploadCandidates, standardUploadCandidates, allowBackgroundUpload )

			return self._continuousUploadTask:throwIfNotRunning( ownerSharedUploadCandidates, standardUploadCandidates, allowBackgroundUpload )

		end )
	
		:map( function( ownerSharedUploadCandidates, standardUploadCandidates, allowBackgroundUpload )
		
			if previousBackgroundUploadStatus and not allowBackgroundUpload then
				disposeAllActiveUploads( self )
			elseif not previousBackgroundUploadStatus and allowBackgroundUpload then
				maybeCancelScheduledAndWaitingBinaryUpload( self )
			end
			
			previousBackgroundUploadStatus = allowBackgroundUpload
			
			logger:debugf( "ownerSharedUploadCandidates raw query result = %s", #ownerSharedUploadCandidates )
			logger:debugf( "standardUploadCandidates raw query result = %s", #standardUploadCandidates )
			
			if #ownerSharedUploadCandidates == 0 and #standardUploadCandidates == 0 then
			
				sendFinalDetailedProgressUpdate( self )
				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 = getHighestPriorityBinaryToUpload( self, ownerSharedUploadCandidates )

			if uploadCandidate == nil then
			
				-- 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.
				
				uploadCandidate = getHighestPriorityBinaryToUpload( self, standardUploadCandidates )
			end

			return uploadCandidate
		
		end )
		
		:filter( function( uploadCandidate )
		
			local hasUploadCandidate = not not uploadCandidate 
			local shouldScheduleUpload = not self._scheduledAndWaitingBinaryUpload
			
			return hasUploadCandidate and shouldScheduleUpload
			
		end )

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


local function uploadSomeBinaries( self )
	
	return getHighestPriorityBinaryToUploadStream( self )

		: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 ]
			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 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 = getUploadingAssetsWithSha256( self, sha256 ) == 0

			addToRecentlyUploaded( self, docId, binaryType, sha256 or true )

			local subscription = SingleAssignmentDisposable()

			local isCopy = not not sourceLinkForCopy
			
			addToScheduledUploads( self, docId, binaryType, subscription, localFileSize, uploadPriority, isCopy )
			
			local isDirectUpload = false
			local copyBinaryUrl = binaryUrl
			
			if self._useDirectUploadIfPossible and mainPartsRequestUrl and checkDirectUploadFailureCount( self, docId, binaryType ) then
				isDirectUpload = true
				binaryUrl = mainPartsRequestUrl
			end
			
			trackUploadForDiagnostics( self, spaceId, binaryType, 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.
				removeFromScheduledUploads( self, docId, binaryType )

			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.
				removeFromScheduledUploads( self, docId, binaryType )

				-- Trigger combineLatest above to start other uploads.
				uploadFinished( self )

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

			end

			local action

			if sha256Error then

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

				-- Handle asynchronous error from a prior chunked upload that is sent via catalog metadata.
				action = resolveSha256ErrorInInputSource( self, 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 = copyOneBinary( self, docId, copyBinaryUrl, binaryType, sha256, contentType, uploadHeaders, sourceLinkForCopy )
				
					:catch( function( responseData, headers )

						local resourceDoesNotExist, newId = errorAndMaybeNewIdIfMasterOrProxyResourceDoesNotExist( binaryType, responseData, headers )
						
						if resourceDoesNotExist then

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

						end

						handleUploadFailure( self, docId, binaryType, copyBinaryUrl, responseData, headers, false, contentType, localFileSize, true, false )
						
						return Observable.throw( responseData, headers )
						
					end )

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

			else

				action = uploadOneBinary( self, docId, localFilePath, localFileSize, binaryUrl, binaryType, contentType, uploadHeaders, sha256, uploadPriority, isDirectUpload )

			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

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

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.

	_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
	
	_useDirectUploadIfPossible = false,
		-- (boolean) Default to false. Will default to true for non-unit tests if direct uplaod 
		-- feature is ready to release.
		
	_enableVideoProxies = false,
		-- (boolean) Default to false. When set to true, will add upload request header to request proxies for video

	_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.

	class = {
	
		generateIndexSchema = generateIndexSchema,
		
		_DEBUG_sanitizeDirectUploadUrlForAnalytics = function( _, url )
			return sanitizeDirectUploadUrlForAnalytics( url )
		end
		
	},

	WFCloseSupport.addCloseSupportMethods,

}

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

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._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._useDirectUploadIfPossible = optUploaderOptions.useDirectUploadIfPossible or false

	self._enableVideoProxies = optUploaderOptions.enableVideoProxies or false

	self._directUploadConsecutiveFailureCounter = 0
	
	self._directUploadFallBackToOzUploadTimerSub = nil
	
	self._directUploadFallBackToOzUploadWindowInSecond = optUploaderOptions.directUploadFallBackToOzUploadWindowInSecond or kDirectUploadFallBackToOzUploadWindowInSecond
	
	self._recentlyUploaded = {}

	self._scheduledUploadSubscriptions = {}

	self._numberOfScheduledBinaryUploads = 0

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

	self._onlyUploadSpaceAssets = false

	local continuousTaskDelays = optTaskDelays or {
		progressiveBackoff = true,
		errorDelay = 10,
		maxDelay = 60,
	}
	
	local continuousTask = function() 
		return uploadSomeBinaries( self ) 
	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._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
				
				disposeOfUploadForDeletedDocId( self, 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:close()

	check( 'WFCatBinaryUploader' )

	disposeAllActiveUploads( self )

	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

end

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

function WFCatBinaryUploader:stopContinuousUploads()

	self._continuousUploadTask:stopContinuousTask()

	disposeAllActiveUploads( self )

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

end

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

function WFCatBinaryUploader:startContinuousUploads()

	if self._onlyUploadSpaceAssets then
		self:stopContinuousUploads()
	end

	self._onlyUploadSpaceAssets = false

	self._continuousUploadTask:startContinuousTask()

end

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

function WFCatBinaryUploader:startContinuousUploadOfOnlySpaceAssets()

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

	self._onlyUploadSpaceAssets = true

	self._continuousUploadTask:startContinuousTask()

end

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

function WFCatBinaryUploader:enableBackgroundUpload( enable )

	self._allowBackgroundUpload:onNext( enable )

end

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

function WFCatBinaryUploader:errorStatus()

	return self._errorStatusStream

end

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

function WFCatBinaryUploader:isDoingWork()

	return self._continuousUploadTask:isTaskDoingWork()

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 guarentees 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

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

function WFCatBinaryUploader:DEBUG_getDiagnosticLogErrors()
	return logger.DEBUG_makeDiagnosticLogErrorsHtmlText()
end

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

function WFCatBinaryUploader:DEBUG_resetDiagnosticLogErrors()
	return logger.DEBUG_resetDiagnosticLogErrors()
end

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

function WFCatBinaryUploader:DEBUG_getInFlightUploadProgressUpdateStream()
	return self._debugInFlightUploadProgressUpdateStream
end

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

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( true ),
			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, binaryTypes in pairs( self._scheduledUploadSubscriptions ) do
			
				inFlightUploads[ docId ] = {}
			
				for binaryType, _ in pairs( binaryTypes ) do
				
					inFlightUploads[ docId ][ binaryType ] = true
				
				end
			
			end
		
			local scheduledAndWaitingBinaryUpload
			
			if self._scheduledAndWaitingBinaryUpload then
			
				scheduledAndWaitingBinaryUpload = {
					docId = self._scheduledAndWaitingBinaryUpload.docId,
					binaryType = self._scheduledAndWaitingBinaryUpload.binaryType,
					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

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

local function makeBinaryUploaderHTML( self, needsBinaryUploadInfo )

	local scheduledUploads = {}
	
	for docId, uploadSubscriptionsForDocId in pairs( self._scheduledUploadSubscriptions ) do
		for binaryType, _ in pairs( uploadSubscriptionsForDocId ) do
			table.insert( scheduledUploads, { id = docId, binaryType = binaryType } )
		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

	return {
			scheduledUploads = scheduledUploads,
			numberOfScheduledBinaryUploads = self._numberOfScheduledBinaryUploads,
			needsBinaryUpload = needsBinaryUploadInfo,
			recentErrors = self._uploadErrorLog,
			uploadHttpRequestCost = uploadHttpRequestCost,
			uploadSpeedStats = finalStats,
			totalBinarySize = self._uploadSpeedStats.totalBinarySize,
			chunkFileAtSize = self._chunkedUploader:chunkFileSize(),
			chunkSize = self._chunkedUploader:chunkSize()
		}
end

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

function WFCatBinaryUploader:sendDiagnosticLogData()

	self._catalog:indexStore():queryOnce {
		sql = "SELECT docId, binaryType, captureDate, fileSize, docType, isReadyToUpload FROM needsBinaryUpload"
	}
	: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._catalog: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 )
		:map( function( needsBinaryUploadInfo )
		
			WFDiagnosticLog.addSectionToLog
				{
					name = "Binary Uploader",
					id = 'WFCatBinaryUploader',
					weight = 510,
					parentSection = WFDiagnosticLog.Sections.SYNC,
					body = makeBinaryUploaderHTML( self, needsBinaryUploadInfo )
				}
				
		end )
		:call( WFRxRunUntilDone )


	return {}

end

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

function WFCatBinaryUploader:DEBUG_uploadCount( bucket )

	local bucketStats = self._uploadSpeedStats.buckets[ bucket ]

	return derefOpt( bucketStats, 'count' ) or 0

end

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

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

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

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 key = string.format( "%s::%s", docId, binaryType )

		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 binaryType == 'xmp_sidecar' 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 )

		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 ) )

	return { currentSizeMeasurement, ozUploadSpeed, #rows }, previousRemainingBinaryUploads

end

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

-- Exposed for testing:

WFCatBinaryUploader.class._TEST_calculateUploadSpeed = function( self, ... )
	return calculateUploadSpeed( ... )
end

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

local function getSizeOfRemainingUploadsStream( self )

	-- 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
		-- 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
						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

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

local function getSizeOfRemainingCopiesStream( self )

	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
							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 ]
				
				if not fileSize then
				
					if binaryType == 'xmp_sidecar' 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 )
				
				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

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

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

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

-- Exposed for testing:

WFCatBinaryUploader.class._TEST_getEstimatedTimeRemainingImp = function( self, ... )
	return _getEstimatedTimeRemainingImp( ... )
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(
			getSizeOfRemainingUploadsStream( self ),
			getSizeOfRemainingCopiesStream( self ) )

		:filter( function( estimatedTimeRemaining, averageUploadSpeed )

			return estimatedTimeRemaining ~= 0 and estimatedTimeRemaining ~= math.huge
	
		end )

		:calm( kCalmValue, WFRxScheduler.sharedScheduler() )

end

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

return Rx.Class 'WFCatBinaryUploader'( WFCatBinaryUploader )
