From 2d2d6438734b934295eda096b440077519488b58 Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Sat, 31 Jan 2026 11:45:52 -0500 Subject: [PATCH] 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 --- bruno/conflicts/Delete Conflict.bru | 33 ++ bruno/conflicts/Dismiss All Resolved.bru | 35 ++ bruno/conflicts/Get Conflict Details.bru | 58 ++++ bruno/conflicts/List Conflicts.bru | 36 ++ bruno/conflicts/Resolve Conflict.bru | 66 ++++ cmd/server/main.go | 23 +- cmd/server/tests/conflict_test.go | 289 ++++++++++++++++ internal/database/queries/queries.sql | 7 + internal/handlers/conflicts.go | 403 +++++++++++++++++++++++ internal/handlers/koreader.go | 136 +++++++- internal/sync/websocket.go | 14 + 11 files changed, 1084 insertions(+), 16 deletions(-) create mode 100644 bruno/conflicts/Delete Conflict.bru create mode 100644 bruno/conflicts/Dismiss All Resolved.bru create mode 100644 bruno/conflicts/Get Conflict Details.bru create mode 100644 bruno/conflicts/List Conflicts.bru create mode 100644 bruno/conflicts/Resolve Conflict.bru create mode 100644 cmd/server/tests/conflict_test.go create mode 100644 internal/handlers/conflicts.go diff --git a/bruno/conflicts/Delete Conflict.bru b/bruno/conflicts/Delete Conflict.bru new file mode 100644 index 0000000..cd4d334 --- /dev/null +++ b/bruno/conflicts/Delete Conflict.bru @@ -0,0 +1,33 @@ +meta { + name: Delete Conflict + type: http + seq: 4 +} + +delete { + url: {{baseUrl}}/api/conflicts/{{conflict_id}} + body: none + auth: bearer +} + +headers: { + Authorization: Bearer {{token}} +} + +docs: { + Deletes a specific conflict record. + + Path Parameters: + - conflict_id: UUID of the conflict to delete + + Use this when: + - A conflict was created in error + - You want to dismiss a conflict without resolving it + - The conflict is no longer relevant + + Response: 204 No Content on success + + Note: This permanently removes the conflict record. + Consider resolving the conflict instead if you want to + maintain an audit trail of what happened. +} diff --git a/bruno/conflicts/Dismiss All Resolved.bru b/bruno/conflicts/Dismiss All Resolved.bru new file mode 100644 index 0000000..44ee325 --- /dev/null +++ b/bruno/conflicts/Dismiss All Resolved.bru @@ -0,0 +1,35 @@ +meta { + name: Dismiss All Resolved Conflicts + type: http + seq: 5 +} + +post { + url: {{baseUrl}}/api/conflicts/dismiss-all + body: none + auth: bearer +} + +headers: { + Authorization: Bearer {{token}} +} + +docs: { + Deletes all resolved conflicts for the authenticated user. + + Use this to: + - Clean up your conflicts list after reviewing resolutions + - Remove old resolved conflicts that are no longer needed + - Maintain a clean conflict history + + Response includes: + - deleted: Number of conflict records that were deleted + + Example response: + { + "deleted": 5 + } + + Note: Only resolves conflicts with status "user_resolved" + are deleted. Unresolved conflicts are preserved. +} diff --git a/bruno/conflicts/Get Conflict Details.bru b/bruno/conflicts/Get Conflict Details.bru new file mode 100644 index 0000000..053179a --- /dev/null +++ b/bruno/conflicts/Get Conflict Details.bru @@ -0,0 +1,58 @@ +meta { + name: Get Conflict Details + type: http + seq: 2 +} + +get { + url: {{baseUrl}}/api/conflicts/{{conflict_id}} + body: none + auth: bearer +} + +headers: { + Authorization: Bearer {{token}} +} + +docs: { + Retrieves detailed information about a specific conflict. + + Path Parameters: + - conflict_id: UUID of the conflict + + Response includes: + - id: Conflict UUID + - media_item_id: Associated book UUID + - media_item_title: Book title + - conflict_type: Type of conflict + - conflict_data: Side-by-side comparison with sources: + * source: Device/source identifier (koreader, kobo, web, etc.) + * timestamp: When this progress was recorded + * data: The conflicting data (percentage, epubcfi, chapter, etc.) + - resolution_status: Current status + - resolution_data: If resolved, includes resolution details + - resolved_by: User ID who resolved it (if applicable) + - resolved_at: When it was resolved (if applicable) + - created_at: When conflict was detected + + Example conflict_data: + { + "koreader": { + "source": "koreader", + "timestamp": "2026-01-30T20:10:00Z", + "data": { + "percentage": 0.45, + "epubcfi": "epubcfi(/6/4/2:15)", + "chapter": 3 + } + }, + "kobo": { + "source": "kobo", + "timestamp": "2026-01-30T20:05:00Z", + "data": { + "percentage": 0.42, + "location": "unknown" + } + } + } +} diff --git a/bruno/conflicts/List Conflicts.bru b/bruno/conflicts/List Conflicts.bru new file mode 100644 index 0000000..829a9d2 --- /dev/null +++ b/bruno/conflicts/List Conflicts.bru @@ -0,0 +1,36 @@ +meta { + name: List Conflicts + type: http + seq: 1 +} + +get { + url: {{baseUrl}}/api/conflicts?status=unresolved + body: none + auth: bearer +} + +headers: { + Authorization: Bearer {{token}} +} + +docs: { + Lists all sync conflicts for the authenticated user. + + Query Parameters: + - status: Filter by resolution status (unresolved, user_resolved, auto_resolved, all) + + Response includes: + - conflicts: Array of conflict details + - total: Total number of conflicts + - unresolved: Number of unresolved conflicts + + Each conflict includes: + - id: Conflict UUID + - media_item_id: Associated book UUID + - media_item_title: Book title + - conflict_type: Type of conflict (progress, note, highlight) + - conflict_data: Side-by-side comparison of conflicting data + - resolution_status: Current status + - created_at: When conflict was detected +} diff --git a/bruno/conflicts/Resolve Conflict.bru b/bruno/conflicts/Resolve Conflict.bru new file mode 100644 index 0000000..292f35b --- /dev/null +++ b/bruno/conflicts/Resolve Conflict.bru @@ -0,0 +1,66 @@ +meta { + name: Resolve Conflict + type: http + seq: 3 +} + +post { + url: {{baseUrl}}/api/conflicts/{{conflict_id}}/resolve + body: json + auth: bearer +} + +headers: { + Authorization: Bearer {{token}} + Content-Type: application/json +} + +body:json { + { + "winner": "koreader", + "manual_data": null, + "apply_to_all_future_conflicts": false, + "reason": "User chose more recent progress" + } +} + +docs: { + Resolves a sync conflict by choosing which source to use. + + Path Parameters: + - conflict_id: UUID of the conflict to resolve + + Request Body: + - winner: Source to choose (koreader, kobo, web, manual) + - manual_data: Required if winner is "manual" - contains the merged data + - apply_to_all_future_conflicts: Whether to auto-resolve future conflicts from this source + - reason: Optional explanation for the resolution + + Example request body for choosing koreader: + { + "winner": "koreader", + "reason": "More recent progress" + } + + Example request body for manual resolution: + { + "winner": "manual", + "manual_data": { + "percentage": 0.43, + "epubcfi": "epubcfi(/6/4/2:20)", + "chapter": 3 + }, + "reason": "Custom merged position" + } + + Response includes: + - conflict_resolved: true if successful + - applied_to: What was updated (progress, annotations) + - devices_synced: List of device IDs that were notified + + After resolution: + - The winning data is applied to the reading progress + - All connected devices are notified via WebSocket + - Conflict status changes to "user_resolved" + - Resolution data is stored for audit trail +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 24022c5..01a7f21 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -59,6 +59,7 @@ func main() { koreaderHandler := handlers.NewKOReaderHandler(queries, connManager) wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware) + conflictHandler := handlers.NewConflictHandler(queries, connManager) e := echo.New() @@ -128,10 +129,20 @@ func main() { // Public library types endpoint (no authentication required) e.GET("/api/libraries/types", libraryHandler.GetLibraryTypes) + // Refresh token endpoint (no authentication required - uses refresh token from body) + e.POST("/api/auth/refresh", authHandler.RefreshAccessToken) + + // Logout endpoint (optional authentication - can revoke tokens if provided) + e.POST("/api/auth/logout", authHandler.Logout) + + // Protected routes + protected = e.Group("/api", jwtMiddleware) + + // Setup ebook handler routes first (so we can use it for library scan) + h = handlers.SetupRoutes(protected, queries, connManager) + protected.GET("/auth/profile", authHandler.GetProfile) protected.PUT("/auth/profile", authHandler.UpdateProfile) - protected.POST("/auth/refresh", authHandler.RefreshAccessToken) - protected.POST("/auth/logout", authHandler.Logout) // Admin-only routes for user and folder management admin := protected.Group("/auth", handlers.AdminMiddleware) @@ -219,6 +230,14 @@ func main() { devices.GET("/approve/:registration_id", deviceHandler.ApproveDevice) devices.POST("/reject/:registration_id", deviceHandler.RejectDevice) + // Conflict resolution routes (protected - require user auth) + conflicts := protected.Group("/conflicts") + conflicts.GET("", conflictHandler.ListConflicts) + conflicts.GET("/:id", conflictHandler.GetConflict) + conflicts.POST("/:id/resolve", conflictHandler.ResolveConflict) + conflicts.DELETE("/:id", conflictHandler.DeleteConflict) + conflicts.POST("/dismiss-all", conflictHandler.DismissAllResolved) + // WebSocket endpoint for real-time sync e.GET("/ws/sync", wsHandler.HandleWebSocket) diff --git a/cmd/server/tests/conflict_test.go b/cmd/server/tests/conflict_test.go new file mode 100644 index 0000000..f3e0cc1 --- /dev/null +++ b/cmd/server/tests/conflict_test.go @@ -0,0 +1,289 @@ +package main + +import ( + "bytes" + "encoding/json" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" +) + +func TestConflictDetection_TriggeringConditions(t *testing.T) { + t.Run("conflict detected when different devices sync within 5 minutes", func(t *testing.T) { + conflictData := map[string]map[string]interface{}{ + "koreader": { + "source": "koreader", + "timestamp": "2026-01-30T20:10:00Z", + "data": map[string]interface{}{ + "percentage": 0.45, + "epubcfi": "epubcfi(/6/4/2:15)", + "chapter": 3, + }, + }, + "kobo": { + "source": "kobo", + "timestamp": "2026-01-30T20:05:00Z", + "data": map[string]interface{}{ + "percentage": 0.42, + "page": 89, + }, + }, + } + + body, err := json.Marshal(conflictData) + assert.NoError(t, err) + + req := httptest.NewRequest("POST", "/api/sync/koreader/progress", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + assert.Equal(t, "POST", req.Method) + assert.Contains(t, string(body), "koreader") + assert.Contains(t, string(body), "kobo") + }) + + t.Run("no conflict when progress difference is less than 1%", func(t *testing.T) { + progressData := map[string]interface{}{ + "percentage": 0.45, + } + + existingProgress := map[string]interface{}{ + "percentage": 0.451, + } + + diff := progressData["percentage"].(float64) - existingProgress["percentage"].(float64) + if diff < 0 { + diff = -diff + } + + assert.Less(t, diff, 0.01, "Should not trigger conflict for small differences") + }) + + t.Run("no conflict when sync timestamps are more than 5 minutes apart", func(t *testing.T) { + timestamp1 := "2026-01-30T20:00:00Z" + timestamp2 := "2026-01-30T20:10:00Z" + + var conflictDetected bool + if timestamp2 > timestamp1 { + conflictDetected = false + } + + assert.False(t, conflictDetected, "Should not trigger conflict for old syncs") + }) +} + +func TestConflictResolution_ChoosingWinner(t *testing.T) { + t.Run("resolve conflict by choosing koreader source", func(t *testing.T) { + conflictID := uuid.New() + + reqBody := map[string]interface{}{ + "winner": "koreader", + "manual_data": nil, + "apply_to_all_future_conflicts": false, + "reason": "More recent progress", + } + + body, err := json.Marshal(reqBody) + assert.NoError(t, err) + + req := httptest.NewRequest("POST", "/api/conflicts/"+conflictID.String()+"/resolve", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + assert.Equal(t, "POST", req.Method) + assert.Contains(t, req.URL.Path, conflictID.String()) + assert.Contains(t, string(body), "koreader") + }) + + t.Run("resolve conflict with manual merge data", func(t *testing.T) { + conflictID := uuid.New() + + manualData := map[string]interface{}{ + "percentage": 0.43, + "epubcfi": "epubcfi(/6/4/2:20)", + "chapter": 3, + "page": 90, + } + + reqBody := map[string]interface{}{ + "winner": "manual", + "manual_data": manualData, + "apply_to_all_future_conflicts": false, + "reason": "Custom merged position", + } + + body, err := json.Marshal(reqBody) + assert.NoError(t, err) + + req := httptest.NewRequest("POST", "/api/conflicts/"+conflictID.String()+"/resolve", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + assert.Contains(t, string(body), "manual") + assert.Contains(t, string(body), "0.43") + }) + + t.Run("error when winner is manual but no manual_data provided", func(t *testing.T) { + conflictID := uuid.New() + + reqBody := map[string]interface{}{ + "winner": "manual", + "manual_data": nil, + "apply_to_all_future_conflicts": false, + "reason": "Test", + } + + body, err := json.Marshal(reqBody) + assert.NoError(t, err) + + req := httptest.NewRequest("POST", "/api/conflicts/"+conflictID.String()+"/resolve", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + assert.Contains(t, string(body), "manual") + }) +} + +func TestConflictListing_Filtering(t *testing.T) { + t.Run("list only unresolved conflicts", func(t *testing.T) { + req := httptest.NewRequest("GET", "/api/conflicts?status=unresolved", nil) + assert.Equal(t, "GET", req.Method) + assert.Contains(t, req.URL.Query().Get("status"), "unresolved") + }) + + t.Run("list all conflicts regardless of status", func(t *testing.T) { + req := httptest.NewRequest("GET", "/api/conflicts?status=all", nil) + assert.Equal(t, "GET", req.Method) + assert.Contains(t, req.URL.Query().Get("status"), "all") + }) + + t.Run("list only resolved conflicts", func(t *testing.T) { + req := httptest.NewRequest("GET", "/api/conflicts?status=user_resolved", nil) + assert.Equal(t, "GET", req.Method) + assert.Contains(t, req.URL.Query().Get("status"), "user_resolved") + }) +} + +func TestConflictResponse_Structure(t *testing.T) { + t.Run("conflict detail response includes all required fields", func(t *testing.T) { + conflictResponse := map[string]interface{}{ + "id": "conflict-uuid-123", + "media_item_id": "book-uuid-456", + "media_item_title": "Test Book Title", + "conflict_type": "progress", + "resolution_status": "unresolved", + "created_at": "2026-01-30T20:10:00Z", + "conflict_data": map[string]interface{}{ + "koreader": map[string]interface{}{ + "source": "koreader", + "data": map[string]interface{}{ + "percentage": 0.45, + }, + }, + "kobo": map[string]interface{}{ + "source": "kobo", + "data": map[string]interface{}{ + "percentage": 0.42, + }, + }, + }, + } + + body, err := json.Marshal(conflictResponse) + assert.NoError(t, err) + + var parsed map[string]interface{} + err = json.Unmarshal(body, &parsed) + assert.NoError(t, err) + + assert.Contains(t, parsed, "id") + assert.Contains(t, parsed, "media_item_id") + assert.Contains(t, parsed, "conflict_data") + assert.Contains(t, parsed["conflict_data"].(map[string]interface{}), "koreader") + assert.Contains(t, parsed["conflict_data"].(map[string]interface{}), "kobo") + }) + + t.Run("conflict list response includes summary counts", func(t *testing.T) { + listResponse := map[string]interface{}{ + "conflicts": []interface{}{ + map[string]string{"id": "conflict-1", "resolution_status": "unresolved"}, + map[string]string{"id": "conflict-2", "resolution_status": "unresolved"}, + }, + "total": 2, + "unresolved": 2, + } + + body, err := json.Marshal(listResponse) + assert.NoError(t, err) + + var parsed map[string]interface{} + err = json.Unmarshal(body, &parsed) + assert.NoError(t, err) + + assert.Equal(t, float64(2), parsed["total"]) + assert.Equal(t, float64(2), parsed["unresolved"]) + }) +} + +func TestConflictDeletion(t *testing.T) { + t.Run("delete single conflict by ID", func(t *testing.T) { + conflictID := uuid.New() + + req := httptest.NewRequest("DELETE", "/api/conflicts/"+conflictID.String(), nil) + assert.Equal(t, "DELETE", req.Method) + assert.Contains(t, req.URL.Path, conflictID.String()) + }) + + t.Run("dismiss all resolved conflicts", func(t *testing.T) { + req := httptest.NewRequest("POST", "/api/conflicts/dismiss-all", nil) + assert.Equal(t, "POST", req.Method) + assert.Contains(t, req.URL.Path, "dismiss-all") + }) +} + +func TestConflictNotification_WebSocketBroadcast(t *testing.T) { + t.Run("conflict detection notification", func(t *testing.T) { + notification := map[string]interface{}{ + "type": "conflict", + "timestamp": "2026-01-30T20:10:00Z", + "data": map[string]interface{}{ + "book_id": "book-uuid-123", + "notification_type": "detection", + "conflict_id": "", + }, + } + + body, err := json.Marshal(notification) + assert.NoError(t, err) + + var parsed map[string]interface{} + err = json.Unmarshal(body, &parsed) + assert.NoError(t, err) + + data := parsed["data"].(map[string]interface{}) + assert.Equal(t, "detection", data["notification_type"]) + }) + + t.Run("conflict resolved notification", func(t *testing.T) { + conflictID := uuid.New() + + notification := map[string]interface{}{ + "type": "conflict", + "timestamp": "2026-01-30T20:15:00Z", + "data": map[string]interface{}{ + "book_id": "book-uuid-123", + "notification_type": "resolved", + "conflict_id": conflictID.String(), + }, + } + + body, err := json.Marshal(notification) + assert.NoError(t, err) + + var parsed map[string]interface{} + err = json.Unmarshal(body, &parsed) + assert.NoError(t, err) + + data := parsed["data"].(map[string]interface{}) + assert.Equal(t, "resolved", data["notification_type"]) + assert.Equal(t, conflictID.String(), data["conflict_id"]) + }) +} diff --git a/internal/database/queries/queries.sql b/internal/database/queries/queries.sql index da8dd31..c2f1fcc 100644 --- a/internal/database/queries/queries.sql +++ b/internal/database/queries/queries.sql @@ -754,6 +754,13 @@ RETURNING *; -- name: DeleteSyncConflict :exec DELETE FROM sync_conflicts WHERE id = $1; +-- name: ListAllConflictsByUserAndStatus :many +SELECT sc.*, mi.title, mi.author +FROM sync_conflicts sc +JOIN media_items mi ON sc.media_item_id = mi.id +WHERE sc.user_id = $1 AND sc.resolution_status = $2 +ORDER BY sc.created_at DESC; + -- ============================================ -- PHASE 3: KOREADER SYNC PROTOCOL (Weeks 7-9) -- ============================================ diff --git a/internal/handlers/conflicts.go b/internal/handlers/conflicts.go new file mode 100644 index 0000000..e9df7fa --- /dev/null +++ b/internal/handlers/conflicts.go @@ -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, + }) +} diff --git a/internal/handlers/koreader.go b/internal/handlers/koreader.go index 2ad9f16..13f14da 100644 --- a/internal/handlers/koreader.go +++ b/internal/handlers/koreader.go @@ -3,11 +3,13 @@ package handlers import ( "bookmann/internal/database" wsync "bookmann/internal/sync" + "encoding/json" "fmt" "net/http" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/labstack/echo/v4" ) @@ -240,6 +242,35 @@ func (h *KOReaderHandler) SyncProgress(c echo.Context) error { } func (h *KOReaderHandler) updateProgressForBook(c echo.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, book KOReaderBookProgress) error { + ctx := c.Request().Context() + + existingProgress, err := h.db.GetReadingProgress(ctx, database.GetReadingProgressParams{ + MediaItemID: mediaItemID, + UserID: userID, + }) + + if err != nil && err != pgx.ErrNoRows { + return err + } + + hasExistingProgress := err != pgx.ErrNoRows + conflictDetected := false + + if hasExistingProgress && existingProgress.LastSyncSource.Valid { + if existingProgress.LastSyncSource.String != "koreader" && existingProgress.LastSyncTimestamp.Valid { + timeDiff := time.Since(existingProgress.LastSyncTimestamp.Time) + if timeDiff < 5*time.Minute { + percentageDiff := book.Percentage - existingProgress.Percentage.Float64 + if percentageDiff < 0 { + percentageDiff = -percentageDiff + } + if percentageDiff > 0.01 { + conflictDetected = true + } + } + } + } + var epubcfi pgtype.Text var chapter pgtype.Int4 var characterOffset pgtype.Int8 @@ -262,7 +293,7 @@ func (h *KOReaderHandler) updateProgressForBook(c echo.Context, userID pgtype.UU totalPages = pgtype.Int4{Int32: int32(*book.TotalPages), Valid: true} } - _, err := h.db.UpdateUniversalProgress(c.Request().Context(), database.UpdateUniversalProgressParams{ + _, err = h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{ MediaItemID: mediaItemID, UserID: userID, Percentage: pgtype.Float8{Float64: book.Percentage, Valid: true}, @@ -274,26 +305,103 @@ func (h *KOReaderHandler) updateProgressForBook(c echo.Context, userID pgtype.UU TotalPages: totalPages, LastSyncDevice: pgtype.Text{String: "koreader", Valid: true}, LastSyncSource: pgtype.Text{String: "koreader", Valid: true}, + ViewportY: pgtype.Float8{}, + ScrollPositionX: pgtype.Float8{}, + ScrollPositionY: pgtype.Float8{}, + PanelNumber: pgtype.Int4{}, + ReadingMode: pgtype.Text{}, + ZoomLevel: pgtype.Float8{}, }) - if err == nil { - // Broadcast progress update to all connected WebSocket clients - deviceInfo := book.DeviceInfo - if deviceInfo.DeviceModel == "" { - deviceInfo.DeviceModel = "KOReader Device" + if err != nil { + return err + } + + if conflictDetected { + koreaderData := map[string]interface{}{ + "source": "koreader", + "timestamp": time.Now(), + "data": map[string]interface{}{ + "percentage": book.Percentage, + }, + } + if book.Epubcfi != nil { + koreaderData["data"].(map[string]interface{})["epubcfi"] = *book.Epubcfi + } + if book.Chapter != nil { + koreaderData["data"].(map[string]interface{})["chapter"] = *book.Chapter + } + if book.Character != nil { + koreaderData["data"].(map[string]interface{})["character"] = *book.Character + } + if book.Page != nil { + koreaderData["data"].(map[string]interface{})["page"] = *book.Page + } + if book.TotalPages != nil { + koreaderData["data"].(map[string]interface{})["total_pages"] = *book.TotalPages } - h.connManager.BroadcastProgressUpdate( - uuid.UUID(mediaItemID.Bytes), - book.Percentage, - wsync.SourceDevice{ - ID: uuid.UUID(userID.Bytes).String(), - Name: deviceInfo.DeviceModel, - Type: "koreader", + existingData := map[string]interface{}{ + "source": existingProgress.LastSyncSource.String, + "timestamp": existingProgress.LastSyncTimestamp.Time, + "data": map[string]interface{}{ + "percentage": existingProgress.Percentage.Float64, }, - ) + } + if existingProgress.Epubcfi.Valid { + existingData["data"].(map[string]interface{})["epubcfi"] = existingProgress.Epubcfi.String + } + if existingProgress.Chapter.Valid { + existingData["data"].(map[string]interface{})["chapter"] = existingProgress.Chapter.Int32 + } + if existingProgress.CharacterOffset.Valid { + existingData["data"].(map[string]interface{})["character"] = existingProgress.CharacterOffset.Int64 + } + if existingProgress.CurrentPage.Valid { + existingData["data"].(map[string]interface{})["page"] = existingProgress.CurrentPage.Int32 + } + if existingProgress.TotalPages.Valid { + existingData["data"].(map[string]interface{})["total_pages"] = existingProgress.TotalPages.Int32 + } + + conflictData := map[string]interface{}{ + "koreader": koreaderData, + "existing": existingData, + } + conflictDataJSON, _ := json.Marshal(conflictData) + + _, err := h.db.CreateSyncConflict(ctx, database.CreateSyncConflictParams{ + MediaItemID: mediaItemID, + UserID: userID, + ConflictType: "progress", + ConflictData: conflictDataJSON, + }) + if err == nil { + h.connManager.BroadcastConflictNotification( + mediaItemID.Bytes, + "detection", + "", + ) + } } + deviceInfo := book.DeviceInfo + if deviceInfo.DeviceModel == "" { + deviceInfo.DeviceModel = "KOReader Device" + } + + h.connManager.BroadcastProgressUpdate( + uuid.UUID(mediaItemID.Bytes), + book.Percentage, + wsync.SourceDevice{ + ID: uuid.UUID(userID.Bytes).String(), + Name: deviceInfo.DeviceModel, + Type: "koreader", + }, + ) + + _, err = h.db.UpdateDeviceLastSync(ctx, pgtype.UUID{Bytes: [16]byte{}, Valid: false}) + return err } diff --git a/internal/sync/websocket.go b/internal/sync/websocket.go index 2451b84..5457f4b 100644 --- a/internal/sync/websocket.go +++ b/internal/sync/websocket.go @@ -121,6 +121,20 @@ func (m *ConnectionManager) BroadcastAnnotationUpdate(bookID uuid.UUID, annotati m.Broadcast(msg) } +// BroadcastConflictNotification broadcasts a conflict notification to all connected clients +func (m *ConnectionManager) BroadcastConflictNotification(bookID [16]byte, notificationType string, conflictID string) { + msg := BroadcastMessage{ + Type: MessageTypeConflict, + Timestamp: time.Now().Format(time.RFC3339), + Data: map[string]interface{}{ + "book_id": uuid.UUID(bookID).String(), + "notification_type": notificationType, + "conflict_id": conflictID, + }, + } + m.Broadcast(msg) +} + // AddConnection adds a new WebSocket connection func (m *ConnectionManager) AddConnection(conn *DeviceConnection) { m.mu.Lock()