import { PlanRecord } from '../../../schema/database/plan' ; import { Import } from '../import' ; import { Writable } from 'stream' ; import logger from '../logger' ; import { db } from '../../../db' ; import { AaNameRecord } from '../../../schema/database/aaName' ; export class DatabaseWriter extends Writable { planCount = 0 ; aaNameCount = 0 ; constructor ( private options ) { super ( { objectMode : true , } ) ; } async _writev ( chunks , callback ) { try { if ( chunks . length ) { if ( this . options . format === 'pbo_aa_names' ) { await this . writeAaNames ( chunks . map ( chunk => chunk . chunk as AaNameRecord ) ) ; this . aaNameCount += chunks . length ; this . aaNameCount += chunks . length ; } else if ( this . options . format === 'pbo_expected_plan' || this . options . format === 'pbo_real_plan' ) { await this . writePlans ( chunks . map ( chunk => chunk . chunk as PlanRecord ) ) ; this . planCount += chunks . length ; } else { throw new Error ( `Unsupported format: ${ this . options . format }` ) ; } } callback ( ) ; } catch ( err ) { callback ( err ) ; } } _final ( callback ) { if ( this . planCount > 0 ) logger . log ( `Written ${ this . planCount } plan records to the DB.` ) ; if ( this . aaNameCount > 0 ) logger . log ( `Written ${ this . aaNameCount } analytic account name records to the DB.` ) ; callback ( ) ; } async writeAaNames ( aaNames ) { await db < AaNameRecord > ( 'data.pbo_aa_names' ) . insert ( aaNames ) . transacting ( this . options . transaction ) ; } async writePlans ( plans ) { if ( this . options . format === 'pbo_expected_plan' ) { await db < PlanRecord > ( 'data.pbo_expected_plans' ) . insert ( plans ) . transacting ( this . options . transaction ) ; } else if ( this . options . format === 'pbo_real_plan' ) { await db < PlanRecord > ( 'data.pbo_real_plans' ) . insert ( plans ) . transacting ( this . options . transaction ) ; } else { throw new Error ( `Unsupported format: ${ this . options . format }` ) ; } } }