bookhoard/internal/sync/queue.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, annotationSvc *AnnotationService) *SyncQueueProcessor {
	return NewSyncQueueProcessorWithConfig(db, progressSvc, annotationSvc, 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, annotationSvc *AnnotationService, interval time.Duration, batchSize int) *SyncQueueProcessor {
	return &SyncQueueProcessor{
		db:            db,
		progressSvc:   progressSvc,
		annotationSvc: annotationSvc,
		progressChan:  make(chan *ProgressUpdate, 100),
		interval:      interval,
		batchSize:     batchSize,
	}
}

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 {
	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 {
	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 {
	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,
	}
}