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

WFCatNotificationSync.lua

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

ADOBE CONFIDENTIAL
------------------
Copyright 2020 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remains 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 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 Systems Incorporated.

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

	-- RxLua
local Rx = require 'Rx'
local Observable = Rx.Observable
local WFRxRunUntilDone = require 'WFRxRunUntilDone'

	-- WF.core
local WFCapabilities = require 'WFCapabilities'

	-- WF.core.rxlua
local WFRxScheduler = require 'WFRxScheduler'

	-- WF.core.logger
local WFLogger = require 'WFLogger'
local WFAnalytics = require 'WFAnalytics'

	-- WF.core.os
local KSDate = require 'KSDate'

	-- WF.core.utils
local WFCloseSupport = require 'WFCloseSupport'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFArguments = require 'WFArguments'
local KSTableUtils = require 'KSTableUtils'
local WFTableArchiving = require "WFTableArchiving"
local KSJsonEncoder = require 'KSJsonEncoder'
local KSJsonDecoder = require 'KSJsonDecoder'
local WFRxDatabase = require 'WFRxDatabase'

	-- WF.web.utils
local WFURI = require 'WFURI'

local WFCatNotification = require 'WFCatNotification'

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

local logger = WFLogger( 'WFCatNotificationSync' )
	-- :setLevel( 'info' ) -- 'debug' is useful for aggregated notifications

local check = assert( WFArguments.check )
local deref = assert( KSTableUtils.derefOpt )
local runUntilDone = WFRxRunUntilDone

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

local ANS_BASE_URL_PROD = "https://notify.adobe.io"
local ANS_BASE_URL_STAGE = "https://notify-stage.adobe.io"

local ANS_TIMELINE_URL = "/ans/v2/notifications/timeline"
local ANS_NOTIFICATION_URL = "/ans/v1/notifications"

local kEventWorkflow = 'Notification'
local kSecondsInADay = 24 * 60 * 60 

local kLrNotificationVersion = 1

local kVisitedTrue =  {
   	entry = {
        {
		    [ 'key-name' ] = 'visited',
        	[ 'key-value' ] = true
        }
	}
}

local kVisitedFalse =  {
	entry = {
		{
			[ 'key-name' ] = 'visited',
			[ 'key-value' ] = false
		}
	}
}


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

local function splitSet( str, splitPattern )

	local results = {}

	local start = 1
	local splitStart, splitEnd = string.find( str, splitPattern, start )

 	while splitStart do
    	results[ string.sub( str, start, splitStart - 1 ) ] = true
    	start = splitEnd + 1
    	splitStart, splitEnd = string.find( str, splitPattern, start )
	end

	results[ string.sub( str, start )  ] = true

	return results
	
end

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

local function getANSAdobeAppId()

	local adobeAppId
	
	if WFPlatformInfo.isDesktop then
		adobeAppId = "LightroomConsumerMac"
	elseif WFPlatformInfo.isiOS then
		adobeAppId = "LightroomConsumeriOS"
	elseif WFPlatformInfo.isAndroid then
		adobeAppId = "LightroomConsumerAndroid"
	end
	
	return adobeAppId

end

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

local function baseURL( catalogSpec )

	local isStageEnv = catalogSpec:isStageEnv()
	
	local url = ANS_BASE_URL_PROD
	
	if isStageEnv then
		url = ANS_BASE_URL_STAGE
	end
	
	return url

end

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

local function trackErrorEvent( trackingCategory, url, rawJsonOrError, headers, notificationId )

	local errorDesc = tostring( rawJsonOrError ) or ""
	
	if notificationId then
		logger:errorf( "Error at %s, url %s, error_desc %s, notificationId %s", trackingCategory, url, errorDesc, notificationId )
	else
		logger:errorf( "Error at %s, url %s, error_desc %s", trackingCategory, url, errorDesc )
	end

	-- Analytics
	WFAnalytics.trackEvent {
		['event.workflow'] = kEventWorkflow,
		['event.type'] = 'error',
		['event.subcategory'] = trackingCategory,
		['event.url'] =  url,
		['event.error_type'] = 'notificationSync.Error',
		['event.error_code'] = headers and headers.status_code or 'Unknown',
		['event.error_desc'] =  errorDesc,
		['source.id'] = notificationId
	}

end

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

local function getLinks( self )
	
	if ( self._previousLink and self._previousLink ~= "" ) or ( self._nextLink and self._nextLink ~= "" ) then
		return Observable.just( self._previousLink, self._nextLink )
	end
	
	return self._notificationDb:query( {
			queryOnce = true,
			sql = "SELECT key, value FROM notificationInfo WHERE key = ? OR key = ?",
			"previousLink", "nextLink"
		} )
		:take( 1 )
		:map( function( rows )
		
			for i = 1, #rows do
			
				local row = rows[ i ]
				local key = row[ 1 ]
				local value = row[ 2 ] 
			
				if key == "previousLink" then
				
					self._previousLink = value
				
				elseif key == "nextLink" then
				
					self._nextLink = value
					
				end
			
			end
			
			return self._previousLink, self._nextLink
			
		end )

end

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

local function updateLinks( self, links )
	
	local previousLink = links.previousLink
	local nextLink = links.nextLink
	
	local update = {}
	
	if previousLink and type( previousLink ) == "string" then
	
		if previousLink == "_delete" then
			self._previousLink = nil
			previousLink = ""
		else
			self._previousLink = previousLink
		end
		
		update[ #update+1 ] = { "previousLink", previousLink }
		
	end
	
	if nextLink and type( nextLink ) == "string" then
	
		if nextLink == "_delete" then
			self._nextLink = nil
			nextLink = ""
		else
			self._nextLink = nextLink
		end
		
		update[ #update+1 ] = { "nextLink", nextLink }
		
	end
	
	return self._notificationDb:bulkInsertOrReplace( "notificationInfo", update )

end

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

local function getPollingInterval( self, startOver, trackRestartAnalytics )

	check( 'WFCatNotificationSync', 'boolean' )
	
	return self._notificationDb:query({
		sql = "SELECT key, value FROM notificationInfo WHERE key LIKE ?", "polling%"
	})
	
	:take( 1 )
	
	:map( function( rows )
	
		local pollingIntervalInit, pollingBackoff, pollingIntervalMax
		
		for _, row in ipairs( rows ) do
		
			local key = row[ 1 ]
			local value = row[ 2 ] 
			
			if type( value ) == 'number' then
			
				if key == "pollingIntervalInit" then
					pollingIntervalInit = value
				elseif key == "pollingBackoff" then
					pollingBackoff = value
				elseif key == "pollingIntervalMax" then
					pollingIntervalMax = value			
				end
				
			end 
			
		end
		
		if not pollingIntervalInit or not pollingBackoff or not pollingIntervalMax then
			
			pollingIntervalInit = 60
			pollingBackoff = 60
			pollingIntervalMax = 300
			
		end 
		
		-- set pollingInterval
		
		if startOver or not self._pollingInterval then
		
			if self._pollingInterval and trackRestartAnalytics then
			
				local pollingBackoffCount = self._pollingBackoffCount
				
				-- Analytics
				WFAnalytics.trackEvent {
					['event.type'] = 'polling-interval',
					['event.workflow'] = kEventWorkflow,
					['event.subtype'] = 'restart',
					['event.subcategory'] = 'query',
					['event.value'] = pollingBackoffCount
				}
				
			end
		
		
			self._pollingInterval = pollingIntervalInit
			self._pollingBackoffCount = 0
		
		else
			
			self._pollingBackoffCount = self._pollingBackoffCount + 1
			
			local maybePollingInterval = self._pollingInterval + pollingBackoff
			
			if maybePollingInterval <= pollingIntervalMax then
				self._pollingInterval = maybePollingInterval
			end
		
		end
	
		return self._pollingInterval
	
	end )
		
end

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

local function localeInit( self, locale )

	return self._notificationDb:query({
		sql = "SELECT value FROM notificationInfo WHERE key = ?", "locale"
	})
	
	:take( 1 )
	
	:map( function( rows )
		
		local storedLocale = rows and rows[ 1 ] and rows[ 1 ][ 1 ]
		local updateLocale = false
		
		if locale then
			if storedLocale ~= locale then
				updateLocale = true
			end
		elseif storedLocale then
			locale = storedLocale
		else
			locale = "en_US"
			updateLocale = true
		end
		
		return updateLocale
		
	end )
	
	:flatMap( function( updateLocale )
	
		if updateLocale then
		
			return updateLinks( self, { previousLink = "_delete", nextLink = "_delete" } )
				:flatMap( function()
					return self._notificationDb:clearTable( "notifications" )
				end )
				:flatMap( function()
					return self._notificationDb:bulkInsertOrReplace( "notificationInfo", {{ "locale", locale }} )
				end )
				
		else
		
			return Observable.just( true )
			
		end
		
	end )
	
	:map( function()
	
		self._locale:onNext( locale )
		return locale
		
	end )
	
end 

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

local function requestOptions()

	local adobeAppId = getANSAdobeAppId()

	local options = {
		requestHeaders = {
			{
				header = 'Content-Type',
				value = 'application/json',
			},
			{
				header = 'Accept',
				value = 'application/json',
			},
			{
				header = 'x-adobe-app-id',
				value = adobeAppId,
			}
		},
		returnHttpErrorsAsData = true,
	}
	
	return options
end

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

local function queryNotificationUrl( catalogSpec, optUrl, optLocale, optInitWithTimelineAPI )

	check( 'WFCatSpec', '?string', '?string', '?boolean' )
	
	local url = optUrl
	
	if not optUrl or optUrl == "" then
	
		if optInitWithTimelineAPI then
			url = WFURI.resolve( baseURL( catalogSpec ), ANS_TIMELINE_URL )
		else
			url = WFURI.resolve( baseURL( catalogSpec ), ANS_NOTIFICATION_URL )
		end
		
		local queryParams = {}
		
		if optInitWithTimelineAPI then
			queryParams.config_version = "0"
		end
		
		if optLocale then
			queryParams.locale = optLocale
		end
		
		local query = WFURI.buildQueryString( queryParams )
		
		if query and query ~= "" then
			url = url .. "?" .. query
		end
		
	else
	
		local parsedURI = WFURI.parse( url )
		local queryParams = WFURI.parseQueryString( parsedURI.query )
		
		if not queryParams.config_version then
			queryParams.config_version = "0"
			
			parsedURI.query = WFURI.buildQueryString( queryParams )
			url = WFURI.buildUri( parsedURI )
		end
	
	end

	return url

end

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

local function waitForLocaleInit( self )
	
	check( 'WFCatNotificationSync' )
	
	return self._locale
		:filter( function( locale ) return not not locale end )
		:take( 1 ) -- this ensures that the locale initialization was complete before we start pulling data
end

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

local function ansBool( value )

	-- [immisch 2020-11-26]: In the metadata for a notification, ANS sends us false as the string 'false'.

	if not value then
		return false
	end

	if type( value ) == 'boolean' then
		return value
	end

	if type( value ) == 'string' then
		return value == 'true'
	end
	
	if type( value ) == 'number' then
		return value ~= 0
	end

end

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

local function queryNotificationOnce( self, optUrl, optLocale, optInitWithTimelineAPI )
	
	check( 'WFCatNotificationSync', '?string', '?string', '?boolean' )
	
	local catalogSpec = self._catalogSpec
	local notificationDb = self._notificationDb
	
	local url = queryNotificationUrl( catalogSpec, optUrl, optLocale, optInitWithTimelineAPI )
	
	local options = requestOptions()
	
	return waitForLocaleInit( self )
	:flatMap( function()
		return catalogSpec:get( url, options )
	end )
	:flatMap( function( json, headers )
			
		if not headers or headers.status_code ~= '200' then
			trackErrorEvent( "query", url, json, headers )
			return Observable.just()
		end
		
		local success, data = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, json )
		if not success or not data then
			trackErrorEvent( "query", url, json, headers )
			return Observable.just()
		end
				
		-- TODO [ylo 2020-04-27]: It's alarming to have different response structure in different API
		-- timeline API notification list is data.notifications, 
		-- and query API notification list is data.notifications.notification
		
		local notificationList = deref( data, 'notifications', 'notification' )
		if not notificationList then
			notificationList = deref( data, 'notifications' )
		end
		
		local responsePreviousLink = deref( data, 'paging', 'previous' )
		local responseNextLink = deref( data, 'paging', 'next' )
		local configParam = deref( data, 'notification-configuration', 'config-params', 'config-param' ) or {}
		local configVersion = deref( data, 'notification-configuration', 'version' )
		
		if not notificationList or type( notificationList ) ~= 'table' then
			trackErrorEvent( "query", url, "notification list is nil or not a table", headers )
			return Observable.just()
		end
		
		if ( not responseNextLink or responseNextLink == "" )
				and ( not responsePreviousLink or responsePreviousLink == "" ) then
			trackErrorEvent( "query", url, "doesn't have next and previous links", headers )
			logger:errorf( "QUERY Notification: doesn't have next and previous links with url: %s, x-request-id: %s", 
							url, headers[ 'x-request-id' ] )
		end 
		
		local insertOrReplaceRows = {}
		local deleteRows = {}
		local configRows = {}
		
		for k, notification in pairs( notificationList ) do
			
			local notificationId = notification[ 'notification-id' ]
			local subtype = notification[ 'sub-type' ]
			
			if not notificationId or not subtype then
				trackErrorEvent( "query", url, "doesn't have notification-id or sub-type", headers )
			else
			
				local time = notification[ 'sort-timestamp' ]
				local state = notification[ 'state' ]
				local payloadJsonString = notification[ 'payload' ]

				local successPayload, _ = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, payloadJsonString )
				if not successPayload then
					trackErrorEvent( "query-invalid-payload", url, json, headers, notificationId )					
				else
				
					local timeString = KSDate.timeInIso8601FormatWithUnixEpochInterval( time/1000 )
					
					if state == 'EXPIRED' then

						logger:infof( 'expired: %s', notificationId )
						
						deleteRows[ #deleteRows+1 ] = notificationId
					
					else
					
						local read = ( state == 'READ' )

						local visited = ansBool( deref( notification, 'metadata', 'visited' ) )

						local needsUpdateOnANS = false

						local now = KSDate.timeInIso8601Format()

						-- WF-1090 [immisch 2020-12-01]: we may receive an update to an existing aggregated notification
						-- that contains a new notification. If the notification is in state 'NEW', we must unset the visited flag
						-- and may have to update ANS
						if not read then
							needsUpdateOnANS = visited
							visited = false
						end

						if KSDate.differenceInDays( timeString, now ) > WFCatNotification.ANS_NOTIFICATION_EXPIRY_IN_DAYS then
							
							logger:info( string.format( 'received locally expired (marking as read) %s %s %s %s %s%s x-request-id %s',
								subtype, notificationId, timeString, read and 'read' or 'unread',
								visited and 'visited' or 'unvisited',
								needsUpdateOnANS and ' needsUpdateOnANS' or '', headers[ 'x-request-id' ] ) )

							needsUpdateOnANS = needsUpdateOnANS or not read
							read = true
							
						elseif logger:isEnabled( 'debug' ) then

							-- pretty-print the payload in debug only

							logger:debug( string.format( 'received %s %s %s %s %s%s x-request-id %s: %s',
								subtype, notificationId, timeString, read and 'read' or 'unread',
								visited and 'visited' or 'unvisited',
								needsUpdateOnANS and ' needsUpdateOnANS' or '', headers[ 'x-request-id' ],
								payloadJsonString ) )

						else
							logger:info( string.format( 'received %s %s %s %s %s%s x-request-id %s',
								subtype, notificationId, timeString, read and 'read' or 'unread',
								visited and 'visited' or 'unvisited',
								needsUpdateOnANS and ' needsUpdateOnANS' or '', headers[ 'x-request-id' ] ) )
						end

						insertOrReplaceRows[ #insertOrReplaceRows+1 ] = { notificationId,
							subtype,
							timeString,
							read,
							payloadJsonString,
							visited,
							needsUpdateOnANS
						}
						end

				end

			end
		
		end
		
		for _, param in pairs( configParam ) do
		
			local key = param.key
			local valueStr = param.value and param.value[ 1 ]
			local value = valueStr and tonumber( valueStr )
			
			if ( param.key == "pollingIntervalInit" 
				or param.key == "pollingBackoff" 
				or param.key == "pollingIntervalMax"
				or param.key == "lrNotificationVersion" ) and value then
				
				configRows[ #configRows + 1 ] = { key, value }
				
			end
		
		end
		
		if configVersion and type( configVersion ) == 'number' then
			
			local updateLinkConfigVersion = function( link )

				local parsedURI = WFURI.parse( link )
				local queryParams = WFURI.parseQueryString( parsedURI.query )
				
				if queryParams.config_version then
					queryParams.config_version = tostring( configVersion )
					
					parsedURI.query = WFURI.buildQueryString( queryParams )
					return WFURI.buildUri( parsedURI )
				end
					
				return link
				
			end
			
			if responseNextLink and responseNextLink ~= "" then
				responseNextLink = updateLinkConfigVersion( responseNextLink )
			end
			
			if responsePreviousLink and responsePreviousLink ~= "" then
				responsePreviousLink = updateLinkConfigVersion( responsePreviousLink )
			end
		end
		
		-- Update notifications and notificationInfo table
		
		local result = Rx.Observable.just()
		local hasNotificationUpdate = false
		
		if #deleteRows > 0 then
			
			hasNotificationUpdate = true
			
			result = notificationDb:deleteWhereIn( "notifications", "notificationId", deleteRows )
		
		end
		
		if #insertOrReplaceRows > 0 then
		
			hasNotificationUpdate = true
			
			result = result
				:flatMap( function( response )
					return notificationDb:bulkInsertOrReplace( "notifications", insertOrReplaceRows )
				end )
		
		end
		
		if #configRows > 0 then
		
			result = result
				:flatMap( function( response )
					return notificationDb:bulkInsertOrReplace( "notificationInfo", configRows )
				end )
		
		end
				
		return result
			:map( function()
				return responsePreviousLink, responseNextLink, hasNotificationUpdate 
			end )
	end )

end

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

--[[
	\brief Update the Notification state on ANS and then update the db
	
	\param notificationsUpdates (table) a table with the \c notificationId as the key. 
		Values are tables with the keys \c read and \c visited and boolean values that indicate the state.
		
	\return (table) An array of notificationIds
]]--

local function updateNotificationStateOnANSOnce( self, notificationUpdates )
	local url = baseURL( self._catalogSpec )
	
	url = url .. ANS_NOTIFICATION_URL
	
	local options = requestOptions()
	
	local notifications = {}
	
	for notificationId, update in pairs( notificationUpdates ) do
	
		notifications[ #notifications + 1 ] = {
			[ 'notification-id' ] = notificationId,
			state = update.read and 'READ' or 'NEW',
			metadata = update.visited and kVisitedTrue or kVisitedFalse
		}
	
	end
	
	local requestBody = {
		notifications = {
			notification = notifications
		}
	}
	
	local json = KSJsonEncoder.prettyPrint( requestBody )
	
	return self._catalogSpec:put( url, json, options )
		:flatMap( function( response, headers )
		
			if not headers or headers.status_code ~= '200' then
			
				trackErrorEvent( "update-ans-status", url, json, headers )
				
				return Rx.Observable.just( {} )

			end
			
			if logger:isEnabled( 'info' ) then
				logger:info( string.format( 'updated ANS state %s', WFTableArchiving.tableToString( notificationUpdates ) ) )
			end

			local notificationIds = KSTableUtils.extractKeys( notificationUpdates )
			
			return self._notificationDb:update( 'notifications', {
					values = {
						needsUpdateOnANS = false
					},
					condition = "WHERE notificationId IN CARRAY( ?, ?, 'char*' )",
					notificationIds,
					#notificationIds
				} )
				:map( function()
					return notificationIds
				end )
			
		end )
	
end

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


local function hasAnyNotifications( self )
	
	return self._notificationDb:query( {
			sql = "SELECT time FROM notifications LIMIT 1"
		} )
	:map( function( rows )
		return #rows > 0
	end )
	:distinctUntilChanged()

end

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

local function notificationExpiryCheckInit( self )

	return hasAnyNotifications( self )
		:flatMapLatest( function( hasAnyNotifications )
			if not hasAnyNotifications then
				return Observable.never()
			else
				return Observable.combineLatest( self._nextNotificationAutoExpiryCheck:flatMapLatest( function( refreshInSeconds )
					return Rx.Observable.timer( refreshInSeconds, WFRxScheduler.sharedScheduler() )
				end ))
				:flatMap( function()
					return WFCatNotification.deleteNotificationsOlderThan( self._notificationDb, WFCatNotification.ANS_NOTIFICATION_EXPIRY_IN_DAYS )
				end )
				:flatMap( function()
			
					return self._notificationDb:query( {
						sql = "SELECT time FROM notifications ORDER BY time ASC LIMIT 1"
					} )
					:take( 1 ) 
					:map( function( rows )
					
							local oldestNotificationTime = deref( rows, 1, 1 )
						
							if oldestNotificationTime then
								-- calculate how many seconds until the next notification expires
								local now = KSDate.timeInIso8601Format()
								local secondsSinceOldestNotificationTime = KSDate.differenceInSeconds( oldestNotificationTime, now )
								local timeUntilOldestNotificationIsExpired = ( WFCatNotification.ANS_NOTIFICATION_EXPIRY_IN_DAYS * kSecondsInADay ) - secondsSinceOldestNotificationTime
								
								-- add a tiny bit of delay just to make sure we check again after the expiration
								timeUntilOldestNotificationIsExpired = timeUntilOldestNotificationIsExpired + 2 
			
								
								-- round up to 10 seconds, no need to check too frequently
								if timeUntilOldestNotificationIsExpired < 10 then
									timeUntilOldestNotificationIsExpired = 10
								end
			
								self._nextNotificationAutoExpiryCheck:onNext( timeUntilOldestNotificationIsExpired )
							end
			
					end )
				end )
			end
		end )

end

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

local function allowNotificationSyncCheckInit( self )

	return self._notificationDb:query( {
			sql = "SELECT value FROM notificationInfo WHERE key = ?", "lrNotificationVersion"
		} )
	:map( function( rows )
		
		local allowNotificationSync = true
		
		if rows[ 1 ] and rows[ 1 ][ 1 ] and rows[ 1 ][ 1 ] > kLrNotificationVersion then
			allowNotificationSync = false
		end
		
		self._allowNotificationSyncCheck:onNext( allowNotificationSync )
		
		return allowNotificationSync
		
	end )

end

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

local WFCatNotificationSyncMethods = {

	class = {
		getANSAdobeAppId = getANSAdobeAppId
	},

	_catalog = false,
		-- (WFCatalog)
	
	_notificationDb = false,
		-- (WFRxDatabase)
		
	_catalogSpec = false,
		-- (WFCatSpec)
		
	_userIdIMS = false,
		-- (string)
		
	_previousLink = false,
		-- (string)
		
	_nextLink = false,
		-- (string)
		
	_pollingInterval = false,
		-- (number)
		
	_pollingBackoffCount = false,
		-- (number)
		
	_locale = false,
		-- (Rx.ReplaySubject)
		
	_localeInitSubscription = false,
		-- localeInit subscription
	
	_nextNotificationSyncCheck = false,
		-- (Rx.BehaviorSubject)
		
	_notificationSyncSubscription = false,
		-- notificationSync poll subscription
	
	_nextNotificationAutoExpiryCheck = false,
		-- (Rx.BehaviorSubject) 
	
	_nextNotificationAutoExpiryCheckSubscription = false,
		-- notification expiry subscription
		
	_allowNotificationSyncCheck = false,
		-- (Rx.BehaviorSubject) 
	
	_allowNotificationSyncCheckSubscription = false,
		-- allow notification sync check subscription
		
	_clientSubtypeAllowList = false,
		-- can be overridden via capabilities
		
		-- notification 
	WFCloseSupport.addCloseSupportMethods,
	
}


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

function WFCatNotificationSyncMethods:__init( catalog, optLocale, optCatSpec, debug_noLocaleInit, debug_optOutExpiryCheck )

	check( 'WFCatNotificationSync', 'WFCatalog', '+string|nil', '?WFCatSpec', '?boolean', '?boolean' )

	self._catalog = assert( catalog )
	
	local rxDatabasePath = catalog:docStore():path():gsub( '%.[^./\\]*$', '' ) .. '.notifdb'
	local rxDatabase = WFRxDatabase( rxDatabasePath, 'LrRxNotificationDbDefinitions.lua' )
	
	self:setOwned {
		_notificationDb = rxDatabase,
	}
	
	self._catalogSpec = optCatSpec or catalog:ozCatalogSpec()
	
	self._catalogSpec:userIdIMS()
		:map( function( userIdIMS )
			self._userIdIMS = userIdIMS
		end )
		:call( runUntilDone )
		
	-- locale setup
	
	optLocale = optLocale or "en_US"
	
	self._locale = Rx.ReplaySubject( 1 )
	self._nextNotificationAutoExpiryCheck = Rx.ReplaySubject( 0 )
	self._allowNotificationSyncCheck = Rx.ReplaySubject( 1 )
		
	if not debug_optOutExpiryCheck then
		self._nextNotificationAutoExpiryCheckSubscription = notificationExpiryCheckInit( self ):subscribe()
		self._nextNotificationAutoExpiryCheck:onNext( 1 )
	end
	
	if debug_noLocaleInit then
		self._locale:onNext( optLocale )
	else
		self._localeInitSubscription = localeInit( self, optLocale ):subscribe()
	end
	
	self._allowNotificationSyncCheckSubscription = allowNotificationSyncCheckInit( self ):subscribe()
	
end

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

function WFCatNotificationSyncMethods:close()
	
	if self._notificationSyncSubscription then
		self._notificationSyncSubscription:dispose()
		self._notificationSyncSubscription = nil
	end
	
	if self._nextNotificationSyncCheck then
		self._nextNotificationSyncCheck:dispose()
		self._nextNotificationSyncCheck = nil
	end
	
	if self._localeInitSubscription then
		self._localeInitSubscription:dispose()
		self._localeInitSubscription = nil
	end
	
	if self._nextNotificationAutoExpiryCheckSubscription then
		self._nextNotificationAutoExpiryCheckSubscription:dispose()
		self._nextNotificationAutoExpiryCheckSubscription = nil
	end
	
	if self._allowNotificationSyncCheckSubscription then
		self._allowNotificationSyncCheckSubscription:dispose()
		self._allowNotificationSyncCheckSubscription = nil 
	end

	self._clientSubtypeAllowList = nil

end

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

function WFCatNotificationSyncMethods:notificationDb()
	return self._notificationDb
end

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

function WFCatNotificationSyncMethods:clientSubtypeAllowList()

	if self._clientSubtypeAllowList then
		return self._clientSubtypeAllowList
	end

	local clientAllowList = WFCapabilities.getCapability( 'WFCatNotification.subtypeAllowList' )
	
	if not clientAllowList then
		return WFCatNotification.defaultSubtypeAllowList()
	else
		self._clientSubtypeAllowList = splitSet( clientAllowList, ',' )
	end

	return self._clientSubtypeAllowList

end

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

function WFCatNotificationSyncMethods:startNotificationSync( options )

	options = options or {}

	check( 'WFCatNotificationSync', '?table' )
	
	if self._notificationSyncSubscription then
		logger:debug( "startNotificationSync already started" )
		return Observable.just()
	end
	
	self._nextNotificationSyncCheck = Rx.BehaviorSubject( 0 )
	local allSyncShouldUsePollingInterval = false
	
	self._notificationSyncSubscription = self._allowNotificationSyncCheck
	
		:flatMapLatest( function( allowNotificationSync )
		
			if not allowNotificationSync then
				self._nextNotificationSyncCheck:onCompleted()
				return Observable.never()
			end
			
			return self._nextNotificationSyncCheck
	
				:flatMap( function ( refreshInSeconds )
					
					-- Need to wait self._locale initiation before getLinks
					
					return self._locale
					
					:flatMap( function( locale )
						
						return getLinks( self )
						
						:map( function( previousLink, nextLink )
							return locale, previousLink, nextLink
						end )
						
					end )
					
					:flatMap( function( locale, previousLink, nextLink ) 
					
						-- If previousLink is available, request immediately
						-- Otherwise, set timer with refreshInSeconds
						
						-- shouldUpdateNextLink would only be true when previousLink doesn't exist
						-- previousLink not exists means it's the top of this update cycle
						
						if previousLink and previousLink ~= "" then
							if options.debug_previousPageDelay or allSyncShouldUsePollingInterval then
							
								local delay = options.debug_previousPageDelay or refreshInSeconds
								return Observable.timer( delay, WFRxScheduler.sharedScheduler() )
									:map( function()
										return previousLink, locale, false
									end )
							else
								return Observable.just( previousLink, locale, false )
							end
							
						elseif not nextLink or nextLink == "" then
							if allSyncShouldUsePollingInterval then
								return Observable.timer( refreshInSeconds, WFRxScheduler.sharedScheduler() )
									:map( function()
										return nil, locale, true
									end )
							else
								return Observable.just( nil, locale, true )
							end
						else
							return Observable.timer( refreshInSeconds, WFRxScheduler.sharedScheduler() )
								:map( function()
									return nextLink, locale, true
								end )
						end
					
					end )
					
				end )
				
				:flatMap( function( url, locale, shouldUpdateNextLink )
				
					local initWithTimelineAPI = not url or url == ""
					
					return queryNotificationOnce( self, url, locale, initWithTimelineAPI )
							
						:flatMap( function( responsePreviousLink, responseNextLink, hasNotificationUpdate )
							
							if ( not responsePreviousLink or responsePreviousLink == "" )
									and ( not responseNextLink or responseNextLink == "" ) then
								allSyncShouldUsePollingInterval = true
								return Observable.just( false )
							end
							
							allSyncShouldUsePollingInterval = false
							
							-- We only use timeline API for initiation sync,
							-- so next link for future cycle, we should use query notification API
							if string.find( responseNextLink, ANS_TIMELINE_URL ) then
								local parsedNextLinkURI = WFURI.parse( responseNextLink )
								parsedNextLinkURI.rawPath = WFURI.resolve( parsedNextLinkURI.rawPath, ANS_NOTIFICATION_URL )
								responseNextLink = WFURI.buildUri( parsedNextLinkURI )
							end
							
							local links = {
								previousLink = responsePreviousLink,
								nextLink = responseNextLink
							}
							
							if not responsePreviousLink or responsePreviousLink == "" then
								links.previousLink = "_delete"
							end
							
							-- We should always get responseNextLink. If not, we should remain the original nextLink
							if not shouldUpdateNextLink or not responseNextLink or responseNextLink == "" then
								links.nextLink = nil
							end 
							
							return updateLinks( self, links )
								:map( function()
									-- notification update and page through previous link should restart polling interval
									return hasNotificationUpdate or not shouldUpdateNextLink
								end )
						
						end )
						
						:flatMap( function( restartPollingInterval )
		
							if options.debug_noANSStatusUpdate then
								return Observable.just( restartPollingInterval )
							end

							-- Update the read/visited state on ANS
							return self:updateNotificationStateOnce()
								:map( function()
									return restartPollingInterval
								end )
		
						end )
						
						:flatMap( function( restartPollingInterval )
				
							if options.debug_updateCount then
								options.debug_updateCount = options.debug_updateCount - 1
								
								if options.debug_updateCount <= 0 then
									self._nextNotificationSyncCheck:onCompleted()
								end
							end
							
							-- Only restart of pollingInterval with first paging request can be tracked
							local trackRestartAnalytics = restartPollingInterval and shouldUpdateNextLink
							
							return getPollingInterval( self, restartPollingInterval, trackRestartAnalytics )
								:map( function( pollingInterval )
								
									local interval = options.checkInterval 
													or WFCapabilities.getCapability( "WFCatNotificationSync.pollInterval" ) 
													or pollingInterval
									
									self._nextNotificationSyncCheck:onNext( interval )
								
								end )
						
						end )
					
				end )
		end )
		
		:subscribe()
		
		
	return self._nextNotificationSyncCheck
end

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

function WFCatNotificationSyncMethods:stopNotificationSync()

	if self._notificationSyncSubscription then
		self._notificationSyncSubscription:dispose()
		self._notificationSyncSubscription = nil
	end
	
	if self._nextNotificationSyncCheck then
		self._nextNotificationSyncCheck:dispose()
		self._nextNotificationSyncCheck = nil
	end
	
	self._pollingInterval = false
	self._pollingBackoffCount = false
	
end

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

function WFCatNotificationSyncMethods:queryNotificationTotalList( optLocale )

	check( 'WFCatNotificationSync', '?string' )
	
	local requestUrlCheck = Rx.BehaviorSubject( "" )
	
	return self._notificationDb:clearTable( "notifications" )
	
		:flatMap( function()
			return self._allowNotificationSyncCheck
		end )
		
		:take( 1 )
		
		:flatMap( function( allowNotificationSync )
			if not allowNotificationSync then
				return Observable.just()
			end
		
			return requestUrlCheck
				
				:flatMap( function( url )
					
					return queryNotificationOnce( self, url, optLocale )
				
				end )
				
				:map( function( previousLink, nextLink )
				
					if previousLink and previousLink ~= "" then
						requestUrlCheck:onNext( previousLink )
					else
						requestUrlCheck:onCompleted()
					end
					
				end )
				
		end )

end

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

function WFCatNotificationSyncMethods:DEBUG_queryNotificationOnce( optUrl, optLocale, optInitWithTimelineAPI )

	check( 'WFCatNotificationSync', '?string', '?string', '?boolean' )

	 return queryNotificationOnce( self, optUrl, optLocale, optInitWithTimelineAPI )

end

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

local function enumerateANSUpdates( notificationDb )

	-- The ANS Update Notifications API takes a maximum of 10 Notifications
	-- https://wiki.corp.adobe.com/display/ANS/ANS+API+Reference#ANSAPIReference-UpdateNotificationsAPI
	
	local sql = [[SELECT notificationId, visited, read FROM notifications 
		WHERE notificationId > ? AND needsUpdateOnANS ORDER BY notificationId ASC LIMIT 10]]

    local lastNotificationIdSubject = Rx.ReplaySubject( 1 )
    lastNotificationIdSubject:onNext( '' )

    return lastNotificationIdSubject:flatMap(function( lastNotificationId )

        return notificationDb:query { sql = sql, 
			queryOnce = true,
			lastNotificationId 
		}
			:take( 1 )
			:tap( function( rows )

				if not next( rows) then
					lastNotificationIdSubject:onCompleted()
				else
					local lastNotificationId = rows[ #rows ][ 1 ]

					lastNotificationIdSubject:onNext( lastNotificationId )
				end
				
			end )

    end )

end

function WFCatNotificationSyncMethods:updateNotificationStateOnce()

	check( 'WFCatNotificationSync' )

	return enumerateANSUpdates( self._notificationDb )
		:flatMap( function( rows )
		
			local notifications = {}
		
			for _, row in ipairs( rows ) do
			
				local notificationId, visited, read = unpack( row )
							
				notifications[ notificationId ] = {
					read = (read == 1) and true or false,
					visited = (visited == 1) and true or false
				}
										
			end
			
			if next( notifications ) then
				return updateNotificationStateOnANSOnce( self, notifications )
			else
				return Rx.Observable.just( {} )
			end
		
		end )

end

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

function WFCatNotificationSyncMethods:DEBUG_getPollingInterval( startOver )

	check( 'WFCatNotificationSync', 'boolean' )

	return getPollingInterval( self, startOver )
	
end

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

function WFCatNotificationSyncMethods:notificationUnreadCount()
	
	check( 'WFCatNotificationSync' )
	
	return WFCatNotification.notificationUnreadCount( self._notificationDb, self:clientSubtypeAllowList() )
	
end

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

function WFCatNotificationSyncMethods:notificationList( optWaitUntilSyncCycleComplete )
	
	check( 'WFCatNotificationSync', '?boolean' )
	
	if not optWaitUntilSyncCycleComplete then
		return WFCatNotification.notificationList( self._notificationDb, self:clientSubtypeAllowList() )
	end
	
	local previousLinkWasEmpty = false
	
	return self._allowNotificationSyncCheck
	
	:take( 1 )
	
	:flatMap( function( allowNotificationSync )
		if not allowNotificationSync then
			return WFCatNotification.notificationList( self._notificationDb, self:clientSubtypeAllowList() )
		end
	
		return self._notificationDb:query({
			sql = "SELECT value FROM notificationInfo WHERE key = ?", "previousLink"
		})
		
		:map( function( rows )
			
			local allowNotificationListUpdate
			local currentSyncCycleIsComplete = false
			
			local previousLink = rows and rows[ 1 ] and rows[ 1 ][ 1 ]
			
			if previousLink and previousLink ~= "" then
			
				-- If it's init sync with timeline API, sort order of response is correct
				-- We can immediately return notification list
				if string.find( previousLink, ANS_TIMELINE_URL ) then
					allowNotificationListUpdate = true
				else
					allowNotificationListUpdate = false
				end
				
				previousLinkWasEmpty = false
				
			else
			
				-- If previousLink is empty, it means current sync cycle is complete
				currentSyncCycleIsComplete = true
				
				if not previousLinkWasEmpty then
					allowNotificationListUpdate = true
				else
					allowNotificationListUpdate = false
				end
				
				previousLinkWasEmpty = true
				
			end
			
			return currentSyncCycleIsComplete, allowNotificationListUpdate

		end )
		
		:filter( function( currentSyncCycleIsComplete, allowNotificationListUpdate )
			return allowNotificationListUpdate
		end )
		
		:flatMap( function( currentSyncCycleIsComplete )
			
			return WFCatNotification.notificationList( self._notificationDb, self:clientSubtypeAllowList() )
				:take( 1 )
				:map( function( notificationList )
					return notificationList, currentSyncCycleIsComplete
				end )
		
		end )
		
	end )

end

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

function WFCatNotificationSyncMethods:setReadNotification( notificationsMarkRead, optNotificationsMarkRead )
	
	return WFCatNotification.setReadNotification( self._notificationDb, notificationsMarkRead, optNotificationsMarkRead )
		:flatMap( function( rowsAffected )
			return self:updateNotificationStateOnce()
				:map( function( notificationIds )
					return #notificationIds
				end )
		end )
		:sum()
	
end

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

function WFCatNotificationSyncMethods:setVisitedNotification( notificationsMarkVisted, optNotificationsMarkVisited )
	
	return WFCatNotification.setVisitedNotification( self._notificationDb, notificationsMarkVisted, optNotificationsMarkVisited )
		:flatMap( function()
			return self:updateNotificationStateOnce()
				:map( function( notificationIds )
					return #notificationIds
				end )
		end )
		:sum()
	
end

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

function WFCatNotificationSyncMethods:notificationKillSwitch()

	return self._allowNotificationSyncCheck
		:map( function( allowNotificationSync )
			return not allowNotificationSync
		end )
		:distinctUntilChanged()

end 

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

function WFCatNotificationSyncMethods:haltAndDeleteNotificationDb()
	
	return self._notificationDb:haltAndDelete()
	
end

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

function WFCatNotificationSyncMethods:_DEBUG_simulateCorruptionNotificationDb()
	
	return self._notificationDb:_DEBUG_simulateCorruption()
	
end

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

return Rx.Class( 'WFCatNotificationSync', WFCatNotificationSyncMethods )
