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

WFCatVideo.lua

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

ADOBE CONFIDENTIAL
___________________

Copyright 2021 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.

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

--============================================================================--
	-- Rx
local Rx = require 'Rx'

	-- WF.core.utils
local WFArguments = require 'WFArguments'
local WFPlatformInfo = require 'WFPlatformInfo'
local KSTableUtils = require 'KSTableUtils'
local WFCatFileSpec = require 'WFCatFileSpec'
local WFAnalytics = require 'WFAnalytics'
local WFRxThrottledBatchedBulkDocValue = require 'WFRxThrottledBatchedBulkDocValue'

	-- WF.web.json
local KSJsonDecoder = require 'KSJsonDecoder'
local WFURI = require 'WFURI'

	-- WF.core.logging
local WFLogger = require 'WFLogger'

	-- WF.managed_catalog
local WFCatIndexUtils = require 'WFCatIndexUtils'

	-- WF.core.utils
local WFCatIdUtils =  require 'WFCatIdUtils'
--------------------------------------------------------------------------------

local KSTableUtils_derefOpt = KSTableUtils.derefOpt
local KSTableUtils_appendArray = KSTableUtils.appendArray

local logger = WFLogger 'WFCatVideo'
--	:setLevel 'DEBUG'

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

-- @mark WFCatVideo namespace

local WFCatVideo = {}

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

local function WFAnalyticsTrackHttpError( docId, optSpaceId, contentType, generateBinaryType, url, headers, responseData )

	local errorEvent = {}

	errorEvent['event.type'] = 'error'
	errorEvent['event.appWorkflow'] = 'SYNC'
	errorEvent['event.subcategory'] = 'Generate_video_' .. generateBinaryType
	errorEvent['event.url'] = url
	errorEvent['event.error_type'] = 'generate_video_' .. generateBinaryType .. '.Error'
	errorEvent['content.id'] = docId
	errorEvent['content.spaceId'] = optSpaceId
	errorEvent['content.type'] = 'video'
	errorEvent['content.mimetype'] = contentType

	local errorCode = headers and headers.status_code or 'nil'

	if not headers then

		-- Modern native HTTP implementations put "wf_err_timeout" in the error JSON on timeout
		if (string.match( responseData, "wf_err_timeout" ) or string.match( responseData, "timed out" )) then
			errorCode = "timeout"
		end

		local annotatedReason = string.format( "%s", tostring( responseData ) )

		errorEvent['event.subtype'] = 'os'
		errorEvent['event.error_code'] = errorCode
		errorEvent['event.error_desc'] = annotatedReason

	else

		local ozErrorResponse = string.format( "%s",
									   tostring( responseData ) )

		local ozRequestId = headers['X-Request-Id']

		errorEvent['event.subtype'] = 'api'
		errorEvent['event.error_code'] = errorCode
		errorEvent['event.error_desc'] = ozErrorResponse
		errorEvent['event.context_guid'] = ozRequestId

	end

	WFAnalytics.trackEvent( errorEvent )

end

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

local function requestOzGenerateProxies( wfCatalog, assetId, optSpaceId, optSpecialProxy )

	WFArguments.check( 'WFCatalog', 'string', '?string', '?string' )

	local xGenerateProxiesValue = 'all'

	if optSpecialProxy == 'bt709' then
		xGenerateProxiesValue = '1080p709'
	end

	logger:debugf( "request oz generate proxy with assetId: %s, spaceId: %s, specialProxy: %s", assetId, optSpaceId, optSpecialProxy )

	local httpOptArgs = {
		requestHeaders = {
			{ header = 'X-Generate-Proxies', value = xGenerateProxiesValue },
	 	},
		requestMethod = 'POST',
		returnHttpErrorsAsData = true
	}

	local catalogId = wfCatalog:ozCatalogSpec():catalogId()

	local docId = assetId
	local revisionId

	return wfCatalog:docStore():docValue( assetId )

		:flatMap( function( docValue )

			local spmLinks = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'links' )
		
			-- If original is not uploaded to Oz yet, ignore this request
			
			if not ( KSTableUtils_derefOpt( spmLinks, '/rels/master', 'href' ) 
					or KSTableUtils_derefOpt( spmLinks, '/rels/master', 'exists' ) ) then
					
				logger:infof( "requestOzGenerateProxies original link does not exist. assetId:%s, spaceId:%s",
								assetId, optSpaceId )

				return Rx.Observable.just( false, 'original does not exist' )
			
			end
			
			-- If requesting special proxy and "/rels/proxy_type/1080p709" link already exists and not invalid,
			-- ignore this request

			if optSpecialProxy == 'bt709' then

				if KSTableUtils_derefOpt( spmLinks, '/rels/proxy_type/1080p709', 'href' )
						and not KSTableUtils_derefOpt( spmLinks, '/rels/proxy_type/1080p709', 'invalid' ) then

					logger:infof( "requestOzGenerateProxies link /rels/proxy_type/1080p709 already exists and valid. AssetId: %s, spaceId: %s",
									assetId, optSpaceId )

					return Rx.Observable.just( false, 'already exists' )

				end

			end

			local revisionIds = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'revision_ids' )
			revisionId = revisionIds and revisionIds[ #revisionIds ] or docValue._revId

			-- oz generate proxy files API: http://apidocs.adbecrsl.com/v2/docs/multi_page/#method_generateproxyfiles
			local generateProxiesUrl

			if optSpaceId then

				generateProxiesUrl = "/v2/spaces/" .. optSpaceId

				docId = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'catalog_doc', 'id' )
			else
				generateProxiesUrl = "/v2/catalogs/" .. catalogId
			end

			generateProxiesUrl = generateProxiesUrl .. "/assets/" .. docId .. "/revisions/" .. revisionId .. "/proxies"

			generateProxiesUrl = WFURI.stringByAppendingPathComponents( wfCatalog:ozCatalogSpec():serverBaseUrl(), generateProxiesUrl )

			return wfCatalog:ozCatalogSpec():put( generateProxiesUrl, '{}', httpOptArgs )

				:map( function( responseData, headers )

					local contentType = WFCatFileSpec.getContentTypeFromDocValue( docValue )

					if not headers or headers.status_code ~= '202' then

						WFAnalyticsTrackHttpError( docId, optSpaceId, contentType, 'proxy_' .. xGenerateProxiesValue, generateProxiesUrl, headers, responseData )

						return error( string.format( "requestOzGenerateProxies: Fail to GET Proxies files with HTTP status code %s", 
										headers and headers.status_code ) )

					end

					local ok, proxyLinksOrError = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, responseData )

					if not ok then

						WFAnalyticsTrackHttpError( docId, optSpaceId, contentType, 'proxy_' .. xGenerateProxiesValue, generateProxiesUrl, headers, proxyLinksOrError )

						return error( string.format( "requestOzGenerateProxies: Fail to decode mainParts response error: %s", 
										proxyLinksOrError ) )

					end

					-- The request API is asynchronous and returns a response before the requested proxies are generated
					if optSpecialProxy == 'bt709' then

						if KSTableUtils_derefOpt( proxyLinksOrError, 'links', '/rels/proxy_type/1080p709', 'href') then
							return true
						else
							return false
						end

					elseif KSTableUtils_derefOpt( proxyLinksOrError, 'links', '/rels/proxy_type/1080p', 'href') then
						return true
					else
						return false
					end

				end )

		end )

end

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

--[[!
	\brief Request Oz to generate proxies for existing 1.0 video
	\param wfCatalog (WFCatalog) wfCatalog
	\param assetId (string) asset ID
	\param optArgs (optional table)
		specialProxy (string) Windows only. Currently only supports 'bt709'. Special proxy name that is requested.
							If non-Windows client adds 'bt709' arg, this API would only request normal proxies
	\return (observable) Return true if Oz has already generated requested proxies.
					Otherwise, it means Oz received the request, but it would take a while to
					generate requested proxies
--]]

function WFCatVideo.generateProxies( wfCatalog, assetId, optArgs )

	WFArguments.check( 'WFCatalog', 'string', '?table' )

	optArgs = optArgs or {}

	local specialProxy

	if optArgs.specialProxy == 'bt709' and WFPlatformInfo.isWindows then
		specialProxy = optArgs.specialProxy
	end

	return requestOzGenerateProxies( wfCatalog, assetId, nil, specialProxy )

end

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

--[[!
	\brief Request Oz to generate proxies for existing 1.0 video in SWY album
	\param wfCatalog (WFCatalog) wfCatalog
	\param spaceId (string) spaceId of the SWY album of asset
	\param assetId (string) asset ID
	\param optArgs (optional table)
		specialProxy (string) Windows only. Currently only supports 'bt709'. Special proxy name that is requested.
							If non-Windows client adds 'bt709' arg, this API would only request normal proxies
	\return (observable) Return true if Oz has already generated requested proxies.
					Otherwise, it means Oz received the request, but it would take a while to
					generate requested proxies
--]]

function WFCatVideo.generateProxiesForSpaceAsset( wfCatalog, spaceId, assetId, optArgs )

	WFArguments.check( 'WFCatalog', 'string', 'string', '?table' )

	optArgs = optArgs or {}

	local specialProxy

	if optArgs.specialProxy == 'bt709' and WFPlatformInfo.isWindows then
		specialProxy = optArgs.specialProxy
	end

	return requestOzGenerateProxies( wfCatalog, assetId, spaceId, specialProxy )

end

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

--[[!
	\brief Request Oz to generate special renditions for video in SWY album
	\param wfCatalog (WFCatalog) wfCatalog
	\param spaceId (string) spaceId of the SWY album of asset
	\param assetId (string) asset ID
	\param specialRendition (string) Windows only. Currently only supports 'bt709'. Special rendition name that is requested
	\return (observable) Return true if Oz has already generated requested proxies.
					Otherwise, it means Oz received the request, but it would take a while to
					generate requested proxies.
					If non-windows device calls this API, false would be returned without actions.

--]]

function WFCatVideo.generateSpecialRenditionForSpaceAsset( wfCatalog, spaceId, assetId, specialRendition )

	WFArguments.check( 'WFCatalog', 'string', 'string', 'string' )

	if not ( specialRendition == 'bt709' and WFPlatformInfo.isWindows ) then

		logger:warnf( "generateSpecialRenditionsForSpaceAsset only supports bt709 for windowns device" )
		return Rx.Observable.just( false )

	end

	logger:debugf( "request oz generate special reditions with assetId: %s, spaceId: %s, specialRendition: %s", assetId, spaceId, specialRendition )

	local xGenerateRenditionsValue = '1080p709'

	local httpOptArgs = {
		requestHeaders = {
			{ header = 'X-Generate-Renditions', value = xGenerateRenditionsValue },
	 	},
		requestMethod = 'POST',
		returnHttpErrorsAsData = true
	}

	local revisionId

	return wfCatalog:docStore():docValue( assetId )

		:flatMap( function( docValue )

			local spmLinks = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'links' )
		
			-- If original is not uploaded to Oz yet, ignore this request
			
			if not ( KSTableUtils_derefOpt( spmLinks, '/rels/master', 'href' ) 
					or KSTableUtils_derefOpt( spmLinks, '/rels/master', 'exists' ) ) then
					
				logger:infof( "generateSpecialRenditionsForSpaceAsset original link does not exist. assetId:%s, spaceId:%s",
								assetId, spaceId )

				return Rx.Observable.just( false, 'original does not exist' )
			
			end
		
			-- If /rels/rendition_type/1080p709 link already exists and not invalid, ignore this request
			
			if KSTableUtils_derefOpt( spmLinks, '/rels/rendition_type/1080p709', 'href' )
					and not KSTableUtils_derefOpt( spmLinks, '/rels/rendition_type/1080p709', 'invalid' ) then

				logger:infof( "generateSpecialRenditionsForSpaceAsset link /rels/rendition_type/1080p709 already exists and valid. assetId:%s, spaceId:%s",
								assetId, spaceId )

				return Rx.Observable.just( false, 'already exists' )

			end

			local revisionIds = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'revision_ids' )
			revisionId = revisionIds and revisionIds[ #revisionIds ] or docValue._revId

			-- oz generate rendition files API: http://apidocs.adbecrsl.com/v2/docs/multi_page/#method_generaterenditionfiles
			local generateRenditionsUrl = "/v2/spaces/" .. spaceId

			-- For space API, we need to use catalog_doc.id as assetId
			assetId = KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'catalog_doc', 'id' )

			generateRenditionsUrl = generateRenditionsUrl .. "/assets/" .. assetId .. "/revisions/" .. revisionId .. "/renditions"

			generateRenditionsUrl = WFURI.stringByAppendingPathComponents( wfCatalog:ozCatalogSpec():serverBaseUrl(), generateRenditionsUrl )

			return wfCatalog:ozCatalogSpec():put( generateRenditionsUrl, '{}', httpOptArgs )

				:map( function( responseData, headers )

					local contentType = WFCatFileSpec.getContentTypeFromDocValue( docValue )

					if not headers or headers.status_code ~= '202' then

						WFAnalyticsTrackHttpError( assetId, spaceId, contentType, 'rendition_' .. xGenerateRenditionsValue, generateRenditionsUrl, headers, responseData )

						return error( string.format( "generateSpecialRenditionsForSpaceAsset: Fail to GET Renditions files with HTTP status code %s", 
										headers and headers.status_code ) )

					end

					local ok, renditionLinksOrError = pcall( KSJsonDecoder.decodeWebSafeJson_stripNull, responseData )

					if not ok then

						WFAnalyticsTrackHttpError( assetId, spaceId, contentType, 'rendition_' .. xGenerateRenditionsValue, generateRenditionsUrl, headers, renditionLinksOrError )

						return error( string.format( "generateSpecialRenditionsForSpaceAsset: Fail to decode mainParts response error: %s", 
										renditionLinksOrError ) )

					end

					-- The request API is asynchronous and returns a response before the requested proxies are generated
					if KSTableUtils_derefOpt( renditionLinksOrError, 'links', '/rels/rendition_type/1080p709', 'href') then
						return true
					else
						return false
					end

				end )

		end )

end



--------------------------------------------------------------------------------
--[[!
	\brief Request Oz to generate special proxy or renditions for the given asset id.
	\param wfCatalog (WFCatalog) wfCatalog
	\param localId (string) local asset ID
	\param spaceId (string) spaceId of the SWY album of asset
	\param isSharedAlbum (boolean) is the current request in a sharedAlbum
	\param canEditInAlbum (boolean) is the current request in an editable album
--]]

function WFCatVideo.requestSpecialProxyOrRenditionGenerationForVideoStream( wfCatalog, localId, spaceId, isSharedAlbum, canEditInAlbum )

	return WFCatIdUtils.getOzAssetIdsForLocalIds( wfCatalog, { localId } )
		:flatMap( function( ozIds )
		
			local ozId = ozIds and ozIds[ 1 ]

			if not ozId then
				return Rx.Observable.just()
			end

			local specialAssetValue = 'bt709'
			local generateProxyArgs = { specialProxy = specialAssetValue }
			
			if not isSharedAlbum then
				return WFCatVideo.generateProxies( wfCatalog, ozId, generateProxyArgs )
			end
			
			if not spaceId then
				return Rx.Observable.just() -- Shared album without spaceId 
			end

			if canEditInAlbum then
				return WFCatVideo.generateProxiesForSpaceAsset( wfCatalog, spaceId, ozId, generateProxyArgs)
			else
				return WFCatVideo.generateSpecialRenditionForSpaceAsset( wfCatalog, spaceId, ozId, specialAssetValue )
			end
			
		end )
	
end

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

if WFPlatformInfo.isMobile() then

local function queryStoreLocallyVideoStream( wfCatalog, optCalm )

	local sql = [[SELECT DISTINCT coreInfo.docId
				FROM coreInfo
					INNER JOIN album_asset ON album_asset.assetId == coreInfo.docId
					INNER JOIN albums ON albums.docId = album_asset.albumId
					LEFT JOIN space ON space.docId = albums.spaceId
					LEFT JOIN space_user ON space.spaceUserId = space_user.fullDocId
				WHERE albums.useOffline == 1
					AND ( albums.spaceId IS NULL OR space_user.role == 'editor' )
					AND coreInfo.isVideo == true]]

	return wfCatalog:indexStore():query{
		sql = sql,
		singleColumnQuery = true,
		calm = optCalm or 5,
		quiesceUntilUpToDate = true,
	}

end

WFCatVideo.DEBUG_queryStoreLocallyVideoStream = queryStoreLocallyVideoStream

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

local function getNeedGenerateProxiesAssetsInfo( wfCatalog, assetIds )

	if next( assetIds ) then

		local includedDocFieldSpec = {
			_id = true,
			type = true,
			subtype = true,
			_spaceId = true,
			_serverProvidedMetadata = {
				links = { ['/rels/proxy_type/1080p'] = { href = true } }
			}
		}

		return WFRxThrottledBatchedBulkDocValue( "WFCatVideo_filterAndRequestProxies", wfCatalog:docStore(), assetIds, 100, 10, includedDocFieldSpec )
			:map( function( assetDocs )

				local needGenerateProxiesAssetsInfo = {}

				for _, docValue in ipairs( assetDocs ) do

					if WFCatIndexUtils.isVideoDocument( docValue ) and
							not KSTableUtils_derefOpt( docValue, '_serverProvidedMetadata', 'links', '/rels/proxy_type/1080p', 'href' ) then

						local assetInfoTable = { docValue._id, docValue._spaceId }
						needGenerateProxiesAssetsInfo[ #needGenerateProxiesAssetsInfo + 1 ] = assetInfoTable

					end

				end

				return needGenerateProxiesAssetsInfo

			end )
			:scan( function( needGenerateProxiesAssetsInfo, batchOfNeedGenerateProxiesAssetsInfo )
				needGenerateProxiesAssetsInfo = KSTableUtils_appendArray( needGenerateProxiesAssetsInfo, batchOfNeedGenerateProxiesAssetsInfo )
				return needGenerateProxiesAssetsInfo
			end, {} )

	end

	return Rx.Observable.just( {} )

end

WFCatVideo.DEBUG_getNeedGenerateProxiesAssetsInfo = getNeedGenerateProxiesAssetsInfo

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

local function generateProxiesForAsetIds( wfCatalog, needGenerateProxiesAssetsInfo )

	if next( needGenerateProxiesAssetsInfo ) then

		return Rx.Observable.from( needGenerateProxiesAssetsInfo )
			:flatMap( function( assetInfoTable )

				local assetId = KSTableUtils_derefOpt( assetInfoTable, 1 )
				local spaceId = KSTableUtils_derefOpt( assetInfoTable, 2 )

				if spaceId then
					return  WFCatVideo.generateProxiesForSpaceAsset( wfCatalog, spaceId, assetId )
				end

				return WFCatVideo.generateProxies( wfCatalog, assetId )
			end)
			:toArray()

	end

	return Rx.Observable.just( true )

end

WFCatVideo.DEBUG_generateProxiesForAsetIds = generateProxiesForAsetIds

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

function WFCatVideo.generateProxiesForStoreLocally( wfCatalog )

	local requestedAssetIdList = {}

	return queryStoreLocallyVideoStream( wfCatalog )

		:distinctUntilChanged()

		:map( function( assetIds )

			-- check if there's any assetIds that has been requested generate proxies
			local filteredAssetIds = {}

			for _, assetId in ipairs( assetIds ) do
				if not requestedAssetIdList[ assetId ] then

					filteredAssetIds[ #filteredAssetIds + 1 ] = assetId
					requestedAssetIdList[ assetId ] = true

				end
			end

			return filteredAssetIds

		end )

		:filter( function( filteredAssetIds )
			return #filteredAssetIds > 0
		end )

		:flatMap( function( filteredAssetIds )
			return getNeedGenerateProxiesAssetsInfo( wfCatalog, filteredAssetIds )
		end )

		:flatMap( function( needGenerateProxiesAssetsInfo )
			return generateProxiesForAsetIds( wfCatalog, needGenerateProxiesAssetsInfo )
		end )

end

end

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

return WFCatVideo
