Fix camera/tracker dedup (device_type at provision + backfill) and finish refresh-token flow

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/auth.py

@@ -112,6 +112,66 @@ async def store_refresh_token(account_id: int, token_hash: str, expires_at: date
         )
 
 
+async def touch_last_login(account_id: int) -> None:
+    pool = await get_pool()
+    async with pool.connection() as conn, conn.cursor() as cur:
+        await cur.execute(
+            "UPDATE auth.accounts SET last_login_at = now() WHERE account_id = %s",
+            (account_id,),
+        )
+
+
+async def rotate_refresh_token(raw_token: str) -> TokenPair | None:
+    """Redeem a refresh token and rotate it.
+
+    Returns a fresh access+refresh pair, or None when the token is unknown,
+    expired, already revoked, or belongs to a deactivated account. The old
+    token is revoked in the same transaction (single-use rotation), so a
+    replayed refresh token never yields a second valid pair — that also gives
+    us reuse detection if we want to act on it later.
+    """
+    token_hash = hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
+    pool = await get_pool()
+    async with pool.connection() as conn, conn.transaction(), conn.cursor() as cur:
+        await cur.execute(
+            """
+            SELECT t.token_id, t.account_id, a.scopes
+              FROM auth.tokens t
+              JOIN auth.accounts a ON a.account_id = t.account_id
+             WHERE t.token_hash = %s
+               AND t.token_type = 'refresh'
+               AND t.revoked_at IS NULL
+               AND t.expires_at > now()
+               AND a.is_active = true
+             FOR UPDATE OF t
+            """,
+            (token_hash,),
+        )
+        row = await cur.fetchone()
+        if row is None:
+            return None
+        token_id, account_id, scopes = int(row[0]), int(row[1]), list(row[2])
+
+        await cur.execute(
+            "UPDATE auth.tokens SET revoked_at = now() WHERE token_id = %s",
+            (token_id,),
+        )
+
+        access, ttl = issue_access_token(account_id, scopes)
+        new_raw, expires_at, new_hash = issue_refresh_token(account_id)
+        await cur.execute(
+            "INSERT INTO auth.tokens (account_id, token_type, token_hash, expires_at) "
+            "VALUES (%s, 'refresh', %s, %s)",
+            (account_id, new_hash, expires_at),
+        )
+        await cur.execute(
+            "UPDATE auth.accounts SET last_login_at = now() WHERE account_id = %s",
+            (account_id,),
+        )
+
+    return TokenPair(access_token=access, refresh_token=new_raw, expires_in=ttl)
+
+
 async def current_account(
     token: Annotated[str, Depends(oauth2_scheme)],
 ) -> AuthAccount:

app/projectors/live_positions.py

@@ -26,6 +26,21 @@ PROJECTED_FLAG_KEY = "live_positions_projected_at"
 
 
 _PLATE_FROM_DEVICE_NAME = re.compile(r"^.* - (.+?)(?:_cam|_CAM)?$")
+_CAMERA_NAME_RE = re.compile(r"_cam$", re.IGNORECASE)
+
+
+def _classify_device_type(device_name: str | None) -> str:
+    """Tracksolid device_name carries a '_cam' suffix for camera units
+    (e.g. "John Mbugua - KDW 573B_cam"). Everything else is a GPS tracker.
+
+    The tracker-first dedup in serve.fn_live_view (PRD F1.6) depends on this
+    classification: when a camera and a tracker share one vehicle_id, the
+    accurate tracker fix must win. Provisioning every device as 'tracker'
+    silently defeats that tie-break, so classify at first sight here.
+    """
+    if device_name and _CAMERA_NAME_RE.search(device_name):
+        return "camera"
+    return "tracker"
 
 
 def _extract_plate_from_device_name(device_name: str | None) -> str | None:
@@ -105,16 +120,18 @@ async def _resolve_device(
             imei=imei, vehicle_id=vehicle_id, plate=plate,
         )
 
+    device_type = _classify_device_type(device_name)
     await cur.execute(
         """INSERT INTO domain.devices
              (imei, account_id, vehicle_id, device_type, lifecycle, activation_at)
-           VALUES (%s, %s, %s, 'tracker', 'active', now())
+           VALUES (%s, %s, %s, %s, 'active', now())
            ON CONFLICT (imei) DO UPDATE
-             SET account_id = EXCLUDED.account_id,
-                 vehicle_id = COALESCE(domain.devices.vehicle_id, EXCLUDED.vehicle_id),
-                 lifecycle  = CASE WHEN domain.devices.lifecycle = 'provisioned'
-                                   THEN 'active' ELSE domain.devices.lifecycle END""",
-        (imei, account_id, vehicle_id),
+             SET account_id  = EXCLUDED.account_id,
+                 vehicle_id  = COALESCE(domain.devices.vehicle_id, EXCLUDED.vehicle_id),
+                 device_type = EXCLUDED.device_type,
+                 lifecycle   = CASE WHEN domain.devices.lifecycle = 'provisioned'
+                                    THEN 'active' ELSE domain.devices.lifecycle END""",
+        (imei, account_id, vehicle_id, device_type),
     )
 
     return vehicle_id

app/routers/auth.py

@@ -5,7 +5,9 @@ from app.auth import (
     fetch_account,
     issue_access_token,
     issue_refresh_token,
+    rotate_refresh_token,
     store_refresh_token,
+    touch_last_login,
     verify_password,
 )
 
@@ -27,9 +29,20 @@ async def issue_token(
     access, ttl = issue_access_token(account_id, scopes)
     refresh, expires_at, refresh_hash = issue_refresh_token(account_id)
     await store_refresh_token(account_id, refresh_hash, expires_at)
+    await touch_last_login(account_id)
 
     return TokenPair(
         access_token=access,
         refresh_token=refresh,
         expires_in=ttl,
     )
+
+
+@router.post("/refresh", response_model=TokenPair)
+async def refresh(
+    refresh_token: str = Form(...),
+) -> TokenPair:
+    pair = await rotate_refresh_token(refresh_token)
+    if pair is None:
+        raise HTTPException(status_code=401, detail="invalid or expired refresh token")
+    return pair

db/migrations/20260601000021_backfill_camera_device_type.sql

@@ -0,0 +1,32 @@
+-- migrate:up
+--
+-- Fix historical mis-classification of camera units.
+--
+-- The auto-provisioner used to insert every device as device_type='tracker'
+-- (app/projectors/live_positions.py). That silently defeated the tracker-first
+-- dedup in serve.fn_live_view (PRD F1.6): when a '_cam' camera and an X3
+-- tracker were consolidated onto one vehicle_id, both rows read 'tracker', so
+-- the dedup tie-break fell through to occurred_at DESC and the map flipped
+-- between the tracker's accurate fix and the camera's. This re-tags any device
+-- whose Tracksolid device_name carries the '_cam' suffix as a camera, matching
+-- the new _classify_device_type() logic in the projector.
+--
+-- Idempotent: only flips rows that aren't already 'camera'.
+
+UPDATE domain.devices d
+   SET device_type = 'camera',
+       updated_at  = now()
+  FROM state.live_positions lp
+ WHERE lp.imei = d.imei
+   AND lp.device_name ~* '_cam$'
+   AND d.device_type <> 'camera';
+
+-- migrate:down
+--
+-- Revert to the prior (incorrect) state: everything a tracker. Kept only so the
+-- migration is reversible; running it re-introduces the dedup bug.
+
+UPDATE domain.devices
+   SET device_type = 'tracker',
+       updated_at  = now()
+ WHERE device_type = 'camera';

tests/test_projector.py

@@ -0,0 +1,19 @@
+import pytest
+
+from app.projectors.live_positions import _classify_device_type
+
+
+@pytest.mark.parametrize(
+    "device_name,expected",
+    [
+        ("John Mbugua - KDW 573B_cam", "camera"),
+        ("John Mbugua - KDW 573B_CAM", "camera"),
+        ("John Mbugua - KDW 573B", "tracker"),
+        ("Parked - KMGK 596V", "tracker"),
+        ("JC400P-92732", "tracker"),
+        ("", "tracker"),
+        (None, "tracker"),
+    ],
+)
+def test_classify_device_type(device_name: str | None, expected: str) -> None:
+    assert _classify_device_type(device_name) == expected

web/fleet-core.js

@@ -18,6 +18,13 @@ const VEHICLE_SOURCE = 'vehicles';
 
 /* ---------- authClient ---------- */
 
+function _saveTokens(payload) {
+  const expiresAt = Math.floor(Date.now() / 1000) + Number(payload.expires_in || 900);
+  localStorage.setItem(STORAGE_ACCESS, payload.access_token);
+  localStorage.setItem(STORAGE_REFRESH, payload.refresh_token);
+  localStorage.setItem(STORAGE_EXPIRES, String(expiresAt));
+}
+
 export const authClient = {
   isAuthenticated() {
     const expiresAt = Number(localStorage.getItem(STORAGE_EXPIRES) || 0);
@@ -35,11 +42,37 @@ export const authClient = {
       const detail = await res.json().catch(() => ({ detail: 'login failed' }));
       throw new Error(detail.detail || 'login failed');
     }
-    const payload = await res.json();
-    const expiresAt = Math.floor(Date.now() / 1000) + Number(payload.expires_in || 900);
-    localStorage.setItem(STORAGE_ACCESS, payload.access_token);
-    localStorage.setItem(STORAGE_REFRESH, payload.refresh_token);
-    localStorage.setItem(STORAGE_EXPIRES, String(expiresAt));
+    _saveTokens(await res.json());
+  },
+
+  // Exchange the stored refresh token for a fresh access+refresh pair.
+  // Returns true on success; on any failure the session is cleared so the
+  // caller can fall back to the login screen. Single-flight: concurrent
+  // callers share one in-flight request so a burst of 401s rotates once.
+  async refresh() {
+    const refreshToken = localStorage.getItem(STORAGE_REFRESH);
+    if (!refreshToken) return false;
+    if (this._refreshing) return this._refreshing;
+    this._refreshing = (async () => {
+      try {
+        const res = await fetch('/api/auth/refresh', {
+          method: 'POST',
+          headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
+          body: new URLSearchParams({ refresh_token: refreshToken }),
+        });
+        if (!res.ok) {
+          this.logout();
+          return false;
+        }
+        _saveTokens(await res.json());
+        return true;
+      } catch {
+        return false;
+      } finally {
+        this._refreshing = null;
+      }
+    })();
+    return this._refreshing;
   },
 
   logout() {
@@ -70,15 +103,24 @@ export async function apiFetch(path, { params, ...opts } = {}) {
       }
     }
   }
-  const token = localStorage.getItem(STORAGE_ACCESS);
-  const res = await fetch(url.toString(), {
-    ...opts,
-    headers: {
-      ...(opts.headers || {}),
-      ...(token ? { Authorization: `Bearer ${token}` } : {}),
-      Accept: 'application/json',
-    },
-  });
+  const send = () => {
+    const token = localStorage.getItem(STORAGE_ACCESS);
+    return fetch(url.toString(), {
+      ...opts,
+      headers: {
+        ...(opts.headers || {}),
+        ...(token ? { Authorization: `Bearer ${token}` } : {}),
+        Accept: 'application/json',
+      },
+    });
+  };
+
+  let res = await send();
+  // Access tokens are short-lived (15 min). On expiry, rotate the refresh
+  // token once and retry transparently rather than dumping the user at /login.
+  if (res.status === 401 && (await authClient.refresh())) {
+    res = await send();
+  }
   if (res.status === 401) {
     authClient.logout();
     window.location.href = '/login.html';
@@ -1121,10 +1163,13 @@ function _fmtNum(v, digits) {
 
 async function _downloadTripsCsv(vehicleId, dateStr) {
   const url = `/api/views/vehicle/${vehicleId}/trips.csv?date=${encodeURIComponent(dateStr)}`;
+  const send = () =>
+    fetch(url, { headers: { Authorization: `Bearer ${authClient.getToken()}` } });
   try {
-    const r = await fetch(url, {
-      headers: { Authorization: `Bearer ${authClient.getToken()}` },
-    });
+    let r = await send();
+    if (r.status === 401 && (await authClient.refresh())) {
+      r = await send();
+    }
     if (!r.ok) throw new Error(`HTTP ${r.status}`);
     const blob = await r.blob();
     const cd = r.headers.get('Content-Disposition') || '';