import { O } from 'ts-toolbelt' ; import { isPromise } from 'is-what' ; import { WriteLock } from '../Magnetar' ; import { handleStream } from './handleStream' ; import { ActionConfig , MagnetarStreamAction } from '../types/actions' ; import { ActionType } from '../types/actionsInternal' ; import { StreamResponse , DoOnStreamFns , isDoOnStream , DoOnStream } from '../types/plugins' ; import { getEventNameFnsMap } from '../types/events' ; import { getModifyPayloadFnsMap } from '../types/modifyPayload' ; import { getModifyReadResponseFnsMap } from '../types/modifyReadResponse' ; import { executeOnFns } from '../helpers/executeOnFns' ; import { throwOnIncompleteStreamResponses , throwIfNoFnsToExecute } from '../helpers/throwFns' ; import { ModuleConfig , GlobalConfig } from '../types/config' ; import { getPluginModuleConfig } from '../helpers/moduleHelpers' ; import { getCollectionWriteLocks } from '../helpers/pathHelpers' ; export function handleStreamPerStore ( [ collectionPath , docId ] , moduleConfig , globalConfig , actionType , streaming , cacheStream , writeLockMap ) { return async function ( payload ? , actionConfig = { } ) : Promise < void > { const foundStream = streaming ( ) ; if ( isPromise ( foundStream ) ) return foundStream ; const writeLock = docId ? writeLockMap . get ( `${ collectionPath }/${ docId }` ) ! : writeLockMap . get ( collectionPath ) ! ; if ( isPromise ( writeLock . promise ) ) await writeLock . promise ; if ( ! docId ) { const collectionWriteLocks = getCollectionWriteLocks ( collectionPath , writeLockMap ) ; await Promise . allSettled ( collectionWriteLocks . map ( ( w ) => w . promise ) ) ; } const eventNameFnsMap = getEventNameFnsMap ( globalConfig . on , moduleConfig . on , actionConfig . on ) ; const modifyPayloadFnsMap = getModifyPayloadFnsMap ( globalConfig . modifyPayloadOn , moduleConfig . modifyPayloadOn , actionConfig . modifyPayloadOn ) ; const modifyReadResponseMap = getModifyReadResponseFnsMap ( globalConfig . modifyReadResponseOn , moduleConfig . modifyReadResponseOn , actionConfig . modifyReadResponseOn ) ; const storesToExecute = actionConfig . executionOrder || ( moduleConfig . executionOrder || { } ) [ 'stream' ] || ( moduleConfig . executionOrder || { } ) [ actionType ] || ( globalConfig . executionOrder || { } ) [ 'stream' ] || ( globalConfig . executionOrder || { } ) [ actionType ] || [ ] ; throwIfNoFnsToExecute ( storesToExecute ) ; for ( const modifyFn of modifyPayloadFnsMap [ 'stream' ] ) { payload = modifyFn ( payload , docId ) ; } const streamInfoPerStore = { } ; const doOnStreamFns = { added : modifyReadResponseMap . added , modified : modifyReadResponseMap . modified , removed : modifyReadResponseMap . removed , } ; /**
         * Last incoming modified docs are cached here temporarily to prevent UI flashing because of the writeLock
         */ const lastIncomingModifiedDocs = new Map ( ) ; /**
         * this is what must be executed by a plugin store that implemented "stream" functionality
         */ const mustExecuteOnRead = { added : async ( _payload , _meta ) => { const _writeLock = writeLockMap . get ( `${ collectionPath }/${ _meta . id }` ) ; if ( _writeLock && isPromise ( _writeLock . promise ) ) { await _writeLock . promise ; } return executeOnFns ( doOnStreamFns . added , _payload , [ _meta ] ) ; } , modified : async ( _payload , _meta ) => { const identifier = `${ collectionPath }/${ _meta . id }` ; lastIncomingModifiedDocs . set ( identifier , [ _payload , _meta ] ) ; const _writeLock = writeLockMap . get ( identifier ) ; if ( _writeLock && isPromise ( _writeLock . promise ) ) { await _writeLock . promise ; } const lastIncoming = lastIncomingModifiedDocs . get ( identifier ) ; if ( ! lastIncoming ) { return ; } const [ __payload , __meta ] = lastIncoming ; lastIncomingModifiedDocs . delete ( identifier ) ; return executeOnFns ( doOnStreamFns . modified , __payload , [ __meta ] ) ; } , removed : async ( _payload , _meta ) => { const identifier = `${ collectionPath }/${ _meta . id }` ; const _writeLock = writeLockMap . get ( identifier ) ; if ( _writeLock && isPromise ( _writeLock . promise ) ) { await _writeLock . promise ; } return executeOnFns ( doOnStreamFns . removed , _payload , [ _meta ] ) ; } , } ; for ( const storeName of storesToExecute ) { const pluginAction = globalConfig . stores [ storeName ] . actions [ 'stream' ] ; const pluginModuleConfig = getPluginModuleConfig ( moduleConfig , storeName ) ; if ( pluginAction ) { const result = await handleStream ( { collectionPath , docId , pluginModuleConfig , pluginAction , actionConfig , payload , eventNameFnsMap , actionName : 'stream' , storeName , mustExecuteOnRead , } ) ; if ( isDoOnStream ( result ) ) { for ( const [ doOn , doFn ] of Object . entries ( result ) ) { if ( doFn ) doOnStreamFns [ doOn as 'added' | 'modified' | 'removed' ] . push ( doFn as any ) ; } } if ( ! isDoOnStream ( result ) ) { streamInfoPerStore [ storeName ] = result ; } } } throwOnIncompleteStreamResponses ( streamInfoPerStore , doOnStreamFns ) ; const streamPromises = Object . values ( streamInfoPerStore ) . map ( ( res ) => res . streaming ) ; const streamPromise = new Promise ( ( resolve , reject ) => { Promise . all ( streamPromises ) . then ( ( ) => resolve ( ) ) . catch ( reject ) ; } ) ; const closeStream = ( ) : void => { Object . values ( streamInfoPerStore ) . forEach ( ( { stop } ) => stop ( ) ) ; cacheStream ( ( ) => { } , null ) ; } ; cacheStream ( closeStream , streamPromise ) ; return streamPromise ; } ; }