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

WFCatImportDuplicateFiles.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
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 Systems Incorporated.

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

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

	-- WF.core.utils
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCrypto = require 'WFCrypto'

	-- WF.oz_client.managed_catalog
local WFCatFileSpec = require 'WFCatFileSpec'
local WFCatImportConstants = require "WFCatImportConstants"
local WFCatIndexUtils = require 'WFCatIndexUtils'

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

local check = assert( WFArguments.check, "Cannot find WFArguments.check" )
local derefOpt = KSTableUtils.derefOpt
local isValidAssetDocument = WFCatIndexUtils.isValidAssetDocument

local kNoLink = 0
local kHasProxyLink = 1
local kHasMasterLink = 2

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

local WFCatImportDuplicateFiles = {}

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

local function docNeedsImportOfMaster( docValue )

	return( not WFCatFileSpec.getOriginalVolumeGuidFromDocValue( docValue )
		or not WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue ) )
		and not WFCatFileSpec.getOriginalLinkFromDocValue( docValue )

	-- If the doc is marked with willNotUploadMaster it is from Lightroom and
	-- doesn't have a master.  However ignore this flag and just check to see 
	-- if the master hasn't been re-imported already by looking at local path and
	-- the oz master link. Files may have been imported and then locally deleted
	-- before the master has had the chance to upload (but the metadata may have
	-- been uploaded), this would cause the user not to be able to re-import the
	-- file to get the master back.

end

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

function WFCatImportDuplicateFiles.generateIndexSchema( indexStoreSchema )

	check( 'WFRxIndexStoreSchema' )

	indexStoreSchema:addTable {

		sql = [[CREATE TABLE importSource( 
				sha256, 
				originalDigest, 
				assetId, 
				fileName, 
				fileSize, 
				needToImportMaster, 
				hasLinkOnOz, 
				captureDate,
				originalSHA256,
				spaceId,
				linkedLocationId,
				linkedLocationSource )]],
				
		updateLazily = false,
		
		relevantTypeSubtypeTuples = {
			{ 'asset', 'image' },
			{ 'asset', 'video' },
		},
		
		map = function( docValue, emit )

			if isValidAssetDocument( docValue ) and 
				( docValue.subtype == 'image' or docValue.subtype == 'video' ) then
				-- Ignore trash assets (deleted_image, deleted_video) so user can re-import.

				local sha256 = derefOpt( docValue, 'importSource', 'sha256' )
				
				if sha256 then
					sha256 = string.lower( sha256 )
				end

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

				if originalSHA256 then
					originalSHA256 = string.lower( originalSHA256 )
				end
				
				local originalDigest = derefOpt( docValue, 'importSource', 'originalDigest' )

				if originalDigest then
					-- ACR (and Oz API spec) has this value in upper case. But there may be some assets 
					-- on Oz that have a lower case originalDigest if they came in from other clients
					originalDigest = string.upper( originalDigest )
				end

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

				fileName = fileName or ""
					-- v2 clients require fileName being set, in v1 it was optional.

					-- if fileName isn't set we may end up incorrectly assuming a file
					-- isn't a duplicate but the sha256 check will catch it.
					-- (This can also happen if we imported a file with a different
					-- name anyways).

					-- Having a nil fileName is unlikley to happen as current clients
					-- (lightroom, lightroom mobile, lightroom web and Nimbus) are
					-- setting the fileName.

					-- Don't use KSPathUtils.leafName( filePath ) as filePath
					-- has platform specific path delimiters and may have come
					-- from a different platform.

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

				local needToImportMaster = docNeedsImportOfMaster( docValue ) and 1 or 0

				local hasLinkOnOz = WFCatFileSpec.getOriginalLinkFromDocValue( docValue ) and kHasMasterLink

				if not hasLinkOnOz then
					hasLinkOnOz = WFCatFileSpec.getProxyLinkFromDocValue( docValue ) and kHasProxyLink or kNoLink
				end


				if hasLinkOnOz ~= kNoLink and docValue.subtype == 'image' and
					( derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/xmp' ) == nil or
					  derefOpt( docValue, '_localOnly', 'xmpSidecarNeedsUpload' ) --[[legacy case where we explicitly marked xmp needing upload]]) then

					-- Don't consider the image to be uploaded to oz until the xmp link has
					-- been generated and the sidecar file has been uploaded (if it has one).

					-- This is done to prevent copying a partially uploaded asset.

					hasLinkOnOz = kNoLink

				end
				
				local captureDate = derefOpt( docValue, 'captureDate' )
				
				local spaceId = docValue._spaceId
				
				local linkedLocationId = derefOpt( docValue, 'importSource', 'linkedLocation', 'id' )
				local linkedLocationSource = derefOpt( docValue, 'importSource', 'linkedLocation', 'source' )

				emit( sha256, originalDigest, docValue._id, fileName, fileSize, needToImportMaster, hasLinkOnOz, captureDate, originalSHA256, spaceId, linkedLocationId, linkedLocationSource )

			end

		end,

		useLocalId = true,
		hasOneRowPerDocument = true,

	}

	indexStoreSchema:addIndex [[CREATE INDEX importSource_sha256 ON importSource( sha256, assetId )]]
	indexStoreSchema:addIndex [[CREATE INDEX importSource_originalDigest ON importSource( originalDigest, assetId )]]
	indexStoreSchema:addIndex [[CREATE INDEX importSource_fileNamefileSize ON importSource( fileName, fileSize )]]
	indexStoreSchema:addIndex [[CREATE INDEX importSource_sha256hasLinkOnOz ON importSource( sha256, hasLinkOnOz )]]
	indexStoreSchema:addIndex [[CREATE INDEX importSource_needToImportMaster on importSource( needToImportMaster, assetId )]]
	indexStoreSchema:addIndex [[CREATE INDEX importSource_linkedLocation on importSource( linkedLocationId, linkedLocationSource )]]
	
end

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

function WFCatImportDuplicateFiles.isDuplicateFile( fileInfo, possibleMatches )

	assert( fileInfo.fileName and fileInfo.fileSize )

	if not possibleMatches then
		return false
	end
	
	local fileNameSet = possibleMatches[ fileInfo.fileName ]
	
	if fileNameSet == nil then
		return false
	elseif fileNameSet[ fileInfo.fileSize ] then
		return true
	else
		return false
	end

end

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

local function makeHashesContainer()
	return {
		hashToDocId = {},
			-- keys are hash (string), value are docId-s (string) if there is only
			-- one entry and table of the format { 'docId1' = true, 'docId2' = true, etc }
			-- if there are duplicates
			
		docIdToHash = {},
			-- keys are docId-s (string), values are hash (string)
	}
end

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

local sha256Hashes
	-- A table keyed on sha256 and values that are typically docIds (strings)
	-- but it will hold a table keyed on docId if mulitiple docIds share the
	-- same sha256 hash.

local sha256Hashes_docsWithoutMaster
	-- A table of hashes for docIds that have the willNotUploadMaster flag set
	-- and don't have an original path or link.
	-- These docIds are imported by Lightroom and will never have their master
	-- image uploaded by lightroom.
	
local originalDigests
	-- Similar container (like sha256Hashes and sha256Hashes_docsWithoutMaster),
	-- but for the duplicate-detection-digest that is computed by ACR

local hashesChangesSubscription
local hashesCloseActionAdded
local hashesLoaded
local hashesLoader

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

local function disposeHashes()

	if sha256Hashes then
		sha256Hashes = nil
	end
	
	if sha256Hashes_docsWithoutMaster then
		sha256Hashes_docsWithoutMaster = nil
	end
	
	if originalDigests then
		originalDigests = nil
	end

	if hashesChangesSubscription then
		hashesChangesSubscription:dispose()
		hashesChangesSubscription = nil
	end

	hashesLoaded = nil
	hashesLoader = nil

end

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

local function hashesCloseAction()

	return function()
		disposeHashes()
		hashesCloseActionAdded = nil
	end

end

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

local function setDocIdForHash( hashesContainer, hash, docId )

	if not hash then
		return
	end
	
	local prevValue = hashesContainer.hashToDocId[ hash ]

	if prevValue ~= docId then

		if prevValue == nil then

			hashesContainer.hashToDocId[ hash ] = docId
			hashesContainer.docIdToHash[ docId ] = hash

		else

			-- Multiple docIds share the same hash make or add to a table
			-- keyed on docIds to can keep track of them.

			local hashCollisonTable

			if type( prevValue ) == 'table' then

				hashCollisonTable = prevValue

			else -- type( prevValue ) == 'string'

				hashCollisonTable = { [ prevValue ] = true }
				hashesContainer.hashToDocId[ hash ] = hashCollisonTable

			end

			hashCollisonTable[ docId ] = true
			hashesContainer.docIdToHash[ docId ] = hash

		end

	end
	
end

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

local function deleteDocIdFromContainer( hashesContainer, docId )

	local newValue = nil
	
	local hash = hashesContainer.docIdToHash[ docId ]
	
	if hash then

		local docIdOrCollisionTable = hashesContainer.hashToDocId[ hash ]

		if type( docIdOrCollisionTable ) == 'table' then

			docIdOrCollisionTable[ docId ] = nil

			if next( docIdOrCollisionTable ) ~= nil then
				newValue = docIdOrCollisionTable
					-- another docId exists, keep the collision table
			end

		elseif docId ~= docIdOrCollisionTable then

			newValue = docIdOrCollisionTable
				-- don't delete hash if docId doesn't match

		end

		hashesContainer.hashToDocId[ hash ] = newValue
		hashesContainer.docIdToHash[ docId ] = nil
	end

end

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

local function getDocIdForHash( hashesContainer, hash, bFetchAll )

	if not hash then
		return nil
	end

	local docId = hashesContainer.hashToDocId[ hash ]

	-- Check to see if this hash has multiple docIds, if so return the first.
	if type( docId ) == 'table' and not bFetchAll then
		docId = next( docId )
	end

	return docId

end

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

local function getHashesListFromContainer( hashesContainer )
	
	local hashesList = {}
	
	for hash in pairs( hashesContainer.hashToDocId ) do
		hashesList[ #hashesList + 1 ] = hash
	end
	
	return hashesList
	
end

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

local function getDocIdsFromHash( hashOfDocsWithoutMaster )

	if type( hashOfDocsWithoutMaster ) == 'string' then
		return { hashOfDocsWithoutMaster }
	end

	local docIdsWithoutMaster = {}

	for k, _ in pairs( hashOfDocsWithoutMaster ) do
		docIdsWithoutMaster[ #docIdsWithoutMaster + 1 ] = k	
	end

	return docIdsWithoutMaster

end

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

local function setDocIdForInContainers( sha256, originalDigest, docId, needToImportMaster )
	
	setDocIdForHash( sha256Hashes, sha256, docId )

	if needToImportMaster then
		setDocIdForHash( sha256Hashes_docsWithoutMaster, sha256, docId )
	end
	
	setDocIdForHash( originalDigests, originalDigest, docId )
	
end

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

local function deleteDocIdFromContainers( docId )

	deleteDocIdFromContainer( sha256Hashes, docId )
	deleteDocIdFromContainer( sha256Hashes_docsWithoutMaster, docId )
	deleteDocIdFromContainer( originalDigests, docId )

end

--------------------------------------------------------------------------------
-- maybeReportSha256Change is called from
-- 	docStore:scannedFilteredMappedChangesFeedSince on a separate worker thread.
--------------------------------------------------------------------------------

function WFCatImportDuplicateFiles.maybeReportSha256Change( 
					hashData, docValue )

	local subtype = docValue.subtype
	if subtype == 'deleted_image' or subtype == 'deleted_video' or docValue._spaceId then
		-- Don't include Trash or space-feed assets in the duplicate-check
		return hashData
	end

	local docId = docValue._id

	local isDeleted = docValue._deleted

	local sha256 = not isDeleted and 
					derefOpt( docValue, 'importSource', 'sha256' )
					
	local originalDigest = derefOpt( docValue, 'importSource', 'originalDigest' )
	
	if originalDigest then
		-- ACR (and Oz API spec) has this value in upper case. But there may be some assets 
		-- on Oz that have a lower case originalDigest if they came in from other clients
		originalDigest = string.upper( originalDigest )
	end


	if not docId or not( isDeleted or sha256 ) then

		-- we don't care about changes unless they are deletion or have
		-- a sha256
		return hashData

	end
	
	hashData.docIdToSha256 = hashData.docIdToSha256 or {}

	if isDeleted then

		hashData.docIdToSha256[ docId ] = nil

		return hashData, "deleted", docId

	end
	
	local currSha256, currNeedToImportMaster =
			unpack( hashData.docIdToSha256[ docId ] or {} )

	local needToImportMaster = docNeedsImportOfMaster( docValue )

	if sha256 == currSha256 and 
		needToImportMaster == currNeedToImportMaster then

		return hashData

	end

	hashData.docIdToSha256[ docId ] = { sha256, needToImportMaster }

	return hashData, "sha256", docId, sha256, originalDigest, needToImportMaster

end

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

local kSha256ChangesFeedDisposed = {}

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

local sha256ChangesFeedReplayForTest

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

local function replaySha256ChangesForTest( ... ) 

	if sha256ChangesFeedReplayForTest then
		sha256ChangesFeedReplayForTest:onNext( ... ) 
	end
end

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

function WFCatImportDuplicateFiles._testOnly_getSha256ChangesFeed()
	
	if not sha256ChangesFeedReplayForTest then
		sha256ChangesFeedReplayForTest = Rx.ReplaySubject( 1 )
	end
	
	return sha256ChangesFeedReplayForTest
				:filter( 
					function( v ) 

						return v ~= kSha256ChangesFeedDisposed 
						
					end )

end

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

local function updateSha256Hashes( changeType, assetId, sha256, originalDigest, needToImportMaster )

	if changeType == "deleted" then

		deleteDocIdFromContainers( assetId )

		return

	end

	setDocIdForInContainers( string.lower( sha256 ), originalDigest, assetId,
							needToImportMaster )

	if not needToImportMaster then

		deleteDocIdFromContainer(
			sha256Hashes_docsWithoutMaster, assetId )

	end

end 

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

local function setupHashesListener( docStore, indexStore )

	if not hashesCloseActionAdded then
		indexStore:addCloseAction( hashesCloseAction() )
		hashesCloseActionAdded = true
	end

	if not sha256ChangesFeedReplayForTest then
		sha256ChangesFeedReplayForTest = Rx.ReplaySubject( 1 )
	end

	local updaterHashesChangesSubscription = docStore
			:scannedFilteredMappedChangesFeed( 
				"WFCatImportDuplicateFiles" , 
				"maybeReportSha256Change", 
				{} )
			:tap( updateSha256Hashes )
			:tap( replaySha256ChangesForTest )
			:subscribe()

	-- create a second disposable that stops the replay updateSha256Hashes is
	-- no longer called
	local sha256ChangesFeedDisposed =
			Rx.Disposable.create(
				function() 

					sha256ChangesFeedReplayForTest
						:onNext( kSha256ChangesFeedDisposed )

				end )

	hashesChangesSubscription = 
		Rx.CompositeDisposable( updaterHashesChangesSubscription,
								sha256ChangesFeedDisposed )

end

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

function WFCatImportDuplicateFiles.addHashesForRecentImage( docStore, indexStore, sha256, originalDigest, docId )

	-- This function is needed because we don't have a way to guarantee our indexstore
	-- will be up to date immediately after setting a doc value.

	if sha256 ~= nil then

		if sha256Hashes == nil then
		
			sha256Hashes = makeHashesContainer()
			sha256Hashes_docsWithoutMaster = makeHashesContainer()
			originalDigests = makeHashesContainer()
		
		end

		setDocIdForInContainers( sha256, originalDigest, docId, false )

	end

end

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

WFCatImportDuplicateFiles.getDocIdsFromHash = getDocIdsFromHash

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

local function loadHashes( docStore, indexStore )

	-- Use hashesLoaded variable as addHashesForRecentImage may create sha256Hashes
	-- before we have a chance to load it from the importSource table.

	if hashesLoaded then
	
		return Observable.just( true )

	else

		if hashesLoader == nil then
		
			setupHashesListener( docStore, indexStore )

			if sha256Hashes == nil then
				
				sha256Hashes = makeHashesContainer()
				sha256Hashes_docsWithoutMaster = makeHashesContainer()
				originalDigests = makeHashesContainer()
				
			end

			hashesLoader = indexStore:queryOnce {
				sql = "SELECT sha256, originalDigest, assetId, needToImportMaster FROM importSource WHERE spaceId IS NULL",
				quiesceUntilUpToDate = true,
			}

			-- make the query complete, sha256Hashes is updated directly afterwards

			:map( function( rows )

				for _, row in ipairs( rows ) do

					local sha256 = row[ 1 ]
					local originalDigest = row[ 2 ]
					local docId = row[ 3 ]
					local needToImportMaster = row[ 4 ] == 1

					if sha256 and docId then
						setDocIdForInContainers( sha256, originalDigest, docId, needToImportMaster )
					end

				end

				hashesLoaded = true
				hashesLoader = nil

				return true

			end )

			:replay( nil, 1 )

			:refCount()

		end

		return hashesLoader

	end

end

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

function WFCatImportDuplicateFiles.findExistingFileBySha256( docStore, indexStore, sha256, bFetchAll )

	return loadHashes( docStore, indexStore )
		
		:map( function()
		
			return getDocIdForHash( sha256Hashes, sha256, bFetchAll ),
				getDocIdForHash( sha256Hashes_docsWithoutMaster, sha256, true )
			
		end )
	
end

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

function WFCatImportDuplicateFiles.findExistingFilesByOriginalDigests( wfCatalog, originalDigestList )

	local docStore = wfCatalog:docStore()
	local indexStore = wfCatalog:indexStore()

	return loadHashes( docStore, indexStore )
		
		:map( function()
			
			local foundHashes = {}
			
			for _, originalDigest in ipairs( originalDigestList ) do
				foundHashes[ originalDigest ] = getDocIdForHash( originalDigests, originalDigest )
			end
			
			return foundHashes
			
		end )

end

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

function WFCatImportDuplicateFiles.findExistingFilesBySha256( wfCatalog, rawImageInfoArray )

	local docStore = wfCatalog:docStore()
	local indexStore = wfCatalog:indexStore()

	return Observable.from( rawImageInfoArray )
		:selectMany( function( image )

			local shaObservable
			if image.sha256 then
				shaObservable = Observable.just( image.sha256 )
			else
				shaObservable = WFCrypto.asyncDigestFile( image.filePath, 'SHA256' )
			end

			return shaObservable
				:flatMapLatest( function( sha256 )

					return WFCatImportDuplicateFiles.findExistingFileBySha256( docStore, indexStore, sha256 )
						:map( function( exists, hasAssetsWithoutMaster )

							exists = exists and not hasAssetsWithoutMaster
								-- Consider it as not imported yet if any of the assets with this sha256 hash
								-- have been imported by Lightroom without a master.

							local ret = {
								id_local = image.id_local,
								sha256 = sha256,
								isDuplicate = exists
									and WFCatImportConstants.kMatchesExistingFile
									or WFCatImportConstants.kNotSimilarToAnyExistingFile,
							}

							return ret

						end )

				end )
		end )
		:toArray()

end

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

function WFCatImportDuplicateFiles.findExistingFiles( wfCatalog, rawImageInfoArray )

	return Observable.from( rawImageInfoArray )

	:selectMany( function( image )

		assert( image.fileName and image.fileSize )

		return wfCatalog:indexStore():queryOnce {
			sql = "SELECT fileName, fileSize FROM importSource WHERE fileName = ? AND fileSize = ? AND spaceId IS NULL",
			image.fileName, image.fileSize
		}

		:map( function( rows )
			if rows and #rows > 0 then
				return { image.id_local, WFCatImportConstants.kMaybeMatchesExistingFile }
			else
				return { image.id_local, WFCatImportConstants.kNotSimilarToAnyExistingFile }
			end
		end )

	end )

	:toArray()

	:map( function( dupInfoPackedWeird )

		local dupInfo = {}

		for _, infoPacket in ipairs( dupInfoPackedWeird ) do
			dupInfo[ infoPacket[ 1 ] ] = infoPacket[ 2 ]
		end

		return dupInfo

	end )

end

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

local function addAssetInfoToSet( fileInfoSet, fileName, fileSize, captureDate )
	
	if not fileInfoSet[ fileName ] then
		fileInfoSet[ fileName ] = {}
	end
	
	local set = fileInfoSet[ fileName ][ fileSize ]
	if not set then
		set = {}
		fileInfoSet[ fileName ][ fileSize ] = set
	end
	
	if not set[ captureDate ] then
		set[ captureDate ] = true
	end
	
end

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

--[[!
	\brief Collect fileName's and fileSize's of all assets in the catalog that don't have the master

	\param wfCatalog the catalog

	\return Rx.Observable( array of tables ) an array of results, one table
		for each fileName. Each table is an array of fileSize's. This format
		would enable a quick lookup for duplicates.
		See WFCatImportDuplicateFiles.isDuplicateFile
--]]
function WFCatImportDuplicateFiles.assetListInfo( wfCatalog )

	-- Return info of all assets arranged by fileName and fileSize (can be used
	-- for a quick, but inaccurate duplicate detection)
	return wfCatalog:indexStore():query {
		sql = [[ SELECT fileName, fileSize, captureDate 
					FROM importSource 
					WHERE needToImportMaster = 0
					AND spaceId IS NULL ]],
		quiesceUntilUpToDate = true,
	}
	:map( function( rows )

		local results = {}
		for _, row in ipairs( rows ) do
			local fileName = row[ 1 ]
			local fileSize = row[ 2 ]
			local captureDate = row[ 3 ]

			if fileName and fileSize and captureDate then
				addAssetInfoToSet( results, fileName, fileSize, captureDate )
			end
		end

		return results

	end )

end

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

--[[!
	\brief Function to return all the hashes of imported files in the catalog.

	\param wfCatalog the catalog

	\return An array of hashes (strings)

--]]

function WFCatImportDuplicateFiles.getAllOriginalDigests( wfCatalog )

	local docStore = wfCatalog:docStore()
	local indexStore = wfCatalog:indexStore()
	
	return loadHashes( docStore, indexStore )
		
		:map( function()
		
			return getHashesListFromContainer( originalDigests )
			
		end )
	
end

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

--[[!
	\brief Function to find an asset that has a master or proxy uploaded on oz.
		This function is used by code to determine if we can copy an asset instead
		of uploading it.

	\param wfCatalog the catalog

	\param sha256 the sha256 for which we want to find an asset that has a
		master or proxy link.

	\param notDocId the id of a document we know that doesn't have a link yet.
		Typically this is an asset that was recently added and is used as a hint
		to eliminate the need to do a query if it is the only asset with the
		sha256.

	\return tuple of of an assetId and boolean.  The assetId is an asset with
		an master or proxy link (meaning it has been uploaded).  The boolean
		will be true if it only has a proxy link.

--]]

function WFCatImportDuplicateFiles.findUploadedAssetWithSha256HintNotDocId( wfCatalog, sha256, notDocId )

	return WFCatImportDuplicateFiles.findExistingFileBySha256( wfCatalog:docStore(), wfCatalog:indexStore(), sha256, true )

	:flatMap( function( assetIds, assetIdsWithoutMaster )

		-- if assetIds is a single enry and it matches the docId we hinted that we don't then there isn't
		-- an existing asset up on oz yet.  This should cover most cases and will be fast.
		if assetIds == notDocId then
			return Observable.just( nil )
		end

		return wfCatalog:indexStore():queryOnce {
			sql = [[SELECT assetId, hasLinkOnOz FROM importSource
				WHERE sha256 = ? AND hasLinkOnOz != 0 AND spaceId IS NULL
				ORDER BY hasLinkOnOz DESC LIMIT 1]],
			sha256,
			quiesceUntilUpToDate = true,
		}

		:map( function( rows )

			if #rows == 0 then
				return nil
			end

			local row = rows[ 1 ]

			local assetId = row[ 1 ]
			local hasLinkOnOz = row[ 2 ]

			return assetId, hasLinkOnOz == kHasProxyLink

		end )

	end )

end

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

return WFCatImportDuplicateFiles
