feat(be): gtfs sync data

apps/backend/src/app.module.ts

@@ -9,6 +9,7 @@ import { cacheModuleConfig } from "src/config/cache-module.config";
 import { configModuleConfig } from "src/config/config-module.config";
 import { GRAPHQL_PATH } from "src/constants/api";
 import { DepartureModule } from "src/modules/departure/departure.module";
+import { GtfsModule } from "src/modules/gtfs/gtfs.module";
 import { ImportModule } from "src/modules/import/import.module";
 import { LoggerModule } from "src/modules/logger/logger.module";
 import { PlatformModule } from "src/modules/platform/platform.module";
@@ -25,6 +26,7 @@ import { StopModule } from "src/modules/stop/stop.module";
         PrismaModule,
         LoggerModule,
         StatusModule,
+        GtfsModule,
         ConfigModule.forRoot(configModuleConfig),
         ScheduleModule.forRoot(),
         CacheModule.registerAsync(cacheModuleConfig),

apps/backend/src/modules/gtfs/gtfs.controller.ts

@@ -0,0 +1,18 @@
+import { Controller, OnModuleInit } from "@nestjs/common";
+import { Cron, CronExpression } from "@nestjs/schedule";
+
+import { GtfsService } from "src/modules/gtfs/gtfs.service";
+
+@Controller("gtfs")
+export class GtfsController implements OnModuleInit {
+    constructor(private readonly gtfsService: GtfsService) {}
+
+    async onModuleInit(): Promise<void> {
+        await this.gtfsService.syncGtfsData();
+    }
+
+    @Cron(CronExpression.EVERY_7_HOURS)
+    async cronSyncStops(): Promise<void> {
+        await this.gtfsService.syncGtfsData();
+    }
+}

apps/backend/src/modules/gtfs/gtfs.module.ts

@@ -0,0 +1,11 @@
+import { Module } from "@nestjs/common";
+
+import { GtfsController } from "src/modules/gtfs/gtfs.controller";
+import { GtfsService } from "src/modules/gtfs/gtfs.service";
+
+@Module({
+    controllers: [GtfsController],
+    providers: [GtfsService],
+    imports: [],
+})
+export class GtfsModule {}

apps/backend/src/modules/gtfs/gtfs.service.ts

@@ -0,0 +1,86 @@
+import { Injectable } from "@nestjs/common";
+import { Open as unzipperOpen } from "unzipper";
+
+import { PrismaService } from "src/modules/prisma/prisma.service";
+import { parseCsvString } from "src/utils/csv.utils";
+
+@Injectable()
+export class GtfsService {
+    constructor(private readonly prisma: PrismaService) {}
+
+    async syncGtfsData() {
+        const response = await fetch("http://data.pid.cz/PID_GTFS.zip");
+        const arrayBuffer = await response.arrayBuffer();
+        const buffer = Buffer.from(arrayBuffer);
+        const directory = await unzipperOpen.buffer(buffer);
+
+        const routes = directory.files.find(
+            (file) => file.path === "routes.txt",
+        );
+        if (!routes) {
+            console.log("routes.txt not found");
+            return;
+        }
+        const routeStops = directory.files.find(
+            (file) => file.path === "route_stops.txt",
+        );
+        if (!routeStops) {
+            console.log("route_stops.txt not found");
+            return;
+        }
+
+        const routesBuffer = await routes.buffer();
+        const routeStopsBuffer = await routeStops.buffer();
+
+        type RouteRecord = {
+            route_id: string;
+            route_short_name: string;
+            route_long_name: string;
+            route_type: string;
+            route_color?: string | undefined;
+            is_night: string;
+            route_url?: string | undefined;
+        };
+        // FIXME: validate with zod
+        const routesData = await parseCsvString<RouteRecord>(
+            routesBuffer.toString(),
+        );
+
+        type RouteStopRecord = {
+            route_id: string;
+            direction_id: string;
+            stop_id: string;
+            stop_sequence: string;
+        };
+        // FIXME: validate with zod
+        const routeStopsData = await parseCsvString<RouteStopRecord>(
+            routeStopsBuffer.toString(),
+        );
+
+        await this.prisma.$transaction(async (transaction) => {
+            await transaction.gtfsRouteStop.deleteMany();
+            await transaction.gtfsRoute.deleteMany();
+
+            await transaction.gtfsRoute.createMany({
+                data: routesData.map((route) => ({
+                    id: route.route_id,
+                    shortName: route.route_short_name,
+                    longName: route.route_long_name ?? null,
+                    type: route.route_type,
+                    isNight: Boolean(route.is_night),
+                    color: route.route_color ?? null,
+                    url: route.route_url ?? null,
+                })),
+            });
+
+            await transaction.gtfsRouteStop.createMany({
+                data: routeStopsData.map((routeStop) => ({
+                    routeId: routeStop.route_id,
+                    directionId: routeStop.direction_id,
+                    stopId: routeStop.stop_id,
+                    stopSequence: Number(routeStop.stop_sequence),
+                })),
+            });
+        });
+    }
+}

apps/backend/src/utils/csv.utils.ts

@@ -0,0 +1,14 @@
+import { parseString } from "@fast-csv/parse";
+
+export async function parseCsvString<T>(csvString: string): Promise<T[]> {
+    return new Promise((resolve) => {
+        const rows: T[] = [];
+
+        parseString(csvString, { headers: true })
+            .on("error", (error) => console.error(error))
+            .on("data", (row) => rows.push(row))
+            .on("end", () => {
+                resolve(rows);
+            });
+    });
+}