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

NmbDCPDocumentProcessor.lua

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

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

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

local KSTableUtils = require 'KSTableUtils'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFArguments = require 'WFArguments'
local Rx = require 'Rx'
local WFCloseSupport = require 'WFCloseSupport'
local WFCatFileSpec = require 'WFCatFileSpec'
local KSPathUtils = require 'KSPathUtils'
local WFLogger = require 'WFLogger'
local WFRxRunUntilDone = require 'WFRxRunUntilDone'
local WFPlatformInfo = require 'WFPlatformInfo'
local WFRxBatchedBulkApplyOverlay = require 'WFRxBatchedBulkApplyOverlay'
local WFRxThrottledBatchedApplyOverlayToIds = require 'WFRxThrottledBatchedApplyOverlayToIds'
local NmbProfileCatalogUtils = require 'NmbProfileCatalogUtils'
local NmbTrackedDataBroadcaster = require 'NmbTrackedDataBroadcaster'
local WFRxScheduler = require 'WFRxScheduler'

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

local check = assert( WFArguments.check )

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

local kNotParsed = "not parsed"

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

local logger = WFLogger( "NmbDCPDocumentProcessor" )

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

local function isOriginalLinkInvalid( docValue )

	return KSTableUtils.derefOpt( docValue, '_serverProvidedMetadata', 'links',
													'/rels/master', 'invalid' )

end

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

local function profileKey( cameraModel, cameraProfileDigest )

	return string.format( "%s_%s", cameraModel, cameraProfileDigest )

end

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

local function generateIndexSchema( _, indexStoreSchema )

	check( '_', 'WFRxIndexStoreSchema' )

	indexStoreSchema:addTable {
	
		sql = [[CREATE TABLE customCameraProfiles( cameraProfileUniqueID,
				cameraProfileDigest, profileName, cameraModel, sha256, filename, href,
				waitingForBinary, profileLinkChecksPending, localPath, assetUpdateDone )]],
				
		hasOneRowPerDocument = true,
		
		relevantTypeSubtypeTuples = {
			{ 'asset', 'camera_profile' },
		},
		
		map = function( docValue, emit )
		
			if not docValue._deleted == true and docValue.type == "asset" 
				and ( docValue.subtype == "camera_profile") then
			
				local fingerprint = KSTableUtils.derefOpt( docValue, "resolvedDigest" )
				local sha256 = WFCatFileSpec.getLocalOriginalSha256WithFallback( docValue )
				local fileName = WFCatFileSpec.getOriginalRelativePathFromDocValue( docValue )
				fileName = fileName or KSTableUtils.derefOpt( docValue, "fileName" )
				fileName = fileName and KSPathUtils.leafName( fileName ) or nil
				local href = not isOriginalLinkInvalid( docValue ) and WFCatFileSpec.getOriginalLinkFromDocValue( docValue ) or nil
				local profileName = KSTableUtils.derefOpt( docValue, "name" )
				local localPath = WFCatFileSpec.buildOriginalPathFromDocValue( docValue )
				
				local profileLinkChecksPending = KSTableUtils.derefOpt( docValue, '_localOnly', 'profileLinkChecksPending' )
				local assetUpdateDone = not not KSTableUtils.derefOpt( docValue, '_localOnly', 'assetUpdateDone' )

				local cameraProfileDigest = KSTableUtils.derefOpt( docValue, "_localOnly", "cameraProfileDigest" )
				cameraProfileDigest = cameraProfileDigest or
					KSTableUtils.derefOpt( docValue, "cameraProfileDigest" )

				local cameraModel = KSTableUtils.derefOpt( docValue, "_localOnly", "cameraModel" )
				cameraModel = cameraModel or
						KSTableUtils.derefOpt( docValue, "model" )
						
				if type( cameraModel ) == 'string' then
					cameraModel = cameraModel:lower()
				end
				
				local ozSha256 = KSTableUtils.derefOpt( docValue, "importSource", "sha256" )
				local localSha256 = KSTableUtils.derefOpt( docValue, "_localOnly", "files", "original", "master", "sha256" )

				local waitingForBinary =
					(  ozSha256 and localSha256 and ozSha256 ~= localSha256 ) -- local file is stale
					or ( not localSha256 and ozSha256 ) -- no local binary, but binary available on Oz
				waitingForBinary = not not waitingForBinary
				
				--logger:infof( "waitingForBinary is %s for %s with ozSha256 = %s and localSha256 = %s",
				--				waitingForBinary, fileName, ozSha256, localSha256 )
				
				if fingerprint and cameraProfileDigest and profileName and fileName then
					
					-- Add the entry only if mandatory fields are present.
					
					emit( fingerprint, cameraProfileDigest, profileName, cameraModel, sha256,
							fileName, href, waitingForBinary, profileLinkChecksPending, 
							localPath, assetUpdateDone )
				end

			end
				
		end,
		
		updateLazily = false,

	}

	indexStoreSchema:addIndex(
		  [[CREATE INDEX customCameraProfiles_fingerprint
				  ON customCameraProfiles( cameraProfileUniqueId, cameraProfileDigest, docId )]] )
				  
	indexStoreSchema:addIndex(
		  [[CREATE INDEX customCameraProfiles_fingerprint_name
				  ON customCameraProfiles( cameraProfileDigest, profileName, docId )]] )

	indexStoreSchema:addIndex(
		  [[CREATE INDEX customCameraProfiles_cameraProfileDigest_cameraModel_href
				  ON customCameraProfiles( cameraProfileDigest, cameraModel, href, filename, profileLinkChecksPending )]] )

	indexStoreSchema:addIndex(
		  [[CREATE INDEX customCameraProfiles_profileLinkChecksPending
				  ON customCameraProfiles( profileLinkChecksPending )]] )

	indexStoreSchema:addIndex(
		  [[CREATE INDEX customCameraProfiles_waitingForBinary
				  ON customCameraProfiles( waitingForBinary )]] )

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

	-- This table is used to identify image assets which may need to have
	-- camera profile links populated in the develop.profiles.camera section.
	-- All nil values, other than docId, is an indication that parsing of the
	-- asset's xmpCameraRaw is needed, but that needs to happen outside of the
	-- map function, for performance reasons. The information obatined via the
	-- parsing is then stuffed into _localOnly fields, which can then be
	-- utilized in another pass through this function to populate the table
	-- with actual values. Those values, in turn can be used for populating
	-- the camera profile section, again outside of the map function.
	--
	-- The xmp parsing happens via maybeParseSomeXmpCameraRaw, and the profile
	-- linke generation happens via getProfileAndAssetInfo/processAssetBatch.
	--
	-- All of this with the goal of getting custom camera profile links populated
	-- for assets which previosly couldn't have profile links because custom
	-- camera profiles were not being synced.
		
	indexStoreSchema:addTable {
	
		sql = [[CREATE TABLE assetsNeedingCameraProfileLinks( cameraProfileDigest, cameraModel )]],
		
		hasOneRowPerDocument = true, -- One or less, really.
		
		relevantTypeSubtypeTuples = {
			{ 'asset', 'image' },
		},
		
		map = function( docValue, emit )
		
			if not WFPlatformInfo.hasACR then
				return
			end
		
			if docValue and docValue._deleted ~= true and docValue.subtype == 'image' then
			
				local cameraModel = KSTableUtils.derefOpt( docValue, "xmp", "tiff", "Model" )
				local cameraProfileDigest = KSTableUtils.derefOpt( docValue, "_localOnly", "customCameraProfile", "profileDigest" )
				local processedKey = KSTableUtils.derefOpt( docValue, "_localOnly", "customCameraProfile", "profileLinkCheckDone" )
				
				if type( cameraModel ) == 'string' then
					cameraModel = cameraModel:lower()
				end
				
				local key = profileKey( cameraModel, cameraProfileDigest )
				
				-- The 'cameraProfileDigest' will be "NA" in _localOnly after parsing
				-- in NmbLocalCameraProfileStatusWorker if
				-- a) the camera profile digest in xmp refers to an Adobe profile
				-- b) there was no camera profile specified in xmp
				-- So, if there's no cameraProfileDigest, that means the asset has not
				-- yet been parsed by NmbLocalCameraProfileStatusWorker.
				
				if processedKey and processedKey == key or cameraProfileDigest == "NA" then
					logger:infof( "NOT emitting for %s (cameraProfileDigest is %s)",
									KSTableUtils.derefOpt( docValue, 'importSource', 'fileName' ),
									cameraProfileDigest )
					return
				elseif not cameraProfileDigest then
					logger:infof( "emitting 'not parsed' for %s", KSTableUtils.derefOpt( docValue, 'importSource', 'fileName' ) )
					emit( kNotParsed, kNotParsed )
				else
					logger:infof( "emitting %s, %s for %s", cameraProfileDigest, cameraModel, KSTableUtils.derefOpt( docValue, 'importSource', 'fileName' ) )
					emit( cameraProfileDigest, cameraModel )
				end

			end
		
		end,
		
		updateLazily = false,

	}

	indexStoreSchema:addIndex(
		  [[CREATE INDEX assetsNeedingCameraProfileLinks_cameraProfileFingerprint
				  ON assetsNeedingCameraProfileLinks( cameraProfileDigest, cameraModel, docId )]] )
				  
	----------------------------------------------------------------------------
				  
	indexStoreSchema:addTable {
	
		sql = [[CREATE TABLE assetReferredCameraProfile( cameraProfileDigest, isMissing, isStandard )]],		
		
		useLocalId = true,
		
		hasOneRowPerDocument = true,
		
		relevantTypeSubtypeTuples = {
			{ 'asset', 'image' },
		},
		
		map = function( docValue, emit )
		
			if not WFPlatformInfo.hasACR then
				return
			end
		
			if docValue and docValue._deleted ~= true and docValue.subtype == 'image' then
	
				-- Maybe cameramodel is not populated yet.
				-- Camera model is needed for finding out
				-- if its a custom profile or not.
				-- Hence noop if its not yet populated
				local cameraModel = KSTableUtils.derefOpt( docValue, "xmp", "tiff", "Model" )
				if not cameraModel then
					return
				end
				
				-- If the new digest is different from the one we have
				-- then we need to populate the profile info again.
				local xmpCameraRawProfileDigest = NmbProfileCatalogUtils.getXmpCameraRawProfileDigest( docValue )
				local oldXmpCameraRawProfileDigest = KSTableUtils.derefOpt( docValue, "_localOnly", "cameraProfileInfo", "xmpCameraRawProfileDigest" )
				
				-- isStandard is not present. Need to recalculate this, since
				-- there is we might have missed storing the standard profile
				-- info.
				local isStandard = KSTableUtils.derefOpt( docValue, "_localOnly", "cameraProfileInfo", "isStandard" )
				
				if xmpCameraRawProfileDigest ~= oldXmpCameraRawProfileDigest or
					isStandard == nil then
					-- Develop settings has changed. 
					-- Recalculate
					emit( nil, nil, nil )
				else
				
					local cameraProfileDigest = KSTableUtils.derefOpt( docValue, "_localOnly", "cameraProfileInfo", "profileDigest" )
					
					if cameraProfileDigest == "NA" then
						return
					end
				
					local cameraProfileMissing = KSTableUtils.derefOpt( docValue, "_localOnly", "cameraProfileInfo", "isMissing" )
					
					emit( cameraProfileDigest, cameraProfileMissing, isStandard )
				end
			end
		
		end,
	}
	
	indexStoreSchema:addIndex(
		  [[CREATE INDEX assetReferredCameraProfile_digest_localId
				  ON assetReferredCameraProfile( cameraProfileDigest, docId )]] )

end

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

local function listenForIndexChange( self, indexStore, indexTableName, columns )

	return indexStore:query {
		sql = string.format(  "SELECT %s FROM %s", columns, indexTableName ),
		quiesceUntilUpToDate = true,
		calm = 30,
	}
	
	:flatMap( function( queryResult )
	
		self._taskTriggerSubject:onNext( true )
		
		return Rx.Observable.empty()
	
	end )
	
	:call( WFRxRunUntilDone )

end

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

local function getProfileAndAssetInfo( self, wfCatalog )

	return wfCatalog:indexStore():queryOnce {
		sql = [[ SELECT cameraProfileDigest, cameraModel, href, filename, assetsNeedingCameraProfileLinks.docId
					FROM customCameraProfiles
					JOIN assetsNeedingCameraProfileLinks USING( cameraProfileDigest, cameraModel )
					WHERE href NOTNULL
						AND profileLinkChecksPending NOTNULL ]],
		quiesceUntilUpToDate = true,
	}
	
	:map( function( queryResult )
	
		local processedResult = {}
		
		for _, row in ipairs( queryResult or {} ) do
		
			table.insert( processedResult, {
				cameraProfileDigest = row[ 1 ],
				cameraModel = row[ 2 ],
				profileHref = row[ 3 ],
				profileFilename = row[ 4 ],
				docId = row[ 5 ],
			} )
		
		end
		
		return processedResult
		
	end )

end

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

local function getCustomCameraProfileOverlaysForAssetDocs( self, wfCatalog, assetInfoByDocId )

	-- In this function we assume that 'assetInfoByDocId' contains the profile
	-- href and file name for each asset. This assumption is valid due to the
	-- nature of the query made in 'getProfileAndAssetInfo' further up the Rx
	-- chain.
	
	-- We also don't have to worry about leaving the profiles section of the
	-- asset document empty, because it follows from the assumption stated above
	-- that we have complete camera profile info for each asset.
	
	local profileOverlays = {}

	for id, assetInfo in pairs( assetInfoByDocId ) do
	
		logger:infof( "OBTAINING camera profile overlay for %s", id )
	
		
		-- If we're not talking about a custom camera profile, we aren't looking
		-- to fix it up at this time. We figure out if the camera profile is custom
		-- by checking if it resides in the user's own catalog.
		
		local catalogId = wfCatalog:ozCatalogId()
		local isCustomProfileHref = assetInfoByDocId[ id ].profileHref:lower():find( string.format( "/%s/", catalogId:lower() ) )
		
		if not isCustomProfileHref then
			logger:infof( "NOT adding camera profile link for %s (%s) because it's not a custom camera profile",
							assetInfoByDocId[ id ].profileFilename, assetInfoByDocId[ id ].profileHref )
		end
		
		local developOverlay = isCustomProfileHref and {
			profiles = {
				camera = {
					filename = assetInfoByDocId[ id ].profileFilename,
					href = assetInfoByDocId[ id ].profileHref
				}
			}
		} or nil
			
		local overlay = {
			_id = id,
			_localOnly = {
				customCameraProfile = isCustomProfileHref and {
					profileLinkCheckDone = profileKey( assetInfoByDocId[ id ].cameraModel, assetInfoByDocId[ id ].cameraProfileDigest )
				} or nil,
			},
			develop = developOverlay
		}
		
		table.insert( profileOverlays, overlay )
	
	end
	
	return profileOverlays

end

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

local function processAssetBatch( self, wfCatalog, assetInfoArray )

	if not next( assetInfoArray ) then
		return Rx.Observable.just( nil )
	end

	local assetIds = {}
	local assetInfoById = {}
	
	for _, item in ipairs( assetInfoArray ) do
		if not self._activelyProcessing[ item.docId ] then
			table.insert( assetIds, item.docId )
			self._activelyProcessing[ item.docId ] = true
			assetInfoById[ item.docId ] = item
		end
	end
	
	local docStore = wfCatalog:docStore()

	local overlays = getCustomCameraProfileOverlaysForAssetDocs( self, wfCatalog, assetInfoById )
	
	return WFRxBatchedBulkApplyOverlay( "processAssetBatch", docStore, overlays )
		
	:finally( function()
	
		for _, item in ipairs( assetInfoArray ) do
			self._activelyProcessing[ item.docId ] = nil
		end
		
	end )

end

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

local function maybeResetProfileLinkChecksPendingForSomeCustomProfiles( self )

	-- Identify entries in 'customCameraProfiles' which have no assets referring
	-- to them from 'assetsNeedingCameraProfileLinks'.
	
	-- TODO: Make sure the performance of this query sucks as little as possible.
	-- I.e. create appropriate indices or change the query itself.
	
	local query = [[ SELECT docId FROM customCameraProfiles
		WHERE NOT exists
			( SELECT 1 FROM assetsNeedingCameraProfileLinks
				WHERE customCameraProfiles.cameraModel = assetsNeedingCameraProfileLinks.cameraModel
					AND assetsNeedingCameraProfileLinks.cameraProfileDigest = customCameraProfiles.cameraProfileDigest )
			AND customCameraProfiles.profileLinkChecksPending = 1
			AND customCameraProfiles.href NOTNULL ]]

	return self._catalog:indexStore():queryOnce {
		sql = query,
		quiesceUntilUpToDate = true,
		singleColumnQuery = true,
	}
	
	:flatMap( function( profileDocIds )
	
		if next( profileDocIds ) then
		
			-- Generate overlays which will set 'profileLinkChecksPending' to nil
			
			local overlay = {
				_localOnly = {
					profileLinkChecksPending = {
						_delete = true
					}
				}
			}
			
			return WFRxThrottledBatchedApplyOverlayToIds( "maybeResetProfileLinkChecksPendingForSomeCustomProfiles", 
					self._catalog:docStore(), overlay, profileDocIds, 100, 10 )
		
		end
		
		return Rx.Observable.just( nil )
	
	end )

end

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

local function areAllAssetsParsed( self )

	return self._catalog:indexStore():queryOnce {
		sql = [[ SELECT COUNT( 1 )
					FROM assetsNeedingCameraProfileLinks
					WHERE cameraProfileDigest = ? ]],
		quiesceUntilUpToDate = true,
		kNotParsed
	}
	
	:map( function( countResult )
	
		local count = countResult and countResult[ 1 ] and countResult[ 1 ][ 1 ]
	
		return count == 0
	
	end )

end

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

-- Identify assets which name a camera profile by cameraProfileDigest which has
-- a different camera model name associated with the camera profile.

local function getCameraModelDiscrepancies( self, wfCatalog )

	return wfCatalog:indexStore():queryOnce {
		sql = [[ SELECT customCameraProfiles.cameraModel,
						assetsNeedingCameraProfileLinks.cameraModel, href,
						customCameraProfiles.cameraProfileUniqueID,
						cameraProfileDigest
					FROM customCameraProfiles
					JOIN assetsNeedingCameraProfileLinks USING( cameraProfileDigest )
					WHERE href NOTNULL
						AND assetsNeedingCameraProfileLinks.cameraModel != customCameraProfiles.cameraModel ]],
		quiesceUntilUpToDate = true,
	}
	
	:map( function( queryResult )
	
		local processedResult = {}
		
		for _, row in ipairs( queryResult or {} ) do
		
			table.insert( processedResult, {
				profileCameraModel = row[ 1 ],
				imageCameraModel = row[ 2 ],
				profileHref = row[ 3 ],
				cameraProfileUniqueID = row[ 4 ],
				cameraProfileDigest = row[ 5 ],
			} )
		
		end
		
		return processedResult
		
	end )

end

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

local function processPossibleCameraModelDiscrepancies( self, infoForAssetsWithPossibleDiscrepancy )
		
	-- Here we can get as fancy as we want. We could go as simple as
	-- checking if the profile's camera model is a substring of the image's
	-- camera model, or vice versa, or we could crack open the negative
	-- and see if we can obtain a cameraProfileUniqueID for the
	-- cameraProfileDigest (which would be very reliable and have the
	-- most overhead).
	
	-- For now, generate headlights data to enable sizing up of the real world
	-- problem.
	
	local discrepancyStats = {}
	
	for _, item in ipairs( infoForAssetsWithPossibleDiscrepancy ) do
	
		local assetCameraModel = item.imageCameraModel
		local profileCameraModel = item.profileCameraModel

		local statKey = string.format( "%s%s",
										profileCameraModel, assetCameraModel )
	
		discrepancyStats[ statKey ] = discrepancyStats[ statKey ] or {
			profileCameraModel = profileCameraModel,
			imageCameraModel = assetCameraModel,
			cameraProfileUniqueID = item.cameraProfileUniqueID,
			profileHref = item.profileHref,
			affectedAssetCount = 0,
			cameraProfileDigest = item.cameraProfileDigest,
		}
		
		discrepancyStats[ statKey ].affectedAssetCount = discrepancyStats[ statKey ].affectedAssetCount + 1
				
	end

	if next( discrepancyStats ) then
		NmbTrackedDataBroadcaster.broadcastMessage( 'CameraModelDiscrepancy', discrepancyStats )
	end
	
	

	return {}

end

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

-- A task to check whether there are entries in the 'assetsNeedingCameraProfileLinks'
-- table which need to have renditions re-generated on Oz. It is the job of this
-- Oz client to request such rendition re-generation if the custom camera
-- profile originated from this client.

local function maybeGenerateCameraProfileOverlays( self )

	if not self._isOnlineAndSyncing then
	
		return Rx.Observable.just( nil )
		
	end
	
	local wfCatalog = self._catalog
	
	return getProfileAndAssetInfo( self, wfCatalog )

	:flatMap( function( assetQueryResult )
	
		return processAssetBatch( self, wfCatalog, assetQueryResult )
		
		:flatMap( function()
		
			if next( assetQueryResult ) then
			
				-- Don't want to prematurely consider resetting the profileLinkChecksPending
				-- flag for any row in the customCameraProfiles table while we are waiting
				-- for the assetsNeedingProfileLinks table to be populated by the first
				-- build of Lr CC that defines it. (We may get some
				-- no-rows results from getProfileAssetInfo before the
				-- assetsNeedingCameraProfileLinks table is populated.)
				
				-- Waiting until we've gotten here might mean that there are rows left
				-- with profileLinkChecksPending unequal to nil in the customCameraProfiles
				-- table even if there aren't any assets in the catalog which reference
				-- the camera profile in question, but that doesn't matter if we don't
				-- actually find ourselves with items to process coming out of
				-- getProfileAndAssetInfo.

				-- To avoid processing asset links in perpetuity. We only want to fix up links
				-- which are missing as a result of assets being synced PRIOR to the DCP being
				-- available on Oz. See https://jira.corp.adobe.com/browse/NMB-7819.
				
				logger:infof( "calling maybeResetProfileLinkChecksPendingForSomeCustomProfiles" )
				
				return areAllAssetsParsed( self )
				
				:flatMap( function ( allassetsParsed )
				
					-- Even if all assets are parsed for customCameraProfile info, there
					-- is still a small chance that some assets need re-parsing due to user
					-- having made changes. Choosing to believe that that risk is small
					-- enough to ignore, for now.
					
					if allassetsParsed then
						return maybeResetProfileLinkChecksPendingForSomeCustomProfiles( self )
					else
						return Rx.Observable.just()
					end
					
				end )
			
			end
			
			logger:infof( "skipping maybeResetProfileLinkChecksPendingForSomeCustomProfiles" )
			return Rx.Observable.just()
		
		end )
		
	end )

	:flatMap( function()

		if self._customCameraModelDiscrepancyReportDone then
			return Rx.Observable.just()
		end
		
		self._customCameraModelDiscrepancyReportDone = true
		
		-- Five minutes into each Lr CC session, report custom camera profile discrepancies,
		-- if any to Highbeam. See https://jira.corp.adobe.com/browse/NMB-8598.
		
		return Rx.Observable.timer( 300, WFRxScheduler.sharedScheduler() )
		
		:take( 1 )
		
		:flatMap( function()
		
			return getCameraModelDiscrepancies( self, wfCatalog )
			
			:map( function( infoForAssetsWithPossibleDiscrepancy )
			
				processPossibleCameraModelDiscrepancies( self, infoForAssetsWithPossibleDiscrepancy )
				
			end )
			
		end )
	
	end )

	:call( WFRxRunUntilDone )
	
end

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

local NmbDCPDocumentProcessor = {
	
	_catalog = false,
	
	_profileCatalog = false,
	
	_activelyProcessing = false,
	
	_continuousTask = false,
	
	_taskTriggerSubject = false,
	
	_isOnlineAndSyncing = false,
	
	_isOnlineAndSyncingStream = false,
	
	_dcpRefreshSubject = false,
	
	_didInitialCheck = false,
	
	_currentlyBeingParsed = false,
	
	_customCameraModelDiscrepancyReportDone = false,
	
	class = {
		generateIndexSchema = generateIndexSchema,
	},
	
	WFCloseSupport.addCloseSupportMethods,
}

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

function NmbDCPDocumentProcessor:__init( catalog, profileCatalog )

	local volumeAgents =  catalog:getAllVolumeAgents()
	local volumeAgent = volumeAgents[ next( volumeAgents ) ]

	self:setOwned {
		_catalog = catalog:relayObject()
	}
	
	self._profileCatalog = profileCatalog
	
	self._activelyProcessing = {}
	
	self._taskTriggerSubject = Rx.ReplaySubject( 1 )
	
	self._dcpRefreshSubject = Rx.ReplaySubject( 1 )
	
	self._currentlyBeingParsed = {}
	
	if volumeAgent then
		self._isOnlineAndSyncingStream = Rx.Observable.combineLatest( volumeAgent:isSyncPausedStream(),
										volumeAgent:hasInternetConnectionStream() )
			:subscribe( function( isSyncPaused, hasInternetConnection )
				if ( not isSyncPaused ) and hasInternetConnection then
					self._isOnlineAndSyncing = true
					
					if not self._didInitialCheck then	
						-- Since we're throttling the checks, do an initial
						-- check on launch to 'catch up' if necessary.
						self._didInitialCheck = true

						maybeGenerateCameraProfileOverlays( self )
					end
					
				else
					self._isOnlineAndSyncing = false
				end
			end )
	end
	
	self._taskTriggerSubject:subscribe( function()

		maybeGenerateCameraProfileOverlays( self )

	end )
	
	local indexStore = self._catalog:indexStore()
	
	-- Start litening for changes in the index table fields of significance.
	
	listenForIndexChange( self, indexStore, 'assetsNeedingCameraProfileLinks',
							'cameraProfileDigest, cameraModel' )

	listenForIndexChange( self, indexStore, 'customCameraProfiles', 'href' )

end

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

function NmbDCPDocumentProcessor:close()

	local continuousTask = self._continuousTask
	
	self._continuousTask = nil
	
	if continuousTask then
		continuousTask:stopContinuousTask()
		continuousTask:close()
	end
	
	local isOnlineOnSyncingStream = self._isOnlineAndSyncingStream
	
	self._isOnlineAndSyncingStream = nil
	
	if isOnlineOnSyncingStream then
		isOnlineOnSyncingStream:dispose()
	end
	
	local taskTriggerSubject = self._taskTriggerSubject
	
	self._taskTriggerSubject = nil
	
	if taskTriggerSubject then
		taskTriggerSubject:dispose()
	end
	
	local dcpRefreshSubject = self._dcpRefreshSubject
	
	self._dcpRefreshSubject = nil
	
	if dcpRefreshSubject then
		dcpRefreshSubject:dispose()
	end
	
end

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

function NmbDCPDocumentProcessor:getDCPRefreshStream()

	return self._dcpRefreshSubject

end

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

NmbDCPDocumentProcessor.class.assetsWithProfilesStatsStream = 
function( _, wfCatalog )

	local profileStatusSQL = [[
			SELECT IIF( isMissing > 0, 'missing_', '' ) ||
					IIF( cameraProfileDigest > '', 
							IIF( isStandard > 0, 'standard', 'custom' ),
							'pending' ) AS status,
				count( * ) 
			FROM assetReferredCameraProfile
			GROUP BY status]]

	return wfCatalog
			:indexStore()
			:queryThenDebounce(  60.0, { sql = profileStatusSQL } )
			:map(
				function( sqlRows )

					-- sqlRows is
					-- {
					-- 	{ status1, count1 },
					-- 		...
					-- 	{ statusN, countN }
					-- }
					--
					-- status will be one of 
					--
					-- "missing_custom", "missing_standard", "standard", "custom", "pending"

					return WFFunctionUtils.reduce( {},
						function( stats, row )

							local status, count = row[ 1 ], row[ 2 ]

							stats[ status ] = count

							return stats

						end,
						sqlRows )

				end )

end

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

return Rx.Class 'NmbDCPDocumentProcessor'( NmbDCPDocumentProcessor )
