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

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

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

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

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

	-- WF.core.rxlua
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.string
local KSStringUtils = require 'KSStringUtils'
local WFSortOrderStrings = require 'WFSortOrderStrings'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFArguments = require 'WFArguments'
local WFCatCollaboration = require 'WFCatCollaboration'
local WFCatIdUtils = require 'WFCatIdUtils'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFAlbumOrdering = require 'WFAlbumOrdering'
local WFRxBatchedBulkApplyOverlay = require 'WFRxBatchedBulkApplyOverlay'
local WFRxBatchedBulkSetDocValue = require 'WFRxBatchedBulkSetDocValue'
local WFLogger = require 'WFLogger'
local WFTableMerge = require 'WFTableMerge'
local WFDiagnosticLog = require 'WFDiagnosticLog'

local WFCatTaskManager = require 'WFCatTaskManager'
local WFCatReorderAssetsTask = require 'WFCatReorderAssetsTask'

	-- WF.oz_client.managed_catalog
local WFCatDelete = require 'WFCatDelete'
local WFCatUtils = require 'WFCatUtils'
local WFCatSyncStatus = require 'WFCatSyncStatus'

local WFRxDocValueWatcher = require 'WFRxDocValueWatcher'

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

local logger = WFLogger( 'WFCatAlbums' )

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

local OZ_MAX_ALBUM_NAME_CHARS = 250

local ROOT_ALBUM_PARENT_ID = 'root'

local SORT_PREV_FOR_INSERT_AS_FIRST = ''

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

local check = assert( WFArguments.check )
local derefOpt = assert( KSTableUtils.derefOpt )

local runUntilDone = WFRxRunUntilDone

--------------------------------------------------------------------------------
	
local WFFunctionUtils_map =
		assert( WFFunctionUtils.map, "WFFunctionUtils.map is missing" )

--------------------------------------------------------------------------------
	
local WFFunctionUtils_reduce =
		assert( WFFunctionUtils.reduce, "WFFunctionUtils.reduce is missing" )

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

local KSDate_timeNowForAnimation =
		assert( KSDate.timeNowForAnimation,
					"KSDate.timeNowForAnimation is missing" )

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

-- @mark WFCatAlbums namespace

local WFCatAlbums = {}

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

local function hasParent( query, match, ... )
	if not match then
		return false
	end
	if query == match then
		return true
	end
	return hasParent( query, ... )
end

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

local function findCycles( rows )

	local allParents = {}
		-- each entry is an array of all the parents for that album
		
	local cycles = {}
		-- array of arrays, each containing the docIds of albums in a cycle

	for _, row in ipairs( rows ) do
	
		local parentId = row[ 7 ]
		local docId = row[ 1 ]
		
		if parentId then
		
			local parentParents = allParents[ parentId ]
			local myParents = { parentParents and unpack( parentParents ) }
			
			myParents[ #myParents + 1 ] = parentId
			
			if hasParent( docId, unpack( myParents ) ) then
				cycles[ #cycles + 1 ] = myParents
			end
			
			allParents[ docId ] = myParents
			
		end
		
	end

	return cycles
	
end

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

local function breakCycles( rows )

	local cycles = findCycles( rows )

	-- If any loops are found (usually none will be) we
	-- need to break them

	if next( cycles ) then
	
		local rowsById = {}
		
		for _, row in ipairs( rows ) do
			rowsById[ row[ 1 ] ] = row
		end
		
		for _, docIds in ipairs( cycles ) do
		
			table.sort( docIds, function( a, b )
				local rowA = rowsById[ a ]
				local rowB = rowsById[ b ]
				local tieBreakerA = rowA[ 5 ] or rowA[ 6 ] or rowA[ 1 ]
				local tieBreakerB = rowB[ 5 ] or rowB[ 6 ] or rowB[ 1 ]
				return tieBreakerA < tieBreakerB
			end )
			
			-- Nil out the parent field on the oldest node
			rowsById[ docIds[ 1 ] ][ 7 ] = nil
			
		end
		
	end
	
end

WFCatAlbums._breakCycles = breakCycles
	-- For unit tests

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

local function createAlbumAssetDocument( albumId, assetId, spaceId )

	local baseDocValue = WFCatUtils.createBaseOverlayWithUserCreatedFields( KSIdGenerator.uuidInLowercaseHex() )

	local albumAssetDocValue = {
		type = 'album_asset',
		album = {
			id = albumId,
		},
		asset = {
			id = assetId,
		},
		_spaceId = spaceId
	}
	
	WFTableMerge.applyOverlay( albumAssetDocValue, baseDocValue )
	
	return albumAssetDocValue
	
end

WFCatAlbums.createAlbumAssetDocument = createAlbumAssetDocument

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

local function intToBool( zeroOrOne )
	return zeroOrOne ~= 0
end

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

local function basicAlbumsList( wfCatalog, quiesceUntilUpToDate )

	local ownerInfoByAlbumIdObservable = 
			WFCatCollaboration.getInfoOfAllAlbumOwnersByAlbumId( wfCatalog )
	
	return Observable.combineLatest( 
			wfCatalog:indexStore():query {
				sql = [[ SELECT albums.docId, albumId, albums.spaceId, name, albums.updateTimestamp, 
								space_user.created, parentId, sortOrder, subtype, hidden, assetSortOrder,
								download, showMetadata, showLocation, allowComments, allowFavorites,
								space_user.role, hasDisplayFilters, title
							FROM albums
							LEFT JOIN space ON albums.spaceId = space.docId
							LEFT JOIN space_user ON space.spaceUserId = space_user.fullDocId ]],
				quiesceUntilUpToDate = quiesceUntilUpToDate,
			}, ownerInfoByAlbumIdObservable )
				
		:map( function( rows, ownerInfoByAlbumId )
			
			local albums = {}
			local docIdsToLocalId = {}
			
			for _, row in ipairs( rows ) do
			
				local albumId = row[ 2 ]
			
				local album = {}
				album.id = row[ 1 ]
				album.fullDocId = albumId
				album.spaceId = row[ 3 ]
				album.name = row[ 4 ]
				album.updateTimestamp = row[ 5 ]
				album.joinedTimestamp = row[ 6 ]
				album.parentDocId = row[ 7 ]
				album.sortOrder = row[ 8 ]
				album.subtype = row[ 9 ]
				album.hidden = intToBool( row[ 10 ] )
				album.assetSortOrder = row[ 11 ]
				album.hasDisplayFilters = row[ 18 ]
				album.title = row[ 19 ]

				if album.spaceId then
				
					album.shareSettings = {
						allowDownloads = intToBool( row[ 12 ] ) or false,
						showMetadata = intToBool( row[ 13 ] ) or false,
						showLocation = intToBool( row[ 14 ] ) or false,
						allowComments = intToBool( row[ 15 ] ) or false,
						allowFavorites = intToBool( row[ 16 ] ) or false
					}
					
					local role = row[ 17 ]
					
					role = WFCatCollaboration.downgradeRoleIfNotSupported( role )
					
					album.permissions = {
						role = role
					}
		
					album.ownerInfo = ownerInfoByAlbumId[ albumId ]
		
				end
				
				docIdsToLocalId[ album.fullDocId ] = album.id 
				
				albums[ #albums + 1 ] = album
				
			end
			
			for _, album in ipairs( albums ) do
			
				album.parentId = docIdsToLocalId[ album.parentDocId ] 
				album.parentDocId = nil -- this just cleans up the full doc id of the parent in 
										-- the album record. passing only local ids to the UI
			end
			
			return albums
			
		end )
	
end

WFCatAlbums._basicAlbumsList = basicAlbumsList

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

local function addAlbumIntoHierarchy( album, hierarchy )
	
	local albums = hierarchy.albums or {}
	local hiddenAlbums = hierarchy.hiddenAlbums or {}
	local projectAlbums = hierarchy.projectAlbums or {}
	local sharedAlbums = hierarchy.sharedAlbums or {}
	local sharedWithYouAlbums = hierarchy.sharedWithYouAlbums or {}
	local localIdToAlbumData = hierarchy.localIdToAlbumData or {}
	local docIdToAlbumData = hierarchy.docIdToAlbumData or {}
	local localIdToChildrenAlbumData = hierarchy.localIdToChildrenAlbumData or {}
	
	localIdToAlbumData[ album.id ] = album
	
	if album.fullDocId then
		docIdToAlbumData[ album.fullDocId ] = album
	end
	
	if album.spaceId then
		
		sharedWithYouAlbums[ #sharedWithYouAlbums + 1 ] = album
	
	elseif album.subtype == 'collection' or album.subtype == 'collection_set' then
		
		if album.hidden then
		
			hiddenAlbums[ #hiddenAlbums + 1 ] = album
	
		else
		
			albums[ #albums + 1 ] = album
	
			local parentId = album.parentId or ROOT_ALBUM_PARENT_ID
			
			local albumsInParent = localIdToChildrenAlbumData[ parentId ] or {}
			albumsInParent[ #albumsInParent + 1 ] = album
		
			localIdToChildrenAlbumData[ parentId ] = albumsInParent
		
		end
	
	elseif album.subtype == 'project' then
		
		projectAlbums[ #projectAlbums + 1 ] = album
	
	end
	
	if album.isShared and not album.hidden then
		sharedAlbums[ #sharedAlbums + 1 ] = album
	end

	hierarchy.albums = albums
	hierarchy.hiddenAlbums = hiddenAlbums
	hierarchy.sharedAlbums = sharedAlbums
	hierarchy.sharedWithYouAlbums = sharedWithYouAlbums
	hierarchy.projectAlbums = projectAlbums
	hierarchy.localIdToAlbumData = localIdToAlbumData
	hierarchy.localIdToChildrenAlbumData = localIdToChildrenAlbumData
	hierarchy.docIdToAlbumData = docIdToAlbumData
	
	
	return hierarchy
end

WFCatAlbums._addAlbumIntoHierarchy = addAlbumIntoHierarchy

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

local function initAlbumHierarchy()

	return {
		albums = {},
		hiddenAlbums = {},
		sharedAlbums = {},
		sharedWithYouAlbums = {},
		projectAlbums = {},
		localIdToChildrenAlbumData = {},
		localIdToAlbumData = {},
		docIdToAlbumData = {},
	}
	
end

WFCatAlbums._initAlbumHierarchy = initAlbumHierarchy

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

local function reparentAlbumInHierarchy( hierarchy, albumId, optNewParentId )
	
	local album = WFCatAlbums._getAlbumFromHierarchyById( hierarchy, albumId )
	local parent = WFCatAlbums._getAlbumFromHierarchyById( hierarchy, optNewParentId )
	
	local oldParentId = album and album.parentId or 'root'
	local newParentId = optNewParentId or 'root'
		
	if album and oldParentId ~= newParentId then 
		-- if we have an album and the parent has changed, move it around in the cache

		-- remove from old parent in cache
		local oldSiblings = hierarchy.localIdToChildrenAlbumData[ oldParentId ] or {}
		for i = #oldSiblings, 1, -1 do -- iterate backwards so we can use table.remove safely
			local sibling = oldSiblings[ i ]
			if sibling.id == album.id then
				table.remove( oldSiblings, i )
				break 
			end
		end
		hierarchy.localIdToChildrenAlbumData[ oldParentId ] = oldSiblings
		
		-- switch album's parent id
		if parent then
			album.parentId = optNewParentId
		else
			album.parentId = nil
		end
		
		-- add to new parent in cache
		local newSiblings = hierarchy.localIdToChildrenAlbumData[ newParentId ] or {}
		newSiblings[ #newSiblings + 1] = album
		hierarchy.localIdToChildrenAlbumData[ newParentId ] = newSiblings
		
	end
	
	return hierarchy

end

WFCatAlbums._reparentAlbumInHierarchy = reparentAlbumInHierarchy

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

local function removeIdsFrom( idsToRemove, albumList )
	
	if not next( idsToRemove ) then
		return
	end
	
	for i = #albumList, 1, -1 do
		local album = albumList[ i ]
		
		if idsToRemove[ album.id ] then
			table.remove( albumList, i )
		end
	end
	
end

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

local function deleteAlbumFromHierarchy( hierarchy, albumId )

	local album = WFCatAlbums._getAlbumFromHierarchyById( hierarchy, albumId )
	
	local idsRemoved = {}
	if album then
		local children = {}
		WFCatAlbums._addAllChildrenOfAlbumTo( hierarchy, album, children )
		
		local idsToRemove = {}
		idsToRemove[ album.id ] = true
		for _, child in ipairs( children ) do
			idsToRemove[ child.id ] = true
		end
		
		for id, _ in pairs( idsToRemove ) do
			
			local anAlbum = hierarchy.localIdToAlbumData[ id ]
			
			if anAlbum then
				hierarchy.localIdToAlbumData[ anAlbum.id ] = nil
				hierarchy.localIdToChildrenAlbumData[ anAlbum.id ] = nil
				if anAlbum.fullDocId then
					hierarchy.docIdToAlbumData[ anAlbum.fullDocId ] = nil
				end
				
				idsRemoved[ #idsRemoved + 1 ] = anAlbum.id
			end
		end
		
		-- remove album from the parent map
		local siblings = hierarchy.localIdToChildrenAlbumData[ album.parentId or ROOT_ALBUM_PARENT_ID ] or {}
		for i = #siblings, 1, -1 do -- iterate backwards
			local sibling = siblings[i]
			if sibling.id == albumId then
				table.remove( siblings, i )
				break 
			end
		end
		
		removeIdsFrom( idsToRemove, hierarchy.albums )
		removeIdsFrom( idsToRemove, hierarchy.hiddenAlbums )
		removeIdsFrom( idsToRemove, hierarchy.sharedAlbums )
		removeIdsFrom( idsToRemove, hierarchy.sharedWithYouAlbums )
		
	end
	
	return hierarchy, idsRemoved
	
end

WFCatAlbums._deleteAlbumFromHierarchy = deleteAlbumFromHierarchy

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

local function deleteProjectAlbumFromHierarchy( hierarchy, albumId )
	
	local album = WFCatAlbums._getAlbumFromHierarchyById( hierarchy, albumId )
	
	local idsRemoved = {}
	if album then
		
		hierarchy.localIdToAlbumData[ album.id ] = nil
		hierarchy.localIdToChildrenAlbumData[ album.id ] = nil
		if album.fullDocId then
			hierarchy.docIdToAlbumData[ album.fullDocId ] = nil
		end
		
		idsRemoved[ #idsRemoved + 1 ] = album.id
	
		removeIdsFrom( { albumId = true }, hierarchy.projectAlbums )
		
	end
	
	return hierarchy, idsRemoved
end

WFCatAlbums._deleteProjectAlbumFromHierarchy = deleteProjectAlbumFromHierarchy

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

local function basicProjectAlbumsList( wfCatalog )
	
	return wfCatalog:indexStore():query {
		sql = [[SELECT docId, albumId, name, updateTimestamp, sourceAlbumId, serviceId, pendingPublish, syncStatus, notifyConnect, coverId, assetSortOrder
				FROM project_albums]],
	}
	:map( function( rows )
		
		local albums = {}
		
		for _, row in ipairs( rows ) do
			local album = {}
			album.id = row[ 1 ]
			album.fullDocId = row[ 2 ]
			album.name = row[ 3 ]
			album.updateTimestamp = row[ 4 ]
			album.sourceAlbumId = row[ 5 ]
			album.serviceId = row[ 6 ]
			album.pendingPublish = row[ 7 ] == 1
			album.syncState = row[ 8 ]
			album.notifyConnect = row[ 9 ] == 1
			album.coverId = row[ 10 ]
			album.assetSortOrder = row[ 11 ]
			-- always project album subtype for now
			album.subtype = 'project'
			
			albums[ #albums + 1 ] = album
		end
	
		return albums
	end ) 

end

WFCatAlbums._basicProjectAlbumsList = basicProjectAlbumsList 

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

function WFCatAlbums.projectAlbums( wfCatalog )
	
	return basicProjectAlbumsList( wfCatalog )
		:map( function( projectAlbums )
			return projectAlbums
		end )

end

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

-- return observable containing the rationalized structure of the
-- album index. Pretty much all album APIs operate by computing the 
-- full album structure first, and then doing simple lua-data 
-- structure interrogation.
local function flattenedAlbumHierarchy( wfCatalog, quiesceUntilUpToDate )

	return Rx.Observable.combineLatest( 
			basicAlbumsList( wfCatalog, quiesceUntilUpToDate ), 
			basicProjectAlbumsList( wfCatalog ) )
			
		:map( function( albums, projectAlbums )
	
			local flattenedHierarchy = initAlbumHierarchy()
	
			-- SPAM [matthewj 2017-01-27]: Need to properly implement breakCycles for albums
			-- not rows (and it doesn't really work for cycles > 2 nodes currently anyway)
			-- breakCycles( albums )
		
			for _, album in ipairs( albums ) do
							
				addAlbumIntoHierarchy( album, flattenedHierarchy )
			
			end
			
			for _, album in ipairs( projectAlbums ) do
							
				addAlbumIntoHierarchy( album, flattenedHierarchy )
			
			end
	
			return flattenedHierarchy
	
		end )
		
end

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

local function getAlbumFromHierarchyById( hierarchy, albumId )
	
	local album
	if type( albumId ) == 'number' then
		album = hierarchy.localIdToAlbumData[ albumId ]
	else
		album = hierarchy.docIdToAlbumData[ albumId ]
	end

	return album
end

WFCatAlbums._getAlbumFromHierarchyById = getAlbumFromHierarchyById

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

local function addAllChildrenOfAlbumTo( hierarchy, album, albumList )
	
	local children = hierarchy.localIdToChildrenAlbumData[ album.id ] or {}
	
	for _, child in ipairs( children ) do
		albumList[ #albumList + 1  ] = child
		addAllChildrenOfAlbumTo( hierarchy, child, albumList )
	end
end

WFCatAlbums._addAllChildrenOfAlbumTo = addAllChildrenOfAlbumTo

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

local function getAlbumAndChildren( wfCatalog, albumId )

	return flattenedAlbumHierarchy( wfCatalog, true )
		:take( 1 )
		:map( function( hierarchy )
			
			
			local album = getAlbumFromHierarchyById( hierarchy, albumId )
			
			local albumAndChildren = {}
			
			if album then
				albumAndChildren[ #albumAndChildren + 1 ] = album
				
				addAllChildrenOfAlbumTo( hierarchy, album, albumAndChildren )
			end
				
			return albumAndChildren
		end )
		:mapSwitch( function( array )
			return Observable.from( array )
		end )

end

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

local function isAlbumNameUniqueAmongSiblings( albumName, albumSubtype, siblings, optIgnoreAlbumId )
	
	local nameIsUnique = true
	local lcAlbumName = KSStringUtils.lower( albumName )
	
	for _, sibling in ipairs( siblings or {} ) do

		-- if optIgnoreAlbumId is given, don't check for name collisions with it

		if optIgnoreAlbumId == nil or sibling.id ~= optIgnoreAlbumId then

			if KSStringUtils.lower( sibling.name ) == lcAlbumName 
				and sibling.subtype == albumSubtype then
		
				nameIsUnique = false
				break
			end
		end
	end

	return nameIsUnique

end

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

local function maybeMakeAlbumNameValid( albumName )

	if KSStringUtils.lengthOfUTF8String( albumName ) > OZ_MAX_ALBUM_NAME_CHARS then
		logger:warnf( 'Album names are limited to %d chars, truncating: %s', OZ_MAX_ALBUM_NAME_CHARS, albumName )
		
		albumName = KSStringUtils.UTF8Substring( albumName, 1, OZ_MAX_ALBUM_NAME_CHARS )
	end
	
	return albumName

end

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

local function isValidParent( album )

	if not album then
		return true -- assumes parent will be converted to root, 
					-- if no parent actually exists
	end
	
	if album.subtype == 'collection_set' then
		return true
	end
	
	return false
end

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

local function distinctTableCompare( _, tbl1, _, tbl2 )

	return KSTableUtils.deepEquals( tbl1, tbl2 )

end

--------------------------------------------------------------------------------
-- maybeMakeAllAlbumDataAsSubsetMapStream
-- 
-- returns a stream that, upon subscription, determines whether or not 
-- to schedule a 
-- 
-- 	makeAllAlbumDataAsSubsetMapStream( wfCatalog, ... )
-- 	
-- If scheduled, it will either be immediate or delayed long enough
-- to ensure rateLimited.pauseBetweenFullUpdateInSeconds have passed
--------------------------------------------------------------------------------

local function maybeMakeAllAlbumDataAsSubsetMapStream( 
				rateLimitData, 
				makeAllAlbumDataAsSubsetMapStream, wfCatalog, ... )

	local args, argsN = { ... }, select( "#", ... )

	return Rx.Observable.create(

		function( observer )

			rateLimitData.pendingMaybeMakeAllAlbumDataAsSubsetMapStream = false

			-- We need to schedule a call to
			-- 
			-- 		makeAllAlbumDataAsSubsetMapStream( wfCatalog, ... )
			-- 	
			local secsSinceLastUpdateAllAlbumDataStreamCompleted =
					not rateLimitData.lastUpdateAllAlbumDataStreamCompletionTime and
						rateLimitData.pauseBetweenFullUpdateInSeconds or
						KSDate_timeNowForAnimation() - 
							rateLimitData.lastUpdateAllAlbumDataStreamCompletionTime

			local maybeDelayedUpdateAllAlbumDataSignalStream =
					secsSinceLastUpdateAllAlbumDataStreamCompleted >= 
							rateLimitData.pauseBetweenFullUpdateInSeconds and
						Rx.Observable.just( true ) or
						Rx.Observable.timer( 
							rateLimitData.pauseBetweenFullUpdateInSeconds - 
								secsSinceLastUpdateAllAlbumDataStreamCompleted,
							WFRxScheduler.sharedScheduler() )
							
			local rateLimitedUpdateAllAlbumDataStream =
				maybeDelayedUpdateAllAlbumDataSignalStream
					:take( 1 )
					:flatMapLatest(
						function()

							return makeAllAlbumDataAsSubsetMapStream( 
									wfCatalog, unpack( args, 1, argsN ) )

						end )
					:tapOnCompleted(
						function()
							
							rateLimitData.lastUpdateAllAlbumDataStreamCompletionTime = 
								KSDate_timeNowForAnimation() 

						end )

				return rateLimitedUpdateAllAlbumDataStream
						:subscribe( observer )

		end )

end

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

local function addTableUpdateTrapSQL(
				sqlQueryToTrapTableUpdates, sqlTableName )

	if not sqlQueryToTrapTableUpdates then

		return string.format( "SELECT julianday( 'now' ), '%s'", sqlTableName )

	end

	return string.format( "%s,'%s'", sqlQueryToTrapTableUpdates, sqlTableName )

end

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

local function incrementValue( val )

	return val + 1

end

--------------------------------------------------------------------------------
-- rateLimitedSQLTablesUpdatedSignalStream
-- 
-- Given 
-- 
-- 	 sqlTableName1, ... 
-- 	 
-- a vararg list of sql table names, this constructs a stream that, on
-- subscription, will immediately emit 1 followed by a monontonically
-- increasing integer values that will be emitted no more than once per
-- updateRateInSeconds if any of the sql names have new index store data.
--------------------------------------------------------------------------------

local function rateLimitedSQLTablesUpdatedSignalStream( 
				indexStore, updateRateInSeconds, sqlTableName1, ... )

	local sqlQueryToTrapTableUpdates =
			WFFunctionUtils_reduce(
				nil,
				addTableUpdateTrapSQL,
					{ sqlTableName1, ... } )

	-- sqlQueryToTrapTableUpdates is now
	-- 
	-- SELECT now(), 'sqlTableName1', ..., 'sqlTableNameN'
	-- 
	-- feeding that to indexStore:query() will cause it to fire when any of the
	-- tables update.

	return indexStore
				:query( {
						sql = sqlQueryToTrapTableUpdates,
						calm = updateRateInSeconds,
					} )
				:scan( incrementValue, 0 )

end

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

local function mergeLatestAlbumData( _, currAlbumDataMap, 
										currAlbumIdMap, albumDataSubsetMap, 
										noAlbumDataSubsetMap )
	
	local didChange = not currAlbumDataMap
	
	currAlbumDataMap = currAlbumDataMap or {}

	if not next( albumDataSubsetMap ) and not next( noAlbumDataSubsetMap ) then

		-- we are starting a new full tree refresh. clear out any 
		-- albums not in currAlbumIdMap

		for albumId in pairs( currAlbumDataMap ) do

			if not currAlbumIdMap[ albumId ] then

				currAlbumDataMap[ albumId ] = nil

				didChange = true

			end
			
		end

	else

		-- we are incrementally updating, merge in latest subset
		for albumId, albumData in pairs( albumDataSubsetMap ) do

			if currAlbumDataMap[ albumId ] ~= albumData then

				currAlbumDataMap[ albumId ] = albumData

				didChange = true

			end

		end

		for albumId in pairs( noAlbumDataSubsetMap ) do

			if currAlbumDataMap[ albumId ] then

				currAlbumDataMap[ albumId ] = nil

				didChange = true

			end

		end

	end

	return didChange, currAlbumDataMap

end

--------------------------------------------------------------------------------
--
-- rateLimitedAlbumDataStream
-- 
-- Given a needsUpdateStream trigger stream and a makeAllAlbumDataAsSubsetMapStream, 
-- this returns a stream which, on subscription watches the needsUpdateStream.
-- 
-- When needsUpdateStream emits a value, it creates an albumDataStream via
-- 
--		 albumDataStream = makeAllAlbumDataAsSubsetMapStream( wfCatalog, ... ) 
--
-- albumDataStream sends tuples as follows
--
-- 		albumIdMap, {}
-- 		albumIdMap, albumDataSubsetMap1
-- 		...
-- 		albumIdMap, albumDataSubsetMapN
--
-- This stream is piped into :scan() where it incrementally updates the
-- albumDataMap for the entire album list by merging the data in each
-- albumDataSubsetMap. The incremental updates occur no more once per
-- refreshRateInSeconds.
-- 
-- Once albumDataStream completes, the returned stream either:
-- 
-- a) waits for new needsUpdateStream values then schedules a new
--    makeAllAlbumDataAsSubsetMapStream() call 
-- 
-- 	or 
-- 	
-- b) if needsUpdateStream emited values while consuming albumDataStream, it
--    schedules a makeAllAlbumDataAsSubsetMapStream() call 
-- 	   
-- In either case above, the scheduling will occur no sooner than
-- pauseBetweenResult seconds after the previous albumDataStream emited the
-- final value.
-- 
-- In practice this runs one of the following as updates come with subject to
-- rate limiting:
-- 
-- 	getAlbumCounts( ... ) 
-- 	
-- 	or 
-- 	
-- 	implicitCoversStream( ... )
--------------------------------------------------------------------------------

local function rateLimitedAlbumDataStream(
				desc,
				needsAlbumDataUpdateSignalStream, 
				pauseBetweenIncUpdateInSeconds,
				pauseBetweenFullUpdateInSeconds,
				makeAllAlbumDataAsSubsetMapStream, 
					indexStore, ... )

	local args, argsN = { ... }, select( "#", ... )

	local rateLimitData = { 

			pauseBetweenFullUpdateInSeconds = pauseBetweenFullUpdateInSeconds 

		}

	local streamOfMaybeUpdateAllAlbumDataStreams =
			needsAlbumDataUpdateSignalStream
				:filter(
					function()

						return not rateLimitData.pendingMaybeMakeAllAlbumDataAsSubsetMapStream 

					end )
				:map(
					function()

						rateLimitData.pendingMaybeMakeAllAlbumDataAsSubsetMapStream = true

						return maybeMakeAllAlbumDataAsSubsetMapStream( 
									rateLimitData,
									makeAllAlbumDataAsSubsetMapStream, 
										indexStore, unpack( args, 1, argsN ) )
								:tapOnCompleted(
									function()

										logger:infof( "Finished full update for: %s", desc )

									end )

					end )

	return streamOfMaybeUpdateAllAlbumDataStreams
			-- :flatMerge( 1 ) prevents overlapping full updates
			:flatMerge( 1 ) 
			-- :scan() merges incremental updates with the ongoing results
			:scan( mergeLatestAlbumData, false, nil )
			-- We are receiving a stream of
			-- 		didChange, albumDataMap
			-- 		didChange, albumDataMap
			-- 		didChange, albumDataMap
			-- we only want to emit when didChange is true
			:filter(
				function( didChange )

					return didChange

				end )
			:map(
				function( _, albumDataMap )

					return albumDataMap

				end )
			-- we are now getting a changed albumDataMap, :calm() prevents from
			-- sending updates too often
			:calm( pauseBetweenIncUpdateInSeconds, 
					WFRxScheduler.sharedScheduler() )

end

--------------------------------------------------------------------------------
-- given an array, makePaddedNElementSubarrayStream emits a stream of subarrays
-- of size exactly n. If #array = m and m is not a multiple of n, the last
-- subarray will n elements with the last n - n mod m elements assigned to
-- padVal
--
-- In practice, this breaks up a list of m ids so that they can be passed as 
-- params to a SQL query that has an IN( ?, ?, ..., ? ) predicate with n
-- placeholders.
--------------------------------------------------------------------------------

local function makePaddedNElementSubarrayStream( n, array, padVal )

	if #array == 0 then

		return Rx.Observable.empty()

	end

	local subArrays = {}

	local currSubArray = {}

	for _, v in ipairs( array ) do

		if #currSubArray == n then

			subArrays[ #subArrays + 1 ] = currSubArray

			currSubArray = {}

		end
	
		currSubArray[ #currSubArray + 1 ] = v

	end

	if #currSubArray > 0 then

		-- right pad the final nonempty array
		for i = #currSubArray + 1, n do

			currSubArray[ i ] = padVal

		end

		subArrays[ #subArrays + 1 ] = currSubArray 

	end

	return Observable.from( subArrays )

end

--------------------------------------------------------------------------------
-- albumDataSubsetMapForAlbumIdSubarrayStream emits a two maps:
-- 
--  albumDataMap
-- 
-- 		{ 
-- 			[ album i + 1 ] = data i + 1, 
-- 			...  
-- 			[ album j ] = data j, 
-- 			
-- 		}
-- 		
--  noAlbumDataMap
--  
-- 		{ 
-- 			[ album k + 1 ] = true,
-- 			...  
-- 			[ album l ] = true,
-- 			
-- 		}

local function albumDataSubsetMapForAlbumIdSubarrayStream( 
				indexStore, albumDataSQLWithPlaceholders, albumIdSubArray )

	return indexStore
			:queryOnce( {
					sql = albumDataSQLWithPlaceholders,
					unpack( albumIdSubArray )
				} )
			:map( 

				function( albumDataRows )

					local albumDataMap, noAlbumDataMap = {}, {}

					for _, albumDataRow in ipairs( albumDataRows ) do

						local albumId, albumData = unpack( albumDataRow )

						if albumData == nil then

							noAlbumDataMap[ albumId ] = true

						else 

							albumDataMap[ albumId ] = albumData

						end

					end

					for _, albumId in ipairs( albumIdSubArray ) do

						if not albumDataMap[ albumId ] then

							noAlbumDataMap[ albumId ] = true

						end

					end

					return albumDataMap, noAlbumDataMap

				end )

end

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

local function unpackQueryRows( packedQueryRows )

	return WFFunctionUtils_map( unpack, packedQueryRows )

end

--------------------------------------------------------------------------------
-- returns a stream that emits these tuples
-- 
-- 		albumIdMap, {}
-- 		albumIdMap, albumDataSubsetMap1
-- 		...
-- 		albumIdMap, albumDataSubsetMapN
-- 		
-- albumIdMap is a set of albumsIds returned
-- 
-- albumDataSubsetMap is an album data map for a subset of albumIdMap
-- 
-- Merging all of the albumDataSubsetMap will give a map of data for all
-- elements in albumIdMap

local function albumDataSubsetMapForAlbumIdsStream( 
				indexStore, albumIdListSQL, albumDataSQLWithPlaceholders,
				albumDataSQLAlbumIdPlaceholderCount )

	-- 1) get the list of all albums for which we wish to gather data
	local albumIdsOnceStream =
			indexStore
				:queryOnce( { sql = albumIdListSQL } )
				:map( unpackQueryRows )
				:publish()
				:refCount()

	-- 2) Convert albumIdsOnceStream to a set
	local albumIdMapOnceStream =
		albumIdsOnceStream
			:map( KSTableUtils.mapArrayToSet )

	-- 3a) Make the albumId list into a list of albumId sub lists
	local albumIdSubArrayStream =
			albumIdsOnceStream
				:flatMapLatest(
					function( albumIds )

						return makePaddedNElementSubarrayStream( 
								albumDataSQLAlbumIdPlaceholderCount, 
								albumIds, '' )

					end )

	-- 3b) Make the albumId sub array list into a series of streams that
	--     produce albumDataSubsetMap
	local streamOfAlbumDataSubsetMapStreams =
			albumIdSubArrayStream
				:map(
					function( albumIdSubArray )

						return albumDataSubsetMapForAlbumIdSubarrayStream( 
								indexStore, albumDataSQLWithPlaceholders,
								albumIdSubArray )

					end )

	-- 3c) Run each of the albumDataSubsetMap streams serially on index store to
	--     produce a stream of album data subset maps
	local albumDataSubsetMapStream =
			streamOfAlbumDataSubsetMapStreams
				:flatMerge( 1 )

	-- 3d) start album data sub map result stream with empty initial result
	albumDataSubsetMapStream =
		Rx.Observable.concat( 
			Rx.Observable.just( {}, {} ), albumDataSubsetMapStream )

	-- 4) return a combined stream that emits these tuples
	-- 
	-- 		albumIdMap, {}, {}
	-- 		albumIdMap, albumDataSubsetMap1, noAlbumDataSubsetMap1
	-- 		...
	-- 		albumIdMap, albumDataSubsetMapN, noAlbumDataSubsetMapN
	return Rx.Observable.combineLatest( 
					albumIdMapOnceStream, albumDataSubsetMapStream )

end

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

local kAlbumIdsWithAssetsSQL =
		[[SELECT DISTINCT( albumId ) 
		  FROM album_asset_v2 
		  GROUP BY albumId]]

--------------------------------------------------------------------------------
-- kNumAlbumIdsToCountEachQuery is the number of album counts to update at a
-- time
-- 
-- Album counts are a low priority UI query and since UI query are FIFO, the
-- best policy for low priority UI query is to keep the run time under 1/16th
-- of a second.
-- 
-- OTOH, we want the album tree to update as fast as possible, so we also try
-- to make the album count as high as possible.
-- 
-- 2019.12.10 bkruse
-- 
-- Based on a 2014 Mbp pro with SSD, I was finding 32 albums at a time on a 
-- 290K asset db with 575+ albums containing 200K assets was averaging about 
-- 1/16 sec.

local kNumAlbumIdsToCountEachQuery = 32

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

local kAlbumSQLPlaceholder= 
		string.format( "%s?", string.rep( "?,", kNumAlbumIdsToCountEachQuery - 1 ) )

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

local kAlbumCountsForAlbumIdsSQL =
	string.format( 
		[[SELECT albumId, COUNT( DISTINCT album_asset_v2.assetId )
		  FROM album_asset_v2 INNER JOIN assets USING( assetId )
		  WHERE albumId IN( %s ) AND assets.isMedia = 1
		  GROUP BY albumId]], kAlbumSQLPlaceholder )

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

local function albumCountSubsetMapsStream( indexStore )

	return albumDataSubsetMapForAlbumIdsStream( 
			indexStore, kAlbumIdsWithAssetsSQL,
			kAlbumCountsForAlbumIdsSQL, kNumAlbumIdsToCountEachQuery )

end

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

local kIncrementalAlbumCountUpdateCalmFactor = 1 / 5 -- 5 times per second

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

local kMinSecondsBetweenUpdateAllAlbumCounts = 1.0

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

function WFCatAlbums.getAlbumCounts( wfCatalog )

	local indexStore = wfCatalog:indexStore()

	-- getAlbumCounts originally computed the entire album tree counts in a
	-- single SQL query. That was expensive (ave 1 sec) for 100K+ albums in 50+
	-- album trees. This expense caused laggy index store behavior while the
	-- index store was being updated because the index store would frequently
	-- block for a second to run the full album tree count query. This blocking
	-- would come at the expense of other UI queries (such as grid contents)
	-- and also at the expense of harvesting doc store changes into index store
	-- tables.
	--
	-- It's been redesigned to collect the counts n albums at a time and
	-- assemble the counts at the end. In addition, it ensures that the
	-- assembly runs at a limited rate.
	--
	-- These changes ensure that the we never block the index store for more
	-- than a .1 s at a time and that other UI index store queries and
	-- harvesting operations can be interleaved with the queries used to
	-- assemble the album counts. 
	-- 

	-- First, set up a stream that fires an increasing integer whenever the
	-- album counts might have changed. 
	local needsNewAlbumCountsSignalStream =
			rateLimitedSQLTablesUpdatedSignalStream( 
				indexStore, kMinSecondsBetweenUpdateAllAlbumCounts,
				"album_asset_v2", "assets" )

	local desc = "album counts"

	-- Now use needsNewAlbumCountsSignalStream to schedule calls to
	-- albumCountSubsetMapsStream. albumCountsOnceStream builds up the album count
	-- tree incrementally.
	return rateLimitedAlbumDataStream(
				desc,
				needsNewAlbumCountsSignalStream, 
				kIncrementalAlbumCountUpdateCalmFactor,
				kMinSecondsBetweenUpdateAllAlbumCounts,
				albumCountSubsetMapsStream, 
					indexStore )

end

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

local kAllAlbumIdsWithAssetsSQL =
		[[SELECT DISTINCT( albums.albumId )
		  FROM albums, album_asset_v2
		  WHERE albums.albumId = album_asset_v2.albumId AND 
		  		subtype = 'collection']]

--------------------------------------------------------------------------------
-- see kNumAlbumIdsToCountEachQuery for more info about
-- kNumAlbumIdsToGetImplicitAlbumCoversEachQuery
--------------------------------------------------------------------------------

local kNumAlbumIdsToGetImplicitAlbumCoversEachQuery = 32

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

local kImplicitAlbumSQLPlaceholder= 
		string.format( "%s?", string.rep( "?,", kNumAlbumIdsToGetImplicitAlbumCoversEachQuery - 1 ) )

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

local kImplicitAlbumCoversSQL = string.format(
			[[SELECT albumId, 
						( SELECT assets.docId 
						  FROM assets, album_asset_v2 
						  WHERE assets.assetId = album_asset_v2.assetId AND
								  albums.albumId = album_asset_v2.albumId AND 
								  assets.isMedia = 1 
						  ORDER BY captureDate ASC, filenameLC ASC,
						  			serverCreated ASC )
			  FROM albums 
			  WHERE albumId IN( %s )]], kImplicitAlbumSQLPlaceholder )

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

local kAllProjectAlbumIdsWithAssetsSQL = 
		[[SELECT DISTINCT( project_albums.albumId )
		  FROM project_albums, album_asset_v2
		  WHERE project_albums.albumId = album_asset_v2.albumId]]

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

local kImplicitProjectAlbumCoversSQL = string.format(
			[[SELECT albumId, 
						( SELECT assets.docId 
						  FROM assets, album_asset_v2 
						  WHERE album_asset_v2.albumId = 
						  			project_albums.albumId AND
						  		assets.assetId = album_asset_v2.assetId AND 
						  		assets.isMedia = 1
						  ORDER BY captureDate ASC, filenameLC ASC,
						  			serverCreated ASC )
			  FROM project_albums
			  WHERE albumId IN( %s )]], kImplicitAlbumSQLPlaceholder )

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

local function implicitCoverSubsetMapsStream( indexStore, maybeAlbumType )

	local albumIdsWithAssetsSQL =
    		maybeAlbumType == 'project' and
				kAllProjectAlbumIdsWithAssetsSQL or kAllAlbumIdsWithAssetsSQL

	local implicitCoverSQL =
    		maybeAlbumType == 'project' and
				kImplicitProjectAlbumCoversSQL or
				kImplicitAlbumCoversSQL

	return albumDataSubsetMapForAlbumIdsStream( 
			indexStore, albumIdsWithAssetsSQL, implicitCoverSQL,
			kNumAlbumIdsToGetImplicitAlbumCoversEachQuery )

end

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

local kIncrementalAlbumCoverUpdateCalmFactor = 1 / 5 -- 5 times per second

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

local kMinSecondsBetweenUpdateAllAlbumCovers = 2.0

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

local function implicitCoversStream( indexStore, maybeAlbumType )

	-- Originally, implicit album covers were computed for the entire album
	-- tree in a single SQL query. That was expensive (ave 1 sec) for 200K+
	-- assets spread over 575+ albums. 
	-- 
	-- While the index store updated, the full tree implicit cover query would
	-- prevent the index store from updating the contents and executing higher
	-- priority queries. Both of those operations drive grid contents, which 
	-- are much more important when a customer is interacting with the grid.
	--
	-- Therefore, implicit covers are collected
	-- kNumAlbumIdsToGetImplicitAlbumCoversEachQuery at a time and the covers
	-- are emitted incrementally.
	--
	-- These changes ensure that the we never block the index store for more
	-- than a .1 s at a time and that other higher priority index store
	-- operations can be interleaved with the queries used to assemble the
	-- implict album covers.

	local albumIdSQLTableName =
			maybeAlbumType == 'project' and "project_albums" or "albums"

	-- First, set up a stream that fires an increasing integer whenever the
	-- implicit album covers might have changed. 
	local needNewImplicitAlbumCoversSignalStream =
			rateLimitedSQLTablesUpdatedSignalStream( 
				indexStore, kMinSecondsBetweenUpdateAllAlbumCounts,
				"album_asset_v2", albumIdSQLTableName )

	-- Now use needNewImplicitAlbumCoversSignalStream to schedule calls to
	-- implicitCoverSubsetMapsStream. implicitCoverSubsetMapsStream builds up the
	-- implicit album cover tree incrementally
	local desc = 
			maybeAlbumType == 'project' and 
				"implicit project covers" or "implicit album covers"

	return rateLimitedAlbumDataStream(
				desc,
				needNewImplicitAlbumCoversSignalStream,
				kIncrementalAlbumCoverUpdateCalmFactor,
				kMinSecondsBetweenUpdateAllAlbumCovers, 
				implicitCoverSubsetMapsStream,
					indexStore, maybeAlbumType )

end

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

function WFCatAlbums.getAlbumCovers( wfCatalog, maybeAlbumType )

	local albumsTable = 
			maybeAlbumType == 'project' and 'project_albums' or 'albums'

	local explicitCoversQuery = 
			wfCatalog
				:indexStore()
				:query {
					sql = string.format( [[SELECT %s.docId, %s.albumId, assets.docId
								FROM %s 
								LEFT JOIN assets
								ON %s.coverId == assets.assetId ]], albumsTable, albumsTable, albumsTable, albumsTable ),
					calm = 0.5
				}
		
	local implicitCoversCaptureDateFileNameQuery = 
				implicitCoversStream( wfCatalog:indexStore(), maybeAlbumType )
	
	return Observable.combineLatest(
		explicitCoversQuery,
		implicitCoversCaptureDateFileNameQuery )
		:map( function( explicitCoversRows, implicitCoversCaptureDateFileNameByAlbumId )
			local covers = {}
			
			for _, row in ipairs( explicitCoversRows ) do
				local albumDocId = row[ 1 ]
				local albumId = row[ 2 ]
				local explicitCoverId = row[ 3 ]
				
				-- prefer explicit cover id
				local coverId = explicitCoverId
				
				-- then implicit by capture date
				if not coverId then
					coverId = implicitCoversCaptureDateFileNameByAlbumId[ albumId ]
				end
				
				covers[ albumDocId ] = coverId
				
			end
			
			return covers
		end )
		:distinctUntilChanged( nil, distinctTableCompare )
   
end

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

function WFCatAlbums.getProjectAlbumCovers( wfCatalog )

	return WFCatAlbums.getAlbumCovers( wfCatalog, "project" )
 
end

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

function WFCatAlbums.getSizeOfAlbumsForLocalUse( wfCatalog, albumIds, includeLocalFiles )
	
	check( 'WFCatalog', 'table', '?boolean' )
	
	local _albumUuids
	
	return WFCatIdUtils.getOzAlbumIdsForLocalId( wfCatalog, albumIds )
	
		:flatMap( function( albumUuids )
	
			_albumUuids = albumUuids
	
			-- Get the size of all non-local originals in the album.
	
			local sql = [[SELECT SUM( originalSize )
							FROM assetPathsAndInfo
								INNER JOIN album_asset_v2 ON album_asset_v2.assetId = assetPathsAndInfo.docId
								INNER JOIN albums ON albums.albumId = album_asset_v2.albumId
								INNER JOIN CARRAY( ?, ?, 'char*' ) AS au ON au.value = albums.albumId
							WHERE originalVolumeGuid IS NULL AND ozHasOriginal = 1 ]]
	
			if includeLocalFiles then
			
				-- Get the size of all originals in the album.
			
				sql = [[SELECT SUM( originalSize )
							FROM assetPathsAndInfo
								INNER JOIN album_asset_v2 ON album_asset_v2.assetId = assetPathsAndInfo.docId
								INNER JOIN albums ON albums.albumId = album_asset_v2.albumId
								INNER JOIN CARRAY( ?, ?, 'char*' ) AS au ON au.value = albums.albumId ]]
			
			end
	
			return wfCatalog:indexStore():queryOnce {
					sql = sql,
					quiesceUntilUpToDate = true,
					albumUuids,
					#albumUuids,
				}
			
		end )
		
		:take( 1 )
		
		:flatMap( function( originalSizeRows )
		
			-- Get the size of all non-local proxies in the album.
		
			local sql = [[SELECT SUM( proxySize )
							FROM assetPathsAndInfo
								INNER JOIN album_asset_v2 ON album_asset_v2.assetId = assetPathsAndInfo.docId
								INNER JOIN albums ON albums.albumId = album_asset_v2.albumId
								INNER JOIN CARRAY(?,?,'char*') AS au ON au.value = albums.albumId
								WHERE proxyVolumeGuid IS NULL ]]
	
			if includeLocalFiles then
			
				-- Get the size of all proxies in the album.
			
				sql = [[SELECT SUM( proxySize )
							FROM assetPathsAndInfo
								INNER JOIN album_asset_v2 ON album_asset_v2.assetId = assetPathsAndInfo.docId
								INNER JOIN albums ON albums.albumId = album_asset_v2.albumId
								INNER JOIN CARRAY(?,?,'char*') AS au ON au.value = albums.albumId ]]
			
			end
		
			return wfCatalog:indexStore():queryOnce {
					sql = sql,
					quiesceUntilUpToDate = true,
					_albumUuids,
					#_albumUuids,
				}
				
				:map( function( proxySizeRows )
				
					local sizeOfOriginals = originalSizeRows[ 1 ][ 1 ] or 0
					local sizeOfProxies = proxySizeRows[ 1 ][ 1 ] or 0

					return sizeOfOriginals + sizeOfProxies

				end )
			
		end )

		:takeLast( 1 )

end

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

local function toBool( int )

	return int and int > 0

end

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

function WFCatAlbums.getAlbumMarkedForOfflineStatus( catalog )

	return catalog:indexStore():query( {
			sql = [[ SELECT albumId, isMarkedForOffline 
					 FROM albums ]],
			calm = 1
		} )
		
		:distinctUntilChanged( nil, distinctTableCompare )
		
		:map( function( rows )
		
			local offlineStatusByAlbumId = {}
		
			for _, row in ipairs( rows ) do
		
				local albumId = row[ 1 ]
				local offlineStatusInt = row[ 2 ]
				
				offlineStatusByAlbumId[ albumId ] = toBool( offlineStatusInt )
		
			end
		
			return offlineStatusByAlbumId

		end )

end

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

function WFCatAlbums.getAlbumListForAssetIds( wfCatalog, assetIds )
	
	if #assetIds > 1 then
	
		return Observable.just( {} )
	
	end
	
	local docId = assetIds[ 1 ]
	
	return wfCatalog:indexStore():query {
		sql = [[SELECT DISTINCT albums.docId, name, albums.isMarkedForOffline
					FROM albums
					INNER JOIN album_asset_v2 using( albumId )
					INNER JOIN assets ON assets.assetId = album_asset_v2.assetId
					WHERE assets.docId = ?
					ORDER BY nameLC COLLATE i18n_collate ASC]],
			calm = 2,
			docId
		}
		:distinctUntilChanged( nil, distinctTableCompare )
		:map( function( rows )
			
			local albumNames = {}
			
			for _, albumAsset in ipairs( rows ) do
			
				local theDocId = albumAsset[ 1 ]
				local albumName = albumAsset[ 2 ]
				local isMarkedForOffline = toBool( albumAsset[ 3 ] )
				
				table.insert( albumNames, {
					docId = theDocId,
					albumName = albumName,
					isMarkedForOffline = isMarkedForOffline
				} )
	
			end
			
			return albumNames
		
		end )

end

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

--[[
	\brief Fetches the list of albums

	\param wfCatalog (WFCatalog) The catalog.

	\return <code>Observable< ></code> An Observable object sends a value whenever
		the albums change.  The result is an array of album nodes, each containing
		the id (as a local integer id), the album name, the parent album (if any),
		subtype, and cover
--]]

function WFCatAlbums.getAlbumsList( wfCatalog )

	return flattenedAlbumHierarchy( wfCatalog, false )
		:pluck( 'albums' )

end

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

local function reorderNewAlbumAndSiblings( newAlbumId, siblings )
	
	-- compute order field
	local ordering = {}

	for i, album in ipairs( siblings or {} ) do
		ordering[ i ] = {
			id = album.id,
			-- in this case it's the local id, but that's ok
			touched = false,
			order = album.sortOrder,
		}
	end

	ordering[ #ordering + 1 ] = {
		id = newAlbumId,
			-- in this case it's the uuid but that's ok too
		touched = true,
	}

	local updatedOrders = WFAlbumOrdering.adjustOrderFields( ordering )
	
	local newAlbumOrder
	local deleteAlbumAssetOverlays = {}
	for i, updatedOrder in pairs( updatedOrders ) do
		if ordering[ i ].id == newAlbumId then
			newAlbumOrder = updatedOrder
		else
		
			local overlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( ordering[ i ].id )
			overlay.order = updatedOrder
			deleteAlbumAssetOverlays[ #deleteAlbumAssetOverlays + 1 ] = overlay
			
		end
	end
	
	return newAlbumOrder, deleteAlbumAssetOverlays
end

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

local function createAlbum_imp( wfCatalog, albumSubtype, albumName, optParentId, optHidden, optImportSource, optDocId, optManifestSourceId )
	
	albumName = maybeMakeAlbumNameValid( albumName )
	
	local docStore = wfCatalog:docStore()
	
	local maybeGetParentDoc = Observable.just( {} )
	
	if optParentId then
	
		maybeGetParentDoc = docStore:docValue( optParentId )
	
	end
	
	return maybeGetParentDoc
		:take( 1 )
		:flatMapLatest( function( parentDoc )

			local optParentDocId = parentDoc._id

			if optParentId and not isValidParent( parentDoc ) then 
				return Observable.throw( 'InvalidParent' )
			end

			local albumId = optDocId or KSIdGenerator.uuidInLowercaseHex()
			local updateTimestamp = KSDate.timeInIso8601Format( nil, false )
			
			local baseDocValue = WFCatUtils.createBaseOverlayWithUserCreatedFields( albumId )

			local albumDocValue = {
				type = 'album',
				subtype = albumSubtype,
				name = albumName,
				hideInOrganize = optHidden, -- use 'not not' if we want ths field to always exist
				parent = optParentDocId and {
					id = optParentDocId,
				},
				importSource = optImportSource,
				_localOnly = {
					updateTimeStamp = updateTimestamp,
					manifest = optManifestSourceId and {
						[ optManifestSourceId ] = true
					},
				},
			}
	
			WFTableMerge.applyOverlay( albumDocValue, baseDocValue )
	
			local result = docStore:setDocValue( albumDocValue )
	
			-- Wait until all of the above operations have completed,
			-- then extract the local doc ID form doc store.
	
			return result
				:takeLast( 0 )
				:concat( docStore:docValue( albumId ) )
				:map( function( docValue )
					return docValue._localId, albumId, docValue
				end )

	end )
	
end

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

--[[!
	\brief Creates an album in the provided catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumName (string) The name of the new album, will truncate to 250 chars as Oz does not allow longer names.
	\param optParentId (string|int|nil) The id of the parent.
	\param optHidden (boolean) true if this album should be hidden from the albums list
	\param optUpdateTimestamp (iso8601date|nil) The date to put in _localOnly.updateTimeStamp.
	\param optDocId (string) The id of the new document if we already know what it should be
	\param optManifestSourceId (string) The id of the source in case request is coming via manifest processing

	\return <code>Observable< ></code> An Observable object that signals the albumId
		and then completes when operation is completed; or resolves to an error
		if an error occurs.
]]--
function WFCatAlbums.createAlbum( wfCatalog, albumName, optParentId, optHidden, optImportSource, optDocId, optManifestSourceId )

	check( 'WFCatalog', '+string', '?+string|number', '?boolean', "?table", "?uuid", '?+string' )
	
	return createAlbum_imp( wfCatalog, 'collection', albumName, optParentId, optHidden, optImportSource, optDocId, optManifestSourceId )
end

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

--[[!
	\brief Creates a folder (album of subtype 'collection_set') in the provided catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param folderName (string) The name of the new folder, will truncate to 250 chars as Oz does not allow longer names.
	\param optParentId (string|int|nil) The id of the parent.
	\param optUpdateTimestamp (iso8601date|nil) The date to put in _localOnly.updateTimestamp.
	\param optDocId (string) The id of the new document if we already know what it should be
	\param optManifestSourceId (string) The id of the source in case request is coming via manifest processing

	\return <code>Observable< ></code> An Observable object that signals the albumId
		and then completes when operation is completed; or resolves to an error
		if an error occurs.
]]--
function WFCatAlbums.createFolder( wfCatalog, folderName, optParentId, optImportSource, optDocId, optManifestSourceId )
	
	check( 'WFCatalog', '+string', '?+string|number', "?table", "?uuid", '?+string' )
	
	return createAlbum_imp( wfCatalog, 'collection_set', folderName, optParentId, false, optImportSource, optDocId, optManifestSourceId )

end

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

local function removeAlbumAssetsAndAlbumForDocId( wfCatalog, albumDocId )
	
	return wfCatalog:indexStore():queryOnce {
			-- get all the album_assets
			sql = [[SELECT docId FROM album_asset_v2 WHERE albumId = ?]],
			quiesceUntilUpToDate = true,
			singleColumnQuery = true,
			albumDocId
		}
		:flatMapLatest( function( docIds )

			return WFCatDelete.deleteDocs( wfCatalog:docStore(), docIds, true )
			
		end )
		:takeLast( 1 )
		:flatMapLatest( function()
			return wfCatalog:docStore():deleteDoc( albumDocId )
		end )
end

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

--[[!
	\brief Deletes an album or folder, but ignores all child album/folders. This should generally
		be used only internally unless you really know you want it.  WFCatAlbums.deleteAlbumOrFolder
		is probably more appropriate for most uses.
		
	\param wfCatalog (WFCatalog) The catalog
	\param albumDocId (string) the UUID of the album/folder to delete.
	
	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--
function WFCatAlbums.deleteIndividualAlbum( wfCatalog, albumDocId )
	check( "WFCatalog", "+string" )
	
	-- PR #3312056: Remove any album share first, then delete the album, too.
	-- Regardless of success/failure of unsharing the album, we're still going
	-- to delete it, hence the catch()
	
	return  WFCatCollaboration.deleteSharedAlbum( wfCatalog, albumDocId )
				:catch( Observable.empty() )
				:concat( Observable.just() )
				:takeLast( 1 )

					-- Remove all album assets from the album
				:flatMapLatest( function()
					return removeAlbumAssetsAndAlbumForDocId( wfCatalog, albumDocId ) 
				end )
				:takeLast( 0 )

end



--------------------------------------------------------------------------------
--[[!
	\brief Deletes an album/folder and all child albums/folders from this catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string|int) The unique or local ID of the album/folder to delete.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--

function WFCatAlbums.deleteAlbumOrFolder( wfCatalog, albumId )

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

	if type( albumId ) == 'string' and #albumId < 32 then
		return Observable.throw( "WFCatAlbums:deleteAlbumOrFolder: The albumId must be a 32-char uuid", 2 )
	end

	return getAlbumAndChildren( wfCatalog, albumId )
		:selectMany( function( album )
		
			local albumDocId = album.fullDocId
			return WFCatAlbums.deleteIndividualAlbum( wfCatalog, albumDocId )
		end )
		:call( runUntilDone )

end

--------------------------------------------------------------------------------
--[[!
	\brief Deletes only the albums/folders specified by id in the params.  It will not 
		remove any children from this catalog that are not explicitly specified in the list
		of albums to remove so you should fully consider
	
	\param wfCatalog (WFCatalog) The catalog.
	\param albumIds (table) Table of local ids of the albums and/or folders to delete.
	
	\return <code>Observable<></code> An Observable object completes when the delete
		operation is completed; or resolves to an error if an error occurs.
]]--
function WFCatAlbums.deleteIndividualAlbums( wfCatalog, folderAndAlbumIds )
	
	check( "WFCatalog", "table" )
	
	return Observable.from( folderAndAlbumIds )
		:flatMap( function( id )
			return WFCatIdUtils.albumUuidForMaybeLocalId( wfCatalog, id )
				:flatMap( function( albumId )
					if albumId then
						return WFCatAlbums.deleteIndividualAlbum( wfCatalog, albumId )
					else 
						return Observable.just()
					end
				end )
		end )
		:toArray()
		:call( runUntilDone )
end

--------------------------------------------------------------------------------
--[[!
	\brief Deletes only the albums/folders specified by id in the params.  It will not 
		remove any children from this catalog that are not explicitly specified in the list
		of albums to remove so you should fully consider
	
	\param wfCatalog (WFCatalog) The catalog.
	\param albumIds (table) Table of local ids of the albums and/or folders to delete.
	
	\return <code>Observable<></code> An Observable object completes when the delete
		operation is completed; or resolves to an error if an error occurs.
]]--
function WFCatAlbums.deleteProjectAlbums( wfCatalog, projectAlbumIds )
	
	check( "WFCatalog", "table" )
	-- SPAM [matthewj 2018-08-15]: Should this be moved to NmbConnectors (similar to how deleting a 
	-- shared album is in WFCatCollaboration)?
	return Observable.from( projectAlbumIds )
		:flatMap( function( id )
			return WFCatIdUtils.albumUuidForMaybeLocalId( wfCatalog, id )
				:flatMap( function( albumId )
					if albumId then
						return removeAlbumAssetsAndAlbumForDocId( wfCatalog, albumId )
					else 
						return Observable.just()
					end
				end )
		end )
		:toArray()
		:call( runUntilDone )
end

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

-- allowList is from the sort orders listed in
-- http://apidocs.adbecrsl.com/v2/data-model/album/collection.html
local allowListedAssetSortOrders = KSTableUtils.mapArrayToSet {
	"",
	"captureDateAsc",
	"captureDateDesc",
	"customAsc",
	"customDesc",
	"fileNameAsc",
	"fileNameDesc",
	"importTimestampAsc",
	"importTimestampDesc",
	"ratingAsc",
	"ratingDesc",
	"userUpdatedAsc",
	"userUpdatedDesc",
}

local function performAssetSortOrderValidation( hierarchy, albumId, assetSortOrder )
	
	local album = getAlbumFromHierarchyById( hierarchy, albumId )
			
	if not album then
		return false, "AlbumDoesNotExist"
	end
	
	if assetSortOrder ~= nil and not allowListedAssetSortOrders[assetSortOrder] then
		return false, "AssetSortOrderIsInvalid"
	end
	
	return true
end

function WFCatAlbums._validateAssetSortOrder( hierarchyObservable, albumId, assetSortOrder )

	return hierarchyObservable
		:flatMap( function( hierarchy )
			
			local isValid, maybeError = performAssetSortOrderValidation( hierarchy, albumId, assetSortOrder )
			
			if not isValid and maybeError then
				return Observable.throw( maybeError )
			end
			
			return Observable.just( hierarchy )
		end )
end
--------------------------------------------------------------------------------
--[[!
	\brief Sets the asset sort order an album in this catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string or number) The id of the album to modify.
	\param assetSortOrder (string) The asset sort order that is documented at 
		http://apidocs.adbecrsl.com/v2/data-model/album/collection.html
		This will be one of the following values:
			- captureDateAsc
			- captureDateDesc
			- customAsc
			- customDesc
			- fileNameAsc
			- fileNameDesc
			- importTimestampAsc
			- importTimestampDesc
			- ratingAsc
			- ratingDesc
			- userUpdatedAsc
			- userUpdatedDesc

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--
function WFCatAlbums.setAssetSortOrder( wfCatalog, albumId, assetSortOrder )

	check( "WFCatalog", '+string|number', '?string' )

	local docStore = wfCatalog:docStore()
		
	return Observable.combineLatest( docStore:docValue( albumId ),
			WFCatAlbums._validateAssetSortOrder( 
					flattenedAlbumHierarchy( wfCatalog, true ):take( 1 ), albumId, assetSortOrder ) )
	
		:flatMapLatest( function( albumDoc )
			
			local currentAssetSortOrder
			
			if albumDoc._spaceId then
			
				-- For space-feed documents, the asset sort order can be changed by the contributor/viewer
				-- but it is a local-only change; they cannot overwrite the album order set by the album owner.
			
				currentAssetSortOrder = derefOpt( albumDoc, '_localOnly', 'universal', 'assetSortOrder' )
				
			else
				currentAssetSortOrder = albumDoc.assetSortOrder
			end
			
			if currentAssetSortOrder == assetSortOrder then
				-- There's no need to set the sort order if it's already set to the
				-- desired value.
				return Observable.empty()
			end
			
			local baseOverlay
			
			if albumDoc._spaceId then 
			
				-- For space-feed docs, we're not allowed to sync changes to album documents.
				-- Instead of changing the assetSortOrder document field, we track the sort
				-- order in the localOnly payload (which is not synced).
			
				baseOverlay = {
					_id = albumId
				}
			
				if assetSortOrder ~= nil and assetSortOrder ~= "" then
					KSTableUtils.setNested( baseOverlay, assetSortOrder, '_localOnly', 'universal', 'assetSortOrder' )
				else
					KSTableUtils.setNested( baseOverlay, { _delete = true }, '_localOnly', 'universal', 'assetSortOrder' )
				end
			
			else
			
				baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( albumId )
			
				if assetSortOrder ~= nil and assetSortOrder ~= "" then
					baseOverlay.assetSortOrder = assetSortOrder
				else
					baseOverlay.assetSortOrder = { _delete = true }
				end
			
			end
				
			return docStore:applyOverlay( baseOverlay )
				:takeLast( 0 )
			
		end )
		
end


function WFCatAlbums._validateRenameAlbumInHierarchy( hierarchyObservable, albumId, albumName )
	
	return hierarchyObservable
		:flatMap( function( hierarchy )
			
			local album = getAlbumFromHierarchyById( hierarchy, albumId )
	
			if not album then
				return Observable.throw( "AlbumDoesNotExist" )
			end
	
			local parentId = album.parentId or ROOT_ALBUM_PARENT_ID
	
			-- ignore collisions with this album's own name so the user can change its capitalization.
			local ignoreAlbumId = album.id

			local nameIsUnique = isAlbumNameUniqueAmongSiblings( albumName, album.subtype, hierarchy.localIdToChildrenAlbumData[ parentId ], ignoreAlbumId )

			if not nameIsUnique then
				return Observable.throw( "NameExists" )
			end
	
			return Observable.just( hierarchy )
	
		end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Renames an album or folder in this catalog.

	It is an error to rename an album or folder to match the name of an existing album,
	regardless of case.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string) The unique or local ID of the album/folder to rename.
	\param name (string) The new name for the album/folder.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--

function WFCatAlbums.renameAlbumOrFolder( wfCatalog, albumId, albumName )

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

	albumName = maybeMakeAlbumNameValid( albumName )
	
	local docStore = wfCatalog:docStore()

	return WFCatAlbums._validateRenameAlbumInHierarchy( flattenedAlbumHierarchy( wfCatalog, true ):take( 1 ), albumId, albumName )
		:flatMapLatest( function()
			
			local baseOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( albumId )
			
			baseOverlay.name = albumName
			
			return docStore:applyOverlay( baseOverlay )
			:takeLast( 0 )
			
		end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Converts a hidden album (i.e., ad-hoc shares for clients) into a 
		visible album.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (number) The local ID of the album/folder to rename.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--

function WFCatAlbums.unhideAlbum( wfCatalog, albumId )

	check( "WFCatalog", 'number' )

	local docStore = wfCatalog:docStore()
	
	return docStore:localIdToDocId( albumId )
		:flatMap( function( docId )
		
			return docStore:applyOverlay( {
				_id = docId,
				hideInOrganize = false,
			} )

		end )
		
		:takeLast( 0 )
end

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

local function wouldMoveCreateACycle( hierarchy, albumId, parent )

	if not parent then
		return false -- if we're moving to the top level, it can't create a loop
	end
	
	local visitedNodes = {}
	
	local node = parent
	while node do
		if node.id == albumId then
			-- if the node we're looking at now is the album we're moving
			-- we've got a loop because we started at its future parent
			return true
		end
		
		if visitedNodes[ node ] then
			-- SPAM [matthewj 2017-01-30]: In the edge case that there's already a 
			-- cycle in the parentage but the moved node isn't the one creating it
			-- we return true here to be overly cautious
			return true
		end
		visitedNodes[ node ] = true
		
		-- get the next node to look at
		if node.parentId then
			node = getAlbumFromHierarchyById( hierarchy, node.parentId )
		else
			node = nil
		end
		
	end
	
	return false
end

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

local function performReparentValidation( hierarchy, albumId, optNewParentId )
	
	local album = getAlbumFromHierarchyById( hierarchy, albumId )
			
	if not album then
		return false, "AlbumDoesNotExist"
	end

	local parent = getAlbumFromHierarchyById( hierarchy, optNewParentId )
	
	if not isValidParent( parent ) then
		return false, "InvalidParent"
	end
	
	if wouldMoveCreateACycle( hierarchy, album.id, parent ) then
		return false, "CycleDetected"
	end
	
	local albumName = album.name
	
	local parentId = parent and parent.id or ROOT_ALBUM_PARENT_ID
	local siblingAlbums = hierarchy.localIdToChildrenAlbumData[ parentId ]
	
	local nameIsUnique = isAlbumNameUniqueAmongSiblings( albumName, album.subtype, siblingAlbums, albumName )
	
	if not nameIsUnique then
		return false, "NameExists"
	end
	
	return true
end

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

function WFCatAlbums._validateReparentAlbumInHierarchy( hierarchyObservable, albumId, optNewParentId )
	
	return hierarchyObservable
		:flatMap( function( hierarchy )
			
			local isValid, maybeError = performReparentValidation( hierarchy, albumId, optNewParentId )
			
			if not isValid and maybeError then
				return Observable.throw( maybeError )
			end
			
			return Observable.just( hierarchy )
		end )

end

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

function WFCatAlbums._validateReparentMultipleAlbumsInHierarchy( hierarchyObservable, albumIds, optNewParents )
	
	return hierarchyObservable
		:flatMap( function( hierarchy )
			
			local maybeErrors = {} 
			for _, albumId in ipairs( albumIds ) do
				
				local maybeParentId 
				if type( optNewParents ) == 'number' or type( optNewParents ) == 'string' then
					maybeParentId = optNewParents
				elseif type( optNewParents ) == 'table' then
					maybeParentId = optNewParents[ albumId ]
				end
				
				local isValid, maybeError = performReparentValidation( hierarchy, albumId, maybeParentId )
				
				if not isValid then
					maybeErrors[ albumId ] = maybeError
				end
			end
			
			maybeErrors = next( maybeErrors ) and maybeErrors or nil
			
			return Observable.just( hierarchy, maybeErrors )
		end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Moves an album or folder onto a different parent

	It is an error to move an album/folder a parent that already has an album with that
	same name. Only folders are currently allowed to contain children so trying to reparent 
	to an album will throw an error.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string) The unique or local ID of the album/folder to rename.
	\param optNewParentId (string|int|nil) The new parent or nil to make this a root album/folder
	\param optOrder (array|nil) Array of album ids (local or global) in the order they should
		be in that new parent.  Any album ids that are specified that are not in that parent
		album are ignored.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--

function WFCatAlbums.reparentAlbumOrFolder( wfCatalog, albumId, optNewParentId )

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

	return WFCatAlbums._validateReparentAlbumInHierarchy( flattenedAlbumHierarchy( wfCatalog, true ):take( 1 ), 
															albumId, optNewParentId )
		:flatMap( function( hierarchy )
			local parent = getAlbumFromHierarchyById( hierarchy, optNewParentId )
			
			local newParentOverlay
			if parent then
				newParentOverlay = {
					id = parent.fullDocId,
				}
			else
				newParentOverlay = {
					_delete = true,
				}
			end

			local parentId = parent and parent.id or ROOT_ALBUM_PARENT_ID
			local siblingAlbums = hierarchy.localIdToChildrenAlbumData[ parentId ]
			
			local newAlbumOrder, siblingAlbumOverlays = reorderNewAlbumAndSiblings( albumId, siblingAlbums )
	
			siblingAlbumOverlays[ #siblingAlbumOverlays + 1 ] = {
				_id = albumId,
				parent = newParentOverlay,
				order = newAlbumOrder,
			}

			return WFRxBatchedBulkApplyOverlay( "reparent album or folder", wfCatalog:docStore(), siblingAlbumOverlays )

		end )
		:takeLast( 0 )

end

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

function WFCatAlbums.reparentMultipleAlbumsOrFolders( wfCatalog, albumIds, optNewParents )
	
	check( 'WFCatalog', 'table' )
	
	return WFCatAlbums._validateReparentMultipleAlbumsInHierarchy( flattenedAlbumHierarchy( wfCatalog, true ):take( 1 ),
																	albumIds, optNewParents )
		:flatMap( function( hierarchy, maybeErrors )
			
			local albumParentOverlays = {}
			
			-- Related to NMB-6314: if attempting to reparent multiple albums and there is an error, we 
			-- reject the full reparent even if some would succeed. Unfortunately, models can currently 
			-- only pass a string for an error event so we need to return the errors table as a data
			-- event and if nil, this should be interpreted as a success.
			if maybeErrors then 
				return Observable.just( maybeErrors )
			end
			
			-- there isn't any validation error so we should be able to add overlays to move the albums
			for _, albumId in ipairs( albumIds ) do
				
				local maybeParentId 
				if type( optNewParents ) == 'number' or type( optNewParents ) == 'string' then
					maybeParentId = optNewParents
				elseif type( optNewParents ) == 'table' then
					maybeParentId = optNewParents[ albumId ]
				end
				
				local parent = getAlbumFromHierarchyById( hierarchy, maybeParentId )
		
				local overlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( albumId )
				
				if parent then
					overlay.parent = {
						id = parent.fullDocId,
					}
				else
					overlay.parent = {
						_delete = true,
					}
				end
			
				albumParentOverlays[ albumId ] = overlay
			end
				
			local overlayObservable
			if next( albumParentOverlays ) then
				local finalOverlays = {}
				for _, overlay in pairs( albumParentOverlays ) do
					finalOverlays[ #finalOverlays + 1 ] = overlay
				end
				overlayObservable = WFRxBatchedBulkApplyOverlay( "reparentMultipleAlbumsOrFolders", wfCatalog:docStore(), finalOverlays )
			else
				overlayObservable = Observable.just()
			end
			
			-- make sure we return nil (which will be interpreted as 'success')
			return overlayObservable:map( function() return nil end )
		end )
		:takeLast( 1 )
end

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

local function addAdditionalAlbumInfoToAlbumsList( albums, counts, covers, statuses )
 
	local albumsWithMoreData = {}
	
	for _, album in ipairs( albums ) do
		album.syncStatus = statuses[ album.fullDocId ]
		album.count = counts[ album.fullDocId ] or 0
		album.cover = covers[ album.id ]
		
		albumsWithMoreData[ #albumsWithMoreData + 1 ] = album
	end
	
	return albumsWithMoreData
end 

 --------------------------------------------------------------------------------
 
local function combineAdditionalAlbumInfoWithAlbumList( wfCatalog, albumListObservable )
     
	return Rx.Observable.combineLatest( 
		function( albums, counts, covers, statuses )
		
			if next( counts ) or next( covers ) or next( statuses ) then
			
				albums = addAdditionalAlbumInfoToAlbumsList( albums, counts, covers, statuses )
			
			end
			
			return albums
		
		end,
	
	albumListObservable,
	WFCatAlbums.getAlbumCounts( wfCatalog ):startWith( {} ),
	WFCatAlbums.getAlbumCovers( wfCatalog ):startWith( {} ),
	WFCatSyncStatus.getSyncStatusForAlbums( wfCatalog, true --[[ignore album status]] ):startWith( {} ) )
 
 end
 
-----------------------------------------------------------------------
--[[
     \brief Reorders the album assets that's passed in relative to the anchor asset.
	 \param wfCatalog (WFCatalog) The catalog.
	 \param albumId (string or number) The id of the album where the assets are contained in.
	 \param anchorAssetId (number) The asset id relative to which the target asets should be rearranged.
	 \param assetIdsToReorder (table) List of assets (by asset ids each of which is a number) that should be reordered.
	 \param insertWhere (string) Relative location signifying the final placement of the target assets.
	 								It is one of either "before" or "last".
									"before" works for all but inserting at the end in which case,
									"last" should be used instead.
	 \param maybeAllAssetIds (table) Sorted list of asset ids according to the current sort order that's applied on the album.
	 \param optUndoStateToken (optional, number) Token identifying a WFCatUndo state to restore.

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

function WFCatAlbums.reorderAssetsInAlbum( wfCatalog, albumId, anchorAssetId, assetIdsToReorder, insertWhere, maybeAllAssetIds, optUndoStateToken )
	
	local task = WFCatReorderAssetsTask.createReorderingTask( wfCatalog, albumId, anchorAssetId, assetIdsToReorder, insertWhere, maybeAllAssetIds, optUndoStateToken )
	
	local manager = WFCatTaskManager.getCatTaskManager( wfCatalog )
	manager:addToQueue( task )
	
	return task:taskObservable()
	
end

-----------------------------------------------------------------------
--[[
     \brief Reorders the album assets that's passed in relative to the anchor asset but forces full ordering.
	 \param wfCatalog (WFCatalog) The catalog.
	 \param albumId (string or number) The id of the album where the assets are contained in.
	 \param anchorAssetId (number) The asset id relative to which the target asets should be rearranged.
	 \param assetIdsToReorder (table) List of assets (by asset ids each of which is a number) that should be reordered.
	 \param insertWhere (string) Relative location signifying the final placement of the target assets.
	 								It is one of either "before" or "last".
									"before" works for all but inserting at the end in which case,
									"last" should be used instead.
	 \param maybeAllAssetIds (table) Sorted list of asset ids according to the current sort order that's applied on the album.
	 \param optUndoStateToken (optional, number) Token identifying a WFCatUndo state to restore.

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

--]]
function WFCatAlbums.forceReorderAssetsInAlbum( wfCatalog, albumId, anchorAssetId, assetIdsToReorder, insertWhere, maybeAllAssetIds, optUndoStateToken )
	
	local task = WFCatReorderAssetsTask.createForceReorderingTask( wfCatalog, albumId, anchorAssetId, assetIdsToReorder, insertWhere, maybeAllAssetIds, optUndoStateToken )
	
	local manager = WFCatTaskManager.getCatTaskManager( wfCatalog )
	manager:addToQueue( task )
	
	return task:taskObservable()
	
end

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


 --[[
     \brief Returns an Observable producing array of top-level albums, each containing fields:
         id : (int) the uuid of the album
         name : (user string) the name of the album
         path : (user string) The name of the parents and album separated by /
         parentId : (uuid, or nil) the id of the albums parent
         children : (array) array of child albums
         
     \param wfCatalog an Observable( WFCatalog )
--]]     
 function WFCatAlbums.albumsWithSyncStatus( wfCatalog )
 
     -- Fallback in case we fail to log in and get a catalog reference.
 
     if wfCatalog == nil then
         return Rx.Observable.just {}
     end
 
     return combineAdditionalAlbumInfoWithAlbumList( wfCatalog, WFCatAlbums.getAlbumsList( wfCatalog ) )
         
 end
 
--------------------------------------------------------------------------------

local function assetUuidsForMaybeLocalAlbumId( wfCatalog, maybeLocalAlbumId, maybeAlbumType )
	
	check( "WFCatalog", '+string|number' )
	
	local albumsTable = 'albums'
    if maybeAlbumType == 'project' then
        albumsTable = 'project_albums'
    end
	
	if type( maybeLocalAlbumId ) == 'number' then
		
		return wfCatalog:indexStore():queryOnce {
                sql = string.format( [[SELECT DISTINCT album_asset_v2.assetId, album_asset_v2.docId 
                            FROM album_asset_v2 
                            JOIN %s as a ON album_asset_v2.albumId == a.albumId
                            WHERE a.docId = ? ]], albumsTable ),
				maybeLocalAlbumId,
				quiesceUntilUpToDate = true,
			}

			:map( function( rows )
				
				logger:debugf( 'assetUuidsForMaybeLocalAlbumId for localId %s complete w/ %d rows', maybeLocalAlbumId, #rows )
			
				local assetIds = {}
				
				for _, albumAsset in ipairs( rows ) do
				
					local assetId = albumAsset[ 1 ]
					local docId = albumAsset[ 2 ]
		
					assetIds[ assetId ] = docId
		
				end
				
				return assetIds
			
			end )
	end
	
	-- maybeLocalAlbumId is already string
	return wfCatalog:docStore():docValue( maybeLocalAlbumId )
		:flatMapLatest( function( docValue )
			if docValue._empty then
				return Observable.throw( "No such album: " .. tostring( maybeLocalAlbumId ) )
			end
			
			assert( type( docValue._localId ) == 'number' )
			return assetUuidsForMaybeLocalAlbumId( wfCatalog, docValue._localId, maybeAlbumType )
		end )
		
end

WFCatAlbums.getAlbumAssetIds = assetUuidsForMaybeLocalAlbumId

WFCatAlbums.getProjectAlbumAssetIds = function( wfCatalog, maybeLocalAlbumId )
    return assetUuidsForMaybeLocalAlbumId( wfCatalog, maybeLocalAlbumId, 'project' )
end

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

local function assetsWithQuotaErrors( wfCatalog, assetIds )
	
	if not next( assetIds ) then
		return Observable.just( {} )
	end
	
	local idColumn
	local carrayType
	if type( assetIds[ 1 ] ) == 'string' then
		idColumn = 'assets.assetId'
		carrayType = 'char*'
	else
		idColumn = 'assets.docId'
		carrayType = 'int32'
	end
	
	return wfCatalog:indexStore():query {
		sql = string.format( [[SELECT assets.assetId FROM quota_exceeded 
				JOIN assets USING( docId )
				JOIN CARRAY( ?, ? , '%s' ) AS qe ON qe.value == %s]] ,carrayType ,idColumn ),
		assetIds,
		#assetIds,
		singleColumnQuery = true
	}
	:map( function( assetIds )
		-- turn into a map for easy lookup
		local assetIdMap = {}
		for _, assetId in ipairs( assetIds ) do
			assetIdMap[ assetId ] = WFCatOzErrors.Errors.kStorageExceededError
		end
		
		return assetIdMap
	end )
	:take( 1 )

end

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

local function highestOrderForAlbum( wfCatalog, albumId )

	return wfCatalog:indexStore():query( {
		sql = [[SELECT assetId, sortOrder FROM album_asset_v2 
				WHERE albumId == ? and sortOrder IS NOT NULL 
				ORDER BY sortOrder DESC limit 1]],
			albumId,
		quiesceUntilUpToDate = true,
	} )
		
		:map( function( row )
		
			if row and #row >= 1 then
				return row[ 1 ][ 2 ]
			end

		end )
end

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

local function getAssetOrderingObservable( wfCatalog, albumId, assetIds )

	-- Docs will be inserted in order after the last doc with
	-- order. Docs without order will not be affected.
	
	-- Fetch the order of last doc with order.
	return highestOrderForAlbum( wfCatalog, albumId )
		:take( 1 )
		:map( function( highestOrder )
		
			local ordering = {}
			local formerOrder = highestOrder or SORT_PREV_FOR_INSERT_AS_FIRST
			
			for _, assetId in ipairs( assetIds ) do
				ordering[ assetId ] = WFSortOrderStrings.insertAfter_lex64( formerOrder, 'low' )
				formerOrder = ordering[ assetId ]
			end
			
			return ordering
		end )
end

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

--[[!
	\brief Adds assets to an album in this catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string|number) The unique ID of the album to modify.
	\param assetIds (array of strings or integers) An array of unique or local
		IDs for the assets to add.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--

local function maybeAddNewAlbumAssetDocument( newAlbumAssetDocs, assetId,
												albumId, existingAssetIds,
												assetsWithQuotaErrorsMap )

	if existingAssetIds[ assetId ] then

		return newAlbumAssetDocs

	end

	local aaDoc = createAlbumAssetDocument( albumId, assetId )

	if assetsWithQuotaErrorsMap[ assetId ] then

		aaDoc = WFCatOzErrors.addOverlayForOzErrorSubtype( 
					aaDoc, WFCatOzErrors.Errors.kStorageExceededError )

	end

	newAlbumAssetDocs[ #newAlbumAssetDocs + 1 ] = aaDoc

	return newAlbumAssetDocs

end

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

local function makeNewlyCreatedAlbumAssetDocs( albumUuid, assetUuids,
												existingAssetUuids,
												assetsWithQuotaErrorsMap )

	return WFFunctionUtils.reduce( 
			{},
			maybeAddNewAlbumAssetDocument,
				assetUuids, albumUuid, existingAssetUuids,
				assetsWithQuotaErrorsMap )

end

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


function WFCatAlbums.addAssetsToAlbum( wfCatalog, maybeLocalAlbumId, maybeLocalAssetIds )

	check( "WFCatalog", '+string|number', 'table' )

	local albumUuidStream = 
			WFCatIdUtils.albumUuidForMaybeLocalId( wfCatalog, maybeLocalAlbumId )

	local assetUuidsStream = 
			WFCatIdUtils.getOzAssetIdsForLocalIds( wfCatalog, maybeLocalAssetIds )

	local assetsWithQuotaErrorsStream = 
			assetsWithQuotaErrors( wfCatalog, maybeLocalAssetIds )

	local existingAssetUuidsStream = 
			assetUuidsForMaybeLocalAlbumId( wfCatalog, maybeLocalAlbumId )
	
	local albumAssetDocsToCreateStream =
			Observable.combineLatest( 
				makeNewlyCreatedAlbumAssetDocs,
					albumUuidStream, assetUuidsStream,
					existingAssetUuidsStream, assetsWithQuotaErrorsStream )

	return albumAssetDocsToCreateStream
			:flatMapLatest( 
				function( albumAssetDocsToCreate )
			
					if not next( albumAssetDocsToCreate ) then

						return Observable.just()

					end
					
					return WFRxBatchedBulkSetDocValue( 
							"addAssetsToAlbum", wfCatalog:docStore(),
							albumAssetDocsToCreate )
				
			end )
			:tap( function()
				logger:debugf( 'addAssetsToAlbum WFRxBatchedBulkSetDocValue COMPLETE' )
			end )
			:takeLast( 0 )

end

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

--[[!
	\brief Adds assets to an album taking care of ids of album_asset documents already
		on Oz. (Example: while migrating a synced album from Lightroom)

		If the asset is already in the album, do not create a new entry. 
		Simply return the id of the existing album_asset document irrespective
		of whether the client sent a doc-id or not.
		If the client provides doc-id of an existing album-asset document in Oz, 
		then create a document with the same id. Otherwise, generate one. 
		Return the new document's id.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string) The unique ID of the album to modify.
	\param assetIds (list of strings) 
	\param albumAssetIdsForAsset (table of strings) A table in the following format
		{
			[ assetId3 ] = guid3, -- If Oz already has this album_asset document with guid3
			[ assetId4 ] = guid4, -- If Oz already has this album_asset document with guid4
		}
	\param obeyOrder (boolean)  If provided, the sort order will be set
		obeying the order of the assets in the input list.

	\return <code>Observable< ></code> An Observable object sends the ids of the album_asset
		documents in the following format.
		{
			[ assetId1 ] = guid1,
			[ assetId2 ] = guid2,
			[ assetId3 ] = guid3,
			[ assetId4 ] = false, -- if there was an error validating the given albumAssetId
		}
]]--

function WFCatAlbums.addAssetsToAlbumGivenDocIds( wfCatalog, albumId, assetIds, albumAssetIdsForAsset, obeyOrder )
	check( "WFCatalog", '+string', 'table', 'table', '?boolean' )
	
	local existingAssetUuidsObservable = assetUuidsForMaybeLocalAlbumId( wfCatalog, albumId )
	
	local existingAlbumAssetIdList = {}
	for _, guid in pairs( albumAssetIdsForAsset ) do
		existingAlbumAssetIdList[ #existingAlbumAssetIdList + 1 ] = guid
	end
	
	local existingAlbumAssetDocsObservable = #existingAlbumAssetIdList > 0
		and wfCatalog:docStore():bulkDocValue( existingAlbumAssetIdList )
		or Observable.just( {} )
	
	local assetOrderingObservable
	
	if obeyOrder then
		-- Docs will be inserted in order after the last doc with
		-- order. Docs without order will not be affected.
		
		assetOrderingObservable = getAssetOrderingObservable( wfCatalog, albumId, assetIds )
	else
		assetOrderingObservable = Observable.just( {} )
	end
	
	return Observable.combineLatest( existingAssetUuidsObservable, existingAlbumAssetDocsObservable, assetOrderingObservable )
	
		:flatMap( function( assetsInAlbum, existingAlbumAssetDocs, ordering )
		
			local existingAlbumAssetLocalIds = {}
			local assetIdsToAdd = {}
			local validationFailedAssets = {}
			local deletedAlbumAssets = {}
			
			local existingAlbumAssetDocMap = {}
			for _, docValue in ipairs( existingAlbumAssetDocs ) do
				if docValue._id then
					
					if docValue._deleted then
						-- Album asset doc has been deleted
						-- Treat this case same as the 
						-- non-existing album-asset case
						deletedAlbumAssets[ docValue._id ] = true
					
					elseif not docValue._empty then
						-- Album asset doc exists
						existingAlbumAssetDocMap[ docValue._id ] = docValue
					end
				end
			end
			
			-- Sort the assets that we want to add into three buckets:
			-- those that already are part of the album, those that are not,
			-- and a degenerate case where an existing album_asset with the
			-- same id points to an unrelated asset or album
			for _, assetId in ipairs( assetIds ) do
			
				local albumAssetId = albumAssetIdsForAsset[ assetId ]
				local isValid = true
				
				if albumAssetId then
				
					local albumAssetDoc = existingAlbumAssetDocMap[ albumAssetId ]
					
					if albumAssetDoc then
						if KSTableUtils.derefOpt( albumAssetDoc, "asset", "id" ) ~= assetId or 
							KSTableUtils.derefOpt( albumAssetDoc, "album", "id" ) ~= albumId then
							
							isValid = false
							validationFailedAssets[ #validationFailedAssets + 1 ] = assetId
						end
					end
				end
				
				if isValid then
					if assetsInAlbum[ assetId ] then
						existingAlbumAssetLocalIds[ #existingAlbumAssetLocalIds + 1 ] = assetsInAlbum[ assetId ]
					else
						assetIdsToAdd[ #assetIdsToAdd + 1] = assetId
					end
				end
			end
			
			-- Create album_asset documents for new assets
			local newDocs, newDocsRet = {}, {}
			for i, assetId in ipairs( assetIdsToAdd ) do
				
				-- Deleted album-assets should be treated similar to
				-- new album-assets. Create the album-asset with a 
				-- new doc id.
				
				local providedDocId = albumAssetIdsForAsset[ assetId ]
				local isDeleted = providedDocId and deletedAlbumAssets[ providedDocId ]
				
				local docId = not isDeleted and providedDocId
					or KSIdGenerator.uuidInLowercaseHex()

				local baseDocValue = WFCatUtils.createBaseOverlayWithUserCreatedFields( docId )

				local docValue = {
					type = 'album_asset',
					order = ordering[ assetId ],
					album = {
						id = albumId,
					},
					asset = {
						id = assetId,
					},
				}
				
				WFTableMerge.applyOverlay( docValue, baseDocValue )
				
				newDocsRet[ assetId ] = docId
				newDocs[ i ] = docValue

			end
			
			-- Create album_asset documents for existing assets
			-- to alter ordering
			
			local newAlbumAssetsDocsObservable = #newDocs > 0
				and WFRxBatchedBulkSetDocValue( "addAssetsToAlbumGivenDocIds", wfCatalog:docStore(), newDocs )
						:map( function()
							return newDocsRet
						end )
				or Observable.just( {} )
			
			-- Fetch the album_asset doc ids of those assets already in the album
			local existingAlbumAssetDocsObservable2
			
			if obeyOrder and #existingAlbumAssetLocalIds > 0 then
			
				-- Already existing album-assets in the input list 
				-- needs to be modified to respect the order
				existingAlbumAssetDocsObservable2 = wfCatalog:docStore():bulkDocValue( existingAlbumAssetLocalIds )
				 	:flatMap( function( docs )
						
						local existingDocs = {}
						for _, doc in ipairs( docs ) do
							
							local overlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( doc._id )
							
							overlay.type = 'album_asset'
							overlay.order = ordering[ doc.asset.id ]

							existingDocs[ #existingDocs + 1 ] = overlay
							
						end
						
						if #existingDocs > 0 then
						
							return WFRxBatchedBulkApplyOverlay( "addAssetsToAlbumGivenDocIds", wfCatalog:docStore(), existingDocs )
								:map( function()
									return docs
								end )
								
						else
							return Observable.just( {} )
						end
					end )
			else
				existingAlbumAssetDocsObservable2 = wfCatalog:docStore():bulkDocValue( existingAlbumAssetLocalIds )
			end
			
			
			local skippedFailedAlbumAssets = Observable.just( validationFailedAssets )
			
			-- Merge the existing docs and new docs stream
			return Observable.combineLatest( function( existingDocs, newDocRevs, failedAssets )
				
				local ret = {}
				
				for assetId, docId in pairs( newDocRevs ) do
					ret[ assetId ] = docId
				end
				
				for _, doc in ipairs( existingDocs ) do
					ret[ doc.asset.id ] = doc._id
				end
				
				for _, assetId in ipairs( failedAssets ) do
					ret[ assetId ] = false
				end
				
				return ret
				
			end, 
			existingAlbumAssetDocsObservable2, 
			newAlbumAssetsDocsObservable,
			skippedFailedAlbumAssets )
			
		end )
end

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

local function assetIdsInStacksAsMap( wfCatalog, stackIds )
	
	return wfCatalog:indexStore():queryOnce {
				sql = [[SELECT assetId FROM stack_assets JOIN CARRAY( ?, ?, 'char*' ) AS t ON t.value = stack_assets.stackId]],
				stackIds,
				#stackIds,
				quiesceUntilUpToDate = true,
				singleColumnQuery = true
			}
			:map( function( assetIds )
				local assetIdMap = {}
				for _, assetId in ipairs( assetIds ) do
					assetIdMap[ assetId ] = true
				end
				return assetIdMap
			end )

end

--------------------------------------------------------------------------------
--[[!
	\brief Removes assets from an album in this catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string) The unique ID of the album to modify.
	\param assetIds (array of strings or integers) An array of unique or local
		IDs for the assets to remove.  It is not an error to remove assets
		that are not in the album -- those asset ids are simply ignored.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--

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

local kNumOfSelectedAlbumAssetToQuery = 64

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

local kSQLPlaceholderString = 
		string.format( "%s?", string.rep( "?,", kNumOfSelectedAlbumAssetToQuery - 1 ) )

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

local function returnAlbumAssetTupleIfSelected( 
					albumAssetTuple, selectAssetIds )

	local assetId = albumAssetTuple[ 2 ]

	if assetId and selectAssetIds[ assetId ] then

		return albumAssetTuple

	end

end

--------------------------------------------------------------------------------
-- getSelectedAlbumAssetsStream has been optimized to fix bad real world 
-- performance when deleting thousands of album assets.
--
-- The original implementation was
--
--		local albumAssets = wfCatalog:indexStore():queryOnce {
--				-- get all the album_assets
--				sql = [[SELECT aa.docId, aa.assetId FROM album_asset_v2 AS aa JOIN TEMP_ids AS t ON t.docId == aa.assetId WHERE aa.albumId = ?]],
--				quiesceUntilUpToDate = true,
--				tempIds = assetIds,
--				albumId,
--			}
--
-- That took multiple seconds for 1000+ images
--
-- The new implementation uses an IN() query if kNumOfSelectedAlbumAssetToQuery assetIds
-- or less.
--
-- Otherwise it simply gets the entire album contents and filters the asset ids
-- to those selected.
--	
--
local function getSelectedAlbumAssetsStream( indexStore, albumId, assetIds )

	if #assetIds == 0 then

		return Observable.just( {} )

	end

	-- we use a single query string with kNumOfSelectedAlbumAssetToQuery placeholders so
	-- that the index store can cache the sql. If we don't have
	-- kNumOfSelectedAlbumAssetToQuery asset ids, we just fill the gap with -1
	if #assetIds <= kNumOfSelectedAlbumAssetToQuery then

		local vals = {}

		for i = 1, kNumOfSelectedAlbumAssetToQuery do

			vals[ i ] = i <= #assetIds and assetIds[ i ] or -1

		end
		
		return indexStore
				:queryOnce( {
						sql = string.format(
								[[SELECT docId, assetId FROM album_asset_v2 
								  WHERE albumId = ? AND assetId IN ( %s )]],
								kSQLPlaceholderString ),
						quiesceUntilUpToDate = true,
						albumId, unpack( vals, 1, kNumOfSelectedAlbumAssetToQuery ) 
					} )

	end

	-- otherwise fetch all of the album asset records and filter out any that
	-- aren't selected. This is much faster than the doing it in SQL via JOIN

	return indexStore
			:queryOnce( {
					sql = [[SELECT docId, assetId FROM album_asset_v2 WHERE albumId = ?]],
					quiesceUntilUpToDate = true,
					albumId 
				} )
			:map(
				function( albumAssetTuples )

					return WFFunctionUtils.filter( 
							returnAlbumAssetTupleIfSelected,
								albumAssetTuples,
								KSTableUtils.mapArrayToSet( assetIds ) )

				end )

end

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

function WFCatAlbums.removeAssetsFromAlbum( wfCatalog, maybeLocalAlbumId, maybeLocalAssetIds )

	check( "WFCatalog", '+string|number', 'table' )

	local albumUuid = WFCatIdUtils.albumUuidForMaybeLocalId( wfCatalog, maybeLocalAlbumId )
	local assetUuidsAndTypesStream = WFCatIdUtils.getOzAssetIdsAndTypeForIds( wfCatalog, maybeLocalAssetIds )

	return Observable.combineLatest( albumUuid, assetUuidsAndTypesStream )
		:map( function( albumId, assetUuidsAndTypes )
		
		local assetIds = {}
		local stacksForRemoval = {}
		-- process the assetIds - we need a list of all asset ids as well as a 
		-- map of which asset ids are for stacks which we'll use later
		for _, assetUuidAndType in pairs( assetUuidsAndTypes ) do
			local assetId = assetUuidAndType[ 1 ]
			local subtype = assetUuidAndType[ 2 ]
			if subtype == 'stack' then
				stacksForRemoval[ #stacksForRemoval + 1 ] = assetId
			end
			assetIds[ #assetIds + 1 ] = assetId
		end
		
		-- fetch all album_assets for the assetIds
		local albumAssets = 
				getSelectedAlbumAssetsStream( 
					wfCatalog:indexStore(), albumId, assetIds )
		
		-- look for the cover asset id as we'll need to know if we delete the cover asset
		local albumCoverAssetUuid = WFCatAlbums.albumCover( wfCatalog, albumId )
					:map( function( _, assetId )
						return assetId 
					end )
		
		local assetIdsForRemovedStacksStream = assetIdsInStacksAsMap( wfCatalog, stacksForRemoval )
		
		-- which stacks (and their stack_assets) are not being removed
		local assetsIdsForStacksRemaining = wfCatalog:indexStore():queryOnce {
				-- get stacks in album that are not being deleted
				sql = [[SELECT aa.assetId FROM album_asset_v2 as aa JOIN assets USING( assetId ) 
							WHERE aa.albumId = ? AND assets.subtype = 'stack' 
						EXCEPT 
							SELECT docId FROM TEMP_ids]],
				quiesceUntilUpToDate = true,
				tempIds = stacksForRemoval,
				singleColumnQuery = true,
				albumId
			}
			:flatMap( function( stackIds )
				return assetIdsInStacksAsMap( wfCatalog, stackIds )
			end )		
		
		return Observable.combineLatest( albumAssets, albumCoverAssetUuid, assetIdsForRemovedStacksStream, assetsIdsForStacksRemaining )
			:take( 1 )
			:map( function( albumAssetRows, albumCoverId, assetIdsForRemovedStacks, assetIdsForStacksRemaining )
				
				local deletedAlbumCover = false
				local overlays = {}
				local overlaysToKeepFromOz = {
						label = 'oz'
					}
				
				for _, row in ipairs( albumAssetRows ) do
					
					local albumAssetId = row[ 1 ]
					local assetId = row[ 2 ]
					
					if assetId  == albumCoverId then
						deletedAlbumCover = true
					end
					
					-- check to see if the asset is in a stack that's remaining. if it is, then it should actually stay.
					if not assetIdsForStacksRemaining[ assetId ] then 
						
						local overlay = {
								_id = albumAssetId,
								_deleted = true
							}
						
						-- asset is not in a stack that is sticking around, so if it's in a stack that's being removed,
						-- give it a label of 'oz' so that we don't send it
						if assetIdsForRemovedStacks[ assetId ] then
							overlaysToKeepFromOz[ #overlaysToKeepFromOz + 1 ] = overlay
						else
							overlays[ #overlays + 1 ] = overlay
						end
						
					end
				end
				
				return deletedAlbumCover, overlays, overlaysToKeepFromOz
			end )
			:flatMap( function( deletedAlbumCover, overlays, overlaysToKeepFromOz )
				
				return WFRxBatchedBulkApplyOverlay( "removeAssetsFromAlbum to to keep from oz", wfCatalog:docStore(), overlays )
					:flatMap( function()
						return WFRxBatchedBulkApplyOverlay( "removeAssetsFromAlbum to to keep from oz", wfCatalog:docStore(), overlaysToKeepFromOz )
					end )
					:map( function()
						return deletedAlbumCover, albumId
					end )
			end )
			
		end )
	:switch()
	:flatMapLatest( function( deletedAlbumCover, albumId )

		-- Not sure if it can happen but delete album cover at the end
		-- so the UI doesn't try defaulting it to a photo that will
		-- subsequently be deleted when deleting multiple assets.

		if deletedAlbumCover then
			return WFCatAlbums.setAlbumCover( wfCatalog, albumId, nil )
		end

		return Observable.just( true )

	end )
	:takeLast( 0 )
	:call( runUntilDone )


end

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

local function maybeGetFullDocId( docStore, maybeLocalId )

	if not maybeLocalId or type( maybeLocalId ) == 'string' then
		return Observable.just( maybeLocalId )
	end
	
	return docStore:localIdToDocId( maybeLocalId )
	
end

--------------------------------------------------------------------------------
--[[!
	\brief Sets the cover image for an album in this catalog.

	\param wfCatalog (WFCatalog) The catalog.
	\param albumId (string or number) The id of the album to modify.
	\param assetId (string or number or nil) The id of the asset to set as cover.
		If nil, any existing cover setting is removed from the album.

	\return <code>Observable< ></code> An Observable object completes when operation
		is completed; or resolves to an error if an error occurs.
]]--
function WFCatAlbums.setAlbumCover( wfCatalog, albumId, maybeAssetId )
	
	check( "WFCatalog", 'string|number', '?string|number' )
	
	local docStore = wfCatalog:docStore()
	
	local albumDocValueStream = docStore:docValue( albumId )
	
	local maybeFullAssetIdStream = maybeGetFullDocId( docStore, maybeAssetId )
	
	return Observable.combineLatest( albumDocValueStream, maybeFullAssetIdStream )
	
			:take( 1 )
			
			:flatMap( function( albumDocValue, maybeFullAssetId )
				
				if albumDocValue._spaceId then
					-- SPAM [jandrew 2019-08-26]: viewers and contributors should not 
					-- be able change the album sort order or the album cover.
					return Observable.just()
				end
				
				local overlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( albumId )
				
				if maybeFullAssetId then
					overlay.cover = { id = maybeFullAssetId }	
				else
					overlay.cover = { _delete = true }
				end
				
				return wfCatalog:docStore():applyOverlay( overlay )

			end )
			
			:takeLast( 0 )
	 
end

--------------------------------------------------------------------------------
--[[!
	\brief Returns the local asset id for the cover image for an album in this catalog.
	Rules governing the selection of the photo:
	If a cover image has been set, and we have the asset in the catalog use that asset
	Select the first photo in the album in chronological order.
	
	\param maybeLocalAlbumId (string|number) The unique ID of the album. This can either be the local 
	id or the global uuid of the album.
	
	\return <code>Observable< number, string ></code> An Observable object whose value is the docId, assetId
	of the current cover photo for this album.
	
	Observable value may be nil if there are no assets in the album or we do not yet have 
	local information for the asset id referred to as the cover.

 ]]--
function WFCatAlbums.albumCover( wfCatalog, maybeLocalAlbumId )

	WFArguments.check( 'WFCatalog', 'string|number' )
	
	local albumIdObservable

	if type( maybeLocalAlbumId ) == 'number' then
		albumIdObservable = wfCatalog:indexStore():queryOnce {
								sql = [[SELECT albumId FROM albums WHERE docId = ?]],
								singleColumnQuery = true,
								maybeLocalAlbumId
							}
							:map( function( rows )
								return derefOpt( rows, 1 )
							end )
	else
		albumIdObservable = Observable.just( maybeLocalAlbumId )
	end
	
	return albumIdObservable
		:map( function( albumId )	
			return wfCatalog:indexStore():query {
	
				sql = [[SELECT assets.docId, assets.assetId
							FROM assets
							INNER JOIN albums ON albums.coverId == assets.assetId
							WHERE albums.albumId = ?
					UNION ALL
						SELECT docId, assetId FROM 
						( SELECT assets.docId, assets.assetId
							FROM assets
							INNER JOIN album_asset_v2 ON album_asset_v2.assetId == assets.assetId
							WHERE album_asset_v2.albumId = ?
							ORDER BY
								case when album_asset_v2.sortOrder is null then 1 else 0 end,
								album_asset_v2.sortOrder, captureDate ASC )
					LIMIT 1]],
				albumId, albumId
			}
			:map( function( rows )
				-- pull out local cover id 
				return derefOpt( rows, 1, 1 ), derefOpt( rows, 1, 2 )
			end )
			
		end )
		:switch()
	
end

function WFCatAlbums.addDiagnosticLogSectionFunc( catalog )

	WFDiagnosticLog.addLogSectionFunc( function()

		return {

			name = "WFCatAlbums",
			id = 'WFCatAlbums',
			weight = 510,

			body = {
			
				albumCycles = basicAlbumsList( catalog, true )
				
					:map( function( albums )
						
						return findCycles( albums )
						
					end )
					
			},

		}

	end )

end

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

function WFCatAlbums.enableProjectAlbumPublishNotification( catalog, projectAlbumId, serviceId, editUrl )

	check( "WFCatalog", "+string|number", "+string", "+string" )

	return catalog:docStore():applyOverlay {
		_id = projectAlbumId,
		_localOnly = {
			publishNotifications = {
				serviceId = serviceId,
				editUrl = editUrl
			}
		}
	}
	
end

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

function WFCatAlbums.removeProjectAlbumPublishNotification( catalog, projectAlbumId )

	check( "WFCatalog", "+string|number" )

	return catalog:docStore():applyOverlay {
			_id = projectAlbumId,
			_localOnly = {
				publishNotifications = {
					_delete = true,
				}
			}
		}
end

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

function WFCatAlbums.ongoingProjectAlbumNotificationStatus( catalog, projectAlbumId )

	return Observable.just( { projectAlbumId } )
	
	:call( WFRxDocValueWatcher.make, catalog:docStore() )
	
	:pluck( "_localOnly", "publishNotifications" )

end

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

return WFCatAlbums
