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

WFCatStuckSyncUtils.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2021 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 KSIdGenerator = require 'KSIdGenerator'
local KSDate = require 'KSDate'
local KSTableUtils = require 'KSTableUtils'
local WFLogger = require 'WFLogger'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxDocValueWatcher = require "WFRxDocValueWatcher"
local WFRxScheduler = require 'WFRxScheduler'
local WFRxThrottledBatchedBulkApplyOverlay = require 'WFRxThrottledBatchedBulkApplyOverlay'
local WFRxThrottledBatchedBulkDocValue = require 'WFRxThrottledBatchedBulkDocValue'
local WFAnalytics = require "WFAnalytics"

	--WF.oz_client.shared.utils
local LrAssetUtils = require 'LrAssetUtils'

	-- WF.managed_catalog
local WFCatMetadataPusher = require 'WFCatMetadataPusher'
local WFCatOzErrors = require 'WFCatOzErrors'
local WFCatSyncStatus = require 'WFCatSyncStatus'

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

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt )

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

local logger = WFLogger 'WFCatStuckSyncUtils'

local kStuckReportingTimeThreshold = 300 -- seconds

local kGlobalDocumentId

if WFPlatformInfo.isMobile() then

	local WFOzSyncConstants = require 'WFOzSyncConstants'

	kGlobalDocumentId = WFOzSyncConstants.kGlobalDocumentId

else

	kGlobalDocumentId = '00000000000000000000000000000000'

end

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

local WFCatStuckSyncUtils = {}

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

local function checkStuckDocumentAgainstThreshold( 
		maybeStuckDocIdsAndTimestamps, stuckTimeThreshold, period )

	-- Every N seconds (period), evaluate the set of stuck doc ids to see if
	-- any of them have been stuck for a time period greater than the
	-- stuck sync time threshold.
	
	if not next( maybeStuckDocIdsAndTimestamps ) then
		return Observable.just( {} )
	end
	
	return Observable.timer( 0, period, WFRxScheduler.sharedScheduler() )
	
		:map( function()
			
			local stuckDocIds = {}

			if next( maybeStuckDocIdsAndTimestamps ) then

				logger:infof( "Now checking %s against the stuck sync threshold of %d", maybeStuckDocIdsAndTimestamps, stuckTimeThreshold )

				local currentTime = KSDate.timeNow()

				for docId, stuckTimeStamp in pairs( maybeStuckDocIdsAndTimestamps ) do

					if currentTime - stuckTimeStamp > stuckTimeThreshold then

						logger:infof( "Asset %s is stuck", docId )
						stuckDocIds[ #stuckDocIds + 1 ] = docId

					else
					
						logger:infof( "Asset %s is not yet considered stuck", docId )
					end

				end

			end

			return stuckDocIds
		
		end )				

end

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

--[[!

	\brief Get a stream that returns asset document IDs that the system believes
		are in a stuck sync state due to non-uploaded metadata. Calling this
		function multiple times will return the same observable stream.
		
	\param wfCatalog (WFCatalog)
	
	\param isNormalSyncRunningStream (Observable) An observable that fires with true
		when the sync processes are not paused, and the app is not offline. It
		should fire with false otherwise.
	
	\param optCalm (optional number) optional number of seconds indicating how often we
		should check for stuck documents. The default is 10 seconds.
		
	\param optStuckTimeThreshold(optional number) optional number of seconds an asset
		must be appear stuck before being recorded as stuck. Defaults to 180 seconds.
		
	Note: Assets may appear to be stuck in the window in between when data is sent to Oz
	and when a response is received. If this proves problematic, we may have to introduce
	a stuck sync time threshold.
		
	\return <code>Observable<table></code> An observable that fires periodically with a 
		list of document Ids that may be in a stuck sync state.
		
--]]

local instanceOfDocsStuckSyncingMetadataStream

function WFCatStuckSyncUtils.getDocsStuckSyncingMetadataStream( 
		wfCatalog, isNormalSyncRunningStream, optCalm, optStuckReportingTimeThreshold )

	if wfCatalog:ozUserId() == kGlobalDocumentId then
		return Observable.never() -- Don't track stuck syncs for freemium users.
	end

	if not instanceOfDocsStuckSyncingMetadataStream then

		local stuckTimeThreshold = optStuckReportingTimeThreshold or kStuckReportingTimeThreshold

		local docStore = wfCatalog:docStore()

		local assetsNeedingMetadataUploadSql

		if WFPlatformInfo.isMobile() then -- iOS & Android
			assetsNeedingMetadataUploadSql = [[SELECT docId FROM coreInfo WHERE syncStatus == 1]]
		else -- Desktop
		
			local syncStatusColumnName = WFCatSyncStatus.getSyncStatusColumnName()
			assetsNeedingMetadataUploadSql = string.format( [[SELECT assetId FROM asset_sync_status_ui WHERE %s == 1]], syncStatusColumnName )
			
		end

		local assetsNeedingMetadataUploadStream = wfCatalog:indexStore():query( {
				sql = assetsNeedingMetadataUploadSql,
				calm = optCalm or 30,
				singleColumnQuery = true
			} )

			:flatMapLatest( function( assetIds )
			
				local assetDocValueStream = WFRxDocValueWatcher.make( Observable.just( assetIds ), docStore )
				
					:bufferWithTimeOrCount( optCalm or 1, 100, WFRxScheduler.sharedScheduler() )
					
					:scan( function( docValues, batchOfDocValues )
					
						for _, docValue in ipairs( batchOfDocValues ) do
						
							docValues[ docValue._id ] = docValue
						
						end
					
						return docValues
						
					end, {} )
					
				return assetDocValueStream
			
			end )

		local catalogPusherLocalDocValueId = WFCatMetadataPusher.getContinueFromLocalDocValueKey( false )
				
		local catalogPusherChangeCounterStream = docStore:ongoingLocalDocValue( catalogPusherLocalDocValueId )
				:pluck( 'continueFrom' )
		
		local spacePusherLocalDocValueId = WFCatMetadataPusher.getContinueFromLocalDocValueKey( true )
		
		local spacePusherChangeCounterStream = docStore:ongoingLocalDocValue( spacePusherLocalDocValueId )
				:pluck( 'continueFrom' )

		local maybeStuckDocIdsAndTimestamps = {}

		isNormalSyncRunningStream = isNormalSyncRunningStream:distinctUntilChanged()

		instanceOfDocsStuckSyncingMetadataStream = Observable.combineLatest( 
				isNormalSyncRunningStream, assetsNeedingMetadataUploadStream,
				catalogPusherChangeCounterStream, spacePusherChangeCounterStream )
		
			:filter( function( isSyncRunning )

				-- To reduce false positives, we don't check for stuck assets unless
				-- sync is enabled ("enabled" meaning not paused, and not offline).
			
				if not isSyncRunning then

					maybeStuckDocIdsAndTimestamps = {}

				end

				return isSyncRunning

			end )

			:flatMapLatest( function( _, maybeStuckDocValues, catalogPusherChangeCounter, spacePusherChangeCounter )

				-- If ever the change counter of the document is smaller than the change counter
				-- of the associated pusher, and the document has unsynced metadata, then the
				-- document may be in a stuck sync state.
				
				-- In order to reduce false positives, we track the timestamp of when the
				-- stuck sync was detected. Only docs that have been stuck for a duration
				-- of time greater than the stuck sync time threshold are emitted by this
				-- code.
			
				catalogPusherChangeCounter = catalogPusherChangeCounter or -1
				spacePusherChangeCounter = spacePusherChangeCounter or -1
				
				for docId, docValue in pairs( maybeStuckDocValues ) do
				
					-- Make sure we're ignoring deleted assets, assets in conflict, and
					-- other assets that should not be sent to Oz. The index store query
					-- should be filtering these documents out, but it feels safer to a guard
					-- here as well.
					
					if not docValue._deleted
							and not docValue._conflict
							and not WFCatOzErrors.shouldSkipUploadDueToError( docValue ) then

						if docValue._spaceId then

							if docValue._changeCounter < spacePusherChangeCounter
									and not LrAssetUtils.isLatestMetadataUploaded( docValue ) then

								if not maybeStuckDocIdsAndTimestamps[ docId ] then

									logger:infof( "SWY asset %s might be stuck", docId )
									maybeStuckDocIdsAndTimestamps[ docId ] = KSDate.timeNow()

								end

							else

								logger:infof( "SWY asset %s is not stuck", docId )
								maybeStuckDocIdsAndTimestamps[ docId ] = nil

							end

						elseif docValue._changeCounter < catalogPusherChangeCounter
								and not LrAssetUtils.isLatestMetadataUploaded( docValue ) then

							if not maybeStuckDocIdsAndTimestamps[ docId ] then

								logger:infof( "Asset %s might be stuck", docId )
								maybeStuckDocIdsAndTimestamps[ docId ] = KSDate.timeNow()

							end

						else

							logger:infof( "Asset %s is not stuck", docId )
							maybeStuckDocIdsAndTimestamps[ docId ] = nil

						end

					end

				end

				-- Stop tracking assets that are no longer stuck

				for docId in pairs( maybeStuckDocIdsAndTimestamps ) do

					if not maybeStuckDocValues[ docId ] then

						maybeStuckDocIdsAndTimestamps[ docId ] = nil
						
						logger:infof( "Asset %s is no longer stuck", docId )
						
					end
					
				end
				
				return checkStuckDocumentAgainstThreshold( maybeStuckDocIdsAndTimestamps, stuckTimeThreshold, optCalm or 10 )
				
			end )

			:replay( nil, 1 )
							
			:refCount()

			:finally( function()
				instanceOfDocsStuckSyncingMetadataStream = nil
			end )

	end
	
	return instanceOfDocsStuckSyncingMetadataStream

end

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

local function wasPreviousRevisionGeneratedToResolveStuckSync( docValue )

	local lastGeneratedRevIdToResolveStuckSync
			= KSTableUtils_derefOpt( docValue, '_localOnly', 'lastGeneratedRevIdToResolveStuckSync' )

	return not not lastGeneratedRevIdToResolveStuckSync

end

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

local function generateRevisionOverlayToResolveStuckSync( docValue )

	local newRevId = KSIdGenerator.uuidInLowercaseHex()

	local overlay = {
	
		_id = docValue._id,
		
		_newRevId = newRevId,
			-- bump the sequence counter in order to trigger a resync.
			
		_localOnly = {
		
			-- Track the auto-generated revision Id. In order to prevent loopers,
			-- we only attempt to resolve stuck syncs for a document once.
			
			lastGeneratedRevIdToResolveStuckSync = newRevId
			
		}
		
	}

	return overlay

end

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

local function trackResolveStuckMetadataDocuments( docId )

	local errorEvent = {}

	errorEvent['event.type'] = 'resolve_error'
	errorEvent['event.workflow'] = 'SYNC'
	errorEvent['event.subcategory'] = 'metadata_pusher'
	errorEvent['event.error_type'] = 'MetadataUpload.AutoResolveStuckSync'
	errorEvent['content.id'] = docId

	WFAnalytics.trackEvent( errorEvent )

end

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

local function trackStuckMetadataDocuments( docId, hasPreviousGeneratedRevision )

	local errorEvent = {}

	errorEvent['event.type'] = 'error'
	errorEvent['event.workflow'] = 'SYNC'
	errorEvent['event.subcategory'] = 'metadata_pusher'
	errorEvent['event.error_type'] = 'MetadataUpload.StuckSync'
	errorEvent['content.hasTriedToAutoResolveStuckSync'] = not not hasPreviousGeneratedRevision
	errorEvent['content.id'] = docId

	WFAnalytics.trackEvent( errorEvent )

end

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

--[[!

	\brief Attempts to resolve stuck syncs for a given list of document IDs. This API
		only works for assets stuck due to non-uploaded metadata.
		
	\param wfCatalog (WFCatalog) 
	\param stuckDocIds (table) a table of document IDs in a stuck sync state.
	
	\return <code>Observable</code> An observable that completes after
		attempting to resolve the stuck sync.
		
--]]

function WFCatStuckSyncUtils.resolveStuckSync( wfCatalog, stuckDocIds )

	local overlayStream =
			WFRxThrottledBatchedBulkDocValue( "resolveStuckSync", wfCatalog:docStore(), stuckDocIds, 100, 10 )

				:scan( function( overlays, batchOfDocValues )

					for _, docValue in ipairs( batchOfDocValues ) do

						local hasPreviousGeneratedRevision = wasPreviousRevisionGeneratedToResolveStuckSync( docValue )

						trackStuckMetadataDocuments( docValue._id, hasPreviousGeneratedRevision )

						if not hasPreviousGeneratedRevision then -- Looper prevention

							trackResolveStuckMetadataDocuments( docValue._id )

							local overlay = generateRevisionOverlayToResolveStuckSync( docValue )

							overlays[ #overlays + 1 ] = overlay

						end

					end

					return overlays

				end, {} )

				:takeLast( 1 )

	return overlayStream

		:flatMap( function( overlays )

			return WFRxThrottledBatchedBulkApplyOverlay(
					"resolveStuckSync", wfCatalog:docStore(), overlays, 100, 10 )

				:takeLast( 1 )

		end )

end

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

--[[!

	\brief Get a stream that returns asset document IDs that the system believes
		are in a stuck sync state due to an non-uploaded original. Calling this
		function multiple times will return the same observable stream.
		
	\param wfCatalog (WFCatalog)
	
	\param isNormalSyncRunningStream (Observable) An observable that fires with true
		when the sync processes are not paused, and the app is not offline. It
		should fire with false otherwise.
	
	\param optCalm (optional number) optional number of seconds indicating how often we
		should check for stuck documents. The default is 10 seconds.
		
	\param optStuckTimeThreshold(optional number) optional number of seconds an asset
		must be appear stuck before being recorded as stuck. Defaults to 180 seconds.
		
	Note: Assets may appear to be stuck in the window in between when data is sent to Oz
	and when a response is received. If this proves problematic, we may have to introduce
	a stuck sync time threshold.
	
	\return <code>Observable<table></code> An observable that fires periodically with a 
		list of document Ids that may be in a stuck sync state.
		
--]]

local instanceOfDocsStuckUploadingOriginalStream

function WFCatStuckSyncUtils.getDocsStuckUploadingOriginalStream( 
		wfCatalog, isNormalSyncRunningStream, optCalm, optStuckReportingTimeThreshold )

	if wfCatalog:ozUserId() == kGlobalDocumentId then
		return Observable.never() -- Don't track stuck syncs for freemium users.
	end

	if not instanceOfDocsStuckUploadingOriginalStream then

		local stuckTimeThreshold = optStuckReportingTimeThreshold or kStuckReportingTimeThreshold

		local assetsNeedingOriginalUploadSql

		if WFPlatformInfo.isMobile() then -- iOS & Android
			assetsNeedingOriginalUploadSql = [[SELECT docId FROM coreInfo WHERE syncStatus == 2]]
		else -- Desktop
		
			local syncStatusColumnName = WFCatSyncStatus.getSyncStatusColumnName()
			assetsNeedingOriginalUploadSql = string.format( [[SELECT assetId FROM asset_sync_status_ui WHERE %s == 2]], syncStatusColumnName )
			
		end

		local assetsNeedingOriginalUploadQuery = wfCatalog:indexStore():query( {
				sql = assetsNeedingOriginalUploadSql,
				calm = optCalm or 10,
				singleColumnQuery = true
			} )
			
		local needsBinaryUploadQuery = wfCatalog:indexStore():query( {
			sql = [[SELECT DISTINCT docId FROM needsBinaryUpload]],
			calm = optCalm or 10,
			singleColumnQuery = true
		} )
			
		local maybeStuckDocIdsAndTimestamps = {}
			
		isNormalSyncRunningStream = isNormalSyncRunningStream:distinctUntilChanged()

		instanceOfDocsStuckUploadingOriginalStream = Observable.combineLatest( 
				isNormalSyncRunningStream, assetsNeedingOriginalUploadQuery, needsBinaryUploadQuery )
			
			:flatMapLatest( function( isSyncRunning, syncStatusAssetIds, needsBinaryUploadAssetIds )
			
				if not isSyncRunning then
					
					maybeStuckDocIdsAndTimestamps = {}
					
					return Observable.just( {} )
						-- To reduce false positives, we don't check for stuck assets unless 
						-- sync is enabled ("enabled" meaning not paused, and not offline).
						
				end
		
				local needsBinaryUploadAssetIdsSet = needsBinaryUploadAssetIds and KSTableUtils.invert( needsBinaryUploadAssetIds )
			
				for _, docId in ipairs( syncStatusAssetIds ) do
				
					if not needsBinaryUploadAssetIdsSet[ docId ] then
					
						if not maybeStuckDocIdsAndTimestamps[ docId ] then
							logger:infof( "Asset %s original might be stuck", docId )
							maybeStuckDocIdsAndTimestamps[ docId ] = KSDate.timeNow()
						end
						
					else
					
						logger:infof( "Asset %s original is not stuck", docId )
						maybeStuckDocIdsAndTimestamps[ docId ] = nil
					
					end
					
				end
				
				-- Stop tracking assets that are no longer stuck
				
				local syncStatusAssetIdsSet = syncStatusAssetIds and KSTableUtils.invert( syncStatusAssetIds )
				
				for docId in pairs( maybeStuckDocIdsAndTimestamps ) do

					if not syncStatusAssetIdsSet[ docId ] then

						logger:infof( "Asset %s original is no longer stuck", docId )
						maybeStuckDocIdsAndTimestamps[ docId ] = nil
						
					end
					
				end
				
				return checkStuckDocumentAgainstThreshold( maybeStuckDocIdsAndTimestamps, stuckTimeThreshold, optCalm or 10 )
				
			end )

			:replay( nil, 1 )
						
			:refCount()

			:finally( function()
				instanceOfDocsStuckUploadingOriginalStream = nil
			end )

	end
	
	return instanceOfDocsStuckUploadingOriginalStream

end

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

function WFCatStuckSyncUtils.trackStuckBinaryDocuments( documentInfos )

	for _, v in ipairs( documentInfos ) do
					
		local errorEvent = {}

		errorEvent['event.type'] = 'error'
		errorEvent['event.workflow'] = 'SYNC'
		errorEvent['event.subcategory'] = 'binary_uploader'
		errorEvent['event.error_type'] = 'BinaryUpload.StuckSync'
		errorEvent['content.id'] = v
		
		WFAnalytics.trackEvent( errorEvent )
	
	end
	
end

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

function WFCatStuckSyncUtils.addDiagnosticLogSectionFunc( wfCatalog )

	local fireWithTrue = Observable.just( true )

	WFDiagnosticLog.addLogSectionFunc( function()
	
		local body = {
		
			_README_ = "Note: Assets may appear to be stuck in the window in between when data is sent to Oz, and when a response is received.",
			
			docsStuckSyncingMetadata
					= WFCatStuckSyncUtils.getDocsStuckSyncingMetadataStream( wfCatalog, fireWithTrue )
						:take( 1 ),
						
			docsStuckUploadingOriginal
					= WFCatStuckSyncUtils.getDocsStuckUploadingOriginalStream( wfCatalog, fireWithTrue )
						:take( 1 )
						
		}

		return {

			name = "Stuck Sync Detection",
			id = 'WFCatStuckSyncUtils',
			weight = 510,
			parentSection = WFDiagnosticLog.Sections.SYNC,
			body = body

		}

	end )

end

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

return WFCatStuckSyncUtils
