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

WFCatConflictResolver.lua

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

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

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

	-- RxLua
local Rx = require 'Rx'

	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.core.utils
local KSIdGenerator = require 'KSIdGenerator'
local KSTableUtils = require 'KSTableUtils'
local WFAnalytics = require 'WFAnalytics'
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local WFDiagnosticLog = require 'WFDiagnosticLog'
local WFTableMerge = require 'WFTableMerge'

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

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

local logger = WFLogger 'WFCatConflictResolver'
--	:setLevel( 'debug' )

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

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

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

local kConflictResolution_delete = 'delete'
local kConflictResolution_deleteLocally = 'deleteLocally'
local kConflictResolution_threeWayMerge = 'threeWayMerge'
local kConflictResolution_localRevByUserUpdated = 'localRevByUserUpdated'
local kConflictResolution_remoteRevByUserUpdated = 'remoteRevByUserUpdated'
local kConflictResolution_localRevWithOzUserUpdated = 'localRevWithOzUserUpdated'
local kConflictResolution_localRevByChangeCounter = 'localRevByChangeCounter'
local kConflictResolution_remoteRevByChangeCounter = 'remoteRevByChangeCounter'
local kConflictResolution_errorArbitrarilyLocalRev = 'errorArbitrarilyLocalRev'

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

-- @mark WFCatConflictResolver namespace

local function generateIndexSchema( _, indexStoreSchema )

	check( '_', 'WFRxIndexStoreSchema' )

	indexStoreSchema:addTable {

		sql = [[CREATE TABLE conflicts( sequence )]],

		relevantTypeSubtypeTuples = {
			{ '*', '*' },
		},

		map = function( docValue, emit )

			if docValue._conflict then
				emit( docValue._changeCounter )
			end

		end,

		hasOneRowPerDocument = true,

	}
	
	indexStoreSchema:addIndex [[CREATE INDEX conflicts_sequence ON conflicts( sequence )]]
	
end

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

local function stripNoiseFromRev( docValue )
	
	if not docValue then
		return
	end

	docValue._changeCounter = nil
	docValue._conflict = nil
	docValue._revId = nil

end

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

local function preserveOzDataFromRemoteRevForLocalRev( remoteRev, localRevId )

	-- Always preserve purgeDate, userCreated, original, and _serverProvidedMetadata from oz

	return {
		_id = remoteRev._id,
		_newRevId = localRevId,
		_serverProvidedMetadata = remoteRev._serverProvidedMetadata,
		purgeDate = remoteRev.purgeDate,
		userCreated = remoteRev.userCreated,
		original = remoteRev.original,
	}

end

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

local function recordConflictResolution( self, resolutionType, id )

	local ids = self._conflictResolutionSummary[ resolutionType ] or {}
	ids[ #ids + 1 ] = id
	self._conflictResolutionSummary[ resolutionType ] = ids

end

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

local function resolveConflictingRevs( self, docStore, localRev, remoteRev )

	-- Note: The variable naming here reflects the *likely* scenario given the
	-- doc store's strategy of making bulk changes (i.e. typically changes received
	-- from Oz) be the losing revision. Not guaranteed, but highly likely.

	local localHistory = docStore:docRevisionHistory( localRev._id, localRev._revId )
	local remoteHistory = docStore:docRevisionHistory( remoteRev._id, remoteRev._revId )

	local remoteRevId = remoteRev._revId
	local localRevId = localRev._revId
	local kRemoteRevAlreadyDeleted = 'remoteRevAlreadyDeleted'

	return Rx.Observable.combineLatest(
		function( localRevList, remoteRevList )

			logger:debugf( "localRevList = %s", localRevList )
			logger:debugf( "remoteRevList = %s", remoteRevList )

			-- Identify the first available revision that occurs in both lists
			-- and use it as base for three-way merge.

			local localRevIds = {}
			for _, info in ipairs( localRevList ) do
				if info.status == 'available' then
					localRevIds[ info.revId ] = true
				end
			end

			for _, info in ipairs( remoteRevList ) do
				if info.status == 'available' and localRevIds[ info.revId ] then
					return info.revId
				end
			end
			
			return nil, #remoteRevList

		end, localHistory, remoteHistory )

		:take( 1 )
		:flatMapLatest( function( baseRevId, numberOfRemoteRevisions )

			if baseRevId then
				return docStore:docValue( localRev._id, baseRevId )
			else
				return Rx.Observable.just( nil, numberOfRemoteRevisions )
			end

		end )

		:flatMapLatest( function( baseRev, numberOfRemoteRevisions )

			logger:debugf( "localRev = %s", localRev )
			logger:debugf( "remoteRev = %s", remoteRev )
			logger:debugf( "baseRev = %s", baseRev )
			
			local localRevChangeCounter = localRev._changeCounter
			local remoteRevChangeCounter = remoteRev._changeCounter
			
			stripNoiseFromRev( localRev )
			stripNoiseFromRev( remoteRev )
			stripNoiseFromRev( baseRev )
				
			if localRev._deleted or remoteRev._deleted then

				-- Cloud Trash has two different types of cases where we don't
				-- want to send a delete back for an asset:

				-- 1. A delete/restore or restore/delete operation was performed
				-- which results in the possibility of pulling an update while
				-- Oz is still processing the second update.

				-- 2. A delete, restore, or delete trash action was taken prior
				-- to getting confirmation that the action has been processed by
				-- Oz (didn't get the _serverProvidedMetadata yet). Cloud Trash
				-- compares the document's _changeCounter against the MetadataPusher
				-- continueFrom counter to determine whether Oz should know about
				-- the document being operated on.

				-- If we issue the delete in either of these cases we may end up
				-- removing documents, binaries, or associated documents that Oz
				-- still needs to complete the prior request.

				-- We could potentially mark the asset that one of these cases
				-- occurred but think it's unlikely that the document is left
				-- behind so we would rather ignore it and if it is a problem a
				-- different client could rectify it by issuing a delete.

				local suppressSendingNewDelete = localRev.type == 'asset'
					and derefOpt( localRev, '_newId' ) ~= nil

				recordConflictResolution( self, suppressSendingNewDelete
					and kConflictResolution_deleteLocally or kConflictResolution_delete, localRev._id )

				-- if either revision has been deleted, that's the change
				-- that should win so explicity delete the doc
				return docStore:resolveConflictWithDelete( localRev._id, suppressSendingNewDelete )

			end
			
			if KSTableUtils.deepEquals( localRev, remoteRev ) then
				-- If the remoteRev and localRev contain the same contents we don't need to do any
				-- complicated merging - just choose the remoteRev (because Oz already has that revId)
				-- and delete the local one. This may seem like it wouldn't happen often, but there
				-- are cases like stacks where local operations changing the stack cover on asset delete
				-- may be mirrored by Oz operations doing the same and ending up with the same new cover
				-- and different revision trees. This prevents a stuck sync in those cases.
				return docStore:deleteRevisionAsNonCurrent( localRev._id, localRevId )
					:map( function()
						return kRemoteRevAlreadyDeleted
					end )
			
			elseif baseRev then
				
				local mergedRev = WFTableMerge.threeWayMerge( baseRev, localRev, remoteRev )

				logger:infof( "mergedRev = %s", mergedRev )
				recordConflictResolution( self, kConflictResolution_threeWayMerge, localRev._id )

				local overlay = WFTableMerge.createOverlay( localRev, mergedRev )
				
				if not next( overlay ) then
					-- The merged revision is the same as the local revision but it doesn't match
					-- the remote revision which means we have changes to send to Oz. But since 
					-- there's not currently anything changing in the localRev 
					-- a) bump the sequence counter 
					-- b) ensure any _dontsendtooz labels are dropped
					return docStore:docValue( localRev._id, localRevId )
						:pluck( '_labels' )
						:flatMap( function( labels )
							if labels and labels._dontsendtooz then
								overlay._newRevId = KSIdGenerator.uuidInLowercaseHex()
							end
							
							overlay._id = localRev._id
							return docStore:applyOverlay( overlay )
						end )
				else 
					overlay._id = localRev._id
					return docStore:applyOverlay( overlay )
				end
				
			else

				-- We can't find a suitable base revision on which to do a three-way
				-- merge.
				
				if ( localRev.type and localRev.type ~= remoteRev.type ) or 
					( localRev.subtype and localRev.subtype ~= remoteRev.subtype ) then
				
					-- https://jira.corp.adobe.com/browse/WF-1583
					-- https://jira.corp.adobe.com/browse/WF-1802
					
					-- If the remoteRev and localRev have different types and/or subtypes, then
					-- we cannot resolve the conflict with any sort of merge or applyOverlay
					-- operation; the doc store does not allow you to change the type or subtype
					-- of a document once set. This scenario was never supposed to happen, though
					-- it appears to have been possible due to a Cloud Trash bug(s).
					
					-- The the case of WF-1583, the remote revision is correct so we just choose
					-- the remoteRev and delete the local one. Resolving to the remoteRev
					-- is likely the safest choice. No conflict resolution leads to a stuck sync.
					
					local event = {}

					event['event.type'] = 'typeOrSubtypeMismatchWarning'
					event['event.workflow'] = 'CONFLICT'
					event['event.subcategory'] = 'typeOrSubtypeMismatch'
					event['event.docId'] = localRev._id
					event['event.localRevType'] = localRev.type
					event['event.localRevSubtype'] = localRev.subtype
					event['event.remoteRevType'] = remoteRev.type
					event['event.remoteRevSubtype'] = remoteRev.subtype
					
					WFAnalytics.trackEvent( event )
					
					return docStore:deleteRevisionAsNonCurrent( localRev._id, localRevId )
						:map( function()
							return kRemoteRevAlreadyDeleted
						end )
						
				end

				local localRevIsMostRecent = false

				if derefOpt( localRev, '_localOnly', 'universal', 'trash' ) then

					if numberOfRemoteRevisions == 1
							and remoteRev._serverProvidedMetadata
							and remoteRev.type == "asset"
							and WFCatTrashUtils.createFromInfo( localRev ) then

						local localRevOzRevisionIds = derefOpt( localRev, '_serverProvidedMetadata', 'revision_ids' ) or {}
						local localRevisionIdIsOnOz = false
						
						for _, revId in ipairs( localRevOzRevisionIds ) do
						
							if revId == localRevId then
							
								localRevisionIdIsOnOz = true
								break
								
							end
						
						end

						if not localRevisionIdIsOnOz then

							-- NMB-11106 / NMB-11058 / WF-1372
							-- Scenario:
							-- 1 User adds an image, lets it sync, pauses sync,
							-- 2 modifies the asset
							-- 3 delete the asset
							-- 4 restore the asset
							-- 5 turn on sync.
							--
							-- In this scenario, a conflict is generated because the restored asset
							-- generated by Oz does not contain the modication from step 2. For this
							-- specific sceario, there is no base revision between the Oz restored asset
							-- and the locally restored asset because the locally restored asset was created
							-- from an asset with an unsynced revision. Furthermore, the userUpdated fields
							-- can be not be used to resolve the conflict correctly because the remote restored
							-- asset has it's userUpdated field updated at the time of document creation; it
							-- appears to be more recently updated than the locally restored asset that contains
							-- the modification.
							--
							-- We can solve this by always choosing the local resvision. This is safe to
							-- do because of the following:
							--
							-- If this is a document that's been through a trash operation and there's
							-- only one remote revision, we can safely assume that the remote revision is 
							-- an Oz generated restored asset. In this case the content of the remote revision
							-- reflects the content at the time the original asset was moved into the trash. 
							-- Therefore, if there's a conflict between the localRev and the remoteRev, we
							-- know that the localRev is going to be more recently updated.

							localRevIsMostRecent = true

						end

					end

					-- For assets that were run through a trash operation use the
					-- userUpdated field to determine which overlay should be
					-- considered the latest modifications.

					if localRev.userUpdated and remoteRev.userUpdated then

						if remoteRev.userUpdated >= localRev.userUpdated and not localRevIsMostRecent then

							logger:warnf( "resolving %s with remoteRev by userUpdated", localRev._id )
							recordConflictResolution( self, kConflictResolution_remoteRevByUserUpdated, localRev._id )

							local overlay = WFTableMerge.createOverlay( localRev, remoteRev )

							overlay._id = localRev._id

							-- We are using the remote rev, so use it's revId to avoid
							-- attempting to send changes back up to oz.
							overlay._newRevId = remoteRevId

							-- Delete the existing remoteRevId as we are adding a new one with the
							-- same revId and don't want to delete it.
							return docStore:deleteRevisionAsNonCurrentAndApplyOverlay( remoteRev._id, remoteRevId, overlay )
							:map( function()
								return kRemoteRevAlreadyDeleted
							end )

						else

							logger:warnf( "resolving %s with localRev by userUpdated", localRev._id )
							recordConflictResolution( self, kConflictResolution_localRevByUserUpdated, localRev._id )

							local overlay = preserveOzDataFromRemoteRevForLocalRev( remoteRev, localRevId )
							return docStore:applyOverlay( overlay )

						end

					elseif localRev.userUpdated == nil and localRev.type == "asset" then
				
						logger:debugf( "resolving %s with localRev and remoteRev Oz data ", localRev._id )
						recordConflictResolution( self, kConflictResolution_localRevWithOzUserUpdated, localRev._id )

						local overlay = preserveOzDataFromRemoteRevForLocalRev( remoteRev, localRevId )
						overlay.userUpdated = remoteRev.userUpdated
						
						return docStore:applyOverlay( overlay )

					end

				end
				
				if localRevChangeCounter and remoteRevChangeCounter then

					-- Use the changeCounter to determine which conflicting rev should win.
					-- The reasoning is, that a higher changeCounter usually means a "newer" revision.

					if remoteRevChangeCounter > localRevChangeCounter and not localRevIsMostRecent then

						logger:warnf( "resolving %s with remoteRev by changeCounter", localRev._id )
						recordConflictResolution( self, kConflictResolution_remoteRevByChangeCounter, localRev._id )

						-- Modify localRev to the content of remoteRev.
						local overlay = WFTableMerge.createOverlay( localRev, remoteRev )
						overlay._id = localRev._id

						-- We are using the remote rev, so use it's revId to avoid
						-- attempting to send changes back up to oz.
						overlay._newRevId = remoteRevId

						-- Delete the existing remoteRevId as we are adding a new one with the
						-- same revId and don't want to delete it.
						return docStore:deleteRevisionAsNonCurrentAndApplyOverlay( remoteRev._id, remoteRevId, overlay )
						:map( function()
							return kRemoteRevAlreadyDeleted
						end )

					else

						logger:warnf( "resolving %s with localRev by changeCounter", localRev._id )
						recordConflictResolution( self, kConflictResolution_localRevByChangeCounter, localRev._id )

						local overlay = preserveOzDataFromRemoteRevForLocalRev( remoteRev, localRevId )
						return docStore:applyOverlay( overlay )

					end

				end

				-- Can we reach this point?
				-- Give up and arbitrary solve conflict by deleting remoteRev in next step.

				logger:errorf( "Arbitrarily resolving conflict for doc ID %s", localRev._id )
				recordConflictResolution( self, kConflictResolution_errorArbitrarilyLocalRev, localRev._id )

				return Rx.Observable.just( nil )

			end

		end )

		:flatMapLatest( function( result )

			if result == kRemoteRevAlreadyDeleted then
				return Rx.Observable.just( nil )
			end

			return docStore:deleteRevisionAsNonCurrent( remoteRev._id, remoteRevId )

		end )
		
end

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

local function resolveOneConflict( self, docStore, docId )

	return docStore:currentRevisionsForDoc( docId, true )
		:tapOnError( function( message )
			logger:errorf( "docStore:currentRevisionsForDoc( %s ) failed with message %s",
							   docId, message )
		end )
		:flatMap( function( currentRevs )
				
				-- If there are more than two current revs, ignore third and successive
				-- revs. We'll catch those on a successive round. Without P2P sync, this
				-- scenario is highly unlikely but I can't completely rule it out.

				if type( currentRevs ) == 'table' and currentRevs[ 2 ] then
					return resolveConflictingRevs( self, docStore, currentRevs[ 1 ], currentRevs[ 2 ] )
				else
					return Rx.Observable.empty()
				end

		end )

end

--------------------------------------------------------------------------------
--[[!
	\class WFCatConflictResolver

	\brief This namespace handles resolving conflicts in the docStore. 
		
	A conflict is created in the docStore when two different revisions are created (via \c applyOverlay, 
	\c setDocValue or \c bulkSetDocValue) with the same parent. This most commonly happens when the local
	client makes an update (using \c applyOverlay) while a different change comes in from Oz for the same 
	asset and is applied with \c bulkSetDocValue. 
	
	After \c bulkSetDocValue the docStore will mark both revisions as 'current' and mark it as a conflict.
	
	Once marked as a conflict, the document will be indexed into the \c conflicts table in the index store
	and a query runs to take each of these conflicted documents and resolve the conflict.
	
	To resolve the conflict the conflict resolver takes the 'local' revision (this isn't guaranteed to be the 
	revision for the change that was made locally in practice, but it is 99+% of the time so it's easier to 
	refer to it as such) and the 'remote' revision and gets the revision history of each.
	
	The resolver then works backward through the revision history lists to find the most recent common ancestor
	revision - the base revision. If no base revision is found, that is ok, but the resolution is a little 
	clumsier.
	
	Reolve the conflict using the following rules:
	
	1. If either the local revision or the remote revision is a delete, then the delete wins. This prevents one
		client from attempting to resurrect a delete made by another.
	2. If the local and remote revisions match, then choose either revision. We choose the remote revision to 
		minimize communication with Oz.
	3. If there is a base revision, use \c WFTableMerge.threeWayMerge to merge the results. Any conflicting fields
		will have the values from the local revision.
	4. If there is no base revision
		* If the asset is a cloud trash related asset
			* If number of Oz revisions is one, and local revision is not on Oz, then we always choose local revision. 
				It's due to a scenario to preserve local changes due to modifications during pause sync 
			* use the \c userUpdated value to determine which change is more recent
				and let that one win
		* If the asset is not a trash asset, choose the revision with the higher changeCounter assuming that the
			most recent addition to the docStore is the most recent change
			
	There may be other edge cases to resolve so refer to the \c resolveConflictingRevs function for how any 
	additional cases are handled.
	
	Note: When resolving a conflict it is important that the resolution causes a new revision row to be written to the
	docStore with a higher sequence number. This ensures that the resolution causes downstream processes 
	(e.g. \c WFCatMetadataPusher) to notice a change has been made and react to the updated data. Not causing this 
	sequence number to be bumped can lead to missed updates.
]]--
local WFCatConflictResolver = {

	_conflictQuerySubscription = false,
		-- (Subscription) holds the conflicts query subscription
	
	_currentConflictResolutions = false,
		-- (table) holds list of observables indexed by docId that are in process of conflict resolution.

	_conflictResolutionSummary = false,
		-- (table) holds list of asset ids that were resolved indexed by kConflictResolution_*

	class = {
		generateIndexSchema = generateIndexSchema,
	},
	
	WFCloseSupport.addCloseSupportMethods,
}

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

function WFCatConflictResolver:__init( docStore, indexStore )

	check( 'WFCatConflictResolver', 'WFRxDocStore', 'WFRxIndexStore' )

	local query = indexStore:query {
				sql = [[SELECT docId, sequence FROM conflicts ORDER BY sequence ASC LIMIT 100]],
				calm = 1,
			}

	-- Initialize tables here to ensure unit tests get a clean table as they
	-- may create a new WFCatConflictResolver for each test.
	self._currentConflictResolutions = {}
	self._conflictResolutionSummary = {}

	self._conflictQuerySubscription = query:subscribe(
		function( docIdRows )
			if #docIdRows > 0 then
				logger:infof( "Conflict resolver identified %d matching documents", #docIdRows )
			end
			for _, docIdRow in ipairs( docIdRows ) do
				local docId = docIdRow[ 1 ]
				
				if not self._currentConflictResolutions[ docId ] then
						logger:infof( "Resolving conflict for %s", docId )
						
						local clearResolutionSubscription = function() 
							self._currentConflictResolutions[ docId ] = nil 
						end
						
						self._currentConflictResolutions[ docId ] = resolveOneConflict( self, docStore, docId )
																:subscribe( nil,
																			clearResolutionSubscription, 
																			clearResolutionSubscription )
							
				else
					logger:infof( "Skipping resolution for %s because one is in progress", docId )
				end
				
			end
		end,
		function( message )
			logger:errorf( "Conflict resolver query failed with message: %s", message )
			self._conflictQuerySubscription = nil
		end,
		function()
			logger:debugf( "Conflict resolver query notified as done" )
			self._conflictQuerySubscription = nil
		end )

	WFDiagnosticLog.addLogProvider( self )

end

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

function WFCatConflictResolver:close()
	
	if self._conflictQuerySubscription then
		self._conflictQuerySubscription:dispose()
		self._conflictQuerySubscription = nil
	end
end

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

function WFCatConflictResolver:sendDiagnosticLogData()

	WFDiagnosticLog.addSectionToLog
		{

			id = string.format( 'WFCatConflictResolver' ),

			name = string.format( "Conflict Resolver" ),
			weight = 511,
			parentSection = WFDiagnosticLog.Sections.SYNC,

			body = self._conflictResolutionSummary

		}

end

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

function WFCatConflictResolver:DEBUG_conflictsResolved()

	-- This function is used by unit tests to verify the correct number of
	-- conflicts are resolved. Improper setting/mocking of revision ids can lead
	-- to unexpected conflicts which may change the behavior of what is actually
	-- happening when running the app with Oz.

	return self._conflictResolutionSummary

end

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

return Rx.Class 'WFCatConflictResolver'( WFCatConflictResolver )
