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

NmbSessionLogger.lua

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

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

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


	-- wf.core.os
local KSPathUtils = require 'KSPathUtils'
local WFFileUtils = require 'WFFileUtils'
local WFPlatformInfo = require 'WFPlatformInfo'

	-- wf.sqlite
local WFSQLiteDatabase = require "WFSQLiteDatabase"
local WFSQLiteDatabaseSchema = require "WFSQLiteDatabaseSchema"

	-- wf.core.utils
local WFFunctionUtils = require 'WFFunctionUtils'

	-- Rx
local Rx = require "Rx"

local WFFunctionUtils_reduce = 
		assert( WFFunctionUtils.reduce,
				"WFFunctionUtils.reduce does not exist" )

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

local logger = require 'WFLogger'( "NmbSessionLogger" )

--==============================================================================
-- @mark db log support
--==============================================================================

local function safeCloseDb( db )
	
	local ok, maybeErr = pcall( function() db:close() end )

	if not ok then

		logger:error( maybeErr )

	end

end

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

local function deleteDbAtPath( path )

	if type( path ) ~= "string" then
		
		return
		
	end

	WFFileUtils.deleteFile( path )

	WFFileUtils.deleteFile( path .. '-shm' )

	WFFileUtils.deleteFile( path .. '-wal' )

end

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

local function handleFatalDbErrorForSessionLogDbAtPath( 
				errType, dbPath, errStr )

	if errType ~= "corrupt" then

		logger:warnf( "Non corrupt error of type '%s' for session logger at path '%s': %s",
						errType, dbPath, errStr )
		
		return

	end

	logger:warnf( "Corrupt db error of type '%s' for session logger at path '%s': %s",
					errType, dbPath, errStr )

	deleteDbAtPath( dbPath )

end

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

local function openDbAtPath( path )

	local dummySchema = WFSQLiteDatabaseSchema( "1.01" )

	local db = WFSQLiteDatabase( path, dummySchema )

	local dbImp = db._db

	if dbImp and pcall( function() return dbImp.setFatalErrorHook end ) then

		pcall( 
			function() 

				dbImp:setFatalErrorHook( handleFatalDbErrorForSessionLogDbAtPath )
				
			end )

	end

	return db

end

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

local function execSQL( db, sql, ... )

	db:execOnce( sql, ... )

	return db

end

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

local function maybeOpenOrCreateDbAtPath( path )

	local db = openDbAtPath( path )

	if not db then

		return

	end

	local schemaSQL = 
		{
			[[

				CREATE TABLE IF NOT EXISTS sessionIds (
					sessionId PRIMARY KEY
				);

			]],
			[[ 

				CREATE TABLE IF NOT EXISTS sessionEvents (
					processId NOT NULL,
					sessionId NOT NULL,
					eventType NOT NULL,
					event NOT NULL,
					eventTimeStamp NOT NULL
				);

			]],

			[[ 

				CREATE INDEX IF NOT EXISTS sessionEventsByTime 
					ON sessionEvents (
						eventTimeStamp, processId, sessionId, eventType, event );

			]],

			[[ 

				CREATE INDEX IF NOT EXISTS sessionEventsBySession
					ON sessionEvents (
						sessionId, processId, eventTimeStamp, eventType, event );

			]],

		}

	local ok, maybeErr = pcall( WFFunctionUtils_reduce, db, execSQL, schemaSQL )

	if not ok then

		safeCloseDb( db )

		logger:errorf( "failed to init session log db: %s", maybeErr )

		return

	end

	return db

end

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

local function pack( ... )

	return select( '#', ... ), { ... }

end

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

local function runWithMaybeDbAtPath( path, f, ... )

	local maybeDb = maybeOpenOrCreateDbAtPath( path )

	local nRet, ret = pack( pcall( f, maybeDb, ... ) )
	
	if maybeDb then

		safeCloseDb( maybeDb )

	end

	if not ret[ 1 ] then

		error( ret[ 2 ] )

	end

	return unpack( ret, 2, nRet )
	
end

--==============================================================================
-- @mark diagnostic log support
--==============================================================================

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

local function maybeRunWithAttachedDbAtPath( 
				db, attachedDbName, attachedDbPath, f, ... )

	local ok, maybeErr = 
			pcall( 
				function() 
					
					db:execOnce( "ATTACH DATABASE ? AS ?",
									attachedDbPath, attachedDbName )

				end )

	if not ok then

		logger:error( maybeErr )

		return

	end

	ok, maybeErr = pcall( f, db, attachedDbName, ... )

	if not ok then

		logger:error( maybeErr )

	end

	ok, maybeErr = 
		pcall( 
			function() 
				
				db:execOnce( "DETACH DATABASE ?", attachedDbName )

			end )

	if not ok then

		logger:error( maybeErr )

	end

end

--------------------------------------------------------------------------------
				
local function copyAllSessionEventsRelativeToTimestamp( 
					destDb, srcDbName, relativeDirection, timestamp )

	local sql =

		relativeDirection == "since" and

			string.format( 
			[[INSERT INTO sessionEvents
				SELECT processId, ? || ';' || sessionId, eventType, event, eventTimeStamp
				FROM %s.sessionEvents
				WHERE eventTimeStamp >= ?]],
				srcDbName ) or

			string.format( 
			[[INSERT INTO sessionEvents
				SELECT processId, ? || ';' || sessionId, eventType, event, eventTimeStamp
				FROM %s.sessionEvents
				WHERE eventTimeStamp < ?]],
				srcDbName )

	destDb:execOnce( sql, srcDbName, timestamp )

end

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

local function mergeInDbInfo( mergedDb, dbInfoToMerge, relativeDirection, 
								timestamp )

	local name, path = unpack( dbInfoToMerge )

	maybeRunWithAttachedDbAtPath(
		mergedDb, name, path,
			copyAllSessionEventsRelativeToTimestamp,
				relativeDirection, timestamp )

	return mergedDb

end

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

local function appendResultsSplittingSessionId( results, eventTimeStamp, 
													elapsed, processId,
													sessionIdPlusLogName,
													eventType, event )

	local _, _, where, sessionId = 
		string.find( tostring( sessionIdPlusLogName ), "([^;]*);([^;]*)" ) 

	where = where or "unknown log"

	sessionId = sessionId or "unknown session"

	results[ #results + 1 ] = 
		{ eventTimeStamp, elapsed, processId, where, tonumber( sessionId ), 
			eventType, event }

	return results

end

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

local function getMergedSessionLogRowDataImp( 
				maybeMergedDb, relativeDirection, timestamp, dbInfosToMerge )

	if not maybeMergedDb then

		return {}

	end

	local mergedDb = maybeMergedDb

	mergedDb = WFFunctionUtils_reduce( mergedDb, 
					mergeInDbInfo, 
						dbInfosToMerge, relativeDirection, timestamp )

	return mergedDb
			:reduceSQLQueryViaValueArrays( {},
				appendResultsSplittingSessionId,
					[[SELECT eventTimeStamp, strftime( "%s", eventTimeStamp ) - strftime( "%s", ? ), processId, sessionId, eventType, event 
					  FROM sessionEvents
					  ORDER BY eventTimeStamp]],
					timestamp )

end

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

local function getMergedSessionLogRowDataRelativeToTimestamp( 
				relativeDirection, timestamp, log1Name, log1Path, log2Name,
				log2Path )

	return runWithMaybeDbAtPath( ':memory:', 
				getMergedSessionLogRowDataImp,
					relativeDirection, timestamp, 
					{ { log1Name, log1Path }, 
					  { log2Name, log2Path } } )

end

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

local function getMergedSessionLogRowDataSinceTimestamp( 
				timestamp, log1Name, log1Path, log2Name, log2Path )

	return getMergedSessionLogRowDataRelativeToTimestamp( 
				"since", timestamp, log1Name, log1Path, log2Name,
				log2Path )

end

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

local function getMergedSessionLogRowDataBeforeTimestamp( 
				timestamp, log1Name, log1Path, log2Name, log2Path )

	return getMergedSessionLogRowDataRelativeToTimestamp( 
				"strictlyBefore", timestamp, log1Name, log1Path, log2Name,
				log2Path )

end

--------------------------------------------------------------------------------
-- Register a hook that is called when a diagnostic log is generated
--
-- This hook adds the last _maxDiagLogSessionCount session entries to the
-- diagnostic log
--------------------------------------------------------------------------------

--==============================================================================
-- @mark SQLite db support for the session log
--==============================================================================

local function maybeGetScalar( db, sql, ... )

	return db:gatherArray( sql, ... )[ 1 ]

end

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

local function getSessionCount( db )

	return maybeGetScalar( db, "SELECT COUNT( * ) FROM sessionIds" )

end

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

local _50Mb = 50 * 1024 * 1024

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

local function maybeVacuum( db )

	local pageCount = tonumber( maybeGetScalar( db, "PRAGMA freelist_count" ) )

	if not pageCount then

		return

	end

	local pageSize = tonumber( maybeGetScalar( db, "PRAGMA page_size" ) )

	if not pageSize then

		return

	end

	if pageCount * pageSize < _50Mb then

		return

	end

	local freeListCount = 
			tonumber( maybeGetScalar( db, "PRAGMA freelist_count" ) )

	if not freeListCount then

		return

	end

	if freeListCount / pageCount < 0.5 then

		return

	end

	db:exec( "VACUUM" )

end

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

local function maybeTrimDaemonSessionLogDb( db, maxPrevSessions )

	maybeVacuum( db )

	local prevSessionCount = getSessionCount( db )

	if prevSessionCount > maxPrevSessions then

		local staleSessions = 
				db:gatherArray( 
					[[SELECT sessionId 
						FROM sessionIds 
						ORDER BY sessionId ASC LIMIT ?]],
						prevSessionCount - maxPrevSessions )

		if #staleSessions > 0 then

			local newestStaleSessionId = staleSessions[ #staleSessions ]

			execSQL( db, "DELETE FROM sessionIds WHERE sessionId <= ?", 
							newestStaleSessionId )

			execSQL( db, "DELETE FROM sessionEvents WHERE sessionId <= ?", 
							newestStaleSessionId )

		end

	end

end

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

local function bootstrapSessionLogDbForWriteImp( path, maxPrevSessions )

	local db = maybeOpenOrCreateDbAtPath( path )

	if not db then

		error( string.format( "failed to open or create db at path: %s", 
								path ) )

	end

	maybeTrimDaemonSessionLogDb( db, maxPrevSessions )

	local prevSessionId =
			maybeGetScalar( 
				db,
				[[SELECT IFNULL( MAX( sessionId ), 0 ) FROM sessionIds]] )

	local sessionId

	local i = 1

	local maxSessionTries = 16

	-- gaurd against race condition of two daemon launching at same time and
	-- requesting same session id. This is very unlikely, but trying
	-- to make this code extra resilient.
	while not sessionId and i <= maxSessionTries do

		local ok, maybeErr = 
			pcall( execSQL, db, "INSERT INTO sessionIds SELECT ?", prevSessionId + i )

		if not ok then

			logger:errorf( "Failed to set sessionId to %s because: %s",
							prevSessionId + i, maybeErr )

		end

		sessionId = ok and prevSessionId + i

		i = i + 1

	end

	return db, sessionId

end

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

local function bootstrapSessionLogDbForWrite( path, maxPrevSessions)

	local ok, dbOrErrStr, sessionId = 
			pcall( bootstrapSessionLogDbForWriteImp, 
					path, maxPrevSessions )

	if not ok then

		logger:error( dbOrErrStr )

		deleteDbAtPath( path )

		ok, dbOrErrStr, sessionId = 
			pcall( bootstrapSessionLogDbForWriteImp, 
					path )

	end

	if not ok then
		
		logger:errorf( "failed twice to open daemon session log: %s", dbOrErrStr )

		return nil, dbOrErrStr

	end

	local db = dbOrErrStr
	
	if not sessionId then

		-- couldn't establish a session, bail
		if db then

			safeCloseDb( db )

		end

		return nil, "no session id"

	end

	return db, sessionId

end

--==============================================================================
-- @mark write to db
--==============================================================================

local function maybeFormatStr( ... )

	if select( '#', ... ) == 1 then

		return ...

	end

	local ok, str = pcall( string.format, ... )

	if not ok then

		return

	end
	
	return str

end

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

local function writeToSessionLogDb( db, sessionId, level, str )

	local ok, maybeErr = 
		pcall( execSQL,
					db, 
					[[INSERT INTO sessionEvents SELECT ?, ?, ?, ?, strftime( '%Y-%m-%dT%H:%M:%f', 'now' )]],
						WFPlatformInfo.getProcessID(), sessionId, level, str )

	if not ok then

		logger:errorf( 
			"failed to write to daemon session log db: %s", maybeErr )

	end

end

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

local function maybeWriteToSessionLogDb( db, sessionId, level, 
											maybeFmtStr, ... )

	if type( level ) ~= "string" then

		return 

	end

	local str = maybeFormatStr( maybeFmtStr, ... )

	if type( str ) ~= "string" then

		return 

	end

	writeToSessionLogDb( db, sessionId, level, str )

end

--==============================================================================
-- @mark public API
--==============================================================================

local function logEventStreamToPathStream( eventStream, path, maxSessionCount )

	maxSessionCount = tonumber( maxSessionCount ) or 15

	return Rx.Observable.create(
			function( observer )

				-- open db when subscription starts
				local db, sessionId = 
						bootstrapSessionLogDbForWrite( path, maxSessionCount )

				if not db then

					-- send an eror if that fails
					observer:onError( sessionId )

					return

				end

				local timestamp = 
						db:gatherArray( [[SELECT strftime( '%Y-%m-%dT%H:%M:%f', 'now' )]] )[ 1 ]

				-- immediately send timestamp of when session started, this
				-- is only data we send
				observer:onNext( timestamp )

				-- subscribe to events and write them to the db as they occur
				local eventStreamSubscription =
					eventStream
						:subscribe(

							function( level, maybeFmtStr, ... )

								maybeWriteToSessionLogDb( db, sessionId, level,
															maybeFmtStr, ... )

							end )

				-- define the subscription:dispose() method
				return Rx.Disposable.create(

					function()

						-- stop listening to events
						eventStreamSubscription:dispose()

						safeCloseDb( db )

					end )
			end )

end

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

local function getSessionLogDbPathWithParentFolder( parentFolder )

	return KSPathUtils.child( parentFolder, "NmbSessionLogger.db" )

end

--==============================================================================
-- @mark exports section
--==============================================================================

--------------------------------------------------------------------------------
--
-- NmbSessionLogger
-- 
-- This namespace allows you to serialize an Rx.Observable event stream to
-- disk via SQLite database. Multiple serialized event streams can later be
-- merged sorted by timestamps.
-- 
-- The main use case is to allow LrCC desktop to stream IPC related messages
-- on the client and the daemon to disk. Later, given the path to each of
-- the serialized dbs, the logs can be combined into a single log with the
-- messages interleaved by timestamps.
-- 

return {

	deleteDbAtPath = deleteDbAtPath,

	-- the getMergedSessionLogRowData*Timestamp API merge the rows in
	-- two different session logs by timestamp.
	-- 
	-- In practice, these are used to generate diagnostic log sections.
	getMergedSessionLogRowDataSinceTimestamp = 
		getMergedSessionLogRowDataSinceTimestamp,

	getMergedSessionLogRowDataBeforeTimestamp = 
		getMergedSessionLogRowDataBeforeTimestamp,

	-- 
	-- Given an Rx.Observable that emits
	-- 
	-- 			levelStr, logStringWithMaybeStringFormatArgs, ...
	-- 			
	-- wher Message format is
	-- 		
	-- 		levelStr 
	-- 		
	-- 			is "info", "warn", or "error"
	-- 			
	-- 		logStringWithMaybeStringFormatArgs
	-- 			
	-- 			is either a plain string or a string.format style string
	-- 			
	-- 		...
	-- 		
	-- 			if there are additional arguments they are assumed they are values
	-- 			for placeholder arguments in logStringWithMaybeStringFormatArgs
	--  
	-- The messages are persisted to a file on disk via SQLite database.
	-- 
	-- The messages can later be harvested via the
	-- getMergedSessionLogRowDataSinceTimestamp and
	-- getMergedSessionLogRowDataBeforeTimestamp APIs
	-- 
	logEventStreamToPathStream = logEventStreamToPathStream,

	getSessionLogDbPathWithParentFolder = getSessionLogDbPathWithParentFolder,

}
