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 progressChan chan *ProgressUpdate interval time.Duration batchSize int } type ProgressUpdate struct { DeviceID pgtype.UUID MediaItemID pgtype.UUID UserID pgtype.UUID Percentage float64 Epubcfi *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 } func NewSyncQueueProcessor(db *database.Queries) *SyncQueueProcessor { return &SyncQueueProcessor{ db: db, progressChan: make(chan *ProgressUpdate, 100), interval: 5 * time.Second, batchSize: 50, } } 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") } } 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.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) } } } func (p *SyncQueueProcessor) getPendingItems(ctx context.Context) []SyncQueueItem { var allItems []SyncQueueItem var offset int32 = 0 batchSize := int32(p.batchSize) for { dbItems, err := p.db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ DeviceID: pgtype.UUID{Valid: false}, Limit: batchSize, }) if err != nil { log.Printf("Error listing pending items: %v", err) break } if len(dbItems) == 0 { break } for _, item := range dbItems { if item.Status.String == SyncStatusPending { allItems = append(allItems, dbItemToSyncQueueItem(item)) } } if int32(len(dbItems)) < batchSize { break } offset += batchSize } 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) 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") } 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} } source := "queue" if v, ok := syncData["source"].(string); ok { source = v } _, 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 } func (p *SyncQueueProcessor) syncNote(ctx context.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, syncData map[string]interface{}) error { return fmt.Errorf("note sync not yet implemented") } func (p *SyncQueueProcessor) syncHighlight(ctx context.Context, userID pgtype.UUID, mediaItemID pgtype.UUID, syncData map[string]interface{}) error { return fmt.Errorf("highlight sync not yet implemented") } 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, 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, } }