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

LrClassicDevelopSettings.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2023 Adobe
All Rights Reserved.

NOTICE: All information contained herein is, and remains
the property of Adobe and its suppliers, if any. The intellectual
and technical concepts contained herein are proprietary to Adobe
and its suppliers and are protected by all applicable intellectual
property laws, including trade secret and copyright laws.
Dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from Adobe.

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

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

	-- WF.core.os
local KSData = require 'KSData'
local KSDate = require 'KSDate'
local WFArguments = require 'WFArguments'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFDeviceInfo = require 'WFDeviceInfo'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFTableMerge = require 'WFTableMerge'
local WFCapabilitiesCache = require "WFCapabilitiesCache"
local WFCrypto = require 'WFCrypto'
local WFAsyncFileUtils = require 'WFAsyncFileUtils'
local WFConfiguration = require 'WFConfiguration'
local WFLogger = require 'WFLogger'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFFileUtils = require 'WFFileUtils'
local WFImageOrientation = require 'WFImageOrientation'

	-- WF.oz_client.managed_catalog
local WFCatAsset = require 'WFCatAsset'
local WFCatAssetAux = require 'WFCatAssetAux'
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatUtils = require 'WFCatUtils'
local WFCatIndexUtils = require 'WFCatIndexUtils'
local WFCapabilities = require 'WFCapabilities'
local WFCatAssetUtils = require 'WFCatAssetUtils'
local WFCatDelete = require 'WFCatDelete'
local WFCatVolumeAgentQueryHelper = require 'WFCatVolumeAgentQueryHelper'

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

local INLINE_XMP_CAMERA_RAW_MAX_LENGTH = WFConfiguration.LrClassic.inlineXmpMaxLength
										or 10 * 1024 --10KB

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

local logger = WFLogger( 'LrClassicDevelopSettings' )

local derefOpt = assert( KSTableUtils.derefOpt )

local KSDate_timeInIso8601FormatNearestSecond = assert( KSDate.timeInIso8601FormatNearestSecond, "KSDate.timeInIso8601FormatNearestSecond does not exist" )

local runUntilDone = WFRxRunUntilDone

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

local WFCatAssetAux_externalAuxModule_getClientAux

local function getExternalClientAux()

	if not WFCatAssetAux_externalAuxModule_getClientAux then

		local externalAuxModule = WFCapabilities.getCapability( "WFCatAssetAux.externalAuxModule" )

		if externalAuxModule then
	
			local WFCatAssetAux_externalAuxModule = require( externalAuxModule )
	
			WFCatAssetAux_externalAuxModule_getClientAux = WFCatAssetAux_externalAuxModule.getClientAux
	
		end
	
	end
	
	return WFCatAssetAux_externalAuxModule_getClientAux

end


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

local function createKey( pixStreamId, assetId )

	return string.format( "%s:%d", pixStreamId, assetId )

end

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

local observersWaitingForDevelopSettings = {}

local retainedDevSettingsPathObservers = {}

local xmpDevelopSettingsPathStreamSubscription

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

local _currentSaveSequencesByAssetId = {}

local _pendingDevelopSettingsSaveListeners = {}

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

local function bootstrapXMPDevelopSettingsPathStreamSubscription( pixelProviderProcessor, pixStreamIdToMatch )

	if not xmpDevelopSettingsPathStreamSubscription then
	
		xmpDevelopSettingsPathStreamSubscription = 
			pixelProviderProcessor
				:pixStreamIdPixelPathStream( pixStreamIdToMatch )
				:subscribe( function( pixStreamId, assetId, success, xmpDevelopPath )
				
					local key = createKey( pixStreamId, assetId )
					
					local observers = observersWaitingForDevelopSettings[ key ] or {}
					
					for _, observer in pairs( observers ) do
					
						observer:onNext( success, xmpDevelopPath )
					
					end
				
					observersWaitingForDevelopSettings[ key ] = nil
				
					pixelProviderProcessor:runUntilDone( 'removeRequest',
															pixStreamId, assetId )
				
				end )

	end

end

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

local function getXmpDevelopPath( pixelProviderProcessor, imageId, needsASAP )

	-- TODO we should have at most two pixelProvider requests per assetId and use separate pixStreamIds for asap vs normal

	local pixStreamId = 'LrClassicDevelopSettings'
	
	return Rx.Observable.create( 

		function( observer )

			bootstrapXMPDevelopSettingsPathStreamSubscription( 
				pixelProviderProcessor, pixStreamId )

			local key = createKey( pixStreamId, imageId )

			local observers = observersWaitingForDevelopSettings[ key ] or KSTableUtils.makeFullyWeak()
			
			observers[ observer ] = observer
			
			observersWaitingForDevelopSettings[ key ] = observers
			
			retainedDevSettingsPathObservers[ observer ] = observer
			
			pixelProviderProcessor:runUntilDone( 
									"requestXmpDevelopForDevelop", 
										pixStreamId,
										imageId,
										needsASAP )

			return Rx.Disposable.create( function() 
			
				retainedDevSettingsPathObservers[ observer ] = nil
			
			end )

		end )

end

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

local function getExternalFileXMPSettingsStream( 
				pixelProviderProcessor, imageId, docValue, 
				docValueXmpDevelopSha256, baseSettings, needsASAP )

	local hasLocalAuxError = WFCatAssetAux.hasLocalAuxErrorForCurrentDevelopSettings( docValue )

	local xmpPathObservable = 
			getXmpDevelopPath( pixelProviderProcessor, imageId, needsASAP )
			:flatMap( 
				function( success, xmpDevelopPath ) 
					
					if not success or not xmpDevelopPath then
						logger:warnf( "Failed to fetch xmp_develop settings for asset %s: %s", imageId, xmpDevelopPath )
						
						baseSettings.settingsXmpStatus = 'failed'
						
						return Observable.just( baseSettings )
					end
					
					return WFAsyncFileUtils.readFile( xmpDevelopPath )
							:map( function( fileData ) 
								
								if fileData then

									baseSettings.settingsXmp = tostring( fileData )

									-- Read the xmp even if we have an invalid mask in case we add the ability
									-- to edit anyways similar to future settings.  Currently WF doesn't have
									-- the ability to recover from an invalid mask without resetting the settings.
									baseSettings.settingsXmpStatus = hasLocalAuxError and 'failed' or 'complete'

									fileData:close()

								else

									logger:warnf( "failed to read xmp from external develop settings file: %s",
													xmpDevelopPath )

									baseSettings.settingsXmpStatus = 'failed'

								end
								
								return baseSettings

							end )
					
				end )
		
	local localXmpDevelopSha256 = 
			derefOpt( WFCatFileSpec.getXmpDevelopInfoFromDocValue( docValue ),
						'sha256' )

	if hasLocalAuxError or localXmpDevelopSha256 ~= docValueXmpDevelopSha256 then

		-- if we don't think we have this locally yet, mark it pending as it
		-- may take a bit and clients might need to know that we're fetching
		-- the settings
		baseSettings.settingsXmpStatus = 'pending'
		
		xmpPathObservable = xmpPathObservable:startWith( baseSettings )
		
	end
	
	return xmpPathObservable

end

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

local function getValidatedVideoSettingsStream( docValue, baseSettings )

	local developSettings = derefOpt( docValue, 'develop', 'adjustments' )
		
	if developSettings ~= nil and type( developSettings ) ~= 'table' then

		logger:warnf( "bad docValue.develop.adjustments value in docstore: adjustments value was not a table" )
	
		baseSettings.settingsXmpStatus = 'failed'
		
		return Observable.just( baseSettings )

	end

	baseSettings.adjustments = developSettings

	baseSettings.settingsXmpStatus = 'complete' 

	return Observable.just( baseSettings )

end

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

local function getValidatedImageSettingsStream( 
				pixelProviderProcessor, imageId, docValue, baseSettings,
				needsASAP )

	local docValueXmpDevelopSha256 = 
			derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )
		
	if docValueXmpDevelopSha256 then
		
		return getExternalFileXMPSettingsStream( 
				pixelProviderProcessor, imageId, docValue,
				docValueXmpDevelopSha256, baseSettings, needsASAP )

	end

	local developSettingsAsStr = 
			derefOpt( docValue, 'develop', 'xmpCameraRaw' )
	
	-- Check if develop is correct format for image
	if developSettingsAsStr ~= nil and 
			type( developSettingsAsStr ) ~= 'string' then

		local badSettingsReason =
					type( developSettingsAsStr ) ~= "table" and
						string.format( "bad type: %s", type( developSettingsAsStr ) ) or
						"xmp was a table without a sha256"

		logger:warnf( "bad docValue.develop.xmpCameraRaw value in docstore: %s",
						badSettingsReason )
	
		baseSettings.settingsXmpStatus = 'failed'
		
		return Observable.just( baseSettings )

	end
	
	baseSettings.settingsXmp = developSettingsAsStr

	baseSettings.settingsXmpStatus = 'complete'

	return Observable.just( baseSettings )

end

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

local function getValidatedSettingsStream( pixelProviderProcessor, imageId,
											docValue, baseSettings, needsASAP )
												
	if WFCatIndexUtils.isVideoDocument( docValue ) then

		return getValidatedVideoSettingsStream( docValue, baseSettings )

	end

	return getValidatedImageSettingsStream( 
			pixelProviderProcessor, imageId, docValue, baseSettings,
			needsASAP )

end

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

local function docValueHasLocalOriginal( docValue )

	local localOriginalVolumeGuid = WFCatFileSpec.getOriginalVolumeGuidFromDocValue( docValue )
	local localOriginalRelativePath = WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue )
	return localOriginalVolumeGuid and localOriginalRelativePath

end

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

local function developContentStreamImp( wfCatalog, pixelProviderProcessor, docValue, needsASAP )
	
	if docValue._empty or docValue.type ~= 'asset' then

		logger:errorf( "Asset does not exist, or has wrong type: id = %s",
								docValue._localId )

		return Observable.empty()

	end
	
	local imageId = docValue._localId

	local originalPath = WFCatFileSpec.buildOriginalPathFromDocValue( docValue )
	
	if originalPath and not WFFileUtils.exists( originalPath ) then
		originalPath = nil
	end
	
	local proxyPath = WFCatFileSpec.buildProxyPathFromDocValue( docValue )
	
	if proxyPath and not WFFileUtils.exists( proxyPath ) then
		proxyPath = nil
	end
	
	local isUnsupportedFormat = WFCatUtils.isUnsupportedFormat( docValue )
	
	-- 2017.07.28 bkruse
	-- 
	-- Nimbus has one asset id per file semantics, use the asset's id as the
	-- fileUUId. This might change in the future, but for now, it's true and
	-- filUUId is used to cache negative paths so we don't want to
	-- assets to have the same fileUUId unless the files are indeed unique.
	-- 
	-- It is possible via catalog migration to have two files with the same
	-- sha256, so we can't rely on the sha256 as unique fileUUId
	local fileUuid = docValue._id -- derefOpt( docValue, 'importSource', 'sha256' ) or docValue._id

	local disallowLocalEdits = false

	local isVideo = derefOpt( docValue, 'subtype') == 'video'
	
	local hasCopiedAssetSourceID = derefOpt( docValue, 'copy', 'copy_id' ) and derefOpt( docValue, '_localOnly', 'sourceAssetId' )
	local isCopiedAsset = hasCopiedAssetSourceID and true or false

	local disallowClientAccessToProxyAndOrig =
			isUnsupportedFormat or
				( disallowLocalEdits )

	-- rootFile is is a vestige of LrClassic's libImage object interface that
	-- the LrD client code still depends on.
	local rootFile

	local haveLocalProxyOrOrig = originalPath or proxyPath 

	if haveLocalProxyOrOrig and not disallowClientAccessToProxyAndOrig then

		rootFile = 
			{
				filePath = originalPath or proxyPath,
				originalPath = originalPath,
				proxyPath = proxyPath,
			}

	end
	
	local contentType = derefOpt( docValue, 'importSource', 'contentType' )
	local fileFormat
	
	if contentType == 'image/jpeg' then
		fileFormat = "JPG"
	elseif isVideo then
		fileFormat = "VIDEO"
	else
		fileFormat = "RAW"
	end
	
	local tiffOrientation = 1
	
	local temporalAdjustments
	
	local videoDevelopVersion
	
	if isVideo then
		
		local vfOrientation = derefOpt( docValue, 'develop', 'adjustments', 'geometric', 'orientation' ) or 'normal'
		tiffOrientation = WFImageOrientation.vfOrientationToTiffOrientation( vfOrientation )
		
		-- Check if we also have temporal data in the adjustments
		temporalAdjustments = derefOpt( docValue, 'develop', 'temporal' )
		
		-- set the VF version for the video asset
		videoDevelopVersion =
			{
				version = derefOpt( docValue, 'develop', 'version' ),
				processVersion = derefOpt( docValue, 'develop', 'processVersion' )
			}
		
	else
		tiffOrientation = derefOpt( docValue, 'develop', 'userOrientation' ) or 1
	end
	
	local baseSettings = {

			changeCounter = tonumber( docValue._changeCounter ),

			fileUuid = fileUuid,

			historySettingsID = KSIdGenerator.uuid(),
	
			fileName = derefOpt( docValue, 'importSource', 'fileName' ),
			fileFormat = fileFormat,
	
			payloadHeight = derefOpt( docValue, 'importSource', 'originalHeight' ),
			payloadWidth = derefOpt( docValue, 'importSource', 'originalWidth' ),
			noOriginalOnOz = derefOpt( docValue, 'importSource', 'willNotUploadMaster' ),
			id_local = docValue._localId,
			isVideo = isVideo,
			numUnlockedSnapshots = 0,
			numVirtualCopies = 0,
	
			tiffOrientation = tiffOrientation,
			
			temporal = temporalAdjustments,
			videoDevelopVersion = videoDevelopVersion,
			
			defaultCrop = derefOpt( docValue, 'importSource', 'originalCrop' ),
			
			rootFile = rootFile,
			
			settingsID = "default",
			
			isCopiedAsset = isCopiedAsset,
			
			virtualCopy = false,
			virtualCopyMap = {},
			
			-- used to show the downloading error screens.
			hasProxyLink = not disallowClientAccessToProxyAndOrig and WFCatFileSpec.getProxyLinkFromDocValue( docValue ) ~= nil,
			hasOriginalLink = not disallowClientAccessToProxyAndOrig and WFCatFileSpec.getOriginalLinkFromDocValue( docValue ) ~= nil,
			
			hasLocalOriginal = docValueHasLocalOriginal( docValue ),
			
			isUnsupportedFormat = isUnsupportedFormat,

			disallowLocalEdits = disallowLocalEdits,
			
		}
		
	return getValidatedSettingsStream( 
			pixelProviderProcessor, imageId, docValue, baseSettings,
			needsASAP )
		:flatMapLatest( function( validatedBaseSettings )

			if validatedBaseSettings.settingsXmpStatus == 'complete' then
				return Observable.just( validatedBaseSettings )
			end

			return WFCatAsset.getAssetLatestEditDetails( wfCatalog, imageId )
				:map( function( latestEditDetails )

					latestEditDetails.thisDevice = WFDeviceInfo.deviceDescription()
					latestEditDetails.hasInvalidMask = WFCatAssetAux.hasLocalAuxErrorForCurrentDevelopSettings( docValue )

					validatedBaseSettings.settingsXmpStatusDetails = latestEditDetails
					return validatedBaseSettings

				end )
		end )

end

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

local function docIsNonEmpty( docValue )

	if not docValue._empty then

		return docValue

	end

end

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

local function nonEmptyDocValues( docValues )

	return WFFunctionUtils.filter( docIsNonEmpty, docValues )

end

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

local function buildOutAssetIdTables( assetIdTables, assetIdChangeCounterTuple )

	if type( assetIdChangeCounterTuple ) ~= "table" then

		return assetIdTables

	end

	local assetId, maybeChangeCounter = unpack( assetIdChangeCounterTuple )

	if not assetId then

		return assetIdTables

	end

	assetIdTables.assetIds[ #assetIdTables.assetIds + 1 ] = assetId

	assetIdTables.assetIdSet[ assetId ] = true

	if maybeChangeCounter then

		assetIdTables.assetIdToChangeCounterSet[ assetId ] = 
			maybeChangeCounter

	end

	return assetIdTables

end

--------------------------------------------------------------------------------
-- isUpdateWanted is called from 
-- 	docStore:scannedFilteredMappedChangesFeed( ... )
-- 	
-- 	it's used to scan *all* ongoing changes to the doc store and wantUpdatesSet
-- 	is used to filter out only those docValues that we are interested in.
-- 	
-- it should always return wantUpdatesSet unchanged. wantUpdatesSet will always
-- hold the set of assetIds for which we want updateds
-- 
-- it should return the docValue as a second value if the docValue is of interest
--------------------------------------------------------------------------------

local function isUpdateWanted( wantUpdatesSet, docValue )

	if docValue._empty or docValue._deleted then

		return wantUpdatesSet

	end

	local localAssetId = docValue._localId

	if not localAssetId then

		return wantUpdatesSet

	end

	if not wantUpdatesSet[ localAssetId ] then

		logger:infof( "%s is ignoring update of %s, it's not watched", 
						wantUpdatesSet.desc, localAssetId )

		return wantUpdatesSet

	end

	return wantUpdatesSet, docValue

end

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

local function addToBatch( batches, assetId, batchSize )

	local currBatchSize = #( batches[ #batches ] )

	if currBatchSize == batchSize then

		currBatchSize = 0

		batches[ #batches + 1 ] = {}

	end

	batches[ #batches ][ currBatchSize + 1 ] = assetId

	return batches

end

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

local function batchedDocValuesStream( _, wfCatalog, assetIds, batchSize )

	local batchedAssetIds = 
			WFFunctionUtils.reduce( { {} }, addToBatch, assetIds, batchSize )

	return Rx.Observable.from( batchedAssetIds )
			:map( 
				function( assetIdsBatch )

					return wfCatalog
							:docStore()
							:bulkDocValue( assetIdsBatch )
							:filter( nonEmptyDocValues )
							:flatMapLatest( Rx.Observable.from )

				end )
			-- flatMerge( 1 ) calls bulkDocValue only after previous completes
			:flatMerge( 1 ) 
					
end

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

local _batchSize = 5

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

local function getDocValuesPlusUpdatesStream( 
				desc, wfCatalog, assetIds, assetIdSet, changeCounterAtStart )

	local initialDocValuesStream =
			batchedDocValuesStream( 
				desc, wfCatalog, assetIds, _batchSize )

	local updatesSinceStartStream =
				wfCatalog
					:docStore()
					:scannedFilteredMappedChangesFeedSince( 
						"LrClassicDevelopSettings", "isUpdateWanted", 
						assetIdSet, changeCounterAtStart )
	
	return Rx.Observable.concat( 
			initialDocValuesStream, updatesSinceStartStream )

end

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

local function makeCalmedDocValueStream( docValueStream, updatesPerSecond )

	-- Note: if we denote the unique per assetId docValues as docValue[A-Z] and 
	-- further distinguish versions of docValue[A-Z] with a subscript, this
	-- will emit a stream of
	--
	--   docValueAi, docValueBi, docValueAj, docValueCi, docValueBj, ...
	--
	-- It is guaranteed the docValueAj is more recent than docValueAi but not
	-- that docValueAi is more recent thant docValueBi. i.e. most recent values
	-- are only guaranteed per assetId docValue. Updates to two different assetId
	-- docValues are not guaranteed to emitted in the order they occurred.

	assert( tonumber( updatesPerSecond ) or 0 > 0, 
			"updatesPerSecond must be positive number" )

	local pauseIntervalInSeconds = 1 / updatesPerSecond

	local mostRecentDocValuePerAssetId = {}

	return docValueStream
			:map(
				function( docValue )

					local assetId = 
						tonumber( derefOpt( docValue, '_localId' ) )

					if not assetId then

						return

					end

					-- coalesce updates by assetId
					mostRecentDocValuePerAssetId[ assetId ] = docValue

				end )
			-- limit how often we deliver coalesced updates
			:calm( pauseIntervalInSeconds, WFRxScheduler.sharedScheduler() ) 
			:flatMapLatest(
				function()
				
					-- "unpack" the coalesced values and deliver one by one
					--
					local mostRecentDocValues = 
							KSTableUtils.extractValues( 
								mostRecentDocValuePerAssetId )

					-- reset the buffer
					mostRecentDocValuePerAssetId = {}

					-- deliver the latest values one by one
					return Rx.Observable.from( mostRecentDocValues )

				end )


end

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

local function makeUnorderedDevelopContentsStream( 
					desc, wfCatalog, pixelProviderProcessor,
					assetIdChangeCounterTuples, needsASAP )

	local assetIdTables =
				WFFunctionUtils.reduce( 
					{
						assetIds = {},
						assetIdSet = {},
						assetIdToChangeCounterSet = {},
					},
					buildOutAssetIdTables,
						assetIdChangeCounterTuples )

	local assetIds = assetIdTables.assetIds

	if #assetIds == 0  then

		return Rx.Observable.empty()

	end

	local assetIdSet = assetIdTables.assetIdSet
	
	local docValueStream = 
			wfCatalog
				:docStore()
				:changeCounter()
				:take( 1 )
				:flatMapLatest(
					function( changeCounterAtStart )

						-- To guard against race conditions, we need to
						-- establish a "start" doc store change counter. 
						-- 
						-- This guards against this race condition:
						-- 
						-- worker: 
						-- 		changeCounter is at n
						-- this thread: 
						-- 		subscribe initial doc value
						-- worker: 
						-- 		send initial docValue for assetId
						-- worker: 
						-- 		receive update for assetId
						-- worker: 
						-- 		changeCounter is at n + 1
						-- this thread: 
						-- 		subscribe to scannedFilteredMappedChangesFeed
						-- worker: 
						-- 		subscribe to all changes after n + 1
						-- this thread:
						-- 		only receives updates for n + 2 and later
						-- 
						-- In the above sequence, we never see the n + 1
						-- update.  With the changeCounterAtStart the above
						-- sequence is:
						-- 
						-- worker: 
						-- 		changeCounter is at n
						-- this thread: 
						-- 		receives changeCounter with value n
						-- this thread: 
						-- 		subscribe initial doc value
						-- worker: 
						-- 		send initial docValue for assetId
						-- worker: 
						-- 		receive update for assetId
						-- worker: 
						-- 		changeCounter is at n + 1
						-- this thread: 
						-- 		subscribe to scannedFilteredMappedChangesFeedSince
						-- worker: 
						-- 		subscribe to all changes after n
						-- worker: 
						-- 		send all changes from n + 1 to current
						-- this thread: 
						-- 		receive docValue with changeCounter of n + 1

						return getDocValuesPlusUpdatesStream( 
								desc, wfCatalog, assetIds, assetIdSet,
								changeCounterAtStart )

					end )

	local assetIdToChangeCounterSet = assetIdTables.assetIdToChangeCounterSet

	-- IMPORTANT Rx note: when makeCalmedDocValueStream( ... ) is piped through flatMap,
	-- it is no longer guaranteed that downstream will receive docValue changes per
	-- assetId in the correct order. 
	return makeCalmedDocValueStream( docValueStream, 12 )
			:flatMap(
				function( docValue )

					local docValueChangeCounter = 
							tonumber( docValue._changeCounter )

					local lastSeenChangeCounterForAssetId =
							tonumber( assetIdToChangeCounterSet[ docValue._localId ] )

					if type( docValueChangeCounter ) == "number" and
							type( lastSeenChangeCounterForAssetId ) == "number" and
							docValueChangeCounter <= lastSeenChangeCounterForAssetId then

						-- we've already seen this either here or on the client, do nothing
						-- nothing
						return Rx.Observable.empty()

					end

					if type( docValueChangeCounter ) == "number" then
							
						-- update the last seen value
						assetIdToChangeCounterSet[ docValue._localId ] =
							docValueChangeCounter

					end

					-- send the docValue to the client
					return developContentStreamImp( wfCatalog,
						pixelProviderProcessor, docValue, needsASAP )

				end )

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

local function makeMostRecentDevelopContentStream( unorderedDevelopContentStream )

	local mostRecentDevelopContentInfo = {}

	return unorderedDevelopContentStream
			:filter( 
				function( developContent )

					-- First, filter out all cases where the developContent
					-- should be ignored

					local assetId = derefOpt( developContent, "id_local" )

					if not assetId then

						logger:warnf( "NOT sending developContent without assetId: %s", developContent )

						return

					end

					local thisDevelopContentChangeCounter = 
							tonumber( derefOpt( developContent, "changeCounter" ) )

					local thisSettingsXmpStatus = derefOpt( developContent, "settingsXmpStatus" )

					local mostRecentDevelopContentChangeCounter, mostRecentSettingsXmpStatus =
							unpack( mostRecentDevelopContentInfo[ assetId ] or {} )

					if not thisDevelopContentChangeCounter and 
							mostRecentDevelopContentChangeCounter then

						-- this is unexpected, warn
						logger:warnf( "NOT sending developContent for assetId %s: no changeCounter, but have sent changeCounter %s", 
										assetId, mostRecentDevelopContentChangeCounter )

						return

					end

					if thisDevelopContentChangeCounter and
							mostRecentDevelopContentChangeCounter then

						-- handle ignorable changes when we have both changeCounters

						if thisDevelopContentChangeCounter <
								mostRecentDevelopContentChangeCounter then

							-- this is expected, ignore
							logger:infof( "NOT sending developContent for assetId %s this changeCounter %s is older than most recent changeCounter %s",
											assetId,
											thisDevelopContentChangeCounter,
											mostRecentDevelopContentChangeCounter )

							return

						end

						if thisDevelopContentChangeCounter ==
									mostRecentDevelopContentChangeCounter and
								thisSettingsXmpStatus == 
									mostRecentSettingsXmpStatus then

							-- this is expected, ignore
							logger:infof( "NOT sending developContent for assetId %s for same changeCounter and same settingsXmpStatus",
											assetId )

							return

						end

						if thisDevelopContentChangeCounter ==
									mostRecentDevelopContentChangeCounter and
								( mostRecentSettingsXmpStatus == 'complete' or
									mostRecentSettingsXmpStatus == 'failed' ) then

							-- this is not expected, warn
							logger:warnf( "NOT sending developContent for assetId %s for same changeCounter because settingsXmpStatus transition %s -> %s is ignored",
											assetId,
											mostRecentSettingsXmpStatus,
											thisSettingsXmpStatus )

							return

						end

					end

					-- at this point, we assume the develop content should be sent

					mostRecentDevelopContentInfo[ assetId ] =
						thisDevelopContentChangeCounter and
							{ 
								thisDevelopContentChangeCounter, 
								thisSettingsXmpStatus 
							} or
							nil

					-- warn when we send a developContent without a changeCounter
					local logLevel = thisDevelopContentChangeCounter and
										"infof" or "warnf"

					logger[ logLevel ]( logger,"sending developContent for assetId %s with changeCounter %s and settingsXmpStatus %s (previous changeCounter %s, previous settingsXmpStatus %s)",
									assetId,
									thisDevelopContentChangeCounter,
									thisSettingsXmpStatus,
									mostRecentDevelopContentChangeCounter,
									mostRecentSettingsXmpStatus )

					return true

				end )

end

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

local function makeOngoingDevelopContentsStream(
					desc, wfCatalog, pixelProviderProcessor,
					needDevelopContentStream )

	return needDevelopContentStream
			:flatMapLatest(
				function( assetIdChangeCounterTuples, needsASAP )

					local unorderedDevelopContentStream =
						makeUnorderedDevelopContentsStream( 
							desc, wfCatalog, pixelProviderProcessor,
							assetIdChangeCounterTuples, needsASAP )

					return makeMostRecentDevelopContentStream(
							unorderedDevelopContentStream )

				end )

end

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

local function addToProtectedRingBuffer( wfCatalog, localId )
	
	local obs = Observable.just( true )
	
	if WFPlatformInfo.isClassic() then
		return obs
	end
	
	local volumeAgents = wfCatalog:getAllVolumeAgents()

	for _, volumeAgent in pairs( volumeAgents ) do
		obs:concat( volumeAgent:addToProtectedRingBuffer( localId ) )
	end

	return obs:takeLast( 1 )

end

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

local function updateFileAccessTimestampStream( 
				wfCatalog, needDevelopContentStream )

	return needDevelopContentStream
			:filter( 
				function( assetIdChangeCounterTuples )

					return 
						type( assetIdChangeCounterTuples ) == "table" and
						#assetIdChangeCounterTuples > 0 and
						type( assetIdChangeCounterTuples[ 1 ] ) == "table" and
						unpack( assetIdChangeCounterTuples[ 1 ] )

				end )
			:flatMap( 
				function( assetIdChangeCounterTuples )

					local localId = unpack( assetIdChangeCounterTuples[ 1 ] )

					return addToProtectedRingBuffer( wfCatalog, localId )
						
						:map( function()
						
							return string.format( "did update file access for %s", localId )
						
						end )

				end )
	
end

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

local function makeOngoingDevelopContentsStreamWithUpdateFileAccess(
					wfCatalog, pixelProviderProcessor,
					needDevelopContentStream )

	return Rx.Observable.create(
			function( observer )

				local updateFileAccessTimestampSubscription =
						updateFileAccessTimestampStream( 
								wfCatalog, needDevelopContentStream )
							:subscribe(
								function( msg )

									logger:info( msg )

								end )

				local devContentsStream = 
					makeOngoingDevelopContentsStream(
						"active image develop content", wfCatalog,
						pixelProviderProcessor, needDevelopContentStream )

				return Rx.CompositeDisposable(
							updateFileAccessTimestampSubscription,
							devContentsStream:subscribe( observer ) )


			end )

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

local function ongoingDevelopContentModel( wfCatalogStream )

	local needDevelopContentStream = Rx.Subject()

	return {
		
		value = wfCatalogStream
					:mapSwitch(
						function( wfCatalog, pixelProviderProcessor )

							return makeOngoingDevelopContentsStreamWithUpdateFileAccess( 
									wfCatalog, pixelProviderProcessor,
									needDevelopContentStream )

						end ),

		events = {

			setAssetId = function( _, assetId, maybeChangeCounter )

				needDevelopContentStream:onNext( { { assetId, maybeChangeCounter } }, true )

			end,

			clearAssetId = function()
				
				needDevelopContentStream:onNext( {}, true )

			end,

		}

	}
	
end
 
--------------------------------------------------------------------------------

local function developContentStream( wfCatalog, pixelProviderProcessor, 
										assetId, needsASAP )

	local docValueStream =
		wfCatalog
			:docStore()
			:docValue( assetId )
			:tap(
				function( docValue )

					logger:infof( "developContentStream got %sdoc for %s",
									not docIsNonEmpty( docValue ) and
										"empty" or "",
									assetId )
										
				end )
			:filter( docIsNonEmpty )

	-- IMPORTANT Rx note: docValueStream is "in order", but piping through flatMap
	-- does not guarantee order is preserved
	local unorderedDevelopContentStream = docValueStream
			:flatMap(
				function( docValue )

					if docValue._localId then

						return addToProtectedRingBuffer( wfCatalog, docValue._localId )
							
							:flatMap( function()
							
								return developContentStreamImp( wfCatalog,
									pixelProviderProcessor, docValue, needsASAP )
							
							end )

					end

				end )

	return makeMostRecentDevelopContentStream( unorderedDevelopContentStream )

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

local function developContentModel( wfCatalogStream, assetId )

	return {
		
		value = wfCatalogStream
					:mapSwitch(
						function( wfCatalog, pixelProviderProcessor )

							return developContentStream( 
									wfCatalog, pixelProviderProcessor, 
									assetId )

						end )

	}
	
end

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

local _developContentWithLocalNegativePathModelId = 0

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

local function settingsXmpStatusIsNotPending( baseSettings )
	
	return baseSettings.settingsXmpStatus ~= "pending"

end

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

local function baseSettingsWithLocalNegativePathStream(
				wfCatalog, pixelProviderProcessor, pixStreamId, assetId ) 

	return pixelProviderProcessor
			:pixStreamIdPixelPathStream( pixStreamId )
			:where( 
				function( _, theAssetId )

					return theAssetId == assetId

				end )
			:tap(
				function( thePixStreamId, localAssetId, havePath )

					logger:infof( "%s.%s state: %s",
									thePixStreamId, localAssetId, 
									havePath and 
										"have local path" or
										"has no local path" )

				end )
			:flatMap(
				function( _, localAssetId, havePath )

					if not havePath then

						return Rx.Observable.throw( 
								string.format( "no local negative path for: %s",
												localAssetId ) )

					end

					return developContentStream( 
							wfCatalog, pixelProviderProcessor, 
							localAssetId, 'asap' )

				end )
			:tap(
				function( baseSettings )

					logger:infof( "%s.%s got preliminary settings with settingsXmpStatus: %s",
									pixStreamId, baseSettings.id_local, 
									baseSettings.settingsXmpStatus )

				end )
			:filter( settingsXmpStatusIsNotPending )
			:tap(
				function( baseSettings )

					logger:infof( "%s.%s got final settings with settingsXmpStatus: %s",
									pixStreamId, baseSettings.id_local, 
									baseSettings.settingsXmpStatus )

				end )
			:take( 1 )

end

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

local function developContentWithLocalNegativePathModel( 
				wfCatalogStream, assetId )

	_developContentWithLocalNegativePathModelId = 
		_developContentWithLocalNegativePathModelId + 1

	local pixStreamId = 
			string.format( "developContentWithLocalNegativePathModel%s", 
							_developContentWithLocalNegativePathModelId )

	return {
		
		value = wfCatalogStream
					:mapSwitch(
						function( wfCatalog, pixelProviderProcessor )

							return Rx.Observable.create( 
								function( observer )

									-- avoid race condition subscribe to
									-- stream before we request a local
									-- negative path
									local baseSettingsWithLocalNegativePathSubscription =
										baseSettingsWithLocalNegativePathStream(
												wfCatalog,
												pixelProviderProcessor,
												pixStreamId, assetId ) 
											:subscribe( observer )

									local negPathSubscription =
										pixelProviderProcessor
											:dispatch( 
												"getLocalNegativePathASAP", 
													pixStreamId, assetId )
											:subscribeOnCompleted(

												function()

													logger:infof( "finished getLocalNegativePathASAP request for %s.%s",
																	pixStreamId, assetId )

												end )

									local disposePixelProviderRequest =
										Rx.Disposable.create(
											function()

												logger:infof( "Cleaning up request: %s",
																pixStreamId )

												pixelProviderProcessor
													:runUntilDone( 
														"abandonAllPixStreamRequests", 
															pixStreamId )

											end )

									return Rx.CompositeDisposable(
												baseSettingsWithLocalNegativePathSubscription,
												negPathSubscription,
												disposePixelProviderRequest )

								end )

						end )

	}
	
end

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

local function ongoingDevelopContentsModel( wfCatalogStream )
	
	local needDevelopContentStream = Rx.Subject()

	return {
		
		value = wfCatalogStream
					:mapSwitch(
						function( wfCatalog, pixelProviderProcessor )

							return makeOngoingDevelopContentsStream( 
									"apron develop contents", wfCatalog,
									pixelProviderProcessor,
									needDevelopContentStream )

						end ),

		events = {

			setAssetIds = function( _, assetIdChangeCounterTuples )

				WFArguments.check( "_", "table" )

				needDevelopContentStream
					:onNext( assetIdChangeCounterTuples )

			end,

		}

	}
	
end

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

local capabilitiesCache = nil

local function getCapabilitiesCache()

	if capabilitiesCache == nil then
		capabilitiesCache = WFCapabilitiesCache()
	end

	return capabilitiesCache
		-- Only used by unit tests.

end

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

local function xmpCameraRawOverlay( wfCatalog, cameraRawXmp, baseOverlay, forceExternalization, maybeCameraRawLimitOverride )

	local settingsLength = cameraRawXmp and string.len( cameraRawXmp ) or 0
	
	if maybeCameraRawLimitOverride and maybeCameraRawLimitOverride < INLINE_XMP_CAMERA_RAW_MAX_LENGTH then
		logger:infof( "Overriding max ACR settings length. Was %s, now %s", INLINE_XMP_CAMERA_RAW_MAX_LENGTH, maybeCameraRawLimitOverride )
		INLINE_XMP_CAMERA_RAW_MAX_LENGTH = maybeCameraRawLimitOverride
	end
	
	-- ensure baseOverlay meets the minimum requirements
	baseOverlay = baseOverlay or {}
	baseOverlay.develop = baseOverlay.develop or {}
	
	if settingsLength < INLINE_XMP_CAMERA_RAW_MAX_LENGTH and ( not forceExternalization ) and
			not getCapabilitiesCache():get( 'WFCatBinaryUploader.atomicDevelopSettings' ) then
		
		baseOverlay.develop.xmpCameraRaw = cameraRawXmp
		
		return Observable.just( baseOverlay )

	end
		
	return Observable.just( baseOverlay )
		:flatMap( function( overlay )
			
			-- too big to send to Oz inline, save to local file, 
			-- send sha256 of develop settings and let the upload 
			-- process attach the file to the asset record on Oz
		
			local settingsAsKSData = KSData.make( cameraRawXmp )
			
			return WFCrypto.asyncDigestData( settingsAsKSData, 'SHA256' )
				:map( function( sha256 )
					
					local pathForSettingsXmp = wfCatalog:fileSpec():pathForSettingsXmp( sha256 )

					-- set _replace = true to squelch logger warn when going from xmp as embedded string to table.
					overlay.develop.xmpCameraRaw = { sha256 = sha256, _replace = true }
		
					overlay._localOnly = { 
						files = {
							xmp_develop = {
								path = pathForSettingsXmp,
								fileSize = settingsLength,
								sha256 = sha256,
							}
						},
						sourceXmpDevelopLink = { _delete = true } -- WF-1323: settings changes should remove sourceXmpDevelopLink as its no longer valid
					}
					
					return overlay, pathForSettingsXmp
					
				end )
				:flatMap( function( overlayWithSha256, path )
					
					return WFAsyncFileUtils.calculatePathSize( path )
						:flatMap( function( pathSize )
							local settingsLen = KSData.length( settingsAsKSData )
							
							if pathSize == settingsLen then -- settings already exist at this location, don't resave
								return Observable.just( overlayWithSha256 )
							end
							
							return WFAsyncFileUtils.saveToFile( settingsAsKSData, path )
								:tapOnError( function( stdError, platformError )
									logger:errorf( "failed to write external develop settings to %s: %s (%s)", path, stdError, platformError )
								end )
								:catch( function()
									return Observable.just( nil )
								end )
								:map( function()
									return overlayWithSha256
								end )
						end )
				:tap( function()
					settingsAsKSData:close()
				end )
				
			end )

		end )

end

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

local function videoDevelopOverlay( videoDevelopSettingsTable, baseOverlay )

	-- SPAM [btran 2022-01-13]
	-- For videos, we are not expecting to externalize settings.
	
	baseOverlay = baseOverlay or {}
	baseOverlay.develop = baseOverlay.develop or {}
	baseOverlay.develop.adjustments = videoDevelopSettingsTable or {}
		
	return Observable.just( baseOverlay )

end

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

local WFXmp
local WFXMLNS

local function developSettingsOverlay ( wfCatalog, assetId, tiffOrientation, croppedWidth, croppedHeight, 
										settingsXmp, devSettingsDate, maybeAuxInfo, maybeNeedsUpdateAISettings, 
										crVersion, optRevId )
	
	-- build xmp from settings to get HDREditMode setting
	--
	-- a much faster alternative is a no regex string.find(), but it makes me feel a little dirty
    -- local HDREditMode = (string.find(settingsXmp, 'crs:HDREditMode="1"', 1, true) ~= nil)
	--
	-- NOTE: with a 500K settings string, parsing the XMP was only 12ms, so we'll stay with
	-- safe approach. 
	
	local startTime = KSDate.timeNowForAnimation()

	WFXmp = WFXmp or require 'WFXmp'
	WFXMLNS = WFXMLNS or require 'WFXMLNS'

	local xmp = WFXmp.WFXmp()
	xmp:parse( settingsXmp )
	local HDREditMode = ( xmp:get( WFXMLNS.crs, 'HDREditMode' ) == "1" )
	xmp:close()
	xmp = nil

	local duration = KSDate.timeNowForAnimation() - startTime
	local threshold = 0.02
	
	if duration > threshold then

		logger:warnf( "xmp parse of settings took %.5f sec (more than threshold of %.5f sec)", duration, threshold )

	end

	local device = WFDeviceInfo.deviceDescription()

	local overlayWithoutXMPCameraRaw = {
		_id = assetId,

		develop = {
			processingModel = 'lightroom',
			crsVersion = crVersion,
			userOrientation = tiffOrientation,
			croppedWidth = croppedWidth,
			croppedHeight = croppedHeight,
			needsUpdateAISettings = maybeNeedsUpdateAISettings or { _delete = true },
			userUpdated = devSettingsDate,
			device = device,
			crsHDREditMode = HDREditMode
		}
	}

	local forceExternalize = true
	return xmpCameraRawOverlay( wfCatalog, settingsXmp, overlayWithoutXMPCameraRaw, forceExternalize )
	
end

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

local function developVideoSettingsOverlay( imageId, temporal, croppedWidth, croppedHeight, settingsTable, dateChanged )
	
	local overlayWithoutAdjustments = {
		_id = imageId,
	}
	
	local develop = {
		-- SPAM [btran 2022-01-13]
		-- We need an equivalent WFVFInfo file to get latest processVersion and version.
		processVersion = 1,
		version = 1,
		croppedWidth = croppedWidth,
		croppedHeight = croppedHeight,
	}
	
	overlayWithoutAdjustments.develop = develop
	-- Either Temporal or Settings table are required in order to make a valid develop section.
	-- for instances of import where both are nil, we do not want to add a develop section to the payload
	
	local device = WFDeviceInfo.deviceDescription()

	if type( temporal ) == 'table' then
		temporal.userUpdated = dateChanged
		temporal.device = device
		overlayWithoutAdjustments.develop.temporal = temporal
	end
	
	if type( settingsTable ) == 'table' then
		settingsTable.userUpdated = dateChanged
		settingsTable.device = device
		return videoDevelopOverlay( settingsTable, overlayWithoutAdjustments )
	end
	
	return Observable.just( overlayWithoutAdjustments )
	
end

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

local function externalizeDevelopSettings( wfCatalog, imageId, maybeMaxLength )

	return wfCatalog:docStore():docValue( imageId )
		:flatMap( function( docValue )
			if docValue._empty then
				-- doc has been deleted, ignore externalizing develop settings
				return Observable.just( true )
			end
			
			local currentSettingsXmp = derefOpt( docValue, 'develop', 'xmpCameraRaw' )
			
			if type( currentSettingsXmp ) ~= 'string' then
				-- doc already has settings externalized, no need to redo
				return Observable.just( true )
			end
			
			local baseOverlay = {
				_id = docValue._localId,
				develop = {
				}
			}
			
			return xmpCameraRawOverlay( wfCatalog, currentSettingsXmp, baseOverlay, true, maybeMaxLength )
				:flatMap( function( overlay )
					if not overlay then
						return Observable.just( false )
					end
					
					return wfCatalog:docStore():applyOverlay( overlay )
						:map( true )
				end )
		end )
	
end

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

local function verifyDevelopHasChanged( currentDocValue, settingsOverlay )

	local copiedCurrentDocValue = KSTableUtils.deepCopy( currentDocValue )
	local copiedSettingsOverlay = KSTableUtils.deepCopy( settingsOverlay )

	-- strip all fields start with "_"

	for k in pairs( copiedSettingsOverlay ) do
		if type( k ) == 'string' and k:match( '^_' ) then
			copiedSettingsOverlay[ k ] = nil
		end
	end

	--If masks or filters  not present then strip that field also
	
	local maybeOverlayDeleteMasks = derefOpt( copiedSettingsOverlay, 'develop', 'masks', '_delete' )
	local maybeOverlayDeleteFilters = derefOpt( copiedSettingsOverlay, 'develop', 'filters', '_delete' )
	
	local currentDocHasMasks = not not derefOpt( copiedCurrentDocValue, 'develop', 'masks' )
	local currentDocHasFilters = not not derefOpt( copiedCurrentDocValue, 'develop', 'filters' )
	
	if not currentDocHasMasks and maybeOverlayDeleteMasks then
		KSTableUtils.setNested( copiedCurrentDocValue, nil, 'develop', 'masks' )
	end
	
	if not currentDocHasFilters and maybeOverlayDeleteFilters then
		KSTableUtils.setNested( copiedCurrentDocValue, nil, 'develop', 'filters' )
	end
	
	-- strip all userUpdated
	
	KSTableUtils.setNested( copiedCurrentDocValue, nil, 'userUpdated' )
	KSTableUtils.setNested( copiedCurrentDocValue, nil, 'develop', 'userUpdated' )
	KSTableUtils.setNested( copiedCurrentDocValue, nil, 'develop', 'temporal', 'userUpdated' )
	KSTableUtils.setNested( copiedCurrentDocValue, nil, 'develop', 'adjustments', 'userUpdated' )

	KSTableUtils.setNested( copiedSettingsOverlay, nil, 'userUpdated' )
	KSTableUtils.setNested( copiedSettingsOverlay, nil, 'develop', 'userUpdated' )
	KSTableUtils.setNested( copiedSettingsOverlay, nil, 'develop', 'temporal', 'userUpdated' )
	KSTableUtils.setNested( copiedSettingsOverlay, nil, 'develop', 'adjustments', 'userUpdated' )

	local _, developHasChanged = WFTableMerge.applyOverlay( copiedCurrentDocValue.develop, copiedSettingsOverlay.develop )

	return developHasChanged

end

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

local function createPendingSaveKey( assetId, saveSequence )

	return string.format( "%s:%d", assetId, saveSequence )
	
end

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

local function waitForPendingDevelopSettingsSaveForAssetIds( assetIds )

	-- For each asset ID, check to see if there are any existing, pending develop
	-- settings saves in progress. If there are, add it to the list of
	-- pending observables (pendingListeners).
	
	local pendingListeners = {}
	
	for _, assetId in ipairs( assetIds ) do
	
		local currentSaveSequenceForAssetId = _currentSaveSequencesByAssetId[ assetId ]
		
		if currentSaveSequenceForAssetId then

			-- A pending save exists for this asset.
			
			local key = createPendingSaveKey( assetId, currentSaveSequenceForAssetId )
			
			local pendingSaveListener = _pendingDevelopSettingsSaveListeners[ key ] or Rx.Subject()
			_pendingDevelopSettingsSaveListeners[ key ] = pendingSaveListener
				-- pendingSaveListener will resolve when the settings have been saved.
				
			pendingListeners[ #pendingListeners + 1 ] = pendingSaveListener
		
		end
	
	end
	
	if #pendingListeners == 0 then
	
		return Observable.just( true )
		
	end
	
	-- Wait for all of the pending saves to complete.
	
	return Observable.from( pendingListeners )
	
		:flatMap( function( pendingListener )
			return pendingListener
		end )
		
		:takeLast( 1 )
	
end

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

local function pendingDevelopSettingsSaveIsComplete( saveSequence, assetId, wasSaveSuccessful )

	-- Upon a successful save, notify anyone waiting for the pending save to complete
	-- that the save was successful.
	
	local currentSaveSequenceForAssetId = _currentSaveSequencesByAssetId[ assetId ]
	
	if saveSequence == currentSaveSequenceForAssetId then
	
		_currentSaveSequencesByAssetId[ assetId ] = nil
			-- There are no more pending saves for this specific asset.
			
	end
	
	local key = createPendingSaveKey( assetId, saveSequence )
	
	local pendingSaveListener = _pendingDevelopSettingsSaveListeners[ key ]
	_pendingDevelopSettingsSaveListeners[ key ] = nil
	
	if pendingSaveListener then

		pendingSaveListener:onNext( wasSaveSuccessful )
		pendingSaveListener:onCompleted()
		
	end
	
end

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

local function pendingDevelopSettingsSaveFailed( saveSequence, assetId )

	pendingDevelopSettingsSaveIsComplete( saveSequence, assetId, false )

end

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

local function pendingDevelopSettingsSaveSucceeded( saveSequence, assetId )

	pendingDevelopSettingsSaveIsComplete( saveSequence, assetId, true )

end

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

local function pendingDevelopSettingsSaveInProgress( saveSequence, assetId )

	-- Keep track of the current save sequence number for each pending save.
	
	_currentSaveSequencesByAssetId[ assetId ] = saveSequence

end

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

local function getSettingsOverlayStream( 
					isVideo, wfCatalog, assetId, tiffOrientation, 
					croppedWidth, croppedHeight,
					settings, devSettingsDate,
					maybeAuxInfo, maybeNeedsUpdateAISettings,
					crVersion, optRevId )
					
	if isVideo then
		return developVideoSettingsOverlay( 
				assetId, croppedWidth, croppedHeight, settings,
				devSettingsDate, crVersion, optRevId )
	else
		return developSettingsOverlay( 
				wfCatalog, assetId, tiffOrientation, croppedWidth, croppedHeight,
				settings, devSettingsDate, maybeAuxInfo, maybeNeedsUpdateAISettings,
				crVersion, optRevId )
	end
					
end

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

local function setCameraRawSettingsStream( 
					wfCatalog, saveSequence, assetId, isVideo,
					croppedWidth, croppedHeight, settings, 
					maybeAuxInfo, maybeNeedsUpdateAISettings,
					cameraModel, tiffOrientation, crVersion, optRevId )
	
	local devSettingsDate = KSDate_timeInIso8601FormatNearestSecond()

	return WFCatAssetAux.filterAuxThatAreNotUploaded( wfCatalog, assetId, maybeAuxInfo or {} )

			:flatMap( function( notUploadedAuxList )

				if not next( notUploadedAuxList ) then
					return Observable.just( true )
				end

				local fnExternalAux = getExternalClientAux()
				local status = false
				
				if fnExternalAux then 
					status = fnExternalAux(notUploadedAuxList)
				end
				
				return Observable.just( status )

			end )

			:flatMap( function()

				local developSettingsOverlayStream = getSettingsOverlayStream(
												isVideo, wfCatalog, assetId, tiffOrientation,
												croppedWidth, croppedHeight,
												settings, devSettingsDate,
												maybeAuxInfo, maybeNeedsUpdateAISettings,
												crVersion, optRevId )

				local auxDataList = maybeAuxInfo or {}

				local auxOverlayObs = WFCatAssetAux.getAuxOverlay(
						wfCatalog, assetId, auxDataList )

				return Observable.combineLatest(
					developSettingsOverlayStream,
					auxOverlayObs )
			end )

			:map( 
				function( baseDevelopOverlay, auxOverlay )

					local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField()

					local baseDevelopOverlayWithUserUpdatedField =
						KSTableUtils.merge( 
								{}, baseOverlay, baseDevelopOverlay )
								
					local developSettingsOverlayWithAux = 
							KSTableUtils.merge( 
								{}, baseDevelopOverlayWithUserUpdatedField, auxOverlay )
					
					return developSettingsOverlayWithAux
					
				end )
			:flatMap(
				function( settingsOverlay )

					return wfCatalog:docStore():docValue( assetId )
						:flatMap( function( docValue )
							
							-- If docValue doesn't exists in docStore, should not save settings
							if not WFPlatformInfo.isUnitTest() and ( docValue._empty or docValue.type ~= 'asset' or docValue._deleted ) then
								return Observable.throw( string.format( "Try to setCameraRawSettings to non-existent assetId %s", assetId ) )
							end
						
							-- If we are requesting to set fromDefaults to false make sure it
							-- previously existed, only change it to false if it did.
		
							-- Since the client code doesn't know whether fromDefaults is present
							-- it always specifies false when an edit occurs. So this code removes
							-- it if it wasn't already in the payload as it should only be set if
							-- the asset was imported or reset to original/default when not using
							-- Adobe Default (catalog.settings.universal.editDefaults.enabled == true).
		
							-- Setting it to false instead of not present will make it easier for
							-- oz engineers to debug issues. I think seeing it as false instead of
							-- not present will allow them to know this asset was originally imported
							-- with editDefaultsEnabled and has now been edited.
							if derefOpt( docValue, 'develop', 'fromDefaults' ) == nil then
								KSTableUtils.setNested( settingsOverlay, nil, 'develop', 'fromDefaults' )
							end
							
							-- If xmp_develop is updated with new file, we should delete old one if sha256 doesn't match
							local newDevelopSha256 = KSTableUtils.derefOpt( settingsOverlay, 'develop', 'xmpCameraRaw', 'sha256' )
							local oldDevelopSha256 = KSTableUtils.derefOpt( docValue, 'develop', 'xmpCameraRaw', 'sha256' )
							
							if oldDevelopSha256 and oldDevelopSha256 ~= newDevelopSha256 then
							
								if not newDevelopSha256 then
									KSTableUtils.setNested( settingsOverlay, { _delete = true }, '_localOnly', 'files', 'xmp_develop' )
								end
								
								return WFCatVolumeAgentQueryHelper.isDevelopSettingsPurgeable( wfCatalog, assetId, oldDevelopSha256 )
									:map( function( isPurgeable )
										local maybePurgeOldXmpDevelopFilePath
										if isPurgeable then
											maybePurgeOldXmpDevelopFilePath = WFCatAssetUtils.getLocalExternalDevelopSettingsPath( docValue )
										end
										return docValue, settingsOverlay, maybePurgeOldXmpDevelopFilePath
									end )
									
							end
							
							return Observable.just( docValue, settingsOverlay )
						
						end )

						:flatMap( function( docValue, settingsOverlay, maybePurgeOldXmpDevelopFilePath )
							
							if WFCapabilities.getCapability( 'LrClassic_avoidEmptyContentUpdate' ) then

								local developHasChanged = verifyDevelopHasChanged( docValue, settingsOverlay )

								if not developHasChanged then
									return Observable.throw( string.format( "Try to setCameraRawSettings with no develop change to assetId %s", docValue._id ) )
								end
								
							end

							return Observable.just( settingsOverlay, maybePurgeOldXmpDevelopFilePath )

						end )
						
				end )
			:flatMap( 
				function( settingsOverlay, maybePurgeOldXmpDevelopFilePath )
					-- optRevId is used by LrClassic, they set the optRevId and in the new changeddocs LrClassic
 					-- will receive the revId and if the revId is the same as they original provided, then
 					-- LrClassic will know they already have the changes and don't need to process the doc again

					settingsOverlay._newRevId = optRevId
					return wfCatalog:docStore():applyOverlay( settingsOverlay )
						:map( function()
							if maybePurgeOldXmpDevelopFilePath then
								WFCatDelete.deleteAssetAndMaybeParentFoldersByAbsolutePath( maybePurgeOldXmpDevelopFilePath )
							end
							return true
						end )
				
				end )
			:call( runUntilDone )
			:map(
				function()

					return saveSequence, assetId

				end )
		
end

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

local function setCameraRawSettings( wfCatalogStream )

	local cameraRawSettingsStream = Rx.Subject()

	return  {

		value = wfCatalogStream
					:mapSwitch(
						function( wfCatalog )

							return cameraRawSettingsStream
									:flatMap(
										function( saveSequence, assetId, isVideo,
													croppedWidth,
													croppedHeight, xmpSettings,
													maybeAuxInfo, maybeNeedsUpdateAISettings,
													cameraModel, tiffOrientation, crVersion, optRevId)

											logger:infof( "starting save of imageId %s", assetId )
					
											return 
												setCameraRawSettingsStream(
													wfCatalog,
													saveSequence, assetId, isVideo,
													croppedWidth,
													croppedHeight, xmpSettings,
													maybeAuxInfo, maybeNeedsUpdateAISettings,
													cameraModel, tiffOrientation, crVersion, optRevId)
													
												:catch(
													function( errorMsg )
													
														pendingDevelopSettingsSaveFailed( saveSequence, assetId )
													
														local failed = true
														return Observable.just( saveSequence, assetId, failed,  errorMsg )
													
													end )

										end )
									:map(
										function( saveSequence, savedAssetId, failed, errorMsg )

											if not failed then
												pendingDevelopSettingsSaveSucceeded( saveSequence, savedAssetId )

												logger:infof( "Saved settings for imageId %s with sequence", 
															savedAssetId, saveSequence )
												errorMsg = nil
											else
												errorMsg = errorMsg or 'unknown'
											end
										
											return { saveSequence, savedAssetId, errorMsg }

										end )

						end ),

		events = {

			save = function( _, saveSequence, assetId, isVideo, 
								croppedWidth, croppedHeight, xmpSettings,
								maybeAuxInfo, maybeNeedsUpdateAISettings,
								cameraModel, tiffOrientation, crVersion, optRevId)

				pendingDevelopSettingsSaveInProgress( saveSequence, assetId )

				cameraRawSettingsStream
					:onNext( saveSequence, assetId, isVideo,
								croppedWidth, croppedHeight, xmpSettings,
								maybeAuxInfo, maybeNeedsUpdateAISettings,
								cameraModel, tiffOrientation, crVersion, optRevId )

			end,
				
		}
	}

end

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

local function saveExternalizedXmp( wfCatalog, xmpKSData, versionDocVal )

	local xmpSha256 = KSTableUtils.derefOpt( versionDocVal, "develop",
													"xmpCameraRaw", "sha256" )

	if not xmpKSData then
		return Observable.throw(
			string.format( "invalid externalized xmp data for version doc %s",
														versionDocVal._id ) )
	end
	
	local path = wfCatalog:fileSpec():pathForSettingsXmp( xmpSha256 )
	
	return WFAsyncFileUtils.saveToFile( xmpKSData, path )
		:map( function( success )
			if success then
				return path, KSData.length( xmpKSData ), xmpSha256
			else
				error( string.format( "failed to save externalized xmp to %s", path ) )
			end
		end )
			
end

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

return {

	isUpdateWanted = isUpdateWanted,

	ongoingDevelopContentModel = ongoingDevelopContentModel,

	developContentModel = developContentModel,

	developContentWithLocalNegativePathModel = 
		developContentWithLocalNegativePathModel,

	ongoingDevelopContentsModel = ongoingDevelopContentsModel,

	developSettingsOverlay = developSettingsOverlay,
	
	developVideoSettingsOverlay = developVideoSettingsOverlay,
	
	externalizeDevelopSettings = externalizeDevelopSettings,

	setCameraRawSettings = setCameraRawSettings,
	
	getSettingsOverlayStream = getSettingsOverlayStream,
	
	saveExternalizedXmp = saveExternalizedXmp,
	
	getExternalClientAux = getExternalClientAux,
		
	waitForPendingDevelopSettingsSaveForAssetIds = waitForPendingDevelopSettingsSaveForAssetIds,

	_DEBUG_pendingDevelopSettingsSaveFailed = pendingDevelopSettingsSaveFailed,

	_DEBUG_pendingDevelopSettingsSaveSucceeded = pendingDevelopSettingsSaveSucceeded,

	_DEBUG_pendingDevelopSettingsSaveInProgress = pendingDevelopSettingsSaveInProgress,

	_TEST_getCapabilitiesCache = getCapabilitiesCache,

	_privateUnitTestExports = {

		makeMostRecentDevelopContentStream = makeMostRecentDevelopContentStream,

	},
	
}
