Skip to content

Commit 74f323e

Browse files
dzarlaxdzarlax
committed
Address import staging review feedback
1 parent be6f1b9 commit 74f323e

4 files changed

Lines changed: 82 additions & 29 deletions

File tree

‎internal/storage/import_run.go‎

Lines changed: 43 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package storage
33
import (
44
"context"
55
"fmt"
6+
"log"
67
"time"
78

89
"github.com/jackc/pgx/v5"
@@ -62,8 +63,11 @@ func (s *DB) BeginAppleHealthXMLImport(opts ImportOptions) (*ImportSession, erro
6263
ctx, cancel := longCtx()
6364
defer cancel()
6465

66+
if err := s.ensureAppleHealthImportSchema(ctx); err != nil {
67+
return nil, fmt.Errorf("ensure import schema: %w", err)
68+
}
6569
if err := s.cleanupAbandonedImportStages(ctx, abandonedRunningImportAge); err != nil {
66-
return nil, fmt.Errorf("cleanup import staging: %w", err)
70+
log.Printf("cleanup abandoned import staging before new import: %v", err)
6771
}
6872

6973
var runID int64
@@ -100,6 +104,26 @@ func (s *DB) BeginAppleHealthXMLImport(opts ImportOptions) (*ImportSession, erro
100104
return session, nil
101105
}
102106

107+
func (s *DB) ensureAppleHealthImportSchema(ctx context.Context) error {
108+
tables := []string{
109+
importRunsTableDDL,
110+
importRunCoverageTableDDL,
111+
importStagePointsTableDDL,
112+
importStageWorkoutsTableDDL,
113+
}
114+
for _, ddl := range tables {
115+
if _, err := s.pool.Exec(ctx, ddl); err != nil {
116+
return fmt.Errorf("create import table: %w", err)
117+
}
118+
}
119+
for _, index := range importStageIndexMigrations() {
120+
if _, err := s.pool.Exec(ctx, index.ddl); err != nil {
121+
return fmt.Errorf("create import index %s: %w", index.name, err)
122+
}
123+
}
124+
return nil
125+
}
126+
103127
func (s *ImportSession) AddPoints(points []MetricPoint) error {
104128
if len(points) == 0 {
105129
return nil
@@ -182,34 +206,19 @@ func (s *ImportSession) Commit() (ImportCounters, error) {
182206
}
183207

184208
if err := s.promoteCoverage(ctx, tx); err != nil {
185-
_ = tx.Rollback(ctx)
186-
_ = s.db.markImportFailedWithCounters(s.runID, err.Error(), s.counters)
187-
_ = s.db.cleanupImportStageRows(s.runID)
188-
return s.counters, err
209+
return s.failCommit(tx, err)
189210
}
190211
if err := s.promotePoints(ctx, tx); err != nil {
191-
_ = tx.Rollback(ctx)
192-
_ = s.db.markImportFailedWithCounters(s.runID, err.Error(), s.counters)
193-
_ = s.db.cleanupImportStageRows(s.runID)
194-
return s.counters, err
212+
return s.failCommit(tx, err)
195213
}
196214
if err := s.promoteWorkouts(ctx, tx); err != nil {
197-
_ = tx.Rollback(ctx)
198-
_ = s.db.markImportFailedWithCounters(s.runID, err.Error(), s.counters)
199-
_ = s.db.cleanupImportStageRows(s.runID)
200-
return s.counters, err
215+
return s.failCommit(tx, err)
201216
}
202217
if err := s.markCommitted(ctx, tx); err != nil {
203-
_ = tx.Rollback(ctx)
204-
_ = s.db.markImportFailedWithCounters(s.runID, err.Error(), s.counters)
205-
_ = s.db.cleanupImportStageRows(s.runID)
206-
return s.counters, err
218+
return s.failCommit(tx, err)
207219
}
208220
if err := s.deleteStageRows(ctx, tx); err != nil {
209-
_ = tx.Rollback(ctx)
210-
_ = s.db.markImportFailedWithCounters(s.runID, err.Error(), s.counters)
211-
_ = s.db.cleanupImportStageRows(s.runID)
212-
return s.counters, err
221+
return s.failCommit(tx, err)
213222
}
214223
if err := tx.Commit(ctx); err != nil {
215224
// The outcome can be ambiguous after Commit returns an error. Leave the
@@ -220,6 +229,19 @@ func (s *ImportSession) Commit() (ImportCounters, error) {
220229
return s.counters, nil
221230
}
222231

232+
func (s *ImportSession) failCommit(tx pgx.Tx, cause error) (ImportCounters, error) {
233+
if err := tx.Rollback(context.Background()); err != nil {
234+
log.Printf("rollback import %d promote transaction: %v", s.runID, err)
235+
}
236+
if err := s.db.markImportFailedWithCounters(s.runID, cause.Error(), s.counters); err != nil {
237+
log.Printf("mark import %d failed: %v", s.runID, err)
238+
}
239+
if err := s.db.cleanupImportStageRows(s.runID); err != nil {
240+
log.Printf("cleanup stage rows for import %d: %v", s.runID, err)
241+
}
242+
return s.counters, cause
243+
}
244+
223245
func (s *ImportSession) Abort(cause error) {
224246
msg := ""
225247
if cause != nil {

‎internal/storage/import_run_test.go‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -387,6 +387,34 @@ func TestAppleHealthXMLImportCleanupAbandonedStagesPreservesRecentRunning(t *tes
387387
}
388388
}
389389

390+
func TestBeginAppleHealthXMLImportEnsuresPersistentStageSchema(t *testing.T) {
391+
db, cleanup := testDB(t)
392+
defer cleanup()
393+
394+
ctx, cancel := queryCtx()
395+
defer cancel()
396+
if _, err := db.pool.Exec(ctx, `DROP TABLE IF EXISTS import_stage_workouts, import_stage_points`); err != nil {
397+
t.Fatalf("drop stage tables: %v", err)
398+
}
399+
defer db.EnsureIndexes()
400+
401+
xml := beginTestXMLImport(t, db, "xml-ensure-stage-schema")
402+
if err := xml.AddPoints([]MetricPoint{{
403+
MetricName: "heart_rate",
404+
Units: "count/min",
405+
Date: "2026-07-01 10:00:00 +0200",
406+
Qty: 61,
407+
Source: "Apple Watch",
408+
}}); err != nil {
409+
t.Fatalf("stage point after schema ensure: %v", err)
410+
}
411+
assertStageCounts(t, db, xml.runID, 1, 0)
412+
413+
if _, err := xml.Commit(); err != nil {
414+
t.Fatalf("commit xml: %v", err)
415+
}
416+
}
417+
390418
func TestAppleHealthXMLImportReplacesSyntheticWorkoutDuplicate(t *testing.T) {
391419
db, cleanup := testDB(t)
392420
defer cleanup()
@@ -449,7 +477,7 @@ func beginTestXMLImport(t *testing.T, db *DB, source string) *ImportSession {
449477
t.Helper()
450478
session, err := db.BeginAppleHealthXMLImport(ImportOptions{
451479
SourceName: source,
452-
SnapshotAt: time.Now(),
480+
SnapshotAt: time.Now().Add(time.Minute),
453481
})
454482
if err != nil {
455483
t.Fatalf("begin xml import: %v", err)

‎internal/storage/import_schema.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,3 +68,12 @@ const importStageWorkoutsTableDDL = `CREATE TABLE IF NOT EXISTS import_stage_wor
6868
hr_z4_sec INTEGER,
6969
hr_z5_sec INTEGER
7070
)`
71+
72+
func importStageIndexMigrations() []indexMigration {
73+
return []indexMigration{
74+
{"idx_import_stage_points_dedup", `CREATE INDEX IF NOT EXISTS idx_import_stage_points_dedup ON import_stage_points (import_run_id, metric_name, date, source, staged_seq DESC)`},
75+
{"idx_import_stage_points_coverage", `CREATE INDEX IF NOT EXISTS idx_import_stage_points_coverage ON import_stage_points (import_run_id, metric_name, source, local_date)`},
76+
{"idx_import_stage_workouts_dedup", `CREATE INDEX IF NOT EXISTS idx_import_stage_workouts_dedup ON import_stage_workouts (import_run_id, external_id, staged_seq DESC)`},
77+
{"idx_import_stage_workouts_synthetic", `CREATE INDEX IF NOT EXISTS idx_import_stage_workouts_synthetic ON import_stage_workouts (import_run_id, name, start_time, end_time)`},
78+
}
79+
}

‎internal/storage/indexes.go‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -128,14 +128,8 @@ func (s *DB) EnsureIndexes() {
128128

129129
// Speeds up WHERE metric_name = $1 AND date-part queries on metric_points
130130
{"idx_points_metric_date", `CREATE INDEX IF NOT EXISTS idx_points_metric_date ON metric_points (metric_name, SUBSTRING(date,1,10))`},
131-
132-
// Persistent Apple Health XML import staging. Each index starts with
133-
// import_run_id because concurrent imports share the same stage tables.
134-
{"idx_import_stage_points_dedup", `CREATE INDEX IF NOT EXISTS idx_import_stage_points_dedup ON import_stage_points (import_run_id, metric_name, date, source, staged_seq DESC)`},
135-
{"idx_import_stage_points_coverage", `CREATE INDEX IF NOT EXISTS idx_import_stage_points_coverage ON import_stage_points (import_run_id, metric_name, source, local_date)`},
136-
{"idx_import_stage_workouts_dedup", `CREATE INDEX IF NOT EXISTS idx_import_stage_workouts_dedup ON import_stage_workouts (import_run_id, external_id, staged_seq DESC)`},
137-
{"idx_import_stage_workouts_synthetic", `CREATE INDEX IF NOT EXISTS idx_import_stage_workouts_synthetic ON import_stage_workouts (import_run_id, name, start_time, end_time)`},
138131
}
132+
indexes = append(indexes, importStageIndexMigrations()...)
139133

140134
existingIndexes, err := s.existingIndexes(indexes)
141135
if err != nil {

0 commit comments

Comments
 (0)