release: v1.32.0 — hardening and scalability (#580)

* release: prepare v1.32.0

* test: stabilize water capture re-entry

prisma/migrations/0264_whoop_owner_identity/migration.sql

@@ -0,0 +1,26 @@
+-- WHOOP provider identities are globally owned by one local account.
+-- Existing duplicate identities keep the earliest connection; ties are broken
+-- by the primary key so every database chooses the same winner.
+ALTER TABLE "whoop_connections"
+    ALTER COLUMN "whoop_user_id" DROP NOT NULL;
+
+WITH ranked_connections AS (
+    SELECT
+        "id",
+        ROW_NUMBER() OVER (
+            PARTITION BY "whoop_user_id"
+            ORDER BY "created_at" ASC, "id" ASC
+        ) AS owner_rank
+    FROM "whoop_connections"
+    WHERE "whoop_user_id" IS NOT NULL
+)
+UPDATE "whoop_connections" AS connection
+SET
+    "whoop_user_id" = NULL,
+    "updated_at" = CURRENT_TIMESTAMP
+FROM ranked_connections AS ranked
+WHERE connection."id" = ranked."id"
+  AND ranked.owner_rank > 1;
+
+CREATE UNIQUE INDEX "whoop_connections_whoop_user_id_key"
+    ON "whoop_connections"("whoop_user_id");

prisma/migrations/0265_withings_subscription_state/migration.sql

@@ -0,0 +1,11 @@
+ALTER TABLE "withings_connections"
+  ADD COLUMN "webhook_subscription_state" JSONB,
+  ADD COLUMN "webhook_subscription_retry_at" TIMESTAMP(3);
+
+-- Existing connections have no durable outcome yet. Mark them due so the
+-- hourly repair reconciles each category; status 294 is idempotent success.
+UPDATE "withings_connections"
+SET "webhook_subscription_retry_at" = CURRENT_TIMESTAMP;
+
+CREATE INDEX "withings_connections_webhook_subscription_retry_at_idx"
+  ON "withings_connections"("webhook_subscription_retry_at");

prisma/migrations/0266_medication_intake_import_jobs/migration.sql

@@ -0,0 +1,36 @@
+-- Durable medication-intake imports keep the validated payload and monotonic
+-- progress in Postgres so queue retries and process restarts cannot duplicate
+-- intake or inventory side effects. This is additive and preserves all rows.
+CREATE TABLE "medication_intake_import_jobs" (
+  "id" TEXT NOT NULL,
+  "user_id" TEXT NOT NULL,
+  "medication_id" TEXT NOT NULL,
+  "pg_boss_job_id" TEXT,
+  "status" TEXT NOT NULL DEFAULT 'queued',
+  "payload" JSONB NOT NULL,
+  "progress" JSONB NOT NULL DEFAULT '{}',
+  "result" JSONB,
+  "failure_reason" TEXT,
+  "heartbeat_at" TIMESTAMP(3),
+  "started_at" TIMESTAMP(3),
+  "completed_at" TIMESTAMP(3),
+  "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+  CONSTRAINT "medication_intake_import_jobs_pkey" PRIMARY KEY ("id"),
+  CONSTRAINT "medication_intake_import_jobs_status_check"
+    CHECK ("status" IN ('queued', 'running', 'done', 'failed')),
+  CONSTRAINT "medication_intake_import_jobs_user_id_fkey"
+    FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE,
+  CONSTRAINT "medication_intake_import_jobs_medication_id_fkey"
+    FOREIGN KEY ("medication_id") REFERENCES "medications"("id") ON DELETE CASCADE ON UPDATE CASCADE
+);
+
+CREATE UNIQUE INDEX "medication_intake_import_jobs_pg_boss_job_id_key"
+  ON "medication_intake_import_jobs"("pg_boss_job_id");
+CREATE INDEX "medication_intake_import_jobs_user_id_created_at_idx"
+  ON "medication_intake_import_jobs"("user_id", "created_at");
+CREATE INDEX "medication_intake_import_jobs_medication_id_created_at_idx"
+  ON "medication_intake_import_jobs"("medication_id", "created_at");
+CREATE INDEX "medication_intake_import_jobs_status_heartbeat_at_idx"
+  ON "medication_intake_import_jobs"("status", "heartbeat_at");

prisma/schema.prisma

@@ -737,6 +737,7 @@ model User {
   // pg-boss job state so the polling endpoint can read live per-
   // phase progress without round-tripping through pg-boss internals.
   importJobs                  ImportJob[]
+  medicationIntakeImportJobs MedicationIntakeImportJob[]
   // v1.5.0 — persistent measurement rollup table. Carries the
   // count/min/max/mean/sd/slope/r2 stats per (type, granularity,
   // bucketStart) so the analytics + comprehensive read paths bypass
@@ -2419,6 +2420,7 @@ model Medication {
 
   schedules         MedicationSchedule[]
   intakeEvents      MedicationIntakeEvent[]
+  intakeImportJobs MedicationIntakeImportJob[]
   phaseConfig       ReminderPhaseConfig?
   telegramMessages  TelegramReminderMessage[]
   doseChanges       MedicationDoseChange[]
@@ -3398,11 +3400,17 @@ model WithingsConnection {
   /// and re-syncs with the corrected END stamp, then sets this so the discovery
   /// query drops the connection. Null = backfill not yet run.
   sleepTimelineBackfillAt DateTime? @map("sleep_timeline_backfill_at")
+  /// v1.32 — durable per-category webhook subscription outcomes. The JSON
+  /// payload is versioned and validated before use; the timestamp mirrors the
+  /// earliest transient/pending category retry for indexed hourly discovery.
+  webhookSubscriptionState   Json?     @map("webhook_subscription_state")
+  webhookSubscriptionRetryAt DateTime? @map("webhook_subscription_retry_at")
   createdAt               DateTime  @default(now()) @map("created_at")
   updatedAt               DateTime  @updatedAt @map("updated_at")
 
   user User @relation(fields: [userId], references: [id], onDelete: Cascade)
 
+  @@index([webhookSubscriptionRetryAt])
   @@map("withings_connections")
 }
 
@@ -3452,7 +3460,7 @@ model WithingsOAuthState {
 model WhoopConnection {
   id                      String    @id @default(cuid())
   userId                  String    @unique @map("user_id")
-  whoopUserId             String    @map("whoop_user_id")
+  whoopUserId             String?   @unique @map("whoop_user_id")
   accessToken             String    @map("access_token") // encrypted at app level
   refreshToken            String    @map("refresh_token") // encrypted at app level
   tokenExpiresAt          DateTime  @map("token_expires_at")
@@ -4333,6 +4341,33 @@ model ImportJob {
   @@map("import_jobs")
 }
 
+// Durable medication-intake CSV imports. The validated, normalized payload
+// stays in Postgres so retries never depend on request-process memory.
+model MedicationIntakeImportJob {
+  id            String    @id @default(cuid())
+  userId        String    @map("user_id")
+  medicationId  String    @map("medication_id")
+  pgBossJobId   String?   @unique @map("pg_boss_job_id")
+  status        String    @default("queued")
+  payload       Json
+  progress      Json      @default("{}")
+  result        Json?
+  failureReason String?   @map("failure_reason")
+  heartbeatAt   DateTime? @map("heartbeat_at")
+  startedAt     DateTime? @map("started_at")
+  completedAt   DateTime? @map("completed_at")
+  createdAt     DateTime  @default(now()) @map("created_at")
+  updatedAt     DateTime  @updatedAt @map("updated_at")
+
+  user       User       @relation(fields: [userId], references: [id], onDelete: Cascade)
+  medication Medication @relation(fields: [medicationId], references: [id], onDelete: Cascade)
+
+  @@index([userId, createdAt])
+  @@index([medicationId, createdAt])
+  @@index([status, heartbeatAt])
+  @@map("medication_intake_import_jobs")
+}
+
 // ─── Rate Limiting ──────────────────────────────────────────
 
 model RateLimit {
@@ -5376,11 +5411,10 @@ model InsightNarrative {
 // `data-arrival` worker after a SALIENT ingest — never by a backfill, which
 // the seam classifier drops before anything is enqueued.
 //
-// The unique index is load-bearing three times over: it is the "just in"
-// marker the reaction surfaces read, it is the at-most-one-generated-line
-// per kind per day claim (inserted with ON CONFLICT DO NOTHING, so a lost
-// race means another worker already owns today's line), and it is the audit
-// trail for what the spine actually reacted to.
+// The unique index is load-bearing twice over: it is the "just in" marker the
+// reaction surfaces read and the durable identity whose newest same-day
+// revision owns generation. A strictly newer arrival atomically invalidates
+// stale generation state; equal or older deliveries are deduplicated.
 model ArrivalReaction {
   id            String    @id @default(cuid())
   userId        String    @map("user_id")
@@ -5417,8 +5451,8 @@ model ArrivalReaction {
 
   user User @relation(fields: [userId], references: [id], onDelete: Cascade)
 
-  // The once-per-kind-per-local-day claim. Do not relax this to include
-  // `refId` — the day scope is the throttle.
+  // One mutable marker per kind and local day. Do not include `refId`: newer
+  // same-day referents replace this row atomically.
   @@unique([userId, kind, localDate])
   // The reaction surfaces read "today's rows for this user".
   @@index([userId, localDate])