import { Pipeline } from '@/services/data/tuples/pipeline-types' ; import { connectSimulatorDB } from '@/services/local-persist/db' ; import dayjs from 'dayjs' ; import { useEffect } from 'react' ; import { useRuntimeEventBus } from '../runtime/runtime-event-bus' ; import { RuntimeEventTypes } from '../runtime/runtime-event-bus-types' ; import { ChangedDataRow , PipelineRunStatus , PipelineRuntimeContext } from '../types' ; import { buildPipelineRuntimeContext } from '../utils' ; import { buildContextBody , createLogWriter } from './utils' ; import { voteNextDynamicPipeline } from './vote' ; export const useCompleted = ( context , pipelines , setMessage ) => { const { on , off , fire } = useRuntimeEventBus ( ) ; useEffect ( ( ) => { const logWrite = createLogWriter ( context , setMessage ) ; const finishPipeline = async ( ) => { await connectSimulatorDB ( ) . pipelines . update ( context . pipelineRuntimeId ! , { status : context . status , context : buildContextBody ( context ) , dataAfter : context . runtimeData , changed : context . changedData , lastModifiedAt : dayjs ( ) . toDate ( ) } ) ; let defeatedPipelines = context . defeatedPipelines || [ ] ; if ( context . changedData . length > 0 || defeatedPipelines . length > 0 ) { const previousChanged = ( context . defeatedPipelines || [ ] ) . map ( dp => dp . triggerData ) ; const newChanged = context . changedData . reduce ( ( merged , changed ) => { if ( ! merged . some ( item => item . after === changed . after ) ) { merged . push ( changed ) ; } return merged ; } , [ ] as Array < ChangedDataRow > ) ; const allMerged = [ ... new Set ( [ ... newChanged , ... previousChanged ] ) ] ; context . changedData = newChanged ; const topicIds = allMerged . map ( merge => merge . topicId ) ; let availablePipelines = pipelines . filter ( p => topicIds . includes ( p . topicId ) ) ; const availablePipelinesOnNewChanged = ( ( ) => { const topicIds = newChanged . map ( changed => changed . topicId ) ; return pipelines . filter ( p => topicIds . includes ( p . topicId ) ) ; } ) ( ) ; let concernedPipelines = availablePipelines . filter ( p => ! availablePipelinesOnNewChanged . includes ( p ) ) ; concernedPipelines = concernedPipelines . filter ( p => { const defeated = defeatedPipelines . find ( dp => dp . triggerData . topicId == p . topicId ) ; if ( ! defeated ) { return true ; } else { return ! defeated . pipelines . includes ( p . pipelineId ) ; } } ) ; availablePipelines = availablePipelines . filter ( p => ! concernedPipelines . includes ( p ) ) ; if ( availablePipelines . length === 0 ) { fire ( RuntimeEventTypes . RUN_NEXT_PIPELINE ) ; } else { const nextDynamicPipeline = voteNextDynamicPipeline ( { candidates : availablePipelines , allPipelines : pipelines } ) ; const topicId = nextDynamicPipeline . topicId ; let trigger ; const defeats = defeatedPipelines . find ( defeated => defeated . triggerData . topicId == topicId ) ; let defeatedPicked = false ; if ( defeats && defeats . pipelines . some ( pipelineId => pipelineId == nextDynamicPipeline . pipelineId ) ) { trigger = defeats . triggerData ; defeats . pipelines = defeats . pipelines . filter ( pipelineId => pipelineId != nextDynamicPipeline . pipelineId ) ; if ( defeats . pipelines . length === 0 ) { defeatedPipelines = defeatedPipelines . filter ( dp => dp !== defeats ) ; } defeatedPicked = true ; } else { trigger = newChanged . find ( changed => changed . topicId == topicId ) ! ; defeatedPicked = false ; } const dynamicPipelineContext = buildPipelineRuntimeContext ( { pipeline : nextDynamicPipeline , topic : context . allTopics [ topicId ] , triggerData : trigger . after , triggerDataOnce : trigger . before , runtimeData : context . runtimeData , allTopics : context . allTopics } ) ; dynamicPipelineContext . defeatedPipelines = [ ... defeatedPipelines , ... newChanged . map ( changed => { let availablePipelines = pipelines . filter ( p => changed . topicId == p . topicId ) ; if ( ! defeatedPicked ) { availablePipelines = availablePipelines . filter ( p => p !== nextDynamicPipeline ) ; } return { triggerData : changed , pipelines : availablePipelines . map ( p => p . pipelineId ) } ; } ) . filter ( x => x . pipelines . length > 0 ) ] ; fire ( RuntimeEventTypes . RUN_DYNAMIC_PIPELINE , dynamicPipelineContext ) ; } } else { fire ( RuntimeEventTypes . RUN_NEXT_PIPELINE ) ; } } ; const onPipelineCompeted = ( status ) => async ( c ) => { if ( c !== context ) { return ; } context . status = status ; await logWrite ( `Pipeline finished on status ${ status }.` ) ; await finishPipeline ( ) ; } ; const onPipelineIgnored = onPipelineCompeted ( PipelineRunStatus . IGNORED ) ; const onPipelineDone = onPipelineCompeted ( PipelineRunStatus . DONE ) ; const onPipelineFailed = async ( c , error ? ) => { if ( c !== context ) { return ; } context . status = PipelineRunStatus . FAIL ; await logWrite ( `Pipeline failed${ error ? ` on error[${ error . message }]` : '' }.` ) ; await finishPipeline ( ) ; } ; on ( RuntimeEventTypes . PIPELINE_IGNORED , onPipelineIgnored ) ; on ( RuntimeEventTypes . PIPELINE_DONE , onPipelineDone ) ; on ( RuntimeEventTypes . PIPELINE_FAILED , onPipelineFailed ) ; return ( ) => { off ( RuntimeEventTypes . PIPELINE_IGNORED , onPipelineIgnored ) ; off ( RuntimeEventTypes . PIPELINE_DONE , onPipelineDone ) ; off ( RuntimeEventTypes . PIPELINE_FAILED , onPipelineFailed ) ; } ; } , [ on , off , fire , context , pipelines , setMessage ] ) ; } ;