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

LrClassicVolumeAgent.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2024 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.

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

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

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFRxContinuousTask = require 'WFRxContinuousTask'
local WFRxPersistentTimer = require 'WFRxPersistentTimer'
local WFRxScheduler = require 'WFRxScheduler'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

	-- WF.oz_client.managed_catalog
local WFCatPurgeHelper = require 'WFCatPurgeHelper'
local WFCatVolumeAgentQueryHelper = require 'WFCatVolumeAgentQueryHelper'

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

local check = assert( WFArguments.check )

local runUntilDone = assert( WFRxRunUntilDone )

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

local kUploadedXmpDevelopAndAuxPurgeFrequency = 60 * 60

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

local function processBatchOfBinariesForPurge( self, wfCatalog, binaryType,
												batchOfBinaries, getNextBatch,
												purgeList, purgeStream )

	-- This function processes a batch of purge candidates and remove entries
	-- that have already been marked for purge.

	check( 'LrClassicVolumeAgent', 'WFCatalog', 'string', 'table', 'function', 'table', 'table' )

	if #batchOfBinaries == 0 then

		purgeStream:onNext( purgeList )
		purgeStream:onCompleted()
		return

	end

	for _, binaryInfo in ipairs( batchOfBinaries ) do

		local docId = binaryInfo[ 2 ]
		local maybePurgeSpecifics = binaryInfo[ 4 ] or true

		if binaryType == 'xmp_develop'
			or ( binaryType == 'aux'
				and type( maybePurgeSpecifics ) == 'table'
				and next( maybePurgeSpecifics ) ) then

			purgeList.binaryType = binaryType
			purgeList[ docId ] = maybePurgeSpecifics

		end

	end

	-- Get the next batch of binaries and continue purging.

	Observable.defer( function()

		if WFCloseSupport.hasBeenOrWillBeClosed( wfCatalog ) then

			purgeStream:onNext( {} )
			purgeStream:onCompleted()

			return Observable.just( true )

		end

		return getNextBatch()

			:map( function( newBatchOfBinaries, newGetNextBatch )

				processBatchOfBinariesForPurge( self, wfCatalog, binaryType,
						newBatchOfBinaries, newGetNextBatch, purgeList, purgeStream )

				return newBatchOfBinaries, newGetNextBatch

			end )

	end )

	:call( runUntilDone )

end

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

local function fetchBinariesToPurge( self, wfCatalog, binaryType )

	-- This function fetches the first batch of purge candidates and starts
	-- processing them. See the API definitions within WFCatVolumeAgentQueryHelper
	-- for more information.

	check( 'LrClassicVolumeAgent', 'WFCatalog', 'string' )

	local processBinariesStream = Rx.ReplaySubject( 1 )

	if WFCloseSupport.hasBeenOrWillBeClosed( wfCatalog ) then

		processBinariesStream:onNext( {} )
		processBinariesStream:onCompleted()

	end

	local purgeList = {}

	WFCatVolumeAgentQueryHelper.getBatchOfPurgeableBinaries( wfCatalog, binaryType )

		:map( function( batchOfBinaries, getNextBatch )

			processBatchOfBinariesForPurge( self, wfCatalog, binaryType,
					batchOfBinaries, getNextBatch, purgeList, processBinariesStream )

			return batchOfBinaries, getNextBatch

		end )

		:call( runUntilDone )

	return processBinariesStream

end

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

local function ongoingUploadedXmpDevelopAndAuxPurge( self, wfCatalog )

	local purgeDelayOnStartup = self._debugMode and 0.01 or 30
		-- Wait 30 seconds before starting purge.

	local purgeFrequency = self._debugMode and 10 or kUploadedXmpDevelopAndAuxPurgeFrequency
		-- Every hour, run uploaded aux purge.

	return Observable.timer( purgeDelayOnStartup, WFRxScheduler.sharedScheduler() )

		:take( 1 )

		:flatMap( WFRxPersistentTimer( wfCatalog:docStore(), 'uploadedXmpDevelopAndAuxPurge', purgeFrequency ) )

		-- Need to throw as purgeStream doesn't ever complete.

		:flatMapLatest( function()

			return self._uploadedXmpDevelopAndAuxPurgeTask:throwIfNotRunning()

		end )

		:flatMap( function()

			return Observable.combineLatest(
					fetchBinariesToPurge( self, wfCatalog, 'xmp_develop' ),
					fetchBinariesToPurge( self, wfCatalog, 'aux' ) )

		end )

		:flatMap( function( xmpDevelopPurgeList, auxPurgeList )

			local purgeList = {}

			if next( xmpDevelopPurgeList ) then
				purgeList[ #purgeList + 1 ] = xmpDevelopPurgeList
			end

			if next( auxPurgeList ) then
				purgeList[ #purgeList + 1 ] = auxPurgeList
			end

			-- If there're purgeable xmp_develop or aux binaries, continue to delete files
			if next( purgeList ) then
				return WFCatPurgeHelper.deleteFilesByPurgeList( wfCatalog, purgeList )
			end

			return Observable.just( true )

		end )

		:catch( function( errorMessage )

			return self._uploadedXmpDevelopAndAuxPurgeTask:throwIfNotStopped( errorMessage )

		end )

end

--============================================================================--
-- @mark WFCatVolumeAgent APIs
--============================================================================--

local LrClassicVolumeAgent = {

	_debugMode = false,
		-- (boolean) If true, the volume agent purge will trigger right away. Used in testing

	_uploadedXmpDevelopAndAuxPurgeTask = false,
		-- (WFRxContinuousTask) A continuous task to handle uploaded aux purging

	WFCloseSupport.addCloseSupportMethods,

}

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

function LrClassicVolumeAgent:__init( optOverrides )

	check( 'LrClassicVolumeAgent', '?table' )

	optOverrides = optOverrides or {}

	self._debugMode = optOverrides.debugMode or false

end

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

function LrClassicVolumeAgent:close()

	check( 'LrClassicVolumeAgent' )

	local uploadedXmpDevelopAndAuxPurgeTask = self._uploadedXmpDevelopAndAuxPurgeTask
	self._uploadedXmpDevelopAndAuxPurgeTask = nil

	if uploadedXmpDevelopAndAuxPurgeTask then
		uploadedXmpDevelopAndAuxPurgeTask:close()
	end

end

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

function LrClassicVolumeAgent:setupUploadedXmpDevelopAndAuxPurgeTask( wfCatalog )

	check( 'LrClassicVolumeAgent', 'WFCatalog' )

	-- purge uploaded aux

	local uploadedXmpDevelopAndAuxPurgeTaskDelays = {
		progressiveBackoff = false,
		errorDelay = 5,
		maxDelay = 60,
	}

	local uploadedXmpDevelopAndAuxPurgeTask = function()

		return ongoingUploadedXmpDevelopAndAuxPurge( self, wfCatalog )

	end

	self._uploadedXmpDevelopAndAuxPurgeTask = WFRxContinuousTask( "LrClassicUserModel:uploadedXmpDevelopAndAuxPurge", uploadedXmpDevelopAndAuxPurgeTask, uploadedXmpDevelopAndAuxPurgeTaskDelays )
	self._uploadedXmpDevelopAndAuxPurgeTask:startContinuousTask()

end

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

return Rx.Class 'LrClassicVolumeAgent'( LrClassicVolumeAgent )
