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

WFCatStorage.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2016 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain 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 may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law.  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
local BehaviorSubject = Rx.BehaviorSubject

	-- WF.core.utils
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFLogger = require 'WFLogger'
local WFConfiguration = require 'WFConfiguration'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFRxScheduler = require 'WFRxScheduler'
local WFRxThrottledBatchedBulkDocValue = require "WFRxThrottledBatchedBulkDocValue"

	-- WF.oz_client.managed_catalog
local WFCatOzErrors = require 'WFCatOzErrors'
	
--------------------------------------------------------------------------------

local DEBUG_STORAGE_USED = WFConfiguration.Debug.debugStorageUsage
local debugStorageUsedStream

local check = assert( WFArguments.check )

local logger = WFLogger( 'WFCatStorage' )

local kLowOnStoragePercent = .9
local kOutOfStoragePercent = .99

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

local function quota_exceeded_table()
	
	return {
	
		sql = [[CREATE TABLE quota_exceeded( docType, spaceId )]],
		
		useLocalId = true,
		
		hasOneRowPerDocument = true,
		
		relevantTypeSubtypeTuples = {
			{ '*', '*' }
		},
		
		map = function( docValue, emit )
			
			if WFCatOzErrors.hasStorageExceededError( docValue ) and not docValue._deleted then
				
				local spaceId = docValue._spaceId
			
				emit( docValue.type, spaceId )
				
			end
			
		end,
	}

end

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

local function generateIndexSchema( _, indexStoreSchema )

	check( '_', 'WFRxIndexStoreSchema' )
	
	indexStoreSchema:addTable( quota_exceeded_table() )

	indexStoreSchema:addIndex
			[[CREATE INDEX quota_exceededSpaceId on quota_exceeded( spaceId )]]

end

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

local function isAccountDocOverStorage( accountDoc )
	
	local maybeOzStorage = KSTableUtils.derefOpt( accountDoc, 'entitlement', 'storage' )

	return maybeOzStorage and ( maybeOzStorage.used > maybeOzStorage.limit ),
															-- quota exceeded
							maybeOzStorage.limit - maybeOzStorage.used
															-- available space

end

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

local function countOverOzStorageQuota( wfCatalog, optCalm )

	return wfCatalog:indexStore():query( 
		{
			sql = [[SELECT COUNT( docId )
						FROM quota_exceeded
						WHERE spaceId IS NULL]],
			calm = optCalm or 60
		} )
		
		:map( function( results ) 
			
			return KSTableUtils.derefOpt( results, 1, 1 ) or 0
			
		end )
	
end

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

local function docsOverOzStorageQuota( catalog, optCalm )
	
	return catalog:indexStore():query( 
		{ 
			sql = [[SELECT docId, docType
						FROM quota_exceeded
						WHERE spaceId IS NULL
						ORDER by docId ASC]],
			calm = optCalm 
		} )
		
		:map( function( docIdRows )
			
			-- [jandrew 2022-01-04]: WF-1644:
			-- WFCatStorage is used by all platforms. Every 60 seconds, the WFCatStorage
			-- code will check to see how much free space is available on Oz. If there's
			-- any free space available, it will go through and attempt to remove StorageExceededError
			-- errors for a portion of the documents. Before, it was possible that the code would  
			-- only attempt to remove the errors for documents with asset references (versions, 
			-- comments, favorites, etc) but not the actual asset documents themselves. In 
			-- theory, this could lead to the retry looper seen in WF-1644. To avoid this, 
			-- we now sort the documents so that Asset document IDs are at the front of the
			-- returned array. Non-Asset Documents w/ "StorageExceededError" errors
			-- often have dependencies on unsynced Asset documents w/ "StorageExceededError"
			-- errors, so frontloading the returned array with Asset Ids ensures
			-- that we will first attempt to remove the "StorageExceededError" errors
			-- from Assets before other document types. In theory, this will help
			-- prevent us from encountering "ResourceNotFoundError" issues, since we
			-- won't clear the "StorageExceededError" error on dependent documents
			-- until we've already cleared it on their referenced Asset documents.
			
			local assetIds = {}
			local otherDocIds = {}
			
			for _, row in ipairs( docIdRows ) do
			
				local docId = row[ 1 ]
				local docType = row[ 2 ]
			
				if docType == 'asset' then
					table.insert( assetIds, docId )
				else
					table.insert( otherDocIds, docId )
				end
				
			end
			
			local docIds = {}
			
			docIds = KSTableUtils.appendArray( docIds, assetIds )
			docIds = KSTableUtils.appendArray( docIds, otherDocIds )
			
			return docIds
			
		end )

end

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

local function spaceNeededByThisBatch( batchOfDocValues )

	local spaceRequired = 0
	local numMissingFileSizes = 0
	
	for _, docVal in ipairs( batchOfDocValues ) do

		local fileSize = KSTableUtils.derefOpt( docVal, 'importSource', 'fileSize' ) or 0

		if fileSize == 0 then
			numMissingFileSizes = numMissingFileSizes + 1
		end

		spaceRequired = spaceRequired + fileSize
	end
	
	-- Just in case...
	if numMissingFileSizes ~= 0 then
		local avgSize = spaceRequired / ( #batchOfDocValues - numMissingFileSizes )
		spaceRequired = spaceRequired + numMissingFileSizes * avgSize
	end

	return spaceRequired
	
end

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

local function createClearErrorOverlaysForThisBatch( batchOfDocValues )

	local bulkOverlay = {}

	for _, docVal in ipairs( batchOfDocValues ) do
		bulkOverlay[ #bulkOverlay + 1 ] = WFCatOzErrors.createOverlayToRemoveOzError( docVal._id )
	end

	return bulkOverlay
end

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

local function resetQuotaExceeded( catalog, availableSpace, optCalm )

	local spaceAvailable = availableSpace
	
	-- How moch to overshoot, to ensure that we hit the quota exceeded condition
	-- again, if there are enough documents from which to lift the quota exceeded
	-- error to do so.
	
	local delta = 0.1 * availableSpace

	docsOverOzStorageQuota( catalog, optCalm or 60 )
		:take( 1 )
		:flatMap( function( docIds )

			return WFRxThrottledBatchedBulkDocValue( "resetQuotaExceeded", catalog:docStore(), docIds, 100, 10 )

				-- note we use map here so we return a stream of observables as flatMerge expects to get observables.
				:map( function( batchOfDocValues )	
					return Observable.defer( function()

						local spaceUsedByThisBatch = spaceNeededByThisBatch( batchOfDocValues )
			
						spaceAvailable = spaceAvailable - spaceUsedByThisBatch 
						
						local clearErrorOverlays = createClearErrorOverlaysForThisBatch( batchOfDocValues )
						
						return catalog:docStore():bulkApplyOverlay( clearErrorOverlays )
							:flatMap( function( overlayResults )
			
								if spaceAvailable < 0 - delta then
									return Observable.throw( "OutOfSpace" )
								end
			
								return Observable.just( true )
			
							end )
					end )	
				end )
				:flatMerge( 1 ) -- only allow 1 batch to operate in parallel
		end )
		:call( WFRxRunUntilDone )

end

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

local function storageEntitlement( wfCatalog )
	
	check( 'WFCatalog' )
	
	return wfCatalog:ozUser():entitlement()
		:filter( function( value ) return not not value end )
		:map( function ( value )
			return value
		end )
		:pluck( 'storage' )
end

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

local function watchForQuotaReset( catalog, options )

	options = options or {}

	return Observable.combineLatest( storageEntitlement( catalog ), 
											countOverOzStorageQuota( catalog, options.calmSeconds ) )
		:map( function( storage, docsQuotaExceeded )
			
			-- check local cache of account document to see whether we think there's storage 
			-- available.
			return storage.used < storage.limit and docsQuotaExceeded > 0
			
		end )
		:sample( options.sampleSeconds or 30, WFRxScheduler.sharedScheduler() ) -- emit no more than every 30s
		:flatMapLatest( function( shouldResetBasedOnLocalInfo )
			
			if shouldResetBasedOnLocalInfo then
				-- So far, we think we should reset the quota. But this is based on the local
				-- understanding of our storage usage on Oz. This might be out of date (NMB-6749)
				-- so get the latest account doc from Oz and see if it agrees in terms of having 
				-- available space.
				return catalog:ozUser():getAccountDocFromOz( options )
					:map( function( updatedAccountDoc )
					
						local isOverStorage, availableSpace = isAccountDocOverStorage( updatedAccountDoc )
					
						return not isOverStorage, availableSpace
					
					end )
			
			else
				return Observable.just( false )
			end
			
		end )
		:catch( function( maybeError )
			logger:errorf( "Unable to check storage space in the user account document: %s", maybeError )
			return Observable.just( false )
		end )			
		:subscribe( function( resetQuota, availableSpace )

			if resetQuota then
				resetQuotaExceeded( catalog, availableSpace, options.calmSeconds )
			end
			
		end )

end

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

local overStorageQuota = function( wfCatalog )

	return Observable.combineLatest( function( overQuotaCount, ozStorage )

		local isOverQuota = overQuotaCount > 0
		
		if not isOverQuota then
		
			if ozStorage and ozStorage.used and ozStorage.display_limit then
				isOverQuota = ozStorage.used >= ozStorage.display_limit
			end
			
		end

		return isOverQuota

	end,
	countOverOzStorageQuota( wfCatalog ), 
	storageEntitlement( wfCatalog ) )
	
end

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

local function storageLimitPercentExceeded( storageUsed, storageLimit, limitPercent )

	local percent = 0

	if storageUsed and storageLimit and storageLimit > 0 then
		percent = storageUsed / storageLimit
	end

	return limitPercent and percent > limitPercent

end

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

local function lowOnStorage( wfCatalog )

	return storageEntitlement( wfCatalog )
	
		:map( function( ozStorage )

			local storage_limit = ozStorage.display_limit
			local storage_warn_percent = kLowOnStoragePercent
			local storage_warn = ozStorage.warn
			
			if storage_limit > 0 and storage_warn > 0 and storage_warn < storage_limit then
				storage_warn_percent = storage_warn / storage_limit
			end

			return storageLimitPercentExceeded( ozStorage.used, ozStorage.display_limit, storage_warn_percent )

		end )

end
	
--------------------------------------------------------------------------------

if DEBUG_STORAGE_USED then

	-- Override storageEntitlement and countOverOzStorageQuota in order
	-- to simplify QE verification of cloud icon and sync activity popup.

	debugStorageUsedStream = BehaviorSubject( 0 )

	local storageEntitlementImp = storageEntitlement
	
	storageEntitlement = function( wfCatalog )

		return Observable.combineLatest( function( ozData, debugUsedAmount )
			ozData.storage_used = debugUsedAmount
			return ozData
		end, 
		storageEntitlementImp( wfCatalog ), 
		debugStorageUsedStream )

	end

	local function outOfStorage( wfCatalog )

		return storageEntitlement( wfCatalog )
		
			:map( function( ozStorage )
				return storageLimitPercentExceeded( ozStorage.used, ozStorage.display_limit, kOutOfStoragePercent )
			end )

	end

	local countOverOzStorageQuotaImp = countOverOzStorageQuota

	countOverOzStorageQuota = function( wfCatalog )

		return Observable.combineLatest( function( countOfAssetsOverStorage, storageLimitExceededByPercent )

			if storageLimitExceededByPercent and countOfAssetsOverStorage == 0 then
				countOfAssetsOverStorage = 1
			end

			return countOfAssetsOverStorage

		end, 
		countOverOzStorageQuotaImp( wfCatalog ), 
		outOfStorage( wfCatalog ) )

	end

end

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

local function debugStorageUsed( amountUsed )

	if debugStorageUsedStream then
		debugStorageUsedStream:onNext( amountUsed )
	end
	
end

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

local function makeStaticMember( fn )

	return function( _, ... ) return fn( ... ) end

end 

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

local WFCatStorage = {
	
	_catalog = false,
	
	_watchQuotaForResetSubscription = false,
		-- (Subscription) watches for available space when docs are over quota to see if they can
		-- be resent
	
	_watchForDocsOverStorageSubscription = false,
		-- (Subscription) watches for StorageExceededErrors so that we can verify how much space is
		-- available
		
	_watchQuotaForOverageSubscription = false,
	
	_isOverStorage = Rx.ReplaySubject( 1 ),
		-- (ReplaySubject) simple behavior tracking whether the client is over storage on Oz or not

	class = {
		generateIndexSchema = generateIndexSchema,
		storageEntitlement = makeStaticMember( storageEntitlement ),
		countOverOzStorageQuota = makeStaticMember( countOverOzStorageQuota ),
		docsOverOzStorageQuota = makeStaticMember( docsOverOzStorageQuota ),
		overStorageQuota = makeStaticMember( overStorageQuota ),
		lowOnStorage = makeStaticMember( lowOnStorage ),
		debugStorageUsed = makeStaticMember( debugStorageUsed )
	},
	
	WFCloseSupport.addCloseSupportMethods,
}

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

function WFCatStorage:__init( wfCatalog, options )

	check( 'WFCatStorage', 'WFCatalog' )
	self:setOwned {
		_catalog = wfCatalog:relayObject()
	}
	
	self._watchQuotaForResetSubscription = watchForQuotaReset( self._catalog, options )
	
	self._isOverStorage:onNext( false )
	
	self._watchForDocsOverStorageSubscription = Observable.combineLatest( storageEntitlement( self._catalog ),
																			countOverOzStorageQuota( self._catalog ) )
		:flatMapLatest( function( localCacheOfOzStorageInfo, countOverStorage )
			if countOverStorage > 0 then
				if ( localCacheOfOzStorageInfo.used <= localCacheOfOzStorageInfo.limit ) then
				
					return self._catalog:ozUser():getAccountDocFromOz( options )
						:catch( Observable.just() )
						:map( function( userAccountDoc )
							local isOverQuota = isAccountDocOverStorage( userAccountDoc )
							self._isOverStorage:onNext( isOverQuota )
						end )
						
				else
					self._isOverStorage:onNext( true )
				end
						
			else
				self._isOverStorage:onNext( false )
			end
			
			return Observable.just()
		end )
		:subscribe()
		
	self._watchQuotaForOverageSubscription = storageEntitlement( self._catalog )
		:map( function( storage )
			return storage.used > storage.limit
		end )
		:subscribe()
		
end

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

function WFCatStorage:close()
	
	if self._watchQuotaForResetSubscription then
		self._watchQuotaForResetSubscription:dispose()
		self._watchQuotaForResetSubscription = nil
	end
	if self._watchForDocsOverStorageSubscription then
		self._watchForDocsOverStorageSubscription:dispose()
		self._watchForDocsOverStorageSubscription = nil
	end
	if self._watchQuotaForOverageSubscription then
		self._watchQuotaForOverageSubscription:dispose()
		self._watchQuotaForOverageSubscription = nil
	end
end

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

function WFCatStorage:isOverStorage()
	return self._isOverStorage
end

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

return  Rx.Class 'WFCatStorage'( WFCatStorage )
