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

NmbDbCorruptionUtils.lua

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

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

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

local KSPathUtils = require "KSPathUtils"
local KSTableUtils = require "KSTableUtils"
local WFFunctionUtils = require "WFFunctionUtils"
local WFFileUtils = require "WFFileUtils"

local WFSQLiteCorruptionLogger = require 'WFSQLiteCorruptionLogger'

local Rx = require "Rx"

local NmbCatalog = require 'NmbCatalog'

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

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

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

local function append( t, v )

	t[ #t + 1 ] = v 

	return t

end

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

local function appendAll( ... )

	assert( select( '#', ... ) > 1 , "appendAll requires multiple array" )

	local ret = KSTableUtils.shallowcopy( select( 1, ... ) )

	for i = 2, select( '#', ... ) do

		ret = WFFunctionUtils.reduce( ret, append, select( i, ... ) )

	end

	return ret

end

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

local function corruptionEventOnce()

	return WFSQLiteCorruptionLogger.getCorruptionEventStream()
			:take( 1 )

end

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

local function lcPatternMatch( s, patt )

	return string.find( string.lower( s ), string.lower( patt ) ) ~= nil

end

--==============================================================================
-- @mark functions that extract db object from wfCatalogStream
--==============================================================================

local function getIndexStoreFromContainerObj( dbContainerObj )

	return dbContainerObj:indexStore()

end

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

local function getMRUDbFromWFCatalogObj( wfCatalog )

	return NmbCatalog:isClassOf( wfCatalog ) and wfCatalog:deletionBacklog()

end

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

local function getDocStoreFromContainerObj( dbObj )

	return dbObj:docStore()

end

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

local function getNotificationsDb( notifDbContainer )

	-- adjust the API to match what we expect in this namespace
	return {

		haltAndDelete = 
			function()
				
				return notifDbContainer:haltAndDeleteNotificationDb()

			end,

		_DEBUG_simulateCorruption = 
			function()
				
				return notifDbContainer:_DEBUG_simulateCorruptionNotificationDb()

			end,

	}

end

--==============================================================================
-- @mark corruption handlers
--==============================================================================

local function tryExplicitSQLiteFileDeletionResultStream( dbName, path )

	if type( path) ~= "string" then

		return Rx.Observable.throw( 
				string.format( 
					"Failed to explicitly delete db %s because no path",
					dbName ) )

	end

	if string.find( path, '^:memory:' ) then

		return Rx.Observable.empty()

	end

	local ok, maybeErr =
		pcall(
			function()

				WFFileUtils.deleteFile( path )

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

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

			end )

	if not ok then

		return Rx.Observable.throw( 
				string.format( 
					"Failed to explicitly delete db %s at %s because: %s",
					dbName, path, maybeErr ) )

	end

	return Rx.Observable.empty()

end

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

local function makeHaltAndDeleteResultStream( dbName, path, dbObj )

	return dbObj
			:haltAndDelete()
			:catch( 
				function( errStr )

					logger:warnf( "Error with :haltAndDelete() on %s: %s",
									dbName, errStr )

					return tryExplicitSQLiteFileDeletionResultStream( 
								dbName, path )

				end )
			:tap(
				function()

					logger:warnf( "Successfully deleted corrupt %s db", dbName )

				end )

end

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

local function raiseUnrecoverableFatalError( dbName )
	
	-- raise exception that will be caught and sent to client as a boolean
	-- that indicates whether custerm intervention is needed
	return Rx.Observable.throw(
			string.format( 
				"Fatal corruption requiring customer intervention seen on %s db", 
				dbName ) )

end

--==============================================================================
-- @mark DB corruption manifests
--==============================================================================

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

--  as we add new SQLite databases that are derived from wfCatalog, add them
--  to the list below to enable corruption recovery
--  
--  sample entry:
--  
--	{ dbName, dbFileNamePattern, 
--		extractDbObjFromContainerFunc, corruptionCleanupResultStreamFunc }

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

local _wfCatalogCorruptionSupportManifest = {

	{ "index store", "Managed Catalog.wfindex$",
		getIndexStoreFromContainerObj, makeHaltAndDeleteResultStream },

	{ "mru db", "Managed Catalog.mru$",
		getMRUDbFromWFCatalogObj, makeHaltAndDeleteResultStream },

	-- we don't delete the docstore
	{ "docstore", "Managed Catalog.mcat$",
		getDocStoreFromContainerObj, raiseUnrecoverableFatalError },

}

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


local _notifDbCorruptionSupportManifest = {

	{ "notifications db", "Managed Catalog.notifdb$",
		getNotificationsDb, makeHaltAndDeleteResultStream },

}

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

local _uuidRegExPattern = string.rep( "[A-Fa-f0-9]", 32 )

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

local _profileCatalogCorruptionSupportManifest = {

	{ "profiles docstore", "profiles/" .. _uuidRegExPattern .. "[.]mcat$",
		getDocStoreFromContainerObj, makeHaltAndDeleteResultStream },

	{ "profiles index store", "profiles/" .. _uuidRegExPattern .. "[.]wfindex$",
		getIndexStoreFromContainerObj, makeHaltAndDeleteResultStream },

}

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

local function addDbNameFromManifestEntry( dbNames, manifestEntry )

	local dbName = manifestEntry[ 1 ]

	return append( dbNames, dbName )

end

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

local _allRootModelDbNames = 
		WFFunctionUtils.reduce( {}, 
			addDbNameFromManifestEntry,
			appendAll( _wfCatalogCorruptionSupportManifest, 
						_profileCatalogCorruptionSupportManifest, 
						_notifDbCorruptionSupportManifest ) )

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

local function getUserInfoCacheDb( wfCatUser )

	return wfCatUser:userInfoCache()

end

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

local _userInfoCorruptionSupportManifest = {

	{ "userInfoCache", "oz[-]user[-]" .. _uuidRegExPattern,
		getUserInfoCacheDb, makeHaltAndDeleteResultStream },

}

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

local function addFilePatternFromManifestEntry( patterns, manifestEntry )

	local filePattern = manifestEntry[ 2 ]

	return append( patterns, filePattern )

end

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

local _allFilePatterns = 
		WFFunctionUtils.reduce( {}, 
			addFilePatternFromManifestEntry,
			appendAll( _wfCatalogCorruptionSupportManifest, 
						_profileCatalogCorruptionSupportManifest, 
						_notifDbCorruptionSupportManifest, 
						_userInfoCorruptionSupportManifest ) )

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

local function maybeGetDbObj( dbName, extractDbObjFromContainerFunc,
								containerObj )

	local ok, dbObj = pcall( extractDbObjFromContainerFunc, containerObj )

	if not ok or not dbObj then

		logger:warnf( "Bad return value when extracting dbObj for %s: %s", 
						dbName, dbObj )

		return

	end

	return dbObj

end

--==============================================================================
-- @mark simulate DB corruption support
--==============================================================================

local function maybeGetDbObjFromDbName( wantedDbObj, dbCorruptManifestEntry,
											containerObj, wantedDbName )

	if wantedDbObj then

		return wantedDbObj

	end

	local dbName, _, extractDbObjFromContainerFunc = 
			unpack( dbCorruptManifestEntry )

	if dbName ~= wantedDbName then

		return

	end

	return maybeGetDbObj( dbName, extractDbObjFromContainerFunc, containerObj )

end

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

local function maybeGetDbObjForDbNameFromManifest(
					dbCorruptionSupportManifest, containerObj, dbName )

	return WFFunctionUtils.reduce(
				nil,
				maybeGetDbObjFromDbName,
					dbCorruptionSupportManifest, containerObj, dbName )

end

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

local function makeSimulateCorruptionForDbResultStream( 
				wfCatalog, profileCatalog, notifDbContainer, dbName )

	local dbObj =
			maybeGetDbObjForDbNameFromManifest(
					_wfCatalogCorruptionSupportManifest, wfCatalog, 
					dbName ) or
			maybeGetDbObjForDbNameFromManifest(
					_profileCatalogCorruptionSupportManifest, profileCatalog,
					dbName ) or
			maybeGetDbObjForDbNameFromManifest(
					_notifDbCorruptionSupportManifest, notifDbContainer,
					dbName )

	if not dbObj then

		return Rx.Observable.throw( 
				string.format( "don't know how to corrupt %s", dbName ) )

	end

	return dbObj
			:_DEBUG_simulateCorruption()
			:catch( 
				function()

					-- swallow expected exception
					return Rx.Observable.empty()

				end )

end

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

local function makeSimulateCorruptUserInfoCacheStream( wfCatUser )

	return wfCatUser
			:userInfoCache()
			:_DEBUG_simulateCorruption()
			:catch( 
				function()

					-- swallow expected exception
					return Rx.Observable.empty()

				end )

end

--==============================================================================
-- @mark DB corruption recovery result stream
--==============================================================================

--------------------------------------------------------------------------------
-- makeCleanupCorruptDbResultStream is defined so that we always emit 4 values
--------------------------------------------------------------------------------

local function makeCleanupCorruptDbResultStream(
				errorType, canBeSolvedByRestart, dbName, descOrErrStr )

	if type( canBeSolvedByRestart ) ~= "boolean" then

		-- make sure canBeSolvedByRestart is either true or false
		canBeSolvedByRestart = not not canBeSolvedByRestart

	end

	return Rx.Observable.just( 
			tostring( errorType ), 
			canBeSolvedByRestart, 
			tostring( dbName ), 
			tostring( descOrErrStr ) )

end

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

local function doCleanupCorruptDbResultStream( 
				dbName, path, dbObj, corruptionCleanupResultStreamFunc, 
				errorType )

	if errorType ~= "corrupt" then

		-- 1) this is non-corrupt but fatal error. do nothing, just relay
		--    success result
		local canBeSolvedByRestart = true

		return makeCleanupCorruptDbResultStream( 
				errorType, canBeSolvedByRestart, dbName, "transient IO err" )

	end
	
	-- 2) try to do something about the corruption
	
	local corruptionCleanupRawResultStream =
				corruptionCleanupResultStreamFunc( dbName, path, dbObj )

	-- 3) even though the corruptionCleanupResultStreamFunc isn't supposed to
	--    emit data, we use takeLast( 0 ) to make 
	--    corruptionCleanupCompleteOrThrowStream either completes without
	--    emitting data or throws
	local corruptionCleanupCompleteOrThrowStream =
			corruptionCleanupRawResultStream
				:takeLast( 0 )

	local restartWillSolveDbProblemResultStream = 
			makeCleanupCorruptDbResultStream( 
				errorType, true, dbName, "deleted corrupt db" )

	-- 4) if no errors were thrown, the db has been shutdown and
	--    deleted. Rx.Observable.concat ensures we return success
	--    after completion
	local corruptionCleanupSuccessResultStream =
		Rx.Observable.concat( 
			corruptionCleanupCompleteOrThrowStream,
			restartWillSolveDbProblemResultStream )

	return corruptionCleanupSuccessResultStream
			:catch(
				function( errMsg )

					-- 5) corruptionCleanupSuccessResultStream can throw. so we
					--    use :catch( ... ) to trap errors that happing during
					--    shutting down and deleting the corrupt db.
					--    
					--    We convert the error into a "data" result that
					--    contains failure error info 

					return makeCleanupCorruptDbResultStream( 
							errorType, false, dbName, errMsg )

				end )

end

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

local function maybeGetCleanupCorruptDbResultStreamForManifestEntry( 
				resultStream, manifestEntry, containerObj, errorType, path )

	if resultStream then

		return resultStream

	end

	local dbName, dbFileNamePattern, extractDbObjFromContainerFunc,
			corruptionCleanupResultStreamFunc = unpack( manifestEntry )

	if not lcPatternMatch( path, dbFileNamePattern ) then

		return

	end

	local dbObj = maybeGetDbObj( dbName, extractDbObjFromContainerFunc, 
									containerObj )

	if not dbObj then

		-- if we matched the file pattern but can't get a dbObj, return 
		-- a result stream with an error
		local canBeSolvedByRestart = errorType ~= "corrupt" 

		return makeCleanupCorruptDbResultStream( 
				errorType, canBeSolvedByRestart, dbName, "couldn't get dbObj" )

	end

	return doCleanupCorruptDbResultStream( 
			dbName, path, dbObj, corruptionCleanupResultStreamFunc, errorType )


end

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

local function maybeGetCleanupCorruptDbResultStream( 
					manifest, containerObj, errorType, path )

	return WFFunctionUtils.reduce( nil,
			maybeGetCleanupCorruptDbResultStreamForManifestEntry,
				manifest, containerObj, errorType, path )


end

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

local function doesDbPathHaveHandler( hasHandler, handledDbPathPattern, dbPath )

	if hasHandler then

		return true

	end

	if not lcPatternMatch( dbPath, handledDbPathPattern ) then

		return false

	end

	return true

end

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

local function maybeGetUnhandledCorruptDbResultStream( errorType, path )

	local hasHandler =
			WFFunctionUtils.reduce( 
				false, 
				doesDbPathHaveHandler, 
					_allFilePatterns, path )

	if hasHandler then

		return

	end

	local canBeSolvedByRestart = errorType ~= "corrupt" 

	local filename = string.find( path, ":memory:" ) and 
						path or KSPathUtils.leafName( path )
	
	local scrubbedFilename = string.gsub( filename, _uuidRegExPattern, "UUID" )

	return makeCleanupCorruptDbResultStream( 
			errorType, canBeSolvedByRestart, scrubbedFilename, "unhandled db" )

end

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

local function handleCorruptNmbCatalogDerivedDbResultStream( 
				corruptionEventStream, wfCatalog, profileCatalog,
				notifDbContainer )

	return corruptionEventStream
			:flatMapLatest(
				function( errorType, path )

					logger:infof( "Saw NmbCatalog derived db corruption %s %s", 
									errorType, path )

					return

						maybeGetCleanupCorruptDbResultStream(
							_wfCatalogCorruptionSupportManifest, wfCatalog, 
							errorType, path ) or

						maybeGetCleanupCorruptDbResultStream(
							_profileCatalogCorruptionSupportManifest,
							profileCatalog, errorType, path ) or

						maybeGetCleanupCorruptDbResultStream(
							_notifDbCorruptionSupportManifest,
							notifDbContainer, errorType, path ) or

						maybeGetUnhandledCorruptDbResultStream( 
							errorType, path ) or

						Rx.Observable.empty()

				end )

end

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

local function handleCorruptUserInfoCacheDbResultStream( 
				corruptionEventStream, wfCatUser )

	return corruptionEventStream
			:flatMapLatest(
				function( errorType, path )

					logger:infof( "Saw user info cache derived db corruption %s %s", 
									errorType, path )

					return

						maybeGetCleanupCorruptDbResultStream(
							_userInfoCorruptionSupportManifest, wfCatUser,
							errorType, path ) or

						Rx.Observable.empty()

				end )

end

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

return {

	simulateCorruptionDbList = _allRootModelDbNames,

	makeSimulateCorruptionForDbResultStream = 
		makeSimulateCorruptionForDbResultStream,

	makeSimulateCorruptUserInfoCacheStream = 
		makeSimulateCorruptUserInfoCacheStream,

	corruptionEventOnce = corruptionEventOnce,

	handleCorruptNmbCatalogDerivedDbResultStream = 
		handleCorruptNmbCatalogDerivedDbResultStream,

	handleCorruptUserInfoCacheDbResultStream =
		handleCorruptUserInfoCacheDbResultStream,

}

