import type { DocumentSnapshot , QuerySnapshot , DocumentChange , QueryDocumentSnapshot , } from 'firebase/firestore' ; import { doc , onSnapshot } from 'firebase/firestore' ; import { StreamResponse , PluginStreamAction , PluginStreamActionPayload } from '@magnetarjs/core' ; import { FirestoreModuleConfig , FirestorePluginOptions } from '../CreatePlugin' ; import { getFirestoreDocPath , getFirestoreCollectionPath } from '../helpers/pathHelpers' ; import { getQueryInstance , docSnapshotToDocMetadata } from '../helpers/queryHelpers' ; export function streamActionFactory ( firestorePluginOptions ) { return function ( { payload , collectionPath , docId , pluginModuleConfig , mustExecuteOnRead , } ) : StreamResponse { const { added , modified , removed } = mustExecuteOnRead ; const { db } = firestorePluginOptions ; let resolveStream ; let rejectStream ; const streaming = new Promise ( ( resolve , reject ) => { resolveStream = resolve ; rejectStream = reject ; } ) ; let closeStreamStream ; if ( docId ) { const documentPath = getFirestoreDocPath ( collectionPath , docId , pluginModuleConfig , firestorePluginOptions ) ; closeStreamStream = onSnapshot ( doc ( db , documentPath ) , ( docSnapshot ) => { const localChange = docSnapshot . metadata . hasPendingWrites ; if ( localChange ) return ; if ( ! docSnapshot . exists ) return ; const docData = docSnapshot . data ( ) as Record < string , any > ; const docMetadata = docSnapshotToDocMetadata ( docSnapshot ) ; added ( docData , docMetadata ) ; } , rejectStream ) ; } else if ( ! docId ) { const _collectionPath = getFirestoreCollectionPath ( collectionPath , pluginModuleConfig , firestorePluginOptions ) ; const queryInstance = getQueryInstance ( _collectionPath , pluginModuleConfig , db ) ; closeStreamStream = onSnapshot ( queryInstance , ( querySnapshot ) => { const localChanges = querySnapshot . metadata . hasPendingWrites ; if ( localChanges ) return ; querySnapshot . docChanges ( ) . forEach ( ( docChange ) => { const docSnapshot = docChange . doc ; const docData = docSnapshot . data ( ) as Record < string , any > ; const docMetadata = docSnapshotToDocMetadata ( docSnapshot ) ; if ( docChange . type === 'added' ) { added ( docData , docMetadata ) ; } if ( docChange . type === 'modified' ) { modified ( docData , docMetadata ) ; } if ( docChange . type === 'removed' ) { removed ( docData , docMetadata ) ; } } ) ; } , rejectStream ) ; } function stop ( ) { if ( resolveStream ) resolveStream ( ) ; closeStreamStream ( ) ; } return { stop , streaming } ; } ; }