feat: introduce routing status tracking for approved roads and telemetry batch queue limits
This commit is contained in:
@@ -5,6 +5,7 @@ import { Repository, DataSource } from 'typeorm';
|
|||||||
import { CandidateRoad } from './candidate-road.entity';
|
import { CandidateRoad } from './candidate-road.entity';
|
||||||
import { RoadSegmentStat } from './road-stat.entity';
|
import { RoadSegmentStat } from './road-stat.entity';
|
||||||
import { ApiKeyGuard } from '../common/guards/api-key.guard';
|
import { ApiKeyGuard } from '../common/guards/api-key.guard';
|
||||||
|
import { RedisService } from '../common/redis.service';
|
||||||
|
|
||||||
@ApiTags('map-refinement-roads')
|
@ApiTags('map-refinement-roads')
|
||||||
@Controller('map-refinement/roads')
|
@Controller('map-refinement/roads')
|
||||||
@@ -18,6 +19,7 @@ export class RoadRefinementController {
|
|||||||
@InjectRepository(RoadSegmentStat)
|
@InjectRepository(RoadSegmentStat)
|
||||||
private readonly roadStatRepo: Repository<RoadSegmentStat>,
|
private readonly roadStatRepo: Repository<RoadSegmentStat>,
|
||||||
private readonly dataSource: DataSource,
|
private readonly dataSource: DataSource,
|
||||||
|
private readonly redisService: RedisService,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
@Get('summary')
|
@Get('summary')
|
||||||
@@ -60,12 +62,15 @@ export class RoadRefinementController {
|
|||||||
const s = status || 'pending';
|
const s = status || 'pending';
|
||||||
try {
|
try {
|
||||||
const rows = await this.dataSource.query(
|
const rows = await this.dataSource.query(
|
||||||
`SELECT id, "uniqueDriverCount", "totalPoints", "averageSpeed", "lengthMeters",
|
`SELECT c.id, c."uniqueDriverCount", c."totalPoints", c."averageSpeed", c."lengthMeters",
|
||||||
confidence, status, source, name, highway, oneway, "discoveredAt", "reviewedAt",
|
confidence, status, source, name, highway, oneway, "discoveredAt", "reviewedAt",
|
||||||
ST_AsGeoJSON(geometry) as geojson
|
ar.routing_status AS "routingStatus", ar.routing_error AS "routingError",
|
||||||
FROM candidate_roads
|
ar.routing_requested_at AS "routingRequestedAt", ar.routed_at AS "routedAt",
|
||||||
WHERE status = $1
|
ST_AsGeoJSON(c.geometry) as geojson
|
||||||
ORDER BY confidence DESC, "discoveredAt" DESC
|
FROM candidate_roads c
|
||||||
|
LEFT JOIN approved_roads ar ON ar.candidate_id = c.id
|
||||||
|
WHERE c.status = $1
|
||||||
|
ORDER BY c.confidence DESC, c."discoveredAt" DESC
|
||||||
LIMIT 100`,
|
LIMIT 100`,
|
||||||
[s]
|
[s]
|
||||||
);
|
);
|
||||||
@@ -75,10 +80,38 @@ export class RoadRefinementController {
|
|||||||
}));
|
}));
|
||||||
} catch (e: any) {
|
} catch (e: any) {
|
||||||
this.logger.error(`Error fetching candidates: ${e.message}`);
|
this.logger.error(`Error fetching candidates: ${e.message}`);
|
||||||
return [];
|
// Before the first approval the optional publication table may not exist;
|
||||||
|
// pending-road review must remain available in that fresh installation.
|
||||||
|
try {
|
||||||
|
const rows = await this.dataSource.query(
|
||||||
|
`SELECT id, "uniqueDriverCount", "totalPoints", "averageSpeed", "lengthMeters",
|
||||||
|
confidence, status, source, name, highway, oneway, "discoveredAt", "reviewedAt",
|
||||||
|
ST_AsGeoJSON(geometry) as geojson
|
||||||
|
FROM candidate_roads WHERE status = $1
|
||||||
|
ORDER BY confidence DESC, "discoveredAt" DESC LIMIT 100`,
|
||||||
|
[s],
|
||||||
|
);
|
||||||
|
return rows.map((r: any) => ({ ...r, geometry: r.geojson ? JSON.parse(r.geojson) : null }));
|
||||||
|
} catch (_) {
|
||||||
|
return [];
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Get('approved/:id/routing-status')
|
||||||
|
@ApiOperation({ summary: 'Get approved road routing publication status' })
|
||||||
|
@ApiParam({ name: 'id', description: 'Approved road or candidate UUID' })
|
||||||
|
async getRoutingStatus(@Param('id') id: string) {
|
||||||
|
const [road] = await this.dataSource.query(`
|
||||||
|
SELECT id, candidate_id AS "candidateId", routing_status AS "routingStatus",
|
||||||
|
routing_error AS "routingError", routing_requested_at AS "routingRequestedAt",
|
||||||
|
routed_at AS "routedAt", start_node AS "startNode", end_node AS "endNode"
|
||||||
|
FROM approved_roads WHERE id::text = $1 OR candidate_id::text = $1
|
||||||
|
`, [id]);
|
||||||
|
if (!road) throw new HttpException('Approved road not found', HttpStatus.NOT_FOUND);
|
||||||
|
return road;
|
||||||
|
}
|
||||||
|
|
||||||
@Post('candidates/manual')
|
@Post('candidates/manual')
|
||||||
@ApiOperation({ summary: 'Submit manual candidate road drawn on map ✍️' })
|
@ApiOperation({ summary: 'Submit manual candidate road drawn on map ✍️' })
|
||||||
async submitManualCandidate(@Body() body: { geojson: any; name?: string; highway?: string }) {
|
async submitManualCandidate(@Body() body: { geojson: any; name?: string; highway?: string }) {
|
||||||
@@ -212,6 +245,10 @@ export class RoadRefinementController {
|
|||||||
oneway SMALLINT DEFAULT 0,
|
oneway SMALLINT DEFAULT 0,
|
||||||
start_node BIGINT,
|
start_node BIGINT,
|
||||||
end_node BIGINT,
|
end_node BIGINT,
|
||||||
|
routing_status VARCHAR(32) NOT NULL DEFAULT 'pending_connection',
|
||||||
|
routing_error TEXT,
|
||||||
|
routing_requested_at TIMESTAMP,
|
||||||
|
routed_at TIMESTAMP,
|
||||||
approved_at TIMESTAMP DEFAULT NOW()
|
approved_at TIMESTAMP DEFAULT NOW()
|
||||||
);
|
);
|
||||||
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS candidate_id UUID;
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS candidate_id UUID;
|
||||||
@@ -223,6 +260,10 @@ export class RoadRefinementController {
|
|||||||
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS start_node BIGINT;
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS start_node BIGINT;
|
||||||
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS end_node BIGINT;
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS end_node BIGINT;
|
||||||
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS approved_at TIMESTAMP DEFAULT NOW();
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS approved_at TIMESTAMP DEFAULT NOW();
|
||||||
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS routing_status VARCHAR(32) NOT NULL DEFAULT 'pending_connection';
|
||||||
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS routing_error TEXT;
|
||||||
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS routing_requested_at TIMESTAMP;
|
||||||
|
ALTER TABLE approved_roads ADD COLUMN IF NOT EXISTS routed_at TIMESTAMP;
|
||||||
CREATE INDEX IF NOT EXISTS approved_roads_geom_idx ON approved_roads USING GIST(geometry);
|
CREATE INDEX IF NOT EXISTS approved_roads_geom_idx ON approved_roads USING GIST(geometry);
|
||||||
CREATE UNIQUE INDEX IF NOT EXISTS approved_roads_cand_uidx ON approved_roads(candidate_id) WHERE candidate_id IS NOT NULL;
|
CREATE UNIQUE INDEX IF NOT EXISTS approved_roads_cand_uidx ON approved_roads(candidate_id) WHERE candidate_id IS NOT NULL;
|
||||||
`);
|
`);
|
||||||
@@ -244,86 +285,68 @@ export class RoadRefinementController {
|
|||||||
reviewedAt: new Date(),
|
reviewedAt: new Date(),
|
||||||
});
|
});
|
||||||
|
|
||||||
// 4. Safely attempt topology node snapping (Non-blocking)
|
const [middleTables] = await this.dataSource.query(`
|
||||||
try {
|
SELECT COUNT(*)::int AS count
|
||||||
await this.dataSource.query(`
|
FROM information_schema.tables
|
||||||
DO $$
|
WHERE table_schema = 'public' AND table_name IN ('planet_osm_line', 'planet_osm_ways', 'planet_osm_nodes')
|
||||||
DECLARE
|
`);
|
||||||
cand RECORD;
|
if (middleTables.count !== 3) {
|
||||||
start_pt GEOMETRY;
|
await this.dataSource.query(`UPDATE approved_roads
|
||||||
end_pt GEOMETRY;
|
SET routing_status = 'blocked_missing_network',
|
||||||
snapped_geom GEOMETRY;
|
routing_error = 'Routing network index is unavailable; retry after the map import completes.'
|
||||||
s_node BIGINT := NULL;
|
WHERE candidate_id = $1::uuid`, [id]);
|
||||||
e_node BIGINT := NULL;
|
return { success: true, id, status: 'approved', routingStatus: 'blocked_missing_network' };
|
||||||
node_rec RECORD;
|
|
||||||
BEGIN
|
|
||||||
SELECT id, geometry::geometry as geom, name, COALESCE(highway, 'residential') as highway, confidence, "uniqueDriverCount", oneway
|
|
||||||
INTO cand
|
|
||||||
FROM candidate_roads WHERE id = $1;
|
|
||||||
|
|
||||||
IF FOUND THEN
|
|
||||||
snapped_geom := cand.geom;
|
|
||||||
start_pt := ST_StartPoint(snapped_geom);
|
|
||||||
end_pt := ST_EndPoint(snapped_geom);
|
|
||||||
|
|
||||||
-- Snap start
|
|
||||||
BEGIN
|
|
||||||
WITH p AS (SELECT ST_Transform(start_pt, 3857) AS pt)
|
|
||||||
SELECT n.id, n.lon/1e7 as lon, n.lat/1e7 as lat
|
|
||||||
INTO node_rec
|
|
||||||
FROM p
|
|
||||||
JOIN planet_osm_line l ON l.highway IS NOT NULL AND l.way && ST_Expand(p.pt, 60)
|
|
||||||
JOIN planet_osm_ways w ON w.id = l.osm_id
|
|
||||||
CROSS JOIN LATERAL unnest(w.nodes) AS wn(node_id)
|
|
||||||
JOIN planet_osm_nodes n ON n.id = wn.node_id
|
|
||||||
WHERE ST_Distance(ST_SetSRID(ST_MakePoint(n.lon/1e7, n.lat/1e7), 4326)::geography, start_pt::geography) < 30
|
|
||||||
ORDER BY ST_Distance(ST_SetSRID(ST_MakePoint(n.lon/1e7, n.lat/1e7), 4326)::geography, start_pt::geography) ASC
|
|
||||||
LIMIT 1;
|
|
||||||
|
|
||||||
IF FOUND THEN
|
|
||||||
s_node := node_rec.id;
|
|
||||||
snapped_geom := ST_SetPoint(snapped_geom, 0, ST_SetSRID(ST_MakePoint(node_rec.lon, node_rec.lat), 4326));
|
|
||||||
END IF;
|
|
||||||
EXCEPTION WHEN OTHERS THEN
|
|
||||||
NULL;
|
|
||||||
END;
|
|
||||||
|
|
||||||
-- Snap end
|
|
||||||
BEGIN
|
|
||||||
WITH p AS (SELECT ST_Transform(end_pt, 3857) AS pt)
|
|
||||||
SELECT n.id, n.lon/1e7 as lon, n.lat/1e7 as lat
|
|
||||||
INTO node_rec
|
|
||||||
FROM p
|
|
||||||
JOIN planet_osm_line l ON l.highway IS NOT NULL AND l.way && ST_Expand(p.pt, 60)
|
|
||||||
JOIN planet_osm_ways w ON w.id = l.osm_id
|
|
||||||
CROSS JOIN LATERAL unnest(w.nodes) AS wn(node_id)
|
|
||||||
JOIN planet_osm_nodes n ON n.id = wn.node_id
|
|
||||||
WHERE ST_Distance(ST_SetSRID(ST_MakePoint(n.lon/1e7, n.lat/1e7), 4326)::geography, end_pt::geography) < 30
|
|
||||||
ORDER BY ST_Distance(ST_SetSRID(ST_MakePoint(n.lon/1e7, n.lat/1e7), 4326)::geography, end_pt::geography) ASC
|
|
||||||
LIMIT 1;
|
|
||||||
|
|
||||||
IF FOUND THEN
|
|
||||||
e_node := node_rec.id;
|
|
||||||
snapped_geom := ST_SetPoint(snapped_geom, ST_NPoints(snapped_geom) - 1, ST_SetSRID(ST_MakePoint(node_rec.lon, node_rec.lat), 4326));
|
|
||||||
END IF;
|
|
||||||
EXCEPTION WHEN OTHERS THEN
|
|
||||||
NULL;
|
|
||||||
END;
|
|
||||||
|
|
||||||
UPDATE approved_roads
|
|
||||||
SET geometry = snapped_geom,
|
|
||||||
start_node = COALESCE(s_node, start_node),
|
|
||||||
end_node = COALESCE(e_node, end_node)
|
|
||||||
WHERE candidate_id = cand.id;
|
|
||||||
END IF;
|
|
||||||
END $$;
|
|
||||||
`, [id]);
|
|
||||||
} catch (snapErr: any) {
|
|
||||||
this.logger.warn(`Topology snapping skipped for ${id}: ${snapErr.message}`);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
this.logger.log(`✅ Road candidate ${id} approved & published.`);
|
// A route can only enter GraphHopper when both endpoints share real OSM nodes.
|
||||||
return { success: true, id, status: 'approved' };
|
// Keep an explicit status instead of silently publishing an isolated line.
|
||||||
|
const [connected] = await this.dataSource.query(`
|
||||||
|
WITH road AS (
|
||||||
|
SELECT id, geometry FROM approved_roads WHERE candidate_id = $1::uuid
|
||||||
|
),
|
||||||
|
endpoints AS (
|
||||||
|
SELECT ST_StartPoint(geometry) AS start_pt, ST_EndPoint(geometry) AS end_pt FROM road
|
||||||
|
),
|
||||||
|
start_node AS (
|
||||||
|
SELECT n.id, ST_SetSRID(ST_MakePoint(n.lon / 1e7, n.lat / 1e7), 4326) AS point
|
||||||
|
FROM endpoints e
|
||||||
|
JOIN planet_osm_line l ON l.highway IS NOT NULL AND l.way && ST_Expand(ST_Transform(e.start_pt, 3857), 60)
|
||||||
|
JOIN planet_osm_ways w ON w.id = l.osm_id
|
||||||
|
CROSS JOIN LATERAL unnest(w.nodes) AS node_id
|
||||||
|
JOIN planet_osm_nodes n ON n.id = node_id
|
||||||
|
WHERE ST_Distance(ST_SetSRID(ST_MakePoint(n.lon / 1e7, n.lat / 1e7), 4326)::geography, e.start_pt::geography) <= 30
|
||||||
|
ORDER BY ST_Distance(ST_SetSRID(ST_MakePoint(n.lon / 1e7, n.lat / 1e7), 4326)::geography, e.start_pt::geography) LIMIT 1
|
||||||
|
),
|
||||||
|
end_node AS (
|
||||||
|
SELECT n.id, ST_SetSRID(ST_MakePoint(n.lon / 1e7, n.lat / 1e7), 4326) AS point
|
||||||
|
FROM endpoints e
|
||||||
|
JOIN planet_osm_line l ON l.highway IS NOT NULL AND l.way && ST_Expand(ST_Transform(e.end_pt, 3857), 60)
|
||||||
|
JOIN planet_osm_ways w ON w.id = l.osm_id
|
||||||
|
CROSS JOIN LATERAL unnest(w.nodes) AS node_id
|
||||||
|
JOIN planet_osm_nodes n ON n.id = node_id
|
||||||
|
WHERE ST_Distance(ST_SetSRID(ST_MakePoint(n.lon / 1e7, n.lat / 1e7), 4326)::geography, e.end_pt::geography) <= 30
|
||||||
|
ORDER BY ST_Distance(ST_SetSRID(ST_MakePoint(n.lon / 1e7, n.lat / 1e7), 4326)::geography, e.end_pt::geography) LIMIT 1
|
||||||
|
)
|
||||||
|
UPDATE approved_roads a
|
||||||
|
SET geometry = ST_SetPoint(ST_SetPoint(a.geometry, 0, s.point), ST_NPoints(a.geometry) - 1, e.point),
|
||||||
|
start_node = s.id, end_node = e.id,
|
||||||
|
routing_status = 'queued', routing_error = NULL, routing_requested_at = NOW()
|
||||||
|
FROM start_node s CROSS JOIN end_node e
|
||||||
|
WHERE a.candidate_id = $1::uuid
|
||||||
|
RETURNING a.id
|
||||||
|
`, [id]);
|
||||||
|
|
||||||
|
if (!connected) {
|
||||||
|
await this.dataSource.query(`UPDATE approved_roads
|
||||||
|
SET routing_status = 'needs_endpoint_connection',
|
||||||
|
routing_error = 'Both road endpoints must be within 30 metres of existing road nodes.'
|
||||||
|
WHERE candidate_id = $1::uuid`, [id]);
|
||||||
|
return { success: true, id, status: 'approved', routingStatus: 'needs_endpoint_connection' };
|
||||||
|
}
|
||||||
|
|
||||||
|
await this.redisService.set('routing_sync_requested', '1');
|
||||||
|
this.logger.log(`✅ Road candidate ${id} queued for routing rebuild.`);
|
||||||
|
return { success: true, id, status: 'approved', routingStatus: 'queued' };
|
||||||
} catch (e: any) {
|
} catch (e: any) {
|
||||||
this.logger.error(`Failed to approve candidate road: ${e.message}`);
|
this.logger.error(`Failed to approve candidate road: ${e.message}`);
|
||||||
throw new HttpException(e.message, HttpStatus.INTERNAL_SERVER_ERROR);
|
throw new HttpException(e.message, HttpStatus.INTERNAL_SERVER_ERROR);
|
||||||
|
|||||||
@@ -50,6 +50,11 @@ export class DriverTelemetryDto {
|
|||||||
return val != null ? Number(val) : 0;
|
return val != null ? Number(val) : 0;
|
||||||
})
|
})
|
||||||
elevation?: number;
|
elevation?: number;
|
||||||
|
|
||||||
|
@ApiPropertyOptional({ description: 'Client GPS capture time as ISO-8601', example: '2026-09-16T10:20:30.000Z' })
|
||||||
|
@IsOptional()
|
||||||
|
@IsString()
|
||||||
|
timestamp?: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
export class DriverTelemetryBatchDto {
|
export class DriverTelemetryBatchDto {
|
||||||
|
|||||||
@@ -25,14 +25,17 @@ export class TelemetryController {
|
|||||||
|
|
||||||
@Post('batch')
|
@Post('batch')
|
||||||
@ApiOperation({
|
@ApiOperation({
|
||||||
summary: 'Batch ingest driver telemetry points 📦',
|
summary: 'Queue navigation telemetry batch 📦',
|
||||||
description: 'Receives an array of telemetry points for offline-buffered sync or high-frequency traces.',
|
description: 'Acknowledges a navigation batch immediately; the server persists it from Redis in the background.',
|
||||||
})
|
})
|
||||||
async ingestBatch(@Body() body: DriverTelemetryBatchDto) {
|
async ingestBatch(@Body() body: DriverTelemetryBatchDto) {
|
||||||
if (!body || !Array.isArray(body.points)) {
|
if (!body || !Array.isArray(body.points)) {
|
||||||
throw new HttpException('Invalid payload: expected { points: [...] }', HttpStatus.BAD_REQUEST);
|
throw new HttpException('Invalid payload: expected { points: [...] }', HttpStatus.BAD_REQUEST);
|
||||||
}
|
}
|
||||||
return this.telemetryService.ingestBatch(body.points);
|
if (body.points.length > 100) {
|
||||||
|
throw new HttpException('A telemetry batch may contain at most 100 points', HttpStatus.BAD_REQUEST);
|
||||||
|
}
|
||||||
|
return this.telemetryService.enqueueBatch(body.points);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Get('nearby')
|
@Get('nearby')
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
|
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
||||||
import { InjectRepository } from '@nestjs/typeorm';
|
import { InjectRepository } from '@nestjs/typeorm';
|
||||||
import { Repository, DataSource } from 'typeorm';
|
import { Repository, DataSource } from 'typeorm';
|
||||||
import { TelemetryLog } from './telemetry.entity';
|
import { TelemetryLog } from './telemetry.entity';
|
||||||
@@ -6,8 +6,11 @@ import { DriverTelemetryDto } from './dto/driver-telemetry.dto';
|
|||||||
import { RedisService } from '../common/redis.service';
|
import { RedisService } from '../common/redis.service';
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
export class TelemetryService implements OnModuleInit {
|
export class TelemetryService implements OnModuleInit, OnModuleDestroy {
|
||||||
private readonly logger = new Logger(TelemetryService.name);
|
private readonly logger = new Logger(TelemetryService.name);
|
||||||
|
private readonly queueKey = 'telemetry:batch:queue';
|
||||||
|
private draining = false;
|
||||||
|
private queueTimer?: NodeJS.Timeout;
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
@InjectRepository(TelemetryLog)
|
@InjectRepository(TelemetryLog)
|
||||||
@@ -43,6 +46,50 @@ export class TelemetryService implements OnModuleInit {
|
|||||||
} catch (err: any) {
|
} catch (err: any) {
|
||||||
this.logger.warn(`Telemetry DB auto-migration check note: ${err.message}`);
|
this.logger.warn(`Telemetry DB auto-migration check note: ${err.message}`);
|
||||||
}
|
}
|
||||||
|
// The request path only queues data. A small background worker persists it
|
||||||
|
// in batches, keeping navigation uploads fast even when PostGIS is busy.
|
||||||
|
this.queueTimer = setInterval(() => void this.drainQueue(), 5_000);
|
||||||
|
void this.drainQueue();
|
||||||
|
}
|
||||||
|
|
||||||
|
onModuleDestroy() {
|
||||||
|
if (this.queueTimer) clearInterval(this.queueTimer);
|
||||||
|
}
|
||||||
|
|
||||||
|
async enqueueBatch(points: DriverTelemetryDto[]) {
|
||||||
|
if (!points?.length) return { success: true, accepted: 0, queued: 0 };
|
||||||
|
if (points.length > 100) {
|
||||||
|
throw new Error('A telemetry batch may contain at most 100 points.');
|
||||||
|
}
|
||||||
|
await this.redisService.getClient().lPush(this.queueKey, JSON.stringify(points));
|
||||||
|
const queuedBatches = await this.redisService.getClient().lLen(this.queueKey);
|
||||||
|
void this.drainQueue();
|
||||||
|
return { success: true, accepted: points.length, queuedBatches };
|
||||||
|
}
|
||||||
|
|
||||||
|
private async drainQueue() {
|
||||||
|
if (this.draining) return;
|
||||||
|
this.draining = true;
|
||||||
|
try {
|
||||||
|
// Limit a turn so requests and other Redis work stay responsive.
|
||||||
|
for (let i = 0; i < 20; i++) {
|
||||||
|
const raw = await this.redisService.getClient().rPop(this.queueKey);
|
||||||
|
if (!raw) break;
|
||||||
|
try {
|
||||||
|
const points = JSON.parse(raw) as DriverTelemetryDto[];
|
||||||
|
await this.ingestBatch(points);
|
||||||
|
} catch (error: any) {
|
||||||
|
// Put the original payload back at the head for a later retry.
|
||||||
|
await this.redisService.getClient().rPush(this.queueKey, raw);
|
||||||
|
this.logger.error(`Telemetry queue persistence failed: ${error.message}`);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (error: any) {
|
||||||
|
this.logger.error(`Telemetry queue worker failed: ${error.message}`);
|
||||||
|
} finally {
|
||||||
|
this.draining = false;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -64,7 +111,7 @@ export class TelemetryService implements OnModuleInit {
|
|||||||
heading,
|
heading,
|
||||||
distance,
|
distance,
|
||||||
elevation,
|
elevation,
|
||||||
timestamp: new Date(),
|
timestamp: this.captureTime(data.timestamp),
|
||||||
location: {
|
location: {
|
||||||
type: 'Point',
|
type: 'Point',
|
||||||
coordinates: [lng, lat],
|
coordinates: [lng, lat],
|
||||||
@@ -127,7 +174,7 @@ export class TelemetryService implements OnModuleInit {
|
|||||||
heading,
|
heading,
|
||||||
distance,
|
distance,
|
||||||
elevation,
|
elevation,
|
||||||
timestamp: new Date(),
|
timestamp: this.captureTime(p.timestamp),
|
||||||
location: {
|
location: {
|
||||||
type: 'Point',
|
type: 'Point',
|
||||||
coordinates: [lng, lat],
|
coordinates: [lng, lat],
|
||||||
@@ -168,6 +215,12 @@ export class TelemetryService implements OnModuleInit {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private captureTime(value?: string): Date {
|
||||||
|
if (!value) return new Date();
|
||||||
|
const date = new Date(value);
|
||||||
|
return Number.isNaN(date.getTime()) ? new Date() : date;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Find nearby active drivers using PostGIS spatial geography search
|
* Find nearby active drivers using PostGIS spatial geography search
|
||||||
*/
|
*/
|
||||||
|
|||||||
@@ -35,6 +35,7 @@ class TelemetryPoint {
|
|||||||
'heading': heading < 0 ? 0.0 : heading,
|
'heading': heading < 0 ? 0.0 : heading,
|
||||||
'elevation': elevation,
|
'elevation': elevation,
|
||||||
'distance': distance,
|
'distance': distance,
|
||||||
|
'timestamp': timestamp.toUtc().toIso8601String(),
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -66,7 +67,7 @@ class TelemetryTrackingService {
|
|||||||
List<TelemetryPoint> get bufferedPoints => List.unmodifiable(_buffer);
|
List<TelemetryPoint> get bufferedPoints => List.unmodifiable(_buffer);
|
||||||
DateTime? get lastFlushTime => _lastFlushTime;
|
DateTime? get lastFlushTime => _lastFlushTime;
|
||||||
|
|
||||||
/// Initializes periodic flush timer
|
/// Initializes periodic flush timer for an active navigation session.
|
||||||
void initialize() {
|
void initialize() {
|
||||||
_periodicFlushTimer?.cancel();
|
_periodicFlushTimer?.cancel();
|
||||||
_lastFlushTime = DateTime.now();
|
_lastFlushTime = DateTime.now();
|
||||||
|
|||||||
@@ -67,8 +67,7 @@ class NavigationCubit extends Cubit<NavigationState> {
|
|||||||
CarPlatformBridge.ensureInitialized();
|
CarPlatformBridge.ensureInitialized();
|
||||||
await ttsService.init();
|
await ttsService.init();
|
||||||
|
|
||||||
// Initialize 3-second telemetry tracking & 2-minute batch uploads
|
// The tracking session begins only once the user starts navigation.
|
||||||
TelemetryTrackingService.instance.initialize();
|
|
||||||
TelemetryTrackingService.instance.checkPeriodicUpdates();
|
TelemetryTrackingService.instance.checkPeriodicUpdates();
|
||||||
|
|
||||||
// Load saved vehicle & search preferences
|
// Load saved vehicle & search preferences
|
||||||
@@ -289,16 +288,6 @@ class NavigationCubit extends Cubit<NavigationState> {
|
|||||||
_startMovementInterpolation(newLoc, heading);
|
_startMovementInterpolation(newLoc, heading);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Continuous 3-second telemetry sampling with stationary filtering and 2-minute batching
|
|
||||||
TelemetryTrackingService.instance.recordPosition(
|
|
||||||
latitude: newLoc.latitude,
|
|
||||||
longitude: newLoc.longitude,
|
|
||||||
speedKmH: speedKmH,
|
|
||||||
heading: heading,
|
|
||||||
elevation: alt,
|
|
||||||
remainingDistance: state.remainingDistance,
|
|
||||||
);
|
|
||||||
|
|
||||||
// Proactively move camera on first acquired GPS lock (only when style is loaded)
|
// Proactively move camera on first acquired GPS lock (only when style is loaded)
|
||||||
if (!_hasInitiallyCenteredCamera && mapController != null && _isMapStyleLoaded) {
|
if (!_hasInitiallyCenteredCamera && mapController != null && _isMapStyleLoaded) {
|
||||||
_hasInitiallyCenteredCamera = true;
|
_hasInitiallyCenteredCamera = true;
|
||||||
@@ -310,6 +299,15 @@ class NavigationCubit extends Cubit<NavigationState> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (state.isNavigating) {
|
if (state.isNavigating) {
|
||||||
|
// Navigation-only telemetry: sampled every 3 seconds and sent as a batch every 2 minutes.
|
||||||
|
TelemetryTrackingService.instance.recordPosition(
|
||||||
|
latitude: newLoc.latitude,
|
||||||
|
longitude: newLoc.longitude,
|
||||||
|
speedKmH: speedKmH,
|
||||||
|
heading: heading,
|
||||||
|
elevation: alt,
|
||||||
|
remainingDistance: state.remainingDistance,
|
||||||
|
);
|
||||||
_processActiveNavigationTick(newLoc, speedKmH, heading);
|
_processActiveNavigationTick(newLoc, speedKmH, heading);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -718,6 +716,7 @@ class NavigationCubit extends Cubit<NavigationState> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
final route = state.currentRoute!;
|
final route = state.currentRoute!;
|
||||||
|
TelemetryTrackingService.instance.initialize();
|
||||||
final steps = route.steps;
|
final steps = route.steps;
|
||||||
print("🧭 [NavigationCubit] Starting navigation along route: ${route.formattedDistance}, ETA ${route.formattedDuration}, ${steps.length} steps");
|
print("🧭 [NavigationCubit] Starting navigation along route: ${route.formattedDistance}, ETA ${route.formattedDuration}, ${steps.length} steps");
|
||||||
|
|
||||||
@@ -774,7 +773,8 @@ class NavigationCubit extends Cubit<NavigationState> {
|
|||||||
|
|
||||||
void stopNavigation() {
|
void stopNavigation() {
|
||||||
print("🛑 [NavigationCubit] stopNavigation triggered.");
|
print("🛑 [NavigationCubit] stopNavigation triggered.");
|
||||||
TelemetryTrackingService.instance.flush(repository: repository);
|
// Stop the two-minute timer and send the final in-memory batch.
|
||||||
|
TelemetryTrackingService.instance.dispose(repository: repository);
|
||||||
_movementInterpolationTimer?.cancel();
|
_movementInterpolationTimer?.cancel();
|
||||||
_movementInterpolationTimer = null;
|
_movementInterpolationTimer = null;
|
||||||
ttsService.stop();
|
ttsService.stop();
|
||||||
|
|||||||
@@ -34,6 +34,7 @@ WITH ways AS (
|
|||||||
start_node, end_node,
|
start_node, end_node,
|
||||||
ROW_NUMBER() OVER (ORDER BY id) AS way_seq
|
ROW_NUMBER() OVER (ORDER BY id) AS way_seq
|
||||||
FROM approved_roads
|
FROM approved_roads
|
||||||
|
WHERE routing_status IN ('queued', 'syncing', 'routed')
|
||||||
),
|
),
|
||||||
pts AS (
|
pts AS (
|
||||||
SELECT w.way_seq, w.confidence, w.drivers, w.highway, w.oneway, w.start_node, w.end_node,
|
SELECT w.way_seq, w.confidence, w.drivers, w.highway, w.oneway, w.start_node, w.end_node,
|
||||||
|
|||||||
@@ -1,31 +1,53 @@
|
|||||||
#!/bin/bash
|
#!/bin/bash
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
# check_routing_sync.sh
|
# check_routing_sync.sh
|
||||||
# Runs via cron every minute to check if the admin dashboard requested
|
# Runs via cron every minute. The Redis request is cleared only after the
|
||||||
# a routing graph rebuild (due to a newly approved road).
|
# rebuilt engine is healthy, so failed rebuilds remain queued for retry.
|
||||||
# --------------------------------------------------------------------------
|
# --------------------------------------------------------------------------
|
||||||
|
|
||||||
APP_DIR="/home/hamzadoctor/app"
|
set -euo pipefail
|
||||||
|
|
||||||
|
APP_DIR="${APP_DIR:-/home/hamzadoctor/app}"
|
||||||
|
DATA_DIR="${APP_DIR}/infrastructure/osm-data"
|
||||||
|
LOCK_FILE="/tmp/map-saas-routing-sync.lock"
|
||||||
cd "${APP_DIR}"
|
cd "${APP_DIR}"
|
||||||
|
|
||||||
|
exec 9>"${LOCK_FILE}"
|
||||||
|
if ! flock -n 9; then
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
|
||||||
# Read flag from Redis using the redis container
|
# Read flag from Redis using the redis container
|
||||||
SYNC_REQ=$(docker compose exec -T redis redis-cli get routing_sync_requested | tr -d '\r')
|
SYNC_REQ=$(docker compose exec -T redis redis-cli get routing_sync_requested | tr -d '\r')
|
||||||
|
|
||||||
if [ "$SYNC_REQ" = "1" ] || [ "$SYNC_REQ" = "\"1\"" ]; then
|
if [ "$SYNC_REQ" = "1" ] || [ "$SYNC_REQ" = "\"1\"" ]; then
|
||||||
echo "$(date): Sync requested! Triggering delta apply and GH rebuild..."
|
echo "$(date -Is): Sync requested; rebuilding connected approved roads..."
|
||||||
|
docker compose exec -T db psql -U mapuser -d mapdb -c \
|
||||||
|
"UPDATE approved_roads SET routing_status='syncing', routing_error=NULL WHERE routing_status='queued';"
|
||||||
|
|
||||||
# 1. Delete the flag immediately so we don't trigger it again
|
if ! bash "${APP_DIR}/infrastructure/scripts/apply-delta.sh" "${APP_DIR}" "${DATA_DIR}/master_map.osm.pbf" "${DATA_DIR}/delta.osm"; then
|
||||||
docker compose exec -T redis redis-cli del routing_sync_requested
|
docker compose exec -T db psql -U mapuser -d mapdb -c \
|
||||||
|
"UPDATE approved_roads SET routing_status='queued', routing_error='Delta export or merge failed; queued for retry.' WHERE routing_status='syncing';"
|
||||||
# 2. Apply delta
|
exit 1
|
||||||
if [ -f "${APP_DIR}/infrastructure/scripts/apply-delta.sh" ]; then
|
|
||||||
bash "${APP_DIR}/infrastructure/scripts/apply-delta.sh"
|
|
||||||
fi
|
fi
|
||||||
|
|
||||||
# 3. Restart GraphHopper to rebuild index
|
|
||||||
docker compose stop routing
|
docker compose stop routing
|
||||||
rm -rf "${APP_DIR}/infrastructure/osm-data/graph-cache" "${APP_DIR}/infrastructure/osm-data/default-gh"
|
rm -rf "${DATA_DIR}/graph-cache" "${DATA_DIR}/default-gh"
|
||||||
docker compose up -d routing
|
docker compose up -d routing
|
||||||
|
|
||||||
echo "$(date): Routing rebuild triggered successfully."
|
for _ in $(seq 1 90); do
|
||||||
|
if curl -fsS http://localhost:8989/health >/dev/null 2>&1; then
|
||||||
|
docker compose exec -T db psql -U mapuser -d mapdb -c \
|
||||||
|
"UPDATE approved_roads SET routing_status='routed', routing_error=NULL, routed_at=NOW() WHERE routing_status='syncing';"
|
||||||
|
docker compose exec -T redis redis-cli del routing_sync_requested >/dev/null
|
||||||
|
echo "$(date -Is): Routing graph is ready; queued roads are routable."
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
sleep 10
|
||||||
|
done
|
||||||
|
|
||||||
|
docker compose exec -T db psql -U mapuser -d mapdb -c \
|
||||||
|
"UPDATE approved_roads SET routing_status='queued', routing_error='GraphHopper did not become healthy; queued for retry.' WHERE routing_status='syncing';"
|
||||||
|
echo "$(date -Is): GraphHopper did not become healthy; retaining sync request for retry." >&2
|
||||||
|
exit 1
|
||||||
fi
|
fi
|
||||||
|
|||||||
@@ -68,15 +68,12 @@ docker compose --profile import run --rm osm-import \
|
|||||||
--database mapdb --host db --user mapuser \
|
--database mapdb --host db --user mapuser \
|
||||||
/data/iraq-latest.osm.pbf
|
/data/iraq-latest.osm.pbf
|
||||||
|
|
||||||
# ── Step 2b: إسقاط جداول osm2pgsql الوسيطة ────────────────────────────────
|
# ── Step 2b: Keep routing topology tables ─────────────────────────────────
|
||||||
# planet_osm_nodes/ways/rels ينشئها --slim لدعم --append فقط، ولا يقرأها
|
# Approved roads must share real OSM node IDs with the base graph. Deleting
|
||||||
# التطبيق ولا martin إطلاقاً (جداول الإخراج هي line/polygon/point/roads).
|
# nodes/ways makes Martin draw a road but leaves GraphHopper unable to connect it.
|
||||||
# قياس 30/07/2026: nodes 5031MB + ways 1400MB ≈ 6.4GB نائمة بين الدورات.
|
echo "🧭 Preserving OSM node/way topology for approved-road routing..."
|
||||||
# آمن لأن --create في الخطوة 2 يعيد إنشاءها من الصفر كل تشغيل.
|
|
||||||
# ⚠️ لا تشغّل osm2pgsql --append بعد هذه النقطة حتى الدورة القادمة.
|
|
||||||
echo "🧹 إسقاط جداول osm2pgsql الوسيطة (تُعاد في الدورة القادمة)..."
|
|
||||||
docker compose exec -T db psql -U mapuser -d mapdb -c \
|
docker compose exec -T db psql -U mapuser -d mapdb -c \
|
||||||
"DROP TABLE IF EXISTS planet_osm_nodes, planet_osm_ways, planet_osm_rels;"
|
"DROP TABLE IF EXISTS planet_osm_rels;"
|
||||||
|
|
||||||
# ── Step 3: Merge all four PBFs into master_map.osm.pbf ────────────────────
|
# ── Step 3: Merge all four PBFs into master_map.osm.pbf ────────────────────
|
||||||
# ✅ FIX: Old script wrote to region.osm.pbf — GraphHopper reads master_map.osm.pbf
|
# ✅ FIX: Old script wrote to region.osm.pbf — GraphHopper reads master_map.osm.pbf
|
||||||
|
|||||||
Reference in New Issue
Block a user