diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..41211aa --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,46 @@ +name: test + +on: + pull_request: + push: + branches: [main] + +permissions: + contents: read + +jobs: + test: + name: cargo test + runs-on: ubuntu-latest + + services: + postgres: + image: postgres:18 + env: + POSTGRES_USER: thq + POSTGRES_PASSWORD: thq + POSTGRES_DB: thq_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U thq -d thq_test" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + + # rustup installs the toolchain pinned by rust-toolchain.toml on first + # use; this step makes the resolved version visible in the log. + - name: Show toolchain + run: rustup show + + - uses: Swatinem/rust-cache@v2 + + - name: cargo test + env: + THQ_TEST_DATABASE_URL: postgres://thq:thq@localhost:5432/thq_test + run: cargo test diff --git a/README.md b/README.md index d5657d5..57be292 100644 --- a/README.md +++ b/README.md @@ -5,11 +5,12 @@ A telemetry server for [TrainLCD](https://github.com/TrainLCD). It provides real ## Features - **WebSocket** — Real-time broadcast of location updates, log events, and interaction events -- **GraphQL** — Event ingestion (`sendLogEvent`, `sendInteractionEvent`, `sendLocation` mutations), history queries (`logEvents`, `interactionEvents`, `locations`) and aggregated per-line accuracy reports (`POST /graphql`) +- **GraphQL** — Event ingestion (`sendLogEvent`, `sendInteractionEvent`, `sendLocation` mutations), history queries (`logEvents`, `interactionEvents`, `locations`), frozen-position detection (`locationFreezes`, `locationFreezeSessions`, `locationFreezeSummary`) and aggregated per-line accuracy reports (`POST /graphql`) - **PostgreSQL persistence** — Optionally stores all events in the database - **Ring buffer** — Keeps the latest N events in memory (default 1000) - **Scoped authentication** — Three shared secrets: observer (WebSocket + history queries), events (log + interaction submission), telemetry (log + interaction + location submission) - **Line topology** — Automatic segment annotation from a CSV topology file +- **Freeze detection** — Finds location log gaps that look like a frozen position and compares them across builds (see [docs/location-freeze-regression.md](./docs/location-freeze-regression.md)) ## Requirements @@ -84,15 +85,15 @@ telemetry_auth_token = "change-me-telemetry" Three shared secrets grant exactly one role each: -| Token | WebSocket subscribe | History queries (`logEvents` / `interactionEvents` / `locations`) | `sendLogEvent` / `sendInteractionEvent` | `sendLocation` | -|---|---|---|---|---| -| Observer | ✅ | ✅ | ❌ | ❌ | -| Events | ❌ | ❌ | ✅ | ❌ | -| Telemetry | ❌ | ❌ | ✅ | ✅ | +| Token | WebSocket subscribe | History queries (`logEvents` / `interactionEvents` / `locations`) | Freeze queries (`locationFreezes` / `locationFreezeSessions` / `locationFreezeSummary`) | `sendLogEvent` / `sendInteractionEvent` | `sendLocation` | +|---|---|---|---|---|---| +| Observer | ✅ | ✅ | ✅ | ❌ | ❌ | +| Events | ❌ | ❌ | ❌ | ✅ | ❌ | +| Telemetry | ❌ | ❌ | ❌ | ✅ | ✅ | - **WebSocket** — send the observer token via subprotocols: `Sec-WebSocket-Protocol: thq, thq-auth-` - **GraphQL mutations** — send the events or telemetry token via `Authorization: Bearer ` -- **GraphQL history queries** — send the observer token via `Authorization: Bearer `; raw event data is exposed only to the observation role that already sees it in real time over WebSocket +- **GraphQL history and freeze queries** — send the observer token via `Authorization: Bearer `; raw event data is exposed only to the observation role that already sees it in real time over WebSocket - **GraphQL aggregated queries** (`accuracyByLine`) — no authentication (aggregated data only) Authentication is always enforced. At least one token must be configured, or the server refuses to start. @@ -168,6 +169,9 @@ mutation { speed: 45.0 } timestamp: 1706000000000 + appVersion: "1.2.3" # optional — same value as sendLogEvent + platform: ios # ios | android | macos | unknown + channel: production # production | canary }) { sessionId warning # set when e.g. the reported accuracy exceeds 100 m @@ -177,6 +181,8 @@ mutation { `stationId` is only meaningful when `state` is `arrived` or `passing` and is ignored otherwise. `batteryLevel` (0.0–1.0) and `batteryState` (`unknown | unplugged | charging | full`) are optional. +`appVersion` / `platform` / `channel` are optional for backwards compatibility with clients that predate them, and carry the same values the client already sends with `sendLogEvent`. Storing them on the location row lets the freeze queries group by build without joining `log_events`; a blank `appVersion` is rejected. For rows that lack them, the freeze queries fall back to the log and interaction events of the same session. + #### `logEvents` / `interactionEvents` / `locations` — History queries Each mutation has a matching query returning the persisted events, newest first. All three require the **observer token** (`Authorization: Bearer `) — the same read-only role that observes events in real time over WebSocket — and a configured database. @@ -213,7 +219,7 @@ query { id sessionId device state stationId lineId coords { latitude longitude accuracy speed } timestamp segmentId fromStationId toStationId - batteryLevel batteryState recordedAt + batteryLevel batteryState appVersion platform channel recordedAt } } ``` @@ -232,6 +238,72 @@ Per-query filters: `logEvents` also accepts `type` and `level`; `interactionEven Columns added to the storage schema over time are nullable in the results: legacy rows recorded before a column existed return `null` for it (e.g. `sessionId`, `appVersion`, or `lineId` on old rows). `recordedAt` is the server-side persistence time, while `timestamp` is the client-reported unix-millisecond value. +#### `locationFreezes` / `locationFreezeSessions` / `locationFreezeSummary` — Frozen-position detection + +Finds stretches where the location log went silent while the app kept running and the device was moving fast — the signature of a position that stopped advancing on screen. All three require the **observer token** and a configured database, and all three take the same `LocationFreezeFilter`. The background, thresholds and known blind spots are documented in [docs/location-freeze-regression.md](./docs/location-freeze-regression.md). + +```graphql +query { + locationFreezes( + filter: { + from: "2026-07-01T00:00:00Z" # required, client-reported timestamp + to: "2026-07-02T00:00:00Z" # required, at most 90 days after from + lineId: 11302 # every other filter is optional + segmentId: "11302:1130201:1130202" + device: "device-001" + sessionId: "d0f7..." + appVersion: "10.4.1(100)" + platform: ios + channel: canary + gapThresholdMs: 60000 # default 60000, minimum 1000 + speedThresholdKmh: 30 # default 30 + requireAppAlive: true # default true + } + limit: 100 # default 100, cap 2000 + ) { + sessionId device lineId segmentId fromStationId toStationId + appVersion platform channel + gapStart gapEnd gapMs speedBeforeGap + coordsBeforeGap { latitude longitude accuracy speed } + coordsAfterGap { latitude longitude accuracy speed } + jumpDistanceMeters aliveEventCount + } +} +``` + +```graphql +query { + locationFreezeSessions(filter: { from: "2026-07-01T00:00:00Z", to: "2026-07-02T00:00:00Z" }) { + sessionId device lineIds appVersion platform channel + startedAt endedAt locationCount maxSpeed + freezeCount maxGapMs totalGapMs + } +} +``` + +```graphql +query { + locationFreezeSummary(filter: { from: "2026-07-01T00:00:00Z", to: "2026-07-02T00:00:00Z" }) { + lineId segmentId fromStationId toStationId device + appVersion platform channel + sessionCount locationCount + freezeSessionCount freezeCount maxGapMs totalGapMs + } +} +``` + +| Filter | Type | Description | +|---|---|---| +| `from` | `DateTime!` | Inclusive lower bound on the client-reported timestamp | +| `to` | `DateTime!` | Exclusive upper bound; at most 90 days after `from`. It bounds the row that *starts* a gap: a gap whose closing row only arrives after `to` is still reported in full | +| `lineId` / `segmentId` / `device` / `sessionId` | — | Matched against the row immediately before the gap | +| `appVersion` / `platform` / `channel` | — | Build filters, applied after the per-session backfill | +| `gapThresholdMs` | `Int` | Gap length that counts as a freeze candidate (default 60000, minimum 1000) | +| `speedThresholdKmh` | `Float` | Minimum speed on the row before the gap (default 30) | +| `requireAppAlive` | `Boolean` | Require log or interaction events inside the gap (default true) | + +`locationFreezes` returns one row per gap, newest first. `locationFreezeSessions` returns exactly one row per session — a ride that crossed several lines is not split up, and the lines it touched are listed in `lineIds`. `locationFreezeSessions` and `locationFreezeSummary` also return sessions and groups with **zero** freezes, so a build can be shown to be clean rather than merely absent from the results. Their counts (`locationCount`, `startedAt`, `endedAt`) cover only rows inside `[from, to)`. Millisecond fields (`gapMs`, `maxGapMs`, `totalGapMs`) are `Int`. + #### `accuracyByLine` — Aggregated accuracy report Returns aggregated accuracy metrics per line. Raw event data is exposed only through the observer-token history queries above; this aggregated report requires no authentication. @@ -365,12 +437,33 @@ When `database_url` / `DATABASE_URL` is provided, the server connects to Postgre | Table | Key columns | |---|---| -| `location_logs` | `id`, `session_id`, `device`, `state`, `station_id`, `line_id`, `segment_id`, `from_station_id`, `to_station_id`, `latitude`, `longitude`, `accuracy`, `speed`, `battery_level`, `battery_state`, `timestamp`, `recorded_at` | +| `location_logs` | `id`, `session_id`, `device`, `state`, `station_id`, `line_id`, `segment_id`, `from_station_id`, `to_station_id`, `latitude`, `longitude`, `accuracy`, `speed`, `battery_level`, `battery_state`, `app_version`, `platform`, `channel`, `timestamp`, `recorded_at` | | `log_events` | `id`, `session_id`, `device`, `app_version`, `platform`, `channel`, `log_type`, `log_level`, `message`, `timestamp`, `recorded_at` | | `interaction_events` | `id`, `session_id`, `device`, `app_version`, `platform`, `channel`, `properties` (JSONB), `event_name`, `timestamp`, `recorded_at` | Without a `database_url` the server still accepts WebSocket traffic but does not persist messages. +## Testing + +```bash +# Unit tests only; the PostgreSQL-backed tests skip themselves +cargo test + +# Formatting and lints +cargo fmt --all -- --check +cargo clippy --all-targets -- -D warnings +``` + +The freeze queries also have an integration test that runs against a real PostgreSQL instance. It is skipped unless `THQ_TEST_DATABASE_URL` points at a database the test may create tables in: + +```bash +THQ_TEST_DATABASE_URL=postgres://thq@127.0.0.1:5433/thq_test cargo test +``` + +Each run uses freshly generated device and session identifiers and filters every query by that device, so it is safe to run against a shared scratch database and to run concurrently. + +CI runs the same suite against a `postgres:18` service container on every pull request and on pushes to `main` (`.github/workflows/test.yml`), so the PostgreSQL-backed tests actually execute there. + ## Project structure ```text @@ -382,6 +475,7 @@ src/ ├── domain.rs # Domain model definitions ├── storage.rs # PostgreSQL persistence layer ├── graphql.rs # GraphQL schema & resolvers +├── freeze.rs # Frozen-position detection queries ├── segment.rs # Line topology & segment inference └── static/ └── join.csv # Line topology data diff --git a/docs/location-freeze-regression.md b/docs/location-freeze-regression.md new file mode 100644 index 0000000..58846a8 --- /dev/null +++ b/docs/location-freeze-regression.md @@ -0,0 +1,164 @@ +# 現在地凍結(location freeze)の検出 + +テスト乗車のテレメトリから「アプリは動いているのに現在地が進まなくなった」区間を抽出し、 +路線・区間・機種・ビルド別に集計するための GraphQL Query 3 本(`locationFreezes` / +`locationFreezeSessions` / `locationFreezeSummary`)について説明します。 + +関連 Issue: [TrainLCD/THQ#30](https://github.com/TrainLCD/THQ/issues/30) + +## シグネチャ(何を「凍結」とみなすか) + +MobileApp は乗車中、最大 1 回/秒の頻度で `sendLocation` を送ります。したがって位置ログが +数分単位で途切れていれば、その間クライアントは位置を更新できていません。ただし途切れ自体は +異常とは限らないため、次の 3 条件を **すべて** 満たしたものだけを凍結候補として扱います。 + +1. **位置ログの欠落**: 同一 `session_id` の連続する 2 行の間隔が `gapThresholdMs`(既定 60000 ms)を超える。 +2. **欠落直前の速度が高い**: 欠落直前の行の OS 由来 `speed` が `speedThresholdKmh`(既定 30 km/h)を超える。 +3. **その間もアプリは生きていた**: 欠落期間中に同一 `session_id` の `log_events` / + `interaction_events` が 1 件以上ある(`requireAppAlive: true`、既定)。 + +各条件が切り分けているものは次のとおりです。 + +- 条件 2 は **駅停車** を除外します。停車中に位置更新が止まっても、それは正常です。 +- 条件 3 は **アプリ・端末の死亡** を除外します。強制終了・電源断・バックグラウンド停止で + ログ全体が止まったのであれば、位置ログだけの問題ではありません。 +- 残るのは「アプリのイベントは流れ続けているのに位置ログだけが止まった」ケース、つまり + 画面上の現在地が高速移動中に凍結した状態です。 + +## 3 つの Query の使い分け + +いずれも観測用トークン(`THQ_OBSERVER_AUTH_TOKEN`)と DB 接続が必要で、同じ +`LocationFreezeFilter` を取ります。`from` / `to` は必須で、差は最大 90 日です。 + +| Query | 粒度 | 並び順 | 主な用途 | +|---|---|---|---| +| `locationFreezes` | 欠落 1 件ごと | `gapStart` 降順 | 個別事象の再現条件を調べる | +| `locationFreezeSessions` | セッションごと | `startedAt` 降順 | 1 回の乗車が何回凍結したかを見る | +| `locationFreezeSummary` | 路線・区間・機種・ビルドごと | `freezeCount` 降順、`sessionCount` 降順 | ビルド間・区間別の比較 | + +`locationFreezeSessions` と `locationFreezeSummary` は **凍結が 0 件のセッション / グループも返します**。 +「結果に出てこない」と「凍結が無かった」を区別できないと、ビルド間比較になりません。 + +### 例: 同一区間を 2 つのビルドで走った結果を並べる + +```graphql +query { + locationFreezeSummary( + filter: { + from: "2026-07-01T00:00:00Z" + to: "2026-07-08T00:00:00Z" + segmentId: "11302:1130201:1130202" + } + ) { + appVersion + platform + device + sessionCount + locationCount + freezeSessionCount + freezeCount + maxGapMs + totalGapMs + } +} +``` + +返る行は例えば次のようになります。 + +```text +appVersion device sessionCount freezeSessionCount freezeCount maxGapMs +10.4.1(100) Pixel 8 1 1 1 300000 +10.4.2(101) Pixel 8 1 0 0 null +``` + +同じ区間・同じ端末で `freezeSessionCount` が 1 → 0 になっているので、その区間については +10.4.2(101) で解消したと読めます。逆に `freezeCount` が増えていれば退行です。 + +個別事象の詳細を見るときは `locationFreezes` を使います。 + +```graphql +query { + locationFreezes( + filter: { + from: "2026-07-01T00:00:00Z" + to: "2026-07-08T00:00:00Z" + appVersion: "10.4.1(100)" + } + ) { + sessionId + segmentId + gapStart + gapEnd + gapMs + speedBeforeGap + coordsBeforeGap { latitude longitude accuracy speed } + coordsAfterGap { latitude longitude accuracy speed } + jumpDistanceMeters + aliveEventCount + } +} +``` + +`jumpDistanceMeters` は `coordsBeforeGap` と `coordsAfterGap` の大円距離(m)で、 +凍結中に表示位置が実位置からどれだけ離れたかの目安です。`aliveEventCount` は欠落期間中に +届いた `log_events` + `interaction_events` の件数で、条件 3 の根拠にあたります。 + +## しきい値の考え方 + +- **`gapThresholdMs`**: 既定の 60000 ms は、送信間隔(最大 1 回/秒)の 60 倍にあたります。 + 地下鉄やトンネル区間では測位が正常に途切れるため、そのままでは正常な欠落を拾います。 + 対象を路線で絞る(`lineId` / `segmentId`)か、しきい値を大きめに取ってください。 + THQ は路線種別(地上/地下)を持っていないため、この判断は自動化できません。 +- **`speedThresholdKmh`**: 既定の 30 km/h は、駅停車・徐行と走行中を分けるための値です。 + 在来線の低速区間を見るときは下げ、新幹線の高速域だけを見るときは上げます。 + なお `speed` は OS 由来の値で、欠落直前の 1 行しか見ていません。 +- **`requireAppAlive`**: `false` にすると条件 3 を外し、アプリが落ちた可能性のある欠落も + 含めて返します。「凍結ではなくクラッシュではないか」を確かめるとき、true と false の + 件数差を見るのが手軽です。 +- **`limit`**: 既定 100、上限 2000 に丸められます。 + +## 検索窓と、検出できない欠落 + +`[from, to)` は **欠落の開始行** を絞ります。つまり `from <= gapStart < to` の欠落が対象で、 +欠落を閉じる行(`gapEnd`)は `to` 以降であっても構いません。窓の末尾で始まった欠落も、 +その後に位置ログが再開していれば `gapMs` まで含めて正しく返ります。 + +仕組み上検出できないのは、**閉じない欠落** だけです。電源断・アプリ終了・乗車終了などで +位置ログが二度と来なかった場合、欠落の長さを測る相手が存在しないため、セッション最後の行に +続く空白は返りません。これは条件 3 で除外したいケースとも重なります。 + +一方、`locationFreezeSessions` / `locationFreezeSummary` が返す件数 +(`locationCount` / `startedAt` / `endedAt`)は **窓の中の行だけ** を数えます。 +`to` をまたいだ欠落を閉じる行は、欠落の測定には使われますが、これらの集計には入りません。 + +また、`session_id` が NULL の行(旧クライアント)は対象外です。 + +## 過去データの `appVersion` 補完 + +`location_logs` の `app_version` / `platform` / `channel` は THQ#30 で追加した列です。 +それ以前の行と、まだ送っていないクライアントの行では NULL になります。 + +このため 3 つの Query は、位置ログの列が NULL のときに **同一 `session_id` の +`log_events` / `interaction_events` からビルド情報を補完** します。補完値は、そのセッションの +**最初のイベント行(`app_version` を持つ行を優先)から 3 列をまとめて取ります**。 +列ごとに `MIN` を取ると、別々の行の `appVersion` と `platform` が混ざった実在しない組み合わせに +なり得るためです。 +`appVersion` / `platform` / `channel` フィルタも、この補完後の値に対して適用されます。 +どちらにも情報が無ければ `null` のままです。 + +補完はセッション単位なので、1 セッションの途中でアプリを更新するようなケースは表現できません。 +実運用では乗車中にビルドが変わることはないため、この単純化を採用しています。 + +## 実装メモ + +- SQL は `src/freeze.rs` の共通 CTE 1 本を 3 つの Query で共有し、末尾の `SELECT` だけを差し替えています。 +- `LEAD()` による前後行の対応付けは、**路線・区間・機種などで絞り込む前に**、セッション内の + 全行に対して計算します。先に区間で絞ると区間境界や路線切替をまたぐ隣接行が消え、 + 実際には存在しない欠落が生まれるためです。絞り込みは「欠落直前の行」に対して後段で適用します。 +- 同じ理由で、`LEAD()` の対象行には **`to` の上限を掛けません**。まず窓の中に該当行を持つ + セッションを `candidate_sessions` で選び、`ordered` ではそのセッションの `from` 以降の行を + すべて読みます。`to` は後段の `scoped` で「欠落開始行」に対してだけ適用します。 +- `locationFreezeSessions` はセッションを路線で分割せず、1 セッション 1 行を返します。 + 乗車中に路線が変わっても行が分かれないよう、路線は `lineIds` 配列(昇順)にまとめます。 +- Postgres 統合テストは `THQ_TEST_DATABASE_URL` が設定されているときだけ走ります + (未設定ならスキップ)。詳細は [README](../README.md) の Testing 節を参照してください。 diff --git a/docs/react-tanstack-query.md b/docs/react-tanstack-query.md index 64a6ebf..4bacebf 100644 --- a/docs/react-tanstack-query.md +++ b/docs/react-tanstack-query.md @@ -12,15 +12,16 @@ thq-server の GraphQL API はエンドポイント `POST /graphql` で公開さ | `sendInteractionEvent` | Mutation | イベント用または遠隔測定用トークン | | `sendLocation` | Mutation | 遠隔測定用トークンのみ | | `logEvents` / `interactionEvents` / `locations` | Query | 観測用トークンのみ | +| `locationFreezes` / `locationFreezeSessions` / `locationFreezeSummary` | Query | 観測用トークンのみ | | `accuracyByLine` | Query | 不要 | -Mutation と履歴取得 Query の認証は `Authorization: Bearer ` ヘッダで行います。 +Mutation と履歴取得 Query、および現在地凍結検出 Query の認証は `Authorization: Bearer ` ヘッダで行います。凍結検出 Query の使い方は [location-freeze-regression.md](./location-freeze-regression.md) を参照してください。 | トークン | できること | |---|---| | イベント用(`THQ_EVENTS_AUTH_TOKEN`) | `sendLogEvent` + `sendInteractionEvent` | | 遠隔測定用(`THQ_TELEMETRY_AUTH_TOKEN`) | `sendLogEvent` + `sendInteractionEvent` + `sendLocation` | -| 観測用(`THQ_OBSERVER_AUTH_TOKEN`) | `logEvents` + `interactionEvents` + `locations`(+ WebSocket 購読) | +| 観測用(`THQ_OBSERVER_AUTH_TOKEN`) | `logEvents` + `interactionEvents` + `locations` + `locationFreezes` + `locationFreezeSessions` + `locationFreezeSummary`(+ WebSocket 購読) | > **セキュリティ上の注意**: ブラウザ向けにビルドした JavaScript に埋め込んだトークンは、利用者全員から見えます。イベント用・遠隔測定用トークンを Web フロントエンドに直接埋め込むのは避け、ネイティブアプリや自前のバックエンド(BFF)経由で扱ってください。認証不要な `accuracyByLine` の表示だけであればトークンは一切不要です。 diff --git a/docs/react-websocket-observer.md b/docs/react-websocket-observer.md index afd78e9..e02761c 100644 --- a/docs/react-websocket-observer.md +++ b/docs/react-websocket-observer.md @@ -64,12 +64,17 @@ const ws = new WebSocket("wss://thq.example.com/ws", [ "from_station_id": 1130201, "to_station_id": 1130202, "battery_level": 0.85, - "battery_state": 2 + "battery_state": 2, + "app_version": "10.4.2(101)", + "platform": "ios | android | macos | unknown", + "channel": "production | canary" } ``` `segment_id` / `from_station_id` / `to_station_id` はサーバー側の区間推定によって付与されます(トポロジ未設定時や推定不能時は `null`)。 +`app_version` / `platform` / `channel` はクライアントが `sendLocation` に付けて送るビルド情報で、`log` / `interaction` と同じ値です。送っていないクライアントでは `null` になります(THQ#30 で追加。詳細は [location-freeze-regression.md](./location-freeze-regression.md))。 + **log** — `sendLogEvent` Mutation で登録されたログ ```json @@ -145,6 +150,9 @@ export interface LocationUpdateEvent { to_station_id: number | null; battery_level: number | null; battery_state: 0 | 1 | 2 | 3 | null; // 0: UNKNOWN, 1: UNPLUGGED, 2: CHARGING, 3: FULL + app_version: string | null; // 送っていないクライアントでは null + platform: "ios" | "android" | "macos" | "unknown" | null; + channel: "production" | "canary" | null; } export interface LogEvent { diff --git a/src/domain.rs b/src/domain.rs index 1195e5c..bcc9386 100644 --- a/src/domain.rs +++ b/src/domain.rs @@ -348,6 +348,12 @@ pub struct OutgoingLocation { pub to_station_id: Option, pub battery_level: Option, pub battery_state: Option, + /// Build metadata mirrored from the log/interaction payloads so freeze + /// detection can group location rows by build without a join (THQ#30). + /// `None` for clients that predate the field. + pub app_version: Option, + pub platform: Option, + pub channel: Option, } #[derive(Debug, Clone, Serialize)] @@ -507,11 +513,17 @@ mod tests { to_station_id: None, battery_level: None, battery_state: None, + app_version: Some("10.4.2(101)".into()), + platform: Some(Platform::Ios), + channel: Some(Channel::Production), }); let json = serde_json::to_value(&msg).unwrap(); assert_eq!(json["type"], "location_update"); assert_eq!(json["device"], "dev"); assert_eq!(json["coords"]["speed"], 3.0); + assert_eq!(json["app_version"], "10.4.2(101)"); + assert_eq!(json["platform"], "ios"); + assert_eq!(json["channel"], "production"); } } diff --git a/src/freeze.rs b/src/freeze.rs new file mode 100644 index 0000000..4522f86 --- /dev/null +++ b/src/freeze.rs @@ -0,0 +1,986 @@ +//! Detection of "frozen position" regressions in the location telemetry. +//! +//! The signature this module looks for (see `docs/location-freeze-regression.md`) +//! is a location log gap that satisfies all three of: +//! +//! 1. no location row for longer than `gap_threshold_ms`, +//! 2. the OS-reported speed on the row right before the gap was high, and +//! 3. the app itself kept running during the gap (log or interaction events +//! were still being submitted for the same session). +//! +//! Taken together those rule out a normal stop at a station (2) and an app or +//! device that simply went away (3), leaving the case where the app is alive +//! but its position stopped advancing. + +use sqlx::{postgres::PgArguments, query::QueryAs, Postgres}; + +use crate::storage::Storage; + +/// Mean Earth radius used by [`haversine_meters`]. +const EARTH_RADIUS_METERS: f64 = 6_371_000.0; + +/// Filters shared by the three freeze queries. The time bounds are +/// client-reported unix milliseconds, matching the `timestamp` column. +pub struct LocationFreezeQuery { + /// Inclusive lower bound on the client-reported timestamp, unix millis. + pub from_ts: i64, + /// Exclusive upper bound on the client-reported timestamp, unix millis. + pub to_ts: i64, + pub session_id: Option, + pub line_id: Option, + pub segment_id: Option, + pub device: Option, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// A location gap longer than this many milliseconds is a freeze candidate. + pub gap_threshold_ms: i64, + /// Only gaps whose preceding row reported a speed above this (km/h) count. + pub speed_threshold_kmh: f64, + /// When true, a gap only counts if the app kept emitting events during it. + pub require_app_alive: bool, + pub limit: i32, +} + +/// One detected gap, with the rows on either side of it. +#[derive(Clone, sqlx::FromRow)] +pub struct LocationFreezeRow { + pub session_id: String, + pub device: String, + pub line_id: Option, + pub segment_id: Option, + pub from_station_id: Option, + pub to_station_id: Option, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// Client timestamp of the last row before the gap, unix millis. + pub gap_start: i64, + /// Client timestamp of the first row after the gap, unix millis. + pub gap_end: i64, + pub gap_ms: i64, + pub speed_before_gap: f64, + pub lat_before_gap: f64, + pub lon_before_gap: f64, + pub accuracy_before_gap: Option, + pub lat_after_gap: f64, + pub lon_after_gap: f64, + pub accuracy_after_gap: Option, + pub speed_after_gap: Option, + pub alive_event_count: i32, +} + +/// Per-session rollup. Sessions without any freeze are included so two builds +/// that ran the same segment can be compared side by side. +#[derive(Clone, sqlx::FromRow)] +pub struct LocationFreezeSessionRow { + pub session_id: String, + pub device: String, + /// Lines the session had location rows on inside the window, ascending. + pub line_ids: Vec, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// Client timestamp of the session's first in-window row, unix millis. + pub started_at: i64, + /// Client timestamp of the session's last in-window row, unix millis. + pub ended_at: i64, + pub location_count: i32, + pub max_speed: Option, + pub freeze_count: i32, + pub max_gap_ms: Option, + pub total_gap_ms: i64, +} + +/// Rollup by line / segment / device / build. Groups without any freeze are +/// included for the same reason as [`LocationFreezeSessionRow`]. +#[derive(Clone, sqlx::FromRow)] +pub struct LocationFreezeSummaryRow { + pub line_id: Option, + pub segment_id: Option, + pub from_station_id: Option, + pub to_station_id: Option, + pub device: String, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + pub session_count: i32, + pub location_count: i32, + pub freeze_session_count: i32, + pub freeze_count: i32, + pub max_gap_ms: Option, + pub total_gap_ms: i64, +} + +/// Common CTE prefix shared by the three queries. +/// +/// `LEAD` is deliberately computed over every row of the session, *before* the +/// line / segment / device / build filters are applied: filtering first would +/// drop the neighbouring rows at a segment boundary or a line change and +/// manufacture gaps that never happened. The filters are applied afterwards, in +/// `scoped`, against the row that precedes the gap. +/// +/// The same reasoning applies to the upper time bound. `candidate_sessions` +/// picks the sessions that have at least one matching row inside `[from, to)`, +/// `ordered` then reads *every* row of those sessions from `from` onwards — no +/// upper bound — so a gap can be closed by a row that only arrives at or after +/// `to`. Without that, `LEAD` would return NULL for the last row inside the +/// window and a gap straddling the boundary would silently disappear. `scoped` +/// restores the bound where it belongs: on the row that *starts* the gap +/// (`o.timestamp < $2`), so the window still decides which gaps are reported +/// and the per-window counts (`location_count`, `started_at`, `ended_at`) still +/// only cover rows inside it. +/// +/// Bind order, fixed for all three queries: +/// `$1` from_ts, `$2` to_ts, `$3` session_id, `$4` line_id, `$5` segment_id, +/// `$6` device, `$7` app_version, `$8` platform, `$9` channel, +/// `$10` gap_threshold_ms, `$11` speed_threshold_kmh, `$12` require_app_alive, +/// `$13` limit. +const COMMON_CTE: &str = r#" +WITH candidate_sessions AS ( + SELECT DISTINCT c.session_id + FROM location_logs c + WHERE c.session_id IS NOT NULL + AND c.timestamp >= $1::bigint AND c.timestamp < $2::bigint + AND ($3::text IS NULL OR c.session_id = $3) + AND ($4::int IS NULL OR c.line_id = $4) + AND ($5::text IS NULL OR c.segment_id = $5) + AND ($6::text IS NULL OR c.device = $6) +), +ordered AS ( + SELECT l.session_id, l.device, l.line_id, l.segment_id, l.from_station_id, l.to_station_id, + l.latitude, l.longitude, l.accuracy, l.speed, l.timestamp, + l.app_version, l.platform, l.channel, + LEAD(l.timestamp) OVER w AS next_timestamp, + LEAD(l.latitude) OVER w AS next_latitude, + LEAD(l.longitude) OVER w AS next_longitude, + LEAD(l.accuracy) OVER w AS next_accuracy, + LEAD(l.speed) OVER w AS next_speed + FROM location_logs l + WHERE l.session_id IN (SELECT session_id FROM candidate_sessions) + AND l.timestamp >= $1::bigint + WINDOW w AS (PARTITION BY l.session_id ORDER BY l.timestamp) +), +session_meta AS ( + SELECT DISTINCT ON (e.session_id) e.session_id, e.app_version, e.platform, e.channel + FROM ( + SELECT session_id, app_version, platform, channel, timestamp FROM log_events + WHERE session_id IN (SELECT session_id FROM candidate_sessions) + UNION ALL + SELECT session_id, app_version, platform, channel, timestamp FROM interaction_events + WHERE session_id IN (SELECT session_id FROM candidate_sessions) + ) e + ORDER BY e.session_id, (e.app_version IS NULL), e.timestamp +), +scoped AS ( + SELECT o.*, + COALESCE(o.app_version, m.app_version) AS eff_app_version, + COALESCE(o.platform, m.platform) AS eff_platform, + COALESCE(o.channel, m.channel) AS eff_channel + FROM ordered o LEFT JOIN session_meta m ON m.session_id = o.session_id + WHERE o.timestamp < $2::bigint + AND ($4::int IS NULL OR o.line_id = $4) + AND ($5::text IS NULL OR o.segment_id = $5) + AND ($6::text IS NULL OR o.device = $6) + AND ($7::text IS NULL OR COALESCE(o.app_version, m.app_version) = $7) + AND ($8::text IS NULL OR COALESCE(o.platform, m.platform) = $8) + AND ($9::text IS NULL OR COALESCE(o.channel, m.channel) = $9) +), +freezes AS ( + SELECT s.*, (s.next_timestamp - s.timestamp) AS gap_ms, + ( + (SELECT COUNT(*) FROM log_events e + WHERE e.session_id = s.session_id AND e.timestamp > s.timestamp AND e.timestamp < s.next_timestamp) + + + (SELECT COUNT(*) FROM interaction_events i + WHERE i.session_id = s.session_id AND i.timestamp > s.timestamp AND i.timestamp < s.next_timestamp) + )::int AS alive_event_count + FROM scoped s + WHERE s.next_timestamp IS NOT NULL + AND s.next_timestamp - s.timestamp > $10::bigint + AND s.speed IS NOT NULL AND s.speed > $11::double precision +), +freezes_alive AS ( + SELECT * FROM freezes WHERE (NOT $12::bool) OR alive_event_count > 0 +) +"#; + +/// Tail of the detail query: one row per detected gap, newest first. +const FREEZES_TAIL: &str = r#" +SELECT f.session_id, + f.device, + f.line_id, + f.segment_id, + f.from_station_id, + f.to_station_id, + f.eff_app_version AS app_version, + f.eff_platform AS platform, + f.eff_channel AS channel, + f.timestamp AS gap_start, + f.next_timestamp AS gap_end, + f.gap_ms, + f.speed AS speed_before_gap, + f.latitude AS lat_before_gap, + f.longitude AS lon_before_gap, + f.accuracy AS accuracy_before_gap, + f.next_latitude AS lat_after_gap, + f.next_longitude AS lon_after_gap, + f.next_accuracy AS accuracy_after_gap, + f.next_speed AS speed_after_gap, + f.alive_event_count +FROM freezes_alive f +ORDER BY f.timestamp DESC +LIMIT $13 +"#; + +/// Tail of the per-session query. `session_stats` covers every session in the +/// window, so sessions with zero freezes still show up. +/// +/// A session is one row here even when it rode several lines: `line_id` would +/// otherwise split the same ride into one row per line and each of those rows +/// would report only part of the session. The lines are exposed as an array +/// instead. +const SESSIONS_TAIL: &str = r#" +, session_stats AS ( + SELECT s.session_id, s.device, + s.eff_app_version, s.eff_platform, s.eff_channel, + COALESCE(array_agg(DISTINCT s.line_id ORDER BY s.line_id) + FILTER (WHERE s.line_id IS NOT NULL), '{}'::int[]) AS line_ids, + MIN(s.timestamp)::bigint AS started_at, + MAX(s.timestamp)::bigint AS ended_at, + COUNT(*)::int AS location_count, + MAX(s.speed) AS max_speed + FROM scoped s + GROUP BY s.session_id, s.device, s.eff_app_version, s.eff_platform, s.eff_channel +), +session_freezes AS ( + SELECT f.session_id, + COUNT(*)::int AS freeze_count, + MAX(f.gap_ms)::bigint AS max_gap_ms, + COALESCE(SUM(f.gap_ms), 0)::bigint AS total_gap_ms + FROM freezes_alive f + GROUP BY f.session_id +) +SELECT s.session_id, + s.device, + s.line_ids, + s.eff_app_version AS app_version, + s.eff_platform AS platform, + s.eff_channel AS channel, + s.started_at, + s.ended_at, + s.location_count, + s.max_speed, + COALESCE(f.freeze_count, 0) AS freeze_count, + f.max_gap_ms, + COALESCE(f.total_gap_ms, 0)::bigint AS total_gap_ms +FROM session_stats s +LEFT JOIN session_freezes f ON f.session_id = s.session_id +ORDER BY s.started_at DESC +LIMIT $13 +"#; + +/// Tail of the aggregated query, grouped by line / segment / device / build. +const SUMMARY_TAIL: &str = r#" +, group_stats AS ( + SELECT s.line_id, s.segment_id, s.device, + s.eff_app_version, s.eff_platform, s.eff_channel, + MIN(s.from_station_id) AS from_station_id, + MIN(s.to_station_id) AS to_station_id, + COUNT(DISTINCT s.session_id)::int AS session_count, + COUNT(*)::int AS location_count + FROM scoped s + GROUP BY s.line_id, s.segment_id, s.device, s.eff_app_version, s.eff_platform, s.eff_channel +), +group_freezes AS ( + SELECT f.line_id, f.segment_id, f.device, + f.eff_app_version, f.eff_platform, f.eff_channel, + COUNT(DISTINCT f.session_id)::int AS freeze_session_count, + COUNT(*)::int AS freeze_count, + MAX(f.gap_ms)::bigint AS max_gap_ms, + COALESCE(SUM(f.gap_ms), 0)::bigint AS total_gap_ms + FROM freezes_alive f + GROUP BY f.line_id, f.segment_id, f.device, f.eff_app_version, f.eff_platform, f.eff_channel +) +SELECT g.line_id, + g.segment_id, + g.from_station_id, + g.to_station_id, + g.device, + g.eff_app_version AS app_version, + g.eff_platform AS platform, + g.eff_channel AS channel, + g.session_count, + g.location_count, + COALESCE(f.freeze_session_count, 0) AS freeze_session_count, + COALESCE(f.freeze_count, 0) AS freeze_count, + f.max_gap_ms, + COALESCE(f.total_gap_ms, 0)::bigint AS total_gap_ms +FROM group_stats g +LEFT JOIN group_freezes f + ON f.device = g.device + AND f.line_id IS NOT DISTINCT FROM g.line_id + AND f.segment_id IS NOT DISTINCT FROM g.segment_id + AND f.eff_app_version IS NOT DISTINCT FROM g.eff_app_version + AND f.eff_platform IS NOT DISTINCT FROM g.eff_platform + AND f.eff_channel IS NOT DISTINCT FROM g.eff_channel +ORDER BY freeze_count DESC, session_count DESC +LIMIT $13 +"#; + +/// Concatenates the shared CTE with a query-specific tail. +fn freeze_sql(tail: &str) -> String { + format!("{COMMON_CTE}{tail}") +} + +/// Applies the 13 shared bind parameters in the order documented on [`COMMON_CTE`]. +fn bind_filter<'q, T>( + query: QueryAs<'q, Postgres, T, PgArguments>, + filter: &'q LocationFreezeQuery, +) -> QueryAs<'q, Postgres, T, PgArguments> { + query + .bind(filter.from_ts) + .bind(filter.to_ts) + .bind(&filter.session_id) + .bind(filter.line_id) + .bind(&filter.segment_id) + .bind(&filter.device) + .bind(&filter.app_version) + .bind(&filter.platform) + .bind(&filter.channel) + .bind(filter.gap_threshold_ms) + .bind(filter.speed_threshold_kmh) + .bind(filter.require_app_alive) + .bind(filter.limit) +} + +/// Great-circle distance between two WGS84 points, in meters. +/// +/// Used for `jumpDistanceMeters`: how far the reported position jumped while +/// it was frozen, i.e. roughly how far the displayed position had drifted from +/// the real one by the time updates resumed. +pub(crate) fn haversine_meters(lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 { + let phi1 = lat1.to_radians(); + let phi2 = lat2.to_radians(); + let delta_phi = (lat2 - lat1).to_radians(); + let delta_lambda = (lon2 - lon1).to_radians(); + + let a = (delta_phi / 2.0).sin().powi(2) + + phi1.cos() * phi2.cos() * (delta_lambda / 2.0).sin().powi(2); + // atan2 form stays accurate for antipodal points, where asin saturates. + 2.0 * EARTH_RADIUS_METERS * a.sqrt().atan2((1.0 - a).max(0.0).sqrt()) +} + +impl Storage { + /// Detected location freezes, newest gap first. + /// + /// # Errors + /// + /// Returns an error when the database is not configured or the query fails. + pub async fn fetch_location_freezes( + &self, + filter: &LocationFreezeQuery, + ) -> anyhow::Result> { + let pool = self.pool()?; + let sql = freeze_sql(FREEZES_TAIL); + let rows = bind_filter(sqlx::query_as::<_, LocationFreezeRow>(&sql), filter) + .fetch_all(pool) + .await?; + Ok(rows) + } + + /// Per-session freeze rollup, newest session first. Sessions with no freeze + /// are included. + /// + /// # Errors + /// + /// Returns an error when the database is not configured or the query fails. + pub async fn fetch_location_freeze_sessions( + &self, + filter: &LocationFreezeQuery, + ) -> anyhow::Result> { + let pool = self.pool()?; + let sql = freeze_sql(SESSIONS_TAIL); + let rows = bind_filter(sqlx::query_as::<_, LocationFreezeSessionRow>(&sql), filter) + .fetch_all(pool) + .await?; + Ok(rows) + } + + /// Freeze rollup by line / segment / device / build, worst group first. + /// Groups with no freeze are included. + /// + /// # Errors + /// + /// Returns an error when the database is not configured or the query fails. + pub async fn fetch_location_freeze_summary( + &self, + filter: &LocationFreezeQuery, + ) -> anyhow::Result> { + let pool = self.pool()?; + let sql = freeze_sql(SUMMARY_TAIL); + let rows = bind_filter(sqlx::query_as::<_, LocationFreezeSummaryRow>(&sql), filter) + .fetch_all(pool) + .await?; + Ok(rows) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn haversine_is_zero_for_the_same_point() { + assert_eq!(haversine_meters(35.6812, 139.7671, 35.6812, 139.7671), 0.0); + } + + #[test] + fn haversine_matches_tokyo_to_shin_osaka() { + // Tokyo station -> Shin-Osaka station, ~400 km great-circle. + let d = haversine_meters(35.681236, 139.767125, 34.733380, 135.500218); + assert!( + (390_000.0..410_000.0).contains(&d), + "unexpected distance: {d}" + ); + } + + #[test] + fn haversine_matches_a_meridian_offset() { + // One degree of latitude is ~111.19 km on a sphere of radius 6371 km. + let d = haversine_meters(35.0, 139.0, 36.0, 139.0); + assert!((d - 111_195.0).abs() < 50.0, "unexpected distance: {d}"); + } + + #[test] + fn common_cte_binds_are_shared_by_every_tail() { + for tail in [FREEZES_TAIL, SESSIONS_TAIL, SUMMARY_TAIL] { + let sql = freeze_sql(tail); + assert!(sql.starts_with("\nWITH candidate_sessions AS")); + assert!(sql.contains("freezes_alive")); + // the limit is always the last bind parameter + assert!(sql.contains("LIMIT $13")); + } + } + + // --------------------------------------------------------------------- + // PostgreSQL integration test. + // + // Skipped unless THQ_TEST_DATABASE_URL points at a database the test may + // create tables in, e.g. + // + // THQ_TEST_DATABASE_URL=postgres://thq@127.0.0.1:5433/thq_test cargo test + // + // Every run uses a fresh uuid device and session ids, and every query + // filters on that device, so concurrent runs cannot see each other's rows. + // --------------------------------------------------------------------- + + use crate::domain::{ + Channel, LogBody, LogLevel, LogType, MovementState, OutgoingCoords, OutgoingInteraction, + OutgoingLocation, OutgoingLog, Platform, + }; + use uuid::Uuid; + + const BASE_TS: i64 = 1_700_000_000_000; + const LAT0: f64 = 35.681236; + const LON0: f64 = 139.767125; + /// ~30 km north of `LAT0` (30 000 m / 6 371 000 m, in degrees). + const LAT_JUMP: f64 = 0.269_795; + + fn location( + session_id: &str, + device: &str, + ts: i64, + lat: f64, + speed: Option, + app_version: Option<&str>, + ) -> OutgoingLocation { + OutgoingLocation { + id: Uuid::new_v4().to_string(), + session_id: session_id.to_string(), + device: device.to_string(), + state: MovementState::Moving, + station_id: None, + line_id: 1, + coords: OutgoingCoords { + latitude: lat, + longitude: LON0, + accuracy: Some(5.0), + speed, + }, + timestamp: ts as u64, + segment_id: Some("1:101:102".to_string()), + from_station_id: Some(101), + to_station_id: Some(102), + battery_level: Some(0.8), + battery_state: None, + app_version: app_version.map(str::to_string), + platform: app_version.map(|_| Platform::Ios), + channel: app_version.map(|_| Channel::Canary), + } + } + + fn log_event(session_id: &str, ts: i64, app_version: &str) -> OutgoingLog { + log_event_on(session_id, ts, app_version, Platform::Ios) + } + + fn log_event_on( + session_id: &str, + ts: i64, + app_version: &str, + platform: Platform, + ) -> OutgoingLog { + OutgoingLog { + id: Uuid::new_v4().to_string(), + session_id: session_id.to_string(), + device: None, + app_version: app_version.to_string(), + platform, + channel: Channel::Canary, + timestamp: ts as u64, + log: LogBody { + r#type: LogType::App, + level: LogLevel::Info, + message: "still alive".to_string(), + }, + } + } + + fn interaction_event(session_id: &str, ts: i64, app_version: &str) -> OutgoingInteraction { + OutgoingInteraction { + id: Uuid::new_v4().to_string(), + session_id: session_id.to_string(), + device: None, + app_version: app_version.to_string(), + platform: Platform::Ios, + channel: Channel::Canary, + timestamp: ts as u64, + event_name: "tab_change".to_string(), + properties: None, + } + } + + fn base_filter(device: &str) -> LocationFreezeQuery { + LocationFreezeQuery { + from_ts: BASE_TS - 60_000, + to_ts: BASE_TS + 3_600_000, + session_id: None, + line_id: None, + segment_id: None, + device: Some(device.to_string()), + app_version: None, + platform: None, + channel: None, + gap_threshold_ms: 60_000, + speed_threshold_kmh: 30.0, + require_app_alive: true, + limit: 100, + } + } + + #[tokio::test] + async fn freeze_queries_detect_the_mobileapp_6883_signature() { + let Ok(url) = std::env::var("THQ_TEST_DATABASE_URL") else { + eprintln!( + "skipping freeze_queries_detect_the_mobileapp_6883_signature: \ + set THQ_TEST_DATABASE_URL to run it" + ); + return; + }; + + let storage = Storage::connect(Some(url)) + .await + .expect("connect to the test database"); + + let device = format!("dev-{}", Uuid::new_v4()); + let session_a = format!("a-{}", Uuid::new_v4()); + let session_b = format!("b-{}", Uuid::new_v4()); + let session_c = format!("c-{}", Uuid::new_v4()); + let session_d = format!("d-{}", Uuid::new_v4()); + let session_e = format!("e-{}", Uuid::new_v4()); + let session_f = format!("f-{}", Uuid::new_v4()); + + // Session A: 320 km/h, 30 s of 1 Hz rows, then a 300 s hole during + // which only log events arrive, then rows resume 30 km further north. + // Its location rows carry no build metadata, so the 10.4.1(100) shown + // by the query has to come from the session's log events. + for i in 0..30 { + storage + .store_location(&location( + &session_a, + &device, + BASE_TS + i * 1_000, + LAT0, + Some(320.0), + None, + )) + .await + .expect("store session A pre-gap row"); + } + for i in 0..30 { + storage + .store_location(&location( + &session_a, + &device, + BASE_TS + 329_000 + i * 1_000, + LAT0 + LAT_JUMP, + Some(320.0), + None, + )) + .await + .expect("store session A post-gap row"); + } + // 10 log events strictly inside the gap (29 000 .. 329 000). + for k in 0..10 { + storage + .store_log(&log_event( + &session_a, + BASE_TS + 54_000 + k * 30_000, + "10.4.1(100)", + )) + .await + .expect("store session A in-gap log event"); + } + // One interaction event outside the gap: it must feed the app_version + // backfill without inflating aliveEventCount. + storage + .store_interaction(&interaction_event( + &session_a, + BASE_TS + 5_000, + "10.4.1(100)", + )) + .await + .expect("store session A interaction event"); + // A later event from a different build. The backfill must take all + // three columns from the *first* event row rather than the minimum of + // each column, otherwise this row's platform ("android", which sorts + // before "ios") would be pasted onto 10.4.1(100). + storage + .store_log(&log_event_on( + &session_a, + BASE_TS + 400_000, + "10.4.9(999)", + Platform::Android, + )) + .await + .expect("store session A late log event"); + + // Session B: the same segment on a newer build, no gap at all. + for i in 0..120 { + storage + .store_location(&location( + &session_b, + &device, + BASE_TS + i * 1_000, + LAT0, + Some(320.0), + Some("10.4.2(101)"), + )) + .await + .expect("store session B row"); + } + + // Session C: stopped at a station, so the same 300 s hole is expected. + for i in 0..10 { + storage + .store_location(&location( + &session_c, + &device, + BASE_TS + i * 1_000, + LAT0, + Some(0.0), + Some("10.4.4(103)"), + )) + .await + .expect("store session C pre-gap row"); + } + for i in 0..10 { + storage + .store_location(&location( + &session_c, + &device, + BASE_TS + 309_000 + i * 1_000, + LAT0, + Some(0.0), + Some("10.4.4(103)"), + )) + .await + .expect("store session C post-gap row"); + } + + // Session D: same hole at speed, but nothing proves the app was alive. + for i in 0..10 { + storage + .store_location(&location( + &session_d, + &device, + BASE_TS + i * 1_000, + LAT0, + Some(320.0), + Some("10.4.3(102)"), + )) + .await + .expect("store session D pre-gap row"); + } + for i in 0..10 { + storage + .store_location(&location( + &session_d, + &device, + BASE_TS + 309_000 + i * 1_000, + LAT0 + LAT_JUMP, + Some(320.0), + Some("10.4.3(102)"), + )) + .await + .expect("store session D post-gap row"); + } + + // Session E: the gap starts just inside the window but is only closed + // by a row that arrives after `to`. The freeze is real, so `LEAD` has + // to see past the upper bound of the search window. + for i in 0..10 { + storage + .store_location(&location( + &session_e, + &device, + BASE_TS + 3_590_000 + i * 1_000, + LAT0, + Some(320.0), + Some("10.4.5(104)"), + )) + .await + .expect("store session E pre-gap row"); + } + storage + .store_location(&location( + &session_e, + &device, + BASE_TS + 3_900_000, + LAT0 + LAT_JUMP, + Some(320.0), + Some("10.4.5(104)"), + )) + .await + .expect("store session E post-window row"); + // 3 log events strictly inside the gap (3 599 000 .. 3 900 000). + for k in 0..3 { + storage + .store_log(&log_event( + &session_e, + BASE_TS + 3_600_000 + k * 30_000, + "10.4.5(104)", + )) + .await + .expect("store session E in-gap log event"); + } + + // Session F: the same shape at speed, but the session simply stops — + // no row ever closes the gap, so there is nothing to measure. + for i in 0..10 { + storage + .store_location(&location( + &session_f, + &device, + BASE_TS + 3_590_000 + i * 1_000, + LAT0, + Some(320.0), + Some("10.4.6(105)"), + )) + .await + .expect("store session F row"); + } + for k in 0..3 { + storage + .store_log(&log_event( + &session_f, + BASE_TS + 3_600_000 + k * 30_000, + "10.4.6(105)", + )) + .await + .expect("store session F post-window log event"); + } + + // --- locationFreezes ------------------------------------------------- + let filter = base_filter(&device); + let freezes = storage + .fetch_location_freezes(&filter) + .await + .expect("fetch freezes"); + + let ids: Vec<&str> = freezes.iter().map(|r| r.session_id.as_str()).collect(); + assert_eq!( + freezes.len(), + 2, + "sessions A and E satisfy all three conditions, got {ids:?}" + ); + // newest gap first, so the boundary-crossing session E leads + let e = &freezes[0]; + assert_eq!(e.session_id, session_e); + assert_eq!(e.gap_start, BASE_TS + 3_599_000); + assert_eq!( + e.gap_end, + BASE_TS + 3_900_000, + "the closing row lies past the window and must still be used" + ); + assert_eq!(e.gap_ms, 301_000); + assert_eq!(e.alive_event_count, 3); + assert_eq!(e.app_version.as_deref(), Some("10.4.5(104)")); + + let a = &freezes[1]; + assert_eq!(a.session_id, session_a); + assert_eq!(a.gap_ms, 300_000); + assert_eq!(a.gap_start, BASE_TS + 29_000); + assert_eq!(a.gap_end, BASE_TS + 329_000); + assert_eq!(a.alive_event_count, 10); + assert_eq!(a.speed_before_gap, 320.0); + assert_eq!(a.segment_id.as_deref(), Some("1:101:102")); + // backfilled from the session's log / interaction events + assert_eq!(a.app_version.as_deref(), Some("10.4.1(100)")); + assert_eq!(a.platform.as_deref(), Some("ios")); + assert_eq!(a.channel.as_deref(), Some("canary")); + let jump = haversine_meters( + a.lat_before_gap, + a.lon_before_gap, + a.lat_after_gap, + a.lon_after_gap, + ); + assert!( + (29_000.0..31_000.0).contains(&jump), + "unexpected jump distance: {jump}" + ); + + // Dropping the liveness requirement also surfaces session D. + let lenient = LocationFreezeQuery { + require_app_alive: false, + ..base_filter(&device) + }; + let freezes = storage + .fetch_location_freezes(&lenient) + .await + .expect("fetch freezes without the liveness requirement"); + let ids: Vec<&str> = freezes.iter().map(|r| r.session_id.as_str()).collect(); + assert_eq!(freezes.len(), 3, "sessions A, D and E, got {ids:?}"); + assert!(ids.contains(&session_a.as_str())); + assert!(ids.contains(&session_d.as_str())); + assert!(ids.contains(&session_e.as_str())); + assert!( + !ids.contains(&session_f.as_str()), + "a gap that never closes cannot be measured, got {ids:?}" + ); + + // --- locationFreezeSessions ----------------------------------------- + let sessions = storage + .fetch_location_freeze_sessions(&filter) + .await + .expect("fetch freeze sessions"); + assert_eq!(sessions.len(), 6, "every session in the window is listed"); + + let by_id = |id: &str| { + sessions + .iter() + .find(|r| r.session_id == id) + .unwrap_or_else(|| panic!("session {id} missing")) + }; + let row_a = by_id(&session_a); + assert_eq!(row_a.freeze_count, 1); + assert_eq!(row_a.max_gap_ms, Some(300_000)); + assert_eq!(row_a.total_gap_ms, 300_000); + assert_eq!(row_a.location_count, 60); + assert_eq!(row_a.started_at, BASE_TS); + assert_eq!(row_a.ended_at, BASE_TS + 358_000); + // the first event row wins, and all three columns come from that one + // row: the later 10.4.9(999) / android event must not leak in + assert_eq!(row_a.app_version.as_deref(), Some("10.4.1(100)")); + assert_eq!(row_a.platform.as_deref(), Some("ios")); + assert_eq!(row_a.channel.as_deref(), Some("canary")); + assert_eq!(row_a.max_speed, Some(320.0)); + assert_eq!( + row_a.line_ids, + vec![1], + "one row per session, lines rolled up" + ); + + let row_b = by_id(&session_b); + assert_eq!(row_b.freeze_count, 0); + assert_eq!(row_b.max_gap_ms, None); + assert_eq!(row_b.total_gap_ms, 0); + assert_eq!(row_b.location_count, 120); + assert_eq!(row_b.app_version.as_deref(), Some("10.4.2(101)")); + + assert_eq!(row_b.line_ids, vec![1]); + + assert_eq!(by_id(&session_c).freeze_count, 0); + assert_eq!(by_id(&session_d).freeze_count, 0); + + let row_e = by_id(&session_e); + assert_eq!(row_e.freeze_count, 1, "the gap crosses the window boundary"); + assert_eq!(row_e.max_gap_ms, Some(301_000)); + assert_eq!(row_e.total_gap_ms, 301_000); + assert_eq!( + row_e.location_count, 10, + "counts only cover rows inside the window" + ); + assert_eq!(row_e.ended_at, BASE_TS + 3_599_000); + assert_eq!(row_e.line_ids, vec![1]); + + let row_f = by_id(&session_f); + assert_eq!(row_f.freeze_count, 0); + assert_eq!(row_f.max_gap_ms, None); + + // --- locationFreezeSummary ------------------------------------------ + let summary = storage + .fetch_location_freeze_summary(&filter) + .await + .expect("fetch freeze summary"); + assert_eq!(summary.len(), 6, "one row per build on this segment"); + + let build = |v: &str| { + summary + .iter() + .find(|r| r.app_version.as_deref() == Some(v)) + .unwrap_or_else(|| panic!("build {v} missing from the summary")) + }; + let old = build("10.4.1(100)"); + assert_eq!(old.freeze_session_count, 1); + assert_eq!(old.freeze_count, 1); + assert_eq!(old.session_count, 1); + assert_eq!(old.location_count, 60); + assert_eq!(old.max_gap_ms, Some(300_000)); + assert_eq!(old.total_gap_ms, 300_000); + assert_eq!(old.segment_id.as_deref(), Some("1:101:102")); + assert_eq!(old.from_station_id, Some(101)); + assert_eq!(old.to_station_id, Some(102)); + + let new = build("10.4.2(101)"); + assert_eq!(new.freeze_session_count, 0); + assert_eq!(new.freeze_count, 0); + assert_eq!(new.session_count, 1); + assert_eq!(new.location_count, 120); + assert_eq!(new.max_gap_ms, None); + assert_eq!(new.total_gap_ms, 0); + + let boundary = build("10.4.5(104)"); + assert_eq!(boundary.freeze_session_count, 1); + assert_eq!(boundary.freeze_count, 1); + assert_eq!(boundary.location_count, 10); + assert_eq!(boundary.max_gap_ms, Some(301_000)); + + let never_closed = build("10.4.6(105)"); + assert_eq!(never_closed.freeze_count, 0); + assert_eq!(never_closed.max_gap_ms, None); + + // the worst groups sort first: both freezing builds lead the clean ones + assert_eq!(summary[0].freeze_count, 1); + assert_eq!(summary[1].freeze_count, 1); + let worst: Vec<&str> = summary[..2] + .iter() + .filter_map(|r| r.app_version.as_deref()) + .collect(); + assert!( + worst.contains(&"10.4.1(100)") && worst.contains(&"10.4.5(104)"), + "unexpected worst groups: {worst:?}" + ); + assert_eq!(summary[2].freeze_count, 0); + } +} diff --git a/src/graphql.rs b/src/graphql.rs index d73c402..7b6e188 100644 --- a/src/graphql.rs +++ b/src/graphql.rs @@ -12,6 +12,10 @@ use crate::{ BatteryState, Channel, LogBody, LogLevel, LogType, MovementState, OutgoingCoords, OutgoingInteraction, OutgoingLocation, OutgoingLog, OutgoingMessage, Platform, Properties, }, + freeze::{ + haversine_meters, LocationFreezeQuery, LocationFreezeRow, LocationFreezeSessionRow, + LocationFreezeSummaryRow, + }, segment::SegmentEstimator, state::TelemetryHub, storage::{ @@ -40,6 +44,14 @@ pub struct RequestAuth { const HARD_LIMIT: i32 = 2000; +/// Widest range the freeze queries accept, mirroring the `hour` bucket cap of +/// `accuracyByLine`. +const FREEZE_MAX_SPAN_DAYS: i64 = 90; + +/// The app sends at most one location per second, so anything below this would +/// flag ordinary jitter rather than a freeze. +const FREEZE_MIN_GAP_THRESHOLD_MS: i32 = 1000; + #[derive(Enum, Copy, Clone, Eq, PartialEq, Debug)] #[graphql(rename_items = "lowercase")] pub enum TimeBucketSize { @@ -170,10 +182,130 @@ pub struct LocationEvent { /// Battery level as a decimal (0.0 to 1.0). pub battery_level: Option, pub battery_state: Option, + /// Build metadata reported alongside the position. Null for rows written + /// before the columns existed or by clients that do not send them. + pub app_version: Option, + pub platform: Option, + pub channel: Option, /// Server-side time the event was persisted. pub recorded_at: DateTime, } +/// Filters shared by `locationFreezes`, `locationFreezeSessions` and +/// `locationFreezeSummary`. See `docs/location-freeze-regression.md`. +#[derive(InputObject)] +pub struct LocationFreezeFilter { + /// Inclusive lower bound on the client-reported timestamp. + pub from: DateTime, + /// Exclusive upper bound on the client-reported timestamp. At most 90 days + /// after `from`. + pub to: DateTime, + pub line_id: Option, + /// Server-assigned segment ID, matched against the row before the gap. + pub segment_id: Option, + pub device: Option, + pub session_id: Option, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// A location gap longer than this many milliseconds counts as a freeze + /// candidate. The app sends at most one location per second, so the default + /// of 60 000 is already two orders of magnitude above normal jitter. + #[graphql(default = 60000)] + pub gap_threshold_ms: i32, + /// Only gaps whose preceding row reported a speed (km/h) above this count, + /// which is what separates a freeze from a stop at a station. + #[graphql(default = 30.0)] + pub speed_threshold_kmh: f64, + /// When true (the default), a gap only counts if the same session kept + /// submitting log or interaction events while the position was missing — + /// evidence that the app itself did not die. + #[graphql(default = true)] + pub require_app_alive: bool, +} + +/// A single location log gap that matches the freeze signature. +#[derive(SimpleObject, Clone)] +pub struct LocationFreeze { + pub session_id: String, + pub device: String, + pub line_id: Option, + pub segment_id: Option, + pub from_station_id: Option, + pub to_station_id: Option, + /// Taken from `location_logs`, falling back to the log / interaction events + /// of the same session; null when neither carries it. + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// Client timestamp of the last row before the gap. + pub gap_start: DateTime, + /// Client timestamp of the first row after the gap. + pub gap_end: DateTime, + /// Length of the gap in milliseconds. + pub gap_ms: u64, + /// Speed (km/h) reported on the row before the gap. + pub speed_before_gap: f64, + /// Coordinates of the row before the gap, i.e. where the display froze. + pub coords_before_gap: Coords, + /// Coordinates of the first row after the gap. + pub coords_after_gap: Coords, + /// Great-circle distance in meters between the two rows above: roughly how + /// far the displayed position had drifted from reality. + pub jump_distance_meters: f64, + /// Log + interaction events of the same session inside the gap. + pub alive_event_count: i32, +} + +/// Per-session rollup. Sessions with zero freezes are included so two builds +/// that rode the same segment can be compared side by side. +#[derive(SimpleObject, Clone)] +pub struct LocationFreezeSession { + pub session_id: String, + pub device: String, + /// Lines the session had location rows on inside the window, ascending; + /// empty when every row lacked a line. + pub line_ids: Vec, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// Client timestamp of the session's first row inside the window. + pub started_at: DateTime, + /// Client timestamp of the session's last row inside the window. + pub ended_at: DateTime, + pub location_count: i32, + pub max_speed: Option, + pub freeze_count: i32, + /// Null when the session has no freeze. + pub max_gap_ms: Option, + /// Zero when the session has no freeze. + pub total_gap_ms: u64, +} + +/// Rollup by line, segment, device and build. Groups with zero freezes are +/// included, so a build can be shown to be clean rather than merely absent. +#[derive(SimpleObject, Clone)] +pub struct LocationFreezeSummary { + pub line_id: Option, + pub segment_id: Option, + pub from_station_id: Option, + pub to_station_id: Option, + pub device: String, + pub app_version: Option, + pub platform: Option, + pub channel: Option, + /// Sessions with at least one location row in this group. + pub session_count: i32, + pub location_count: i32, + /// Sessions in this group with at least one freeze. + pub freeze_session_count: i32, + pub freeze_count: i32, + /// Null when the group has no freeze. + pub max_gap_ms: Option, + /// Zero when the group has no freeze. + pub total_gap_ms: u64, +} + /// Builds the application GraphQL schema with storage, telemetry, and segment-estimation dependencies. /// /// # Examples @@ -469,6 +601,169 @@ impl QueryRoot { Ok(rows.into_iter().map(LocationEvent::from).collect()) } + + /// Location log gaps that match the frozen-position signature, newest gap + /// first. Requires the observer token and a configured database. + /// + /// See `docs/location-freeze-regression.md` for the three conditions and + /// for the gaps this deliberately cannot see (a gap whose closing row falls + /// outside the window, and a session that never comes back). + async fn location_freezes( + &self, + ctx: &Context<'_>, + filter: LocationFreezeFilter, + #[graphql(default = 100)] limit: i32, + ) -> Result> { + let (storage, query) = freeze_query(ctx, filter, limit)?; + + let started = Instant::now(); + let rows = storage + .fetch_location_freezes(&query) + .await + .map_err(|e| format!("failed to fetch location freezes: {e}"))?; + + info!( + count = rows.len(), + limit = query.limit, + gap_threshold_ms = query.gap_threshold_ms, + speed_threshold_kmh = query.speed_threshold_kmh, + require_app_alive = query.require_app_alive, + duration_ms = started.elapsed().as_millis(), + "locationFreezes resolver completed" + ); + + Ok(rows.into_iter().map(LocationFreeze::from).collect()) + } + + /// Per-session freeze rollup, newest session first. Sessions without a + /// freeze are included. Requires the observer token. + async fn location_freeze_sessions( + &self, + ctx: &Context<'_>, + filter: LocationFreezeFilter, + #[graphql(default = 100)] limit: i32, + ) -> Result> { + let (storage, query) = freeze_query(ctx, filter, limit)?; + + let started = Instant::now(); + let rows = storage + .fetch_location_freeze_sessions(&query) + .await + .map_err(|e| format!("failed to fetch location freeze sessions: {e}"))?; + + info!( + count = rows.len(), + limit = query.limit, + gap_threshold_ms = query.gap_threshold_ms, + speed_threshold_kmh = query.speed_threshold_kmh, + require_app_alive = query.require_app_alive, + duration_ms = started.elapsed().as_millis(), + "locationFreezeSessions resolver completed" + ); + + Ok(rows.into_iter().map(LocationFreezeSession::from).collect()) + } + + /// Freeze rollup by line, segment, device and build, worst group first. + /// Groups without a freeze are included. Requires the observer token. + async fn location_freeze_summary( + &self, + ctx: &Context<'_>, + filter: LocationFreezeFilter, + #[graphql(default = 100)] limit: i32, + ) -> Result> { + let (storage, query) = freeze_query(ctx, filter, limit)?; + + let started = Instant::now(); + let rows = storage + .fetch_location_freeze_summary(&query) + .await + .map_err(|e| format!("failed to fetch location freeze summary: {e}"))?; + + info!( + count = rows.len(), + limit = query.limit, + gap_threshold_ms = query.gap_threshold_ms, + speed_threshold_kmh = query.speed_threshold_kmh, + require_app_alive = query.require_app_alive, + duration_ms = started.elapsed().as_millis(), + "locationFreezeSummary resolver completed" + ); + + Ok(rows.into_iter().map(LocationFreezeSummary::from).collect()) + } +} + +/// Prepares the validated filter shared by the three freeze queries. +/// +/// Applies the same observer-token and storage checks as `history_query`, then +/// the freeze-specific range and threshold rules. +/// +/// # Errors +/// +/// Returns an error when authorization or storage configuration is missing, +/// history queries are disabled, the range is inverted or wider than 90 days, +/// or a threshold is out of range. +fn freeze_query<'a>( + ctx: &'a Context<'_>, + filter: LocationFreezeFilter, + limit: i32, +) -> Result<(&'a Storage, LocationFreezeQuery)> { + let auth = ctx + .data::() + .map_err(|_| "auth context is missing")?; + if !auth.can_read_events { + return Err("unauthorized: a valid observer bearer token is required".into()); + } + + if filter.from >= filter.to { + return Err("from must be earlier than to".into()); + } + + let max_span = ChronoDuration::days(FREEZE_MAX_SPAN_DAYS); + if filter.to - filter.from > max_span { + return Err(format!( + "requested span exceeds maximum for location freeze queries: max {} days", + max_span.num_days() + ) + .into()); + } + + if filter.gap_threshold_ms < FREEZE_MIN_GAP_THRESHOLD_MS { + return Err( + format!("gapThresholdMs must be at least {FREEZE_MIN_GAP_THRESHOLD_MS}").into(), + ); + } + + if !filter.speed_threshold_kmh.is_finite() || filter.speed_threshold_kmh < 0.0 { + return Err("speedThresholdKmh must be a finite value >= 0".into()); + } + + let storage = ctx + .data::() + .map_err(|_| "storage is not configured; DATABASE_URL is required")?; + if !storage.enabled() { + return Err("database-backed storage is disabled; history queries are unavailable".into()); + } + + Ok(( + storage, + LocationFreezeQuery { + from_ts: filter.from.timestamp_millis(), + to_ts: filter.to.timestamp_millis(), + session_id: filter.session_id, + line_id: filter.line_id, + segment_id: filter.segment_id, + device: filter.device, + app_version: filter.app_version, + platform: filter.platform.map(|p| p.as_str().to_string()), + channel: filter.channel.map(|c| c.as_str().to_string()), + gap_threshold_ms: i64::from(filter.gap_threshold_ms), + speed_threshold_kmh: filter.speed_threshold_kmh, + require_app_alive: filter.require_app_alive, + limit: limit.clamp(1, HARD_LIMIT), + }, + )) } /// Prepares the validated filter used by raw history queries. @@ -603,6 +898,12 @@ pub struct LocationEventInput { /// Battery level as a decimal (0.0 to 1.0). pub battery_level: Option, pub battery_state: Option, + /// Application version string (e.g. "1.2.3"), same value as the one sent + /// with `sendLogEvent`. Optional for backwards compatibility with clients + /// that predate it; a blank string is rejected. + pub app_version: Option, + pub platform: Option, + pub channel: Option, } #[derive(SimpleObject)] @@ -885,6 +1186,12 @@ impl MutationRoot { } } + if let Some(version) = &input.app_version { + if version.trim().is_empty() { + return Err("appVersion must not be blank".into()); + } + } + // station_id is only meaningful when not moving/approaching let station_id = if matches!( input.state, @@ -924,6 +1231,9 @@ impl MutationRoot { to_station_id: None, battery_level: input.battery_level, battery_state: input.battery_state, + app_version: input.app_version, + platform: input.platform, + channel: input.channel, }; let loc = segmenter.annotate(loc).await; @@ -1076,11 +1386,109 @@ impl From for LocationEvent { to_station_id: row.to_station_id, battery_level: row.battery_level, battery_state: row.battery_state.and_then(BatteryState::from_i16), + app_version: row.app_version, + platform: row.platform.as_deref().and_then(Platform::parse), + channel: row.channel.as_deref().and_then(Channel::parse), recorded_at: row.recorded_at, } } } +/// Converts a client-reported unix-millisecond timestamp into a `DateTime`, +/// falling back to the epoch for values outside the representable range. +fn millis_to_datetime(millis: i64) -> DateTime { + DateTime::from_timestamp_millis(millis).unwrap_or_else(|| DateTime::from_timestamp_nanos(0)) +} + +/// Converts a non-negative millisecond count into the `u64` the schema exposes. +/// The values come from `MAX`/`SUM` over timestamp differences, so a negative +/// result would mean corrupt data; degrade to zero rather than fail the query. +fn millis_to_u64(millis: i64) -> u64 { + u64::try_from(millis).unwrap_or(0) +} + +impl From for LocationFreeze { + fn from(row: LocationFreezeRow) -> Self { + let jump_distance_meters = haversine_meters( + row.lat_before_gap, + row.lon_before_gap, + row.lat_after_gap, + row.lon_after_gap, + ); + + Self { + session_id: row.session_id, + device: row.device, + line_id: row.line_id, + segment_id: row.segment_id, + from_station_id: row.from_station_id, + to_station_id: row.to_station_id, + app_version: row.app_version, + platform: row.platform.as_deref().and_then(Platform::parse), + channel: row.channel.as_deref().and_then(Channel::parse), + gap_start: millis_to_datetime(row.gap_start), + gap_end: millis_to_datetime(row.gap_end), + gap_ms: millis_to_u64(row.gap_ms), + speed_before_gap: row.speed_before_gap, + coords_before_gap: Coords { + latitude: row.lat_before_gap, + longitude: row.lon_before_gap, + accuracy: row.accuracy_before_gap, + speed: Some(row.speed_before_gap), + }, + coords_after_gap: Coords { + latitude: row.lat_after_gap, + longitude: row.lon_after_gap, + accuracy: row.accuracy_after_gap, + speed: row.speed_after_gap, + }, + jump_distance_meters, + alive_event_count: row.alive_event_count, + } + } +} + +impl From for LocationFreezeSession { + fn from(row: LocationFreezeSessionRow) -> Self { + Self { + session_id: row.session_id, + device: row.device, + line_ids: row.line_ids, + app_version: row.app_version, + platform: row.platform.as_deref().and_then(Platform::parse), + channel: row.channel.as_deref().and_then(Channel::parse), + started_at: millis_to_datetime(row.started_at), + ended_at: millis_to_datetime(row.ended_at), + location_count: row.location_count, + max_speed: row.max_speed, + freeze_count: row.freeze_count, + max_gap_ms: row.max_gap_ms.map(millis_to_u64), + total_gap_ms: millis_to_u64(row.total_gap_ms), + } + } +} + +impl From for LocationFreezeSummary { + fn from(row: LocationFreezeSummaryRow) -> Self { + Self { + line_id: row.line_id, + segment_id: row.segment_id, + from_station_id: row.from_station_id, + to_station_id: row.to_station_id, + device: row.device, + app_version: row.app_version, + platform: row.platform.as_deref().and_then(Platform::parse), + channel: row.channel.as_deref().and_then(Channel::parse), + session_count: row.session_count, + location_count: row.location_count, + freeze_session_count: row.freeze_session_count, + freeze_count: row.freeze_count, + max_gap_ms: row.max_gap_ms.map(millis_to_u64), + total_gap_ms: millis_to_u64(row.total_gap_ms), + } + } +} + /// Calculates the number of time buckets needed to cover a time range. /// /// # Examples @@ -1135,6 +1543,29 @@ mod tests { async_graphql::Request::new(query.to_string()).data(auth) } + /// A syntactically valid freeze query with the supplied extra filter + /// fields, so the tests exercise the resolver rather than schema parsing. + fn freeze_query_str(field: &str, selection: &str, extra: &str) -> String { + format!( + r#"query {{ + {field}(filter: {{ + from: "2026-07-01T00:00:00Z", + to: "2026-07-02T00:00:00Z"{extra} + }}) {{ {selection} }} + }}"# + ) + } + + /// The three freeze queries with a valid filter, for the shared + /// authorization and storage checks. + fn freeze_queries() -> [String; 3] { + [ + freeze_query_str("locationFreezes", "sessionId", ""), + freeze_query_str("locationFreezeSessions", "sessionId", ""), + freeze_query_str("locationFreezeSummary", "device", ""), + ] + } + fn location_mutation(state: &str, extra: &str) -> String { format!( r#"mutation {{ @@ -1682,11 +2113,16 @@ mod tests { let hub = Arc::new(TelemetryHub::new(10)); let schema = test_schema(hub); - for query in [ - r#"query { logEvents { id } }"#, - r#"query { interactionEvents { id } }"#, - r#"query { locations { id } }"#, - ] { + let queries: Vec = [ + r#"query { logEvents { id } }"#.to_string(), + r#"query { interactionEvents { id } }"#.to_string(), + r#"query { locations { id } }"#.to_string(), + ] + .into_iter() + .chain(freeze_queries()) + .collect(); + + for query in &queries { for auth in [EVENTS_ONLY, TELEMETRY] { let resp = schema.execute(request(query, auth)).await; assert!(!resp.errors.is_empty(), "expected rejection for {query}"); @@ -1724,11 +2160,16 @@ mod tests { let hub = Arc::new(TelemetryHub::new(10)); let schema = test_schema(hub); - for query in [ - r#"query { logEvents { id } }"#, - r#"query { interactionEvents { id } }"#, - r#"query { locations { id } }"#, - ] { + let queries: Vec = [ + r#"query { logEvents { id } }"#.to_string(), + r#"query { interactionEvents { id } }"#.to_string(), + r#"query { locations { id } }"#.to_string(), + ] + .into_iter() + .chain(freeze_queries()) + .collect(); + + for query in &queries { let resp = schema.execute(request(query, OBSERVER)).await; assert!(!resp.errors.is_empty(), "expected error for {query}"); assert!( @@ -1746,6 +2187,210 @@ mod tests { assert_eq!(TimeBucketSize::Day.max_duration().num_days(), 365); } + #[tokio::test] + async fn freeze_queries_reject_inverted_time_range() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub); + + let resp = schema + .execute(request( + r#"query { + locationFreezes(filter: { + from: "2026-07-02T00:00:00Z", + to: "2026-07-01T00:00:00Z" + }) { sessionId } + }"#, + OBSERVER, + )) + .await; + + assert!(!resp.errors.is_empty()); + assert!(resp.errors[0] + .message + .contains("from must be earlier than to")); + } + + #[tokio::test] + async fn freeze_queries_reject_span_beyond_ninety_days() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub); + + let resp = schema + .execute(request( + r#"query { + locationFreezeSessions(filter: { + from: "2026-01-01T00:00:00Z", + to: "2026-05-01T00:00:00Z" + }) { sessionId } + }"#, + OBSERVER, + )) + .await; + + assert!(!resp.errors.is_empty()); + assert!( + resp.errors[0].message.contains("90 days"), + "unexpected message: {}", + resp.errors[0].message + ); + } + + #[tokio::test] + async fn freeze_queries_reject_sub_second_gap_threshold() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub); + + let resp = schema + .execute(request( + &freeze_query_str("locationFreezes", "sessionId", ", gapThresholdMs: 999"), + OBSERVER, + )) + .await; + + assert!(!resp.errors.is_empty()); + assert!( + resp.errors[0].message.contains("gapThresholdMs"), + "unexpected message: {}", + resp.errors[0].message + ); + } + + #[tokio::test] + async fn freeze_queries_reject_negative_speed_threshold() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub); + + let resp = schema + .execute(request( + &freeze_query_str( + "locationFreezeSummary", + "device", + ", speedThresholdKmh: -1.0", + ), + OBSERVER, + )) + .await; + + assert!(!resp.errors.is_empty()); + assert!( + resp.errors[0].message.contains("speedThresholdKmh"), + "unexpected message: {}", + resp.errors[0].message + ); + } + + #[tokio::test] + async fn freeze_queries_accept_defaulted_thresholds() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub); + + // storage is disabled in tests, so passing validation surfaces as the + // storage error rather than a threshold complaint + let resp = schema + .execute(request( + &freeze_query_str("locationFreezes", "sessionId gapMs", ""), + OBSERVER, + )) + .await; + + assert!(!resp.errors.is_empty()); + assert!( + resp.errors[0].message.contains("storage is disabled"), + "unexpected message: {}", + resp.errors[0].message + ); + } + + #[tokio::test] + async fn send_location_broadcasts_build_metadata() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub.clone()); + + let resp = schema + .execute(request( + r#"mutation { + sendLocation(input: { + sessionId: "sess-1", + device: "dev", + state: moving, + lineId: 1, + coords: { latitude: 35.6812, longitude: 139.7671, speed: 320.0 }, + timestamp: 1706000000000, + appVersion: "10.4.2(101)", + platform: ios, + channel: canary + }) { sessionId } + }"#, + TELEMETRY, + )) + .await; + + assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors); + let snapshot = hub.snapshot().await; + assert_eq!(snapshot.len(), 1); + let v: serde_json::Value = serde_json::from_str(&snapshot[0]).unwrap(); + assert_eq!(v["app_version"], "10.4.2(101)"); + assert_eq!(v["platform"], "ios"); + assert_eq!(v["channel"], "canary"); + } + + #[tokio::test] + async fn send_location_omits_build_metadata_when_not_sent() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub.clone()); + + let resp = schema + .execute(request(&location_mutation("moving", ""), TELEMETRY)) + .await; + + assert!(resp.errors.is_empty(), "errors: {:?}", resp.errors); + let snapshot = hub.snapshot().await; + assert_eq!(snapshot.len(), 1); + let v: serde_json::Value = serde_json::from_str(&snapshot[0]).unwrap(); + assert!(v["app_version"].is_null()); + assert!(v["platform"].is_null()); + assert!(v["channel"].is_null()); + } + + #[tokio::test] + async fn send_location_rejects_blank_app_version() { + let hub = Arc::new(TelemetryHub::new(10)); + let schema = test_schema(hub.clone()); + + let resp = schema + .execute(request( + r#"mutation { + sendLocation(input: { + sessionId: "sess-1", + device: "dev", + state: moving, + lineId: 1, + coords: { latitude: 35.6812, longitude: 139.7671 }, + timestamp: 1, + appVersion: " " + }) { sessionId } + }"#, + TELEMETRY, + )) + .await; + + assert!(!resp.errors.is_empty()); + assert!(resp.errors[0].message.contains("appVersion")); + assert!(hub.snapshot().await.is_empty()); + } + + #[test] + fn millis_round_trip_through_the_freeze_conversions() { + assert_eq!(millis_to_datetime(0).timestamp_millis(), 0); + assert_eq!( + millis_to_datetime(1_706_000_000_000).timestamp_millis(), + 1_706_000_000_000 + ); + assert_eq!(millis_to_u64(300_000), 300_000); + // corrupt data degrades to zero instead of failing the query + assert_eq!(millis_to_u64(-1), 0); + } + #[test] fn estimate_bucket_count_rounds_up() { let from = Utc::now(); diff --git a/src/main.rs b/src/main.rs index bee70d7..5b7506a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,6 @@ mod config; mod domain; +mod freeze; mod graphql; mod segment; mod server; diff --git a/src/segment.rs b/src/segment.rs index 0307078..6209b68 100644 --- a/src/segment.rs +++ b/src/segment.rs @@ -509,6 +509,9 @@ mod tests { to_station_id: None, battery_level: None, battery_state: None, + app_version: None, + platform: None, + channel: None, }; let second = OutgoingLocation { @@ -549,6 +552,9 @@ mod tests { to_station_id: None, battery_level: None, battery_state: None, + app_version: None, + platform: None, + channel: None, }; let second = OutgoingLocation { @@ -654,6 +660,9 @@ mod tests { to_station_id: None, battery_level: None, battery_state: None, + app_version: None, + platform: None, + channel: None, }; // first annotate stores track @@ -696,6 +705,9 @@ mod tests { to_station_id: None, battery_level: None, battery_state: None, + app_version: None, + platform: None, + channel: None, }; let annotated = estimator.annotate(loc).await; diff --git a/src/storage.rs b/src/storage.rs index de15858..55ee350 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -83,6 +83,11 @@ pub struct LocationEventRow { pub timestamp: i64, pub battery_level: Option, pub battery_state: Option, + /// Build metadata added by THQ#30; NULL on rows written before the + /// column existed or by clients that do not send it yet. + pub app_version: Option, + pub platform: Option, + pub channel: Option, pub recorded_at: sqlx::types::chrono::DateTime, } @@ -117,6 +122,18 @@ impl Storage { self.pool.is_some() } + /// Borrows the configured connection pool for query modules that build + /// their own SQL (see `crate::freeze`). + /// + /// # Errors + /// + /// Returns an error when no database is configured. + pub(crate) fn pool(&self) -> anyhow::Result<&PgPool> { + self.pool + .as_ref() + .ok_or_else(|| anyhow::anyhow!("database is not configured")) + } + /// Initializes the configured database schema and required indexes. /// /// Does nothing when no database pool is configured. Database errors are propagated. @@ -153,6 +170,9 @@ impl Storage { timestamp BIGINT NOT NULL, battery_level DOUBLE PRECISION, battery_state SMALLINT, + app_version TEXT, + platform TEXT, + channel TEXT, recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); "#, @@ -191,6 +211,17 @@ impl Storage { sqlx::query("ALTER TABLE location_logs ADD COLUMN IF NOT EXISTS session_id TEXT;") .execute(pool) .await?; + // THQ#30: build metadata on location rows so freeze detection can group + // by build without joining log_events. + sqlx::query("ALTER TABLE location_logs ADD COLUMN IF NOT EXISTS app_version TEXT;") + .execute(pool) + .await?; + sqlx::query("ALTER TABLE location_logs ADD COLUMN IF NOT EXISTS platform TEXT;") + .execute(pool) + .await?; + sqlx::query("ALTER TABLE location_logs ADD COLUMN IF NOT EXISTS channel TEXT;") + .execute(pool) + .await?; sqlx::query( r#" @@ -315,6 +346,26 @@ impl Storage { .execute(pool) .await?; + // THQ#30: the freeze queries walk each session in timestamp order and + // count the events that fall inside a gap. + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_location_logs_session_timestamp ON location_logs (session_id, timestamp);", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_log_events_session_timestamp ON log_events (session_id, timestamp);", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_interaction_events_session_timestamp ON interaction_events (session_id, timestamp);", + ) + .execute(pool) + .await?; + Ok(()) } @@ -342,7 +393,7 @@ impl Storage { let ts = i64::try_from(loc.timestamp).unwrap_or(i64::MAX); sqlx::query( - "INSERT INTO location_logs (id, session_id, device, state, station_id, line_id, segment_id, from_station_id, to_station_id, latitude, longitude, accuracy, speed, timestamp, battery_level, battery_state) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) ON CONFLICT (id) DO NOTHING", + "INSERT INTO location_logs (id, session_id, device, state, station_id, line_id, segment_id, from_station_id, to_station_id, latitude, longitude, accuracy, speed, timestamp, battery_level, battery_state, app_version, platform, channel) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19) ON CONFLICT (id) DO NOTHING", ) .bind(&loc.id) .bind(&loc.session_id) @@ -360,6 +411,9 @@ impl Storage { .bind(ts) .bind(loc.battery_level) .bind(loc.battery_state.as_ref().map(battery_state_i16)) + .bind(&loc.app_version) + .bind(loc.platform.map(|p| p.as_str())) + .bind(loc.channel.map(|c| c.as_str())) .execute(pool) .await .context("failed to insert location log")?; @@ -696,7 +750,8 @@ impl Storage { SELECT id, session_id, device, state, station_id, line_id, segment_id, from_station_id, to_station_id, latitude, longitude, accuracy, speed, - timestamp, battery_level, battery_state, recorded_at + timestamp, battery_level, battery_state, + app_version, platform, channel, recorded_at FROM location_logs WHERE ($1::text IS NULL OR session_id = $1) AND ($2::text IS NULL OR device = $2)