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:
1 parent
28bd323505
commit
e83c00f869
9 files changed
+25
-281
No files matched your search
+3
-6
@@ -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)
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|
||||||
|
|||||||
@@ -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,7 +442,6 @@ 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,
|
||||||
@@ -460,15 +455,6 @@ func (h *KoboHandler) Markup(c *echo.Context) error {
|
|||||||
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,7 +566,6 @@ 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,
|
||||||
@@ -594,17 +579,6 @@ func (h *KoboHandler) Markup(c *echo.Context) error {
|
|||||||
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,7 +771,6 @@ 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,
|
||||||
@@ -809,15 +782,6 @@ func (h *KoboHandler) AnalyticsGettests(c *echo.Context) error {
|
|||||||
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,7 +824,6 @@ 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,
|
||||||
@@ -872,15 +835,6 @@ func (h *KoboHandler) SyncFromServer(c *echo.Context) error {
|
|||||||
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++
|
||||||
|
|||||||
@@ -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,7 +875,6 @@ 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
|
||||||
@@ -943,71 +939,11 @@ func (h *KOReaderHandler) updateProgressForBook(c *echo.Context, deviceID pgtype
|
|||||||
ClearLocator: clearLocator,
|
ClearLocator: clearLocator,
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err := h.progressSvc.SaveProgress(ctx, saveReq)
|
if _, err := h.progressSvc.SaveProgress(ctx, saveReq); err != nil {
|
||||||
if err != nil {
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
h.processBookAnnotations(ctx, deviceID, userID, mediaItemID, book)
|
h.processBookAnnotations(ctx, deviceID, userID, mediaItemID, book)
|
||||||
return nil
|
return nil
|
||||||
}
|
|
||||||
|
|
||||||
_, err := h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{
|
|
||||||
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 {
|
|
||||||
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 {
|
type KOReaderResolveResponse struct {
|
||||||
|
|||||||
@@ -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,7 +1281,6 @@ 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},
|
||||||
@@ -1297,50 +1296,6 @@ func (mh *MediaHandler) UpdateMediaReadingProgress(c *echo.Context) error {
|
|||||||
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, 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},
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
|
||||||
}
|
|
||||||
|
|
||||||
return c.JSON(http.StatusOK, progress)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// DeleteMediaReadingProgress handles DELETE /api/media-items/:id/progress
|
// DeleteMediaReadingProgress handles DELETE /api/media-items/:id/progress
|
||||||
|
|||||||
@@ -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}
|
||||||
|
|||||||
@@ -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")
|
||||||
|
|||||||
+7
-56
@@ -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,7 +432,6 @@ 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,
|
||||||
@@ -467,54 +466,6 @@ func (p *SyncQueueProcessor) syncProgress(ctx context.Context, userID pgtype.UUI
|
|||||||
req.TotalPages = &tp
|
req.TotalPages = &tp
|
||||||
}
|
}
|
||||||
_, err := p.progressSvc.SaveProgress(ctx, req)
|
_, err := p.progressSvc.SaveProgress(ctx, req)
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
var epubcfi pgtype.Text
|
|
||||||
if v, ok := syncData["epubcfi"].(string); ok {
|
|
||||||
epubcfi = pgtype.Text{String: v, Valid: true}
|
|
||||||
}
|
|
||||||
|
|
||||||
var chapter pgtype.Int4
|
|
||||||
if v, ok := syncData["chapter"].(float64); ok {
|
|
||||||
chapter = pgtype.Int4{Int32: int32(v), Valid: true}
|
|
||||||
}
|
|
||||||
|
|
||||||
var characterOffset pgtype.Int8
|
|
||||||
if v, ok := syncData["character"].(float64); ok {
|
|
||||||
characterOffset = pgtype.Int8{Int64: int64(v), Valid: true}
|
|
||||||
}
|
|
||||||
|
|
||||||
var currentPage pgtype.Int4
|
|
||||||
if v, ok := syncData["page"].(float64); ok {
|
|
||||||
currentPage = pgtype.Int4{Int32: int32(v), Valid: true}
|
|
||||||
}
|
|
||||||
|
|
||||||
var totalPages pgtype.Int4
|
|
||||||
if v, ok := syncData["total_pages"].(float64); ok {
|
|
||||||
totalPages = pgtype.Int4{Int32: int32(v), Valid: true}
|
|
||||||
}
|
|
||||||
|
|
||||||
_, 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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in new issue
Block a user