feat: implement incremental tour update

service/data-hub/data-connector/src/app.controller.ts

@@ -63,6 +63,15 @@ export class AppController {
       .then(() => this.anomaliesService.validateAnomalies());
   }
 
+  @Get('updated-tours')
+  async updatedTours(@Query('from') from: string, @Query('to') to: string) {
+    const interval = {
+      start: new Date(from),
+      end: new Date(to),
+    };
+    return await this.importer.runTourUpdateCheck(interval);
+  }
+
   @Get('schichtplanung')
   async syncSchichtplanung(@Query('day') day: string) {
     return await this.importer.runSchichtplanungSync(new Date(day));
@@ -120,9 +129,7 @@ export class AppController {
             ],
           },
           {
-            check: {
+            check: 2,
-              not: 3,
-            },
           },
           {
             deletedAt: null,

service/data-hub/data-connector/src/app.module.ts

@@ -1,4 +1,6 @@
 import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo';
+import { HttpModule } from '@nestjs/axios';
+import { BullModule } from '@nestjs/bullmq';
 import { Module } from '@nestjs/common';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { GraphQLModule } from '@nestjs/graphql';
@@ -13,14 +15,15 @@ import { DatabaseModule } from './core/database/database.module';
 import { PrismaModule } from './core/database/prisma.module';
 import { ApplicationAccountingModule } from './modules/app-accounting/app-accounting.module';
 import { ApplicationControlCenterModule } from './modules/app-control-center/app-control-center.module';
+import { AppHomeModule } from './modules/app-home/app-home.module';
 import { ManagerModule } from './modules/app-manager/app-manager.module';
 import { ApplicationTicketSystemModule } from './modules/app-ticket-system/app-ticket-system.module';
 import { AnomalyDetectionModule } from './modules/feat-anomalies/anomaly-detection.module';
+import { FeatCalculationModule } from './modules/feat-calculation/feat-calculation.module';
 import { FeatureEtlModule } from './modules/feat-etl/feat-etl.module';
 import { FeatReportsModule } from './modules/feat-reports/feat-reports.module';
 import { FeatTicketSystemModule } from './modules/feat-tickets/ticket-system.module';
-import { FeatCalculationModule } from './modules/feat-calculation/feat-calculation.module';
+import { FeatBusinessObjectsModule } from './modules/feat-business-objects/feat-business-objects.module';
-import { AppHomeModule } from './modules/app-home/app-home.module';
 
 @Module({
   imports: [
@@ -48,7 +51,7 @@ import { AppHomeModule } from './modules/app-home/app-home.module';
       // isGlobal: true,
       dialect: new PostgresDialect({
         pool: new Pool({
-          host: 'db',
+          host: 'localhost',
           port: 5432,
           database: 'avicenna',
           user: 'johndoe',

service/data-hub/data-connector/src/app.service.ts

@@ -26,7 +26,7 @@ export class AppService {
     private readonly schedulerRegistry: SchedulerRegistry,
   ) {}
 
-  @Cron('0 */5 2-20 * * *')
+  // @Cron('0 */5 2-20 * * *')
   async fetchDataContinously() {
     const now = new Date();
     const fetchInterval = {

service/data-hub/data-connector/src/modules/feat-business-objects/entities/tour/tour.repository.ts

@@ -6,6 +6,7 @@ import { KYSELY_CLIENT_PROVIDER_TOKEN } from 'src/core/database/database.module'
 import { PrismaService } from 'src/core/database/prisma.service';
 import { DB } from 'src/core/database/types';
 import { AccountingState } from 'src/modules/app-accounting/accounting-state';
+import { chunk } from 'src/utils';
 import {
   createPrismaQueryFromAccountingFilters,
   TourAccountingFilterDto,
@@ -127,6 +128,37 @@ export class TourRepository {
     });
   }
 
+  async findUpdatedAtBetween(
+    firstId: string,
+    lastId: string,
+  ): Promise<{ id: string; updatedAt: Date }[]> {
+    return await this.db
+      .with('boundary_dates', (db) =>
+        db
+          .selectFrom('Tour')
+          .select('startDate')
+          .where('id', 'in', [firstId, lastId]),
+      )
+      .selectFrom('Tour')
+      .select(['Tour.id', 'operationId', 'updatedAt'])
+      .where((eb) =>
+        eb.and([
+          eb('Tour.startDate', '>=', (eb) =>
+            eb
+              .selectFrom('boundary_dates')
+              .select(eb.fn.min('startDate').as('min_date')),
+          ),
+          eb('Tour.startDate', '<=', (eb) =>
+            eb
+              .selectFrom('boundary_dates')
+              .select(eb.fn.max('startDate').as('max_date')),
+          ),
+        ]),
+      )
+      .orderBy('Tour.updatedAt', 'asc')
+      .execute();
+  }
+
   public findInIntervalWithEmployee(
     { start, end }: Interval,
     employeeId: string,
@@ -210,6 +242,44 @@ export class TourRepository {
     });
   }
 
+  async createTour(tourInput: Prisma.TourCreateInput): Promise<Tour> {
+    return await this.prisma.tour.create({
+      data: {
+        ...tourInput,
+      },
+    });
+  }
+
+  async updateTour(
+    id: string,
+    tourInput: Prisma.TourUpdateInput,
+  ): Promise<Tour> {
+    return await this.prisma.tour.update({
+      where: { id },
+      data: {
+        ...tourInput,
+      },
+    });
+  }
+
+  async deleteTours(ids: string[]): Promise<Prisma.BatchPayload[]> {
+    const chunks = chunk(ids, 2000);
+    return await Promise.all(
+      chunks.map((chunk) =>
+        this.prisma.tour.updateMany({
+          where: {
+            id: {
+              in: chunk,
+            },
+          },
+          data: {
+            deletedAt: new Date(),
+          },
+        }),
+      ),
+    );
+  }
+
   // -------------------- Ticket System ------------------- //
   async findFirstRecurring(tour: Tour): Promise<Tour | null> {
     const institutionStreet =

service/data-hub/data-connector/src/modules/feat-business-objects/entities/tour/tour.service.ts

@@ -1,8 +1,8 @@
 import { Injectable } from '@nestjs/common';
-import { Tour } from '@prisma/client';
+import { Prisma, Tour } from '@prisma/client';
 import { Interval } from 'date-fns';
-import { TourRepository } from './tour.repository';
 import { AccountingState } from 'src/modules/app-accounting/accounting-state';
+import { TourRepository } from './tour.repository';
 import { TourAccountingFilterDto } from './tours.filter';
 
 @Injectable()
@@ -16,6 +16,13 @@ export class TourService {
     return this.repository.findInInterval(interval, ordinanceTypes);
   }
 
+  async findUpdatedAtBetween(
+    firstId: string,
+    lastId: string,
+  ): Promise<{ id: string; updatedAt: Date }[]> {
+    return this.repository.findUpdatedAtBetween(firstId, lastId);
+  }
+
   public findInIntervalWithEmployee(
     interval: Interval,
     employeeId: string,
@@ -60,6 +67,22 @@ export class TourService {
     return this.repository.removeTourFromTicket(tourId);
   }
 
+  async createTour(tourInput: Prisma.TourCreateInput): Promise<Tour> {
+    return this.repository.createTour(tourInput);
+  }
+
+
+  async updateTour(
+    id: string,
+    tourInput: Prisma.TourUpdateInput,
+  ): Promise<Tour> {
+    return this.repository.updateTour(id, tourInput);
+  }
+
+  async deleteTours(ids: string[]): Promise<Prisma.BatchPayload[]> {
+    return this.repository.deleteTours(ids);
+  }
+
   async findByAccountingState(
     state: AccountingState,
     filter: TourAccountingFilterDto,

service/data-hub/data-connector/src/modules/feat-business-objects/entities/tour/tour.tokens.ts

@@ -0,0 +1 @@
+export const TOURS_QUEUE = 'tours';

service/data-hub/data-connector/src/modules/feat-etl/feat-extractors/dispo-live-extractors.service.ts

@@ -24,6 +24,7 @@ import { EmployeeDispoLiveRaw, PatientRaw, TourRaw } from './source.interface';
 import { SourceRepository } from './source.repository';
 import { chunk } from 'src/utils';
 import { ConfigService } from 'src/core/config/config.service';
+import { TourExtractor } from './extractors/tour.extractor';
 
 @Injectable()
 export class DispoLiveExtractorsService {
@@ -35,6 +36,10 @@ export class DispoLiveExtractorsService {
     private readonly config: ConfigService,
   ) {}
 
+  extractTour(id: string, jobId: string) {
+    return this.fetchTour(id).pipe(map((tour) => ({ ...tour, jobId })));
+  }
+
   extractTours(
     interval: Interval,
     jobId: string,
@@ -54,6 +59,67 @@ export class DispoLiveExtractorsService {
     );
   }
 
+  private fetchTour(id: string): Observable<TourRaw> {
+    const fahrberichteEndpoint = 'data/fahrberichte';
+    const carsEndpoint = 'data/stammdatenFahrzeuge';
+
+    const fetchCars = (
+      authCookies: string,
+    ): Observable<{ id: string; name: string }[]> =>
+      this.http
+        .post<{ data: { _id: string; name: string }[] }>(
+          `${this.config.dispoliveBaseUrl()}/${carsEndpoint}`,
+          {},
+          {
+            headers: {
+              'Content-Type': 'application/json',
+              Cookie: authCookies,
+            },
+          },
+        )
+        .pipe(
+          map((response) =>
+            response.data.data.map(({ _id, name }) => ({ id: _id, name })),
+          ),
+        );
+
+    const fetchTour = (id: string, authCookies: string) =>
+      this.http
+        .post<{ data: TourRaw[] }>(
+          `${this.config.dispoliveBaseUrl()}/${fahrberichteEndpoint}`,
+          {
+            filter: { _id: id },
+          },
+          {
+            headers: {
+              'Content-Type': 'application/json',
+              Cookie: authCookies,
+            },
+          },
+        )
+        .pipe(
+          map((response) => TourExtractor.extract(response.data.data?.at(0))),
+        );
+
+    return this.dispoliveLogin(
+      this.config.dispoliveUsername(),
+      this.config.dispolivePassword(),
+    ).pipe(
+      switchMap((authCookies) =>
+        forkJoin([of(authCookies), fetchCars(authCookies)]),
+      ),
+      switchMap(([authCookies, cars]) =>
+        fetchTour(id, authCookies).pipe(
+          map(({ carNo, ...tour }) => ({
+            ...tour,
+            carNo,
+            carName: cars.find(({ id }) => id === carNo)?.name ?? tour.carName,
+          })),
+        ),
+      ),
+    );
+  }
+
   private fetchTours(intervals: Interval[]): Observable<TourRaw[]> {
     const archiveEndpoint = 'custom/getArchive/getArchive';
     const carsEndpoint = 'data/stammdatenFahrzeuge';
@@ -95,156 +161,7 @@ export class DispoLiveExtractorsService {
         )
         .pipe(
           map((response) =>
-            response.data.map((tour) => {
+            response.data.map((tour) => TourExtractor.extract(tour)),
-              const {
-                _id,
-                check,
-                createTime,
-                lastUpdate,
-                empfangenVonId,
-                endgen,
-                serienId,
-                serienKategorName,
-                serienKategorie,
-                zadStatus,
-                abrDate,
-                abrNum,
-                abrTime,
-                ausrueckZeit,
-                startTime,
-                startDate,
-                date,
-                date_iso,
-                dauergenehmigung,
-                direction,
-                distanceValue,
-                besetztKm,
-                gefahren,
-                gesamtPreis,
-                infektion,
-                itDesc,
-                itKrz,
-                itName,
-                kkId,
-                krankenkasse,
-                patId,
-                patName,
-                patSurname,
-                patStreet,
-                patZip,
-                patCity,
-                startInstitution,
-                startStreet,
-                startZip,
-                startCity,
-                targetInstitution,
-                targetStreet,
-                targetCity,
-                targetZip,
-                transportKrz,
-                transportName,
-                transportart,
-                verordnungsId,
-                verordnungsName,
-                type,
-                startBegin,
-                startEnd,
-                frei,
-                ziel,
-                zielBegin,
-                endTime,
-                endDate,
-                einsatzNummer,
-                carName,
-                carNo,
-                beifahrer,
-                beifahrerName,
-                fahrer,
-                fahrerName,
-                color,
-              } = tour;
-              return {
-                id: _id ? `${_id}` : undefined,
-                check: check ? `${check}` : undefined,
-                createTime: createTime ? `${createTime}` : undefined,
-                lastUpdate: lastUpdate ? `${lastUpdate}` : undefined,
-                empfangenVonId: empfangenVonId
-                  ? `${empfangenVonId}`
-                  : undefined,
-                endgen: endgen ? `${endgen}` : undefined,
-                serienId: serienId ? `${serienId}` : undefined,
-                serienKategorName: serienKategorName
-                  ? `${serienKategorName}`
-                  : undefined,
-                serienKategorie: serienKategorie
-                  ? `${serienKategorie}`
-                  : undefined,
-                zadStatus: zadStatus ? `${zadStatus}` : undefined,
-                abrDate: abrDate ? `${abrDate}` : undefined,
-                abrNum: abrNum ? `${abrNum}` : undefined,
-                abrTime: abrTime ? `${abrTime}` : undefined,
-                ausrueckZeit: ausrueckZeit ? `${ausrueckZeit}` : undefined,
-                startTime: startTime ? `${startTime}` : undefined,
-                startDate: startDate ? `${startDate}` : undefined,
-                date: date ? `${date}` : undefined,
-                date_iso: date_iso ? `${date_iso}` : undefined,
-                dauergenehmigung: dauergenehmigung
-                  ? `${dauergenehmigung}`
-                  : undefined,
-                direction: direction ? `${direction}` : undefined,
-                distanceValue: distanceValue ? `${distanceValue}` : undefined,
-                besetztKm: besetztKm ? `${besetztKm}` : undefined,
-                gefahren: gefahren ? `${gefahren}` : undefined,
-                gesamtPreis: gesamtPreis ? `${gesamtPreis}` : undefined,
-                infektion: infektion ? `${infektion}` : undefined,
-                itDesc: itDesc ? `${itDesc}` : undefined,
-                itKrz: itKrz ? `${itKrz}` : undefined,
-                itName: itName ? `${itName}` : undefined,
-                kkId: kkId ? `${kkId}` : undefined,
-                krankenkasse: krankenkasse ? `${krankenkasse}` : undefined,
-                patId: patId ? `${patId}` : undefined,
-                patName: patName ? `${patName}` : undefined,
-                patSurname: patSurname ? `${patSurname}` : undefined,
-                patStreet: patStreet ? `${patStreet}` : undefined,
-                patZip: patZip ? `${patZip}` : undefined,
-                patCity: patCity ? `${patCity}` : undefined,
-                startInstitution: startInstitution
-                  ? `${startInstitution}`
-                  : undefined,
-                startStreet: startStreet ? `${startStreet}` : undefined,
-                startZip: startZip ? `${startZip}` : undefined,
-                startCity: startCity ? `${startCity}` : undefined,
-                targetInstitution: targetInstitution
-                  ? `${targetInstitution}`
-                  : undefined,
-                targetStreet: targetStreet ? `${targetStreet}` : undefined,
-                targetCity: targetCity ? `${targetCity}` : undefined,
-                targetZip: targetZip ? `${targetZip}` : undefined,
-                transportKrz: transportKrz ? `${transportKrz}` : undefined,
-                transportName: transportName ? `${transportName}` : undefined,
-                transportart: transportart ? `${transportart}` : undefined,
-                verordnungsId: verordnungsId ? `${verordnungsId}` : undefined,
-                verordnungsName: verordnungsName
-                  ? `${verordnungsName}`
-                  : undefined,
-                type: type ? `${type}` : undefined,
-                startBegin: startBegin ? `${startBegin}` : undefined,
-                startEnd: startEnd ? `${startEnd}` : undefined,
-                frei: frei ? `${frei}` : undefined,
-                ziel: ziel ? `${ziel}` : undefined,
-                zielBegin: zielBegin ? `${zielBegin}` : undefined,
-                endTime: endTime ? `${endTime}` : undefined,
-                endDate: endDate ? `${endDate}` : undefined,
-                einsatzNummer: einsatzNummer ? `${einsatzNummer}` : undefined,
-                carName: carName ? `${carName}` : undefined,
-                carNo: carNo ? `${carNo}` : undefined,
-                beifahrer: beifahrer ? `${beifahrer}` : undefined,
-                beifahrerName: beifahrerName ? `${beifahrerName}` : undefined,
-                fahrer: fahrer ? `${fahrer}` : undefined,
-                fahrerName: fahrerName ? `${fahrerName}` : undefined,
-                color: color ? `${color}` : undefined,
-              };
-            }),
           ),
         );
 
@@ -326,7 +243,7 @@ export class DispoLiveExtractorsService {
                 !!qualifikation && typeof qualifikation === 'object'
                   ? //@ts-expect-error ts is wrong here
                     qualifikation?._id
-                  : qualifikation ?? undefined,
+                  : (qualifikation ?? undefined),
             };
           }),
         ),

service/data-hub/data-connector/src/modules/feat-etl/feat-extractors/extractors/tour.extractor.ts

@@ -0,0 +1,141 @@
+import { Prisma } from '@prisma/client';
+import { TourRaw } from '../source.interface';
+
+export class TourExtractor {
+  public static extract(tour: TourRaw): Partial<Prisma.TourSourceCreateInput> {
+    const {
+      _id,
+      check,
+      createTime,
+      lastUpdate,
+      empfangenVonId,
+      endgen,
+      serienId,
+      serienKategorName,
+      serienKategorie,
+      zadStatus,
+      abrDate,
+      abrNum,
+      abrTime,
+      ausrueckZeit,
+      startTime,
+      startDate,
+      date,
+      date_iso,
+      dauergenehmigung,
+      direction,
+      distanceValue,
+      besetztKm,
+      gefahren,
+      gesamtPreis,
+      infektion,
+      itDesc,
+      itKrz,
+      itName,
+      kkId,
+      krankenkasse,
+      patId,
+      patName,
+      patSurname,
+      patStreet,
+      patZip,
+      patCity,
+      startInstitution,
+      startStreet,
+      startZip,
+      startCity,
+      targetInstitution,
+      targetStreet,
+      targetCity,
+      targetZip,
+      transportKrz,
+      transportName,
+      transportart,
+      verordnungsId,
+      verordnungsName,
+      type,
+      startBegin,
+      startEnd,
+      frei,
+      ziel,
+      zielBegin,
+      endTime,
+      endDate,
+      einsatzNummer,
+      carName,
+      carNo,
+      beifahrer,
+      beifahrerName,
+      fahrer,
+      fahrerName,
+      color,
+    } = tour;
+    return {
+      id: _id ? `${_id}` : undefined,
+      check: check ? `${check}` : undefined,
+      createTime: createTime ? `${createTime}` : undefined,
+      lastUpdate: lastUpdate ? `${lastUpdate}` : undefined,
+      empfangenVonId: empfangenVonId ? `${empfangenVonId}` : undefined,
+      endgen: endgen ? `${endgen}` : undefined,
+      serienId: serienId ? `${serienId}` : undefined,
+      serienKategorName: serienKategorName ? `${serienKategorName}` : undefined,
+      serienKategorie: serienKategorie ? `${serienKategorie}` : undefined,
+      zadStatus: zadStatus ? `${zadStatus}` : undefined,
+      abrDate: abrDate ? `${abrDate}` : undefined,
+      abrNum: abrNum ? `${abrNum}` : undefined,
+      abrTime: abrTime ? `${abrTime}` : undefined,
+      ausrueckZeit: ausrueckZeit ? `${ausrueckZeit}` : undefined,
+      startTime: startTime ? `${startTime}` : undefined,
+      startDate: startDate ? `${startDate}` : undefined,
+      date: date ? `${date}` : undefined,
+      date_iso: date_iso ? `${date_iso}` : undefined,
+      dauergenehmigung: dauergenehmigung ? `${dauergenehmigung}` : undefined,
+      direction: direction ? `${direction}` : undefined,
+      distanceValue: distanceValue ? `${distanceValue}` : undefined,
+      besetztKm: besetztKm ? `${besetztKm}` : undefined,
+      gefahren: gefahren ? `${gefahren}` : undefined,
+      gesamtPreis: gesamtPreis ? `${gesamtPreis}` : undefined,
+      infektion: infektion ? `${infektion}` : undefined,
+      itDesc: itDesc ? `${itDesc}` : undefined,
+      itKrz: itKrz ? `${itKrz}` : undefined,
+      itName: itName ? `${itName}` : undefined,
+      kkId: kkId ? `${kkId}` : undefined,
+      krankenkasse: krankenkasse ? `${krankenkasse}` : undefined,
+      patId: patId ? `${patId}` : undefined,
+      patName: patName ? `${patName}` : undefined,
+      patSurname: patSurname ? `${patSurname}` : undefined,
+      patStreet: patStreet ? `${patStreet}` : undefined,
+      patZip: patZip ? `${patZip}` : undefined,
+      patCity: patCity ? `${patCity}` : undefined,
+      startInstitution: startInstitution ? `${startInstitution}` : undefined,
+      startStreet: startStreet ? `${startStreet}` : undefined,
+      startZip: startZip ? `${startZip}` : undefined,
+      startCity: startCity ? `${startCity}` : undefined,
+      targetInstitution: targetInstitution ? `${targetInstitution}` : undefined,
+      targetStreet: targetStreet ? `${targetStreet}` : undefined,
+      targetCity: targetCity ? `${targetCity}` : undefined,
+      targetZip: targetZip ? `${targetZip}` : undefined,
+      transportKrz: transportKrz ? `${transportKrz}` : undefined,
+      transportName: transportName ? `${transportName}` : undefined,
+      transportart: transportart ? `${transportart}` : undefined,
+      verordnungsId: verordnungsId ? `${verordnungsId}` : undefined,
+      verordnungsName: verordnungsName ? `${verordnungsName}` : undefined,
+      type: type ? `${type}` : undefined,
+      startBegin: startBegin ? `${startBegin}` : undefined,
+      startEnd: startEnd ? `${startEnd}` : undefined,
+      frei: frei ? `${frei}` : undefined,
+      ziel: ziel ? `${ziel}` : undefined,
+      zielBegin: zielBegin ? `${zielBegin}` : undefined,
+      endTime: endTime ? `${endTime}` : undefined,
+      endDate: endDate ? `${endDate}` : undefined,
+      einsatzNummer: einsatzNummer ? `${einsatzNummer}` : undefined,
+      carName: carName ? `${carName}` : undefined,
+      carNo: carNo ? `${carNo}` : undefined,
+      beifahrer: beifahrer ? `${beifahrer}` : undefined,
+      beifahrerName: beifahrerName ? `${beifahrerName}` : undefined,
+      fahrer: fahrer ? `${fahrer}` : undefined,
+      fahrerName: fahrerName ? `${fahrerName}` : undefined,
+      color: color ? `${color}` : undefined,
+    };
+  }
+}

service/data-hub/data-connector/src/modules/feat-etl/feat-extractors/source.interface.ts

@@ -1,5 +1,6 @@
 export interface TourRaw {
   _id?: string;
+  id?: string;
   check?: string;
   createTime?: string; // yyyy-MM-dd HH:mm:ss  'Europe/Berlin'  OR   yyyy-MM-dd  'Europe/Berlin'
   lastUpdate?: string; // yyyy-MM-dd HH:mm:ss  'Europe/Berlin'
@@ -149,3 +150,31 @@ export interface TourSummaryDto {
   check: string;
   lastUpdate?: string;
 }
+
+export interface FahrzeugplanungItemDto {
+  _id?: string;
+  className: string;      // 'greenItem' OR 'blocked'
+  group: string;          // Fahrzeug ID
+  mitarbeiterId: string;  // Mitarbeiter ID
+  start: string;          // ISO 8601 TS (UTC)
+  end: string;            // ISO 8601 TS (UTC)
+}
+
+export interface FahrzeugItemDto {
+  _id: string
+  bezeichnung: string
+  marke: string
+  tuev: string
+  conn: string
+  carStatus: string
+  kaufDatum: string
+  erstzulassung: string
+  status: string
+  firmaId: string
+  kennzeichen: string
+  kaufPreis: string
+  kaufKm: string
+  fahrzeugIdentNr: string
+  kfzNr: string
+  verordnungsart: string
+}

service/data-hub/data-connector/src/modules/feat-etl/feat-importers/feature-importer.module.ts

@@ -1,12 +1,15 @@
 import { BullModule } from '@nestjs/bullmq';
 import { Module } from '@nestjs/common';
+import { TOURS_QUEUE } from 'src/modules/feat-business-objects/entities/tour/tour.tokens';
 import { FeatBusinessObjectsModule } from 'src/modules/feat-business-objects/feat-business-objects.module';
 import { DataAccessExtractorsModule } from '../feat-extractors/data-access-extractors.module';
 import { DataAccessLoadersModule } from '../feat-loaders/data-access-loaders.module';
 import { DataAccessTransformersModule } from '../feat-transformers/data-access-transformers.module';
 import { ImporterService } from './importer.service';
 import { SchichtplanungConsumer } from './processors/schichtplanung.consumer';
+import { ToursConsumer } from './processors/tours.consumer';
 import { SchichtplanungSynchronizer } from './producer/schichtplanung';
+import { TourUpdateChecker } from './producer/tour-update';
 import { SCHICHTPLANUNG_QUEUE } from './tokens';
 
 @Module({
@@ -15,10 +18,13 @@ import { SCHICHTPLANUNG_QUEUE } from './tokens';
     DataAccessExtractorsModule,
     DataAccessTransformersModule,
     DataAccessLoadersModule,
+    BullModule.registerQueue({ name: TOURS_QUEUE }),
     BullModule.registerQueue({ name: SCHICHTPLANUNG_QUEUE }),
   ],
   providers: [
     ImporterService,
+    ToursConsumer,
+    TourUpdateChecker,
     SchichtplanungSynchronizer,
     SchichtplanungConsumer,
   ],

service/data-hub/data-connector/src/modules/feat-etl/feat-importers/importer.service.ts

@@ -11,6 +11,7 @@ import { ETLEventTypes } from './events/event-types.enum';
 import { ExtractionDoneEvent } from './events/extraction-done.event';
 import { LoadingDoneEvent } from './events/loading-done.event';
 import { TransformationDoneEvent } from './events/transformation-done.event';
+import { TourUpdateChecker } from './producer/tour-update';
 import { SchichtplanungSynchronizer } from './producer/schichtplanung';
 
 @Injectable()
@@ -21,9 +22,13 @@ export class ImporterService {
     private readonly transformersService: TransformersService,
     private readonly loadersService: LoadersService,
     private readonly eventEmitter: EventEmitter2,
+    private readonly tourUpdateChecker: TourUpdateChecker,
     private readonly schichtplanungSynchronizer: SchichtplanungSynchronizer,
   ) {}
 
+  runTourUpdateCheck(interval: Interval) {
+    return this.tourUpdateChecker.fetchTourUpdates(interval);
+  }
 
   runSchichtplanungSync(day: Date) {
     return this.schichtplanungSynchronizer.syncSchichtplanung(day);

service/data-hub/data-connector/src/modules/feat-etl/feat-importers/processors/tours.consumer.ts

@@ -0,0 +1,63 @@
+import { InjectQueue, Processor, WorkerHost } from '@nestjs/bullmq';
+import { Job, Queue, Worker } from 'bullmq';
+import { TourService } from 'src/modules/feat-business-objects/entities/tour/tour.service';
+import { TOURS_QUEUE } from 'src/modules/feat-business-objects/entities/tour/tour.tokens';
+import { DispoLiveExtractorsService } from '../../feat-extractors/dispo-live-extractors.service';
+import { firstValueFrom, from, lastValueFrom, map, switchMap, tap } from 'rxjs';
+import { TourTransformer } from '../../feat-transformers/transformers/tour.transformer';
+import { TourSource } from '@prisma/client';
+import { ToursCalculationAdapater } from '../../feat-transformers/adapter/tours-calculation.adapter';
+import { Logger } from '@nestjs/common';
+
+@Processor(TOURS_QUEUE, {
+  limiter: {
+    duration: 1000,
+    max: 10,
+  },
+})
+export class ToursConsumer extends WorkerHost {
+  private readonly logger = new Logger(ToursConsumer.name);
+
+  constructor(
+    private readonly service: TourService,
+    private readonly apiService: DispoLiveExtractorsService,
+    private readonly tourCalculationAdapter: ToursCalculationAdapater,
+  ) {
+    super();
+  }
+
+  async process(job: Job, token?: string): Promise<any> {
+    let isCreate = false;
+    switch (job.name) {
+      case 'Create':
+        isCreate = true;
+      case 'Update':
+        this.logger.log(`Working on Job(${job.id}): ${job.data?.id}`);
+        const tour = await lastValueFrom(
+          this.apiService.extractTour(job.data?.id, job.id).pipe(
+            map((tour) =>
+              TourTransformer.transform({
+                ...tour,
+                id: tour._id ?? tour.id,
+                insertedAt: new Date(),
+                internalId: tour.jobId,
+              } as TourSource),
+            ),
+            switchMap((tour) =>
+              from(
+                this.tourCalculationAdapter.addRevenueAndConsumptionCostsToTours(
+                  [tour],
+                ),
+              ).pipe(map(([tour]) => tour)),
+            ),
+          ),
+        );
+        if (isCreate || job.name === 'Create') {
+          return await this.service.createTour(tour).then(() => this.logger.log(`Created tour ${tour.id} successfully`));
+        } 
+        return await this.service.updateTour(tour.id, tour).then(() => this.logger.log(`Updated tour ${tour.id} successfully`));
+      default:
+        console.log(`No case matched`, job);
+    }
+  }
+}

service/data-hub/data-connector/src/modules/feat-etl/feat-importers/producer/tour-update.ts

@@ -0,0 +1,92 @@
+import { InjectQueue } from '@nestjs/bullmq';
+import { Injectable } from '@nestjs/common';
+import { Queue } from 'bullmq';
+import { Interval } from 'date-fns';
+import { catchError, from, map, switchMap, tap, throwError } from 'rxjs';
+import { TourService } from 'src/modules/feat-business-objects/entities/tour/tour.service';
+import { TOURS_QUEUE } from 'src/modules/feat-business-objects/entities/tour/tour.tokens';
+import { DispoLiveApiExtractorsService } from '../../feat-extractors/dispo-live-api-extractors.service';
+import {
+  parseCETDateTime,
+  pick,
+  transformProperties,
+} from '../../feat-transformers/transformer.functions';
+
+@Injectable()
+export class TourUpdateChecker {
+  constructor(
+    private readonly tourService: TourService,
+    private readonly dispoLiveApiService: DispoLiveApiExtractorsService,
+    @InjectQueue(TOURS_QUEUE) private readonly toursQueue: Queue,
+  ) {}
+
+  async fetchTourUpdates(interval: Interval) {
+    return await this.dispoLiveApiService.toursFindByRange(interval).pipe(
+      switchMap((summaries) =>
+        from(
+          this.tourService.findUpdatedAtBetween(
+            summaries.data.at(0)._id,
+            summaries.data.at(-1)._id,
+          ),
+        ).pipe(
+          map((savedTours) => {
+            const deleted = [];
+            const updated = [];
+            const created = [];
+
+            for (const summary of summaries.data) {
+              const savedTour = savedTours.find(({ id }) => summary._id === id);
+              if (!savedTour) created.push({ id: summary._id });
+              const lastUpdateDate = summary?.lastUpdate
+                ? transformProperties(parseCETDateTime('yyyy-MM-dd HH:mm:ss'))(
+                    pick(['lastUpdate'])(
+                      summary as Record<'lastUpdate', string>,
+                    ),
+                  )
+                : undefined;
+
+              if (
+                lastUpdateDate?.lastUpdate.toISOString() !==
+                savedTour?.updatedAt?.toISOString()
+              ) {
+                updated.push({ id: summary._id });
+              }
+            }
+
+            for (const savedTour of savedTours) {
+              const summary = summaries.data.find(
+                ({ _id }) => savedTour.id === _id,
+              );
+              if (!summary) deleted.push({ id: savedTour.id });
+            }
+
+            return {
+              created,
+              updated,
+              deleted,
+            };
+          }),
+          switchMap(({ created, updated, deleted }) =>
+            from(
+              Promise.all([
+                ...created.map(
+                  async (summary) =>
+                    await this.toursQueue.add(`Create`, summary),
+                ),
+                ...updated.map(
+                  async (summary) =>
+                    await this.toursQueue.add(`Update`, summary),
+                ),
+                this.tourService.deleteTours(deleted.map(({ id }) => id)),
+              ]),
+            ),
+          ),
+        ),
+      ),
+      catchError((err) => {
+        console.log(err);
+        return throwError(() => new Error('Something went wrong'));
+      }),
+    );
+  }
+}

service/data-hub/data-connector/src/modules/feat-etl/feat-transformers/data-access-transformers.module.ts

@@ -9,6 +9,6 @@ import { TransformersService } from './transformers.service';
 @Module({
   imports: [PrismaModule, DataAccessExtractorsModule, FeatCalculationModule],
   providers: [TransformersService, StagingRepository, ToursCalculationAdapater],
-  exports: [TransformersService],
+  exports: [TransformersService, ToursCalculationAdapater],
 })
 export class DataAccessTransformersModule {}