diff --git a/cmd/server/main.go b/cmd/server/main.go index 7eb991f6..30ba500b 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -130,8 +130,7 @@ func main() { maintenanceCancel := annotationService.StartDailyMaintenance() defer maintenanceCancel() - queueProcessor := sync.NewSyncQueueProcessorWithConfig(queries, registry.SyncQueueConfig().Interval, registry.SyncQueueConfig().BatchSize) - queueProcessor.SetProgressService(progressService) + queueProcessor := sync.NewSyncQueueProcessorWithConfig(queries, progressService, registry.SyncQueueConfig().Interval, registry.SyncQueueConfig().BatchSize) queueProcessor.SetAnnotationService(annotationService) // Create library service @@ -148,8 +147,7 @@ func main() { worker := services.NewWorkerWithConfig(workerCfg.Size, workerCfg.QueueCap, connManager) services.WorkerInstance = worker - koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor) - koreaderHandler.SetProgressService(progressService) + koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor, progressService) koreaderHandler.SetAnnotationService(annotationService) koreaderHandler.SetLibraryService(libraryService) wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware) @@ -169,8 +167,7 @@ func main() { dashboardHandler := handlers.NewDashboardHandler(queries) seriesHandler := handlers.NewSeriesHandler(queries) filtersHandler := handlers.NewFiltersHandler(queries) - mediaHandler := handlers.NewMediaHandler(queries, libraryService, worker) - mediaHandler.SetProgressService(progressService) + mediaHandler := handlers.NewMediaHandler(queries, libraryService, progressService) mediaHandler.SetAnnotationService(annotationService) matchingHandler := handlers.NewMatchingHandler(queries, connManager) jobsHandler := handlers.NewJobsHandler(queries, worker) diff --git a/cmd/server/tests/test_helpers_test.go b/cmd/server/tests/test_helpers_test.go index 9863fe83..c10de780 100644 --- a/cmd/server/tests/test_helpers_test.go +++ b/cmd/server/tests/test_helpers_test.go @@ -483,13 +483,11 @@ func setupTestServer(t *testing.T) *TestServerSetup { progressService := wsync.NewProgressService(queries, connManager) annotationService := wsync.NewAnnotationService(queries, connManager) - queueProcessor := wsync.NewSyncQueueProcessor(queries) - queueProcessor.SetProgressService(progressService) + queueProcessor := wsync.NewSyncQueueProcessor(queries, progressService) queueCtx, queueCancel := context.WithCancel(context.Background()) go queueProcessor.Start(queueCtx) - koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor) - koreaderHandler.SetProgressService(progressService) + koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor, progressService) koreaderHandler.SetAnnotationService(annotationService) wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware) conflictHandler := handlers.NewConflictHandler(queries, connManager) @@ -510,8 +508,7 @@ func setupTestServer(t *testing.T) *TestServerSetup { statisticsService := services.NewStatisticsService(queries) statsHandler := handlers.NewStatsHandler(statisticsService) seriesHandler := handlers.NewSeriesHandler(queries) - mediaHandler := handlers.NewMediaHandler(queries, libraryService, worker) - mediaHandler.SetProgressService(progressService) + mediaHandler := handlers.NewMediaHandler(queries, libraryService, progressService, worker) mediaHandler.SetAnnotationService(annotationService) matchingHandler := handlers.NewMatchingHandler(queries, connManager) diff --git a/internal/handlers/kobo.go b/internal/handlers/kobo.go index 155e11f3..3ddb4477 100644 --- a/internal/handlers/kobo.go +++ b/internal/handlers/kobo.go @@ -26,12 +26,8 @@ type KoboHandler struct { bookResolver *services.BookResolver } -func NewKoboHandler(db *database.Queries, connManager *wsync.ConnectionManager) *KoboHandler { - return &KoboHandler{db: db, connManager: connManager, bookResolver: services.NewBookResolver(db)} -} - -func (h *KoboHandler) SetProgressService(svc *wsync.ProgressService) { - h.progressSvc = svc +func NewKoboHandler(db *database.Queries, connManager *wsync.ConnectionManager, progressSvc *wsync.ProgressService) *KoboHandler { + return &KoboHandler{db: db, connManager: connManager, progressSvc: progressSvc, bookResolver: services.NewBookResolver(db)} } func (h *KoboHandler) SetAnnotationService(svc *wsync.AnnotationService) { @@ -446,29 +442,19 @@ func (h *KoboHandler) Markup(c *echo.Context) error { } } - if h.progressSvc != nil { - _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Source: "kobo", - DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, - Percentage: &percentage, - CurrentPage: currentPage, - TotalPages: totalPages, - DeviceType: "kobo", - DeviceName: device.DeviceName, - Broadcast: true, - ClearLocator: true, - }) - } else { - _, err = h.db.UpdateUniversalProgress(c.Request().Context(), database.UpdateUniversalProgressParams{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Percentage: pgtype.Float8{Float64: percentage, Valid: true}, - LastSyncDevice: pgtype.Text{String: "kobo", Valid: true}, - LastSyncSource: pgtype.Text{String: "kobo", Valid: true}, - }) - } + _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ + MediaItemID: pgMediaUUID, + UserID: pgUserID, + Source: "kobo", + DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, + Percentage: &percentage, + CurrentPage: currentPage, + TotalPages: totalPages, + DeviceType: "kobo", + DeviceName: device.DeviceName, + Broadcast: true, + ClearLocator: true, + }) if err == nil { markupsSynced++ @@ -580,31 +566,19 @@ func (h *KoboHandler) Markup(c *echo.Context) error { epubcfi = *convertedCFI } - if h.progressSvc != nil { - _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Source: "kobo", - DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, - Epubcfi: &epubcfi, - ContextText: contextText, - Chapter: &chapter, - ChapterProgress: &chapterProgress, - DeviceType: "kobo", - DeviceName: device.DeviceName, - Broadcast: false, - }) - } else { - _, err = h.db.UpdateUniversalProgress(c.Request().Context(), database.UpdateUniversalProgressParams{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Epubcfi: pgtype.Text{String: epubcfi, Valid: epubcfi != ""}, - Chapter: pgtype.Int4{Int32: int32(bookmarkSync.Chapter), Valid: true}, - ChapterProgress: pgtype.Float8{Float64: 0.5, Valid: true}, - LastSyncDevice: pgtype.Text{String: "kobo", Valid: true}, - LastSyncSource: pgtype.Text{String: "kobo", Valid: true}, - }) - } + _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ + MediaItemID: pgMediaUUID, + UserID: pgUserID, + Source: "kobo", + DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, + Epubcfi: &epubcfi, + ContextText: contextText, + Chapter: &chapter, + ChapterProgress: &chapterProgress, + DeviceType: "kobo", + DeviceName: device.DeviceName, + Broadcast: false, + }) if err != nil { fmt.Printf("Failed to store last-read-place: %v", err) } @@ -797,27 +771,17 @@ func (h *KoboHandler) AnalyticsGettests(c *echo.Context) error { pgMediaUUID := pgtype.UUID{Bytes: bookhoardUUID, Valid: true} percentage := test.PercentRead / 100.0 - if h.progressSvc != nil { - _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Source: "kobo", - DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, - Percentage: &percentage, - DeviceType: "kobo", - DeviceName: device.DeviceName, - Broadcast: true, - ClearLocator: true, - }) - } else { - _, err = h.db.UpdateUniversalProgress(c.Request().Context(), database.UpdateUniversalProgressParams{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Percentage: pgtype.Float8{Float64: percentage, Valid: true}, - LastSyncDevice: pgtype.Text{String: "kobo", Valid: true}, - LastSyncSource: pgtype.Text{String: "kobo", Valid: true}, - }) - } + _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ + MediaItemID: pgMediaUUID, + UserID: pgUserID, + Source: "kobo", + DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, + Percentage: &percentage, + DeviceType: "kobo", + DeviceName: device.DeviceName, + Broadcast: true, + ClearLocator: true, + }) } _, err := h.db.UpdateDeviceLastSync(c.Request().Context(), device.ID) @@ -860,27 +824,17 @@ func (h *KoboHandler) SyncFromServer(c *echo.Context) error { pgMediaUUID := pgtype.UUID{Bytes: bookhoardUUID, Valid: true} percentage := syncData.PercentRead / 100.0 - if h.progressSvc != nil { - _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Source: "bookhoard", - DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, - Percentage: &percentage, - DeviceType: "kobo", - DeviceName: device.DeviceName, - Broadcast: false, - ClearLocator: true, - }) - } else { - _, err = h.db.UpdateUniversalProgress(c.Request().Context(), database.UpdateUniversalProgressParams{ - MediaItemID: pgMediaUUID, - UserID: pgUserID, - Percentage: pgtype.Float8{Float64: percentage, Valid: true}, - LastSyncDevice: pgtype.Text{String: "kobo", Valid: true}, - LastSyncSource: pgtype.Text{String: "bookhoard", Valid: true}, - }) - } + _, err = h.progressSvc.SaveProgress(c.Request().Context(), wsync.SaveProgressRequest{ + MediaItemID: pgMediaUUID, + UserID: pgUserID, + Source: "bookhoard", + DeviceID: pgtype.UUID{Bytes: deviceID, Valid: true}, + Percentage: &percentage, + DeviceType: "kobo", + DeviceName: device.DeviceName, + Broadcast: false, + ClearLocator: true, + }) if err == nil { booksSynced++ diff --git a/internal/handlers/koreader.go b/internal/handlers/koreader.go index f2d63626..4532ed2d 100644 --- a/internal/handlers/koreader.go +++ b/internal/handlers/koreader.go @@ -36,19 +36,16 @@ type LibraryPathResolver interface { ResolveMediaPath(ctx context.Context, libraryID pgtype.UUID, relativePath string) (string, error) } -func NewKOReaderHandler(db *database.Queries, connManager *wsync.ConnectionManager, queue *wsync.SyncQueueProcessor) *KOReaderHandler { +func NewKOReaderHandler(db *database.Queries, connManager *wsync.ConnectionManager, queue *wsync.SyncQueueProcessor, progressSvc *wsync.ProgressService) *KOReaderHandler { return &KOReaderHandler{ db: db, connManager: connManager, queue: queue, + progressSvc: progressSvc, bookResolver: services.NewBookResolver(db), } } -func (h *KOReaderHandler) SetProgressService(svc *wsync.ProgressService) { - h.progressSvc = svc -} - func (h *KOReaderHandler) SetAnnotationService(svc *wsync.AnnotationService) { h.annotationSvc = svc } @@ -878,138 +875,77 @@ func (h *KOReaderHandler) updateProgressForBook(c *echo.Context, deviceID pgtype deviceModel = "KOReader Device" } - if h.progressSvc != nil { - epubcfi := book.Epubcfi - // No position pointer from the device, or a conversion that - // declined below: the write is percentage-only by intent. The - // service then clears any inherited stale anchor instead of - // re-pairing it with the new percentage. - clearLocator := epubcfi == nil - if epubcfi != nil && wsync.IsCREXPointer(*epubcfi) { - // Same facade every annotation uses: cached converter, - // structural-first resolution, guarded fallbacks. The facade - // passes non-reflowable formats through untouched. - ec := h.loadAnnotationEpub(ctx, mediaItemID) - if ec.convertible() { - pct := 0.0 - if book.Percentage >= 0 { - pct = book.Percentage - } - contextText := "" - if book.ContextText != nil { - contextText = *book.ContextText - } - loc := wsync.ConvertToCanonical(wsync.LocatorSourceKOReader, *epubcfi, pct, contextText, ec.mediaItem.FormatGroup, ec.epubPath, "") - switch { - case strings.HasPrefix(loc.CFI, "epubcfi(") && - (loc.Precision == "structural" || loc.Precision == "exact"): - converted := loc.CFI - epubcfi = &converted - log.Printf("Bookhoard: CRE→CFI converted progress (%s) to %s", loc.Precision, converted) - case loc.CFI != "" && loc.CFI != *epubcfi && - (loc.Precision == "element" || loc.Precision == "section"): - // Fragment-ID positions resolve to a section href: keep - // serving it (legacy behavior). - converted := loc.CFI - epubcfi = &converted - log.Printf("Bookhoard: CRE→CFI converted progress (%s) to href %s", loc.Precision, converted) - default: - // Low-confidence (percentage/fallback): store no - // canonical locator — the row's percentage restores - // approximately instead of a confidently-wrong CFI, - // and the device keeps its own native position. - log.Printf("Bookhoard: CRE→CFI conversion low-confidence (%s) for %s; storing percentage only", loc.Precision, *epubcfi) - epubcfi = nil - clearLocator = true - } + epubcfi := book.Epubcfi + // No position pointer from the device, or a conversion that + // declined below: the write is percentage-only by intent. The + // service then clears any inherited stale anchor instead of + // re-pairing it with the new percentage. + clearLocator := epubcfi == nil + if epubcfi != nil && wsync.IsCREXPointer(*epubcfi) { + // Same facade every annotation uses: cached converter, + // structural-first resolution, guarded fallbacks. The facade + // passes non-reflowable formats through untouched. + ec := h.loadAnnotationEpub(ctx, mediaItemID) + if ec.convertible() { + pct := 0.0 + if book.Percentage >= 0 { + pct = book.Percentage + } + contextText := "" + if book.ContextText != nil { + contextText = *book.ContextText + } + loc := wsync.ConvertToCanonical(wsync.LocatorSourceKOReader, *epubcfi, pct, contextText, ec.mediaItem.FormatGroup, ec.epubPath, "") + switch { + case strings.HasPrefix(loc.CFI, "epubcfi(") && + (loc.Precision == "structural" || loc.Precision == "exact"): + converted := loc.CFI + epubcfi = &converted + log.Printf("Bookhoard: CRE→CFI converted progress (%s) to %s", loc.Precision, converted) + case loc.CFI != "" && loc.CFI != *epubcfi && + (loc.Precision == "element" || loc.Precision == "section"): + // Fragment-ID positions resolve to a section href: keep + // serving it (legacy behavior). + converted := loc.CFI + epubcfi = &converted + log.Printf("Bookhoard: CRE→CFI converted progress (%s) to href %s", loc.Precision, converted) + default: + // Low-confidence (percentage/fallback): store no + // canonical locator — the row's percentage restores + // approximately instead of a confidently-wrong CFI, + // and the device keeps its own native position. + log.Printf("Bookhoard: CRE→CFI conversion low-confidence (%s) for %s; storing percentage only", loc.Precision, *epubcfi) + epubcfi = nil + clearLocator = true } } - - saveReq := wsync.SaveProgressRequest{ - MediaItemID: mediaItemID, - UserID: userID, - Source: "koreader", - DeviceID: deviceID, - Percentage: &book.Percentage, - Epubcfi: epubcfi, - ContextText: book.ContextText, - Chapter: book.Chapter, - CharacterOffset: book.Character, - CurrentPage: book.Page, - TotalPages: book.TotalPages, - DeviceType: "koreader", - DeviceName: deviceModel, - Broadcast: true, - ClearLocator: clearLocator, - } - - _, err := h.progressSvc.SaveProgress(ctx, saveReq) - if err != nil { - return err - } - h.processBookAnnotations(ctx, deviceID, userID, mediaItemID, book) - return nil } - _, err := h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{ + saveReq := wsync.SaveProgressRequest{ MediaItemID: mediaItemID, UserID: userID, - Percentage: pgtype.Float8{Float64: book.Percentage, Valid: true}, - Epubcfi: textPtrToPgText(book.Epubcfi), - Chapter: intPtrToPgInt4(book.Chapter), - ChapterProgress: pgtype.Float8{Float64: book.Percentage, Valid: true}, - CharacterOffset: int64PtrToPgInt8(book.Character), - CurrentPage: intPtrToPgInt4(book.Page), - TotalPages: intPtrToPgInt4(book.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 { + Source: "koreader", + DeviceID: deviceID, + Percentage: &book.Percentage, + Epubcfi: epubcfi, + ContextText: book.ContextText, + Chapter: book.Chapter, + CharacterOffset: book.Character, + CurrentPage: book.Page, + TotalPages: book.TotalPages, + DeviceType: "koreader", + DeviceName: deviceModel, + Broadcast: true, + ClearLocator: clearLocator, + } + + if _, err := h.progressSvc.SaveProgress(ctx, saveReq); err != nil { return err } - - h.connManager.BroadcastProgressUpdate( - mediaItemID.Bytes, - book.Percentage, - wsync.SourceDevice{ - ID: uuid.UUID(deviceID.Bytes).String(), - Name: deviceModel, - Type: "koreader", - }, - ) - h.processBookAnnotations(ctx, deviceID, userID, mediaItemID, book) - return nil } -func textPtrToPgText(s *string) pgtype.Text { - if s != nil { - return pgtype.Text{String: *s, Valid: true} - } - return pgtype.Text{} -} - -func intPtrToPgInt4(i *int) pgtype.Int4 { - if i != nil { - return pgtype.Int4{Int32: int32(*i), Valid: true} - } - return pgtype.Int4{} -} - -func int64PtrToPgInt8(i *int64) pgtype.Int8 { - if i != nil { - return pgtype.Int8{Int64: *i, Valid: true} - } - return pgtype.Int8{} -} - type KOReaderResolveResponse struct { BookUUID string `json:"book_uuid"` SHA256 string `json:"sha256,omitempty"` diff --git a/internal/handlers/media.go b/internal/handlers/media.go index 1c53149b..451e14ff 100644 --- a/internal/handlers/media.go +++ b/internal/handlers/media.go @@ -180,11 +180,15 @@ type MediaHandler struct { annotationSvc *wsync.AnnotationService } -func NewMediaHandler(db *database.Queries, libraryService *services.LibraryService, worker ...*services.Worker) *MediaHandler { +// NewMediaHandler requires the shared ProgressService — progress writes +// must funnel through SaveProgress (the LWW-seeded canonical writer); +// a nil service has nowhere to fall back to and must not be wired. +func NewMediaHandler(db *database.Queries, libraryService *services.LibraryService, progressSvc *wsync.ProgressService, worker ...*services.Worker) *MediaHandler { mh := &MediaHandler{ db: db, libraryService: libraryService, searchService: services.NewSearchService(db), + progressSvc: progressSvc, } if len(worker) > 0 && worker[0] != nil { mh.worker = worker[0] @@ -192,10 +196,6 @@ func NewMediaHandler(db *database.Queries, libraryService *services.LibraryServi return mh } -func (mh *MediaHandler) SetProgressService(svc *wsync.ProgressService) { - mh.progressSvc = svc -} - func (mh *MediaHandler) SetAnnotationService(svc *wsync.AnnotationService) { mh.annotationSvc = svc } @@ -1281,66 +1281,21 @@ func (mh *MediaHandler) UpdateMediaReadingProgress(c *echo.Context) error { req.DeviceName = deviceLabel req.Broadcast = true - if mh.progressSvc != nil { - if req.Percentage != nil && *req.Percentage < 0.005 { - existing, err := mh.db.GetUniversalProgress(c.Request().Context(), database.GetUniversalProgressParams{ - MediaItemID: pgtype.UUID{Bytes: mediaUUID, Valid: true}, - UserID: pgtype.UUID{Bytes: userUUID, Valid: true}, - }) - if err == nil && existing.Percentage.Valid && existing.Percentage.Float64 > 0.01 { - return c.JSON(http.StatusOK, map[string]string{"status": "ignored"}) - } + if req.Percentage != nil && *req.Percentage < 0.005 { + existing, err := mh.db.GetUniversalProgress(c.Request().Context(), database.GetUniversalProgressParams{ + MediaItemID: pgtype.UUID{Bytes: mediaUUID, Valid: true}, + UserID: pgtype.UUID{Bytes: userUUID, Valid: true}, + }) + if err == nil && existing.Percentage.Valid && existing.Percentage.Float64 > 0.01 { + return c.JSON(http.StatusOK, map[string]string{"status": "ignored"}) } - - result, err := mh.progressSvc.SaveProgress(c.Request().Context(), req) - if err != nil { - return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()}) - } - return c.JSON(http.StatusOK, result) } - percentage := 0.0 - if req.Percentage != nil { - percentage = *req.Percentage - } - epubcfi := "" - if req.Epubcfi != nil { - epubcfi = *req.Epubcfi - } - currentPage := int32(0) - if req.CurrentPage != nil { - currentPage = int32(*req.CurrentPage) - } - totalPages := int32(0) - if req.TotalPages != nil { - totalPages = int32(*req.TotalPages) - } - - progress, err := mh.db.UpdateUniversalProgress(c.Request().Context(), database.UpdateUniversalProgressParams{ - MediaItemID: pgtype.UUID{Bytes: mediaUUID, Valid: true}, - UserID: pgtype.UUID{Bytes: userUUID, Valid: true}, - Percentage: pgtype.Float8{Float64: percentage, Valid: true}, - CharacterOffset: pgtype.Int8{Valid: false}, - Epubcfi: pgtype.Text{String: epubcfi, Valid: epubcfi != ""}, - Chapter: pgtype.Int4{Valid: false}, - ChapterProgress: pgtype.Float8{Valid: false}, - ViewportX: pgtype.Float8{Valid: false}, - ViewportY: pgtype.Float8{Valid: false}, - ZoomLevel: pgtype.Float8{Valid: false}, - ScrollPositionX: pgtype.Float8{Valid: false}, - ScrollPositionY: pgtype.Float8{Valid: false}, - PanelNumber: pgtype.Int4{Valid: false}, - ReadingMode: pgtype.Text{Valid: false}, - LastSyncDevice: pgtype.Text{String: "web", Valid: true}, - LastSyncSource: pgtype.Text{String: "web", Valid: true}, - CurrentPage: pgtype.Int4{Int32: currentPage, Valid: true}, - TotalPages: pgtype.Int4{Int32: totalPages, Valid: totalPages > 0}, - }) + result, err := mh.progressSvc.SaveProgress(c.Request().Context(), req) if err != nil { return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()}) } - - return c.JSON(http.StatusOK, progress) + return c.JSON(http.StatusOK, result) } // DeleteMediaReadingProgress handles DELETE /api/media-items/:id/progress diff --git a/internal/handlers/progress_test.go b/internal/handlers/progress_test.go index fa754a00..453494a0 100644 --- a/internal/handlers/progress_test.go +++ b/internal/handlers/progress_test.go @@ -30,51 +30,6 @@ func TestGetDeviceIcon_Unknown(t *testing.T) { assert.Equal(t, "📚", result) } -func TestTextPtrToPgText(t *testing.T) { - t.Run("nil returns invalid", func(t *testing.T) { - result := textPtrToPgText(nil) - assert.False(t, result.Valid) - }) - t.Run("non-nil returns valid", func(t *testing.T) { - s := "epubcfi(/6/4/2:10)" - result := textPtrToPgText(&s) - assert.True(t, result.Valid) - assert.Equal(t, s, result.String) - }) - t.Run("empty string returns valid", func(t *testing.T) { - s := "" - result := textPtrToPgText(&s) - assert.True(t, result.Valid) - assert.Equal(t, "", result.String) - }) -} - -func TestIntPtrToPgInt4(t *testing.T) { - t.Run("nil returns invalid", func(t *testing.T) { - result := intPtrToPgInt4(nil) - assert.False(t, result.Valid) - }) - t.Run("non-nil returns valid", func(t *testing.T) { - v := 5 - result := intPtrToPgInt4(&v) - assert.True(t, result.Valid) - assert.Equal(t, int32(5), result.Int32) - }) -} - -func TestInt64PtrToPgInt8(t *testing.T) { - t.Run("nil returns invalid", func(t *testing.T) { - result := int64PtrToPgInt8(nil) - assert.False(t, result.Valid) - }) - t.Run("non-nil returns valid", func(t *testing.T) { - v := int64(10000) - result := int64PtrToPgInt8(&v) - assert.True(t, result.Valid) - assert.Equal(t, int64(10000), result.Int64) - }) -} - func TestFloat64PtrHelpers(t *testing.T) { t.Run("pgtype float64 valid", func(t *testing.T) { v := pgtype.Float8{Float64: 0.5, Valid: true} diff --git a/internal/router/sync.go b/internal/router/sync.go index cf62c2e9..bafe94bc 100644 --- a/internal/router/sync.go +++ b/internal/router/sync.go @@ -37,8 +37,7 @@ func registerSyncRoutes(cfg *Config) { // Kobo sync routes (device authentication required) // Kobo devices use URL path: /api/sync/kobo/{token}/markup // API clients can use Authorization header: Authorization: Bearer {token} - koboHandler := handlers.NewKoboHandler(cfg.Queries, cfg.ConnManager) - koboHandler.SetProgressService(cfg.ProgressService) + koboHandler := handlers.NewKoboHandler(cfg.Queries, cfg.ConnManager, cfg.ProgressService) koboHandler.SetAnnotationService(cfg.AnnotationService) koboHandler.SetLibraryService(cfg.LibraryService) koboSync := e.Group("/api/sync/kobo/:token") diff --git a/internal/sync/queue.go b/internal/sync/queue.go index 6f7bf152..a0601018 100644 --- a/internal/sync/queue.go +++ b/internal/sync/queue.go @@ -73,26 +73,26 @@ type SyncQueueItem struct { ProcessedAt *time.Time } -func NewSyncQueueProcessor(db *database.Queries) *SyncQueueProcessor { - return NewSyncQueueProcessorWithConfig(db, 5*time.Second, 50) +// NewSyncQueueProcessor requires the shared ProgressService — queue +// progress entries flush through SaveProgress (the LWW-seeded canonical +// writer); a nil service has nowhere to fall back to and must not be wired. +func NewSyncQueueProcessor(db *database.Queries, progressSvc *ProgressService) *SyncQueueProcessor { + return NewSyncQueueProcessorWithConfig(db, progressSvc, 5*time.Second, 50) } // NewSyncQueueProcessorWithConfig constructs a processor with the given flush // interval and batch size. Used at startup to source values from the settings // registry. -func NewSyncQueueProcessorWithConfig(db *database.Queries, interval time.Duration, batchSize int) *SyncQueueProcessor { +func NewSyncQueueProcessorWithConfig(db *database.Queries, progressSvc *ProgressService, interval time.Duration, batchSize int) *SyncQueueProcessor { return &SyncQueueProcessor{ db: db, + progressSvc: progressSvc, progressChan: make(chan *ProgressUpdate, 100), interval: interval, batchSize: batchSize, } } -func (p *SyncQueueProcessor) SetProgressService(svc *ProgressService) { - p.progressSvc = svc -} - func (p *SyncQueueProcessor) SetAnnotationService(svc *AnnotationService) { p.annotationSvc = svc } @@ -432,89 +432,40 @@ func (p *SyncQueueProcessor) syncProgress(ctx context.Context, userID pgtype.UUI source = v } - if p.progressSvc != nil { - req := SaveProgressRequest{ - MediaItemID: mediaItemID, - UserID: userID, - Source: source, - Percentage: &percentage, - Broadcast: false, - } - if v, ok := syncData["epubcfi"].(string); ok { - req.Epubcfi = &v - } - // Entries without a CFI are percentage-only by intent: declare it - // so the service clears any inherited stale anchor instead of - // re-pairing it with this percentage. - req.ClearLocator = req.Epubcfi == nil - if v, ok := syncData["context_text"].(string); ok { - req.ContextText = &v - } - if v, ok := syncData["chapter"].(float64); ok { - ch := int(v) - req.Chapter = &ch - } - if v, ok := syncData["character"].(float64); ok { - co := int64(v) - req.CharacterOffset = &co - } - if v, ok := syncData["page"].(float64); ok { - pg := int(v) - req.CurrentPage = &pg - } - if v, ok := syncData["total_pages"].(float64); ok { - tp := int(v) - req.TotalPages = &tp - } - _, err := p.progressSvc.SaveProgress(ctx, req) - return err + req := SaveProgressRequest{ + MediaItemID: mediaItemID, + UserID: userID, + Source: source, + Percentage: &percentage, + Broadcast: false, } - - var epubcfi pgtype.Text if v, ok := syncData["epubcfi"].(string); ok { - epubcfi = pgtype.Text{String: v, Valid: true} + req.Epubcfi = &v + } + // Entries without a CFI are percentage-only by intent: declare it + // so the service clears any inherited stale anchor instead of + // re-pairing it with this percentage. + req.ClearLocator = req.Epubcfi == nil + if v, ok := syncData["context_text"].(string); ok { + req.ContextText = &v } - - var chapter pgtype.Int4 if v, ok := syncData["chapter"].(float64); ok { - chapter = pgtype.Int4{Int32: int32(v), Valid: true} + ch := int(v) + req.Chapter = &ch } - - var characterOffset pgtype.Int8 if v, ok := syncData["character"].(float64); ok { - characterOffset = pgtype.Int8{Int64: int64(v), Valid: true} + co := int64(v) + req.CharacterOffset = &co } - - var currentPage pgtype.Int4 if v, ok := syncData["page"].(float64); ok { - currentPage = pgtype.Int4{Int32: int32(v), Valid: true} + pg := int(v) + req.CurrentPage = &pg } - - var totalPages pgtype.Int4 if v, ok := syncData["total_pages"].(float64); ok { - totalPages = pgtype.Int4{Int32: int32(v), Valid: true} + tp := int(v) + req.TotalPages = &tp } - - _, err := p.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: source, Valid: true}, - LastSyncSource: pgtype.Text{String: source, Valid: true}, - ViewportX: pgtype.Float8{}, - ScrollPositionX: pgtype.Float8{}, - ScrollPositionY: pgtype.Float8{}, - PanelNumber: pgtype.Int4{}, - ReadingMode: pgtype.Text{}, - ZoomLevel: pgtype.Float8{}, - }) - + _, err := p.progressSvc.SaveProgress(ctx, req) return err } diff --git a/internal/sync/queue_test.go b/internal/sync/queue_test.go index f916e0e0..669f936f 100644 --- a/internal/sync/queue_test.go +++ b/internal/sync/queue_test.go @@ -8,7 +8,7 @@ import ( ) func TestCalculateNextRetry(t *testing.T) { - processor := NewSyncQueueProcessor(nil) + processor := NewSyncQueueProcessor(nil, nil) tests := []struct { name string