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

WFCatSharedWithYouAlbums.lua

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

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

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

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

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

	-- WF.core.utils
local KSDate = require 'KSDate'
local KSJsonDecoder = require 'KSJsonDecoder'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFLogger = require 'WFLogger'
local WFCatDelete = require 'WFCatDelete'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxBatchedBulkSetDocValue = require 'WFRxBatchedBulkSetDocValue'
local WFRxThrottledBatchedBulkDocValue = require 'WFRxThrottledBatchedBulkDocValue'
local WFURI = require 'WFURI'

	-- WF.oz_client.managed_catalog]
local WFCatAlbums = require 'WFCatAlbums'
local WFCatCollaboration = require 'WFCatCollaboration'
local WFCatCopies = require 'WFCatCopies'
local WFCatSharedWithYouUtils = require 'WFCatSharedWithYouUtils'
local WFCatUtils = require 'WFCatUtils'

local WFRxRunUntilDone = require 'WFRxRunUntilDone'

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

local logger = WFLogger( 'WFCatSharedWithYouAlbums' )

local check = assert( WFArguments.check )

local spaceIdForAlbumId = {}
	-- key: Album ID (local ID or full doc ID)
	-- value: Associated space ID. Else false if there is no associated space ID.

local runUntilDone = assert( WFRxRunUntilDone )

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )
local KSTableUtils_setNested = assert( KSTableUtils.setNested )

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

local kContributor = 'contributor'
local kEditor = 'editor'

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

local function makeCatalogBasedObservableIntoSubModel( fn )
	
	return function( model, ... )
	
		local params = { ... }
		local nParams = select( '#', ... )

		local catalogStream = model._wfCatalog:take( 1 )
		
		local value = catalogStream
		
				:mapSwitch( function( wfCatalog )
					
					return fn( wfCatalog, unpack( params, 1, nParams ) )
					
				end )

		return {
		
			_wfCatalog = model._wfCatalog,
			value = value,
			
		}
		
	end
	
end


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

-- @mark WFCatSharedWithYouAlbums namespace

local WFCatSharedWithYouAlbums = {}

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

--[[!
	\brief Get the space Id for a given album.

	\param albumId (string|number) The local or full album doc ID.

	\return <code>Observable</code> Resolves to the space ID of the album. If
		the album is not in a space, resolves to false.
		
--]]

function WFCatSharedWithYouAlbums.getSpaceIdOfAlbumId( wfCatalog, albumId )

	check( "WFCatalog", "+string|number" )
	
	local cachedSpaceId = spaceIdForAlbumId[ albumId ]
		
	if cachedSpaceId ~= nil then
		return Observable.just( cachedSpaceId )
	end
	
	local albumUuidStream = Observable.just( albumId )
	
	if type( albumId ) == 'number' then
		albumUuidStream = wfCatalog:docStore():localIdToDocId( albumId )
	end
	
	return albumUuidStream
	
		:flatMap( function( fullAlbumId )
		
			local columnName = 'albumId'
			
			if WFPlatformInfo.isMobile() then
				columnName = 'docId'
			end
			
			return wfCatalog:indexStore():queryOnce {
					-- get all the album_assets
					sql = [[SELECT spaceId FROM albums WHERE ]] .. columnName .. [[ = ?]],
					quiesceUntilUpToDate = true,
					singleColumnQuery = true,
					fullAlbumId
				}
				:map( function( spaceIds )
				
					local spaceId = KSTableUtils_derefOpt( spaceIds, 1 ) or false
					
					spaceIdForAlbumId[ albumId ] = spaceId
					spaceIdForAlbumId[ fullAlbumId ] = spaceId
					
					return spaceId
					
				end )
				
		end )
	
end

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

--[[!
	\brief Get all of the sha256s of the originals within a specified space
		that were imported by the current (syncing) user.

	\param wfCatalog (WFCatalog)
	\param spaceId (string) The full space ID.
		
	\return <code>Observable</code> Resolves to a table that contains all of
		the sha256 of the originals in the space that were contributed by you.
--]]

function WFCatSharedWithYouAlbums.setOfSha256sInSpaceId( wfCatalog, spaceId )
	
	check( "WFCatalog", '+string' )
	
	local sql = [[ SELECT DISTINCT originalSHA256, sha256
						FROM coreInfo
						WHERE spaceId = ? AND importedByUser IS NULL ]]
	
	if WFPlatformInfo.isDesktop then
	
		sql = [[ SELECT DISTINCT originalSHA256, sha256
					FROM assets
					INNER JOIN importSource USING( assetId )
					WHERE assets.spaceId = ? AND importedByUser IS NULL ]]
					
	end

	return wfCatalog:indexStore():queryOnce {
			sql = sql,
			quiesceUntilUpToDate = true,
			spaceId
		}

		:map( function( rows )
			
			logger:debugf( 'setOfSha256sInSpaceId for spaceId %s complete w/ %d rows', spaceId, #rows )
		
			local sha256s = {}
			
			for _, row in ipairs( rows ) do
			
				local sha256 = row[ 1 ] or row[ 2 ]
	
				if sha256 then
					sha256s[ sha256 ] = true
				end
				
			end
			
			return sha256s
		
		end )
		
end

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

local function assetUuidsForAlbumUuid( wfCatalog, albumUuid )
	
	check( "WFCatalog", '+string' )
	
	local albumAssetTableName = 'album_asset'
	
	if WFPlatformInfo.isDesktop then
		albumAssetTableName = 'album_asset_v2'
	end
	
	return wfCatalog:indexStore():queryOnce {
			sql = string.format( 
					[[SELECT assetId 
						FROM %s 
						WHERE albumId = ?]], albumAssetTableName ),
			albumUuid,
			quiesceUntilUpToDate = true,
		}

		:map( function( rows )
			
			local assetIds = {}
			
			for _, albumAsset in ipairs( rows ) do
			
				local assetId = albumAsset[ 1 ]
				assetIds[ assetId ] = assetId
	
			end
			
			return assetIds
		
		end )
		
end

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

--[[!
	\brief Add Asset(s) to a shared-with-you album

	\param wfCatalog (WFCatalog)
	\param spaceId (string) The full space ID.
	\param albumId (string|number) The local or full album doc ID.
	\param assetIds (table) A table containing local or full asset IDs
		to add to the specified album.
	\param redactionOptions (table) A table of redaction options:
		redactLocation (boolean)
		redactKeywords (boolean)
		redactRatingsReviews (boolean)
		
	\return <code>Observable</code> Resolves to three tables. The first table contains 
		the NEW copied asset IDs. The second table contains a list of asset IDs that
		were not added to the album because they were duplicates. The third table contains
		a list of asset IDs which have been declared "invalid" by Oz.

--]]

function WFCatSharedWithYouAlbums.addAssetsToAlbum( wfCatalog, spaceId, albumId, assetIds, redactionOptions, copyRenditionsFn )

	check( "WFCatalog", "+string", "+string|number", "table", "?table", '?function' )
	
	logger:infof( "Adding %d assets to album %s in space %s", #assetIds, albumId, spaceId )
		
	if #assetIds == 0 then
		return Observable.just( {}, {} )
	end
	
	-- SPAM [jandrew 2019-07-22]: This function may seem complex but there are a number
	-- of cases that it has to cover:
	-- Case 1.) The asset document is invalid in Oz
	--
	-- Case 2.) The asset document does not exist in the space or in the album. In this case, the
	-- asset is copied into the space, and the copied asset is added to the album.
	--
	-- Case 3.) The asset document already exists within the space, but is not in the album. For
	-- this case, the existing asset is added to the album.
	--
	-- Case 4.) The asset document already exists within the space and the album. This is a no-op.
	--
	-- Case 5.) The image (not the asset document) already exists in the space. If we
	-- detect a SHA256 duplication, we won't copy the asset into the space or add the
	-- existing image in the space into the album. The develop settings for the image
	-- in the space may be different, so we intentionally treat this as a no-op.
	
	local albumUuidStream = Observable.just( albumId )
	
	if type( albumId ) == 'number' then
		albumUuidStream = wfCatalog:docStore():localIdToDocId( albumId )
	end
		
	-- Check to see if each asset already exists in the space. If it does, then we don't 
	-- need to copy it into the space. Also check to see if the asset already exists 
	-- within the album.
	
	-- This also performs basic dupe detection by comparing the SHA256 of the assets to
	-- copy against the existing assets in the space.
	
	local assetIdStream = albumUuidStream
	
		:flatMap( function( albumUuid )
		
			local assetDocValueStream = 
					WFRxThrottledBatchedBulkDocValue( "WFCatSharedWithYouAlbums_addAssetsToAlbum", wfCatalog:docStore(), assetIds, 100, 10 )
					
						:scan( function( docValues, batchOfDocValues )
				
							docValues = KSTableUtils.appendArray( docValues, batchOfDocValues )
							return docValues
							
						end, {} )

						:takeLast( 1 )
						
			local assetUuidsForAlbumStream = assetUuidsForAlbumUuid( wfCatalog, albumUuid )
			local existingAssetSha256s = WFCatSharedWithYouAlbums.setOfSha256sInSpaceId( wfCatalog, spaceId )
			
			return Observable.combineLatest( 
					Observable.just( albumUuid ),
					assetDocValueStream, 
					assetUuidsForAlbumStream,
					existingAssetSha256s )
			
		end )
		
		:take( 1 )
		
		:flatMap( function( albumUuid, batchOfDocValues, assetUuidsForAlbum, existingSha256s )
		
			local assetIdsToCopyAndAddToAlbum = {}
			local assetIdsToAddToAlbum = {}
			local duplicateAssetIds = {}
			local invalidAssetIds = {}
			
			return Observable.from( batchOfDocValues )
		
				:map( function( docValue )

					local isNonTrashAssetDocument = function ( doc )
						local subtypeOfAsset = doc and doc.type == 'asset' and doc.subtype
						return subtypeOfAsset == 'video' or subtypeOfAsset == 'image'
					end
					
					if isNonTrashAssetDocument( docValue ) then

						local assetSpaceId = docValue._spaceId
						local assetUuid = docValue._id
						local sha256BadMatch = KSTableUtils_derefOpt( docValue,  'importSource', 'status', 'error', 'all', 'reason', 'sha256' )  == 'does not match'
						local invalidOriginalOnOz = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/master', 'invalid' ) == true
						local invalidLocalOriginal = KSTableUtils_derefOpt( docValue, '_localOnly', 'localOriginalError' ) == 'InvalidLocalOriginal'
						
						local invalidDocument = sha256BadMatch or invalidOriginalOnOz or invalidLocalOriginal
						
						if invalidDocument then
						
							-- Case 1: The asset has been marked as "invalid" by Oz an should not be added to the album.

							invalidAssetIds[ #invalidAssetIds + 1 ] = assetUuid
						
						elseif assetSpaceId == spaceId then
							
							-- The asset document already exists within the specified space.
					
							if assetUuidsForAlbum[ assetUuid ] then
							
								-- Case 4: No-op! The asset document already exists within the album 
								-- and the space.
								
								duplicateAssetIds[ #duplicateAssetIds + 1 ]  = assetUuid
							
							else
							
								-- Case 3: The asset document needs to be added to the album.
								
								assetIdsToAddToAlbum[ #assetIdsToAddToAlbum + 1 ] = assetUuid
								
							end
						
						else

							local sha256 = KSTableUtils_derefOpt( docValue, 'importSource', 'originalSHA256' )
											or KSTableUtils_derefOpt( docValue, 'importSource', 'sha256' )
						
							-- NOTE: Each album member is allowed to contribute any image once, 
							-- even if has been contributed by another member.
						
							if not existingSha256s[ sha256 ] then
							
								-- Case 2: The image needs to be copied to the space and added to the album.
							
								assetIdsToCopyAndAddToAlbum[ #assetIdsToCopyAndAddToAlbum + 1 ] = assetUuid
							
							else
							
								-- Case 5 : No-op! The image already exists within the space.
							
								duplicateAssetIds[ #duplicateAssetIds + 1 ]  = assetUuid
							
							end
						end
					end

					return albumUuid, assetIdsToCopyAndAddToAlbum, assetIdsToAddToAlbum, duplicateAssetIds, invalidAssetIds

				end )

		end )
		
		:takeLast( 1 )
		
	-- Clone the asset documents. When we add an asset into a space, we
	-- create new asset documents within the doc store. Apply redactions
	-- as specified by the user.
		
	local spaceDocStream = wfCatalog:docStore():docValue( spaceId )
		
	-- In order to know whether or not we should copy across binaries for the asset,
	-- we need to know the user's role. Editors have full access to
	-- the original/proxies while contributors cannot access either.
	
	local currentUserRoleStream = wfCatalog:indexStore():queryOnce( 
		{ 
			sql = [[SELECT role
						FROM space
						INNER JOIN space_user ON space_user.fullDocId == space.spaceUserId
						where space.docId == ?]],
			quiesceUntilUpToDate = true,
			spaceId
		} )
		
		:map( function( rows )
		
			local role = KSTableUtils_derefOpt( rows, 1, 1 ) or kContributor
			
			role = WFCatCollaboration.downgradeRoleIfNotSupported( role )
			
			return role
			
		end )
		
	return Observable.combineLatest( spaceDocStream, currentUserRoleStream, assetIdStream )
	
		:flatMap( function( spaceDoc, currentUserRole, albumUuid, assetIdsToCopyAndAddToAlbum, assetIdsToAddToAlbum, duplicateAssetIds, invalidAssetIds )
	
			if #assetIdsToCopyAndAddToAlbum == 0 or 
				( currentUserRole ~= kContributor and currentUserRole ~= kEditor ) then
				return Observable.just( albumUuid, assetIdsToAddToAlbum, duplicateAssetIds, invalidAssetIds )
			end
			
			local userCreatedUpdatedTimestamp = KSDate.timeInIso8601Format()
			
			local copiedDocModifierFn = function( copiedDocValue )
			
				local currentUserId = wfCatalog:ozUserId()
			
				copiedDocValue._spaceId = spaceId
				copiedDocValue.userCreated = userCreatedUpdatedTimestamp
				copiedDocValue.userUpdated = userCreatedUpdatedTimestamp
				
				KSTableUtils_setNested( copiedDocValue, userCreatedUpdatedTimestamp, 'importSource', 'importTimestamp' )
				
				KSTableUtils_setNested( copiedDocValue, currentUserId, 'importSource', 'importedBy' )
					-- Make sure the importedBy field is set to the contributors ID. LrW uses this field.
					
				local function removeSectionIfEmpty( docValue, ... )
					
					-- If redacting information results in an empty table being created in the document,
					-- we must remove the empty table as well. This function isn't recursive,
					-- but we do manually call this function on the parent tables as necessary. 
	
					local maybeTable = KSTableUtils_derefOpt( docValue, ... )
	
					if maybeTable and not next( maybeTable ) then
						KSTableUtils_setNested( copiedDocValue, nil, ... )
					end
	
				end
			
				if copiedDocValue.type == 'asset' then
					
					-- Always redact certain fields:
					
					KSTableUtils_setNested( copiedDocValue, nil, 'copy' )
					
					KSTableUtils_setNested( copiedDocValue, nil, 'importSource', 'lrcatStoreProviderId' )
					KSTableUtils_setNested( copiedDocValue, nil, 'importSource', 'lrcatImageId' )
					KSTableUtils_setNested( copiedDocValue, nil, 'importSource', 'status' )
					
					removeSectionIfEmpty( copiedDocValue, 'importSource' )
					
					KSTableUtils_setNested( copiedDocValue, nil, 'mergedAssets' )
					
					if redactionOptions.redactLocation ~= false then
						
						-- Track that location info was redacted
						KSTableUtils_setNested( copiedDocValue, true, '_localOnly', 'universal', 'sharedwithyou', 'contribution', 'redactLocation' )
						
						-- Location metadata is redacted by default.
						
						-- All location data, user entered, camera lat/long, and reverse 
						-- geocoded should be removed.
						
						KSTableUtils_setNested( copiedDocValue, nil, 'location' )
						
						KSTableUtils_setNested( copiedDocValue, nil, 'xmp', 'iptcCore', 'CountryCode' )
						KSTableUtils_setNested( copiedDocValue, nil, 'xmp', 'iptcCore', 'Location' )

						removeSectionIfEmpty( copiedDocValue, 'xmp', 'iptcCore' )

						KSTableUtils_setNested( copiedDocValue, nil, 'xmp', 'photoshop', 'City' )
						KSTableUtils_setNested( copiedDocValue, nil, 'xmp', 'photoshop', 'Country' )
						KSTableUtils_setNested( copiedDocValue, nil, 'xmp', 'photoshop', 'State' )
						
						removeSectionIfEmpty( copiedDocValue, 'xmp', 'photoshop' )
						
					end
					
					if redactionOptions.redactKeywords then
						
						-- Track that keyword info was redacted
						KSTableUtils_setNested( copiedDocValue, true, '_localOnly', 'universal', 'sharedwithyou', 'contribution', 'redactKeywords' )
						
						-- redact keywords
						KSTableUtils_setNested( copiedDocValue, nil, 'xmp', 'dc', 'subject' )
						
						removeSectionIfEmpty( copiedDocValue, 'xmp', 'dc' )
						
					end
					
					if redactionOptions.redactRatingsReviews then
						
						-- Track that rating/review info was redacted
						KSTableUtils_setNested( copiedDocValue, true, '_localOnly', 'universal', 'sharedwithyou', 'contribution', 'redactRatingsReviews' )
						
						-- redact ratings/reviews
						KSTableUtils_setNested( copiedDocValue, nil, 'ratings' )
						KSTableUtils_setNested( copiedDocValue, nil, 'reviews' )
						
					else
					
						-- If the ratings/reviews have not been redacted, then rewrite the fields
						-- so that it appears with the correct owner ID.
						
						local spaceOwnerId = KSTableUtils_derefOpt( spaceDoc, '_serverProvidedMetadata', 'owner', 'id' )
						
						-- Ratings
						
						local currentRatings = KSTableUtils_derefOpt( copiedDocValue, 'ratings' ) or {}
					
						local currentRating = currentRatings[ currentUserId ]
					
						if currentRating and spaceOwnerId then
						
							local newRatings = {}
						
							newRatings[ spaceOwnerId ] = currentRating
							
							KSTableUtils_setNested( copiedDocValue, newRatings, 'ratings' )
							
						else
						
							KSTableUtils_setNested( copiedDocValue, nil, 'ratings' )
							
						end
						
						-- Reviews
						
						local currentReviews = KSTableUtils_derefOpt( copiedDocValue, 'reviews' ) or {}
					
						local currentReview = currentReviews[ currentUserId ]
					
						if currentReview and spaceOwnerId then
						
							local newReviews = {}
						
							newReviews[ spaceOwnerId ] = currentReview
							
							KSTableUtils_setNested( copiedDocValue, newReviews, 'reviews' )
							
						else
						
							KSTableUtils_setNested( copiedDocValue, nil, 'reviews' )
							
						end
					
					end
				
					removeSectionIfEmpty( copiedDocValue, 'xmp' )
						-- The above redaction options may have caused us to remove all of the xmp
						-- section's sub-sections.
						
				end

				return copiedDocValue
			
			end
		
			local optArgs = {
				isContributorCopyingToSharedAlbum = currentUserRole == kContributor,
				isXmpDevelopCopyingToNewPath = WFPlatformInfo.isAndroid and true or nil,
			}
		
			return WFCatCopies.createCopiesOfAssets( wfCatalog, assetIdsToCopyAndAddToAlbum, 
					nil, nil, copiedDocModifierFn, false, copyRenditionsFn, optArgs )
	
				:map( function( newAssetIds )
				
					assetIdsToAddToAlbum = KSTableUtils.appendArray( assetIdsToAddToAlbum, newAssetIds )
				
					return albumUuid, assetIdsToAddToAlbum, duplicateAssetIds, invalidAssetIds
				
				end )
	
		end )
		
		:flatMap( function( albumUuid, assetIdsToAddToAlbum, duplicateAssetIds, invalidAssetIds )
		
			-- Create the new album asset documents.
		
			local assetObservables =  Observable.just( assetIdsToAddToAlbum, duplicateAssetIds, invalidAssetIds )
			
			if #assetIdsToAddToAlbum == 0 then
			
        		return assetObservables

			else

				local newAlbumAssets = {}

				for _, assetUuid in ipairs( assetIdsToAddToAlbum ) do

					local albumAsset = WFCatAlbums.createAlbumAssetDocument( albumUuid, assetUuid, spaceId )
					newAlbumAssets[ #newAlbumAssets + 1 ] = albumAsset

				end

				return WFRxBatchedBulkSetDocValue( "WFCatSharedWithYouAlbums.addAssetsToAlbum",
						wfCatalog:docStore(), newAlbumAssets )

					:concat( assetObservables )
			end
		end )

		:takeLast( 1 )

		:tapOnCompleted( function()
			logger:debugf( 'WFCatSharedWithYouAlbums.addAssetsToAlbum COMPLETE' )
		end )
		
end

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

--[[!
	\brief Remove Asset(s) from a shared-with-you album. The user must
		be online to call this function.

	\param wfCatalog (WFCatalog)
	\param spaceId (string) The full space ID.
	\param assetIds (table) A table containing local or full asset IDs
		to remove from the specified space.
		
	\return <code>Observable</code> Resolves to a table containing a list
		of asset IDs that were not able to be deleted.
--]]

function WFCatSharedWithYouAlbums.deleteAssets( wfCatalog, spaceId, assetIds )

	check( "WFCatalog", "+string", "table" )
	
	logger:infof( "Removing %d assets from space %s", #assetIds, spaceId )
	
	if #assetIds == 0 then
		return Observable.just( {} )
	end
	
	local catSpec = wfCatalog:ozCatalogSpec()
	local docStore = wfCatalog:docStore()
	local indexStore = wfCatalog:indexStore()
	
	return indexStore:queryOnce( 
		{ 
			sql = [[SELECT role
						FROM space
						INNER JOIN space_user ON space_user.fullDocId == space.spaceUserId
						where space.docId == ?]],
			quiesceUntilUpToDate = true,
			spaceId
		} )
		
		:map( function( rows )
		
			local role = KSTableUtils_derefOpt( rows, 1, 1 ) or kContributor
			
			role = WFCatCollaboration.downgradeRoleIfNotSupported( role )
			
			return role
			
		end )
		
		:flatMap( function( role )
		
			if role ~= kContributor and role ~= kEditor then
				return Observable.just( assetIds )
			end
			
			return WFRxThrottledBatchedBulkDocValue( "WFCatSharedWithYouAlbums_deleteAssets", docStore, assetIds, 100, 10 )
				:map( function( batchOfDocValues )
					local assetsInDeletionWindow = {}
					local assetIdsToCatalogDocIds = {}
					local undeletedAssets = {}
			
					-- Check to make sure that:
						-- assets are a part of the space.
						-- user has the permission to delete the asset
					
					for _, docValue in ipairs( batchOfDocValues ) do
			
						local assetSpaceId = docValue._spaceId
			
						if assetSpaceId == spaceId 
							and WFCatSharedWithYouUtils.canDeleteAsset( docValue ) then
							
							assetsInDeletionWindow[ #assetsInDeletionWindow + 1 ] = docValue._id
							-- Since Oz guarantees unique ids sent to clients in the space feeds, 
							-- what we know the doc as and what Oz knows the doc as may be slightly 
							-- different. When we call the REST API to delete we need to use Oz's id
							-- so stash it here. If it's not on Oz yet though, just use the assetId
							local catalogDocId =  KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata' ) and
													KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'catalog_doc', 'id' ) or
													docValue._id
							assetIdsToCatalogDocIds[ docValue._id ] = catalogDocId
							
						else
						
							undeletedAssets[ #undeletedAssets + 1 ] = docValue._id
							
						end
			
					end
			
					return assetsInDeletionWindow, assetIdsToCatalogDocIds, undeletedAssets
			
				end )
				:map( function( assetsInDeletionWindow, assetIdsToCatalogDocIds, undeletedAssets )
				
					local assetsToDeleteLocally = {}
				
					if #assetsInDeletionWindow == 0 then
						return Observable.just( assetsToDeleteLocally, undeletedAssets )
					end
					
					-- Attempt to delete all the assets from Oz. We even attempt to delete the
					-- ones we don't believe are on Oz; this is to address a race condition
					-- where our local state does not match the Oz state: WF-693
					
					return Observable.from( assetsInDeletionWindow )
				
						:map( function( assetId )
				
							-- Issue a REST call to Oz to delete the assets - one per asset.
							-- We send the deletion request without changing any local state.
						
							local options = {
								requestHeaders = {
									{ header = 'Content-Type', value = 'application/json' },
									{ header = 'Accept', value = 'application/json' },
								},
								requestMethod = 'DELETE',
								bypassRateLimiter = true,
								returnHttpErrorsAsData = true
							}
							
							local catalogDocId = assetIdsToCatalogDocIds[ assetId ]
							
							if not catalogDocId then
								-- we don't have a catalog doc id so we can't call the REST API and so
								-- we must not delete this doc locally since we can't delete it from Oz
								undeletedAssets[ #undeletedAssets + 1 ] = assetId
								return Observable.just( assetsToDeleteLocally, undeletedAssets )
							end
							
							local deleteAssetUrl = WFURI.stringByAppendingPathComponents( catSpec:serverBaseUrl(),
										string.format( "/v2/spaces/%s/assets/%s", spaceId, catalogDocId ) )
								
							return catSpec:get( deleteAssetUrl, options )
								
								:map( function( data, headers )
								
									if headers and headers.status_code == '204' then
		
										-- Once the REST call has resolved, we know the deletion was successful.
		
										assetsToDeleteLocally[ #assetsToDeleteLocally + 1 ] = assetId
		
									elseif headers and headers.status_code == '404' and headers[ 'X-Result-Code' ] == '1000' then
		
										local ok, response = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty, data )
		
										local resourceDoesNotExist = 
												ok and KSTableUtils_derefOpt( response, 'subtype' ) == "ResourceNotFoundError"
								
										if resourceDoesNotExist then
										
											-- The asset didn't exist on Oz so we can safely delete it locally.
											assetsToDeleteLocally[ #assetsToDeleteLocally + 1 ] = assetId
											
										else
											undeletedAssets[ #undeletedAssets + 1 ] = assetId
										end
		
									else
		
										undeletedAssets[ #undeletedAssets + 1 ] = assetId
									
									end
									
									return assetsToDeleteLocally, undeletedAssets
									
								end )
								
						end )
						
						:flatMerge( 2 )
							-- Limit the REST calls to happen 2 at a time.
					
						:takeLast( 1 )
							-- Wait for all of the REST calls to complete.
				
				end )
				:flatMerge( 1 )
				:flatMap( function( assetsToDeleteLocally, undeletedAssets )
				
					-- Now that all the documents are deleted on Oz, mark the local documents
					-- as deleted. We could wait on the sync feed for the deletions, but
					-- deleting them right away allows for the UI to update more quickly.
					
					local dontSyncToOz = true
						-- Since Oz already has the deletion, we don't need to sync this.
					
					return WFCatDelete.deleteDocs( docStore, assetsToDeleteLocally, dontSyncToOz )
					
						:map( function() 
						
							return undeletedAssets 
							
						end )
		
				end )
				:scan( function( undeletedAssets, batchOfUndeletedAssetsBatch )
					undeletedAssets = KSTableUtils.appendArray( undeletedAssets, batchOfUndeletedAssetsBatch )
					return undeletedAssets
				end, {} )
				:takeLast( 1 )
				
		end )
			
end

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

--[[!
	\brief Remove yourself from a shared space.

	\param wfCatalog (WFCatalog)
	\param spaceId (string) The full space ID.
		
	\return <code>Observable</code> Resolves to true.
--]]

function WFCatSharedWithYouAlbums.leaveSpace( wfCatalog, spaceId )

	check( "WFCatalog", "+string" )
			
	logger:infof( "Leaving space %s", spaceId )
	
	local docStore = wfCatalog:docStore()
	
	local kAbandoned = 'abandoned'
		-- http://apidocs.adbecrsl.com/v2/data-model/space_user/index.html
	
	-- Fetch the space_user ID associated with the user.
	
	return wfCatalog:indexStore():queryOnce {
			sql = [[SELECT spaceUserId, albumId
						FROM space
						LEFT JOIN space_album ON space.docId == space_album.spaceId
						WHERE space.docId = ?]],
			spaceId,
			quiesceUntilUpToDate = true,
		}
	
		:map( function( rows )
		
			local overlays = {}
			
			for _, row in ipairs( rows ) do
			
				-- Change the space_user status to 'abandoned'.
			
				-- This will sync to Oz, at which point Oz will generate a tombstone for the
				-- space on the feed-of-feeds.
				
				-- This also has the effect of stopping all sync-down operations relating to the
				-- space.
			
				local spaceUserId = row[ 1 ]
				
				if spaceUserId then
				
					local abandonSpaceOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( spaceUserId )
					abandonSpaceOverlay.status = kAbandoned
					
					overlays[ #overlays + 1 ] = abandonSpaceOverlay
					
				end
				
				-- Delete the album. This is actually not necessary, since changing the space_user status 
				-- to 'abandoned' will cause a chain reaction, deleting all related documents.
				-- However, deleting the album document here has the immediate effect of removing the album
				-- from the albums list, which makes the operation appear very fast.
				
				local albumId = row[ 2 ]
				
				if albumId then
				
					local deleteAlbumOverlay = WFCatUtils.createBaseOverlayWithUserUpdatedField( albumId )
					deleteAlbumOverlay._deleted = true
					deleteAlbumOverlay.label = '_dontsendtooz'
						-- There's no need to send this deletion to Oz since we're leaving the space. The album
						-- isnt' actually being deleted, it's just being removed from the current user.
						
					overlays[ #overlays + 1 ] = deleteAlbumOverlay
			
				end
				
			end
		
			return overlays
		
		end )

		:flatMap( function( overlays )
			
			if next( overlays ) then
					
				-- There's no need to batch or throttle the following bulkApplyOverlay. The number of
				-- overlays should be small. In most cases, it should be two.
					
				return docStore:bulkApplyOverlay( overlays, "WFCatSharedWithYouAlbums_leaveSpace" )
				
					:map( function()
						return true
					end )
					
			else
			
				-- This should be impossible. Overlays should always contain an entry for the spaceUser
				-- doc associated with the space. If the spaceUser doc doesn't exist, something elsewhere
				-- has gone horribly wrong.
				
				return Observable.just( true )
				
					
			end
			
		end )
		
		:call( runUntilDone )

end

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

--[[!
	\brief Fetch a URL to report abuse within a shared-with-you album.
	
	\param wfCatalog (WFCatalog)
	\param docId (string|number) The local or full doc ID of an abusive
		album or asset.
		
	\return <code>Observable</code> Resolves with a URL. All abuse reporting
		should simply link out to LrW using this URL.
--]]

function WFCatSharedWithYouAlbums.reportAbuse( wfCatalog, docId )

	check( "WFCatalog", "+string|number" )
				
	return wfCatalog:docStore():docValue( docId )
	
		:map( function( docValue )
		
			local docType = docValue.type
			
			if docType == 'asset' then
				logger:infof( "Fetching URL to report abuse in asset %s", docId )
			elseif docType == 'album' then
				logger:infof( "Fetching URL to report abuse in album %s", docId )
			else
				assert( false, "Only assets and albums can be reported for abuse." )
			end
			
			local reportAbuseUrl = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/web_abuse_report', 'href' )
		
			return reportAbuseUrl
			
		end )
	
end

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

--[[!

	\brief A model made up of submodels used to get and interact with 
		albums that have been shared with you.
	
	\param wfCatalogStream (Observable) a stream of WFCatalog objects.
	
	\return WF Model
	
--]]

function WFCatSharedWithYouAlbums.makeModel( wfCatalogStream )

	return {
	
		_wfCatalog = wfCatalogStream,
		
		priority = 'high',
		
		value = {},

		models = {

			leaveSpace = makeCatalogBasedObservableIntoSubModel( function( wfCatalog, spaceId )
				
				return WFCatSharedWithYouAlbums.leaveSpace( wfCatalog, spaceId )

			end ),
	
			reportAbuse = makeCatalogBasedObservableIntoSubModel( function( wfCatalog, spaceId, albumId )
			
				return WFCatSharedWithYouAlbums.reportAbuse( wfCatalog, spaceId, albumId )
				
			end ),
			
		},

	}
	
end

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

--[[!

	\brief A model made up of submodels used to get and interact with 
		albums that have been shared with you.
	
	\param wfCatalog (WFCatalog)
	
	\return WF Model
	
--]]

function WFCatSharedWithYouAlbums.makeModelWithCatalog( wfCatalog )

	local wfCatalogStream = Observable.just( wfCatalog )

	return WFCatSharedWithYouAlbums.makeModel( wfCatalogStream ) 
	
end

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

return WFCatSharedWithYouAlbums
