import { isReadTopicAction , isWriteTopicAction } from '@/services/data/tuples/pipeline-stage-unit-action/pipeline-stage-unit-action-utils' ; import { Pipeline } from '@/services/data/tuples/pipeline-types' ; import { TopicId } from '@/services/data/tuples/topic-types' ; interface PipelineIO { pipeline : Pipeline ; read : Array < TopicId > ; write : Array < TopicId > ; trails : Array < Pipeline > ; } const findNextTrails = ( from , breakOut , ioMap ) : Array < Pipeline > => { return from . trails . map ( trail => { const nextTrails = ioMap . get ( trail . pipelineId ) ! . trails ; if ( nextTrails . length === 0 ) { return [ ] ; } else { return nextTrails . filter ( x => x !== breakOut ) . map ( trail => ioMap . get ( trail . pipelineId ) ! ) . map ( io => findNextTrails ( io , breakOut , ioMap ) ) . flat ( ) ; } } ) . flat ( ) ; } ; export const voteNextDynamicPipeline = ( options ) : Pipeline => { const { candidates , allPipelines } = options ; const topicPipelineMap = allPipelines . reduce ( ( map , pipeline ) => { let pipelines = map . get ( pipeline . topicId ) ; if ( ! pipelines ) { pipelines = [ ] ; map . set ( pipeline . topicId , pipelines ) ; } pipelines . push ( pipeline ) ; return map ; } , new Map < string , Array < Pipeline > > ( ) ) ; const pipelineIOs = allPipelines . map ( pipeline => { const io = { pipeline , read : [ ] , write : [ ] , trails : [ ] } ; pipeline . stages . forEach ( stage => { stage . units . forEach ( unit => { unit . do . forEach ( action => { if ( isReadTopicAction ( action ) && action . topicId != pipeline . topicId ) { io . read . push ( action . topicId ) ; } else if ( isWriteTopicAction ( action ) && action . topicId != pipeline . topicId ) { io . write . push ( action . topicId ) ; io . trails . push ( ... ( topicPipelineMap . get ( action . topicId ) || [ ] ) ) ; } } ) ; } ) ; } ) ; io . read = [ ... new Set ( io . read . filter ( x => x != null && x . trim ( ) . length !== 0 && x != pipeline . topicId ) ) ] ; io . write = [ ... new Set ( io . write . filter ( x => x != null && x . trim ( ) . length !== 0 && x != pipeline . topicId ) ) ] ; io . trails = [ ... new Set ( io . trails . filter ( x => x != null && x !== pipeline ) ) ] ; return io ; } ) ; const ioMap = pipelineIOs . reduce ( ( map , io ) => { map . set ( io . pipeline . pipelineId , io ) ; return map ; } , new Map ( ) ) ; pipelineIOs . forEach ( io => { const nextTrails = findNextTrails ( io , io . pipeline , ioMap ) ; io . trails = [ ... new Set ( [ ... io . trails , ... nextTrails ] ) ] ; } ) ; const topicWriters = pipelineIOs . reduce ( ( map , io ) => { io . write . forEach ( topicId => { let writers = map . get ( topicId ) ! ; if ( ! writers ) { writers = [ ] ; map . set ( topicId , writers ) ; writers . push ( io ) ; } else if ( ! writers . includes ( io ) ) { writers . push ( io ) ; } } ) ; return map ; } , new Map < string , Array < PipelineIO > > ( ) ) ; const picked = candidates . find ( pipeline => { const otherPipelines = candidates . filter ( p => p !== pipeline ) ; const io = ioMap . get ( pipeline . pipelineId ) ! ; const dependTopicIds = io . read ; const writers = dependTopicIds . map ( topicId => topicWriters . get ( topicId ) || [ ] ) . flat ( ) ; const writePipelines = writers . map ( writer => writer . pipeline ) ; const found = otherPipelines . some ( one => writePipelines . includes ( one ) ) ; if ( found ) { return false ; } const allTrails = otherPipelines . map ( pipeline => ioMap . get ( pipeline . pipelineId ) ?. trails ?? [ ] ) . flat ( ) . filter ( p => p !== pipeline ) ; return allTrails . some ( one => writePipelines . includes ( one ) ) ; } ) ; return picked ?? candidates [ 0 ] ; } ;