CR: feat/watch-movement-on-pod

app/api/biometrics/watch-motion/[id]/route.ts

@@ -0,0 +1,14 @@
+import { biometricsDb } from '@/src/db'
+import { getWatchMotionFile } from '@/src/biometrics/references/watch-motion'
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+/** The stored, trimmed canonical file: byte-for-byte what training imports. */
+export async function GET(_request: Request, { params }: { params: Promise<{ id: string }> }): Promise<Response> {
+  const { id } = await params
+  if (!/^[1-9]\d{0,15}$/.test(id)) return Response.json({ error: 'Invalid id' }, { status: 400 })
+  const file = getWatchMotionFile(biometricsDb, Number(id))
+  if (file === null) return Response.json({ error: 'Not found' }, { status: 404 })
+  return new Response(file, { headers: { 'Content-Type': 'application/json' } })
+}

app/api/biometrics/watch-motion/route.test.ts

@@ -0,0 +1,168 @@
+// @vitest-environment node
+import { createHash } from 'node:crypto'
+import { beforeEach, expect, it, vi } from 'vitest'
+
+const state = vi.hoisted(() => ({ db: null as unknown, raw: null as { exec: (sql: string) => void } | null }))
+vi.mock('@/src/db', async () => {
+  const Database = (await import('better-sqlite3')).default
+  const { drizzle } = await import('drizzle-orm/better-sqlite3')
+  const { migrate } = await import('drizzle-orm/better-sqlite3/migrator')
+  const schema = await import('@/src/db/biometrics-schema')
+  const raw = new Database(':memory:')
+  const db = drizzle(raw, { schema })
+  migrate(db, { migrationsFolder: 'src/db/biometrics-migrations' })
+  state.db = db
+  state.raw = raw
+  return { biometricsDb: db }
+})
+
+const { POST, GET } = await import('./route')
+const { GET: GET_ONE } = await import('./[id]/route')
+
+const FROM = 1_793_618_344_598
+const TO = FROM + 8 * 3600_000
+const FIRST_BIN = Math.floor(FROM / 2000) * 2000 // overlaps the start edge, so it is kept
+
+function file(version: 'watch-motion-v1' | 'watch-motion-v2', starts: number[], extra: Record<string, unknown> = {}) {
+  const axes = version === 'watch-motion-v2' ? { meanX: starts.map(() => 0.1), meanY: starts.map(() => -0.2), meanZ: starts.map(() => 0.97) } : {}
+  return {
+    version, recordedFromMs: FROM, recordedToMs: TO, timeZone: 'America/Los_Angeles', utcOffsetSeconds: -28800,
+    sampleRateHz: 50, binSeconds: 2, jerkThresholdGPerSecond: 1, device: 'Apple Watch · watchOS 26.6',
+    bins: { startMs: starts, samples: starts.map(() => 100), meanMagnitude: starts.map(() => 1), magnitudeVariance: starts.map((_, i) => i * 1e-6),
+      maxJerk: starts.map((_, i) => i % 7), movingSamples: starts.map((_, i) => i % 3), ...axes },
+    ...extra,
+  }
+}
+const night = Array.from({ length: 90 }, (_, i) => FIRST_BIN + i * 2000)
+/** The same file with a bin prepended to every column, as the Watch recorder's leftovers. */
+function withStray(f: ReturnType<typeof file>, startMs: number) {
+  return { ...f, bins: Object.fromEntries(Object.entries(f.bins).map(([column, values]) => [column, [column === 'startMs' ? startMs : 1, ...values]])) }
+}
+
+const post = (body: unknown, side = 'left', headers: Record<string, string> = {}) => POST(new Request(`http://pod/api/biometrics/watch-motion?side=${side}`, {
+  method: 'POST', headers: { 'content-type': 'application/json', ...headers }, body: typeof body === 'string' ? body : JSON.stringify(body),
+}))
+
+beforeEach(() => state.raw?.exec('DELETE FROM watch_motion_captures'))
+
+it('accepts a v1 file and serves the stored canonical file back byte-for-byte', async () => {
+  const response = await post(file('watch-motion-v1', night))
+  expect(response.status).toBe(200)
+  const body = await response.json()
+  expect(body).toEqual({ id: expect.any(Number), side: 'left', version: 'watch-motion-v1', recordedFromMs: FROM, recordedToMs: TO, bins: 90, trimmedBins: 0, status: 'created' })
+  const stored = await GET_ONE(new Request('http://pod'), { params: Promise.resolve({ id: String(body.id) }) })
+  expect(stored.headers.get('content-type')).toBe('application/json')
+  const text = await stored.text()
+  expect(JSON.parse(text)).toEqual(file('watch-motion-v1', night))
+  // Keys sorted at every level, like the iOS encoder.
+  expect(text.indexOf('"binSeconds"')).toBeLessThan(text.indexOf('"bins"'))
+  expect(text.indexOf('"magnitudeVariance"')).toBeLessThan(text.indexOf('"startMs"'))
+  const [listed] = await (await GET(new Request('http://pod/api/biometrics/watch-motion?side=left'))).json()
+  expect(listed).toMatchObject({ id: body.id, sha256: createHash('sha256').update(text).digest('hex'), bins: 90, trimmedBins: 0 })
+})
+
+it('accepts a v2 file with axis means', async () => {
+  const response = await post(file('watch-motion-v2', night), 'right')
+  expect(response.status).toBe(200)
+  const body = await response.json()
+  expect(body).toMatchObject({ side: 'right', version: 'watch-motion-v2', bins: 90, status: 'created' })
+  const stored = JSON.parse(await (await GET_ONE(new Request('http://pod'), { params: Promise.resolve({ id: String(body.id) }) })).text())
+  expect(stored.bins.meanZ).toHaveLength(90)
+})
+
+it('trims a stray bin from an earlier recording and counts it', async () => {
+  const stray = FROM - 20 * 3600_000 - (FROM - 20 * 3600_000) % 2000
+  const after = Math.ceil(TO / 2000) * 2000
+  const response = await post(file('watch-motion-v2', [stray, stray + 2000, ...night, after]))
+  expect(await response.json()).toMatchObject({ bins: 90, trimmedBins: 3, status: 'created' })
+  const [listed] = await (await GET(new Request(`http://pod/api/biometrics/watch-motion?startDate=${FROM}&endDate=${new Date(TO).toISOString()}`))).json()
+  expect(listed).toMatchObject({ bins: 90, trimmedBins: 3, recordedFromMs: FROM })
+  const stored = JSON.parse(await (await GET_ONE(new Request('http://pod'), { params: Promise.resolve({ id: String(listed.id) }) })).text())
+  expect(stored.bins.startMs[0]).toBe(FIRST_BIN)
+  expect(stored.bins.meanX).toHaveLength(90)
+  // Each kept bin keeps its own values.
+  expect(stored.bins.maxJerk).toEqual(night.map((_, i) => (i + 2) % 7))
+})
+
+it('reports unchanged for identical content and replaced for new content of the same night', async () => {
+  const first = await (await post(file('watch-motion-v1', night))).json()
+  // A fixed app resending without the stray bin stores the same trimmed file.
+  const again = await (await post(withStray(file('watch-motion-v1', night), FIRST_BIN - 40 * 3600_000))).json()
+  expect(again).toMatchObject({ id: first.id, status: 'unchanged', trimmedBins: 1 })
+  const replaced = await (await post(file('watch-motion-v2', night))).json()
+  expect(replaced).toMatchObject({ id: first.id, status: 'replaced', version: 'watch-motion-v2' })
+  expect(await (await GET(new Request('http://pod/api/biometrics/watch-motion'))).json()).toHaveLength(1)
+  // The other side is a separate capture.
+  expect(await (await post(file('watch-motion-v1', night), 'right')).json()).toMatchObject({ status: 'created' })
+})
+
+it('rejects raw readings, oversize bodies and invalid files', async () => {
+  const rawResponse = await post(file('watch-motion-v1', night, { raw: [[FROM, 0, 0, 1]] }))
+  expect(rawResponse.status).toBe(400)
+  expect(await rawResponse.json()).toEqual({ error: 'raw readings are not accepted' })
+
+  expect((await post('x', 'left', { 'content-length': String(8 * 1024 * 1024 + 1) })).status).toBe(413)
+  const chunk = new Uint8Array(1)
+  Object.defineProperty(chunk, 'byteLength', { value: 8 * 1024 * 1024 + 1 })
+  const stream = new ReadableStream({ start(controller) {
+    controller.enqueue(chunk)
+    controller.close()
+  } })
+  const streamed = await POST(new Request('http://pod/api/biometrics/watch-motion?side=left', { method: 'POST', body: stream, duplex: 'half' } as RequestInit))
+  expect(streamed.status).toBe(413)
+
+  const mismatched = file('watch-motion-v1', night)
+  mismatched.bins.maxJerk.pop()
+  const cases: Array<[unknown, RegExp]> = [
+    [mismatched, /bins\.maxJerk differ/],
+    [file('watch-motion-v1', night, { version: 'watch-motion-v3' }), /version/],
+    [file('watch-motion-v1', night, { binSeconds: 1 }), /binSeconds/],
+    [file('watch-motion-v2', night, { bins: { ...file('watch-motion-v1', night).bins } }), /v2 requires/],
+    [file('watch-motion-v1', night, { bins: file('watch-motion-v2', night).bins }), /v1 must not/],
+    [file('watch-motion-v1', [FIRST_BIN, FIRST_BIN + 1000]), /multiple of 2000/],
+    [file('watch-motion-v1', [FIRST_BIN + 2000, FIRST_BIN]), /strictly increasing/],
+    [file('watch-motion-v1', [FIRST_BIN, FIRST_BIN + 2000], { bins: { ...file('watch-motion-v1', [FIRST_BIN, FIRST_BIN + 2000]).bins, movingSamples: [1, -1] } }), /movingSamples/],
+    [file('watch-motion-v1', [FIRST_BIN, FIRST_BIN + 2000], { bins: { ...file('watch-motion-v1', [FIRST_BIN, FIRST_BIN + 2000]).bins, samples: [1.5, 2] } }), /samples/],
+    [file('watch-motion-v1', [FIRST_BIN - 3600_000]), /No bins fall inside/],
+    [file('watch-motion-v1', night, { recordedToMs: FROM }), /recordedToMs must be after/],
+    [file('watch-motion-v1', night, { extra: 1 }), /extra/],
+  ]
+  for (const [body, message] of cases) {
+    const response = await post(body)
+    expect(response.status, String(message)).toBe(400)
+    expect((await response.json()).error).toMatch(message)
+  }
+  // JSON.parse turns 1e999 into Infinity.
+  const infinite = JSON.stringify(file('watch-motion-v1', [FIRST_BIN])).replace('"maxJerk":[0]', '"maxJerk":[1e999]')
+  expect((await post(infinite)).status).toBe(400)
+  expect((await post('{not json')).status).toBe(400)
+  expect((await post('')).status).toBe(400)
+  expect((await POST(new Request('http://pod/api/biometrics/watch-motion', { method: 'POST', body: '{}' }))).status).toBe(400)
+  expect(await (await GET(new Request('http://pod/api/biometrics/watch-motion'))).json()).toEqual([])
+})
+
+it('rejects cross-origin and DNS-rebinding uploads but accepts the same-origin web app', async () => {
+  for (const origin of ['http://evil', 'http://192.168.1.88:8080', 'null']) {
+    const response = await post(file('watch-motion-v1', night), 'left', { origin, host: '192.168.1.88:3000' })
+    expect(response.status).toBe(403)
+  }
+  const rebound = await post(file('watch-motion-v1', night), 'left', { origin: 'http://attacker.example.com:3000', host: 'attacker.example.com:3000' })
+  expect(rebound.status).toBe(403)
+  expect(await (await GET(new Request('http://pod/api/biometrics/watch-motion'))).json()).toEqual([])
+  const same = await post(file('watch-motion-v1', night), 'left', { origin: 'http://192.168.1.88:3000', host: '192.168.1.88:3000' })
+  expect(same.status).toBe(200)
+})
+
+it('lists by overlap and validates list and fetch parameters', async () => {
+  await post(file('watch-motion-v1', night))
+  const list = (query: string) => GET(new Request(`http://pod/api/biometrics/watch-motion?${query}`))
+  expect(await (await list(`startDate=${TO}`)).json()).toEqual([])
+  expect(await (await list(`endDate=${FROM - 1}`)).json()).toEqual([])
+  expect(await (await list(`startDate=${TO - 1}&endDate=${TO + 1}&side=left`)).json()).toHaveLength(1)
+  expect(await (await list('side=right')).json()).toEqual([])
+  expect((await list('side=middle')).status).toBe(400)
+  expect((await list('startDate=yesterday')).status).toBe(400)
+  expect((await list(`startDate=${TO}&endDate=${FROM}`)).status).toBe(400)
+  expect((await GET_ONE(new Request('http://pod'), { params: Promise.resolve({ id: '999' }) })).status).toBe(404)
+  expect((await GET_ONE(new Request('http://pod'), { params: Promise.resolve({ id: '../x' }) })).status).toBe(400)
+})

app/api/biometrics/watch-motion/route.ts

@@ -0,0 +1,103 @@
+import { z } from 'zod'
+import { biometricsDb } from '@/src/db'
+import { rejectUnknownHost } from '@/src/lib/hostAllowlist'
+import {
+  listWatchMotion, parseWatchMotion, saveWatchMotion, WATCH_MOTION_MAX_BYTES, WatchMotionError,
+} from '@/src/biometrics/references/watch-motion'
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+
+const sideQuery = z.object({ side: z.enum(['left', 'right']) })
+/** ISO 8601 text or Unix milliseconds. */
+const dateParam = z.string().transform((value, ctx) => {
+  const date = new Date(/^\d+$/.test(value) ? Number(value) : value)
+  if (Number.isNaN(date.getTime())) {
+    ctx.addIssue({ code: 'custom', message: 'must be ISO 8601 or Unix milliseconds' })
+    return z.NEVER
+  }
+  return date
+})
+const listQuery = z.object({ side: z.enum(['left', 'right']).optional(), startDate: dateParam.optional(), endDate: dateParam.optional() }).strict()
+
+/** Same check as watch-upload: Next reports the 0.0.0.0 bind address in request.url, so compare Origin with Host. */
+function isSameOrigin(request: Request): boolean {
+  const origin = request.headers.get('origin')
+  if (origin === null) return true
+  try {
+    return new URL(origin).host === request.headers.get('host')
+  }
+  catch {
+    return false
+  }
+}
+
+class TooLarge extends Error {}
+
+/** The body as text, refusing to buffer more than the limit even without a Content-Length. */
+async function readBody(request: Request): Promise<string> {
+  if (Number(request.headers.get('content-length')) > WATCH_MOTION_MAX_BYTES) throw new TooLarge()
+  if (!request.body) return ''
+  const chunks: Uint8Array[] = []
+  let bytes = 0
+  const reader = request.body.getReader()
+  for (;;) {
+    const { done, value } = await reader.read()
+    if (done) break
+    bytes += value.byteLength
+    if (bytes > WATCH_MOTION_MAX_BYTES) {
+      await reader.cancel()
+      throw new TooLarge()
+    }
+    chunks.push(value)
+  }
+  return Buffer.concat(chunks).toString('utf8')
+}
+
+/**
+ * One night's watch-motion-v1/v2 file from the iOS app, as-is. Stored as a
+ * movement label only (see docs/biometrics-reference-imports.md).
+ */
+export async function POST(request: Request): Promise<Response> {
+  const rejectedHost = rejectUnknownHost(request)
+  if (rejectedHost) return rejectedHost
+  if (!isSameOrigin(request)) return Response.json({ error: 'Cross-origin upload is not allowed' }, { status: 403 })
+  const query = sideQuery.safeParse(Object.fromEntries(new URL(request.url).searchParams))
+  if (!query.success) return Response.json({ error: 'Specify side=left|right' }, { status: 400 })
+  let body: string
+  try {
+    body = await readBody(request)
+  }
+  catch (error) {
+    if (error instanceof TooLarge) return Response.json({ error: 'Upload exceeds 8 MiB' }, { status: 413 })
+    return Response.json({ error: 'Upload interrupted' }, { status: 400 })
+  }
+  if (!body) return Response.json({ error: 'Send the watch-motion JSON file as the request body' }, { status: 400 })
+  try {
+    const parsed = parseWatchMotion(body)
+    const { capture, status } = saveWatchMotion(biometricsDb, query.data.side, parsed)
+    return Response.json({
+      id: capture.id,
+      side: capture.side,
+      version: capture.version,
+      recordedFromMs: capture.recordedFromMs,
+      recordedToMs: capture.recordedToMs,
+      bins: capture.bins,
+      trimmedBins: parsed.trimmedBins,
+      status,
+    })
+  }
+  catch (error) {
+    if (error instanceof WatchMotionError) return Response.json({ error: error.message }, { status: error.status })
+    return Response.json({ error: 'Could not store the recording' }, { status: 500 })
+  }
+}
+
+/** Capture metadata overlapping [startDate, endDate], oldest first. */
+export async function GET(request: Request): Promise<Response> {
+  const query = listQuery.safeParse(Object.fromEntries(new URL(request.url).searchParams))
+  if (!query.success) return Response.json({ error: 'Use side=left|right and startDate/endDate as ISO 8601 or Unix milliseconds' }, { status: 400 })
+  const { side, startDate, endDate } = query.data
+  if (startDate && endDate && startDate > endDate) return Response.json({ error: 'startDate must be before or equal to endDate' }, { status: 400 })
+  return Response.json(listWatchMotion(biometricsDb, { side, start: startDate, end: endDate }))
+}

docs/biometrics-reference-imports.md

@@ -273,3 +273,32 @@ RMSSD read failure is reported while the other readings are sent. A newer app
 and core are needed for RMSSD sync. Neither deployment nor local training
 imports a Health export automatically; the user must initiate/authorize an
 import. No Watch is required for Pod-only charts or model inference.
+
+### Apple Watch wrist movement
+
+`POST /api/biometrics/watch-motion?side=left|right` stores one night's
+`watch-motion-v1` or `watch-motion-v2` file from the iOS app, sent as-is
+(format: sleepypod-ios `docs/watch-motion-format.md`). It is a movement label
+only: it never feeds sleep staging, movement scoring or hardware control.
+Bodies over 8 MiB are rejected (413), and so are files carrying `raw`
+readings (400). `binSeconds` must be 2, every `bins` column must match
+`startMs` in length, `startMs` must lie on the 2 s Unix grid and strictly
+increase, counts must be nonnegative integers, and v2 requires (v1 forbids)
+`meanX`/`meanY`/`meanZ`. Bins outside [`recordedFromMs`, `recordedToMs`) are
+leftovers of an earlier recording and are trimmed, not rejected; a bin
+overlapping the start edge stays. The response reports `trimmedBins`.
+
+One capture is kept per side and `recordedFromMs`. Resending identical
+trimmed content returns `status: "unchanged"`; different content for the same
+night replaces the stored file in place (`"replaced"`, same `id`). The stored
+file is the trimmed file as sorted-key compact JSON; `sha256` is its hash.
+`GET /api/biometrics/watch-motion?side=&startDate=&endDate=` lists capture
+metadata overlapping the window (dates as ISO 8601 or Unix ms), and
+`GET /api/biometrics/watch-motion/{id}` returns the stored file byte for byte
+for the training workspace. `biometrics.getWatchMotionSeries`
+(`GET /api/biometrics/watch-motion-series`) sums the bins per minute
+(`movingSamples`, max `maxJerk`, mean `magnitudeVariance`, `samples`) for at
+most 31 days; minutes without bins are absent rather than zero. The night
+chart draws it over the pod's movement, scaled to each night. Unlike the other
+reference tables, captures (~0.5 MB a night) are pruned with the biometrics
+retention window by `recordedToMs`; the iPhone keeps the original files.

scripts/biometrics-archive/README.md

@@ -119,6 +119,7 @@ Rules (`TABLES` in `merge.ts`):
 | `reference_stages` | side, start, source | state: a corrected import replaces the label |
 | `reference_samples` | side, metric, timestamp, source | state (Watch breathing rate and HRV SDNN) |
 | `reference_vitals` | side, timestamp, source | state |
+| `watch_motion_captures` | side, recorded_from | state: a resent night replaces the file |
 | `reference_imports` | id (text) | state (deleted_at_ms) |
 | `biometric_processing_runs`, `biometric_experiments`, `biometric_backtest_jobs`, `biometric_raw_sources`, `biometric_shadow_runs`, `biometric_dataset_raw_artifacts` | id (text) | state |
 | `biometric_backtest_results` | job_id | state |

scripts/biometrics-archive/merge.ts

@@ -56,6 +56,7 @@ export const TABLES: TableSpec[] = [
   { table: 'reference_stages', key: ['side', 'start', 'source'], mode: 'state' },
   { table: 'reference_samples', key: ['side', 'metric', 'timestamp', 'source'], mode: 'state' },
   { table: 'reference_vitals', key: ['side', 'timestamp', 'source'], mode: 'state' },
+  { table: 'watch_motion_captures', key: ['side', 'recorded_from'], mode: 'state' },
   { table: 'biometric_processing_runs', key: ['id'], mode: 'state' },
   { table: 'biometric_runtime_configs', key: ['side', 'generation'], mode: 'append' },
   { table: 'biometric_runtime_receipts', key: ['side', 'component', 'generation'], mode: 'state' },