import { getStoreKey , pluginName , getStorageSecretUrl , getOperationManifestUrl , } from './common' ; import loglevel from 'loglevel' ; import fetcher from 'make-fetch-happen' ; import { HttpRequestCache } from './cache' ; import type { InMemoryLRUCache } from 'apollo-server-caching' ; import type { OperationManifest } from './ApolloServerPluginOperationRegistry' ; import type { Logger , ApolloConfig , WithRequired } from 'apollo-server-types' ; import type { Response , RequestInit , fetch } from 'apollo-server-env' ; const DEFAULT_POLL_SECONDS = 30 ; const SYNC_WARN_TIME_SECONDS = 60 ; export interface AgentOptions { logger ? : Logger ; fetcher ? : typeof fetch ; pollSeconds ? : number ; apollo : WithRequired < ApolloConfig , 'keyHash' | 'graphRef' > ; store : InMemoryLRUCache ; } type SignatureStore = Set < string > ; const callToAction = `Ensure this server's schema has been published with 'apollo service:push' and that operations have been registered with 'apollo client:push'.` ; export default class Agent { private fetcher : typeof fetch ; private timer ? : NodeJS . Timer ; private logger : Logger ; private requestInFlight : Promise < any > | null = null ; private lastSuccessfulCheck ? : Date ; private storageSecret ? : string ; public _timesChecked : number = 0 ; private lastOperationSignatures : SignatureStore = new Set ( ) ; private readonly options : AgentOptions = Object . create ( null ) ; private readonly graphId : string ; readonly graphVariant : string ; constructor ( options ) { Object . assign ( this . options , options ) ; const { graphRef } = this . options . apollo ; const at = graphRef . indexOf ( '@' ) ; if ( at === - 1 ) { this . graphId = graphRef ; this . graphVariant = 'current' ; } else { this . graphId = graphRef . substring ( 0 , at ) ; this . graphVariant = graphRef . substring ( at + 1 ) ; } this . logger = this . options . logger || loglevel . getLogger ( pluginName ) ; this . fetcher = this . options . fetcher || getDefaultGcsFetcher ( ) ; } async requestPending ( ) { return this . requestInFlight ; } private pollSeconds ( ) { return this . options . pollSeconds || DEFAULT_POLL_SECONDS ; } async start ( ) { this . logger . debug ( 'Starting operation registry agent...' ) ; const pulse = async ( ) => await this . checkForUpdate ( ) ; try { await pulse ( ) ; } catch ( err ) { console . error ( 'The operation manifest could not be fetched. Retries will continue, but requests will be forbidden until the manifest is fetched.' , ( err as Error ) . message || err ) ; } this . timer = this . timer || setInterval ( function ( ) { pulse ( ) . catch ( ( err ) => console . error ( err . message || err ) ) ; } , this . pollSeconds ( ) * 1000 ) ; this . timer . unref ( ) ; } public stop ( ) { if ( this . timer ) { clearInterval ( this . timer ) ; } } private timeSinceLastSuccessfulCheck ( ) { if ( ! this . lastSuccessfulCheck ) { return - Infinity ; } return new Date ( ) . getTime ( ) - this . lastSuccessfulCheck . getTime ( ) ; } private warnWhenLossOfSync ( ) { if ( this . timeSinceLastSuccessfulCheck ( ) > SYNC_WARN_TIME_SECONDS * 1000 ) { console . warn ( `WARNING: More than ${ SYNC_WARN_TIME_SECONDS } seconds has elapsed since a successful fetch of the manifest. (Last success: ${ this . lastSuccessfulCheck })` ) ; } } private async fetchAndUpdateStorageSecret ( ) { const storageSecretUrl = getStorageSecretUrl ( this . graphId , this . options . apollo . keyHash ) ; const response = await this . fetcher ( storageSecretUrl , this . fetchOptions ) ; if ( response . status === 304 ) { this . logger . debug ( 'The storage secret was the same as the previous attempt.' ) ; return this . storageSecret ; } if ( ! response . ok ) { const responseText = await response . text ( ) ; this . logger . debug ( `Could not fetch storage secret ${ responseText }` ) ; return ; } this . storageSecret = await response . json ( ) ; return this . storageSecret ; } private fetchOptions : RequestInit = { timeout : this . pollSeconds ( ) * 3 * 1000 , } ; private async fetchManifest ( ) { this . logger . debug ( `Checking for storageSecret` ) ; const storageSecret = await this . fetchAndUpdateStorageSecret ( ) ; if ( ! storageSecret ) { throw new Error ( 'No storage secret found' ) ; } const storageSecretManifestUrl = getOperationManifestUrl ( this . graphId , storageSecret , this . graphVariant ) ; this . logger . debug ( `Checking for manifest changes at ${ storageSecretManifestUrl }` ) ; const response = await this . fetcher ( storageSecretManifestUrl , this . fetchOptions ) ; if ( response . status === 404 || response . status === 403 ) { throw new Error ( `No manifest found for tag "${ this . graphVariant }" at ` + `${ storageSecretManifestUrl }. ${ callToAction }` ) ; } return response ; } private async tryUpdate ( ) { this . _timesChecked ++ ; let response ; try { response = await this . fetchManifest ( ) ; if ( response . status === 304 ) { return false ; } if ( ! response . ok ) { const responseText = await response . text ( ) ; if ( responseText . includes ( '<Code>AccessDenied</Code>' ) ) { throw new Error ( `No manifest found. ${ callToAction }` ) ; } throw new Error ( `Unexpected status: ${ responseText }` ) ; } const contentType = response . headers . get ( 'content-type' ) ; if ( contentType && contentType !== 'application/json' ) { throw new Error ( `Unexpected 'Content-Type' header: ${ contentType }` ) ; } } catch ( err ) { const ourErrorPrefix = `Unable to fetch operation manifest for graph ID '${ this . graphId }': ${ err }` ; ( err as Error ) . message = `${ ourErrorPrefix }: ${ err }` ; throw err ; } await this . updateManifest ( await response . json ( ) ) ; return true ; } public async checkForUpdate ( ) { this . warnWhenLossOfSync ( ) ; if ( this . requestInFlight ) { return this . requestInFlight ; } this . requestInFlight = this . tryUpdate ( ) ; const resetRequestInFlight = ( ) => ( this . requestInFlight = null ) ; return this . requestInFlight . then ( ( result ) => { this . lastSuccessfulCheck = new Date ( ) ; resetRequestInFlight ( ) ; return result ; } ) . catch ( ( err ) => { resetRequestInFlight ( ) ; throw err ; } ) ; } public async updateManifest ( manifest ) { if ( ! manifest || manifest . version !== 2 || ! Array . isArray ( manifest . operations ) ) { throw new Error ( 'Invalid manifest format.' ) ; } const incomingOperations = new Map ( ) ; const replacementSignatures = new Set ( ) ; for ( const { signature , document } of manifest . operations ) { incomingOperations . set ( signature , document ) ; replacementSignatures . add ( signature ) ; if ( ! this . lastOperationSignatures . has ( signature ) ) { this . logger . debug ( `Incoming manifest ADDs: ${ signature }` ) ; this . options . store . set ( getStoreKey ( signature ) , document ) ; } } for ( const signature of this . lastOperationSignatures ) { if ( ! incomingOperations . has ( signature ) ) { this . logger . debug ( `Incoming manifest REMOVEs: ${ signature }` ) ; this . options . store . delete ( getStoreKey ( signature ) ) ; } } this . lastOperationSignatures = replacementSignatures ; } } const GCS_RETRY_COUNT = 5 ; function getDefaultGcsFetcher ( ) { return fetcher . defaults ( { cacheManager : new HttpRequestCache ( ) , headers : { 'user-agent' : [ require ( '../package.json' ) . name , require ( '../package.json' ) . version , ] . join ( '/' ) , } , retry : { retries : GCS_RETRY_COUNT , factor : 2 , minTimeout : 1000 , randomize : true , } , } ) ; }