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

WFLuaTestThreadRunner.lua

--------------------------------------------------------------------------------
ADOBE CONFIDENTIAL
------------------
Copyright 2023 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'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFCloseSupport = require 'WFCloseSupport'
local KSDate = require 'KSDate'

	-- WF.core.lua.threading
local WFWorkerThread = require 'WFWorkerThread'

	-- WF.web.json
local KSJsonDecoder = require "KSJsonDecoder"
local KSJsonEncoder = require "KSJsonEncoder"

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

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

local logger = WFLogger "WFLuaTestThreadRunner"
	:setLevel( 'testrun' )

local check = assert( WFArguments.check, "Cannot find WFArguments.check" )

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

local prioritySuiteNameKeys = {
	Trash = true,
	Uploader = true,
	Downloader = true,
}

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

local function getPrioritySuiteNames( testSuiteAndNames )

	local prioritySuiteNames = {}
	
	local function checkSuiteName( suiteName )
		for key in pairs( prioritySuiteNameKeys ) do
			if suiteName:find( key ) then
				prioritySuiteNames[ suiteName ] = true
			end
		end
	end
	
	for suiteName in pairs( testSuiteAndNames ) do
		checkSuiteName( suiteName )
	end
	
	return prioritySuiteNames
end

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

local function waitForWorkerThreadsToInitialize( self )

	return self._areWorkerThreadsInitialized
		:filter( function( value )
			return value
		end )
		:take( 1 )

end

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

local function getIdleWorkerId( self )

	return waitForWorkerThreadsToInitialize( self )
		:flatMap( function()
			return self._idleWorkerObservable
		end )

end

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

local WFLuaTestThreadRunner = {

	_testSuiteAndNames = false,
	
	_priorityTestSuiteNames = false,
	
	_maxConcurrentThread = false,
	
	_coverageDirPath = false,
	
	_testSuiteAndResults = false,
	
	_testSuiteResultLogIsReadyObservable = false,
	
	_logTestSuiteResultSubscription = false,
	
	_runTestsStartTime = false,
	
	_testRunnerWorkerThreads = false,
	
	_runTestsSubscription = false,
	
	_runTestsOnThreadSubscriptions = false,
	
	_idleWorkerObservable = false,
	
	_areWorkerThreadsInitialized = false,
	
	_testWorkerCompleteObservable = false,
	
	_testWorkerCompleteSubscription = false,
	
	WFCloseSupport.addCloseSupportMethods,
	
}

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

function WFLuaTestThreadRunner:__init( testSuiteAndNames, maxConcurrentThread, coverageDirPath )

	check( 'WFLuaTestThreadRunner', 'table', '?number', '?string' )

	self._testSuiteAndNames = testSuiteAndNames
	self._priorityTestSuiteNames = getPrioritySuiteNames( testSuiteAndNames )
	self._maxConcurrentThread = maxConcurrentThread or 1
	self._coverageDirPath = coverageDirPath
	
	self._areWorkerThreadsInitialized = Rx.ReplaySubject( 1 )
	self._areWorkerThreadsInitialized:onNext( false )
	
	-- init test runner worker threads
	
	self:initWorkerThreads()

	-- init _idleWorkerObservable for all workerId
	
	self._idleWorkerObservable = Rx.ReplaySubject( self._maxConcurrentThread )
	
	for i = 1, self._maxConcurrentThread do
		self._idleWorkerObservable:onNext( i )
	end
	
	-- setup test suite result logging
	
	self._testSuiteResultLogIsReadyObservable = Rx.ReplaySubject( 1 )
	
	self:logTestSuiteResult()
	
	-- setup run test complete notify
	
	self._testWorkerCompleteObservable = Rx.ReplaySubject( 1 )
	
	self:setUpNotifyWhenTestsComplete()
	
	-- run tests
	
	self._testSuiteAndResults = {}
	self._runTestsStartTime = KSDate.timeNow()

	self:runTests()

end

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

function WFLuaTestThreadRunner:initWorkerThreads()

	self._testRunnerWorkerThreads = {}
	local initializedThreads = 0
	
	for i = 1, self._maxConcurrentThread do
	
		local workerThread = WFWorkerThread( 'WFLuaTestThreadRunnerImp.lua', tostring( i ) )

		local workerThreadInitRequest = workerThread:request( 'initTestThreadRunner' )
			:subscribe( function()

				table.insert( self._testRunnerWorkerThreads, workerThread )

				-- Lazily assuming that if the last worker thread is initialized they all are.
				-- It's not perfect here, but it's close enough to get started passing out
				-- query requests
				initializedThreads = initializedThreads + 1
				if initializedThreads == self._maxConcurrentThread then
					self._areWorkerThreadsInitialized:onNext( true )
				end

			end )

		self:addCloseAction( function()
			workerThreadInitRequest:dispose()
		end )
		
	end

end

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

function WFLuaTestThreadRunner:setUpNotifyWhenTestsComplete()

	local completedThreadIds = {}
	local completedThreads = 0
	
	self._testWorkerCompleteSubscription = self._testWorkerCompleteObservable
		
		:filter( function( completeThreadId )
			
			if not completedThreadIds[ completeThreadId ] then
				completedThreads = completedThreads + 1
			end
			
			completedThreadIds[ completeThreadId ] = true
			
			logger:testrunf( "completedThreads: %d", completedThreads )
			
			return completedThreads == self._maxConcurrentThread
			
		end )

		
		:subscribe( function()
		
			local runTestsDuration = KSDate.timeNow() - self._runTestsStartTime
			
			local totalSuccessCount = 0
			local totalFailureCount = 0
			local totalExecutionTime = 0
			
			local failureTests = {}
			
			for testSuiteName, testSuiteResultInfo in pairs( self._testSuiteAndResults ) do
			
				totalSuccessCount = totalSuccessCount + testSuiteResultInfo.successCount or 0
				totalFailureCount = totalFailureCount + testSuiteResultInfo.failureCount or 0
				totalExecutionTime = totalExecutionTime + testSuiteResultInfo.executionTime or 0
				
				if testSuiteResultInfo.failureCount > 0 then
					
					for testName, testResultInfo in pairs( testSuiteResultInfo.testResults ) do
					
						if testResultInfo.failure then
							
							local testFullName = testSuiteName .. '.' .. testName
							failureTests[ #failureTests + 1 ] = {
								testName = testFullName,
								failureMessage = testResultInfo.failureMessage
							}
							
						end
					
					end
					
				end
			
			end
			
			local failureTestsJsonStr
			if( next( failureTests ) ) then
				failureTestsJsonStr = KSJsonEncoder.prettyPrint( failureTests, { pretty = true } )
			end
			
			logger:testrunf( "Tests finished. Took %.4fs (%.4fs in total test time). %d/%d (%d failures)",
							runTestsDuration, totalExecutionTime, totalSuccessCount, totalSuccessCount + totalFailureCount, totalFailureCount )
			
			require 'WFLuaTestMonitor'.notifyRunTestsComplete( totalSuccessCount, totalFailureCount, totalExecutionTime, failureTestsJsonStr,
																		( self._maxConcurrentThread > 1 ) and runTestsDuration or nil )
		end )

end

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

function WFLuaTestThreadRunner:runTests()

	self._runTestsOnThreadSubscriptions = {}
	
	self._runTestsSubscription = getIdleWorkerId( self )
		
		:subscribe( function( threadId )
		
			local testSuiteName = next( self._priorityTestSuiteNames ) or next( self._testSuiteAndNames )
			
			if testSuiteName then
			
				local testNames = self._testSuiteAndNames[ testSuiteName ]
				
				self._priorityTestSuiteNames[ testSuiteName ] = nil
				self._testSuiteAndNames[ testSuiteName ] = nil
				
				if next( testNames ) then
				
					local workerThread = self._testRunnerWorkerThreads[ threadId ]
					
					self._runTestsOnThreadSubscriptions[ threadId ] = workerThread
						:request( "runTestsInSuite", testSuiteName, testNames, self._coverageDirPath )
						:subscribe( function( successCount, failureCount, executionTime, testResults )
							
							local testSuiteResultInfo = {
								successCount = successCount,
								failureCount = failureCount,
								executionTime = executionTime,
								testResults = testResults
							}
							
							self._testSuiteAndResults[ testSuiteName ] = testSuiteResultInfo
							
							self._testSuiteResultLogIsReadyObservable:onNext( testSuiteName )
							
							require 'WFLuaTestMonitor'.notifyTestSuiteComplete( testSuiteName, successCount, failureCount, executionTime )
							
							local threadSubscription
							
							if self._runTestsOnThreadSubscriptions[ threadId ] then
								threadSubscription = self._runTestsOnThreadSubscriptions[ threadId ]
								self._runTestsOnThreadSubscriptions[ threadId ] = nil
							end
							
							self._idleWorkerObservable:onNext( threadId )
							
							if threadSubscription then
								threadSubscription:dispose()
							end
							
						end )
				
				else
					
					self._idleWorkerObservable:onNext( threadId )
					
				end
			
			else
			
				self._testWorkerCompleteObservable:onNext( threadId )
				
			end
			
		end )

end

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

function WFLuaTestThreadRunner:logTestSuiteResult()
	
	self._logTestSuiteResultSubscription = self._testSuiteResultLogIsReadyObservable
	
		:filter( function( testSuiteName )
			return testSuiteName
		end )
		
		:subscribe( function( testSuiteName )
		
			local testSuiteResult = self._testSuiteAndResults[ testSuiteName ]
			
			for testName, testResult in pairs( testSuiteResult.testResults ) do
				
				logger:testrunf( string.format( "Starting %s/%s", testSuiteName, testName ) )
				
				if testResult.logMessage and testResult.logMessage ~= "" then
					logger:testrunf( "%s", testResult.logMessage )
				else
					logger:testrunf( " " )
				end
				
				logger:testrunf( "Test log end" )
				
				if testResult.success then
				
					logger:testrunf( string.format( "OK (%.4fs)", testResult.executionTime ) )
				
				else
				
					if testResult.failureMessage then
						logger:testrunf( string.format( "%s:%s::%s", testSuiteName, testName, testResult.failureMessage ) )
					end
					
					logger:testrunf( string.format( "FAIL (%.4fs)", testResult.executionTime ) )
					
				end
				
			end

		end )
	
end

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

function WFLuaTestThreadRunner:close()

	for i = 1, self._maxConcurrentThread do
	
		local workerThread = self._testRunnerWorkerThreads[ i ]
		self._testRunnerWorkerThreads[ i ] = nil
		
		if workerThread then
			workerThread:close()
		end
		
		if self._runTestsOnThreadSubscriptions[ i ] then
			self._runTestsOnThreadSubscriptions[ i ]:dispose()
			self._runTestsOnThreadSubscriptions[ i ] = nil
		end
	
	end
	
	if self._runTestsSubscription then
		self._runTestsSubscription:dispose()
		self._runTestsSubscription = nil
	end

	if self._idleWorkerObservable then
		self._idleWorkerObservable:dispose()
		self._idleWorkerObservable = nil
	end


	if self._areWorkerThreadsInitialized then
		self._areWorkerThreadsInitialized:dispose()
		self._areWorkerThreadsInitialized = nil
	end
	
	if self._testWorkerCompleteObservable then
		self._testWorkerCompleteObservable:dispose()
		self._testWorkerCompleteObservable = nil
	end
	
	if self._testWorkerCompleteSubscription then
		self._testWorkerCompleteSubscription:dispose()
		self._testWorkerCompleteSubscription = nil
	end
	
	if self._testSuiteResultLogIsReadyObservable then
		self._testSuiteResultLogIsReadyObservable:dispose()
		self._testSuiteResultLogIsReadyObservable = nil
	end
	
	if self._logTestSuiteResultSubscription then
		self._logTestSuiteResultSubscription:dispose()
		self._logTestSuiteResultSubscription = nil
	end
	
end

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

WFLuaTestThreadRunner = Rx.Class 'WFLuaTestThreadRunner'( WFLuaTestThreadRunner )

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

local initMessageProcessor = function( testSuiteAndNamesJsonStr )
	
	local testSuiteAndNames = KSJsonDecoder.decodeWebSafeJson_stripNull_ignoreEmpty( testSuiteAndNamesJsonStr )
	
	local maxConcurrentThread = testSuiteAndNames.maxConcurrentThread or 1
	testSuiteAndNames.maxConcurrentThread = nil
	
	local coverageDirPath
	if testSuiteAndNames.coverageDir and  testSuiteAndNames.coverageDir ~= "" then
		coverageDirPath = testSuiteAndNames.coverageDir
	end
	testSuiteAndNames.coverageDir = nil
	
	local testThreadRunner = WFLuaTestThreadRunner( testSuiteAndNames, maxConcurrentThread, coverageDirPath )
	
	WFCloseSupport.mustCloseAtThreadTermination( testThreadRunner )
	
end

return {
	initMessageProcessor = initMessageProcessor
}
