fix(conflicts): resolution writes funnel through SaveProgress — audio position survives

Batch H1 step 3b — THE data fix. Conflict resolution was the one live
erasure writer: applyProgressResolution and its verbatim twin
applyResolution both wrote the winner via a raw UpdateUniversalProgress
whose param list omits audio_offset_seconds/context_text/viewports, and
the SQL's DO UPDATE SET ... = EXCLUDED... NULLed exactly those. The web
UI resolves through POST /api/conflicts/:id/resolve and
/api/conflicts/bulk-resolve (main.js resolveConflict/bulkResolve), and
the reader's mark-as-read fires a resolution first when the reader SSR
stamps data-conflict-id/-winner — so every resolution today erased the
audiobook position and reader context.

Now: ProgressService is a required ConflictHandler dependency; BOTH
call sites (single resolve + bulk resolve) share one
applyProgressResolution that funnels through SaveProgress (the verbatim
twin is deleted). Deliberate choices, per the plan:
- SkipConflictDetection on the resolution write: both flows apply
  BEFORE the conflict row is marked resolved, so the funnel's LWW
  detector would otherwise file a duplicate pending conflict
  (HasRecentConflictResolution is false at apply time).
- Source "conflict_resolution" — the row's last-write attribution; the
  winner stays durably recorded in resolution_data.winner.
- Broadcast: true — live devices jump to the winning position.
- The winner snapshot maps percentage/epubcfi/context_text/chapter/
  character/page/total_pages; a winner without a CFI clears the locator
  pair (ClearLocator) instead of re-pairing with the loser's stale
  anchor. Everything the snapshot doesn't carry SURVIVES via the LWW
  seed: audio_offset_seconds above all. chapter_progress keeps its
  stored value (was overwritten with the book percentage), an absent
  percentage survives (was written as 0%).
- Accepted: the resolution captures one reading_history session
  (attributed to conflict_resolution; the 2h idle ceiling applies).

Verification: TestH1_ConflictResolutionPreservesAudio drives the real
flow — web push with audio 300s, diverging push from a second source
detects a conflict, resolve with the second source as winner, GET shows
percentage 0.3 AND audio_offset_seconds 300 (previously NULLed), and no
duplicate pending conflict remains. Full contract pin suite green.
The host-run TestConflict*/TestProgress* failures are environmental
(container-only /app/uploads path + db hostname; fail identically at
baseline b4ae423) — the same suites pass in the test-runner container.

cmd/server/main.go

@@ -151,7 +151,7 @@ func main() {
 	koreaderHandler.SetAnnotationService(annotationService)
 	koreaderHandler.SetLibraryService(libraryService)
 	wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware)
-	conflictHandler := handlers.NewConflictHandler(queries, connManager)
+	conflictHandler := handlers.NewConflictHandler(queries, connManager, progressService)
 	analyticsHandler := handlers.NewAnalyticsHandler(queries)
 	statisticsService := services.NewStatisticsService(queries)
 	statsHandler := handlers.NewStatsHandler(statisticsService)

cmd/server/tests/contract_pins_test.go

@@ -146,11 +146,19 @@ func createContractFixture(t *testing.T, setup *TestServerSetup) contractFixture
 // (testuser) and returns its auth token. Bearer-token auth for the
 // KOReader endpoints, URL-token auth for the Kobo endpoints.
 func createContractDevice(t *testing.T, setup *TestServerSetup, deviceType string) string {
+	return createContractDeviceNamed(t, setup, "Test Contract Device", deviceType)
+}
+
+// createContractDeviceNamed registers a device with an explicit name —
+// the device NAME is what the web progress push adopts as its sync
+// source (X-Device-Token attribution), so conflict tests can drive a
+// second named source without a separate account.
+func createContractDeviceNamed(t *testing.T, setup *TestServerSetup, name, deviceType string) string {
 	t.Helper()
 	token := fmt.Sprintf("dev_contract_%s", uuid.New().String())
 	device, err := setup.DB.CreateDevice(context.Background(), database.CreateDeviceParams{
 		UserID:               pgtype.UUID{Bytes: getTestUserID(t, setup.DB), Valid: true},
-		DeviceName:           "Test Contract Device",
+		DeviceName:           name,
 		DeviceType:           deviceType,
 		DeviceIdentifier:     fmt.Sprintf("contract-%s-%d", deviceType, time.Now().UnixNano()),
 		AuthToken:            token,
@@ -822,3 +830,98 @@ func TestContractPin_SSRPages(t *testing.T) {
 		assert.Contains(t, html, "- Bookhoard</title>")
 	})
 }
+
+// ---------------------------------------------------------------------------
+// 6. H1 verification — resolving a conflict must preserve unknown fields
+// ---------------------------------------------------------------------------
+
+// TestH1_ConflictResolutionPreservesAudio is the executable check for the
+// hardening plan's H1 done-when: a resolution applied via
+// POST /api/conflicts/:id/resolve (the endpoint the web UI and the
+// reader's mark-as-read both use) must NOT erase fields the winner's
+// snapshot doesn't carry (before the funnel, the resolution write NULLed
+// audio_offset_seconds), and the funneled resolution write must not file
+// a duplicate pending conflict on the item it just resolved.
+func TestH1_ConflictResolutionPreservesAudio(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+	setup := setupTestServer(t)
+	fx := createContractFixture(t, setup)
+	progress := progressURL(setup.Server.URL, fx.itemID)
+
+	// Push 1: web push carrying an audiobook-style position (source "web")
+	// — audio_offset_seconds is stored but NOT part of conflict snapshots.
+	resp := doReq(t, "PUT", progress, map[string]interface{}{
+		"percentage":           0.5,
+		"audio_offset_seconds": 300,
+	}, setup.Token)
+	defer resp.Body.Close()
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	// Push 2 within the 5-minute window from a second source (a
+	// device-token push attributes the write to the device's name) with a
+	// diverging percentage and NO audio field: the funnel detects the
+	// source mismatch + pct diff and records a conflict.
+	deviceToken := createContractDeviceNamed(t, setup, "koreader", "koreader")
+	push2Req, err := http.NewRequest("PUT", progress,
+		strings.NewReader(`{"percentage": 0.3}`))
+	require.NoError(t, err)
+	push2Req.Header.Set("Content-Type", "application/json")
+	push2Req.Header.Set("Authorization", "Bearer "+setup.Token)
+	push2Req.Header.Set("X-Device-Token", deviceToken)
+	resp2, err := (&http.Client{}).Do(push2Req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp2.StatusCode)
+	resp2.Body.Close()
+
+	// Find the freshly detected progress conflict.
+	listResp := doReq(t, "GET", setup.Server.URL+"/api/conflicts", nil, setup.Token)
+	defer listResp.Body.Close()
+	require.Equal(t, http.StatusOK, listResp.StatusCode)
+	list := decodeJSON(t, listResp)
+	conflicts, ok := list["conflicts"].([]interface{})
+	require.True(t, ok, "conflicts should be an array")
+	var conflictID string
+	for _, raw := range conflicts {
+		col := raw.(map[string]interface{})
+		if col["media_item_id"] == fx.itemID && col["conflict_type"] == "progress" && col["resolution_status"] == "unresolved" {
+			conflictID, _ = col["id"].(string)
+		}
+	}
+	require.NotEmpty(t, conflictID, "the two-source push should have detected a progress conflict")
+
+	// Resolve with the second source as the winner — the endpoint and
+	// payload shape the web UI posts (web/static/main.js resolveConflict).
+	resolve := doReq(t, "POST", setup.Server.URL+"/api/conflicts/"+conflictID+"/resolve", map[string]interface{}{
+		"winner": "koreader",
+	}, setup.Token)
+	defer resolve.Body.Close()
+	require.Equal(t, http.StatusOK, resolve.StatusCode)
+	resolved := decodeJSON(t, resolve)
+	require.Equal(t, true, resolved["conflict_resolved"])
+
+	// The saved row carries the winner's percentage AND the audio field
+	// the winner never knew about.
+	get := doReq(t, "GET", progress, nil, setup.Token)
+	defer get.Body.Close()
+	require.Equal(t, http.StatusOK, get.StatusCode)
+	row := decodeJSON(t, get)
+	require.Equal(t, 0.3, getFloatField(t, row, "percentage"))
+	require.Equal(t, float64(300), getFloatField(t, row, "audio_offset_seconds"),
+		"resolution must preserve audio_offset_seconds (LWW seed), not NULL it")
+
+	// No duplicate pending conflict may remain for this item (the funnel's
+	// own LWW detector must skip the resolution write).
+	get2 := doReq(t, "GET", setup.Server.URL+"/api/conflicts", nil, setup.Token)
+	defer get2.Body.Close()
+	require.Equal(t, http.StatusOK, get2.StatusCode)
+	list2 := decodeJSON(t, get2)
+	for _, raw := range list2["conflicts"].([]interface{}) {
+		col := raw.(map[string]interface{})
+		if col["media_item_id"] == fx.itemID && col["conflict_type"] == "progress" {
+			require.Equal(t, "user_resolved", col["resolution_status"],
+				"the resolved conflict must not have spawned a duplicate pending conflict")
+		}
+	}
+}

cmd/server/tests/test_helpers_test.go

@@ -490,7 +490,7 @@ func setupTestServer(t *testing.T) *TestServerSetup {
 	koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor, progressService)
 	koreaderHandler.SetAnnotationService(annotationService)
 	wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware)
-	conflictHandler := handlers.NewConflictHandler(queries, connManager)
+	conflictHandler := handlers.NewConflictHandler(queries, connManager, progressService)
 	analyticsHandler := handlers.NewAnalyticsHandler(queries)
 	queueHandler := handlers.NewQueueHandler(queries, queueProcessor)
 	systemSettingsHandler := handlers.NewSystemSettingsHandler(queries)

internal/handlers/conflicts.go

@@ -18,12 +18,18 @@ import (
 type ConflictHandler struct {
 	db          *database.Queries
 	connManager *wsync.ConnectionManager
+	progressSvc *wsync.ProgressService
 }
 
-func NewConflictHandler(db *database.Queries, connManager *wsync.ConnectionManager) *ConflictHandler {
+// NewConflictHandler requires the shared ProgressService: applying a
+// resolution is a progress WRITE and must funnel through SaveProgress so
+// the winner's snapshot lands without erasing row fields it doesn't carry
+// (audio_offset_seconds survives the write).
+func NewConflictHandler(db *database.Queries, connManager *wsync.ConnectionManager, progressSvc *wsync.ProgressService) *ConflictHandler {
 	return &ConflictHandler{
 		db:          db,
 		connManager: connManager,
+		progressSvc: progressSvc,
 	}
 }
 
@@ -294,71 +300,64 @@ func (h *ConflictHandler) ResolveConflict(c *echo.Context) error {
 	return c.JSON(http.StatusOK, response)
 }
 
+// applyProgressResolution writes the winning snapshot through the
+// SaveProgress funnel. Deliberate choices (hardening H1):
+//   - Source "conflict_resolution": the row's last-write attribution; the
+//     winner's identity is durably recorded on the conflict row
+//     (resolution_data.winner).
+//   - SkipConflictDetection: both flows apply BEFORE the conflict row is
+//     marked resolved, so the funnel's LWW detector would otherwise file a
+//     duplicate pending conflict on the item just resolved
+//     (HasRecentConflictResolution is false at apply time).
+//   - Broadcast: live devices jump to the winning position, same as a
+//     device push; the write also captures one reading_history session
+//     (idle-gap ceiling applies) — accepted as the resolution's own event.
+//   - Every field the winner's snapshot doesn't carry (audio_offset_seconds,
+//     viewports, zoom, reading_mode, and an absent percentage) SURVIVES via
+//     the service's LWW seed instead of being NULLed — that is the fix.
+//     chapter_progress keeps its stored value instead of being overwritten
+//     with the book-level percentage.
 func (h *ConflictHandler) applyProgressResolution(mediaItemID pgtype.UUID, userID pgtype.UUID, winnerSource string, data map[string]interface{}) error {
-	ctx := context.Background()
+	req := wsync.SaveProgressRequest{
-
+		MediaItemID:           mediaItemID,
-	existingProgress, err := h.db.GetReadingProgress(ctx, database.GetReadingProgressParams{
+		UserID:                userID,
-		MediaItemID: mediaItemID,
+		Source:                "conflict_resolution",
-		UserID:      userID,
+		SkipConflictDetection: true,
-	})
+		Broadcast:             true,
-	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		// A winner without a CFI (percentage-only sources, e.g. Kobo)
-		return err
+		// declares no locator: clear the inherited anchor pair instead of
+		// re-pairing the new percentage with the loser's stale CFI/
+		// context. A winner WITH a CFI keeps the stored pair coherent.
+		ClearLocator: true,
 	}
-
-	percentage := 0.0
 	if p, ok := data["percentage"].(float64); ok {
-		percentage = p
+		req.Percentage = &p
 	}
-
-	var epubcfi pgtype.Text
 	if e, ok := data["epubcfi"].(string); ok {
-		epubcfi = pgtype.Text{String: e, Valid: true}
+		req.Epubcfi = &e
+		req.ClearLocator = false
+	}
+	if ct, ok := data["context_text"].(string); ok {
+		req.ContextText = &ct
 	}
-
-	var chapter pgtype.Int4
 	if c, ok := data["chapter"].(float64); ok {
-		chapter = pgtype.Int4{Int32: int32(c), Valid: true}
+		ch := int(c)
+		req.Chapter = &ch
 	}
-
-	var characterOffset pgtype.Int8
 	if c, ok := data["character"].(float64); ok {
-		characterOffset = pgtype.Int8{Int64: int64(c), Valid: true}
+		co := int64(c)
+		req.CharacterOffset = &co
 	}
-
-	var currentPage pgtype.Int4
-	var totalPages pgtype.Int4
-	if err == nil {
-		currentPage = existingProgress.CurrentPage
-		totalPages = existingProgress.TotalPages
-	}
-
 	if p, ok := data["page"].(float64); ok {
-		currentPage = pgtype.Int4{Int32: int32(p), Valid: true}
+		pg := int(p)
+		req.CurrentPage = &pg
 	}
 	if p, ok := data["total_pages"].(float64); ok {
-		totalPages = pgtype.Int4{Int32: int32(p), Valid: true}
+		tp := int(p)
+		req.TotalPages = &tp
 	}
 
-	_, err = h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{
+	_, err := h.progressSvc.SaveProgress(context.Background(), req)
-		MediaItemID:     mediaItemID,
-		UserID:          userID,
-		Percentage:      pgtype.Float8{Float64: percentage, Valid: true},
-		Epubcfi:         epubcfi,
-		Chapter:         chapter,
-		ChapterProgress: pgtype.Float8{Float64: percentage, Valid: true},
-		CharacterOffset: characterOffset,
-		CurrentPage:     currentPage,
-		TotalPages:      totalPages,
-		LastSyncDevice:  pgtype.Text{String: "conflict_resolution", Valid: true},
-		LastSyncSource:  pgtype.Text{String: winnerSource, Valid: true},
-		ViewportY:       pgtype.Float8{},
-		ScrollPositionX: pgtype.Float8{},
-		ScrollPositionY: pgtype.Float8{},
-		PanelNumber:     pgtype.Int4{},
-		ReadingMode:     pgtype.Text{},
-		ZoomLevel:       pgtype.Float8{},
-	})
-
 	return err
 }
 
@@ -699,7 +698,7 @@ func (h *ConflictHandler) BulkResolveConflicts(c *echo.Context) error {
 			continue
 		}
 
-		if err := h.applyResolution(conflict.MediaItemID, conflict.UserID, winningSource, winnerData); err != nil {
+		if err := h.applyProgressResolution(conflict.MediaItemID, conflict.UserID, winningSource, winnerData); err != nil {
 			results = append(results, ConflictResult{
 				ConflictID: conflictIDStr,
 				Status:     "error",
@@ -783,74 +782,6 @@ func (h *ConflictHandler) getHighestProgressSource(conflictData map[string]Confl
 	return highestSource, highestData
 }
 
-func (h *ConflictHandler) applyResolution(mediaItemID pgtype.UUID, userID pgtype.UUID, winnerSource string, data map[string]interface{}) error {
-	ctx := context.Background()
-
-	existingProgress, err := h.db.GetReadingProgress(ctx, database.GetReadingProgressParams{
-		MediaItemID: mediaItemID,
-		UserID:      userID,
-	})
-	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
-		return err
-	}
-
-	percentage := 0.0
-	if p, ok := data["percentage"].(float64); ok {
-		percentage = p
-	}
-
-	var epubcfi pgtype.Text
-	if e, ok := data["epubcfi"].(string); ok {
-		epubcfi = pgtype.Text{String: e, Valid: true}
-	}
-
-	var chapter pgtype.Int4
-	if c, ok := data["chapter"].(float64); ok {
-		chapter = pgtype.Int4{Int32: int32(c), Valid: true}
-	}
-
-	var characterOffset pgtype.Int8
-	if c, ok := data["character"].(float64); ok {
-		characterOffset = pgtype.Int8{Int64: int64(c), Valid: true}
-	}
-
-	var currentPage pgtype.Int4
-	var totalPages pgtype.Int4
-	if err == nil {
-		currentPage = existingProgress.CurrentPage
-		totalPages = existingProgress.TotalPages
-	}
-
-	if p, ok := data["page"].(float64); ok {
-		currentPage = pgtype.Int4{Int32: int32(p), Valid: true}
-	}
-	if p, ok := data["total_pages"].(float64); ok {
-		totalPages = pgtype.Int4{Int32: int32(p), Valid: true}
-	}
-
-	_, err = h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{
-		MediaItemID:     mediaItemID,
-		UserID:          userID,
-		Percentage:      pgtype.Float8{Float64: percentage, Valid: true},
-		Epubcfi:         epubcfi,
-		Chapter:         chapter,
-		ChapterProgress: pgtype.Float8{Float64: percentage, Valid: true},
-		CharacterOffset: characterOffset,
-		CurrentPage:     currentPage,
-		TotalPages:      totalPages,
-		LastSyncDevice:  pgtype.Text{String: "conflict_resolution", Valid: true},
-		LastSyncSource:  pgtype.Text{String: winnerSource, Valid: true},
-		ViewportY:       pgtype.Float8{},
-		ScrollPositionX: pgtype.Float8{},
-		ScrollPositionY: pgtype.Float8{},
-		PanelNumber:     pgtype.Int4{},
-		ReadingMode:     pgtype.Text{},
-		ZoomLevel:       pgtype.Float8{},
-	})
-
-	return err
-}
-
 func (h *ConflictHandler) BulkDismissConflicts(c *echo.Context) error {
 	user := c.Get("user").(database.Users)
 

internal/sync/progress.go

@@ -341,6 +341,14 @@ type SaveProgressRequest struct {
 	// percentage restores are approximate, and the next anchor-bearing
 	// write re-establishes the pair. Request-supplied fields always win.
 	ClearLocator bool `json:"-"`
+
+	// SkipConflictDetection declares this write APPLIES a conflict
+	// resolution rather than continuing the sync stream: the two flows
+	// (single + bulk) apply the winner BEFORE the conflict row is marked
+	// resolved, so the funnel's LWW detector would see the losing source
+	// within the 5-minute window and file a duplicate pending conflict on
+	// the item the user just resolved. Server-owned (json:"-").
+	SkipConflictDetection bool `json:"-"`
 }
 
 func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequest) (database.ReadingProgress, error) {
@@ -529,7 +537,7 @@ func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequ
 	}
 
 	conflictDetected := false
-	if hasExisting && existing.LastSyncSource.Valid && existing.LastSyncSource.String != req.Source {
+	if !req.SkipConflictDetection && hasExisting && existing.LastSyncSource.Valid && existing.LastSyncSource.String != req.Source {
 		if existing.LastSyncTimestamp.Valid {
 			timeDiff := time.Since(existing.LastSyncTimestamp.Time)
 			if timeDiff < 5*time.Minute {