import SubscriptionClass , { Repositories , Repository , RepositoryData , SyncStatus } from "../models/subscription" ; import { RepoSyncState , Subscription } from "../models" ; import getJiraClient from "../jira/client" ; import { getRepositorySummary } from "./jobs" ; import enhanceOctokit from "../config/enhance-octokit" ; import statsd from "../config/statsd" ; import getPullRequests from "./pull-request" ; import getBranches from "./branches" ; import getCommits from "./commits" ; import { Application , GitHubAPI } from "probot" ; import { metricSyncStatus , metricTaskStatus } from "../config/metric-names" ; import { booleanFlag , BooleanFlags , isBlocked } from "../config/feature-flags" ; import { LoggerWithTarget } from "probot/lib/wrap-logger" ; import { Deduplicator , DeduplicatorResult , RedisInProgressStorageWithTimeout } from "./deduplicator" ; import Redis from "ioredis" ; import getRedisInfo from "../config/redis-info" ; import GitHubClient from "../github/client/github-client" ; import { BackfillMessagePayload } from "../sqs/backfill" ; import { Hub } from "@sentry/types/dist/hub" ; import sqsQueues from "../sqs/queues" ; import { getCloudInstallationId } from "../github/client/installation-id" ; const tasks = { pull : getPullRequests , branch : getBranches , commit : getCommits } ; interface TaskProcessors { [ task ] : ( logger : LoggerWithTarget , github : GitHubAPI , newGithub : GitHubClient , jiraHost : string , repository : Repository , cursor ? : string | number , perPage ? : number ) => Promise < { edges : any [ ] ; jiraPayload : any ; } > ; } type TaskType = "pull" | "commit" | "branch" ; const taskTypes = Object . keys ( tasks ) as TaskType [ ] ; export const sortedRepos = ( repos ) : [ string , RepositoryData ] [ ] => Object . entries ( repos ) . sort ( ( a , b ) => new Date ( b [ 1 ] . repository ?. updated_at || 0 ) . getTime ( ) - new Date ( a [ 1 ] . repository ?. updated_at || 0 ) . getTime ( ) ) ; const getNextTask = async ( subscription ) : Promise < Task | undefined > => { const repos = await RepoSyncState . findAllFromSubscription ( subscription , { order : [ [ "repoUpdatedAt" , "DESC" ] ] } ) ; const sorted = repos . map ( repo => [ repo . repoId . toString ( ) , repo . toRepositoryData ( ) ] ) ; for ( const [ repositoryId , repoData ] of sorted ) { const task = taskTypes . find ( ( taskType ) => repoData [ getStatusKey ( taskType ) ] === undefined || repoData [ getStatusKey ( taskType ) ] === "pending" ) ; if ( ! task ) continue ; const { repository , [ getCursorKey ( task ) ] : cursor } = repoData ; return { task , repositoryId , repository : repository as Repository , cursor : cursor as any } ; } return undefined ; } ; export interface Task { task : TaskType ; repositoryId : string ; repository : Repository ; cursor ? : string | number ; } const upperFirst = ( str ) => str . substring ( 0 , 1 ) . toUpperCase ( ) + str . substring ( 1 ) ; const getCursorKey = ( type ) => `last${ upperFirst ( type ) }Cursor` ; const getStatusKey = ( type ) => `${ type }Status` ; const updateJobStatus = async ( data , edges , task , repositoryId , logger , scheduleNextTask ) => { const { installationId , jiraHost } = data ; const subscription = await Subscription . getSingleInstallation ( jiraHost , installationId ) ; if ( ! subscription ) { logger . info ( "Organization has been deleted. Other active syncs will continue." ) ; return ; } const status = edges ?. length ? "pending" : "complete" ; logger . info ( { status } , "Updating job status" ) ; await subscription . updateRepoSyncStateItem ( repositoryId , getStatusKey ( task ) , status ) ; if ( edges ?. length ) { await subscription . updateRepoSyncStateItem ( repositoryId , getCursorKey ( task ) , edges [ edges . length - 1 ] . cursor ) ; scheduleNextTask ( 0 ) ; } else if ( ! ( await getNextTask ( subscription ) ) ) { await subscription . update ( { syncStatus : SyncStatus . COMPLETE } ) ; const endTime = Date . now ( ) ; const startTime = data ?. startTime || 0 ; const timeDiff = startTime ? endTime - Date . parse ( startTime ) : 0 ; if ( startTime ) { statsd . histogram ( metricSyncStatus . fullSyncDuration , timeDiff ) ; } logger . info ( { startTime , endTime , timeDiff } , "Sync status is complete" ) ; } else { logger . info ( "Sync status is pending" ) ; scheduleNextTask ( 0 ) ; } } ; const getEnhancedGitHub = async ( app , installationId ) => enhanceOctokit ( await app . auth ( installationId ) ) ; param err export const isRetryableWithSmallerRequest = ( err ) : boolean => { if ( err . errors ) { const retryableErrors = err . errors . filter ( ( error ) => { return "MAX_NODE_LIMIT_EXCEEDED" == error . type || error . message ?. startsWith ( "Something went wrong while executing your query" ) ; } ) ; return retryableErrors . length ; } else { return false ; } } ; export const isNotFoundError = ( err , logger ) : boolean | undefined => { const isNotFoundErrorType = err ?. errors && err . errors ?. filter ( ( error ) => error . type === "NOT_FOUND" ) ; const isNotFoundError = isNotFoundErrorType ?. length > 0 || err ?. status === 404 ; isNotFoundError && logger . info ( "Repository deleted after discovery, skipping initial sync" ) ; return isNotFoundError ; } ; async function doProcessInstallation ( app , data , sentry , installationId , jiraHost , logger , scheduleNextTask ) { const subscription = await Subscription . getSingleInstallation ( jiraHost , installationId ) ; if ( ! subscription ) return ; const jiraClient = await getJiraClient ( subscription . jiraHost , installationId , logger ) ; const newGithub = new GitHubClient ( getCloudInstallationId ( installationId ) , logger ) ; const github = await getEnhancedGitHub ( app , installationId ) ; const nextTask = await getNextTask ( subscription ) ; if ( ! nextTask ) { await subscription . update ( { syncStatus : "COMPLETE" } ) ; statsd . increment ( metricSyncStatus . complete ) ; logger . info ( "Sync complete" ) ; return ; } await subscription . update ( { syncStatus : "ACTIVE" } ) ; const { task , repositoryId , cursor } = nextTask ; let { repository } = nextTask ; if ( ! repository ) { const { data : repo } = await github . request ( "GET /repositories/:id" , { id : repositoryId } ) ; repository = getRepositorySummary ( repo ) ; await subscription . updateSyncState ( { repos : { [ repository . id ] : { repository } } } ) ; } logger . info ( { task : nextTask } , "Starting task" ) ; const processor = tasks [ task ] ; const execute = async ( ) => { if ( await booleanFlag ( BooleanFlags . SIMPLER_PROCESSOR , true ) ) { return await processor ( logger , github , newGithub , jiraHost , repository , cursor , 20 ) ; } else { for ( const perPage of [ 20 , 10 , 5 , 1 ] ) { try { return await processor ( logger , github , newGithub , jiraHost , repository , cursor , perPage ) ; } catch ( err ) { logger . error ( { err , payload : data , github , repository , cursor , task } , `Error processing job with page size ${ perPage }, retrying with next smallest page size` ) ; if ( isRetryableWithSmallerRequest ( err ) ) { continue ; } else { throw err ; } } } } throw new Error ( `Error processing GraphQL query: installationId=${ installationId }, repositoryId=${ repositoryId }, task=${ task }` ) ; } ; try { const { edges , jiraPayload } = await execute ( ) ; if ( jiraPayload ) { try { await jiraClient . devinfo . repository . update ( jiraPayload , { preventTransitions : true } ) ; } catch ( err ) { if ( err ?. response ?. status === 400 ) { sentry . setExtra ( "Response body" , err . response . data . errorMessages ) ; sentry . setExtra ( "Jira payload" , err . response . data . jiraPayload ) ; } if ( err . request ) { sentry . setExtra ( "Request" , { host : err . request . domain , path : err . request . path , method : err . request . method } ) ; } if ( err . response ) { sentry . setExtra ( "Response" , { status : err . response . status , statusText : err . response . statusText , body : err . response . body } ) ; } throw err ; } } await updateJobStatus ( data , edges , task , repositoryId , logger , scheduleNextTask ) ; statsd . increment ( metricTaskStatus . complete , [ `type: ${ nextTask . task }` ] ) ; } catch ( err ) { const rateLimit = Number ( err ?. headers ?. [ "x-ratelimit-reset" ] ) ; const delay = Math . max ( Date . now ( ) - rateLimit * 1000 , 0 ) ; if ( delay ) { logger . info ( { delay } , `Delaying job for ${ delay }ms` ) ; scheduleNextTask ( delay ) ; return ; } if ( String ( err ) . includes ( "connect ETIMEDOUT" ) ) { logger . warn ( "ETIMEDOUT error, retrying in 5 seconds" ) ; scheduleNextTask ( 5000 ) ; return ; } if ( String ( err . message ) . includes ( "You have triggered an abuse detection mechanism" ) ) { logger . warn ( "Abuse detection triggered. Retrying in 60 seconds" ) ; scheduleNextTask ( 60000 ) ; return ; } if ( isNotFoundError ( err , logger ) ) { const edgesLeft = [ ] ; await updateJobStatus ( data , edgesLeft , task , repositoryId , logger , scheduleNextTask ) ; return ; } const host = subscription . jiraHost || "none" ; logger . warn ( { err , jiraHost : host } , "Task failed, continuing with next task" ) ; await subscription . updateRepoSyncStateItem ( nextTask . repositoryId , getStatusKey ( nextTask . task as TaskType ) , "failed" ) ; statsd . increment ( metricTaskStatus . failed , [ `type: ${ nextTask . task }` ] ) ; scheduleNextTask ( 0 ) ; } } export async function maybeScheduleNextTask ( jobData , nextTaskDelaysMs , logger ) { if ( nextTaskDelaysMs . length > 0 ) { nextTaskDelaysMs . sort ( ) . reverse ( ) ; if ( nextTaskDelaysMs . length > 1 ) { logger . warn ( "Multiple next jobs were scheduled, scheduling one with the highest priority" ) ; } const delayMs = nextTaskDelaysMs . shift ( ) ! ; logger . info ( "Scheduling next job with a delay = " + delayMs ) ; await sqsQueues . backfill . sendMessage ( jobData , ( delayMs || 0 ) / 1000 , logger ) ; } } const redis = new Redis ( getRedisInfo ( "installations-in-progress" ) ) ; const RETRY_DELAY_BASE_SEC = 60 ; export const processInstallation = ( app ) => { const inProgressStorage = new RedisInProgressStorageWithTimeout ( redis ) ; const deduplicator = new Deduplicator ( inProgressStorage , 1000 ) ; return async ( data , sentry , logger ) : Promise < void > => { const { installationId , jiraHost } = data ; try { if ( await isBlocked ( installationId , logger ) ) { logger . warn ( "blocking installation job" ) ; return ; } sentry . setUser ( { gitHubInstallationId : installationId , jiraHost } ) ; const nextTaskDelaysMs = [ ] ; const result = await deduplicator . executeWithDeduplication ( "i-" + installationId + "-" + jiraHost , ( ) => doProcessInstallation ( app , data , sentry , installationId , jiraHost , logger , ( delay ) => nextTaskDelaysMs . push ( delay ) ) ) ; switch ( result ) { case DeduplicatorResult . E_OK : logger . info ( "Job was executed by deduplicator" ) ; maybeScheduleNextTask ( data , nextTaskDelaysMs , logger ) ; break ; case DeduplicatorResult . E_NOT_SURE_TRY_AGAIN_LATER : { logger . warn ( "Possible duplicate job was detected, rescheduling" ) ; await sqsQueues . backfill . sendMessage ( data , RETRY_DELAY_BASE_SEC , logger ) ; break ; } case DeduplicatorResult . E_OTHER_WORKER_DOING_THIS_JOB : { logger . warn ( "Duplicate job was detected, rescheduling" ) ; await sqsQueues . backfill . sendMessage ( data , RETRY_DELAY_BASE_SEC + RETRY_DELAY_BASE_SEC * Math . random ( ) , logger ) ; break ; } } } catch ( err ) { logger . warn ( { err } , "Process installation failed" ) ; } } ; } ;