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

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

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

local Rx = require 'Rx'

local Observable = Rx.Observable

local WFCatAssetInfoCollector = require "WFCatAssetInfoCollector"
local WFCatSyncStatus = require 'WFCatSyncStatus'

local WFCatAssetUtils = require 'WFCatAssetUtils'

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

	-- WF.core.reactive
local WFRxDocValueWatcher = require 'WFRxDocValueWatcher'
local WFRxScheduler = require "WFRxScheduler"

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFLogger = require 'WFLogger'
local KSTableUtils = require 'KSTableUtils'
local WFImageOrientation = require 'WFImageOrientation'

local WFCatSearchByImportDate = require 'WFCatSearchByImportDate'
local WFCatSearchByCaptureDate = require 'WFCatSearchByCaptureDate'
local WFCatFileSpec = require "WFCatFileSpec"
local WFCatPendingUpdates = require "WFCatPendingUpdates"
local WFCatUndo = require "WFCatUndo"
local WFCatSearchByKeywords = require "WFCatSearchByKeywords"
local WFCatSearchByCamera = require "WFCatSearchByCamera"
local WFCatSearchByLocation = require "WFCatSearchByLocation"
local WFCatSearchBySyncStatus = require "WFCatSearchBySyncStatus"
local WFCatSearchByAssetType = require "WFCatSearchByAssetType"
local WFCatSearchByExtension = require "WFCatSearchByExtension"
local WFCatSharedWithYouUtils = require 'WFCatSharedWithYouUtils'
local WFRxBatchedBulkApplyOverlay = require "WFRxBatchedBulkApplyOverlay"
local WFCatTaskManager = require 'WFCatTaskManager'
local WFCatKeywordTask = require 'WFCatKeywordTask'
local WFCatCaptureDateTask = require 'WFCatCaptureDateTask'
local WFCatUtils = require 'WFCatUtils'
local WFTableMerge = require 'WFTableMerge'
local WFCatCollaborationActivity = require "WFCatCollaborationActivity"


local derefOpt = assert( KSTableUtils.derefOpt )
local check = assert( WFArguments.check )
local logger = WFLogger( "WFCatAsset" )

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

local kMixed = "mixed"

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

local collectedRatingsMap = {}
local collectedFlagsMap = {}

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

local WFCatAsset = {}

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

local function collectKeywordsFromDocValue( docValue )
	local keywords = {}
	local keywordSet = derefOpt( docValue, 'xmp', 'dc', 'subject' ) or {}

	for keyword, _ in pairs( keywordSet ) do
		table.insert( keywords, keyword )
	end
	
	return keywords
end

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

local function hasValidLocalId( docValue )
	
	local isValid = not not docValue._localId
	local isDeleted = not not docValue._deleted
	
	if not isDeleted and not isValid then
		logger:errorf( 'valid localId required, docValue: %s', docValue )
	end
	
	return isValid
end

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

local function canBeDeletedCountConsolidator( assetId, key, value, consolidatedValues )

	local consolidatedValue = consolidatedValues[ key ] or 0
	
	if value then
		consolidatedValue = consolidatedValue + 1
	end
	
	consolidatedValues[ key ] = consolidatedValue
end

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

local function getRatingFromAnyUser( docValue )
	local ratingsSection = docValue.ratings
	if ratingsSection then
		local _, ratingsTable = next( ratingsSection )
		return ratingsTable and ratingsTable.rating
	end
end

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

local function getFlagFromAnyUser( docValue )
	local flagsSection = docValue.reviews
	if flagsSection then
		local _, flagsTable = next( flagsSection )
		return flagsTable and flagsTable.flag
	end
end

--------------------------------------------------------------------------------
-- Uses the docStore to collect ongoing info about an array of assets.
-- (see collectInfoForAssetsFromDocStore for more documentation)
--
-- this function will coalesce multiple results into a single result
--
-- Lower latency than collectInfoForAssetsWithIndexStore when showing value changes, but 
-- uses more memory.

local function collectCoalescedInfoForAssetsFromDocStore( wfCatalog, params )

	local assetIds = params.assetIds
	assert( type( assetIds ) == 'table', 'params assetIds must be an array of integer assetIds' )
	
	if #assetIds == 0 then
		return Observable.just( {} )
	end
	
	local collector = WFCatAssetInfoCollector.makeCollector()

	local collectAssetInfo = function( docValue )
		
		local assetId = docValue._localId
		
		if params.ratings then	
			local rating = getRatingFromAnyUser( docValue )
			collector:consolidateValue( assetId, 'rating', rating, nil )
		end
		
		if params.flags then
			local flag = getFlagFromAnyUser( docValue )
			collector:consolidateValue( assetId, 'flag', flag, nil )
		end

		if params.keywords then
			local keywords = collectKeywordsFromDocValue( docValue )
			
			collector:accumulateValues( assetId, 'keywords', 'partiallyAppliedKeywords', keywords, true )
		end
	
		-- Note that we currently only return auto-tags if exactly one photo is selected to avoid performance cost
		-- of large selections. Auto tags are only made visible for debugging purposes, so this limitation is acceptable.

		if params.autoTags and #assetIds == 1 then

			local autoTags = {}
			local autoTagsValue = derefOpt( docValue,'autoTags', 'tags' ) or {}

			for tag, confidence in pairs( autoTagsValue ) do
				table.insert( autoTags , { tag = tag, confidence = confidence } )
			end

			-- (no need to accumulate since we are restricted to a single photo)
			collector:consolidateValue( assetId, 'autoTags', autoTags )
		end
		
		if params.isUnsupportedFormat then
			local isUnsupported = WFCatUtils.isUnsupportedFormat( docValue )
			collector:consolidateValue( assetId, 'isUnsupportedFormat', isUnsupported, true )
		end

		if params.isEdited then
			local isEdited = WFCatUtils.isEdited( docValue )
			collector:consolidateValue( assetId, 'isEdited', isEdited )
		end

		if params.needsUpdateAISettings then
			local needsUpdateAISettings = WFCatUtils.needsUpdateAISettings( docValue )
			collector:consolidateValue( assetId, 'needsUpdateAISettings', needsUpdateAISettings )
		end
		
		if params.canBeDeleted then
			local canBeDeleted = WFCatSharedWithYouUtils.canDeleteAsset( docValue )
			collector:consolidateValue( assetId, 'canBeDeleted', canBeDeleted, false, canBeDeletedCountConsolidator )
		end
		
	end
	
	local function getCollectedValues()
		return collector:getCollectedValues( #assetIds )
	end

	local scheduler = WFRxScheduler.sharedScheduler()

	local n = 0
	
	-- Give an immediate result for every N results, where N is the number of assets being observed.
	-- This improves the responsiveness of the the common case of applying a change to the entire selection
	-- (ex: rating) while still enforcing a throttled rate of updates.

	local function throttleUntilN()	

		local t = 0.1

		n = n + 1

		if n % #assetIds == 0 then
			t = 0
		end

		return Rx.Observable.timer( t, scheduler )
	end

	return Observable.of( assetIds )
		:call( WFRxDocValueWatcher.make, wfCatalog:docStore(), "collectInfo" )
		:filter( hasValidLocalId )
		:map( collectAssetInfo )
		:debounce( throttleUntilN, scheduler )
		:map( getCollectedValues )
end

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

local function createCollectAssetInfoFn( params, ozUserId, assetInfoMap )

	return function( docValue )
		
		local currentAsset = {}
		
		local assetId = docValue._localId
		
		if params.includeIds then
			currentAsset.id = assetId
		end
		
		if params.ratings then
			local rating = getRatingFromAnyUser( docValue )
			currentAsset.rating = rating or 0
			collectedRatingsMap[ assetId ] = currentAsset.rating
		end
		
		if params.flags then
			local flag = getFlagFromAnyUser( docValue )
			currentAsset.flag = flag or 'unflagged'
			collectedFlagsMap[ assetId ] = currentAsset.flag
		end

		if params.keywords then
			local keywords = collectKeywordsFromDocValue( docValue )
			currentAsset.keywords = keywords
		end
		
		if params.captureDateTime then
			local dateTime = derefOpt( docValue, 'captureDate' )
			currentAsset.captureDateTime = dateTime
		end
		
		if params.isEdited then
			local isEdited = WFCatUtils.isEdited( docValue )
			currentAsset.isEdited = isEdited
		end

		if params.needsUpdateAISettings then
			local needsUpdateAISettings = WFCatUtils.needsUpdateAISettings( docValue )
			currentAsset.needsUpdateAISettings = needsUpdateAISettings
		end
		
		if params.dateTimeOriginal then
			local dateTime = derefOpt( docValue, 'xmp', 'exif', 'DateTimeOriginal' )
			currentAsset.dateTimeOriginal = dateTime
		end
		
		if params.dateCreated then
			local dateTime = derefOpt( docValue, 'xmp', 'photoshop', 'DateCreated' )
			currentAsset.dateCreated = dateTime
		end
		
		if params.createDate then
			local dateTime = derefOpt( docValue, 'xmp', 'xmp', 'CreateDate' )
			currentAsset.createDate = dateTime
		end
		
		if params.temporal then
			local temporal = derefOpt( docValue, 'develop', 'temporal' )
			currentAsset.temporal = temporal
		end
		
				
		if params.video then
			local videoData = derefOpt( docValue, 'video' ) or {}
			currentAsset.video = videoData
		end
		
		if params.isUnsupportedFormat then
			local isUnsupported = WFCatUtils.isUnsupportedFormat( docValue )
			currentAsset.isUnsupportedFormat = isUnsupported
		end
		
		if params.importedByUser then
			local importedByUser = derefOpt( docValue, '_serverProvidedMetadata', 'importedByUser', 'id' )
			if not importedByUser and docValue._spaceId then
				importedByUser = ozUserId
			end
			currentAsset.importedByUser = importedByUser
		end
		
		if params.spaceInfo then
			local spaceId = derefOpt( docValue, '_spaceId' )
			currentAsset.spaceId = spaceId
		end
		
		if params.canBeDeleted then
			currentAsset.canBeDeleted = WFCatSharedWithYouUtils.canDeleteAsset( docValue )
		end
		
		local aux = derefOpt( docValue, 'xmp', 'aux' )

		if aux then		
			if params.enhanceDetails then
				currentAsset.enhanceDetails = aux.EnhanceDetailsAlreadyApplied
			end
				
			if params.mergeDetails then
				local isHDR = aux.IsMergedHDR
				local isPano = aux.IsMergedPanorama
				local mergeDetails
				if isHDR and isPano then
					mergeDetails = "hdrpano"
				elseif isHDR then	
					mergeDetails = "hdr"
				elseif isPano then
					mergeDetails = "pano"
				end
				currentAsset.mergeDetails = mergeDetails			
			end
		end

		assetInfoMap[ assetId ] = currentAsset
		
	end
	
end 

--------------------------------------------------------------------------------
-- Uses the docStore to collect ongoing info about an array of assets.
-- (see collectInfoForAssetsFromDocStore for more documentation)
--
-- this function will return an array of results representing each asset
--
-- Lower latency than collectInfoForAssetsWithIndexStore when showing value changes, but 
-- uses more memory.

local function collectRawInfoForAssetsFromDocStore( wfCatalog, params )

	local assetIds = params.assetIds
	assert( type( assetIds ) == 'table', 'params assetIds must be an array of integer assetIds' )
	
	if #assetIds == 0 then
		return Observable.just( {} )
	end
	
	local assetInfoMap = {}
	
	local collectAssetInfo = createCollectAssetInfoFn( params, wfCatalog:ozUserId(), assetInfoMap )
	
	local function getCollectedValues()
		return assetInfoMap
	end

	local scheduler = WFRxScheduler.sharedScheduler()

	local n = 0
	
	-- Give an immediate result for every N results, where N is the number of assets being observed.
	-- This improves the responsiveness of the the common case of applying a change to the entire selection
	-- (ex: rating) while still enforcing a throttled rate of updates.

	local function throttleUntilN()	

		local t = 0.1

		n = n + 1

		if n % #assetIds == 0 then
			t = 0
		end

		return Rx.Observable.timer( t, scheduler )
	end

	return Observable.of( assetIds )
		:call( WFRxDocValueWatcher.make, wfCatalog:docStore(), "collectRawInfo" )
		:filter( hasValidLocalId )
		:map( collectAssetInfo )
		:debounce( throttleUntilN, scheduler )
		:map( getCollectedValues )
end

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

local function currentAssetInfoFromDocStore( wfCatalog, params )
	
	local assetIds = params.assetIds
	assert( type( assetIds ) == 'table', 'params assetIds must be an array of integer assetIds' )
	
	if #assetIds == 0 then
		return Observable.just( {} )
	end
	
	local assetInfoMap = {}
	
	local collectAssetInfo = createCollectAssetInfoFn( params, wfCatalog:ozUserId(), assetInfoMap )
	
	local function getCollectedValues()
		return assetInfoMap
	end

	return wfCatalog:docStore():bulkDocValue( assetIds )
		:map( function( docValues )
		
			for i, docValue in ipairs( docValues ) do
				
				collectAssetInfo( docValue )
			
			end
			
		end )
		:map( getCollectedValues )

end

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

local function addChangedAtInfoFn( time )

	return function( docValue )
	
		if docValue then
			
			docValue.changedAtTime = time
			docValue.changedOnDevice = require 'WFDeviceInfo'.deviceName()
		
		end
		
		return docValue
	end
end

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

local function saveAssetInfoUndoState( wfCatalog, assetIds, undoTokenCallback, field, collectedValueMap )

	local values = {}

	local function saveState()

		local undoStateToken = WFCatUndo.saveUndoState {
			ids = assetIds,
			[ field ] = values,
		}

		undoTokenCallback( undoStateToken )

		return Observable.from( assetIds )
	end

	local assetIdsWithUnknownValue = {}
	
	for _, id in ipairs( assetIds ) do
		local value = collectedValueMap[ id ]
	
		if value == nil then			
			table.insert( assetIdsWithUnknownValue, id )
		else
			values[ id ] = value
		end
	end

	logger:trace( "number of assets with unknown values:", #assetIdsWithUnknownValue )

	if #assetIdsWithUnknownValue > 0 then
	
		local params = {
			assetIds = assetIdsWithUnknownValue,
			[ field ] = true,
		}

		logger:trace( "collecting info..." )

		return currentAssetInfoFromDocStore( wfCatalog, params )
			:flatMapLatest( function( infos )

				logger:trace( "...done collecting info" )

				for _, id in ipairs( assetIdsWithUnknownValue ) do
					local value = collectedValueMap[ id ]
					--assert( value ~= nil )						
					if value ~= nil then
						values[ id ] = value
					end
				end

				return saveState()

			end )
	end

	return saveState()
	
end

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

local function saveRatingUndoState( wfCatalog, assetIds, undoTokenCallback )	
	return saveAssetInfoUndoState( wfCatalog, assetIds, undoTokenCallback, "ratings", collectedRatingsMap )
end

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

local function saveFlagUndoState( wfCatalog, assetIds, undoTokenCallback )	
	return saveAssetInfoUndoState( wfCatalog, assetIds, undoTokenCallback, "flags", collectedFlagsMap )
end

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

--[[!
	Update the rating on a list of assetIds.

	\param wfCatalog the catalog
	\param assetIds an array of local ids
	\param newRating the new rating to assign to all of the ids
	\param optUndoStateToken (optional, number) Token identifying a WFCatUndo state to restore. If provided,
		then 'assetIds' and 'newRating' args are not used.

	\return RxObservable( result ) sends onCompleted when done with a table containing an undoToken.
--]]

function WFCatAsset.setRatings( wfCatalog, assetIds, newRating, optUndoStateToken )
	
	check( 'WFCatalog', '?table', '?number', '?number' )
	
	local perImageValues = {}
	local undoToken

	local assetIdsObservable
	
	if optUndoStateToken then
		local undoState = WFCatUndo.getUndoState( optUndoStateToken )
		assetIds = assert( undoState.ids, "imageIds required in history state" )
		perImageValues = assert( undoState.ratings, "settings required in history state" )

		assetIdsObservable = Observable.from( assetIds )
	else
		assert( assetIds, "assetIds required" )
	
		assetIdsObservable = saveRatingUndoState( wfCatalog, assetIds, function( token )
			undoToken = token
		end )
	end

	local userId = wfCatalog:ozUserId()
	local ratingTimestamp = KSDate.timeInIso8601Format()

	return assetIdsObservable
		:map( function( assetId )
		
			local imageValue = perImageValues[ assetId ] or newRating or 0
		
			local ratingT = {
				-- Oz does not want removed ratings to be deleted, they should instead 
				-- cause the rating to be set to 0 and include the timestamp when this 
				-- action was taken.
				rating = imageValue,
				date = ratingTimestamp
			}
			
			local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( assetId )
			
			baseOverlay.ratings = {
				[ userId ] = ratingT
			}
			
			return baseOverlay
			
		end )
		:map( addChangedAtInfoFn( KSDate.timeInIso8601Format( nil, false ) ) )
		:toArray()
		:flatMapLatest( function( overlays )
			logger:trace( "applying overlay..." )
			return WFRxBatchedBulkApplyOverlay( "set ratings", wfCatalog:docStore(), overlays )
		end )
		:map( function( result )
			logger:trace( "returning token:", undoToken )
			return {
				undoToken = undoToken
			}
		end )
end

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

--[[!
	Update the filename on an assetIds.
	\param wfCatalog the catalog
	\param assetIds represents ozAssetIds
	\param newFilenames the new rating to assign to the id
	\return RxObservable( result ) sends onCompleted when done with a table containing an undoToken.
--]]

function WFCatAsset.setFilename( wfCatalog, assetIds, newFileNames )

	check( 'WFCatalog', 'table', 'table' )

	assert( #assetIds > 0, "imageIds must contain at least one image ID" )
	
	assert( #assetIds == #newFileNames, "imageIds table size must be same as newFileNames table size" )

	local overlays = {} 
	
	for i, assetId in ipairs( assetIds ) do 
		local baseDocValue = WFCatUtils.createBaseOverlayWithUserCreatedFields( assetId ) 
		KSTableUtils.setNested( baseDocValue, newFileNames[ i ], 'importSource', 'fileName' ) 
		overlays[ #overlays + 1 ] = baseDocValue 
	end 
	
	return WFRxBatchedBulkApplyOverlay( "updateAssetFileNames", wfCatalog:docStore(), overlays ) 

end


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

local kAllowedFlags = {
	pick = true,
	unflagged = true,
	reject = true,
}

--[[!
	Update the flags on a list of assetIds.

	\param wfCatalog the catalog
	\param assetIds an array of local ids
	\param (string) newFlag the new rating to assign to all of the ids, one of: "pick", "unflagged", or "reject".
	\param (optional, string) date an optional ISO 8601 date (must be in UTC
		time zone) to use.  If omitted then the current time is used.
	\param optUndoStateToken (optional, number) Token identifying a WFCatUndo state to restore. If provided,
		then 'assetIds' and 'newRating' args are not used.

	\return RxObservable( result ) sends onCompleted when done with a table containing an undoToken.
--]]

function WFCatAsset.setFlags( wfCatalog, assetIds, newFlag, date, optUndoStateToken )
	check( "WFCatalog", "?table", "?string", "?string" )

	logger:trace( "WFCatAsset.setFlags:", newFlag )

	if newFlag then
		assert( kAllowedFlags[ newFlag ], 'Invalid Flag value.  Must be one of pick, unflagged, or reject' )
	end

	local perImageValues = {}
	local undoToken

	local assetIdsObservable
	
	if optUndoStateToken then
		local undoState = WFCatUndo.getUndoState( optUndoStateToken )
		assetIds = assert( undoState.ids, "imageIds expected in history state" )
		perImageValues = assert( undoState.flags, "flags expected in history state" )

		assetIdsObservable = Observable.from( assetIds )
	else
		assert( assetIds, "assetIds required" )
		assert( newFlag, "newFlag required" )
	
		assetIdsObservable = saveFlagUndoState( wfCatalog, assetIds, function( token )
			undoToken = token
		end )
	end

	date = date or KSDate.timeInIso8601Format( nil, false )
	local userId = wfCatalog:ozUserId()

	return assetIdsObservable
		:map( function( assetId )

			local imageValue = perImageValues[ assetId ] or newFlag
		
			local flagT = {
				flag = imageValue,
				date = date,
			}
			if not imageValue then
				flagT._delete = true
			end
			
			local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( assetId )
			
			baseOverlay.reviews = {
				[ userId ] = flagT,
			}
			
			return baseOverlay
			
		end )
		:map( addChangedAtInfoFn( date ) )
		:toArray()
		:flatMapLatest( function( overlays )
			logger:trace( "applying overlay..." )
			return WFRxBatchedBulkApplyOverlay( "set flags", wfCatalog:docStore(), overlays )
		end )
		:map( function( result )
			logger:trace( "returning token:", undoToken )
			return {
				undoToken = undoToken
			}
		end )
end

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

function WFCatAsset.toggleFlags( wfCatalog, assetIds, flag, date, optUndoStateToken )

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

	logger:trace( "WFCatAsset.toggleFlags:", flag )

	local alternate

	if flag then
		assert( kAllowedFlags[ flag ], 'Invalid Flag value.  Must be one of pick, unflagged, or reject' )
	end

	local perImageValues = {}
	local undoToken

	local assetIdsObservable
	
	if optUndoStateToken then
		local undoState = WFCatUndo.getUndoState( optUndoStateToken )
		assetIds = assert( undoState.ids, "imageIds expected in history state" )
		perImageValues = assert( undoState.flags, "flags expected in history state" )

		assetIdsObservable = Observable.from( assetIds )
	else
		assert( assetIds, "assetIds required" )
		assert( flag, "newFlag required" )
	
		assetIdsObservable = saveFlagUndoState( wfCatalog, assetIds, function( token )
			undoToken = token
			
			local undoState = WFCatUndo.getUndoState( token )
			local multipleValues = false
			
			for assetId, value in pairs( undoState.flags ) do
				if value ~= flag then
					multipleValues = true
					break
				end
			end
			
			if not multipleValues then
				flag = 'unflagged'
				alternate = true
			end
		end )
	end

	date = date or KSDate.timeInIso8601Format( nil, false )
	local userId = wfCatalog:ozUserId()

	return assetIdsObservable
		:map( function( assetId )

			local imageValue = perImageValues[ assetId ] or flag
		
			local flagT = {
				flag = imageValue,
				date = date,
			}
			if not imageValue then
				flagT._delete = true
			end
			
			local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( assetId )
			
			baseOverlay.reviews = {
				[ userId ] = flagT,
			}
			
			return baseOverlay
			
		end )
		:map( addChangedAtInfoFn( date ) )
		:toArray()
		:flatMapLatest( function( overlays )
			logger:trace( "applying overlay..." )
			return WFRxBatchedBulkApplyOverlay( "toggle flags", wfCatalog:docStore(), overlays )
		end )
		:map( function( result )
			logger:trace( "returning token:", undoToken )
			return {
				undoToken = undoToken,
				alternate = alternate,
			}
		end )
end

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

function WFCatAsset.adjustFlags( wfCatalog, assetIds, direction, date, optUndoStateToken )
	check( "WFCatalog", "?table", "?number", "?string" )

	logger:trace( "WFCatAsset.adjustFlags:", direction )

	if direction then
		assert( direction == -1 or direction == 1, 'Invalid direction value.  Must be 1 (increase) or -1 (decrease)' )
	end

	local perImageValues = {}
	local undoToken

	local assetIdsObservable
	
	if optUndoStateToken then
		local undoState = WFCatUndo.getUndoState( optUndoStateToken )
		assetIds = assert( undoState.ids, "imageIds expected in history state" )
		perImageValues = assert( undoState.flags, "flags expected in history state" )

		assetIdsObservable = Observable.from( assetIds )
	else
		assert( assetIds, "assetIds required" )
		assert( direction, "direction required" )
	
		assetIdsObservable = saveFlagUndoState( wfCatalog, assetIds, function( token )
			undoToken = token
			
			local undoState = WFCatUndo.getUndoState( token )
			perImageValues = undoState.flags
		end )
	end

	date = date or KSDate.timeInIso8601Format( nil, false )
	local userId = wfCatalog:ozUserId()

	return assetIdsObservable
		:map( function( assetId )

			local imageValue = perImageValues[ assetId ]

			if not optUndoStateToken then
				assert( direction, "direction required" )
				
				if direction > 0 then
					if imageValue == 'reject' then
						imageValue = 'unflagged'
					else
						imageValue = 'pick'
					end
				else
					if imageValue == 'pick' then
						imageValue = 'unflagged'
					else
						imageValue = 'reject'
					end
				end
			end
			
			local flagT = {
				flag = imageValue,
				date = date,
			}
			if not imageValue then
				flagT._delete = true
			end

			local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( assetId )
			
			baseOverlay.reviews = {
				[ userId ] = flagT,
			}
			
			return baseOverlay
			
		end )
		:map( addChangedAtInfoFn( date ) )
		:toArray()
		:flatMapLatest( function( overlays )
			logger:trace( "applying overlay..." )
			return WFRxBatchedBulkApplyOverlay( "adjust flags", wfCatalog:docStore(), overlays )
		end )
		:map( function( result )
			logger:trace( "returning token:", undoToken )
			return {
				undoToken = undoToken
			}
		end )
end

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

--[[!
	Update the orientation (via rotate) on a list of assetIds.

	\param wfCatalog the catalog
	\param assetIds an array of local ids
	\param rotation a string indicating rotation direction
	\param optUndoStateToken (optional, number) Token identifying a WFCatUndo state to restore. If provided,
		then 'assetIds' and 'rotation' args are not used.

	\return RxObservable( result ) sends onCompleted when done with a table containing an undoToken.
--]]

function WFCatAsset.rotateOrFlipImages( wfCatalog, assetIds, rotation, flip, optUndoStateToken )

	local orientationArray
	local undoVerb
	
	if optUndoStateToken then
		-- we're doing undo, fetch orientation values from undo mgr
		local undoState = WFCatUndo.getUndoState( optUndoStateToken )
		assetIds = assert( undoState.ids, "imageIds expected in history state" )
		orientationArray = assert( undoState.orientations, "orientations expected in history state" )
		undoVerb = undoState.verb
	else
		assert( assetIds, "assetIds required" )
		assert( rotation or flip, "rotation or flip required" )
	
		orientationArray = {}
	end

	local docStore = wfCatalog:docStore()

	local date = KSDate.timeInIso8601Format( nil, false )

	return docStore:bulkDocValue( assetIds )

		:flatMapLatest( function( docValueArray ) ---

			local arrayOfOverlays = {}
					
			for _, docValue in ipairs( docValueArray or {} ) do			

				local assetId = docValue._id
				
				local assetSubtype = derefOpt( docValue, 'subtype' )
				
				local ozUserOrientation, ozCroppedWidth, ozCroppedHeight
				
				if assetSubtype == "video" then
					local ozVFUserOrientation = derefOpt( docValue, 'develop', 'adjustments', 'geometric', 'orientation') or "normal"
					ozUserOrientation = WFImageOrientation.vfOrientationToTiffOrientation( ozVFUserOrientation )
					-- This needs to be updated once we add video crop. For now, use the original dimensions.
					ozCroppedWidth = derefOpt( docValue, 'develop', 'croppedWidth' )
					ozCroppedHeight = derefOpt( docValue, 'develop', 'croppedHeight' )
				else
				
					ozUserOrientation = derefOpt( docValue, 'develop', 'userOrientation' ) or 1
					ozCroppedWidth = derefOpt( docValue, 'develop', 'croppedWidth' )
					ozCroppedHeight = derefOpt( docValue, 'develop', 'croppedHeight' )
					
				end
				
				if optUndoStateToken then
					-- fetch orientation value from undo token
					ozUserOrientation = orientationArray[ assetId ] or 1
				else
					-- tuck away current orientation for undo
					orientationArray[ assetId ] = ozUserOrientation
					
					-- compute new orienation
					local agOrientation = WFImageOrientation.tiffOrientationToAgOrientation( ozUserOrientation )
					if rotation then
						agOrientation = WFImageOrientation.rotate( agOrientation, rotation )
					elseif flip then
						agOrientation = WFImageOrientation.flip( agOrientation, flip )
					end
					ozUserOrientation = WFImageOrientation.AgOrientationTOTiffOrientation( agOrientation )
				end
				
				if rotation or undoVerb == "rotation" then
					-- because we're rotating 90 degrees, width and height simply swap. lazy
					ozCroppedWidth, ozCroppedHeight = ozCroppedHeight, ozCroppedWidth
				end


				local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( assetId )
				
				if assetSubtype == "video" then
					
					local vfOrientation = WFImageOrientation.tiffOrientationToVFOrientation( ozUserOrientation )
					local vfAdjustments = derefOpt( docValue, 'develop', 'adjustments' ) or {}
					local vfGeometric = derefOpt( docValue, 'develop', 'adjustments', 'geometric' ) or {}
					
					-- Set to the same version or if a version does not exist, set to the current version
					local currentProcessVersion = derefOpt( docValue, 'develop', 'processVersion' ) or 1
					local currentVersion = derefOpt( docValue, 'develop', 'version' ) or 1
					
					vfGeometric.orientation = vfOrientation
					vfAdjustments.geometric = vfGeometric
					
					-- userUpdated in baseOverlay.develop is not supported for video.
					-- but it is supported in baseOverlay.adjustments instead.
					vfAdjustments.userUpdated = date
					
					baseOverlay.develop = {
						processVersion = currentProcessVersion,
						version = currentVersion,
						adjustments = vfAdjustments,
						croppedWidth = ozCroppedWidth,
						croppedHeight = ozCroppedHeight,
					}

				else
				
					baseOverlay.develop = {
						processingModel = 'lightroom',
						userUpdated = date,
						userOrientation = ozUserOrientation,
						croppedWidth = ozCroppedWidth,
						croppedHeight = ozCroppedHeight,
					}
					
				end

				table.insert( arrayOfOverlays, baseOverlay )
			
			end
			
			local undoStateToken
			
			if optUndoStateToken == nil then
				-- this is not an undo, construct an undo state, register it, and prepare to return token
				undoStateToken = WFCatUndo.saveUndoState {
					ids = assetIds,
					verb = rotation and "rotation" or "flip",
					orientations = orientationArray,
				}
			end
			
			return WFRxBatchedBulkApplyOverlay( "rotate or flip", docStore, arrayOfOverlays )
			:map( function( result )
				logger:trace( "returning token:", undoStateToken )
				return {
					undoToken = undoStateToken
				}
			end )
			
		end )

end

--------------------------------------------------------------------------------
--[[!
	\brief copy Metadata and set Enhance Details For Images

	\param wfCatalog (WFCatalog) catalog
	\param imageIdMap. Ex: source asset id = destination asset id

	\return Observable which fires only once.
--]]

function WFCatAsset.copyMetadataAndSetEnhanceDetailsForImages( wfCatalog, imageIdMap )
	check( 'WFCatalog', 'table' )
	
	local assetIds = {}
	for k, v in pairs( imageIdMap ) do
		assetIds[#assetIds + 1] = k
	end
	
	local docStore = wfCatalog:docStore()
	
	return docStore:bulkDocValue( assetIds )
		:flatMap( function( docValueArray )
			local overlays = {}
			for _,docValue in ipairs( docValueArray ) do
				local xmp = derefOpt( docValue, 'xmp' ) or {}
				
				-- exclude 'ModifyDate' from being copied over from
				-- the original image. We want ModifyDate to be the 
				-- date which ACR provides in the DNG's XMP which 
				-- matches the creation date of the DNG.
				KSTableUtils.setNested( xmp, nil, 'xmp', 'ModifyDate' )
				
				local xmpNND = {
					aux = {
						EnhanceDetailsAlreadyApplied = true,
					}
				}
				
				WFTableMerge.applyOverlay( xmp, xmpNND )
				
				local reviews = derefOpt( docValue, 'reviews' )
				local ratings = derefOpt( docValue, 'ratings' )
				local location = derefOpt( docValue, 'location' )
				local captureDate = derefOpt( docValue, 'captureDate' )
				
				local overlay = {
					_id = imageIdMap[ docValue._localId ],
					xmp = xmp,
					reviews = reviews,
					ratings = ratings,
					location = location,
					captureDate = captureDate,
				}
				
				overlays[#overlays + 1] = overlay
			end
			
			if #overlays > 0 then
				return WFRxBatchedBulkApplyOverlay( "copyMetadataAndSetEnhanceDetailsForImages", wfCatalog:docStore(), overlays )
			else
				return Observable.just( {} )
			end
		end )
		:take( 1 ) 
end

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

function WFCatAsset.setKeywords( wfCatalog, assetIds, keywords, delete, optUndoStateToken )
	
	local task = WFCatKeywordTask.createKeywordTask( wfCatalog, assetIds, keywords, delete, optUndoStateToken )
	
	local manager = WFCatTaskManager.getCatTaskManager( wfCatalog )
	manager:addToQueue( task )
	
	return task:taskObservable()
	
end

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

function WFCatAsset.editCaptureDate( wfCatalog, assetIds, timeDelta, newDate, optUndoStateToken )
	
	local task = WFCatCaptureDateTask.createCaptureDateTask( wfCatalog, assetIds, timeDelta, newDate, optUndoStateToken )
	
	local manager = WFCatTaskManager.getCatTaskManager( wfCatalog )
	manager:addToQueue( task )
	
	return task:taskObservable()
	
end

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

local function applyPendingUpdatesForKeyword( assetResultForField, valueKey, update, translateFieldValueFn )
	-- Check if value is present
	local index = -1
	for i, keyword in ipairs( assetResultForField ) do
		if valueKey == keyword then
			index = i
			break
		end
	end
	
	local isAlreadyApplied
	
	-- Handle deletion
	if update.isDelete then
		if index == -1 then
			isAlreadyApplied = true
		else
			isAlreadyApplied = false
			table.remove( assetResultForField, index )
		end
		
		return isAlreadyApplied, assetResultForField
	end
	
	-- Handle addition
	if index ~=-1 then
		-- Already present
		isAlreadyApplied = true
	else
		isAlreadyApplied = false
		local prevFieldValue = assetResultForField

		assetResultForField = 
			translateFieldValueFn( update.value, prevFieldValue )
	end
	
	return isAlreadyApplied, assetResultForField
end

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

local function translateKeywords( value, oldValue )
	local list = oldValue or {}
	list[ #list + 1 ] = value
	return list
end

--------------------------------------------------------------------------------
-- Inorder to enable update tracking and addition, 
-- implement and set the values in the below
-- table for each field. You will need to add
-- to the pending update when any of the set
-- api is called.

local kFieldPendingAddDeleteFn = {
	keywords = applyPendingUpdatesForKeyword
}

local kTranslateFieldValueFn = {
	keywords = translateKeywords
}

local kUpdateEnabledFields = {
	keywords = true
}

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

local function applyPendingUpdatesForAssetForField( assetId, fieldKey, assetIdResult )
	-- Check for pending updates
		
	-- Iterating over the pending list, assuming that the
	-- pending list will normally smaller than the assetlist.
	
	-- assetIdResults will contain all the assetIds which are
	-- queried for, even if it doesnt have a value
	
	local translateFieldValue = kTranslateFieldValueFn[ fieldKey ]
	
	-- Apply the update
	local updates = WFCatPendingUpdates.getPendingUpdatesForAsset( assetId, fieldKey ) or {}
	
	for valueKey, update in pairs( updates ) do
		local isAlreadyApplied, assetResultForField = 
			kFieldPendingAddDeleteFn[ fieldKey ]( assetIdResult[ fieldKey ], valueKey, update, translateFieldValue )
		assetIdResult[ fieldKey ] = assetResultForField
		
		if isAlreadyApplied then
			-- Remove from pending updates
			WFCatPendingUpdates.removeFromPendingUpdates( assetId, fieldKey, valueKey )
		end
	end

end

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

local function makeTranslateWithDefaultValueFormatter( 
				field, translateFieldValue, defaultFieldValue )

	check( "string", "function" )

	-- Given an array of { docId, maybeRawValue } from a SQL query, it emits a
	-- lua table:
	--
	-- {
	--   	[ docId1 ] = { [ field ] = someDerivedValue1 }
	--		[ docId2 ] = { [ field ] = someDerivedValue2 }
	--		...
	-- }
	return function( arrayOfArrayResults )
	
		local assetIdResults = {}
	
		for _, resultArray in ipairs( arrayOfArrayResults ) do
		
			local docId = resultArray[ 1 ]

			if type( docId ) ~= "string" and type( docId ) ~= "number" then

				logger:warnf( "Missing expected docId (it had actual type '%s') when formatting the result of %s", type( docId ), field )

			else

				 assetIdResults[ docId ] = assetIdResults[ docId ] or {}

				local newFieldValue = resultArray[ 2 ] or defaultFieldValue

				local prevFieldValue = assetIdResults[ docId ][ field ]

				assetIdResults[ docId ][ field ] = 
					translateFieldValue( newFieldValue, prevFieldValue )

			end

		end
		
		return assetIdResults
		
	end
	
end

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

local function justReturnFirst( v )

	return v

end

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

local function makeDefaultValueFormatter( field, defaultValue )

	check( "string" )

	return makeTranslateWithDefaultValueFormatter( 
			field, justReturnFirst, defaultValue )
	
end

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

local function makeTranslaterFormatter( field, translateFieldValue )

	check( "string", "function" )

	return makeTranslateWithDefaultValueFormatter( 
			field, translateFieldValue, nil )
	
end

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

local function coalesceValues( assetInfos )

	-- walk assetinfo array and return one coalesced set of values to represent
	-- everything (using "mixed" wherever values do not coincide)
	
	local avgRating, avgFlag, avgStatus

	for _, assetInfo in ipairs( assetInfos ) do
		local rating = assetInfo.rating
		local flag = assetInfo.flag
		local status = assetInfo.status
		
		if avgRating == nil then
			avgRating = rating
		elseif avgRating ~= rating then
			avgRating = kMixed
			if avgFlag == kMixed and avgStatus == kMixed then
				break
			end
		end
		
		if avgFlag == nil then
			avgFlag = flag
		elseif avgFlag ~= flag then
			avgFlag = kMixed
			if avgRating == kMixed and avgStatus == kMixed then
				break
			end
		end
		
		if avgStatus == nil then
			avgStatus = status
		elseif avgStatus ~= status then
			avgStatus = kMixed
			if avgRating == kMixed and avgFlag == kMixed then
				break
			end
		end
	end

	return {
		rating = avgRating,
		flag = avgFlag,
		status = avgStatus,
		numberOfAssets = #assetInfos
	}
end

--------------------------------------------------------------------------------
--[Rani] the makeKeywordResultFormatter works easily for index based transation values
-- since the rating is used as a number using a method for translation

local _translatedReviewValues = { 'reject', 'unflagged', 'pick' }

local _translatedReviewValueToNumericValue = KSTableUtils.invert(
												_translatedReviewValues )

local function translateAssetReview( assetReviewFlag)
	
	local flag
	if type(assetReviewFlag) == "string" then
		flag = tonumber(assetReviewFlag)
	else
		flag = assetReviewFlag
	end
	
	return _translatedReviewValues[flag]
	
end

--------------------------------------------------------------------------------
local function convertItemsArray_To_NamedFieldItemsArray( detailsInfoArray, fieldNamesInOrder )
	
	local itemDetailsArray = {}
	local filedCount = fieldNamesInOrder and #fieldNamesInOrder
	
	for _, itemInfo in ipairs( detailsInfoArray ) do
	
		if itemInfo and #itemInfo >= filedCount then
		
			local itemDetailsObj = {}
			
			for  index, fieldName in ipairs( fieldNamesInOrder ) do
				itemDetailsObj[ fieldName ] = itemInfo[ index ]
			end
			
			table.insert( itemDetailsArray, itemDetailsObj ) 
		end
	end
				
	return itemDetailsArray
end
--------------------------------------------------------------------------------
function WFCatAsset.collectSortedUniqueImportTimeAndCountOfAssets( wfCatalog, params )

	 return WFCatSearchByImportDate.importedDates( wfCatalog, params )
	 
	 		:flatMapLatest( function( importTimeInfoArray )
		
				local importTimeItemsArray = convertItemsArray_To_NamedFieldItemsArray( 
							importTimeInfoArray, { 'importTime', 'count' }  )
							
				return Observable.just( importTimeItemsArray )
			
		end)
end


--------------------------------------------------------------------------------
function WFCatAsset.collectSortedUniqueCaptureDatesAndCountOfAssets( wfCatalog, params )

	return  WFCatSearchByCaptureDate.captureDates( wfCatalog, params )
	
			:flatMapLatest( function( captureDatesInfoArray )
	
				local captureDatesItemsArray = convertItemsArray_To_NamedFieldItemsArray( 
							captureDatesInfoArray, { 'captureDate', 'count' }  )
							
				return Observable.just( captureDatesItemsArray )
				
			end)

end

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

function WFCatAsset.collectKeywordDetailsOfRequiredAssets( wfCatalog, assetIds )

	return  WFCatSearchByKeywords.collectKeywordDetailsOfRequiredAssets( wfCatalog, assetIds )
	
end

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

function WFCatAsset.collectLCKeywordDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchByKeywords.collectLCKeywordDetailsOfAllAssets( wfCatalog )
	
end

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

function WFCatAsset.collectKeywordDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchByKeywords.collectKeywordDetailsOfAllAssets( wfCatalog )
	
end

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

function WFCatAsset.collectCameraDetailsOfRequiredAssets( wfCatalog, assetIds )

	return  WFCatSearchByCamera.collectCameraDetailsOfRequiredAssets( wfCatalog, assetIds )
	
end

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

function WFCatAsset.collectCameraDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchByCamera.collectCameraDetailsOfAllAssets( wfCatalog )
	
end

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

function WFCatAsset.collectLocationDetailsOfRequiredAssets( wfCatalog, assetIds )

	return  WFCatSearchByLocation.collectLocationDetailsOfRequiredAssets( wfCatalog, assetIds )
	
end

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

function WFCatAsset.collectLocationDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchByLocation.collectLocationDetailsOfAllAssets( wfCatalog )
	
end

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

function WFCatAsset.collectSyncStatusDetailsOfRequiredAssets( wfCatalog, assetIds )

	return  WFCatSearchBySyncStatus.collectSyncStatusDetailsOfRequiredAssets( wfCatalog, assetIds )
	
end

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

function WFCatAsset.collectSyncStatusDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchBySyncStatus.collectSyncStatusDetailsOfAllAssets( wfCatalog )
	
end

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

function WFCatAsset.collectAssetTypeDetailsOfRequiredAssets( wfCatalog, assetIds )

	return  WFCatSearchByAssetType.collectAssetTypeDetailsOfRequiredAssets( wfCatalog, assetIds )
	
end

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

function WFCatAsset.collectAssetTypeDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchByAssetType.collectAssetTypeDetailsOfAllAssets( wfCatalog )
	
end

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

function WFCatAsset.collectExtensionDetailsOfRequiredAssets( wfCatalog, assetIds )

	return  WFCatSearchByExtension.collectExtensionDetailsOfRequiredAssets( wfCatalog, assetIds )
	
end

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

function WFCatAsset.collectExtensionDetailsOfAllAssets( wfCatalog )

	return  WFCatSearchByExtension.collectExtensionDetailsOfAllAssets( wfCatalog )
	
end

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

--[[!
	\brief Collect ratings and flags on a list of asset ids from the Index store
	\param wfCatalog the catalog
	\param params a table:
		assetIds; an array of local ids
		ratings: optional (boolean) Whether to gather ratings
		flags: optional (boolean) Whether to gather flags
		keywords: optional (boolean) Whether to gather keywords
		autoTags: optional (boolean) whether to gather autotags (only works when querying for one asset)
		statuses: optional (boolean) whether to gather sync status values
		localFileTypes: optional (boolean) whether to gather local file type values (original, proxy, none)
		coalesce: optional (boolean) if true, then a single result is returned using
		"mixed" where values are different

	\return Rx.Observable( array of ratings and/or flags ) an array of results, one
		for each assetId.  Ratings are represented by the key 'rating' and values from 0-5.
		Flags are represented by the key 'flag' and values 'reject', 'unflagged', or 'pick'.
		If coalesce is true, then the result is not an array, but one result table,
		which additionally holds a value by key 'numberOfAssets' indicating how many
		assets were scanned to compute the result.
--]]
function WFCatAsset.collectInfoForAssetsFromIndexStore( wfCatalog, params )

	local assetIds = params.assetIds
	assert( type( assetIds ) == 'table', 'params assetIds must be an array of integer assetIds' )

	local queries = {}

	-- 2016.08.26 bkruse 
	--
	-- Note the use of LEFT JOIN below. 
	-- 
	-- 'TEMP_ids LEFT JOIN some_index_store_table' 
	-- 
	-- is used to force a result set that has at least one record for each
	-- assetId in TEMP_ids, even if some_index_store_table has no corresponding
	-- record for the assetId. 
	--
	-- This allows to iterate over every assetId in TEMP_ids and fix up
	-- a default value if the assetId doesn't have one.
	-- 
	-- If we simply ignore empty values, then we should *NOT* use LEFT JOIN,
	-- just use 
	-- 
	-- 'TEMP_ids JOIN some_index_store_table'
	-- 
	-- instead. 
	-- 
	-- The  result formatter will run faster on plain JOINs because it will
	-- only run on assetIds that actually have values
	--
	-- Finally if you *do* need empty values for each assetId and are using
	-- LEFT JOIN, then it is critical that you use
	--
	-- SELECT TEMP_ids.docId, some_index_store_table.fieldColumn
	-- FROM ....
	--
	-- With  LEFT JOIN, TEMP_ids.docId will always have a value but
	-- some_index_store_table.docId will be NULL for the empty values

	if params.ratings then
		queries[ #queries + 1 ] = wfCatalog:indexStore():query {
			sql = [[
				SELECT ai.value, ratings.rating
				FROM CARRAY( ?, ?, 'int32') AS ai
					LEFT JOIN ratings ON ai.value = ratings.docId 
				WHERE ratings.userId == ?]],
			
			assetIds,
			#assetIds,
			wfCatalog:ozUserId(),
				-- SQL ? params must be the last entry, because the array
				-- portion of an inline table will be truncated by any
				-- subsequent table key-value definitions.

		}
		:map( makeDefaultValueFormatter( 'rating', 0  ) )

	end

	if params.flags then
		queries[ #queries + 1 ] = wfCatalog:indexStore():query {
			sql = [[
				SELECT rv.value, reviews.flag
				FROM CARRAY( ?, ?, 'int32') AS rv
					LEFT JOIN reviews ON reviews.docId = rv.value
				WHERE reviews.userId == ?]],
			assetIds,
			#assetIds,
			wfCatalog:ozUserId(),
				-- SQL ? params must be the last entry, because the array
				-- portion of an inline table will be truncated by any
				-- subsequent table key-value definitions.
		}
		:map( makeTranslateWithDefaultValueFormatter( 
					'flag', translateAssetReview, 
					_translatedReviewValueToNumericValue[ 'unflagged' ]  ) )
	end
	
	if params.statuses then
	
		-- Asset status is determined by both what's in the index store,
		-- as well as the current connection status of the volumes used by Nimbus,
		-- as well as the global sync status ( whether or not there is an internet
		-- connection and whether or not sync is paused ).
		
		-- For example, there exists a unique status for an asset that needs to be
		-- downloaded, as well as a unique status for an asset that needs to be
		-- downloded but cannot be because the volume that stores originals is offline,
		-- or because there's no internet connection.
		
		local keepLocallyConfiguration = WFCatSyncStatus.getKeepLocallyConfiguration()
		
		local assetSyncStatusQuery = wfCatalog:indexStore():query {
			sql = [[
				SELECT assu.value, asset_sync_status_ui.prioritizedSyncStatus,
						asset_sync_status_ui.volumeGuidUsedByOriginal,
						asset_sync_status_ui.volumeGuidUsedByProxy,
						asset_sync_status_ui.spaceId
					FROM CARRAY( ?, ?, 'int32') AS assu
						LEFT JOIN asset_sync_status_ui ON asset_sync_status_ui.docId = assu.value
					WHERE keepLocallyConfiguration == ? ]],
			assetIds,
			#assetIds,
			keepLocallyConfiguration
		}
	
		--[[ 
	
		True asset sync status is made up 4 separate data points.
		
		1. The current state of the asset as generated by "getAssetStatus". This
			is the same status that is stored in sync status index store table.

		2. The state of all spaces that have been shared with the user. If someone
			has shared a space with a user, and that space owner has an expired account,
			nothing can be synced to or from that space.

		3. The current state of the volumes where the assets are stored. If an asset
		 	needs to be downloaded, but the volume we need to download it to is
			offline, that's a new unique state.
		
		4. The current global state of sync; is sync paused and/or offline.

		]]--
			
		local mappedVolumeStatuses = WFCatSyncStatus.getVolumeInfoForSyncStatus( wfCatalog )

		local volumeAgents = wfCatalog:getAllVolumeAgents()
		local volumeAgent = volumeAgents[ next( volumeAgents ) ]
		local syncPausedStream = volumeAgent and volumeAgent:isSyncPausedStream() or Observable.just( false )
		local internetConnectionStream = volumeAgent and volumeAgent:hasInternetConnectionStream() or Observable.just( true )
		local blockedSpaceIdsStream = WFCatSyncStatus.getBlockedSpaceIdsMapStream( wfCatalog )

		local assetSyncStatusWithDriveStatus 
				= Observable.combineLatest(
						assetSyncStatusQuery,
						blockedSpaceIdsStream,
						mappedVolumeStatuses )
				
				:map( function( syncStatusAndVolumeInfoRows, blockedSpaceIds, volumeGuidForOriginals,
						volumeGuidForProxies, onlineVolumes, outOfSpaceVolumes, knownVolumes )
				
					local modifiedRows = {}
				
					local isVolumeForOriginalsOnline = onlineVolumes[ volumeGuidForOriginals ]
					local isVolumeForProxiesOnline = onlineVolumes[ volumeGuidForProxies ]
				
					local isVolumeForOriginalsOutOfSpace = outOfSpaceVolumes[ volumeGuidForOriginals ]
					local isVolumeForProxiesOutOfSpace = outOfSpaceVolumes[ volumeGuidForProxies ]
				
					for _, syncStatusAndVolumeInfoRow in ipairs( syncStatusAndVolumeInfoRows ) do
					
						local docId = syncStatusAndVolumeInfoRow[ 1 ]
						local syncStatus = syncStatusAndVolumeInfoRow[ 2 ]
						local volumeGuidUsedByOriginal = syncStatusAndVolumeInfoRow[ 3 ]
						local volumeGuidUsedByProxy = syncStatusAndVolumeInfoRow[ 4 ]
						local spaceId = syncStatusAndVolumeInfoRow[ 5 ]
						
						local isOriginalOnKnownVolume = knownVolumes[ volumeGuidUsedByOriginal ]
						local isProxyOnKnownVolume = knownVolumes[ volumeGuidUsedByProxy ]
						
						local isOriginalOnOnlineVolume = onlineVolumes[ volumeGuidUsedByOriginal ]
						local isProxyOnOnlineVolume = onlineVolumes[ volumeGuidUsedByProxy ]
						
						local isBlockedSpace = spaceId and blockedSpaceIds[ spaceId ]

						-- Calculate the new sync status.
						
						syncStatus = WFCatSyncStatus.calculateSyncStatusWithVolumeStatus( 
								syncStatus, volumeGuidUsedByOriginal, volumeGuidUsedByProxy,
								isOriginalOnKnownVolume, isProxyOnKnownVolume,
								isVolumeForOriginalsOnline, isVolumeForProxiesOnline,
								isVolumeForOriginalsOutOfSpace, isVolumeForProxiesOutOfSpace,
								isOriginalOnOnlineVolume, isProxyOnOnlineVolume, isBlockedSpace )
						
						table.insert( modifiedRows, { docId, syncStatus } )
						
					end
				
					return modifiedRows
				
				end )
			
			-- Account for the global sync status ( paused / no internet )
			
			local assetSyncStatusWithGlobalStatus = 
					Observable.combineLatest( assetSyncStatusWithDriveStatus, syncPausedStream, internetConnectionStream )
					:map( function( rows, isSyncPaused, isOnline )
					
						local modifiedRows = {}
					
						for _, row in ipairs( rows ) do
						
							local docId = row[ 1 ]
							local syncStatus = row[ 2 ]
						
							syncStatus = WFCatSyncStatus.calculateSyncStatusWithGlobalStatus( 
									syncStatus, isSyncPaused, not isOnline )
								
							table.insert( modifiedRows, { docId, syncStatus } )
								
						end
					
						return modifiedRows
					
					end )
					
		queries[ #queries + 1 ] = assetSyncStatusWithGlobalStatus
			:map( makeTranslaterFormatter( 'status', WFCatSyncStatus.translateAssetSyncStatus ) )
		
	end

	if params.comments and params.albumId then
	
		queries[ #queries + 1 ] = WFCatCollaborationActivity.getCommentsForAlbumByAssetId( wfCatalog, params.albumId, assetIds )
			:map( function( data )
				
				local mappedData = {}
				
				for assetId, status in pairs( data ) do
					mappedData[ assetId ] = { comment = status }
				end
				
				return mappedData
				
			end )
		
	end
	
	if params.favorites and params.albumId then
		
		queries[ #queries + 1 ] = WFCatCollaborationActivity.getFavoritesForAlbumByAssetId( wfCatalog, params.albumId, assetIds )
			:map( function( data )
				
				local mappedData = {}
				
				for assetId, status in pairs( data ) do
					mappedData[ assetId ] = { favorite = status }
				end
				
				return mappedData
				
			end )

	end
	
	if params.localFileTypes then
		queries[ #queries + 1 ] = wfCatalog:indexStore():query {
			sql = [[
			SELECT ai.value, asset_sync_local_file_type_ui.localFileType
				FROM CARRAY( ?, ?, 'int32') AS ai
					LEFT JOIN asset_sync_local_file_type_ui ON asset_sync_local_file_type_ui.docId = ai.value
			]],
			assetIds,
			#assetIds,
		}
		:map( makeTranslaterFormatter( 'localFileType', WFCatSyncStatus.translateAssetSyncLocalFileType ) )
	end

	if params.keywords then
		
		queries[ #queries + 1 ] = wfCatalog:indexStore():query {
			sql = [[
				SELECT ai.value, keywords.keyword
				FROM CARRAY( ?, ?, 'int32') AS ai
					LEFT JOIN keywords ON keywords.docId = ai.value
				]],

			assetIds,
			#assetIds,
		}
		:map( makeTranslaterFormatter( 'keywords', kTranslateFieldValueFn[ 'keywords' ] ) )
		
	end
	
	if params.autoTags then

		-- TODO bkruse 2016.08.26 the autoTags search should be removed from this method.
		--
		-- Instead we should add 
		--		WFCatAsset.collectAutoTagInfoForAssetId( wfCatalog, assetId )
		--
		-- Also, this search works completely different than the others and will completely
		-- break it it's combined with other criteria.
		--
		-- It emits { { autoTags = { ... }  } all other search
		

		local errStr = "Auto tags must be refactored!"

		logger:error( errStr )

		error( errStr )

		local singleSelectedAsset = { assetIds[1] }		-- only get autotags of first asset
		
		queries[ #queries + 1 ] = wfCatalog:indexStore():query {
			sql = [[
				SELECT autoTags.autoTag, autoTags.confidence
				FROM CARRAY( ?, ?, 'int32') AS ai, autoTags 
				WHERE autoTags.docId = ai.value
				ORDER BY autoTags.confidence DESC
				]],

			singleSelectedAsset,
			#singleSelectedAsset,
		}
		:map( 
			function( rows )
				local autoTags = {}
				for _, row in ipairs( rows ) do
					table.insert(autoTags , { tag = row[ 1 ], confidence = row[ 2 ] } )
				end
				
				return { { autoTags = autoTags } }
			end )

	end
	
	local updatestriggerStream
	if params.applyUpdates then
		updatestriggerStream = WFCatPendingUpdates.pendingUpdatesTriggerStream()
	else
		updatestriggerStream = Observable.just( false )
	end
	
	-- Generic combining function combines results from above.

	-- TODO 2016.08.26 bkruse this is potentially expensive if we combine many
	-- queries of many results when the queries are updating.
	--
	-- it is O(nm) where n is the number of queries and m is the number of
	-- results and it will happen any time any of the queries emit updated results.
	--
	-- We could make this faster by using :scan to remember the previous
	-- results and only update the results for newly changed queries.
	--
	-- That would make the cost O( m ). Ping me if this is a hotspot.
	return Observable.combineLatest( function( hasUpdates, ... )
	
			local correctOrder = KSTableUtils.invert( assetIds )
	
			local results = {}
			
			for i = 1, select( '#', ... ) do
			
				local result = select( i, ... )
				
				for assetId, assetInfo in pairs( result ) do
				
					local orderedKey = correctOrder[ assetId ] 
				
					local combined = results[ orderedKey ] or {}
					
					if params.includeIds then
						combined.id = assetId
					end
					
					if assetInfo.rating then
						collectedRatingsMap[ assetId ] = assetInfo.rating
					end

					if assetInfo.flag then
						collectedFlagsMap[ assetId ] = assetInfo.flag
					end
					
					for k, v in pairs( assetInfo ) do
					
						combined[ k ] = v
						
						if params.applyUpdates and kUpdateEnabledFields[ k ] then
							applyPendingUpdatesForAssetForField( assetId, k, assetInfo )
						end
					end
					
					results[ orderedKey ] = combined
					
				end
				
			end
			
			if params.coalesce then
				results = coalesceValues( results )
			end
			
			
			return results
			
		end, updatestriggerStream, unpack( queries ) )
end

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

--[[!
	\brief Collect ratings and flags on a list of asset ids from the Doc store

	\param wfCatalog the catalog
	\param params a table:
		assetIds; an array of local ids
		ratings: optional (boolean) Whether to gather ratings
		flags: optional (boolean) Whether to gather flags
		isEdited: optional (boolean) Whether to gather isEdited booleans based on userUpdated dates
		autotags: not supported
		coalesce: optional (boolean) if true, then a single result is returned using
		"mixed" where values are different

	\return Rx.Observable( array of ratings and/or flags ) an array of results, one
		for each assetId.  Ratings are represented by the key 'rating' and values from 0-5.
		Flags are represented by the key 'flag' and values 'reject', 'unflagged', or 'pick'.
		If coalesce is true, then the result is not an array, but one result table,
		which additionally holds a value by key 'numberOfAssets' indicating how many
		assets were scanned to compute the result.
--]]

function WFCatAsset.collectInfoForAssetsFromDocStore( wfCatalog, params )

	local assetIds = params.assetIds
	assert( type( assetIds ) == 'table', 'params assetIds must be an array of integer assetIds' )
		
	if params.coalesce then
		return collectCoalescedInfoForAssetsFromDocStore( wfCatalog, params )
	else
		return collectRawInfoForAssetsFromDocStore( wfCatalog, params )
	end
end

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

function WFCatAsset.collectPathsForAssetFromDocStore( wfCatalog, assetId )
	local function getPaths(docValue)
		local paths = {
			original = WFCatFileSpec.buildOriginalPathFromDocValue(docValue),
			proxy = WFCatFileSpec.buildProxyPathFromDocValue(docValue),
		}
		return paths
	end
	return wfCatalog:docStore():docValue( assetId )
		:map( getPaths )
end


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

function WFCatAsset.collectCaptureDateTimeRangeForAssets( wfCatalog, assetIds )
	
	return wfCatalog:indexStore():query {
			sql = [[
				SELECT captureDate, completeTime FROM CARRAY( ?, ?, 'int32') AS ai
				JOIN assets ON assets.docId = ai.value
				JOIN captureDate ON captureDate.docId = ai.value
			]],
			assetIds,
			#assetIds,
			calm = 1,
		}
		:map( function( rows )
			local processedImages, numOfEmptyCaptureDate = 0, 0
			local minCaptureDate, maxCaptureDate
			local minCaptureTime, maxCaptureTime
			
			for _, row in ipairs( rows ) do
				
				local rowCaptureDate = row[ 1 ]
				local rowCaptureTime = row[ 2 ]
				
				local captureDate =  KSDate.maybeKSDateFromString( rowCaptureDate )
				
				processedImages = processedImages + 1
				if rowCaptureTime == 0 then
					numOfEmptyCaptureDate = numOfEmptyCaptureDate + 1
				else
					if not minCaptureTime or rowCaptureTime < minCaptureTime then
						minCaptureTime = rowCaptureTime
						minCaptureDate = captureDate
					end
					
					if not maxCaptureTime or rowCaptureTime > maxCaptureTime then
						maxCaptureTime = rowCaptureTime
						maxCaptureDate = captureDate
					end
				end
			end
			
			local captureDateTimeRange = {
					maxCaptureDate = maxCaptureDate,
					maxTime = maxCaptureTime,
					minCaptureDate = minCaptureDate,
					minTime = minCaptureTime,
					numOfEmptyCaptureDate = numOfEmptyCaptureDate,
					processedImages = processedImages,
				}
				
			return { captureDateTimeRange = captureDateTimeRange, numberOfAssets = processedImages }
			
		end )

end

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

local function getLatestEditDetailsForAsset( wfCatalog, docIds )

	return wfCatalog:docStore():bulkDocValue( docIds )

		:flatMap( function( docValues )

			return Observable.from( docValues )

		end )

		:flatMap( function( assetDocValue )

			local userIds = {}

			local tempDevelopInfo = WFCatAssetUtils.getMostRecentTempDevelopInfo( assetDocValue )
			local userId = derefOpt( tempDevelopInfo, 'uploadedByUser', 'id' )
			if userId then
				userIds[ #userIds + 1 ] = userId
			end

			userId = derefOpt( assetDocValue, '_serverProvidedMetadata', 'developedByUser', 'id' )
			if userId and userId ~= userIds[ 1 ] then
				userIds[ #userIds + 1 ] = userId
			end

			local editDetailsObs

			if #userIds > 0 then

				editDetailsObs = wfCatalog:docStore():bulkDocValue( userIds )

					:map( function( feedbackDocs )
						return WFCatAssetUtils.getLatestEditDetails( assetDocValue, feedbackDocs, tempDevelopInfo )
					end )

			else

				editDetailsObs = Observable.just( WFCatAssetUtils.getLatestEditDetails( assetDocValue, {}, tempDevelopInfo ) )

			end

			return editDetailsObs

				:map( function( ... )
					return assetDocValue._id, ...
				end )

		end )

		:scan( function( detailsByDocId, assetId, editDetails )

			detailsByDocId[ assetId ] = editDetails

			return detailsByDocId

		end, {} )

		:takeLast( 1 )

end

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

--[[!
	\brief Describe the latest edits by a user, for a single document.

	\param assetId (string|number) A local or full document Id.

	\return Observable that fires with the latest edit details:
		{
			userUpdated = <date>,
			userDevice = <device>,
			userName = <name>,
			tempBinary = {
				userUpdated = <date>,
				userDevice = <device>,
				userName = <name>,
			},
		}
--]]

function WFCatAsset.getAssetLatestEditDetails( wfCatalog, assetId )

	check( 'WFCatalog', 'string|number' )

	return getLatestEditDetailsForAsset( wfCatalog, { assetId } )

		:map( function( editDetailsByDocId )

			local fullDocId = next( editDetailsByDocId )

			return fullDocId and editDetailsByDocId[ fullDocId ] or {}

		end )

end

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

--[[!
	\brief Describe the latest edits by a user, for multiple documents.

	\param docIds (table) Table of document IDs

	\return Observable that fires with the latest edit details for all documents,
		keyed by document Id.
		{
			<docId1> = {
				userUpdated = <date>,
				userDevice = <device>,
				userName = <name>,
				tempBinary = {
					userUpdated = <date>,
					userDevice = <device>,
					userName = <name>,
				},
			},
			<docId2> = {
				userUpdated = <date>,
				userDevice = <device>,
				userName = <name>
			},
			...
		}

--]]

function WFCatAsset.getLatestEditDetailsForAssets( wfCatalog, docIds )

	check( 'WFCatalog', 'table' )

	return getLatestEditDetailsForAsset( wfCatalog, docIds )

end

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

--[[!
	\brief Idetifies if temporary edits by a different client but same user
		exist the specified document.

	\param assetId (string|number) A local or full document Id.

	\return Observable that fires with the user's other client edit details:
		{
			hasTempEdits <boolean>
			userDeviceWithoutId <device>
			userUpdatedAsString <string>
		}
--]]

function WFCatAsset.getOtherClientEditDetails( wfCatalog, assetId )

	check( 'WFCatalog', 'string|number' )

	return WFCatAsset.getAssetLatestEditDetails( wfCatalog, assetId )

		:map( function( details )

			local thisDevice =  require 'WFDeviceInfo'.deviceDescription()
			local userDevice = derefOpt( details, 'tempBinary', 'userDevice' )

			-- Only return info for edits made on other devices from the same user.
			local hasTempEdits = details.tempBinary
					and userDevice ~= thisDevice
					and details.tempBinary.userName == nil

			local formattedDate

			if hasTempEdits then

				local deviceIdPos = userDevice and string.find( userDevice, ' {' )
				if deviceIdPos then
					userDevice = string.sub( userDevice, 1, deviceIdPos - 1 )
				end

				local userUpdated = derefOpt( details, 'tempBinary', 'userUpdated' )
				if userUpdated then
					local userUpdatedAsNumber = KSDate.iso8601ToTimeIntervalSinceReference( userUpdated )
					userUpdated = KSDate.formatDateTimeAsLocal( userUpdatedAsNumber )
					formattedDate = KSDate.formatDateTimeFromISO8601DateString( userUpdated, nil, 'short' )
				end

			end

			return {
				hasTempEdits = not not hasTempEdits, -- force true or false
				userDeviceWithoutId = hasTempEdits and userDevice or nil,
				userUpdatedAsString = formattedDate,
			}

		end )

end

--[[!
	\brief Idetifies if temporary edits by a different client but same user
		exist the specified document. Sends a new payload every time the 
		document being watched changes.

	\param assetId (string|number) A local or full document Id.

	\return Observable that fires with the user's other client edit details:
		{
			hasTempEdits <boolean>
			userDeviceWithoutId <device>
			userUpdatedAsString <string>
		}
--]]

function WFCatAsset.ongoingGetOtherClientEditDetails( wfCatalog, assetId )

	return Rx.Observable.just( wfCatalog )
		-- every time the asset changes, get new edit details
		:flatMapLatest( function( wfCatalog )

			return wfCatalog:docStore()
				:ongoingDocValue( assetId )
				:flatMapLatest( function(_)

					return WFCatAsset.getOtherClientEditDetails( wfCatalog, assetId )

				end)

	end )

end

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

return WFCatAsset
