-
-
- Hi, {currentUser?.full_name || currentUser?.email} 👋
-
-
- Welcome back, nice to see you again!!!
-
+
+
+ {health.level === "good"
+ ? "Good"
+ : health.level === "fair"
+ ? "Fair"
+ : "Poor"}{" "}
+ ({health.score}%)
+
+ )
+}
+
+function projectCoord(
+ lat: number,
+ lon: number,
+ useGcj02: boolean,
+): [number, number] {
+ if (!useGcj02) return [lat, lon]
+ return wgs84ToGcj02(lat, lon)
+}
+
+/* ---------- map sub-components ---------- */
+
+function FitBoundsToDevices({
+ coordinates,
+}: {
+ coordinates: Array<[number, number]>
+}) {
+ const map = useMap()
+ const hasFitted = useRef(false)
+
+ useEffect(() => {
+ if (coordinates.length === 0 || hasFitted.current) return
+ hasFitted.current = true
+ if (coordinates.length === 1) {
+ map.setView(coordinates[0], 14, { animate: true })
+ return
+ }
+ map.fitBounds(coordinates, { padding: [40, 40] })
+ }, [coordinates, map])
+
+ return null
+}
+
+/** Badges-only overlay — lives inside MapContainer, no dropdown needed */
+function MapStatusBadges({
+ isLive,
+ status,
+ deviceCount,
+ disconnectedDuration,
+}: {
+ isLive: boolean
+ status: ConnectionState
+ deviceCount: number
+ disconnectedDuration: number
+}) {
+ return (
+
+
+
+ {isLive && (
+
+
+
+
+ )}
+ {!isLive && }
+ {isLive
+ ? "Live"
+ : status === "retrying"
+ ? `Reconnecting${disconnectedDuration > 0 ? ` (${disconnectedDuration}s)` : ""}`
+ : "Connecting"}
+
+
+ Devices: {deviceCount}
+
+
+
+ )
+}
+
+/** FitBounds helper — must live inside MapContainer to call useMap() */
+function FitBoundsControl({
+ coordinates,
+}: {
+ coordinates: Array<[number, number]>
+}) {
+ const map = useMap()
+ const fitAll = useCallback(() => {
+ if (!coordinates.length) return
+ if (coordinates.length === 1) {
+ map.setView(coordinates[0], 14, { animate: true })
+ } else {
+ map.fitBounds(coordinates, { padding: [40, 40] })
+ }
+ }, [coordinates, map])
+
+ const focusDevice = useCallback(
+ (coordinate: [number, number]) => {
+ const maxZoom = map.getMaxZoom()
+ const targetZoom = Number.isFinite(maxZoom) ? maxZoom : 18
+ map.setView(coordinate, targetZoom, { animate: true })
+ },
+ [map],
+ )
+
+ useEffect(() => {
+ const ref = window as unknown as MapControlBridge
+ ref.__mapFitAll = fitAll
+ ref.__mapFocusDevice = focusDevice
+ return () => {
+ if (ref.__mapFitAll === fitAll) {
+ delete ref.__mapFitAll
+ }
+ if (ref.__mapFocusDevice === focusDevice) {
+ delete ref.__mapFocusDevice
+ }
+ }
+ }, [fitAll, focusDevice])
+
+ return null
+}
+
+/* ---------- WebSocket hook ---------- */
+
+function useTelemetryWS() {
+ const [snapshot, setSnapshot] = useState
(null)
+ const [status, setStatus] = useState("connecting")
+ const [disconnectedAt, setDisconnectedAt] = useState(null)
+ const retryCount = useRef(0)
+ const lastMessageAt = useRef(0)
+
+ useEffect(() => {
+ const token = localStorage.getItem("access_token")
+ if (!token) {
+ setStatus("retrying")
+ return
+ }
+
+ let reconnectTimer: number | undefined
+ let staleTimer: number | undefined
+ let socket: WebSocket | undefined
+ let shouldReconnect = true
+
+ const STALE_TIMEOUT_MS = 10_000
+
+ const resetStaleTimer = () => {
+ if (staleTimer) window.clearTimeout(staleTimer)
+ staleTimer = window.setTimeout(() => {
+ socket?.close()
+ }, STALE_TIMEOUT_MS)
+ }
+
+ const connect = () => {
+ setStatus((prev) => (prev === "live" ? prev : "connecting"))
+ socket = new WebSocket(buildTelemetryWsUrl(token))
+
+ socket.onopen = () => {
+ setStatus("live")
+ setDisconnectedAt(null)
+ retryCount.current = 0
+ lastMessageAt.current = Date.now()
+ resetStaleTimer()
+ }
+
+ socket.onmessage = (event) => {
+ try {
+ const payload = JSON.parse(event.data) as TelemetrySnapshot
+ setSnapshot(payload)
+ setStatus("live")
+ setDisconnectedAt(null)
+ lastMessageAt.current = Date.now()
+ resetStaleTimer()
+ } catch {
+ // ignore malformed messages
+ }
+ }
+
+ socket.onclose = () => {
+ if (staleTimer) window.clearTimeout(staleTimer)
+ if (!shouldReconnect) return
+ setStatus("retrying")
+ setDisconnectedAt((prev) => prev ?? Date.now())
+ const backoffMs = Math.min(2000 * 2 ** retryCount.current, 30000)
+ retryCount.current++
+ reconnectTimer = window.setTimeout(connect, backoffMs)
+ }
+
+ socket.onerror = () => {
+ socket?.close()
+ }
+ }
+
+ connect()
+
+ return () => {
+ shouldReconnect = false
+ if (reconnectTimer) window.clearTimeout(reconnectTimer)
+ if (staleTimer) window.clearTimeout(staleTimer)
+ if (socket) socket.close()
+ }
+ }, [])
+
+ return { snapshot, status, disconnectedAt }
+}
+
+/* ---------- Dashboard ---------- */
+
+function Dashboard() {
+ const { snapshot, status, disconnectedAt } = useTelemetryWS()
+ const [baseMap, setBaseMap] = useState("amap")
+ const [activeDeviceId, setActiveDeviceId] = useState("")
+ const [isPanelExpanded, setIsPanelExpanded] = useState(true)
+ const activeDeviceIdRef = useRef(activeDeviceId)
+ activeDeviceIdRef.current = activeDeviceId
+
+ const isGcj02 = BASE_MAPS[baseMap].gcj02
+
+ const devicesWithCoords = useMemo(
+ () =>
+ (snapshot?.devices ?? []).filter(
+ (device) =>
+ device.lat !== null &&
+ device.lon !== null &&
+ device.lat >= -90 &&
+ device.lat <= 90 &&
+ device.lon >= -180 &&
+ device.lon <= 180,
+ ),
+ [snapshot],
+ )
+
+ useEffect(() => {
+ if (!devicesWithCoords.length) {
+ if (activeDeviceIdRef.current) setActiveDeviceId("")
+ return
+ }
+ const exists = devicesWithCoords.some(
+ (d) => d.device_id === activeDeviceIdRef.current,
+ )
+ if (!exists && !activeDeviceIdRef.current) {
+ setActiveDeviceId(devicesWithCoords[0].device_id)
+ }
+ }, [devicesWithCoords])
+
+ const activeDevice = useMemo(
+ () =>
+ devicesWithCoords.find((d) => d.device_id === activeDeviceId) ??
+ devicesWithCoords[0] ??
+ null,
+ [devicesWithCoords, activeDeviceId],
+ )
+
+ const mapPoints = useMemo(
+ () =>
+ devicesWithCoords.map((d) =>
+ projectCoord(d.lat as number, d.lon as number, isGcj02),
+ ),
+ [devicesWithCoords, isGcj02],
+ )
+
+ const handleDeviceClick = useCallback(
+ (deviceId: string, coordinate: [number, number]) => {
+ setActiveDeviceId(deviceId)
+ const mapBridge = window as unknown as MapControlBridge
+ mapBridge.__mapFocusDevice?.(coordinate)
+ },
+ [],
+ )
+
+ const isLive = status === "live"
+
+ // Disconnected time ticker
+ const [, forceUpdate] = useState(0)
+ useEffect(() => {
+ if (!disconnectedAt) return
+ const interval = window.setInterval(() => forceUpdate((n) => n + 1), 1000)
+ return () => window.clearInterval(interval)
+ }, [disconnectedAt])
+
+ const disconnectedDuration = disconnectedAt
+ ? Math.floor((Date.now() - disconnectedAt) / 1000)
+ : 0
+
+ return (
+
+ {/* Map area — fills available space */}
+
+ {/* Controls live OUTSIDE MapContainer to avoid Leaflet z-index stacking issues */}
+
+
+
+
+
+
+ {(
+ Object.entries(BASE_MAPS) as Array<
+ [BaseMapType, (typeof BASE_MAPS)[BaseMapType]]
+ >
+ ).map(([key, provider]) => (
+ setBaseMap(key)}>
+ {provider.label} {baseMap === key && "✓"}
+
+ ))}
+
+
+
+
+
+
+
+
+
+
+
+
+ {devicesWithCoords.map((device) => {
+ const isOffline = checkIsOffline(device.last_event_time)
+ const isSelected = activeDeviceId === device.device_id
+ const projectedCoordinate = projectCoord(
+ device.lat as number,
+ device.lon as number,
+ isGcj02,
+ )
+ const [projLat, projLon] = projectedCoordinate
+ const color = deviceMarkerColor(device.fix_status, isOffline)
+ const icon = buildDeviceIcon(color, isSelected, isOffline)
+
+ return (
+
+ handleDeviceClick(device.device_id, projectedCoordinate),
+ }}
+ key={device.device_id}
+ >
+
+
+
+ {device.device_id}
+ {isOffline && (
+
+ (Offline)
+
+ )}
+
+
+ Lat:
+ {formatCoord(device.lat)}
+
+ Lon:
+ {formatCoord(device.lon)}
+
+ Alt:
+
+ {device.alt_m !== null
+ ? `${device.alt_m.toFixed(1)} m`
+ : "-"}
+
+
+ Fix:
+ {getFixStatusLabel(device.fix_status)}
+
+ Sats:
+ {device.satellites_used ?? "-"}
+
+
+ Last seen:{" "}
+ {device.last_event_time
+ ? new Date(device.last_event_time).toLocaleString()
+ : "Unknown"}
+
+
+
+
+ {/* Selected Device Panel Overlay */}
+
+
+ setIsPanelExpanded(!isPanelExpanded)}
+ >
+
+ Selected Device
+ {activeDevice?.signal_health &&
+ !isPanelExpanded &&
+ signalHealthBadge(activeDevice.signal_health)}
+
+
+ {activeDevice?.signal_health &&
+ isPanelExpanded &&
+ signalHealthBadge(activeDevice.signal_health)}
+
+
+
+ {isPanelExpanded && (
+
+
+
Device ID
+
+ {activeDevice?.device_id ?? "-"}
+
+
+
Latitude
+
+ {formatCoord(activeDevice?.lat ?? null)}
+
+
+
Longitude
+
+ {formatCoord(activeDevice?.lon ?? null)}
+
+
+
+
+ Altitude
+
+
+ {activeDevice?.alt_m !== null &&
+ activeDevice?.alt_m !== undefined
+ ? `${activeDevice.alt_m.toFixed(1)} m`
+ : "-"}
+
+
+
Fix Status
+
+ {getFixStatusLabel(activeDevice?.fix_status ?? null)}
+
+
+
+
+ Satellites
+
+
+ {activeDevice?.satellites_used ?? "-"}
+
+
+
+
+ SNR
+
+
+ {activeDevice?.signal_health
+ ? `${activeDevice.signal_health.score}%`
+ : "-"}
+
+
+
+
+ Last Seen
+
+
+ {timeAgo(activeDevice?.last_event_time ?? null)}
+
+
+
- You don't have any locations yet
- Add a new location to get started
+
+ You don't have any locations yet
+
+
+ Add a new location to get started
+
)
}
@@ -59,7 +63,9 @@ function Locations() {
Locations
-
Create and manage your locations
+
+ Create and manage your locations
+
diff --git a/frontend/src/routes/_layout/mqtt-raw.tsx b/frontend/src/routes/_layout/mqtt-raw.tsx
new file mode 100644
index 0000000..4530dbe
--- /dev/null
+++ b/frontend/src/routes/_layout/mqtt-raw.tsx
@@ -0,0 +1,411 @@
+import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"
+import { createFileRoute } from "@tanstack/react-router"
+import type { ColumnDef } from "@tanstack/react-table"
+import { Search } from "lucide-react"
+import { useMemo, useState } from "react"
+
+import { type MqttRawDaySummary, MqttRawService } from "@/client"
+import { DataTable } from "@/components/Common/DataTable"
+import PendingSkeleton from "@/components/Pending/PendingSkeleton"
+import { Button } from "@/components/ui/button"
+import {
+ Card,
+ CardContent,
+ CardDescription,
+ CardHeader,
+ CardTitle,
+} from "@/components/ui/card"
+import {
+ Dialog,
+ DialogClose,
+ DialogContent,
+ DialogDescription,
+ DialogFooter,
+ DialogHeader,
+ DialogTitle,
+} from "@/components/ui/dialog"
+import { Input } from "@/components/ui/input"
+import { LoadingButton } from "@/components/ui/loading-button"
+import { Slider } from "@/components/ui/slider"
+import useCustomToast from "@/hooks/useCustomToast"
+
+const MINUTE_STEPS_PER_DAY = 24 * 60
+
+export const Route = createFileRoute("/_layout/mqtt-raw")({
+ component: MqttRawPage,
+ head: () => ({
+ meta: [
+ {
+ title: "MQTT Raw Data - FastAPI Template",
+ },
+ ],
+ }),
+})
+
+function toUtc8Date(value: Date): Date {
+ const utcMillis = value.getTime() + value.getTimezoneOffset() * 60 * 1000
+ return new Date(utcMillis + 8 * 60 * 60 * 1000)
+}
+
+function todayIsoDateUtc8(): string {
+ const now = new Date()
+ return toUtc8Date(now).toISOString().slice(0, 10)
+}
+
+function formatDateTime(value: string | null | undefined): string {
+ if (!value) {
+ return "-"
+ }
+ const utc8 = toUtc8Date(new Date(value))
+ return `${utc8.toISOString().slice(0, 19).replace("T", " ")} (UTC+8)`
+}
+
+function formatMinuteLabel(minute: number): string {
+ const clamped = Math.max(0, Math.min(minute, MINUTE_STEPS_PER_DAY))
+ const hours = Math.floor(clamped / 60)
+ const minutes = clamped % 60
+ return `${String(hours).padStart(2, "0")}:${String(minutes).padStart(2, "0")}`
+}
+
+function triggerFileDownload(blob: Blob, filename: string): void {
+ const url = URL.createObjectURL(blob)
+ const link = document.createElement("a")
+ link.href = url
+ link.download = filename
+ document.body.appendChild(link)
+ link.click()
+ document.body.removeChild(link)
+ URL.revokeObjectURL(url)
+}
+
+function toDownloadBlob(value: unknown): Blob {
+ if (value instanceof Blob) {
+ return value
+ }
+ if (value instanceof ArrayBuffer) {
+ return new Blob([value], { type: "application/octet-stream" })
+ }
+ if (typeof value === "string") {
+ return new Blob([value], { type: "application/octet-stream" })
+ }
+ return new Blob([JSON.stringify(value)], { type: "application/octet-stream" })
+}
+
+function MqttRawPage() {
+ const queryClient = useQueryClient()
+ const { showSuccessToast, showErrorToast } = useCustomToast()
+
+ const [selectedDay, setSelectedDay] = useState(todayIsoDateUtc8())
+ const [rangeStartMinute, setRangeStartMinute] = useState(0)
+ const [rangeEndMinute, setRangeEndMinute] = useState(MINUTE_STEPS_PER_DAY)
+ const [deleteDayOpen, setDeleteDayOpen] = useState(false)
+ const [deleteAllOpen, setDeleteAllOpen] = useState(false)
+ const [deleteAllConfirmText, setDeleteAllConfirmText] = useState("")
+
+ const {
+ data: daysResponse,
+ isPending,
+ isError,
+ error,
+ } = useQuery({
+ queryFn: () => MqttRawService.listDays({ limit: 200, skip: 0 }),
+ queryKey: ["mqtt-raw-days"],
+ })
+
+ const days = daysResponse?.data ?? []
+ const isFullDayRange =
+ rangeStartMinute === 0 && rangeEndMinute === MINUTE_STEPS_PER_DAY
+ const startSeconds = rangeStartMinute * 60
+ const endSeconds = rangeEndMinute * 60
+
+ const downloadMutation = useMutation({
+ mutationFn: async () => {
+ const data = await MqttRawService.download({
+ day: selectedDay,
+ startSeconds: isFullDayRange ? undefined : startSeconds,
+ endSeconds: isFullDayRange ? undefined : endSeconds,
+ })
+
+ const filename = isFullDayRange
+ ? `mqtt_raw_${selectedDay}.txt`
+ : `mqtt_raw_${selectedDay}_${startSeconds}_${endSeconds}.txt`
+ return {
+ blob: toDownloadBlob(data),
+ filename,
+ }
+ },
+ onSuccess: ({ blob, filename }) => {
+ triggerFileDownload(blob, filename)
+ showSuccessToast("Raw MQTT data downloaded")
+ },
+ onError: (downloadError) => {
+ const message =
+ downloadError instanceof Error
+ ? downloadError.message
+ : "Download failed"
+ showErrorToast(message)
+ },
+ })
+
+ const deleteDayMutation = useMutation({
+ mutationFn: () => MqttRawService.deleteDay({ day: selectedDay }),
+ onSuccess: (result) => {
+ showSuccessToast(result.message)
+ setDeleteDayOpen(false)
+ queryClient.invalidateQueries({ queryKey: ["mqtt-raw-days"] })
+ },
+ onError: (deleteError) => {
+ const message =
+ deleteError instanceof Error ? deleteError.message : "Delete failed"
+ showErrorToast(message)
+ },
+ })
+
+ const deleteAllMutation = useMutation({
+ mutationFn: () => MqttRawService.deleteAll({ confirm: true }),
+ onSuccess: (result) => {
+ showSuccessToast(result.message)
+ setDeleteAllConfirmText("")
+ setDeleteAllOpen(false)
+ queryClient.invalidateQueries({ queryKey: ["mqtt-raw-days"] })
+ },
+ onError: (deleteError) => {
+ const message =
+ deleteError instanceof Error ? deleteError.message : "Delete failed"
+ showErrorToast(message)
+ },
+ })
+
+ const columns = useMemo
[]>(
+ () => [
+ {
+ accessorKey: "day",
+ header: "Day (UTC+8)",
+ cell: ({ row }) => (
+
+ ),
+ },
+ {
+ accessorKey: "message_count",
+ header: "Messages",
+ cell: ({ row }) => (
+ {row.original.message_count}
+ ),
+ },
+ {
+ accessorKey: "first_received_at",
+ header: "First Message",
+ cell: ({ row }) => formatDateTime(row.original.first_received_at),
+ },
+ {
+ accessorKey: "last_received_at",
+ header: "Last Message",
+ cell: ({ row }) => formatDateTime(row.original.last_received_at),
+ },
+ ],
+ [],
+ )
+
+ return (
+
+
+
MQTT Raw Data
+
+ Download saved payload bytes by UTC+8 day/time range and clean old
+ data.
+
+
+
+
+
+ Download
+
+ Pick a UTC+8 day, then download the full day or a custom time range.
+
+
+
+
+
+
Date (UTC+8)
+
setSelectedDay(event.target.value)}
+ type="date"
+ value={selectedDay}
+ />
+
+
+
+ downloadMutation.mutate()}
+ type="button"
+ >
+ Download
+
+
+
+
+
+ Start: {formatMinuteLabel(rangeStartMinute)}
+
+
+ End: {formatMinuteLabel(rangeEndMinute)}
+
+
+
{
+ if (values.length !== 2) {
+ return
+ }
+ setRangeStartMinute(values[0])
+ setRangeEndMinute(values[1])
+ }}
+ step={1}
+ value={[rangeStartMinute, rangeEndMinute]}
+ />
+
+ {isFullDayRange
+ ? "Current selection: full day (00:00-24:00, UTC+8)."
+ : `Current selection: ${formatMinuteLabel(rangeStartMinute)}-${formatMinuteLabel(rangeEndMinute)}.`}
+
+
+
+
+
+ Delete
+
+ Delete data for one UTC+8 day or delete all saved MQTT raw data.
+
+
+
+
+
+
+
+
+
+
Saved Days (UTC+8)
+
+ Click a day to load it into the controls above.
+
+ {isPending ? (
+
+ ) : isError ? (
+
+ {error instanceof Error
+ ? error.message
+ : "Failed to load data days."}
+
+ ) : days.length === 0 ? (
+
+
+
+
+
No raw data saved yet
+
+ MQTT payloads will appear here after ingestion.
+
+
+ ) : (
+
+ )}
+
+
+
+
+
+
@@ -22,8 +22,7 @@
- 📫 Email based password recovery.
- 📬 [Mailcatcher](https://mailcatcher.me) for local email testing during development.
- ✅ Tests with [Pytest](https://pytest.org).
-- 📞 [Traefik](https://traefik.io) as a reverse proxy / load balancer.
-- 🚢 Deployment instructions using Docker Compose, including how to set up a frontend Traefik proxy to handle automatic HTTPS certificates.
+- 🚢 Deployment instructions using Docker Compose.
- 🏭 CI (continuous integration) and CD (continuous deployment) based on GitHub Actions.
### Dashboard Login
@@ -146,66 +145,6 @@ python -c "import secrets; print(secrets.token_urlsafe(32))"
Copy the content and use that as password / secret key. And run that again to generate another secure key.
-## How To Use It - Alternative With Copier
-
-This repository also supports generating a new project using [Copier](https://copier.readthedocs.io).
-
-It will copy all the files, ask you configuration questions, and update the `.env` files with your answers.
-
-### Install Copier
-
-You can install Copier with:
-
-```bash
-pip install copier
-```
-
-Or better, if you have [`pipx`](https://pipx.pypa.io/), you can run it with:
-
-```bash
-pipx install copier
-```
-
-**Note**: If you have `pipx`, installing copier is optional, you could run it directly.
-
-### Generate a Project With Copier
-
-Decide a name for your new project's directory, you will use it below. For example, `my-awesome-project`.
-
-Go to the directory that will be the parent of your project, and run the command with your project's name:
-
-```bash
-copier copy https://github.com/fastapi/full-stack-fastapi-template my-awesome-project --trust
-```
-
-If you have `pipx` and you didn't install `copier`, you can run it directly:
-
-```bash
-pipx run copier copy https://github.com/fastapi/full-stack-fastapi-template my-awesome-project --trust
-```
-
-**Note** the `--trust` option is necessary to be able to execute a [post-creation script](https://github.com/fastapi/full-stack-fastapi-template/blob/master/.copier/update_dotenv.py) that updates your `.env` files.
-
-### Input Variables
-
-Copier will ask you for some data, you might want to have at hand before generating the project.
-
-But don't worry, you can just update any of that in the `.env` files afterwards.
-
-The input variables, with their default values (some auto generated) are:
-
-- `project_name`: (default: `"FastAPI Project"`) The name of the project, shown to API users (in .env).
-- `stack_name`: (default: `"fastapi-project"`) The name of the stack used for Docker Compose labels and project name (no spaces, no periods) (in .env).
-- `secret_key`: (default: `"changethis"`) The secret key for the project, used for security, stored in .env, you can generate one with the method above.
-- `first_superuser`: (default: `"admin@example.com"`) The email of the first superuser (in .env).
-- `first_superuser_password`: (default: `"changethis"`) The password of the first superuser (in .env).
-- `smtp_host`: (default: "") The SMTP server host to send emails, you can set it later in .env.
-- `smtp_user`: (default: "") The SMTP server user to send emails, you can set it later in .env.
-- `smtp_password`: (default: "") The SMTP server password to send emails, you can set it later in .env.
-- `emails_from_email`: (default: `"info@example.com"`) The email account to send emails from, you can set it later in .env.
-- `postgres_password`: (default: `"changethis"`) The password for the PostgreSQL database, stored in .env, you can generate one with the method above.
-- `sentry_dsn`: (default: "") The DSN for Sentry, if you are using it, you can set it later in .env.
-
## Backend Development
Backend docs: [backend/README.md](./backend/README.md).
@@ -222,7 +161,7 @@ Deployment docs: [deployment.md](./deployment.md).
General development docs: [development.md](./development.md).
-This includes using Docker Compose, custom local domains, `.env` configurations, etc.
+This includes using Docker Compose and `.env` configurations.
## Release Notes
@@ -230,4 +169,4 @@ Check the file [release-notes.md](./release-notes.md).
## License
-The Full Stack FastAPI Template is licensed under the terms of the MIT license.
+The SpatialHub is licensed under the terms of the MIT license.
diff --git a/backend/README.md b/backend/README.md
index 521ee3f..f7caf70 100644
--- a/backend/README.md
+++ b/backend/README.md
@@ -170,3 +170,40 @@ The email templates are in `./backend/app/email-templates/`. Here, there are two
Before continuing, ensure you have the [MJML extension](https://github.com/mjmlio/vscode-mjml) installed in your VS Code.
Once you have the MJML extension installed, you can create a new email template in the `src` directory. After creating the new email template and with the `.mjml` file open in your editor, open the command palette with `Ctrl+Shift+P` and search for `MJML: Export to HTML`. This will convert the `.mjml` file to a `.html` file and now you can save it in the build directory.
+
+## MQTT Ingestion Pipeline
+
+This project includes a two-stage MQTT pipeline:
+
+- `mqtt-ingestor` subscribes to `terminal/{device_id}/raw` and stores payload bytes losslessly in `terminal_message_raw`.
+- `mqtt-parser-worker` consumes unparsed rows (`parsed_at IS NULL` and `parse_error IS NULL`) and updates `terminal_device_state`.
+- Parse is single-pass by design in the current implementation: failed rows keep `parse_error` and are not auto-retried with backoff.
+
+### Run locally with Docker Compose
+
+```console
+$ docker compose up -d db backend mqtt-ingestor mqtt-parser-worker
+```
+
+### Unit tests for MQTT modules
+
+The existing `tests/conftest.py` expects a PostgreSQL connection. For parser/repository unit tests that run with in-memory SQLite fixtures, use:
+
+```console
+$ cd backend
+$ PYTHONPATH=. ../.venv/bin/pytest tests/mqtt --confcutdir=tests/mqtt -q
+```
+
+### Verify end-to-end ingestion quickly
+
+1. Start services (`db`, `backend`, `mqtt-ingestor`, `mqtt-parser-worker`).
+2. Publish a sample message to `terminal/device-001/raw`.
+3. Query PostgreSQL tables:
+ - `terminal_message_raw`
+ - `terminal_device_state`
+
+### Archive run (manual trigger)
+
+```console
+$ docker compose run --rm mqtt-parser-worker python -m app.mqtt.archive_job
+```
diff --git a/backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py b/backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py
new file mode 100644
index 0000000..f6088d9
--- /dev/null
+++ b/backend/app/alembic/versions/a1b2c3d4e5f6_add_mqtt_pipeline_tables.py
@@ -0,0 +1,223 @@
+"""add mqtt pipeline tables
+
+Revision ID: a1b2c3d4e5f6
+Revises: 0b117fab4208
+Create Date: 2026-03-17 12:30:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+
+# revision identifiers, used by Alembic.
+revision = "a1b2c3d4e5f6"
+down_revision = "0b117fab4208"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.create_table(
+ "terminal_message_raw",
+ sa.Column("id", sa.Uuid(), nullable=False),
+ sa.Column(
+ "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+ ),
+ sa.Column("topic", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+ sa.Column("received_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("payload_bytes", sa.LargeBinary(), nullable=False),
+ sa.Column("payload_size", sa.Integer(), nullable=False),
+ sa.Column(
+ "payload_sha256", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False
+ ),
+ sa.Column("qos", sa.Integer(), nullable=False),
+ sa.Column("retain", sa.Boolean(), nullable=False),
+ sa.Column(
+ "parse_status", sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False
+ ),
+ sa.Column("parse_attempts", sa.Integer(), nullable=False),
+ sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True),
+ sa.Column(
+ "parse_error", sqlmodel.sql.sqltypes.AutoString(length=1024), nullable=True
+ ),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_device_id"),
+ "terminal_message_raw",
+ ["device_id"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_next_retry_at"),
+ "terminal_message_raw",
+ ["next_retry_at"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_parse_status"),
+ "terminal_message_raw",
+ ["parse_status"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_payload_sha256"),
+ "terminal_message_raw",
+ ["payload_sha256"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_received_at"),
+ "terminal_message_raw",
+ ["received_at"],
+ unique=False,
+ )
+
+ op.create_table(
+ "terminal_observation_event",
+ sa.Column("id", sa.Uuid(), nullable=False),
+ sa.Column("raw_id", sa.Uuid(), nullable=False),
+ sa.Column(
+ "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+ ),
+ sa.Column("event_time", sa.DateTime(timezone=True), nullable=False),
+ sa.Column(
+ "event_type", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False
+ ),
+ sa.Column("payload_json", sa.JSON(), nullable=False),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.ForeignKeyConstraint(["raw_id"], ["terminal_message_raw.id"]),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_device_id"),
+ "terminal_observation_event",
+ ["device_id"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_event_time"),
+ "terminal_observation_event",
+ ["event_time"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_event_type"),
+ "terminal_observation_event",
+ ["event_type"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_raw_id"),
+ "terminal_observation_event",
+ ["raw_id"],
+ unique=False,
+ )
+
+ op.create_table(
+ "terminal_device_state",
+ sa.Column(
+ "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+ ),
+ sa.Column("last_event_time", sa.DateTime(timezone=True), nullable=True),
+ sa.Column("fix_status", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
+ sa.Column("lat", sa.Float(), nullable=True),
+ sa.Column("lon", sa.Float(), nullable=True),
+ sa.Column("alt_m", sa.Float(), nullable=True),
+ sa.Column("satellites_used", sa.Integer(), nullable=True),
+ sa.Column("satellite_signal_summary", sa.JSON(), nullable=True),
+ sa.Column("rtcm_type_summary", sa.JSON(), nullable=True),
+ sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+ sa.PrimaryKeyConstraint("device_id"),
+ )
+
+ op.create_table(
+ "terminal_raw_archive_manifest",
+ sa.Column("id", sa.Uuid(), nullable=False),
+ sa.Column(
+ "device_id", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+ ),
+ sa.Column("range_start", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("range_end", sa.DateTime(timezone=True), nullable=False),
+ sa.Column(
+ "file_path", sqlmodel.sql.sqltypes.AutoString(length=1024), nullable=False
+ ),
+ sa.Column("file_size", sa.Integer(), nullable=False),
+ sa.Column("sha256", sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
+ sa.Column("record_count", sa.Integer(), nullable=False),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.PrimaryKeyConstraint("id"),
+ sa.UniqueConstraint("file_path"),
+ )
+ op.create_index(
+ op.f("ix_terminal_raw_archive_manifest_device_id"),
+ "terminal_raw_archive_manifest",
+ ["device_id"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_raw_archive_manifest_range_end"),
+ "terminal_raw_archive_manifest",
+ ["range_end"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_raw_archive_manifest_range_start"),
+ "terminal_raw_archive_manifest",
+ ["range_start"],
+ unique=False,
+ )
+
+
+def downgrade() -> None:
+ op.drop_index(
+ op.f("ix_terminal_raw_archive_manifest_range_start"),
+ table_name="terminal_raw_archive_manifest",
+ )
+ op.drop_index(
+ op.f("ix_terminal_raw_archive_manifest_range_end"),
+ table_name="terminal_raw_archive_manifest",
+ )
+ op.drop_index(
+ op.f("ix_terminal_raw_archive_manifest_device_id"),
+ table_name="terminal_raw_archive_manifest",
+ )
+ op.drop_table("terminal_raw_archive_manifest")
+ op.drop_table("terminal_device_state")
+ op.drop_index(
+ op.f("ix_terminal_observation_event_raw_id"),
+ table_name="terminal_observation_event",
+ )
+ op.drop_index(
+ op.f("ix_terminal_observation_event_event_type"),
+ table_name="terminal_observation_event",
+ )
+ op.drop_index(
+ op.f("ix_terminal_observation_event_event_time"),
+ table_name="terminal_observation_event",
+ )
+ op.drop_index(
+ op.f("ix_terminal_observation_event_device_id"),
+ table_name="terminal_observation_event",
+ )
+ op.drop_table("terminal_observation_event")
+ op.drop_index(
+ op.f("ix_terminal_message_raw_received_at"), table_name="terminal_message_raw"
+ )
+ op.drop_index(
+ op.f("ix_terminal_message_raw_payload_sha256"),
+ table_name="terminal_message_raw",
+ )
+ op.drop_index(
+ op.f("ix_terminal_message_raw_parse_status"), table_name="terminal_message_raw"
+ )
+ op.drop_index(
+ op.f("ix_terminal_message_raw_next_retry_at"), table_name="terminal_message_raw"
+ )
+ op.drop_index(
+ op.f("ix_terminal_message_raw_device_id"), table_name="terminal_message_raw"
+ )
+ op.drop_table("terminal_message_raw")
diff --git a/backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py b/backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py
new file mode 100644
index 0000000..7d60ea8
--- /dev/null
+++ b/backend/app/alembic/versions/b4c5d6e7f809_simplify_mqtt_parse_tracking.py
@@ -0,0 +1,108 @@
+"""simplify mqtt parse tracking
+
+Revision ID: b4c5d6e7f809
+Revises: a1b2c3d4e5f6
+Create Date: 2026-03-19 17:30:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "b4c5d6e7f809"
+down_revision = "a1b2c3d4e5f6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.add_column(
+ "terminal_message_raw",
+ sa.Column("parsed_at", sa.DateTime(timezone=True), nullable=True),
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_parsed_at"),
+ "terminal_message_raw",
+ ["parsed_at"],
+ unique=False,
+ )
+
+ # Preserve historical parse outcomes before removing state-machine columns.
+ op.execute(
+ sa.text(
+ """
+ UPDATE terminal_message_raw
+ SET parsed_at = COALESCE(received_at, created_at)
+ WHERE parse_status <> 'pending'
+ """
+ )
+ )
+
+ op.drop_index(
+ op.f("ix_terminal_message_raw_next_retry_at"),
+ table_name="terminal_message_raw",
+ )
+ op.drop_index(
+ op.f("ix_terminal_message_raw_parse_status"),
+ table_name="terminal_message_raw",
+ )
+ op.drop_column("terminal_message_raw", "next_retry_at")
+ op.drop_column("terminal_message_raw", "parse_attempts")
+ op.drop_column("terminal_message_raw", "parse_status")
+
+
+def downgrade() -> None:
+ op.add_column(
+ "terminal_message_raw",
+ sa.Column(
+ "parse_status",
+ sa.String(length=32),
+ nullable=False,
+ server_default="pending",
+ ),
+ )
+ op.add_column(
+ "terminal_message_raw",
+ sa.Column("parse_attempts", sa.Integer(), nullable=False, server_default="0"),
+ )
+ op.add_column(
+ "terminal_message_raw",
+ sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True),
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_parse_status"),
+ "terminal_message_raw",
+ ["parse_status"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_message_raw_next_retry_at"),
+ "terminal_message_raw",
+ ["next_retry_at"],
+ unique=False,
+ )
+
+ op.execute(
+ sa.text(
+ """
+ UPDATE terminal_message_raw
+ SET parse_status = CASE
+ WHEN parsed_at IS NULL THEN 'pending'
+ WHEN parse_error IS NULL THEN 'succeeded'
+ ELSE 'failed'
+ END,
+ parse_attempts = CASE
+ WHEN parsed_at IS NULL THEN 0
+ ELSE 1
+ END
+ """
+ )
+ )
+
+ op.drop_index(
+ op.f("ix_terminal_message_raw_parsed_at"),
+ table_name="terminal_message_raw",
+ )
+ op.drop_column("terminal_message_raw", "parsed_at")
diff --git a/backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py b/backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py
new file mode 100644
index 0000000..15dd9f9
--- /dev/null
+++ b/backend/app/alembic/versions/c6d7e8f90123_drop_terminal_observation_event.py
@@ -0,0 +1,60 @@
+"""drop terminal observation event table
+
+Revision ID: c6d7e8f90123
+Revises: b4c5d6e7f809
+Create Date: 2026-03-19 18:20:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "c6d7e8f90123"
+down_revision = "b4c5d6e7f809"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.drop_table("terminal_observation_event")
+
+
+def downgrade() -> None:
+ op.create_table(
+ "terminal_observation_event",
+ sa.Column("id", sa.Uuid(), nullable=False),
+ sa.Column("raw_id", sa.Uuid(), nullable=False),
+ sa.Column("device_id", sa.String(length=255), nullable=False),
+ sa.Column("event_time", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("event_type", sa.String(length=64), nullable=False),
+ sa.Column("payload_json", sa.JSON(), nullable=False),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.ForeignKeyConstraint(["raw_id"], ["terminal_message_raw.id"]),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_device_id"),
+ "terminal_observation_event",
+ ["device_id"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_event_time"),
+ "terminal_observation_event",
+ ["event_time"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_event_type"),
+ "terminal_observation_event",
+ ["event_type"],
+ unique=False,
+ )
+ op.create_index(
+ op.f("ix_terminal_observation_event_raw_id"),
+ "terminal_observation_event",
+ ["raw_id"],
+ unique=False,
+ )
diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py
index c2b83c8..adccc4a 100644
--- a/backend/app/api/deps.py
+++ b/backend/app/api/deps.py
@@ -27,7 +27,7 @@ SessionDep = Annotated[Session, Depends(get_db)]
TokenDep = Annotated[str, Depends(reusable_oauth2)]
-def get_current_user(session: SessionDep, token: TokenDep) -> User:
+def resolve_user_from_token(*, session: Session, token: str) -> User:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
@@ -46,6 +46,10 @@ def get_current_user(session: SessionDep, token: TokenDep) -> User:
return user
+def get_current_user(session: SessionDep, token: TokenDep) -> User:
+ return resolve_user_from_token(session=session, token=token)
+
+
CurrentUser = Annotated[User, Depends(get_current_user)]
diff --git a/backend/app/api/main.py b/backend/app/api/main.py
index 4ea6cac..9b8c989 100644
--- a/backend/app/api/main.py
+++ b/backend/app/api/main.py
@@ -1,6 +1,14 @@
from fastapi import APIRouter
-from app.api.routes import locations, login, private, users, utils
+from app.api.routes import (
+ locations,
+ login,
+ mqtt_raw,
+ private,
+ telemetry,
+ users,
+ utils,
+)
from app.core.config import settings
api_router = APIRouter()
@@ -8,8 +16,9 @@ api_router.include_router(login.router)
api_router.include_router(users.router)
api_router.include_router(utils.router)
api_router.include_router(locations.router)
+api_router.include_router(telemetry.router)
+api_router.include_router(mqtt_raw.router)
if settings.ENVIRONMENT == "local":
api_router.include_router(private.router)
-
diff --git a/backend/app/api/routes/mqtt_raw.py b/backend/app/api/routes/mqtt_raw.py
new file mode 100644
index 0000000..fcd9c2b
--- /dev/null
+++ b/backend/app/api/routes/mqtt_raw.py
@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+from datetime import date, datetime, time, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query
+from fastapi.responses import Response
+from sqlmodel import col, delete, func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.models import (
+ DeleteResult,
+ MqttRawDaysPublic,
+ MqttRawDaySummary,
+ TerminalMessageRaw,
+ TerminalRawArchiveManifest,
+)
+
+router = APIRouter(prefix="/mqtt-raw", tags=["mqtt_raw"])
+UTC_PLUS_8 = timezone(timedelta(hours=8))
+
+
+def _normalize_day(value: Any) -> date:
+ if isinstance(value, date) and not isinstance(value, datetime):
+ return value
+ if isinstance(value, datetime):
+ return value.date()
+ if isinstance(value, str):
+ return date.fromisoformat(value)
+ raise ValueError(f"Unsupported day value: {value!r}")
+
+
+def _day_bounds(day: date) -> tuple[datetime, datetime]:
+ start_local = datetime.combine(day, time.min, tzinfo=UTC_PLUS_8)
+ end_local = start_local + timedelta(days=1)
+ return start_local.astimezone(timezone.utc), end_local.astimezone(timezone.utc)
+
+
+def _local_day_expr(session: SessionDep, field: Any) -> Any:
+ bind = session.get_bind()
+ dialect_name = bind.dialect.name if bind is not None else ""
+
+ if dialect_name == "sqlite":
+ return func.date(func.datetime(field, "+8 hours"))
+ if dialect_name in {"postgresql", "postgres"}:
+ return func.date(func.timezone("Asia/Shanghai", field))
+ return func.date(field)
+
+
+def _build_window(
+ *, day: date, start_seconds: int | None, end_seconds: int | None
+) -> tuple[datetime, datetime, int, int]:
+ start_of_day_local = datetime.combine(day, time.min, tzinfo=UTC_PLUS_8)
+ start_sec = 0 if start_seconds is None else start_seconds
+ end_sec = 24 * 3600 if end_seconds is None else end_seconds
+
+ if not 0 <= start_sec <= 24 * 3600:
+ raise HTTPException(status_code=400, detail="start_seconds must be 0..86400")
+ if not 0 <= end_sec <= 24 * 3600:
+ raise HTTPException(status_code=400, detail="end_seconds must be 0..86400")
+ if start_sec >= end_sec:
+ raise HTTPException(
+ status_code=400, detail="start_seconds must be smaller than end_seconds"
+ )
+
+ start = (start_of_day_local + timedelta(seconds=start_sec)).astimezone(timezone.utc)
+ end = (start_of_day_local + timedelta(seconds=end_sec)).astimezone(timezone.utc)
+ return start, end, start_sec, end_sec
+
+
+def _unlink_file(path_str: str) -> None:
+ path = Path(path_str)
+ try:
+ path.unlink(missing_ok=True)
+ except OSError:
+ return
+
+
+@router.get(
+ "/days",
+ response_model=MqttRawDaysPublic,
+ operation_id="mqtt_raw_list_days",
+)
+def list_mqtt_raw_days(
+ session: SessionDep,
+ _current_user: CurrentUser,
+ skip: int = Query(default=0, ge=0),
+ limit: int = Query(default=30, ge=1, le=200),
+) -> MqttRawDaysPublic:
+ days: dict[date, MqttRawDaySummary] = {}
+
+ raw_day_expr = _local_day_expr(session, TerminalMessageRaw.received_at)
+ raw_statement = (
+ select(
+ raw_day_expr,
+ func.count(),
+ func.min(TerminalMessageRaw.received_at),
+ func.max(TerminalMessageRaw.received_at),
+ )
+ .group_by(raw_day_expr)
+ .order_by(raw_day_expr.desc())
+ )
+ for day_value, message_count, first_at, last_at in session.exec(
+ raw_statement
+ ).all():
+ day_key = _normalize_day(day_value)
+ days[day_key] = MqttRawDaySummary(
+ day=day_key,
+ message_count=int(message_count or 0),
+ first_received_at=first_at,
+ last_received_at=last_at,
+ )
+
+ archive_day_expr = _local_day_expr(session, TerminalRawArchiveManifest.range_start)
+ archive_statement = (
+ select(
+ archive_day_expr,
+ func.sum(TerminalRawArchiveManifest.record_count),
+ func.min(TerminalRawArchiveManifest.range_start),
+ func.max(TerminalRawArchiveManifest.range_end),
+ )
+ .group_by(archive_day_expr)
+ .order_by(archive_day_expr.desc())
+ )
+ for day_value, message_count, first_at, last_at in session.exec(
+ archive_statement
+ ).all():
+ day_key = _normalize_day(day_value)
+ existing = days.get(day_key)
+ archive_count = int(message_count or 0)
+ if existing is None:
+ days[day_key] = MqttRawDaySummary(
+ day=day_key,
+ message_count=archive_count,
+ first_received_at=first_at,
+ last_received_at=last_at,
+ )
+ continue
+ existing.message_count += archive_count
+ if first_at and (
+ existing.first_received_at is None or first_at < existing.first_received_at
+ ):
+ existing.first_received_at = first_at
+ if last_at and (
+ existing.last_received_at is None or last_at > existing.last_received_at
+ ):
+ existing.last_received_at = last_at
+
+ ordered = sorted(days.values(), key=lambda item: item.day, reverse=True)
+ paged = ordered[skip : skip + limit]
+ return MqttRawDaysPublic(data=paged, count=len(ordered))
+
+
+@router.get(
+ "/download",
+ operation_id="mqtt_raw_download",
+ responses={
+ 200: {
+ "content": {
+ "application/octet-stream": {
+ "schema": {"type": "string", "format": "binary"}
+ }
+ }
+ }
+ },
+)
+def download_mqtt_raw(
+ session: SessionDep,
+ _current_user: CurrentUser,
+ day: date,
+ start_seconds: int | None = Query(default=None),
+ end_seconds: int | None = Query(default=None),
+) -> Response:
+ range_start, range_end, start_sec, end_sec = _build_window(
+ day=day,
+ start_seconds=start_seconds,
+ end_seconds=end_seconds,
+ )
+
+ chunks: list[tuple[datetime, bytes]] = []
+
+ raw_statement = (
+ select(TerminalMessageRaw)
+ .where(TerminalMessageRaw.received_at >= range_start)
+ .where(TerminalMessageRaw.received_at < range_end)
+ .order_by(col(TerminalMessageRaw.received_at), col(TerminalMessageRaw.id))
+ )
+ raw_rows = session.exec(raw_statement).all()
+ for row in raw_rows:
+ chunks.append((row.received_at, row.payload_bytes))
+
+ manifest_statement = (
+ select(TerminalRawArchiveManifest)
+ .where(TerminalRawArchiveManifest.range_start >= range_start)
+ .where(TerminalRawArchiveManifest.range_end <= range_end)
+ .order_by(
+ col(TerminalRawArchiveManifest.range_start),
+ col(TerminalRawArchiveManifest.range_end),
+ )
+ )
+ manifests = session.exec(manifest_statement).all()
+ for manifest in manifests:
+ path = Path(manifest.file_path)
+ if not path.exists() or not path.is_file():
+ continue
+ chunks.append((manifest.range_start, path.read_bytes()))
+
+ chunks.sort(key=lambda item: item[0])
+ payload = b"".join(chunk for _, chunk in chunks)
+
+ name_suffix = (
+ day.isoformat()
+ if start_sec == 0 and end_sec == 24 * 3600
+ else f"{day.isoformat()}_{start_sec}_{end_sec}"
+ )
+ filename = f"mqtt_raw_{name_suffix}.txt"
+ return Response(
+ content=payload,
+ media_type="application/octet-stream",
+ headers={"Content-Disposition": f'attachment; filename="{filename}"'},
+ )
+
+
+@router.delete(
+ "/day/{day}",
+ response_model=DeleteResult,
+ operation_id="mqtt_raw_delete_day",
+)
+def delete_mqtt_raw_by_day(
+ session: SessionDep,
+ _current_user: CurrentUser,
+ day: date,
+) -> DeleteResult:
+ range_start, range_end = _day_bounds(day)
+
+ raw_statement = (
+ select(TerminalMessageRaw.id)
+ .where(TerminalMessageRaw.received_at >= range_start)
+ .where(TerminalMessageRaw.received_at < range_end)
+ )
+ raw_ids = list(session.exec(raw_statement).all())
+ if raw_ids:
+ session.exec(
+ delete(TerminalMessageRaw).where(col(TerminalMessageRaw.id).in_(raw_ids))
+ )
+
+ manifest_statement = (
+ select(TerminalRawArchiveManifest)
+ .where(TerminalRawArchiveManifest.range_start >= range_start)
+ .where(TerminalRawArchiveManifest.range_end <= range_end)
+ )
+ manifests = session.exec(manifest_statement).all()
+ for manifest in manifests:
+ _unlink_file(manifest.file_path)
+ session.delete(manifest)
+
+ session.commit()
+
+ deleted_count = len(raw_ids) + len(manifests)
+ return DeleteResult(
+ deleted_count=deleted_count,
+ message=f"Deleted {deleted_count} record(s) for {day.isoformat()}",
+ )
+
+
+@router.delete(
+ "/all",
+ response_model=DeleteResult,
+ operation_id="mqtt_raw_delete_all",
+)
+def delete_all_mqtt_raw(
+ session: SessionDep,
+ _current_user: CurrentUser,
+ confirm: bool = Query(default=False),
+) -> DeleteResult:
+ if not confirm:
+ raise HTTPException(
+ status_code=400,
+ detail="Set confirm=true to delete all saved MQTT raw data",
+ )
+
+ raw_ids = list(session.exec(select(TerminalMessageRaw.id)).all())
+ if raw_ids:
+ session.exec(
+ delete(TerminalMessageRaw).where(col(TerminalMessageRaw.id).in_(raw_ids))
+ )
+
+ manifests = session.exec(select(TerminalRawArchiveManifest)).all()
+ for manifest in manifests:
+ _unlink_file(manifest.file_path)
+ session.delete(manifest)
+
+ session.commit()
+
+ deleted_count = len(raw_ids) + len(manifests)
+ return DeleteResult(
+ deleted_count=deleted_count,
+ message=f"Deleted {deleted_count} record(s) from MQTT raw storage",
+ )
diff --git a/backend/app/api/routes/telemetry.py b/backend/app/api/routes/telemetry.py
new file mode 100644
index 0000000..9342db4
--- /dev/null
+++ b/backend/app/api/routes/telemetry.py
@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query, WebSocket, status
+from fastapi.encoders import jsonable_encoder
+from fastapi.websockets import WebSocketDisconnect
+from sqlmodel import Session
+
+from app.api.deps import CurrentUser, SessionDep, resolve_user_from_token
+from app.core.db import engine
+from app.mqtt.telemetry import build_telemetry_snapshot
+
+router = APIRouter(prefix="/telemetry", tags=["telemetry"])
+STREAM_INTERVAL_SECONDS = 1.0
+
+
+@router.get("/latest")
+def read_telemetry_latest(
+ session: SessionDep,
+ _current_user: CurrentUser,
+ device_id: str | None = None,
+ limit: int = Query(default=50, ge=1, le=200),
+) -> dict[str, Any]:
+ """
+ Return the latest parsed GNSS telemetry snapshot.
+ """
+ return build_telemetry_snapshot(session=session, device_id=device_id, limit=limit)
+
+
+@router.websocket("/ws")
+async def telemetry_websocket(
+ websocket: WebSocket,
+ token: str | None = Query(default=None),
+ device_id: str | None = Query(default=None),
+ limit: int = Query(default=50, ge=1, le=200),
+) -> None:
+ if not token:
+ await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+ return
+
+ with Session(engine) as session:
+ try:
+ resolve_user_from_token(session=session, token=token)
+ except HTTPException:
+ await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+ return
+
+ await websocket.accept()
+
+ try:
+ while True:
+ with Session(engine) as session:
+ payload = build_telemetry_snapshot(
+ session=session,
+ device_id=device_id,
+ limit=limit,
+ )
+ await websocket.send_json(jsonable_encoder(payload))
+ await asyncio.sleep(STREAM_INTERVAL_SECONDS)
+ except (WebSocketDisconnect, RuntimeError):
+ return
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index 650b9f7..59c25a6 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -56,6 +56,17 @@ class Settings(BaseSettings):
POSTGRES_PASSWORD: str = ""
POSTGRES_DB: str = ""
+ MQTT_HOST: str = "101.35.119.226"
+ MQTT_PORT: int = 1883
+ MQTT_USERNAME: str = ""
+ MQTT_PASSWORD: str = ""
+ MQTT_TOPIC_PATTERN: str = "terminal/+/raw"
+ MQTT_CLIENT_ID: str = "spatialhub-mqtt-ingestor"
+ RAW_HOT_RETENTION_DAYS: int = 90
+ RAW_ARCHIVE_BASE_DIR: str = "/data/archive/gnss"
+ PARSER_BATCH_SIZE: int = 200
+ PARSER_POLL_INTERVAL_MS: int = 300
+
@computed_field # type: ignore[prop-decorator]
@property
def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn:
diff --git a/backend/app/models.py b/backend/app/models.py
index 147d591..e61a315 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -1,8 +1,9 @@
import uuid
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
+from typing import Any
from pydantic import EmailStr
-from sqlalchemy import DateTime
+from sqlalchemy import JSON, DateTime, LargeBinary
from sqlmodel import Field, Relationship, SQLModel
@@ -53,7 +54,9 @@ class User(UserBase, table=True):
default_factory=get_datetime_utc,
sa_type=DateTime(timezone=True), # type: ignore
)
- locations: list["Location"] = Relationship(back_populates="owner", cascade_delete=True)
+ locations: list["Location"] = Relationship(
+ back_populates="owner", cascade_delete=True
+ )
# Properties to return via API, id is always required
@@ -108,11 +111,98 @@ class LocationsPublic(SQLModel):
count: int
+class MqttRawDaySummary(SQLModel):
+ day: date
+ message_count: int
+ first_received_at: datetime | None = None
+ last_received_at: datetime | None = None
+
+
+class MqttRawDaysPublic(SQLModel):
+ data: list[MqttRawDaySummary]
+ count: int
+
+
+class TerminalMessageRaw(SQLModel, table=True):
+ __tablename__ = "terminal_message_raw"
+
+ id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+ device_id: str = Field(max_length=255, index=True)
+ topic: str = Field(max_length=255)
+ received_at: datetime = Field(
+ default_factory=get_datetime_utc,
+ sa_type=DateTime(timezone=True), # type: ignore
+ )
+ payload_bytes: bytes = Field(sa_type=LargeBinary)
+ payload_size: int = 0
+ payload_sha256: str = Field(default="", max_length=64, index=True)
+ qos: int = 0
+ retain: bool = False
+ parsed_at: datetime | None = Field(
+ default=None,
+ sa_type=DateTime(timezone=True), # type: ignore
+ index=True,
+ )
+ parse_error: str | None = Field(default=None, max_length=1024)
+ created_at: datetime = Field(
+ default_factory=get_datetime_utc,
+ sa_type=DateTime(timezone=True), # type: ignore
+ )
+
+
+class TerminalDeviceState(SQLModel, table=True):
+ __tablename__ = "terminal_device_state"
+
+ device_id: str = Field(max_length=255, primary_key=True)
+ last_event_time: datetime | None = Field(
+ default=None, sa_type=DateTime(timezone=True)
+ ) # type: ignore[call-overload]
+ fix_status: str | None = Field(default=None, max_length=64)
+ lat: float | None = None
+ lon: float | None = None
+ alt_m: float | None = None
+ satellites_used: int | None = None
+ satellite_signal_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)
+ rtcm_type_summary: dict[str, Any] | None = Field(default=None, sa_type=JSON)
+ updated_at: datetime = Field(
+ default_factory=get_datetime_utc,
+ sa_type=DateTime(timezone=True), # type: ignore
+ )
+
+
+class TerminalRawArchiveManifest(SQLModel, table=True):
+ __tablename__ = "terminal_raw_archive_manifest"
+
+ id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+ device_id: str = Field(max_length=255, index=True)
+ range_start: datetime = Field(
+ sa_type=DateTime(timezone=True), # type: ignore
+ index=True,
+ )
+ range_end: datetime = Field(
+ sa_type=DateTime(timezone=True), # type: ignore
+ index=True,
+ )
+ file_path: str = Field(max_length=1024, unique=True)
+ file_size: int = 0
+ sha256: str = Field(max_length=64)
+ record_count: int = 0
+ created_at: datetime = Field(
+ default_factory=get_datetime_utc,
+ sa_type=DateTime(timezone=True), # type: ignore
+ )
+
+
# Generic message
class Message(SQLModel):
message: str
+class DeleteResult(SQLModel):
+ deleted_count: int
+ message: str
+
+
# JSON payload containing access token
class Token(SQLModel):
access_token: str
diff --git a/backend/app/mqtt/__init__.py b/backend/app/mqtt/__init__.py
new file mode 100644
index 0000000..527a764
--- /dev/null
+++ b/backend/app/mqtt/__init__.py
@@ -0,0 +1 @@
+"""MQTT ingestion pipeline package."""
diff --git a/backend/app/mqtt/archive_job.py b/backend/app/mqtt/archive_job.py
new file mode 100644
index 0000000..a8ff69d
--- /dev/null
+++ b/backend/app/mqtt/archive_job.py
@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import hashlib
+import logging
+from datetime import datetime, timedelta
+from pathlib import Path
+
+from sqlmodel import Session, col, select
+
+from app.core.config import settings
+from app.core.db import engine
+from app.models import TerminalMessageRaw, TerminalRawArchiveManifest, get_datetime_utc
+
+logger = logging.getLogger(__name__)
+
+
+def _format_timestamp(value: datetime) -> str:
+ return value.strftime("%Y-%m-%dT%H-%M-%S")
+
+
+def _build_archive_path(
+ *,
+ archive_base_dir: Path,
+ device_id: str,
+ range_start: datetime,
+ range_end: datetime,
+) -> Path:
+ day = range_start.strftime("%Y-%m-%d")
+ output_dir = archive_base_dir / device_id / day
+ output_dir.mkdir(parents=True, exist_ok=True)
+ filename = (
+ f"device_{device_id}_{_format_timestamp(range_start)}"
+ f"_{_format_timestamp(range_end)}.txt"
+ )
+ return output_dir / filename
+
+
+def run_archive_once(
+ *,
+ session: Session,
+ archive_base_dir: Path | str,
+ retention_days: int,
+ now: datetime | None = None,
+) -> list[TerminalRawArchiveManifest]:
+ archive_root = Path(archive_base_dir)
+ ts_now = now or get_datetime_utc()
+ cutoff = ts_now - timedelta(days=retention_days)
+
+ statement = (
+ select(TerminalMessageRaw)
+ .where(TerminalMessageRaw.received_at < cutoff)
+ .order_by(
+ col(TerminalMessageRaw.device_id), col(TerminalMessageRaw.received_at)
+ )
+ )
+ rows = session.exec(statement).all()
+ if not rows:
+ return []
+
+ grouped: dict[tuple[str, str], list[TerminalMessageRaw]] = {}
+ for row in rows:
+ key = (row.device_id, row.received_at.strftime("%Y-%m-%d"))
+ grouped.setdefault(key, []).append(row)
+
+ manifests: list[TerminalRawArchiveManifest] = []
+ for (_, _), group_rows in grouped.items():
+ range_start = group_rows[0].received_at
+ range_end = group_rows[-1].received_at
+ output_path = _build_archive_path(
+ archive_base_dir=archive_root,
+ device_id=group_rows[0].device_id,
+ range_start=range_start,
+ range_end=range_end,
+ )
+
+ digest = hashlib.sha256()
+ record_count = 0
+ with output_path.open("wb") as out:
+ for row in group_rows:
+ out.write(row.payload_bytes)
+ digest.update(row.payload_bytes)
+ record_count += 1
+
+ manifest = TerminalRawArchiveManifest(
+ device_id=group_rows[0].device_id,
+ range_start=range_start,
+ range_end=range_end,
+ file_path=str(output_path),
+ file_size=output_path.stat().st_size,
+ sha256=digest.hexdigest(),
+ record_count=record_count,
+ )
+ session.add(manifest)
+ manifests.append(manifest)
+
+ for row in group_rows:
+ session.delete(row)
+
+ session.commit()
+
+ for manifest in manifests:
+ session.refresh(manifest)
+
+ return manifests
+
+
+def run() -> None:
+ logging.basicConfig(level=logging.INFO)
+ with Session(engine) as session:
+ manifests = run_archive_once(
+ session=session,
+ archive_base_dir=settings.RAW_ARCHIVE_BASE_DIR,
+ retention_days=settings.RAW_HOT_RETENTION_DAYS,
+ )
+ logger.info("Archived %s file(s)", len(manifests))
+
+
+if __name__ == "__main__":
+ run()
diff --git a/backend/app/mqtt/ingestor.py b/backend/app/mqtt/ingestor.py
new file mode 100644
index 0000000..152eb85
--- /dev/null
+++ b/backend/app/mqtt/ingestor.py
@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.core.db import engine
+from app.mqtt.repository import insert_raw_message
+
+logger = logging.getLogger(__name__)
+TOPIC_PATTERN = re.compile(r"^terminal/(?P