feat(Session DB): local SQLDelight session storage + upload outbox

docs/API_V3.md

@@ -174,6 +174,39 @@ The two uploads look symmetric and are not. Three traps, in the order they bite:
 Everything else is forgiving: resends are free, so retry on any transport error rather than trying to
 reason about what got through.
 
+### Our outbox — the local `pending_uploads` table
+
+Upload state is **not** stored on `measurements`. It lives in a separate outbox table holding only
+keys — `pending_uploads(stream_id, time)`, composite FK → `measurements` `ON DELETE CASCADE`. Each
+measurement is stored once, in `measurements`; an outbox row is just "this one hasn't reached the
+server yet". Written by `MeasurementRepository.insertRecorded`, read via `UploadOutbox.pending`,
+cleared via `UploadOutbox.acknowledge` (a `DELETE` — the data row is untouched; `measurements` stays
+insert-only, per `DOMAIN_KNOWLEDGE.md` rule #4). The traps above shape how it may be used:
+
+1. **Only data recorded on this phone is enqueued.** Fixed sessions are uploaded by the AirBeam itself
+   (with `session_token`); downloaded, followed or Explore-saved sessions belong to the server (often
+   another user). Those paths call `insertAll`, which never touches the outbox — so the safe default is
+   "not uploaded". Only the recording path calls `insertRecorded`, which stores **and** enqueues in one
+   transaction, enqueuing only rows actually newly stored (an `OR IGNORE`-skipped duplicate was
+   enqueued when first recorded).
+2. **Acknowledged means "accepted in a `200` batch", not "stored".** See trap 1. So it is only truthful
+   if the uploader **never sends a frame the server will silently drop**:
+   - **Split against `finished_at` before sending** (trap 3). Frames after the session's server-side
+     `finished_at` are never stored — don't send them, don't acknowledge them. They also must not sit in
+     the outbox forever. **Open decision:** bound `pending` by `finished_at`, or add a `state` column to
+     `pending_uploads` (e.g. "unuploadable"). Either way the change stays in the outbox; the
+     measurement itself is never deleted.
+3. **Sync the clock before the first batch** (trap 2, empty body → `X-Server-Time`). On mobile one
+   out-of-range timestamp fails the **whole batch** with `400 invalid_epoch`, and a naive "retry the
+   oldest pending rows" loop then resends the same poisoned batch forever. The uploader must detect
+   `invalid_epoch` and handle the bad frames rather than retry unchanged.
+4. **Track per row `(stream_id, time)`, never a time cursor.** Mini V2 reconnect-sync backfills
+   *older* timestamps after newer ones were already uploaded; a per-session "uploaded up to T" cursor
+   would treat them as done and lose them.
+5. **Retry is otherwise blind and safe.** On transport errors, timeouts and `5xx`, leave the outbox
+   rows in place — they go out again next pass. Resends are deduped server-side on
+   `(stream, timestamp)`. State lives in SQLite, so it survives process death.
+
 ### Error codes
 
 All v3 errors share `{ error_code, message, fields? }` (`fields` only when the request shape is wrong).

docs/DOMAIN_KNOWLEDGE.md

@@ -112,13 +112,13 @@ Enums:
 2. **Configure** — user fills details → `StartRecordingEvent`.
 3. **Start** — orchestrator sets status, inserts session in DB, posts `ConfigureSession`;
    for fixed, first creates the backend session (below). Mobile also schedules averaging +
-   starts the measurement observer.
+   starts the measurement observer. *(Averaging step dropped in the rewrite — §7.)*
 4. **Record** — connector writes config to device; measurements flow back to DB + UI.
 5. **Disconnect** — unexpected BLE drop → `SensorDisconnectedUnexpectedlyEvent`; session →
    `DISCONNECTED`; a 2s-interval reconnect loop starts (no hard cap).
 6. **Resume** — reconnect success → `RECORDING`. User can also disconnect deliberately
    (standalone mode) keeping the session.
-7. **Finish** — `StopRecordingEvent` → final averaging, delete active-session rows, status
+7. **Finish** — `StopRecordingEvent` → final averaging *(dropped in the rewrite — §7)*, delete active-session rows, status
    `FINISHED`, backend sync. All services/connectors self-tear-down.
 
 Orchestration classes (all **CHANGE-LIKELY** at the EventBus edges, logic **KEEP**):
@@ -190,7 +190,8 @@ transports (BLE, WiFi-SoftAP, file I/O) are **PLATFORM**; HTTP is **CHANGE-LIKEL
    download, one-shot followed-session download.
 
 **Dedup differs:** V1 mobile filters `time > lastMeasurementTime`; V2 relies on the DB
-unique index `(session_id, stream_id, time)` + `IGNORE`. Preserve the index.
+unique index `(session_id, stream_id, time)` + `IGNORE`. Preserve the dedup — in the rewrite
+the key is `PRIMARY KEY (stream_id, time)` (see rule #5).
 
 ---
 
@@ -219,7 +220,7 @@ graph/map.
 ### Graph — **MPAndroidChart `v3.1.0`** (PLATFORM rendering)
 Mini card chart + full-screen detail graph with pinch-zoom/pan, colored **target zones**
 (threshold bands), note markers, midnight limit lines. Data shaping (`GraphDataGenerator`,
-`ChartData`, averaging, date formatting, note hit-testing) is **KEEP**; the `LineChart`/
+`ChartData`, averaging *(read-side only — §7)*, date formatting, note hit-testing) is **KEEP**; the `LineChart`/
 `CombinedChart` rendering + gestures are **PLATFORM**.
 
 ### Map — **Google Maps** + **Places** (PLATFORM rendering)
@@ -328,9 +329,36 @@ v37_38 (`tracked_locations`). Missing 29_30 & 32_33 hops rely on destructive fal
 Sessions, Measurements, MeasurementStreams, ActiveSessionMeasurements, Thresholds, Notes.
 `ExpandedCardsRepository` is SharedPreferences-backed (not DB) → multiplatform-settings.
 
-### Averaging — **KEEP** (algorithm), **PLATFORM** (scheduling/DAO)
+### Averaging — ⛔ **DROPPED as storage mutation; display-only in the rewrite**
+
+> **Decision (2026-10-15).** The rewrite **never averages, down-samples, or deletes stored
+> measurements.** Every measurement received from the device (live or sync) or downloaded is
+> kept at native resolution, exactly as recorded. Averaging exists **only on the read side**:
+> the UI (graph, map, heatmap, stats) buckets raw rows into windows when it renders.
+>
+> Why: the legacy in-place averaging was lossy and irreversible, needed scheduling during
+> recording, and caused real data loss (the sparse-interval bug that wiped whole sessions —
+> V2 guide, "Sparse-interval averaging gotcha"). Storage is cheap (1s × 5 streams × 24h ≈
+> 432k rows/day — trivial for SQLite); lost data is not.
+>
+> Consequences for the rewrite:
+> - No `AveragingService`, no averaging scheduler, no `averaging_frequency` column — the
+>   recording engine just saves.
+> - The **native-interval gate (old rule #4) is moot** — there's nothing destructive to guard.
+>   `measurement_interval` is still stored on the session: display bucketing uses it to avoid
+>   choosing a window finer than the native rate, and to tell gaps from sparse sampling.
+> - Read-side averaging must be **cheap on large streams** — bucket in SQL
+>   (`GROUP BY time / window`) or over a time-bounded query, not by loading a 10h session
+>   into memory per frame.
+> - **Uploads send raw measurements.** The backend gets full resolution (the legacy app could
+>   upload already-averaged rows for long sessions). Expect more v3 upload batches
+>   (mobile cap: 3000 frames/batch).
+>
+> The window values below remain a reasonable *display* default; the mutation described
+> after them is legacy behavior, kept here for reference only.
+
 `AveragingWindow`: `ZERO(1s)`, `FIRST(5s)`, `SECOND(60s)`. `TimeThreshold`: FIRST=2h, SECOND=9h.
-Rule: session > 2h → 5s averages; > 9h → 60s averages. Down-samples in place (mutates
+*Legacy:* session > 2h → 5s averages; > 9h → 60s averages. Down-samples in place (mutates
 `measurements` rows, sets `averaging_frequency`, deletes the rest). **Native-interval gate:**
 if `session.measurementInterval >= window.value`, averaging is skipped (prevents wiping
 sparse V2 data — the reason the v37 column exists). See V2 guide §5D.
@@ -401,8 +429,13 @@ Mini V2 mobile. Level banding lives in `Measurement.getLevel`.
    binary → epoch. Getting this wrong offsets every graph. (V2 guide §8a.)
 2. **HTTPS force-upgrade + port 80→443** in base-URL construction.
 3. **V2 sync UUID validation** — discard sync chunks whose device session UUID ≠ app session.
-4. **Averaging native-interval gate** — skip when `measurementInterval >= window`.
-5. **Measurements unique index** `(session_id, stream_id, time)` — the V2 dedup strategy.
+4. **Raw measurements are never averaged, down-sampled, or deleted in storage** — averaging is
+   display-only (§7). *Supersedes the legacy "averaging native-interval gate" rule, which only
+   existed to limit damage from in-place averaging.*
+5. **Measurements dedup on `(stream, time)`** — the V2 dedup strategy, and the backend's own v3
+   rule. Legacy index was `(session_id, stream_id, time)`; the rewrite uses
+   `PRIMARY KEY (stream_id, time)` + `INSERT OR IGNORE` (`session_id` was redundant). There is
+   no separate active-session buffer table — one `measurements` table, one write path.
 6. **Backend URL/port preserved on logout.**
 7. **Two measurement "drain" flags for V2 finish-while-syncing** (V2 guide §9).
 8. **V2→V1 fallback disconnect-path guards** — keep the service/EventBus alive across the

docs/KMP_ARCHITECTURE.md

@@ -41,7 +41,7 @@
 :core            kotlinx-datetime, coroutines setup, Result types, logging, expect utils
 :domain          pure models + enums + state machines + business rules   (NO platform deps)
                  └ Session/Measurement/Stream/Threshold/Note, Level banding,
-                   averaging algorithm, recording state machine, threshold color levels
+                   display averaging (read-side bucketing), recording state machine, threshold color levels
 :data:db         SQLDelight schema (.sq) + queries + column adapters      (KEEP + PLATFORM driver)
 :data:network    Ktor client + kotlinx.serialization DTOs + API service   (KEEP + PLATFORM engine)
 :data:settings   multiplatform-settings wrapper (auth token, prefs)       (PLATFORM secure storage)
@@ -154,7 +154,6 @@ class RecordingEngine(
     private val driver: SensorDriver,
     private val sessions: SessionRepository,
     private val measurements: MeasurementRepository,
-    private val averaging: AveragingService,
     private val scope: CoroutineScope,
 ) {
     val state: StateFlow<RecordingState>            // NEW→RECORDING→DISCONNECTED→FINISHED
@@ -168,8 +167,10 @@ class RecordingEngine(
   flow through one `saveMeasurements()` path with the dedup rule (`INSERT … IGNORE` on the
   unique index). No more V1-vs-V2 fork.
 - **State machine** is `Session.Status` transitions, pure `commonMain`.
-- **Averaging** (`AveragingService` + windows) ports as-is — pure algorithm; inject a
-  dispatcher instead of `Dispatchers.Default`; keep the native-interval gate.
+- **No averaging in the engine.** The legacy `AveragingService` (in-place down-sampling of
+  stored rows) is **not ported**. Stored measurements stay at native resolution forever;
+  averaging is a read-side, display-only concern (graph/map/stats bucket raw rows when
+  rendering). See `DOMAIN_KNOWLEDGE.md` §7 and load-bearing rule #4.
 - **Background execution** is the one platform escape hatch: recording must survive
   backgrounding. `expect` a `BackgroundSessionHost` — Android = foreground service
   (`FOREGROUND_SERVICE_TYPE_CONNECTED_DEVICE`), iOS = Core Bluetooth background mode +
@@ -183,9 +184,17 @@ class RecordingEngine(
 ## 5. Data Layer
 
 ### Persistence — SQLDelight (`:data:db`)
-- Port the Room schema (v38) to `.sq` files. Preserve: the unique measurements index
-  (`session_id, stream_id, time`), `measurement_interval`, `session_token` + fixed indices,
-  the active-measurements rolling window (>998-row guard → `transaction {}`).
+- **Raw measurements are immutable** — never averaged/down-sampled/deleted by the app
+  (only by session deletion cascade). No `averaging_frequency` column.
+- Port the Room schema (v38) to `.sq` files — as a redesign, not a copy. Preserve the
+  measurements dedup key, now `PRIMARY KEY (stream_id, time)` on `measurements` (the legacy
+  `session_id` in that index was redundant — a stream belongs to one session — and v3 dedups
+  on `(stream, timestamp)` too), plus `measurement_interval` and `session_token`.
+- **No `active_sessions_measurements` table.** The legacy rolling live buffer (10h of rows,
+  >998-row guard) existed because Room couldn't cheaply tail a growing table. The live view is
+  now a windowed query on `measurements` (`ORDER BY time DESC LIMIT ?`, an index walk on the
+  PK), so there is one table and one write path. Batch inserts still go in one
+  `transaction {}`.
 - Column adapters for `Date`→`Instant`, tags list, `Session.Status/Type`, `DeviceType`.
 - `expect` the driver: `AndroidSqliteDriver` / `NativeSqliteDriver`.
 - DAOs → SQLDelight queries returning `Flow` (`.asFlow().mapToList()`). The
@@ -218,7 +227,7 @@ plaintext today. Preserve backend URL/port on logout.
 ### Repositories (`:data:repo`)
 Interfaces in `:domain`, impls here. Extract interfaces for all current repos (only
 `MeasurementsRepository` is one today). Business rules (`getLevel`, threshold find-or-create,
-rolling-window sizing, status transitions) move into `:domain`.
+status transitions) move into `:domain`.
 
 ---
 
@@ -386,7 +395,7 @@ These change the skeleton — resolve early:
 
 Vertical slices, each provable end-to-end, lowest-risk first:
 
-1. **Foundation:** `:core` + `:domain` (models, enums, state machine, averaging, thresholds)
+1. **Foundation:** `:core` + `:domain` (models, enums, state machine, display averaging, thresholds)
    with unit tests. No platform code — proves the shared logic compiles KMP.
 2. **Persistence:** `:data:db` (SQLDelight schema + queries) on both platforms.
 3. **Networking:** `:data:network` (Ktor + DTOs) + **timestamp round-trip tests** + auth/login.

docs/ble_mobile_app_guide.md

@@ -276,6 +276,10 @@ Defaults `DEFAULT_MOBILE_INTERVAL_SECONDS = 1` / `DEFAULT_FIXED_INTERVAL_SECONDS
 `AirBeamMiniV2Configurator`'s companion object and apply only when the param is null
 (legacy/non-V2 paths).
 
+> **Legacy only.** The KMP rewrite does not average stored measurements at all — averaging
+> is display-only (`DOMAIN_KNOWLEDGE.md` §7), so this bug class cannot occur. Kept as the
+> motivating example for that decision.
+
 **Sparse-interval averaging gotcha (`AveragingService`).** Averaging assumes the native
 sample rate is finer-grained than the averaging window. When that is violated (a
 5-second-native session in the 5s FIRST window, or any session with native ≥ window),

shared/build.gradle.kts

@@ -85,6 +85,9 @@ kotlin {
       implementation(libs.turbine)
       implementation(libs.multiplatform.settings.test)
     }
+    getByName("androidHostTest").dependencies {
+      implementation(libs.sqldelight.sqlite.driver)
+    }
   }
 }
 

shared/src/androidHostTest/kotlin/pl/llp/aircasting/data/local/MeasurementRepositoryTest.kt

@@ -0,0 +1,136 @@
+package pl.llp.aircasting.data.local
+
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.test.runTest
+import pl.llp.aircasting.domain.GeoLocation
+import pl.llp.aircasting.domain.session.Measurement
+import pl.llp.aircasting.domain.session.Session
+import pl.llp.aircasting.domain.session.SessionStatus
+import pl.llp.aircasting.domain.session.SessionType
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.Instant
+
+class MeasurementRepositoryTest {
+
+  private val db = testDatabase()
+  private val sessions = LocalSessionRepository(db.sessionQueries)
+  private val streams = LocalMeasurementStreamRepository(db.measurementStreamQueries)
+  private val measurements = LocalMeasurementRepository(db.measurementQueries, db.pendingUploadQueries)
+
+  private val uuid = "22222222-2222-2222-2222-222222222222"
+  private val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000)
+
+  private suspend fun seedSession() = sessions.insert(
+    Session(
+      uuid = uuid,
+      type = SessionType.MOBILE,
+      status = SessionStatus.RECORDING,
+      name = "Test",
+      tags = emptyList(),
+      startTime = t0,
+      timeZone = "UTC",
+      measurementInterval = 1.seconds,
+    )
+  )
+
+  @Test
+  fun `findOrCreate returns the same stream id for the same sensor`() = runTest {
+    seedSession()
+    val first = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    val second = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    assertEquals(first, second)
+    assertEquals(1, streams.findBySession(uuid).size)
+  }
+
+  @Test
+  fun `insertAll ignores a measurement that already exists for stream and time`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+
+    assertEquals(1, measurements.insertAll(listOf(Measurement(stream, t0, value = 12.0))))
+    assertEquals(0, measurements.insertAll(listOf(Measurement(stream, t0, value = 99.0))))
+
+    val stored = measurements.observeRecent(stream, limit = 10).first()
+    assertEquals(1, stored.size)
+    assertEquals(12.0, stored.single().value)
+  }
+
+  @Test
+  fun `insertAll counts only newly stored rows in a partially overlapping batch`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    measurements.insertAll(listOf(Measurement(stream, t0, 1.0)))
+
+    val stored = measurements.insertAll(
+      listOf(
+        Measurement(stream, t0, 1.0),
+        Measurement(stream, t0 + 1.seconds, 2.0),
+        Measurement(stream, t0 + 2.seconds, 3.0),
+      )
+    )
+
+    assertEquals(2, stored)
+  }
+
+  @Test
+  fun `two streams may hold the same timestamp`() = runTest {
+    seedSession()
+    val pm = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    val rh = streams.findOrCreate(uuid, "AirBeam-RH", "%")
+
+    measurements.insertAll(
+      listOf(Measurement(pm, t0, 12.0), Measurement(rh, t0, 45.0))
+    )
+
+    assertEquals(1, measurements.observeRecent(pm, 10).first().size)
+    assertEquals(1, measurements.observeRecent(rh, 10).first().size)
+  }
+
+  @Test
+  fun `a measurement round-trips its location`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    val here = GeoLocation(50.06, 19.94)
+
+    measurements.insertAll(listOf(Measurement(stream, t0, 12.0, location = here)))
+
+    assertEquals(here, measurements.observeRecent(stream, 10).first().single().location)
+  }
+
+  @Test
+  fun `lastTime returns the newest timestamp for the stream`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    measurements.insertAll(
+      listOf(
+        Measurement(stream, t0, 1.0),
+        Measurement(stream, t0 + 2.seconds, 2.0),
+        Measurement(stream, t0 + 1.seconds, 3.0),
+      )
+    )
+    assertEquals(t0 + 2.seconds, measurements.lastTime(stream))
+  }
+
+  @Test
+  fun `lastTime is null for an empty stream`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    assertNull(measurements.lastTime(stream))
+  }
+
+  @Test
+  fun `deleting a session removes its streams and measurements`() = runTest {
+    seedSession()
+    val stream = streams.findOrCreate(uuid, "AirBeam-PM2.5", "µg/m³")
+    measurements.insertAll(listOf(Measurement(stream, t0, 12.0)))
+
+    sessions.delete(uuid)
+
+    assertTrue(streams.findBySession(uuid).isEmpty())
+    assertTrue(measurements.observeRecent(stream, 10).first().isEmpty())
+  }
+}