import { DateTime } from 'luxon' ; import { remove } from 'fs-extra' ; import { ImportRecord } from '../../schema/database/import' ; import { db } from '../../db' ; import { Import } from './import' ; import { YearRecord } from '../../schema' ; import logger from './logger' ; import { importCityvizor } from './cityvizor/importer' ; import { importInternetStream } from './internetstream/importer' ; import { importPbo } from './pbo/importer' ; export async function checkImportQueue ( ) { const runningJob = await db < ImportRecord > ( 'app.imports' ) . where ( { status : 'processing' } ) . first ( ) ; if ( runningJob ) { const startedBeforeMinutes = runningJob . started ? - 1 * DateTime . fromJSDate ( runningJob . started ) . diffNow ( 'minutes' ) . minutes : 0 ; if ( startedBeforeMinutes < 1 ) return ; else { console . log ( '[WORKER] Found a stale job in queue, removing.' ) ; const updateDataStale = { status : 'error' , error : 'Job timed out without proper finish.' , finished : DateTime . local ( ) . toJSDate ( ) , } ; await db < ImportRecord > ( 'app.imports' ) . where ( { id : runningJob . id } ) . update ( updateDataStale ) ; await remove ( runningJob . importDir ) ; } } const currentJob = await db < ImportRecord > ( 'app.imports' ) . where ( { status : 'pending' } ) . orderBy ( 'created' , 'asc' ) . first ( ) ; if ( ! currentJob ) return ; console . log ( `[WORKER] ${ DateTime . local ( ) . toJSDate ( ) } Found a new ${ currentJob . format } job, starting import.` ) ; await db < ImportRecord > ( 'app.imports' ) . where ( { id : currentJob . id } ) . update ( { status : 'processing' , started : DateTime . local ( ) . toJSDate ( ) } ) ; logger . log ( 'Starting the DB transaction.' ) ; const trx = await db . transaction ( ) ; const options = { profileId : currentJob . profileId , year : currentJob . year , transaction : trx , importDir : currentJob . importDir , append : currentJob . append , format : currentJob . format , } ; let error = null ; try { if ( currentJob . format === 'cityvizor' ) { await importCityvizor ( options ) ; } else if ( currentJob . format === 'internetstream' ) { await importInternetStream ( options ) ; } else if ( currentJob . format === 'pbo_expected_plan' || currentJob . format === 'pbo_real_plan' || currentJob . format === 'pbo_aa_names' ) { await importPbo ( options ) ; } else { throw Error ( `Unsupported import type: ${ currentJob . format }` ) ; } await db < YearRecord > ( 'app.years' ) . where ( { profileId : currentJob . profileId , year : currentJob . year } ) . update ( { validity : currentJob . validity } ) ; } catch ( err ) { console . error ( `[WORKER] ${ DateTime . local ( ) . toJSDate ( ) } Import error` , err ) ; logger . log ( `Import failed: ${ err . message }` ) ; logger . log ( `Additonal information: ${ err . detail }` ) ; error = err ; } finally { if ( error ) { logger . log ( 'Aborting the DB transaction, no changes made to the DB.' ) ; trx . rollback ( ) ; } else { logger . log ( 'Import successful, committing the DB transaction.' ) ; trx . commit ( ) ; } await remove ( currentJob . importDir ) ; } console . log ( '___LOGS____' ) ; console . log ( logger . getLogs ( ) ) ; console . log ( '_____________' ) ; const updateData = { status : error ? 'error' : 'success' , error : error ? error . message : null , finished : DateTime . local ( ) . toJSDate ( ) , logs : logger . getLogs ( ) , } as Partial < ImportRecord > ; logger . clear ( ) ; await db < ImportRecord > ( 'app.imports' ) . where ( { id : currentJob . id } ) . update ( updateData ) ; console . log ( `[WORKER] ${ DateTime . local ( ) . toJSDate ( ) } Import finished.` ) ; }