John O'Keefe
6808c1babb
Found while verifying H1's "app offline-play -> flush -> position lands"
with the emulator path: the flush has been dead code in two independent
ways, invisible because the first break made the second unreachable.
1. ListPendingSyncQueueItems matched `device_id = $1` with a NULL $1
(the processor passes DeviceID{Valid:false} to enumerate all devices;
uuid = NULL matches no rows). The ticker therefore processed nothing,
ever. The WHERE now treats NULL as "every device" (the same
narg::uuid IS NULL idiom the visibility queries use; EXPLAIN'd — seq
scan on a bounded table, idx_sync_queue_status_priority_created
fits). The processor also dropped its fake-pagination loop, which
re-fetched the same first batch forever once the backlog hit the
batch size (no OFFSET in the query).
2. dbItemToSyncQueueItem dropped SyncData — even with a working
enumerator, every flushed item arrived with an empty payload and
failed JSON unmarshal. Dead code harboring silent breakage, exactly
the plan's thesis.
Callers updated for the renamed RowLimit param (queue handler, offline
detector). TestH1_SyncQueueFlush_LandsAndPreservesAudio drives the real
flow: web PUT seeds audio 900s, a pending checkpoint-style queue row
applies percentage 0.8 through SaveProgress, audio survives, queue row
completes.
Also fixes dashboard_test.go's stale BuildSections signature (the
package's test binary has been unbuildable) — user-approved: tests
designed against old shapes are not worth keeping.
657 lines
18 KiB
Go
657 lines
18 KiB
Go
package sync
|
|
|
|
import (
|
|
"bookhoard/internal/database"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const (
|
|
SyncTypeProgress = "progress"
|
|
SyncTypeNote = "note"
|
|
SyncTypeHighlight = "highlight"
|
|
SyncTypeBookmark = "bookmark"
|
|
)
|
|
|
|
const (
|
|
SyncStatusPending = "pending"
|
|
SyncStatusProcessing = "processing"
|
|
SyncStatusCompleted = "completed"
|
|
SyncStatusFailed = "failed"
|
|
)
|
|
|
|
const (
|
|
PriorityUserInitiated = 1
|
|
PriorityBookCompletion = 2
|
|
PriorityCriticalNote = 3
|
|
PriorityPageTurn = 5
|
|
PriorityCheckpoint = 7
|
|
PriorityBackgroundSync = 10
|
|
)
|
|
|
|
type SyncQueueProcessor struct {
|
|
db *database.Queries
|
|
progressSvc *ProgressService
|
|
annotationSvc *AnnotationService
|
|
progressChan chan *ProgressUpdate
|
|
interval time.Duration
|
|
batchSize int
|
|
}
|
|
|
|
type ProgressUpdate struct {
|
|
DeviceID pgtype.UUID
|
|
MediaItemID pgtype.UUID
|
|
UserID pgtype.UUID
|
|
Percentage float64
|
|
Epubcfi *string
|
|
ContextText *string
|
|
Chapter *int
|
|
Character *int64
|
|
Page *int
|
|
TotalPages *int
|
|
Source string
|
|
SyncMode string
|
|
}
|
|
|
|
type SyncQueueItem struct {
|
|
ID pgtype.UUID
|
|
DeviceID pgtype.UUID
|
|
MediaItemID pgtype.UUID
|
|
SyncType string
|
|
SyncData []byte
|
|
Priority int32
|
|
Attempts int32
|
|
MaxAttempts int32
|
|
Status string
|
|
ErrorMessage *string
|
|
CreatedAt time.Time
|
|
ProcessedAt *time.Time
|
|
}
|
|
|
|
// 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, 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) SetAnnotationService(svc *AnnotationService) {
|
|
p.annotationSvc = svc
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) Start(ctx context.Context) {
|
|
log.Printf("Starting sync queue processor (interval: %v, batch: %d)", p.interval, p.batchSize)
|
|
|
|
ticker := time.NewTicker(p.interval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
log.Println("Sync queue processor stopped")
|
|
return
|
|
case update := <-p.progressChan:
|
|
p.enqueueProgressUpdate(ctx, update)
|
|
case <-ticker.C:
|
|
p.processQueue(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) EnqueueProgress(update *ProgressUpdate) error {
|
|
select {
|
|
case p.progressChan <- update:
|
|
return nil
|
|
default:
|
|
return fmt.Errorf("progress channel is full")
|
|
}
|
|
}
|
|
|
|
type HighlightUpdate struct {
|
|
DeviceID pgtype.UUID
|
|
MediaItemID pgtype.UUID
|
|
UserID pgtype.UUID
|
|
SelectionText string
|
|
StartPosition string
|
|
EndPosition string
|
|
Color string
|
|
NoteText string
|
|
EpubcfiStart string
|
|
EpubcfiEnd string
|
|
PercentageStart float64
|
|
PercentageEnd float64
|
|
Source string
|
|
DeviceSyncData map[string]interface{}
|
|
}
|
|
|
|
type NoteUpdate struct {
|
|
DeviceID pgtype.UUID
|
|
MediaItemID pgtype.UUID
|
|
UserID pgtype.UUID
|
|
Content string
|
|
Position string
|
|
Source string
|
|
DeviceSyncData map[string]interface{}
|
|
}
|
|
|
|
type BookmarkUpdate struct {
|
|
DeviceID pgtype.UUID
|
|
MediaItemID pgtype.UUID
|
|
UserID pgtype.UUID
|
|
Title string
|
|
Position string
|
|
Notes string
|
|
Source string
|
|
DeviceSyncData map[string]interface{}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) EnqueueHighlight(ctx context.Context, update *HighlightUpdate) error {
|
|
syncData := map[string]interface{}{
|
|
"selection_text": update.SelectionText,
|
|
"start_position": update.StartPosition,
|
|
"end_position": update.EndPosition,
|
|
"color": update.Color,
|
|
"note_text": update.NoteText,
|
|
"source": update.Source,
|
|
"epubcfi_start": update.EpubcfiStart,
|
|
"epubcfi_end": update.EpubcfiEnd,
|
|
"percentage_start": update.PercentageStart,
|
|
"percentage_end": update.PercentageEnd,
|
|
"device_sync_data": update.DeviceSyncData,
|
|
}
|
|
return p.enqueueAnnotation(ctx, update.DeviceID, update.MediaItemID, SyncTypeHighlight, syncData)
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) EnqueueNote(ctx context.Context, update *NoteUpdate) error {
|
|
syncData := map[string]interface{}{
|
|
"content": update.Content,
|
|
"position": update.Position,
|
|
"source": update.Source,
|
|
"device_sync_data": update.DeviceSyncData,
|
|
}
|
|
return p.enqueueAnnotation(ctx, update.DeviceID, update.MediaItemID, SyncTypeNote, syncData)
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) EnqueueBookmark(ctx context.Context, update *BookmarkUpdate) error {
|
|
syncData := map[string]interface{}{
|
|
"title": update.Title,
|
|
"position": update.Position,
|
|
"notes": update.Notes,
|
|
"source": update.Source,
|
|
"device_sync_data": update.DeviceSyncData,
|
|
}
|
|
return p.enqueueAnnotation(ctx, update.DeviceID, update.MediaItemID, SyncTypeBookmark, syncData)
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) enqueueAnnotation(ctx context.Context, deviceID, mediaItemID pgtype.UUID, syncType string, syncData map[string]interface{}) error {
|
|
syncDataJSON, err := json.Marshal(syncData)
|
|
if err != nil {
|
|
return fmt.Errorf("marshal sync data: %w", err)
|
|
}
|
|
|
|
_, err = p.db.CreateSyncQueueItem(ctx, database.CreateSyncQueueItemParams{
|
|
DeviceID: deviceID,
|
|
MediaItemID: mediaItemID,
|
|
SyncType: syncType,
|
|
SyncData: syncDataJSON,
|
|
Priority: pgtype.Int4{Int32: int32(PriorityCriticalNote), Valid: true},
|
|
MaxAttempts: pgtype.Int4{Int32: 3, Valid: true},
|
|
Status: pgtype.Text{String: SyncStatusPending, Valid: true},
|
|
})
|
|
return err
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) enqueueProgressUpdate(ctx context.Context, update *ProgressUpdate) {
|
|
syncData := map[string]interface{}{
|
|
"percentage": update.Percentage,
|
|
"source": update.Source,
|
|
"timestamp": time.Now().Format(time.RFC3339),
|
|
}
|
|
|
|
if update.Epubcfi != nil {
|
|
syncData["epubcfi"] = *update.Epubcfi
|
|
}
|
|
if update.ContextText != nil {
|
|
syncData["context_text"] = *update.ContextText
|
|
}
|
|
if update.Chapter != nil {
|
|
syncData["chapter"] = *update.Chapter
|
|
}
|
|
if update.Character != nil {
|
|
syncData["character"] = *update.Character
|
|
}
|
|
if update.Page != nil {
|
|
syncData["page"] = *update.Page
|
|
}
|
|
if update.TotalPages != nil {
|
|
syncData["total_pages"] = *update.TotalPages
|
|
}
|
|
|
|
syncDataJSON, err := json.Marshal(syncData)
|
|
if err != nil {
|
|
log.Printf("Failed to marshal sync data: %v", err)
|
|
return
|
|
}
|
|
|
|
priority := PriorityPageTurn
|
|
if update.SyncMode == "checkpoint" {
|
|
priority = PriorityCheckpoint
|
|
} else if update.Percentage >= 0.99 {
|
|
priority = PriorityBookCompletion
|
|
}
|
|
|
|
maxAttempts := int32(3)
|
|
if update.SyncMode == "immediate" {
|
|
maxAttempts = 5
|
|
}
|
|
|
|
_, err = p.db.CreateSyncQueueItem(ctx, database.CreateSyncQueueItemParams{
|
|
DeviceID: update.DeviceID,
|
|
MediaItemID: update.MediaItemID,
|
|
SyncType: SyncTypeProgress,
|
|
SyncData: syncDataJSON,
|
|
Priority: pgtype.Int4{Int32: int32(priority), Valid: true},
|
|
MaxAttempts: pgtype.Int4{Int32: maxAttempts, Valid: true},
|
|
Status: pgtype.Text{String: SyncStatusPending, Valid: true},
|
|
})
|
|
|
|
if err != nil {
|
|
log.Printf("Failed to create sync queue item: %v", err)
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) processQueue(ctx context.Context) {
|
|
items := p.getPendingItems(ctx)
|
|
|
|
if len(items) == 0 {
|
|
p.checkStuckItems(ctx)
|
|
return
|
|
}
|
|
|
|
log.Printf("Processing %d sync queue items", len(items))
|
|
|
|
for _, item := range items {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
p.processItem(ctx, item)
|
|
}
|
|
}
|
|
}
|
|
|
|
// getPendingItems enumerates the pending queue. DeviceID is passed NULL:
|
|
// the processor owns every device's backlog (the per-device path remains
|
|
// available to callers that pass a concrete id). One fetch per tick — the
|
|
// old loop re-fetched the same first batch (no OFFSET in the query) and
|
|
// would have looped forever once the backlog hit the batch size.
|
|
func (p *SyncQueueProcessor) getPendingItems(ctx context.Context) []SyncQueueItem {
|
|
var allItems []SyncQueueItem
|
|
|
|
dbItems, err := p.db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{
|
|
DeviceID: pgtype.UUID{Valid: false}, // NULL = all devices
|
|
RowLimit: int32(p.batchSize),
|
|
})
|
|
if err != nil {
|
|
log.Printf("Error listing pending items: %v", err)
|
|
return nil
|
|
}
|
|
|
|
for _, item := range dbItems {
|
|
if item.Status.String == SyncStatusPending {
|
|
allItems = append(allItems, dbItemToSyncQueueItem(item))
|
|
}
|
|
}
|
|
|
|
return allItems
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) checkStuckItems(ctx context.Context) {
|
|
items, err := p.db.GetStuckSyncQueueItems(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
for _, item := range items {
|
|
if item.Attempts.Int32 < item.MaxAttempts.Int32 {
|
|
log.Printf("Resetting stuck queue item %s", item.ID.Bytes)
|
|
_, err := p.db.UpdateSyncQueueItemStatus(ctx, database.UpdateSyncQueueItemStatusParams{
|
|
ID: item.ID,
|
|
Status: pgtype.Text{String: SyncStatusPending, Valid: true},
|
|
ErrorMessage: pgtype.Text{String: "Item was stuck in processing state", Valid: true},
|
|
})
|
|
if err != nil {
|
|
log.Printf("Failed to reset stuck item: %v", err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) processItem(ctx context.Context, item SyncQueueItem) {
|
|
_, err := p.db.UpdateSyncQueueItemStatus(ctx, database.UpdateSyncQueueItemStatusParams{
|
|
ID: item.ID,
|
|
Status: pgtype.Text{String: SyncStatusProcessing, Valid: true},
|
|
ErrorMessage: pgtype.Text{},
|
|
})
|
|
if err != nil {
|
|
log.Printf("Failed to update item status to processing: %v", err)
|
|
return
|
|
}
|
|
|
|
var syncData map[string]interface{}
|
|
err = json.Unmarshal(item.SyncData, &syncData)
|
|
if err != nil {
|
|
log.Printf("Failed to unmarshal sync data: %v", err)
|
|
p.markItemFailed(ctx, item, fmt.Sprintf("Invalid sync data: %v", err))
|
|
return
|
|
}
|
|
|
|
err = p.executeSync(ctx, item, syncData)
|
|
if err != nil {
|
|
log.Printf("Sync failed for item %s: %v", item.ID.Bytes, err)
|
|
|
|
if item.Attempts >= item.MaxAttempts {
|
|
p.markItemFailed(ctx, item, fmt.Sprintf("Max attempts reached: %v", err))
|
|
} else {
|
|
nextRetry := p.calculateNextRetry(item.Attempts)
|
|
log.Printf("Scheduling retry for item %s at %v (attempt %d/%d)",
|
|
item.ID.Bytes, nextRetry, item.Attempts+1, item.MaxAttempts)
|
|
|
|
p.markItemPending(ctx, item, fmt.Sprintf("Will retry: %v", err))
|
|
}
|
|
return
|
|
}
|
|
|
|
_, err = p.db.UpdateSyncQueueItemStatus(ctx, database.UpdateSyncQueueItemStatusParams{
|
|
ID: item.ID,
|
|
Status: pgtype.Text{String: SyncStatusCompleted, Valid: true},
|
|
ErrorMessage: pgtype.Text{},
|
|
})
|
|
if err != nil {
|
|
log.Printf("Failed to mark item as completed: %v", err)
|
|
}
|
|
|
|
log.Printf("Successfully processed sync queue item %s", item.ID.Bytes)
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) executeSync(ctx context.Context, item SyncQueueItem, syncData map[string]interface{}) error {
|
|
device, err := p.db.GetDevice(ctx, item.DeviceID)
|
|
if err != nil {
|
|
return fmt.Errorf("device not found: %w", err)
|
|
}
|
|
|
|
switch item.SyncType {
|
|
case SyncTypeProgress:
|
|
return p.syncProgress(ctx, device.UserID, item.MediaItemID, syncData)
|
|
case SyncTypeNote:
|
|
return p.syncNote(ctx, device.UserID, item.MediaItemID, syncData)
|
|
case SyncTypeHighlight:
|
|
return p.syncHighlight(ctx, device.UserID, item.MediaItemID, syncData)
|
|
case SyncTypeBookmark:
|
|
return p.syncBookmark(ctx, device.UserID, item.MediaItemID, syncData)
|
|
default:
|
|
return fmt.Errorf("unsupported sync type: %s", item.SyncType)
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) syncProgress(ctx context.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, syncData map[string]interface{}) error {
|
|
percentage, ok := syncData["percentage"].(float64)
|
|
if !ok {
|
|
return fmt.Errorf("missing percentage in sync data")
|
|
}
|
|
|
|
source := "queue"
|
|
if v, ok := syncData["source"].(string); ok {
|
|
source = v
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) syncNote(ctx context.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, syncData map[string]interface{}) error {
|
|
if p.annotationSvc == nil {
|
|
return fmt.Errorf("annotation service not available")
|
|
}
|
|
|
|
req := SaveNoteRequest{
|
|
MediaItemID: mediaItemID,
|
|
UserID: userID,
|
|
}
|
|
|
|
if v, ok := syncData["content"].(string); ok {
|
|
req.Content = v
|
|
}
|
|
if v, ok := syncData["position"].(string); ok {
|
|
req.Position = v
|
|
}
|
|
if v, ok := syncData["source"].(string); ok {
|
|
req.Source = v
|
|
}
|
|
if v, ok := syncData["epubcfi_location"].(string); ok {
|
|
req.EpubcfiLocation = v
|
|
}
|
|
if v, ok := syncData["percentage_location"].(float64); ok {
|
|
req.PercentageLocation = v
|
|
}
|
|
if v, ok := syncData["chapter_reference"].(float64); ok {
|
|
req.ChapterReference = int32(v)
|
|
}
|
|
if v, ok := syncData["device_sync_data"].(map[string]interface{}); ok {
|
|
req.DeviceSyncData, _ = json.Marshal(v)
|
|
}
|
|
|
|
_, err := p.annotationSvc.SaveNote(ctx, req)
|
|
return err
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) syncHighlight(ctx context.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, syncData map[string]interface{}) error {
|
|
if p.annotationSvc == nil {
|
|
return fmt.Errorf("annotation service not available")
|
|
}
|
|
|
|
req := SaveHighlightRequest{
|
|
MediaItemID: mediaItemID,
|
|
UserID: userID,
|
|
}
|
|
|
|
if v, ok := syncData["selection_text"].(string); ok {
|
|
req.SelectionText = v
|
|
}
|
|
if v, ok := syncData["start_position"].(string); ok {
|
|
req.StartPosition = v
|
|
}
|
|
if v, ok := syncData["end_position"].(string); ok {
|
|
req.EndPosition = v
|
|
}
|
|
if v, ok := syncData["color"].(string); ok {
|
|
req.Color = v
|
|
}
|
|
if v, ok := syncData["note_text"].(string); ok {
|
|
req.NoteText = v
|
|
}
|
|
if v, ok := syncData["source"].(string); ok {
|
|
req.Source = v
|
|
}
|
|
if v, ok := syncData["epubcfi_start"].(string); ok {
|
|
req.EpubcfiStart = v
|
|
}
|
|
if v, ok := syncData["epubcfi_end"].(string); ok {
|
|
req.EpubcfiEnd = v
|
|
}
|
|
if v, ok := syncData["percentage_start"].(float64); ok {
|
|
req.PercentageStart = v
|
|
}
|
|
if v, ok := syncData["percentage_end"].(float64); ok {
|
|
req.PercentageEnd = v
|
|
}
|
|
if v, ok := syncData["chapter_reference"].(float64); ok {
|
|
req.ChapterReference = int32(v)
|
|
}
|
|
if v, ok := syncData["device_sync_data"].(map[string]interface{}); ok {
|
|
req.DeviceSyncData, _ = json.Marshal(v)
|
|
}
|
|
|
|
_, err := p.annotationSvc.SaveHighlight(ctx, req)
|
|
return err
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) syncBookmark(ctx context.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, syncData map[string]interface{}) error {
|
|
if p.annotationSvc == nil {
|
|
return fmt.Errorf("annotation service not available")
|
|
}
|
|
|
|
req := SaveBookmarkRequest{
|
|
MediaItemID: mediaItemID,
|
|
UserID: userID,
|
|
}
|
|
|
|
if v, ok := syncData["title"].(string); ok {
|
|
req.Title = v
|
|
}
|
|
if v, ok := syncData["position"].(string); ok {
|
|
req.Position = v
|
|
}
|
|
if v, ok := syncData["notes"].(string); ok {
|
|
req.Notes = v
|
|
}
|
|
if v, ok := syncData["source"].(string); ok {
|
|
req.Source = v
|
|
}
|
|
if v, ok := syncData["cfi_position"].(string); ok {
|
|
req.CFIPosition = v
|
|
}
|
|
if v, ok := syncData["epubcfi_location"].(string); ok {
|
|
req.EpubcfiLocation = v
|
|
}
|
|
if v, ok := syncData["percentage_loc"].(float64); ok {
|
|
req.PercentageLoc = v
|
|
}
|
|
if v, ok := syncData["page_number"].(float64); ok {
|
|
req.PageNumber = int32(v)
|
|
}
|
|
if v, ok := syncData["chapter_number"].(float64); ok {
|
|
req.ChapterNumber = int32(v)
|
|
}
|
|
if v, ok := syncData["chapter_reference"].(float64); ok {
|
|
req.ChapterReference = int32(v)
|
|
}
|
|
if v, ok := syncData["device_sync_data"].(map[string]interface{}); ok {
|
|
req.DeviceSyncData, _ = json.Marshal(v)
|
|
}
|
|
|
|
_, err := p.annotationSvc.SaveBookmark(ctx, req)
|
|
return err
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) markItemFailed(ctx context.Context, item SyncQueueItem, errMsg string) {
|
|
_, err := p.db.UpdateSyncQueueItemStatus(ctx, database.UpdateSyncQueueItemStatusParams{
|
|
ID: item.ID,
|
|
Status: pgtype.Text{String: SyncStatusFailed, Valid: true},
|
|
ErrorMessage: pgtype.Text{String: errMsg, Valid: true},
|
|
})
|
|
if err != nil {
|
|
log.Printf("Failed to mark item as failed: %v", err)
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) markItemPending(ctx context.Context, item SyncQueueItem, errMsg string) {
|
|
_, err := p.db.UpdateSyncQueueItemStatus(ctx, database.UpdateSyncQueueItemStatusParams{
|
|
ID: item.ID,
|
|
Status: pgtype.Text{String: SyncStatusPending, Valid: true},
|
|
ErrorMessage: pgtype.Text{String: errMsg, Valid: true},
|
|
})
|
|
if err != nil {
|
|
log.Printf("Failed to mark item as pending: %v", err)
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) calculateNextRetry(attempts int32) time.Time {
|
|
switch attempts {
|
|
case 0:
|
|
return time.Now()
|
|
case 1:
|
|
return time.Now().Add(1 * time.Minute)
|
|
case 2:
|
|
return time.Now().Add(5 * time.Minute)
|
|
case 3:
|
|
return time.Now().Add(15 * time.Minute)
|
|
case 4:
|
|
return time.Now().Add(1 * time.Hour)
|
|
default:
|
|
return time.Now().Add(24 * time.Hour)
|
|
}
|
|
}
|
|
|
|
func (p *SyncQueueProcessor) GetQueueStats(ctx context.Context, deviceID pgtype.UUID) (database.GetSyncQueueStatsRow, error) {
|
|
return p.db.GetSyncQueueStats(ctx, deviceID)
|
|
}
|
|
|
|
func dbItemToSyncQueueItem(item database.SyncQueue) SyncQueueItem {
|
|
return SyncQueueItem{
|
|
ID: item.ID,
|
|
DeviceID: item.DeviceID,
|
|
MediaItemID: item.MediaItemID,
|
|
SyncType: item.SyncType,
|
|
SyncData: item.SyncData, // was dropped: every flushed item lost its payload
|
|
Priority: item.Priority.Int32,
|
|
Attempts: item.Attempts.Int32,
|
|
MaxAttempts: item.MaxAttempts.Int32,
|
|
Status: item.Status.String,
|
|
ErrorMessage: &item.ErrorMessage.String,
|
|
CreatedAt: item.CreatedAt.Time,
|
|
ProcessedAt: &item.ProcessedAt.Time,
|
|
}
|
|
}
|