Merge pull request #196 from Dzarlax-AI/codex/healthkit-import-dedup-batching

Implement staged Apple Health XML import

Author: dzarlax

AGENTS.md

@@ -60,11 +60,11 @@ Single binary HTTP server (`cmd/server/main.go`) that wires together several pac
   ```
   `MaybeFireAll` is called from the morning scheduler tick (see `runMorningSmartRetry` in `cmd/server/main.go`). It honours per-rule cadence + eligibility and never bubbles errors — a misbehaving rule cannot block the morning report path. Render returning `""` is treated as "skip with no persist" (idiomatic when the render itself re-evaluates a condition). See `digest.go` and `energy_nudge.go` for production examples.
 
-- **`internal/applehealth`** — streaming XML parser for Apple Health export files (`export.xml` or `.zip`). Memory-efficient, maps 100+ HK metric types to internal metric names. Normalizes fraction-based percentage metrics (SpO₂, body fat, etc.) to 0–100 scale during import.
+- **`internal/applehealth`** — streaming XML parser for Apple Health export files (`export.xml` or `.zip`). Memory-efficient, maps 100+ HK metric types to internal metric names. Normalizes fraction-based percentage metrics (SpO₂, body fat, etc.) to 0–100 scale during import. Admin UI and CLI XML imports stage through `storage.ImportSession`: one logical `import_runs` row + one `health_records` row per import, `pgx.CopyFrom` temp staging, coverage metadata, set-based promote, and one cache rebuild after commit. XML snapshot freshness is time-based using `HealthData@exportDate` when available: XML supersedes older live/HAE exact rows, later live/mobile exact rows remain fresher, and a later XML snapshot may supersede them again. Do not reintroduce broad `RemoveAutoExportForRange` cleanup for XML imports.
 
 - **`cmd/backfill`** — standalone CLI to rebuild caches. Flags: `--force` / `-f`.
 
-- **`cmd/import`** — standalone CLI to import Apple Health export files. Flags: `--file`, `--batch`, `--pause`, `--dry-run`. Streams XML to avoid memory overload.
+- **`cmd/import`** — standalone CLI to import Apple Health export files. Flags: `--file`, `--batch`, `--pause`, `--dry-run`; `--pause` is retained for compatibility but staged imports ignore per-batch sleeps. Streams XML to avoid memory overload and uses the same `storage.ImportSession` path as Admin UI.
 
 ## Data Flow
 

CLAUDE.md

@@ -60,11 +60,11 @@ Single binary HTTP server (`cmd/server/main.go`) that wires together several pac
   ```
   `MaybeFireAll` is called from the morning scheduler tick (see `runMorningSmartRetry` in `cmd/server/main.go`). It honours per-rule cadence + eligibility and never bubbles errors — a misbehaving rule cannot block the morning report path. Render returning `""` is treated as "skip with no persist" (idiomatic when the render itself re-evaluates a condition). See `digest.go` and `energy_nudge.go` for production examples.
 
-- **`internal/applehealth`** — streaming XML parser for Apple Health export files (`export.xml` or `.zip`). Memory-efficient, maps 100+ HK metric types to internal metric names. Normalizes fraction-based percentage metrics (SpO₂, body fat, etc.) to 0–100 scale during import.
+- **`internal/applehealth`** — streaming XML parser for Apple Health export files (`export.xml` or `.zip`). Memory-efficient, maps 100+ HK metric types to internal metric names. Normalizes fraction-based percentage metrics (SpO₂, body fat, etc.) to 0–100 scale during import. Admin UI and CLI XML imports stage through `storage.ImportSession`: one logical `import_runs` row + one `health_records` row per import, `pgx.CopyFrom` temp staging, coverage metadata, set-based promote, and one cache rebuild after commit. XML snapshot freshness is time-based using `HealthData@exportDate` when available: XML supersedes older live/HAE exact rows, later live/mobile exact rows remain fresher, and a later XML snapshot may supersede them again. Do not reintroduce broad `RemoveAutoExportForRange` cleanup for XML imports.
 
 - **`cmd/backfill`** — standalone CLI to rebuild caches. Flags: `--force` / `-f`.
 
-- **`cmd/import`** — standalone CLI to import Apple Health export files. Flags: `--file`, `--batch`, `--pause`, `--dry-run`. Streams XML to avoid memory overload.
+- **`cmd/import`** — standalone CLI to import Apple Health export files. Flags: `--file`, `--batch`, `--pause`, `--dry-run`; `--pause` is retained for compatibility but staged imports ignore per-batch sleeps. Streams XML to avoid memory overload and uses the same `storage.ImportSession` path as Admin UI.
 
 ## Data Flow
 

README.md

@@ -225,7 +225,7 @@ The endpoint stores summary fields only (duration, distance, energy, average / m
 
 If `HEALTH_HR_ZONES_BPM` is configured, the ingest path also computes time-in-zone (Z1..Z5) from the per-minute HR samples in the payload and stores five integer columns per workout. Pick zone borders that match your physiology — the Karvonen (HR Reserve) method using observed MaxHR and resting HR is more accurate than the textbook `220 - age` formula.
 
-Apple Health XML imports (`cmd/import` and the Admin import UI) also ingest `<Workout>` summaries into the same `workouts` table. XML exports provide nested `WorkoutStatistics` for distance, energy, and aggregate heart-rate fields on some workouts, so imported rows may include `distance_km`, `energy_kcal`, `avg_hr_bpm`, and `max_hr_bpm`. GPX workout routes are deliberately not imported.
+Apple Health XML imports (`cmd/import` and the Admin import UI) also ingest `<Workout>` summaries into the same `workouts` table. XML exports provide nested `WorkoutStatistics` for distance, energy, and aggregate heart-rate fields on some workouts, so imported rows may include `distance_km`, `energy_kcal`, `avg_hr_bpm`, and `max_hr_bpm`. GPX workout routes are deliberately not imported. XML workouts with stable IDs are upserted by `external_id`; synthetic `applexml:` workout IDs are additionally deduplicated by stable workout window fields so corrected XML snapshots do not leave duplicate rows behind.
 
 The `/health` endpoint (no suffix) still accepts all metrics unfiltered for backward compatibility.
 
@@ -402,11 +402,21 @@ Import a full Apple Health export (from iPhone Settings > Health > Export All He
 
 **Via CLI**:
 ```bash
-DATABASE_URL=postgres://... go run ./cmd/import --file path/to/export.zip
+DATABASE_URL=postgres://... go run ./cmd/import --file path/to/export.zip --batch 100000
 ```
 
 The import streams the XML to avoid memory issues with large files. Percentage metrics (SpO2, body fat, walking asymmetry, etc.) are automatically normalized from Apple Health's fraction format (0.96) to percentage scale (96%).
 
+Admin UI and CLI imports share the same staged storage path:
+
+1. create one logical `import_runs` row and one `health_records` row for the export;
+2. read `HealthData@exportDate` as the snapshot freshness timestamp, falling back to import start only when it is unavailable;
+3. stream parsed metric points and workouts into temporary staging tables with bulk database operations;
+4. promote staged rows in one final transaction, recording coverage metadata and provenance;
+5. invalidate affected aggregate rows and run one cache rebuild after commit.
+
+If XML parsing or staging fails, the import is marked failed and no cleanup/backfill is run. The old row-by-row `BulkInsertPoints` path is retained only for narrow compatibility helpers.
+
 After import, run `make energy-backfill` (CLI) or open **Settings → Historical EnergyBank** (web UI) to compute retrospective EnergyBank snapshots from the imported daily scores. This unlocks per-user verdict band calibration; without it, the cold-start defaults are used until the live orchestrator has accumulated 30+ days of snapshots on its own.
 
 ## Multi-Device Source Priority
@@ -429,7 +439,14 @@ This priority is applied consistently across dashboard, daily scores, readiness
 
 ### Apple Health Import & Auto Export Conflict Resolution
 
-When re-importing from Apple Health (e.g. to fill gaps from missed Auto Export days), the system automatically removes Auto Export (`Health dash - Hourly`/`Vitals`) data for overlapping dates. Apple Health export is treated as ground truth. A full cache rebuild (force backfill) runs after each import.
+Apple Health XML export is treated as a timestamped HealthKit snapshot, not as a permanent lock on historical rows:
+
+- XML import supersedes older live / Health Auto Export rows on the same exact `(metric_name, date, source)` key.
+- Live/mobile data ingested after that XML snapshot is considered fresher and can overwrite the exact key.
+- A later XML export may supersede those rows again as a newer HealthKit snapshot.
+- Rows missing from a newer XML snapshot are cleaned up conservatively only within coverage groups proven by staging metadata; the importer does not delete every row in a broad `min_date` / `max_date` range.
+
+This lets full exports repair old HealthKit history while still allowing the first-party app to keep recent data fresh.
 
 ## Readiness Score
 

cmd/import/main.go

@@ -2,17 +2,16 @@
 //
 // Usage:
 //
-//	import --db /app/data/health.db --file export.zip [--batch 500] [--pause 200ms] [--dry-run]
+//	import --file export.zip [--batch 100000] [--dry-run]
 //
 // The importer streams the XML so it never loads the full file into memory.
-// Batches are inserted with a configurable pause between them so the running
-// server is not starved of DB connections during a large historical import.
+// Parsed rows are staged with bulk database operations, then promoted in one
+// final transaction so parse errors do not leave partial imports behind.
 package main
 
 import (
 	"context"
 	"flag"
-	"fmt"
 	"log"
 	"os"
 	"time"
@@ -22,17 +21,19 @@ import (
 )
 
 func main() {
-	filePath := flag.String("file", "", "Apple Health export (.zip or export.xml) — required")
-	batchSize := flag.Int("batch", 500, "metric points or workouts per DB transaction")
-	pauseDur := flag.Duration("pause", 150*time.Millisecond, "sleep between batches (rate-limits DB load)")
-	dryRun := flag.Bool("dry-run", false, "parse only — do not write to DB")
+	filePath := flag.String("file", "", "Apple Health export (.zip or export.xml) - required")
+	batchSize := flag.Int("batch", 100000, "metric points or workouts per staging flush")
+	pauseDur := flag.Duration("pause", 0, "deprecated; staged import ignores per-batch sleeps")
+	dryRun := flag.Bool("dry-run", false, "parse only - do not write to DB")
 	flag.Parse()
+	_ = pauseDur
 
 	if *filePath == "" {
 		log.Fatal("--file is required")
 	}
 
 	var db *storage.DB
+	var session *storage.ImportSession
 	if !*dryRun {
 		dbURL := os.Getenv("DATABASE_URL")
 		if dbURL == "" {
@@ -44,79 +45,80 @@ func main() {
 			log.Fatalf("open db: %v", err)
 		}
 		defer db.Close()
+
+		snapshotAt := time.Now()
+		var exportDateErr error
+		var exportDateFound bool
+		switch {
+		case len(*filePath) > 4 && (*filePath)[len(*filePath)-4:] == ".zip":
+			if t, ok, err := applehealth.ExportDateFromZip(*filePath); err != nil {
+				exportDateErr = err
+			} else if ok {
+				snapshotAt = t
+				exportDateFound = true
+			}
+		default:
+			if t, ok, err := applehealth.ExportDateFromXMLFile(*filePath); err != nil {
+				exportDateErr = err
+			} else if ok {
+				snapshotAt = t
+				exportDateFound = true
+			}
+		}
+		if exportDateErr != nil {
+			log.Printf("could not read Apple Health exportDate, using import start as snapshot time: %v", exportDateErr)
+		} else if !exportDateFound {
+			log.Printf("Apple Health exportDate not found, using import start as snapshot time")
+		}
+
+		session, err = db.BeginAppleHealthXMLImport(storage.ImportOptions{
+			SourceName: "apple-health-cli-import",
+			SnapshotAt: snapshotAt,
+		})
+		if err != nil {
+			log.Fatalf("begin import: %v", err)
+		}
 	}
 
 	var (
 		totalParsed         int
-		totalInserted       int
 		totalWorkoutsParsed int
-		totalWorkoutsUpsert int
-		totalWorkoutsFailed int
-		batchN              int
-		workoutBatchN       int
 		pending             []storage.MetricPoint
 		pendingWorkouts     []storage.Workout
 		startTime           = time.Now()
 	)
 
 	flush := func(pts []storage.MetricPoint) {
-		batchN++
 		totalParsed += len(pts)
-
 		if *dryRun {
-			if totalParsed%100_000 == 0 || totalParsed < 1000 {
-				log.Printf("[dry-run] parsed %d points so far…", totalParsed)
+			if totalParsed%100000 == 0 || totalParsed < 1000 {
+				log.Printf("[dry-run] parsed %d points so far", totalParsed)
 			}
 			return
 		}
-
-		desc := fmt.Sprintf("apple-health-import batch %d", batchN)
-		n, err := db.BulkInsertPoints(desc, pts)
-		if err != nil {
-			log.Printf("batch %d insert error: %v (continuing)", batchN, err)
+		if err := session.AddPoints(pts); err != nil {
+			session.Abort(err)
+			log.Fatalf("stage points: %v", err)
 		}
-		totalInserted += n
-
-		elapsed := time.Since(startTime).Round(time.Second)
-		log.Printf("batch %d: %d parsed / %d new (total %d inserted, %s elapsed)",
-			batchN, len(pts), n, totalInserted, elapsed)
-
-		if *pauseDur > 0 {
-			time.Sleep(*pauseDur)
+		if totalParsed%100000 == 0 {
+			log.Printf("staged %d points so far", totalParsed)
 		}
 	}
 
 	flushWorkouts := func(workouts []storage.Workout) {
-		workoutBatchN++
 		totalWorkoutsParsed += len(workouts)
-
 		if *dryRun {
 			if totalWorkoutsParsed%100 == 0 || totalWorkoutsParsed < 20 {
-				log.Printf("[dry-run] parsed %d workouts so far…", totalWorkoutsParsed)
+				log.Printf("[dry-run] parsed %d workouts so far", totalWorkoutsParsed)
 			}
 			return
 		}
-
-		var failed int
-		for _, w := range workouts {
-			if err := db.UpsertWorkout(0, w); err != nil {
-				failed++
-				log.Printf("workout batch %d upsert %s error: %v", workoutBatchN, w.ExternalID, err)
-			}
-		}
-		totalWorkoutsFailed += failed
-		totalWorkoutsUpsert += len(workouts) - failed
-
-		elapsed := time.Since(startTime).Round(time.Second)
-		log.Printf("workout batch %d: %d parsed / %d upserted / %d failed (total %d upserted, %s elapsed)",
-			workoutBatchN, len(workouts), len(workouts)-failed, failed, totalWorkoutsUpsert, elapsed)
-
-		if *pauseDur > 0 {
-			time.Sleep(*pauseDur)
+		if err := session.AddWorkouts(workouts); err != nil {
+			session.Abort(err)
+			log.Fatalf("stage workouts: %v", err)
 		}
 	}
 
-	// Collect points up to batchSize, then flush.
 	emit := func(pts []storage.MetricPoint) {
 		pending = append(pending, pts...)
 		for len(pending) >= *batchSize {
@@ -133,8 +135,7 @@ func main() {
 		}
 	}
 
-	log.Printf("starting import from %s (batch=%d pause=%s dry-run=%v)",
-		*filePath, *batchSize, *pauseDur, *dryRun)
+	log.Printf("starting import from %s (batch=%d dry-run=%v)", *filePath, *batchSize, *dryRun)
 
 	opts := applehealth.EmitOptions{Points: emit, Workouts: emitWorkouts}
 	var parseErr error
@@ -150,22 +151,30 @@ func main() {
 	if len(pendingWorkouts) > 0 {
 		flushWorkouts(pendingWorkouts)
 	}
-
 	if parseErr != nil {
-		log.Printf("parse error (partial import may have succeeded): %v", parseErr)
+		if session != nil {
+			session.Abort(parseErr)
+		}
+		log.Fatalf("parse error: %v", parseErr)
 	}
 
 	elapsed := time.Since(startTime).Round(time.Second)
 	if *dryRun {
 		log.Printf("dry-run complete: %d points and %d workouts parsed in %s", totalParsed, totalWorkoutsParsed, elapsed)
 		return
 	}
-	log.Printf("import complete: %d points parsed, %d inserted, %d duplicates skipped; %d workouts parsed, %d upserted, %d failed in %s",
-		totalParsed, totalInserted, totalParsed-totalInserted, totalWorkoutsParsed, totalWorkoutsUpsert, totalWorkoutsFailed, elapsed)
 
-	// Trigger a full backfill so daily_scores and caches are up to date.
-	log.Println("running backfill (this may take a few minutes)…")
+	counters, err := session.Commit()
+	if err != nil {
+		log.Fatalf("commit import: %v", err)
+	}
+	log.Printf("import committed: run=%d %d points parsed, %d inserted, %d updated, %d stale deleted, %d duplicate-stage skipped; %d workouts parsed, %d upserted, %d stale deleted in %s",
+		counters.ImportRunID, counters.ParsedPoints, counters.InsertedPoints, counters.UpdatedPoints,
+		counters.DeletedPoints, counters.SkippedPoints, counters.ParsedWorkouts, counters.UpsertedWorkouts,
+		counters.DeletedWorkouts, elapsed)
+
+	log.Println("running backfill (this may take a few minutes)")
 	db.BackfillAggregates(true)
 	db.BackfillScores(true)
-	log.Println("backfill done — import complete")
+	log.Println("backfill done - import complete")
 }