refactor(progress): ProgressService required at construction; delete the 9 unreachable erase-fallbacks

Batch H1 step 3a. The ceiling census (exhaustive UpdateUniversalProgress
grep) found 9 nil-guarded `if progressSvc != nil { funnel } else { raw
erase-write }` fallbacks — the survey's "four" under-counted kobo.go,
which has FOUR sites (Markup ReadingSync, Markup last-read-place,
AnalyticsGettests, SyncFromServer), plus one each in media.go, queue.go
and koreader.go. All unreachable: the service is created unconditionally
at main.go:127 and wired into every handler and the queue processor; the
test harness wires them too.

The erasure pattern is now unrepresentable:
- NewMediaHandler / NewKOReaderHandler / NewKoboHandler / both
  NewSyncQueueProcessor flavors take the ProgressService as a REQUIRED
  constructor arg; the SetProgressService setters are gone.
- Every fallback body (raw UpdateUniversalProgress with explicit-NULL
  pgtypes for viewport/scroll/panel/reading_mode/zoom and no
  audio_offset_seconds/context_text) is deleted; each write funnels
  through SaveProgress with the LWW seed, so fields the writer doesn't
  know survive.
- Removed the now-dead koreader pg-type helpers (textPtrToPgText,
  intPtrToPgInt4, int64PtrToPgInt8) and their self-tests.

Behavior deltas (stored-row content, no wire shape changes —
KOReader/Kobo response pins stayed green):
- KOReader fallback used to write explicit NULL viewports and drop the
  context_text the koplugin sends; the funneled mainline already maps
  book.ContextText.
- Queue flushes no longer overwrite chapter_progress with the book
  percentage (bogus data; preserved via the seed).
- A web PUT without percentage no longer writes 0% via the deleted
  fallback; percentage seeds from the stored row.

Contract suite: KOReader×4 + Kobo×3 pins green.
This commit is contained in:
John O'Keefe committed 2026-10-16 17:28:06 -04:00
1 parent 28bd323505
commit e83c00f869
9 files changed
+163 -419

No files matched your search

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