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

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

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

	-- Rx
local Rx = require "Rx"

local KSTableUtils = require "KSTableUtils"

	-- Utilities
local WFCloseSupport = require 'WFCloseSupport'
local WFLogger = require "WFLogger"
local KSLuaUtils = require "KSLuaUtils"

local LriBlobConsistencyModel = {}

local _, LrmTHLibrary = KSLuaUtils.requireIfExists 'LrmTHLibrary'

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

local derefOpt = assert( KSTableUtils.derefOpt )

local logger = WFLogger "LriFixCacheModel"
--:setLevel 'info'

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

-- page through cacheReferences.

local function enumerateCacheReferences( indexStore )

	local sql = [[SELECT DISTINCT docId FROM cacheReferences
			WHERE docId > ?
			ORDER BY docId ASC LIMIT 50 ]]

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

    return lastDocIdSubject:flatMap(function( lastDocId, lastRenditionName )

        return indexStore:queryOnce {
				sql = sql,
				quiesceUntilUpToDate = true,
				lastDocId,
				lastRenditionName
			}

			:map( function( rows )

				if not next( rows ) then

					lastDocIdSubject:onCompleted()

				else

					local lastDocId = rows[ #rows ][ 1 ]

					lastDocIdSubject:onNext( lastDocId )

				end

				return rows

			end )

    end )

end

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

local renditionNames = { 'thumbnail2x', 'preview', 'proxy', 'original', 'xmp_develop' }

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


local function localRenditions( docValue )
	-- return an array of rendition names that should have local blobs

	local renditions = {}

	for _, renditionName in ipairs( renditionNames) do

		local renditionInfo = derefOpt( docValue, '_localOnly', 'files', renditionName )
		if renditionInfo and renditionInfo.path then
			renditions[ #renditions + 1 ] = renditionName
		end

	end

	return renditions

end


--------------------------------------------------------------------------------
--[[!
	\brief Cross-check the actual availability of blobs between cacheReferences and the blobProvider.

	\param catalog (WFCatalog) catalog

	\param mockLocallyAvailable (function) Used for testing. A function with signature:
		
		function isLocallyAvailable( strippedDocValue, renditionList )
			return renditionList
		end
]]--
function LriBlobConsistencyModel.makeModel( wfCatalog, mockLocallyAvailable )

	if not LrmTHLibrary or not LrmTHLibrary.isLocallyAvailable and not mockLocallyAvailable then
		return {
			value = Rx.Observable.throw( "LrmTHLibrary not loaded or LrmTHLibrary.isLocallyAvailable not implemented" )
		}
	end

	local isLocallyAvailable = mockLocallyAvailable
	
	if not isLocallyAvailable then
		isLocallyAvailable = LrmTHLibrary.isLocallyAvailable
	end

	-- We need all fields that are required for contentId and foldernames/filenames for Originals
	local fieldSpec = {
		_id = true,
		_spaceId = true,
		_revId = true,
		_newRevId = true,
		userUpdated = true,
		captureDate = true,
		develop = {
			xmpCameraRaw = true,
			userOrientation = true
		},
		importSource = {
			sha256 = true,
			importTimestamp = true,
			fileName = true;
		},
		_localOnly = true
	}

	local missingBlobs = 0

	local consistencyCheck = enumerateCacheReferences( wfCatalog:indexStore() )
		:takeWhile( function( rows )
			return next( rows )
		end )
		:flatMap( function( rows )

			local docIds = {}

			for _, row in ipairs( rows ) do
				docIds[ #docIds + 1 ] = row[ 1 ]
			end

			if WFCloseSupport.hasBeenOrWillBeClosed( wfCatalog ) then
				-- Safeguard against premature catalog shutdown. This is most likely to
				-- show up in unit tests.
				return Rx.Observable.just( 0 )
			end

			return wfCatalog:docStore():bulkDocValue( docIds, fieldSpec )
				:map( function( strippedDocs )

					for _, strippedDoc in pairs( strippedDocs) do
						local missingRenditions = isLocallyAvailable( strippedDoc, localRenditions( strippedDoc ) )

						if missingRenditions and next( missingRenditions ) then
							missingBlobs = missingBlobs + 1
						end
					end

					return #rows

				end )
		end )
		:sum() -- sums until completed
		:defaultIfEmpty(0)
		:map( function( count )
			logger:infof( "%d of %d documents have missing blobs", missingBlobs, count )

			return { totalCount = count, missingBlobs = missingBlobs }
		end )

	return {
		value = consistencyCheck
	}

end

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

return LriBlobConsistencyModel