feat: implement geofence using redis pub-sub
This commit is contained in:
@@ -12,3 +12,7 @@ DB_PORT=5432
|
||||
DB_USER=transit_user
|
||||
DB_PASSWORD=26ab87247a55f86032c943b7e361c475cd3ca8b3b772d82e2fe6a9b5ee0839a3
|
||||
DB_NAME=tripz_transit
|
||||
|
||||
# --- Redis (Shared with Main Tripz Backend) ---
|
||||
REDIS_HOST=tripz-redis
|
||||
REDIS_PORT=6379
|
||||
|
||||
@@ -22,6 +22,13 @@ services:
|
||||
- .env
|
||||
depends_on:
|
||||
- transit-db
|
||||
networks:
|
||||
- default
|
||||
- tripz-net
|
||||
|
||||
volumes:
|
||||
transit_pgdata:
|
||||
|
||||
networks:
|
||||
tripz-net:
|
||||
external: true
|
||||
|
||||
@@ -13,8 +13,9 @@
|
||||
"@nestjs/config": "^3.0.0",
|
||||
"@nestjs/core": "^10.0.0",
|
||||
"@nestjs/platform-express": "^10.0.0",
|
||||
"@nestjs/typeorm": "^10.0.0",
|
||||
"pg": "^8.11.0",
|
||||
"@nestjs/typeorm": "^11.0.0",
|
||||
"ioredis": "^5.3.2",
|
||||
"pg": "^8.13.1",
|
||||
"reflect-metadata": "^0.1.13",
|
||||
"rxjs": "^7.8.1",
|
||||
"typeorm": "^0.3.17"
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
import { Injectable, OnModuleInit, Inject, Logger } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { TransitBusTrip } from './entities/transit-bus-trip.entity';
|
||||
import { TransitStation } from './entities/transit-station.entity';
|
||||
import Redis from 'ioredis';
|
||||
|
||||
const GEOFENCE_RADIUS_METERS = 150;
|
||||
|
||||
@Injectable()
|
||||
export class TransitGeofenceService implements OnModuleInit {
|
||||
private readonly logger = new Logger(TransitGeofenceService.name);
|
||||
|
||||
constructor(
|
||||
@Inject('REDIS_CLIENT') private readonly redis: Redis,
|
||||
@InjectRepository(TransitBusTrip) private readonly tripsRepo: Repository<TransitBusTrip>,
|
||||
@InjectRepository(TransitStation) private readonly stationsRepo: Repository<TransitStation>,
|
||||
) {}
|
||||
|
||||
onModuleInit() {
|
||||
this.redis.subscribe('channel:bus-locations', (err, count) => {
|
||||
if (err) {
|
||||
this.logger.error('Failed to subscribe to bus-locations channel', err);
|
||||
} else {
|
||||
this.logger.log(`Subscribed to ${count} channels.`);
|
||||
}
|
||||
});
|
||||
|
||||
this.redis.on('message', async (channel, message) => {
|
||||
if (channel === 'channel:bus-locations') {
|
||||
try {
|
||||
const data = JSON.parse(message);
|
||||
await this.processBusLocation(data.driverId, data.lat, data.lng);
|
||||
} catch (error) {
|
||||
this.logger.error('Error processing bus location', error);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private async processBusLocation(driverId: string, lat: number, lng: number) {
|
||||
// 1. Find active bus trip for this driver
|
||||
const trip = await this.tripsRepo.findOne({
|
||||
where: { driver_id: driverId, status: 'active' },
|
||||
});
|
||||
|
||||
if (!trip) return; // No active trip, ignore
|
||||
|
||||
// 2. Fetch stations for the route
|
||||
// Note: To optimize, this could be cached in memory
|
||||
const stations = await this.stationsRepo.find({
|
||||
where: { route_id: trip.route_id },
|
||||
order: { order_index: 'ASC' },
|
||||
});
|
||||
|
||||
// 3. Find if bus is within any station's radius
|
||||
for (const station of stations) {
|
||||
const distance = this.haversineDistance(lat, lng, Number(station.latitude), Number(station.longitude));
|
||||
|
||||
if (distance <= GEOFENCE_RADIUS_METERS) {
|
||||
// Bus is inside the geofence of this station!
|
||||
if (trip.current_station_id !== station.id) {
|
||||
this.logger.log(`🚌 Bus for driver ${driverId} arrived at station: ${station.name}`);
|
||||
|
||||
// Update trip status
|
||||
trip.current_station_id = station.id;
|
||||
await this.tripsRepo.save(trip);
|
||||
|
||||
// Publish event so main Tripz backend can notify passengers via WebSockets
|
||||
this.redis.publish(
|
||||
'channel:bus-arrived',
|
||||
JSON.stringify({
|
||||
tripId: trip.id,
|
||||
driverId,
|
||||
routeId: trip.route_id,
|
||||
stationId: station.id,
|
||||
stationName: station.name,
|
||||
}),
|
||||
);
|
||||
}
|
||||
break; // Found the station, no need to check others
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculates the great-circle distance between two points on the Earth's surface.
|
||||
* Returns distance in meters.
|
||||
*/
|
||||
private haversineDistance(lat1: number, lon1: number, lat2: number, lon2: number): number {
|
||||
const toRad = (x: number) => (x * Math.PI) / 180;
|
||||
const R = 6371e3; // Earth's radius in meters
|
||||
|
||||
const dLat = toRad(lat2 - lat1);
|
||||
const dLon = toRad(lon2 - lon1);
|
||||
|
||||
const a =
|
||||
Math.sin(dLat / 2) * Math.sin(dLat / 2) +
|
||||
Math.cos(toRad(lat1)) * Math.cos(toRad(lat2)) * Math.sin(dLon / 2) * Math.sin(dLon / 2);
|
||||
|
||||
const c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));
|
||||
return R * c;
|
||||
}
|
||||
}
|
||||
@@ -7,6 +7,9 @@ import { TransitBusTrip } from './entities/transit-bus-trip.entity';
|
||||
import { TransitTicket } from './entities/transit-ticket.entity';
|
||||
import { TransitController } from './transit.controller';
|
||||
import { TransitService } from './transit.service';
|
||||
import { TransitGeofenceService } from './transit-geofence.service';
|
||||
import Redis from 'ioredis';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
@@ -19,7 +22,20 @@ import { TransitService } from './transit.service';
|
||||
]),
|
||||
],
|
||||
controllers: [TransitController],
|
||||
providers: [TransitService],
|
||||
providers: [
|
||||
TransitService,
|
||||
TransitGeofenceService,
|
||||
{
|
||||
provide: 'REDIS_CLIENT',
|
||||
inject: [ConfigService],
|
||||
useFactory: (cfg: ConfigService) => {
|
||||
return new Redis({
|
||||
host: cfg.get<string>('REDIS_HOST', 'tripz-redis'),
|
||||
port: cfg.get<number>('REDIS_PORT', 6379),
|
||||
});
|
||||
},
|
||||
},
|
||||
],
|
||||
exports: [TypeOrmModule, TransitService],
|
||||
})
|
||||
export class TransitModule {}
|
||||
|
||||
Reference in New Issue
Block a user