import { ObjectId } from 'mongodb' ; import { io } from '../socket/socket' ; import { Job } from '../../commons/types/job' ; import { Jobs } from './jobs' ; import { notifyRunnersOfAvailableJobs } from '../runners/notify-runners-of-available-jobs' ; import { Pipelines , pipelineSocketRoom } from '../pipelines/pipelines' ; import { cancelNonRunningJobs } from '../pipelines/cancel/cancel-non-running-jobs' ; import { arraySet } from '../../commons/utils/array-set' ; import { serializeJob } from './serialize-job' ; import { Logger } from '../../commons/logger/logger' ; const logger = new Logger ( 'metroline.server:addJobs' ) ; export async function addJobs ( pipelineId , jobs ) { const date = new Date ( ) ; jobs . forEach ( job => { job . createdAt = date ; } ) ; return Jobs ( ) . insertMany ( jobs ) . then ( ( ) => { arraySet ( jobs . map ( job => job . pipelineId ) ) . forEach ( id => { const pipelineJobs = jobs . filter ( job => job . pipelineId === id ) ; io . to ( pipelineSocketRoom ( id ) ) . emit ( `pipeline.${ id }.jobs` , pipelineJobs . map ( job => serializeJob ( job ) ) ) ; } ) ; } ) . then ( ( ) => Pipelines ( ) . findOne ( { _id : pipelineId } ) ) . then ( pipeline => { if ( pipeline . cancelledAt ) { setTimeout ( ( ) => { cancelNonRunningJobs ( pipelineId . toHexString ( ) ) ; } , 1000 ) ; } else { notifyRunnersOfAvailableJobs ( ) ; } } ) . catch ( err => logger . error ( 'Could not add jobs' , err ) ) ; }