diff --git a/internal/sync/queue.go b/internal/sync/queue.go new file mode 100644 index 0000000..981a9d8 --- /dev/null +++ b/internal/sync/queue.go @@ -0,0 +1,431 @@ +package sync + +import ( + "bookmann/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: (*string)(&item.ErrorMessage.String), + CreatedAt: item.CreatedAt.Time, + ProcessedAt: (*time.Time)(&item.ProcessedAt.Time), + } +} diff --git a/internal/sync/queue_test.go b/internal/sync/queue_test.go new file mode 100644 index 0000000..c2fc4bc --- /dev/null +++ b/internal/sync/queue_test.go @@ -0,0 +1,137 @@ +package sync + +import ( + "bookmann/internal/database" + "context" + "encoding/json" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSyncQueueProcessor_EnqueueProgress(t *testing.T) { + ctx := context.Background() + db := setupTestDB(t) + defer teardownTestDB(t, db) + + processor := NewSyncQueueProcessor(db) + + userID := pgtype.UUID{Bytes: uuid.New(), Valid: true} + deviceID := pgtype.UUID{Bytes: uuid.New(), Valid: true} + mediaItemID := pgtype.UUID{Bytes: uuid.New(), Valid: true} + + percentage := 0.45 + chapter := 3 + update := &ProgressUpdate{ + DeviceID: deviceID, + MediaItemID: mediaItemID, + UserID: userID, + Percentage: percentage, + Chapter: &chapter, + Source: "koreader", + SyncMode: "immediate", + } + + err := processor.EnqueueProgress(update) + require.NoError(t, err, "should enqueue progress update") + + items, err := db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ + DeviceID: deviceID, + Limit: 10, + }) + require.NoError(t, err) + assert.Len(t, items, 1, "should have one queue item") + + item := items[0] + assert.Equal(t, "progress", item.SyncType) + assert.Equal(t, SyncStatusPending, item.Status.String) + assert.Equal(t, int32(PriorityPageTurn), item.Priority.Int32) + + var syncData map[string]interface{} + err = json.Unmarshal(item.SyncData, &syncData) + require.NoError(t, err) + assert.Equal(t, percentage, syncData["percentage"]) + assert.Equal(t, "koreader", syncData["source"]) +} + +func TestCalculateNextRetry(t *testing.T) { + processor := &SyncQueueProcessor{} + + tests := []struct { + name string + attempts int32 + minDelay time.Duration + maxDelay time.Duration + }{ + {"Attempt 0", 0, 0, 1 * time.Second}, + {"Attempt 1", 1, 59 * time.Second, 61 * time.Second}, + {"Attempt 2", 2, 4*time.Minute + 59*time.Second, 5*time.Minute + 1*time.Second}, + {"Attempt 3", 3, 14*time.Minute + 59*time.Second, 15*time.Minute + 1*time.Second}, + {"Attempt 4", 4, 59*time.Minute + 59*time.Second, 60*time.Minute + 1*time.Second}, + {"Attempt 5", 5, 23*time.Hour + 59*time.Minute, 24*time.Hour + 1*time.Minute}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + nextRetry := processor.calculateNextRetry(tt.attempts) + delay := nextRetry.Sub(time.Now()) + + assert.GreaterOrEqual(t, delay, tt.minDelay, "delay should be at least minDelay") + assert.LessOrEqual(t, delay, tt.maxDelay, "delay should be at most maxDelay") + }) + } +} + +func TestSyncTypeConstants(t *testing.T) { + assert.Equal(t, "progress", SyncTypeProgress) + assert.Equal(t, "note", SyncTypeNote) + assert.Equal(t, "highlight", SyncTypeHighlight) + assert.Equal(t, "bookmark", SyncTypeBookmark) +} + +func TestSyncStatusConstants(t *testing.T) { + assert.Equal(t, "pending", SyncStatusPending) + assert.Equal(t, "processing", SyncStatusProcessing) + assert.Equal(t, "completed", SyncStatusCompleted) + assert.Equal(t, "failed", SyncStatusFailed) +} + +func TestPriorityConstants(t *testing.T) { + assert.Equal(t, int32(1), PriorityUserInitiated) + assert.Equal(t, int32(2), PriorityBookCompletion) + assert.Equal(t, int32(3), PriorityCriticalNote) + assert.Equal(t, int32(5), PriorityPageTurn) + assert.Equal(t, int32(7), PriorityCheckpoint) + assert.Equal(t, int32(10), PriorityBackgroundSync) +} + +func setupTestDB(t *testing.T) *database.Queries { + ctx := context.Background() + + dbURL := "postgresql://postgres:postgres@localhost:5432/bookmann?sslmode=disable" + dbPool, err := pgxpool.New(ctx, dbURL) + require.NoError(t, err, "Failed to connect to test database") + + db := database.New(dbPool) + + t.Cleanup(func() { + _, _ = dbPool.Exec(ctx, "DELETE FROM sync_queue WHERE true") + _, _ = dbPool.Exec(ctx, "DELETE FROM reading_progress WHERE true") + _, _ = dbPool.Exec(ctx, "DELETE FROM media_items WHERE title LIKE 'Test %'") + _, _ = dbPool.Exec(ctx, "DELETE FROM libraries WHERE name LIKE 'Test %'") + _, _ = dbPool.Exec(ctx, "DELETE FROM devices WHERE device_name LIKE 'Test %'") + _, _ = dbPool.Exec(ctx, "DELETE FROM users WHERE email LIKE 'test%'") + dbPool.Close() + }) + + return db +} + +func teardownTestDB(t *testing.T, db *database.Queries) { + // Cleanup is handled in setupTestDB via t.Cleanup +}