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

WFCatWatermark.lua

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

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

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

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

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

local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local KSStringUtils = require 'KSStringUtils'
local WFCatUtils = require 'WFCatUtils'
local WFLogger = require 'WFLogger'
local WFCatDelete = require 'WFCatDelete'
local WFArguments = require 'WFArguments'
local WFTableMerge = require 'WFTableMerge'
local WFOzDocUtils = require 'WFOzDocUtils'
local WFFileUtils = require 'WFFileUtils'
local WFCrypto = require 'WFCrypto'
local WFPlatformInfo = require 'WFPlatformInfo'
local KSPathUtils = require 'KSPathUtils'

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

local logger = WFLogger( 'WFCatWatermark' )
local check = WFArguments.check
local derefOpt = KSTableUtils.derefOpt
local deepCopy = KSTableUtils.deepCopy
local deepEquals = KSTableUtils.deepEquals

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

local OZ_MAX_WATERMARK_TEXT_CHARS = 1024
local CLIENT_MAX_WATERMARK_NAME_CHARS = 250
local OZ_MAX_WATERMARK_IMAGE_BINARY_SIZE_IN_BYTES = ( 5 * 1024 * 1024 )
local kOzSupportedExtensionInfo = { "png", "jpg", "jpeg", "jpe", "jfif" }

local WFCatWatermark = {}

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

local function getWatermarkInfoFromDocValue( docValue )

	local id = derefOpt( docValue, '_id' )
	local master = derefOpt( docValue, '_localOnly', 'universal', 'files', 'original', 'master' )
	local importSource = derefOpt( docValue, 'importSource' )
	local href = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href' )
	local isWatermarkInvalid = not not derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'invalid' )

	local watermarkInfo = deepCopy( docValue )
	watermarkInfo = WFOzDocUtils.clearNonPayloadKeys( watermarkInfo )
	watermarkInfo.id = id

	if( importSource ) then
		watermarkInfo.filePath = nil
		watermarkInfo.originalLinkAvailable = false
		-- set the filePath appropriately
		if( master and master.sha256 == importSource.sha256 ) then
			-- available locally.
			-- update the path
			watermarkInfo.filePath = master.relativePath
		 end
		 -- set the originalLinkAvailable appropriately
		if( href and not isWatermarkInvalid ) then
			watermarkInfo.originalLinkAvailable = true
		 end
	end

	local activeWatermarkType = watermarkInfo.activeWatermarkType

	-- preventive check: make sure valid activeWatermarkType being sent to clients
	-- This check ensures that WF won't send invalid data to clients incase of the mix of down sync from old version.
	-- if "activewatermarkType" is nil or not set to "image" or "text", then "fallback" to text
	if not ( activeWatermarkType and activeWatermarkType == "text" or activeWatermarkType == "image" ) then

		logger:warnf( "watermark unknown activeWatermarkType = %s detected for docId = %s and it will be defaulted to text",
		activeWatermarkType, watermarkInfo.id )

		watermarkInfo.activeWatermarkType = "text"

	end

	return watermarkInfo

end

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

local function truncateStringLength( str, maxChars)

	if KSStringUtils.lengthOfUTF8String( str ) > maxChars then
		logger:warnf( 'Watermark: string lengths are subject to ceiling, truncating: %s to %d chars', str, maxChars )
		str = KSStringUtils.UTF8Substring( str, 1, maxChars )
	end

	return str

end

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

local function checkForDuplicateNameAndOptId( wfCatalog, name, optId )

	check( 'WFCatalog', '+string', '?string' )

	-- run the query once : covers all the cases, like optId may be nil and there may be duplicate name or Id or both.
	-- Wow!, The most beautiful query:)
	return wfCatalog:indexStore():queryOnce {
		sql = [[SELECT COUNT(name), COUNT(docId) FROM watermark WHERE watermark.name == ? OR watermark.docId == ?]],
		name,
		optId,
		quiesceUntilUpToDate = true,
	}
	:map( function( row )
		-- if you ever wonder how could checking the first row helps to detect the duplicate in name or in docId.
		-- idea, if docId exist, there exist a associated name.
		-- Hence if docId matched, the first row will be returned 1 as well.

		local duplicates = derefOpt( row, 1, 1 ) or 0
		return duplicates > 0

	end )

end

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

local function getAbsolutePathAndVolumeGuid( wfCatalog, path )

	-- The argument "path" here refers absolute path for Desktop Client and
	-- relative path for Mobile clients
	-- volumeGuid is applicable only for Desktop Client.

	local absolutePath, volumeGuid

	if WFPlatformInfo.isUnitTest() then
		local standardPath = KSPathUtils.getStandardFilePath( "testdata" )
		absolutePath = KSPathUtils.child( standardPath, path )
		if WFPlatformInfo.isDesktop then
			volumeGuid = WFFileUtils.getVolumeGuidForPath( path )
		end

	elseif WFPlatformInfo.isMobile() then
		local catalogId = wfCatalog:ozCatalogId()
		local standardPath = KSPathUtils.child( KSPathUtils.getStandardFilePath( 'carouselDocuments' ), catalogId )
		absolutePath = KSPathUtils.child( standardPath, path )

	elseif WFPlatformInfo.isDesktop then
		absolutePath = WFFileUtils.createFullPathFromRelativePathAndCatalogPath( wfCatalog, path )
		volumeGuid = WFFileUtils.getVolumeGuidForPath( absolutePath )
	end

	return absolutePath, volumeGuid

end

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

local function getFileSize( absolutePath )

	local fileAttributes = absolutePath and WFFileUtils.fileAttributes( absolutePath )
	if not fileAttributes or not fileAttributes.fileSize then
		error( "File does not exist" )
	end
	return fileAttributes.fileSize

end

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

local function checkAndGetFileSizeIfWithinTheMaxLimit( absolutePath )

	local fileSize = getFileSize( absolutePath )
	if fileSize > OZ_MAX_WATERMARK_IMAGE_BINARY_SIZE_IN_BYTES then
		error( "Image watermark file size in bytes( "
			.. fileSize .. " ) must not exceed " .. OZ_MAX_WATERMARK_IMAGE_BINARY_SIZE_IN_BYTES )

	end
	return fileSize

end

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

local function isEmpty( t )

	check( 'table|string|nil' )
	if( type( t ) == "table" ) then
		return ( next( t ) == nil )
	end
	return ( t == nil or t == "" )

end

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

local function validateFileExtension( filePath, fileName )

	check ( '+string', '+string' )
	-- make sure clients sets the extension correctly
	local pathExtension = filePath:match( "%.([^%.]+)$" ):lower()
	local fileExtension = fileName:match( "%.([^%.]+)$" ):lower()

	-- filePath extension and fileName extension should match
	if not ( pathExtension == fileExtension ) then
		error( "Mismatch of filePath and fileName extension, " .. pathExtension .. ", " .. fileExtension )
	end

	-- since the previous comparison make sure both the extensions are same!.
	-- Further validating any one of the extensions is good enough!.
	for _, v in ipairs( kOzSupportedExtensionInfo ) do
		if( v == pathExtension ) then
			return true
		end
	end
	error( "Unsupported image file extension, " .. pathExtension )

end

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

local function buildLocalOnly( action, wfCatalog, relativePath, fileName, docValue )

	check( 'string', 'WFCatalog', '?string', '?string', '?table' )

	local absolutePath, volumeGuid = getAbsolutePathAndVolumeGuid( wfCatalog, relativePath )

	local fileSize = checkAndGetFileSizeIfWithinTheMaxLimit( absolutePath )

	local sha256Digest = WFCrypto.digestFile( absolutePath, 'SHA256' )

	if action == "update" then
		local master = derefOpt( docValue, '_localOnly', 'universal', 'files', 'original', 'master' )
		local importSource = derefOpt( docValue, 'importSource' )
		if ( master and master.fileSize == fileSize and master.sha256 == sha256Digest and
			importSource and importSource.fileName == fileName ) then

			logger:warnf( "watermark update watermark, no change in binary and hence binary will not be uploaded" )
			return {}

		end
	end

	local localOnly = {
		universal = {
			files = {
				original = {
					master = {
						relativePath = relativePath,
						sha256 = sha256Digest,
						fileSize = fileSize,
						volumeGuid = volumeGuid,
					}
				}
			}
		}
	}
	return localOnly

end

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

local function validateActiveWatermarkTypeParam( activeWatermarkType )

	if not activeWatermarkType then
		error( "activeWatermarkType shall not be nil" )
	end
	if activeWatermarkType == "text" or activeWatermarkType == "image" then
		return activeWatermarkType
	else
		error( "Unknown activeWatermarkType received, " .. activeWatermarkType )
	end

end

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

local function isTextNonEmpty( text )

	-- check if text present then it shall not be empty.
	-- Flag an error if it is empty
	if text and text == "" then
		error( "Text field, if it is present, then it shall not be empty" )
	end
	return true

end

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

local function processSettings( action, wfCatalog, name, settings, optFilePath, optFileName, docValue )

	-- check if table is empty
	if isEmpty( settings ) then
		error( "Invalid argument, empty settings received in " .. action .. " watermark" )
	end

	-- check if valid activeWatermarkType is set
	local activeWatermarkType = validateActiveWatermarkTypeParam( settings.activeWatermarkType )

	local watermarkDoc = deepCopy( settings )

	logger:infof( action .. " watermark name = %s, activeWatermarkType = %s, optFilePath = %s",
			name, activeWatermarkType, optFilePath )

	-- Allow processing of optFilePath and text specific fields if it is not nil
	-- irrespective of whether activeWatermarkType points to image or text because of unified object model.
	-- validate file extension: must be jpg or png only. case insensitive.
	if( not isEmpty( optFilePath ) and
		not isEmpty( optFileName ) and
		validateFileExtension( optFilePath, optFileName ) ) then
		watermarkDoc._localOnly = buildLocalOnly( action, wfCatalog, optFilePath, optFileName, docValue )
	end

	-- limit name size to max of CLIENT_MAX_WATERMARK_NAME_CHARS
	name = truncateStringLength( name, CLIENT_MAX_WATERMARK_NAME_CHARS )

	-- check if text field present, then it shall not be empty
	if watermarkDoc.text and isTextNonEmpty( watermarkDoc.text ) then
		-- limit text size to max of OZ_MAX_WATERMARK_TEXT_CHARS
		watermarkDoc.text = truncateStringLength( watermarkDoc.text, OZ_MAX_WATERMARK_TEXT_CHARS )
	end

	watermarkDoc.displayName = name
	watermarkDoc.type = "watermark"
	-- subtype will stay as text for both text and image watermark because of single object model.
	watermarkDoc.subtype = "text"

	-- remove if inadvertently Client included filePath or fileName in the settings.
	-- Though this check very much applicable only for update watermark,
	-- no harm in keeping it for create watermark as well.
	-- because of unified Object model, better to keep this check common to both create and update APIs
	if watermarkDoc.filePath then
		logger:warnf( "watermark filePath shall not be part of the settings. It will be removed")
		watermarkDoc["filePath"] = nil
	end

	if watermarkDoc.fileName then
		logger:warnf( "watermark fileName shall not be part of the settings. It will be removed")
		watermarkDoc["fileName"] = nil
	end

	return watermarkDoc

end

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

local function addImportSourceToDocument( watermarkDoc, optFileName, optDocValue )

	check( 'table', '?string', '?table' )

	if( not isEmpty( watermarkDoc._localOnly ) and optFileName ) then
		-- local only update happened. construct/update new importSource
		local master = derefOpt( watermarkDoc, '_localOnly', 'universal', 'files', 'original', 'master' )
		watermarkDoc.importSource = {
			fileName = optFileName,
			sha256 = master.sha256,
			fileSize = master.fileSize,
			importTimestamp = watermarkDoc.userCreated,
		}
		logger:debug( "watermark importSource has been updated for docId = %s", watermarkDoc._id )
	-- no localOnly updated detected. check if there is importSource exist already, copy it.
	elseif( not isEmpty( optDocValue ) and not isEmpty( optDocValue.importSource ) ) then
		watermarkDoc.importSource = deepCopy( optDocValue.importSource )
		watermarkDoc.importSource.importTimestamp = watermarkDoc.userCreated
		logger:debug( "watermark existing importSource has been copied for docId = %s", watermarkDoc._id )
	end

	return watermarkDoc

end

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

--[[!
\brief Create a new text watermark.

\param wfCatalog (WFCatalog)
\param name (string ) the name of the watermark.
\param settings ( table ) the watermark settings.
	{
		opacity = 73,
		rotation = 90,
		position = "center",
		xOffset =  0.23,
		yOffset =  0.40,
		size = 22,
		text = "@ John Warnock",
		font = {
			name = "SourceSansPro",
			bold = true,
			italic = true,
		},
		shadow = {
			enabled = "simple"
		},
		rgbColor = { 0.4, 0.333212235, 0.0 },

		-- Mandatory param( type string ).Valid values are image or text
		-- WF will return an error if this param is missing or not set to text or image.
		activeWatermarkType,
	}

\param optFilePath( string ) - the optional image watermark file path

	Mobile Clients:
		It is a relative path relative to carouselDocuments/catalogId
		eq: if the absolute path is
		"/Users/dharmara/Library/Developer/CoreSimulator/Devices
		/D862232B-AEB3-4947-8661-C54020B00852/data/Containers/Data
		/Application/C4E9E634-C4B1-4ACD-8820-54672E5A757F/Documents
		/c1737ec7bfe942dd95d8c733f682b5df/Originals/2020/2020-06-05/IMG_8749.JPG"
		Then the relative path shall be "Originals/2020/2020-06-05/IMG_8749.JPG"

	For desktop:
		It is a relative path with respect to the catalog path
		eq: "/Users/dharmara/Pictures/Lightroom Library.lrlibrary
		/c1737ec7bfe942dd95d8c733f682b5df/originals/2020/2020-06-05/IMG_8749.JPG"
		then the path shall be "originals/2020/2020-06-05/IMG_8749.JPG"
		Note: volumeGuid is not passed by client. It is driven in WF for the given path.

\param optFileName( string ) - the file name
	Eq. IMG_8749.JPG"

\param optID( string ) - UUID
	Eq. c1737ec7bfe942dd95d8c733f682b5df

\return <code>Observable</code> The watermark ID of the created watermark doc.
--]]

function WFCatWatermark.createWatermark( wfCatalog, name, settings, optFilePath, optFileName, optId )

	check( 'WFCatalog', '+string', 'table', '?string', '?string', '?string' )

	return checkForDuplicateNameAndOptId( wfCatalog, name, optId )
		:flatMap( function( isDuplicate )

			if isDuplicate then
				return Observable.throw( "Invalid argument, duplicate watermark name or Id received" )
			end

			local watermarkDoc = processSettings( "create", wfCatalog, name, settings, optFilePath, optFileName )
			watermarkDoc._id = optId or KSIdGenerator.uuidInLowercaseHex()
			watermarkDoc = WFCatUtils.addUserCreatedFields( watermarkDoc )
			watermarkDoc = addImportSourceToDocument( watermarkDoc, optFileName )

			return wfCatalog:docStore():setDocValue( watermarkDoc )
				:map( function( revId )
					logger:debug( "watermark createWatermark docStore update successful for docId = %s", watermarkDoc._id )
					return watermarkDoc._id
				end )
				:tapOnError( function( err )
					logger:errorf( "Adding to catalog failed: err:%s", err )
				end )

		end )

end

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

--[[!
\brief Update the text watermark.

\param wfCatalog (WFCatalog)
\param id (string) The full doc ID of the watermark
\param name (string ) the name of the watermark( expected even when no change in name ).
\param settings (table ) the watermark settings( expected full settings ).
	{
		opacity = 73,
		rotation = 90,
		position = "center",
		xOffset =  0.23,
		yOffset =  0.40,
		size = 22,
		text = "@ John Warnock",
		font = {
			name = "SourceSansPro",
			bold = true,
			italic = true,
		},
		shadow = {
			enabled = "simple"
		},
		rgbColor = { 0.4, 0.333212235, 0.0 },

		-- Mandatory param( type string ). Valid values are image or text
		-- WF will return an error if this param is missing or not set to text or image.
		activeWatermarkType,
	}
\param optFilePath( string ) - the optional image watermark file path

	Mobile Clients:
		It is a relative path relative to carouselDocuments/catalogId
		eq: if the absolute path is
		"/Users/dharmara/Library/Developer/CoreSimulator/Devices
		/D862232B-AEB3-4947-8661-C54020B00852/data/Containers/Data
		/Application/C4E9E634-C4B1-4ACD-8820-54672E5A757F/Documents
		/c1737ec7bfe942dd95d8c733f682b5df/Originals/2020/2020-06-05/IMG_8749.JPG"
		Then the relative path shall be "Originals/2020/2020-06-05/IMG_8749.JPG"

	For desktop:
		It is a relative path with respect to the catalog path
		eq: "/Users/dharmara/Pictures/Lightroom Library.lrlibrary
		/c1737ec7bfe942dd95d8c733f682b5df/originals/2020/2020-06-05/IMG_8749.JPG"
		then the path shall be "originals/2020/2020-06-05/IMG_8749.JPG"
		Note: volumeGuid is not passed by client. It is driven in WF for the given path.

\param optFileName( string ) - the file name
	Eq. IMG_8749.JPG"

\return <code>Observable</code> The revision ID of the updated watermark doc.
--]]

function WFCatWatermark.updateWatermark( wfCatalog, id, name, settings, optFilePath, optFileName )

	check( 'WFCatalog', 'string', '+string', 'table', '?string', '?string' )

	return wfCatalog:docStore():docValue( id )
		:flatMap( function( docValue )

			local watermarkDoc = processSettings( "update", wfCatalog, name, settings, optFilePath, optFileName, docValue )
			watermarkDoc.userCreated = watermarkDoc.userCreated or docValue.userCreated
			watermarkDoc = WFCatUtils.addUserUpdatedFields( watermarkDoc )
			watermarkDoc._id = id
			watermarkDoc = addImportSourceToDocument( watermarkDoc, optFileName, docValue )

			local baseDocValue = WFOzDocUtils.clearNonPayloadKeys( docValue )

			local watermarkOverlay = WFTableMerge.createOverlay( baseDocValue, watermarkDoc )
			watermarkOverlay._id = id
			logger:debug( "watermark updateWatermark docStore update successful for docId = %s", watermarkDoc._id )
			return wfCatalog:docStore():applyOverlay( watermarkOverlay )

		end )

end

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

--[[!
\brief Get watermark info for a watermark .

\param wfCatalog (WFCatalog)
\param watermarkId (string) The full doc ID of the watermark

\return <code>Observable</code> The watermark info(table).

	{
		displayName = "my first watermark"
		id = "ca09100b9fe0ba5c14e784e886642606"
		subtype = "text"
		opacity = 73,
		rotation = 90,
		position = "center",
		xOffset =  0.23,
		yOffset =  0.40,
		size = 22,
		text = "@ John Warnock",
		font = {
			name = "SourceSansPro",
			bold = true,
			italic = true,
		},
		shadow = {
			enabled = "off"
		},
		rgbColor = { 0.4, 0.333212235, 0.0 },
		userCreated = 2020-02-20T18:49:26.676Z,
		userUpdated = 2020-02-20T18:49:26.676Z,
		-- filePath is relative path for Mobile clients and absolute Path for Desktop.
		-- Please refer either create watermark or update watermark APIs for more details and examples.
		filePath = Originals/2020/2020-06-05/IMG_8749.JPG,

	}

--]]

function WFCatWatermark.getWatermarkInfo( wfCatalog, watermarkId )

	check( 'WFCatalog', 'string' )

	return wfCatalog:docStore():docValue( watermarkId )
		:map( function( docValue )

			if( docValue.type ~= 'watermark' ) then
				logger:warnf( "watermark the docId = %s doesn't corrospond to watermark doc", watermarkId )
				return {}
			end

			local watermarkInfo = getWatermarkInfoFromDocValue( docValue )

			return watermarkInfo

		end )
end

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

--[[!
\brief Get All the watermarks Info.

\param wfCatalog (WFCatalog)

 -- please refer the getWatermarkInfo API for more details about the structure of watermarkInfo.
\return <code>Observable</code> List of watermarks Info(table).
--]]

function WFCatWatermark.getAllWatermarksInfo( wfCatalog )

	check( 'WFCatalog' )

	return wfCatalog:indexStore():query {
		-- get all the watermark Ids sorted by user created
		sql =  [[ SELECT docId, created FROM watermark ORDER BY created ASC ]],
		quiesceUntilUpToDate = true,
		calm = 2,
	}
	:map( function( rows )
		-- Array of watermark Ids
		local watermarkIds = {}
		for i = 1, #rows do
			watermarkIds[i] = rows[i][1]
		end
		return watermarkIds
	end )
	:flatMapLatest( function( watermarkIds )

		if( isEmpty( watermarkIds ) ) then
			-- This is the case when getAllWatermarksInfo being opened when
			-- there is no watermarks being created WF should return empty table as per
			-- agreement with clients
			return Observable.just( watermarkIds )
		end

		-- For Oct 2020, Max Release, only one watermark is supported.
		-- When multiple watermarks support enabled monitoring many documents may hit a performance.
		-- Hence currently restrict the monitoring to max of 100, which is more than sufficient.

		return Observable.defer( function()

			-- Create the new ongoingDocValue observables within an observable in the flatMapLatest
			-- above so it can dispose of the existing ongoingDocValues before creating new ongoingDocValues.
			-- The idea is if we create them outside of the subscription we will refer to ones that
			-- will be disposed prior to their use.

			local maxLimit = math.min( #watermarkIds, 100 )
			local docValueObservables = {}

			local function distinctTableCompare( _, tbl1, _, tbl2 )
				return deepEquals( tbl1, tbl2 )
			end

			for i = 1, maxLimit  do
				docValueObservables[i] = wfCatalog:docStore():ongoingDocValue( watermarkIds[ i ] )
				:map( getWatermarkInfoFromDocValue )
				:distinctUntilChanged( nil, distinctTableCompare )
			end
			return Observable.just( docValueObservables )

		end )
		:flatMapLatest( function( docValueObservables )

			return Observable.combineLatest( unpack( docValueObservables ) )
			:map( function( ... )
				return { ... }

			end )

		end )

	end )

end

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

local function returnTrue()
	return true
end

------------------------------------------------------------
--[[!
\brief Delete the list of watermarks.

\param catalog (WFCatalog)
\param watermarkIdsArray (array). The list of watermark Ids

\return <code>Observable</code> Resolves to true once all of the watermark Ids
	have been deleted.
--]]

function WFCatWatermark.deleteWatermarks( wfCatalog, watermarkIdsArray )

	-- check if the array is empty, simply ignore and return true
	if not watermarkIdsArray or #watermarkIdsArray < 1 then
		return Observable.just( true )
	end

	return wfCatalog:indexStore():queryOnce {
		-- get all the watermark Ids sorted by user created
		sql = [[ SELECT docId FROM watermark ORDER BY docId ASC]],
		quiesceUntilUpToDate = true,
		singleColumnQuery = true
	}
	:map( function( docIds )
		local existingIdsTbl = {}
		for i = 1, #docIds do
			local id = docIds[i]
			existingIdsTbl[id] = true
		end
		return existingIdsTbl
	end )
	:flatMap( function( existingIdsTbl )

		local deleteList = {}
		for i = 1, #watermarkIdsArray do
			-- delete only if the watermark ids present, if not, no action required.
			local id = watermarkIdsArray[i]
			 if( existingIdsTbl[id] == true ) then
				logger:infof( "watermark, the docId = %s will be deleted", id )
				table.insert( deleteList, id )
			end
		end

		return WFCatDelete.deleteDocs( wfCatalog:docStore(), deleteList )

	end )
	:map( returnTrue )
end

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

--[[!
	\brief Download Image watermark

	\param catalog (WFCatalog)

	\param docId (string)  full doc ID of the image watermark

	\param saveFunc (function) A function that takes three arguments: 'catalog',
		a KSData instance containing the image watermark and the watermark's docValue
		This function should return the path of the saved image watermark,
		the file size of the saved file, and the sha256 on success, or throw on failure.

	\return <code>Observable</code> Resolves to the path of the saved image watermark on success or throws on error.
--]]

function WFCatWatermark.downloadImageWatermark( catalog, docId, saveFunc )

	check( "WFCatalog", "+string", "function" )

	return catalog:docStore():docValue( docId )

		:flatMapLatest( function( docValue )

			local href = derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'href' )
			if not href then
				return Observable.throw( "No image watermark href" )
			else
				local url = string.format( "%s%s", catalog:ozUser():serverBaseUrl(), href )
				return catalog:ozUser():get( url )
					:map( function( ... )
						return docValue, ...
					end )
					:catch( function( response, headers )
						return Observable.just( docValue, response, headers )
					end )
			end

		end )

		:flatMapLatest( function( docValue, response, headers )

			if type( headers ) == "table" and headers.status_code == "200" then

				return saveFunc( catalog, response, docValue )

					:flatMapLatest( function( filePath, fileSize, sha256 )

						if type( filePath ) == "string" then

							local volumeGuid

							if WFPlatformInfo.isDesktop then
								volumeGuid = WFFileUtils.getVolumeGuidForPath( WFFileUtils.createFullPathFromRelativePathAndCatalogPath( catalog, filePath ) )
							end

							local localOnly = {
								universal = {
									files = {
										original = {
											master = {
												relativePath = filePath,
												sha256 = sha256,
												fileSize = fileSize,
												volumeGuid = volumeGuid,
											}
										}
									}
								}
							}
							return catalog:docStore():applyOverlay {
									_id = docId,
									_localOnly = localOnly
								}

								:map( function()
									return ( { status = "success", filePath = filePath } )
								end )

						else
							return Observable.throw( "saveFunc returned non-string path" )
						end

					end )

			else

				-- WF-1288: Error cases, response is in the form of userdata and an issue has been reported
				-- by desktop that it has difficuluty in getting userdata to clients through deamon.
				-- response is of no use in this case to clients. so, Just send header only!.
				return Observable.just( { status = "error", headers = headers } )

			end

			-- Just in case....
			return Observable.just( { status = "error", headers = headers } )

		end )


end

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

return WFCatWatermark

