Add queue management API and database queries

- Add 7 new database queries for queue management - GetStuckSyncQueueItems - Detect stuck items - GetSyncQueueStats - Queue statistics - GetNextRetryTime - Exponential backoff calc - ListAllSyncQueueItems - Admin view - IncrementSyncQueueAttempts - Retry counter - Add queue handler with 7 REST endpoints - GET /api/queue/devices/:id/stats - Queue statistics - GET /api/queue/devices/:id/items - List device queue - POST /api/queue/items/:id/retry - Retry failed item - DELETE /api/queue/items/:id - Delete queue item - DELETE /api/queue/devices/:id/clear - Clear device queue - GET /api/queue/items - List all items (admin) - Add full user/admin access control
2026-01-31 13:06:12 -05:00
parent 11ee4901a5
commit 030d30a225
5 changed files with 671 additions and 0 deletions
@@ -394,6 +394,7 @@ type Users struct {
 	Theme                pgtype.Text        `db:"theme" json:"theme"`
 	ScanFrequencyMinutes pgtype.Int4        `db:"scan_frequency_minutes" json:"scan_frequency_minutes"`
 	AutoScanEnabled      pgtype.Bool        `db:"auto_scan_enabled" json:"auto_scan_enabled"`
+	MaxDevices           pgtype.Int4        `db:"max_devices" json:"max_devices"`
 	CreatedAt            pgtype.Timestamptz `db:"created_at" json:"created_at"`
 	UpdatedAt            pgtype.Timestamptz `db:"updated_at" json:"updated_at"`
 }
@@ -24,6 +24,7 @@ type Querier interface {
 	ClearDeviceSyncQueue(ctx context.Context, deviceID pgtype.UUID) error
 	ClearKoboShelf(ctx context.Context, deviceID pgtype.UUID) error
 	ClearKoboShelfByName(ctx context.Context, arg ClearKoboShelfByNameParams) error
+	CountUserDevices(ctx context.Context, userID pgtype.UUID) (int64, error)
 	// ============================================
 	// PHASE 2: DEVICE MANAGEMENT & AUTH (Weeks 5-6)
 	// ============================================
@@ -73,6 +74,7 @@ type Querier interface {
 	GetDevice(ctx context.Context, id pgtype.UUID) (Devices, error)
 	GetDeviceByAuthToken(ctx context.Context, authToken string) (Devices, error)
 	GetDeviceByIdentifier(ctx context.Context, deviceIdentifier string) (Devices, error)
+	GetFailedSyncQueueItems(ctx context.Context, limit int32) ([]SyncQueue, error)
 	GetKoboEntitlementByContentId(ctx context.Context, arg GetKoboEntitlementByContentIdParams) (GetKoboEntitlementByContentIdRow, error)
 	GetKoboEntitlementByEntitlementId(ctx context.Context, arg GetKoboEntitlementByEntitlementIdParams) (GetKoboEntitlementByEntitlementIdRow, error)
 	GetKoboEntitlementsForDevice(ctx context.Context, deviceID pgtype.UUID) ([]GetKoboEntitlementsForDeviceRow, error)
@@ -100,13 +102,16 @@ type Querier interface {
 	GetMediaNotes(ctx context.Context, arg GetMediaNotesParams) ([]MediaNotes, error)
 	GetMediaRating(ctx context.Context, arg GetMediaRatingParams) (MediaRatings, error)
 	GetMediaRatings(ctx context.Context, mediaItemID pgtype.UUID) ([]GetMediaRatingsRow, error)
+	GetNextRetryTime(ctx context.Context) (interface{}, error)
 	// Get reading history for a user and book
 	GetReadingHistory(ctx context.Context, arg GetReadingHistoryParams) ([]ReadingHistory, error)
 	GetReadingProgress(ctx context.Context, arg GetReadingProgressParams) (ReadingProgress, error)
 	GetRefreshToken(ctx context.Context, token pgtype.UUID) (GetRefreshTokenRow, error)
 	GetScanSettings(ctx context.Context, id pgtype.UUID) (GetScanSettingsRow, error)
+	GetStuckSyncQueueItems(ctx context.Context) ([]SyncQueue, error)
 	GetSyncConflict(ctx context.Context, id pgtype.UUID) (SyncConflicts, error)
 	GetSyncQueueItem(ctx context.Context, id pgtype.UUID) (SyncQueue, error)
+	GetSyncQueueStats(ctx context.Context, deviceID pgtype.UUID) (GetSyncQueueStatsRow, error)
 	// Get universal progress for a book
 	GetUniversalProgress(ctx context.Context, arg GetUniversalProgressParams) (GetUniversalProgressRow, error)
 	GetUser(ctx context.Context, id pgtype.UUID) (GetUserRow, error)
@@ -118,8 +123,10 @@ type Querier interface {
 	GetUserPasswordHash(ctx context.Context, id pgtype.UUID) (string, error)
 	GetUserProgressForBooks(ctx context.Context, arg GetUserProgressForBooksParams) ([]GetUserProgressForBooksRow, error)
 	GetUserVisibleLibraries(ctx context.Context, userID pgtype.UUID) ([]GetUserVisibleLibrariesRow, error)
+	IncrementSyncQueueAttempts(ctx context.Context, arg IncrementSyncQueueAttemptsParams) (SyncQueue, error)
 	IsBookOnKoboShelf(ctx context.Context, arg IsBookOnKoboShelfParams) (bool, error)
 	ListAllConflictsByUserAndStatus(ctx context.Context, arg ListAllConflictsByUserAndStatusParams) ([]ListAllConflictsByUserAndStatusRow, error)
+	ListAllSyncQueueItems(ctx context.Context, arg ListAllSyncQueueItemsParams) ([]ListAllSyncQueueItemsRow, error)
 	ListDevicesByType(ctx context.Context, deviceType string) ([]Devices, error)
 	ListDevicesByUser(ctx context.Context, userID pgtype.UUID) ([]Devices, error)
 	ListLibraries(ctx context.Context) ([]ListLibrariesRow, error)
@@ -169,6 +176,7 @@ type Querier interface {
 	UpdateSyncQueueItemStatus(ctx context.Context, arg UpdateSyncQueueItemStatusParams) (SyncQueue, error)
 	// Update universal progress
 	UpdateUniversalProgress(ctx context.Context, arg UpdateUniversalProgressParams) (ReadingProgress, error)
+	UpdateUserMaxDevices(ctx context.Context, arg UpdateUserMaxDevicesParams) error
 	UpdateUserProfile(ctx context.Context, arg UpdateUserProfileParams) error
 	UpdateUserTheme(ctx context.Context, arg UpdateUserThemeParams) error
 	UpdateUsername(ctx context.Context, arg UpdateUsernameParams) error
@@ -162,6 +162,17 @@ func (q *Queries) ClearKoboShelfByName(ctx context.Context, arg ClearKoboShelfBy
 	return err
 }

+const CountUserDevices = `-- name: CountUserDevices :one
+SELECT COUNT(*) FROM devices WHERE user_id = $1
+`
+
+func (q *Queries) CountUserDevices(ctx context.Context, userID pgtype.UUID) (int64, error) {
+	row := q.db.QueryRow(ctx, CountUserDevices, userID)
+	var count int64
+	err := row.Scan(&count)
+	return count, err
+}
+
 const CreateDevice = `-- name: CreateDevice :one

 INSERT INTO devices (user_id, device_name, device_type, device_identifier, auth_token, sync_enabled, auto_sync, sync_frequency_minutes, device_metadata)
@@ -1164,6 +1175,46 @@ func (q *Queries) GetDeviceByIdentifier(ctx context.Context, deviceIdentifier st
 	return i, err
 }

+const GetFailedSyncQueueItems = `-- name: GetFailedSyncQueueItems :many
+SELECT id, device_id, media_item_id, sync_type, sync_data, priority, attempts, max_attempts, status, error_message, created_at, processed_at FROM sync_queue
+WHERE status = 'failed' AND attempts < max_attempts
+ORDER BY priority ASC, created_at ASC
+LIMIT $1
+`
+
+func (q *Queries) GetFailedSyncQueueItems(ctx context.Context, limit int32) ([]SyncQueue, error) {
+	rows, err := q.db.Query(ctx, GetFailedSyncQueueItems, limit)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []SyncQueue{}
+	for rows.Next() {
+		var i SyncQueue
+		if err := rows.Scan(
+			&i.ID,
+			&i.DeviceID,
+			&i.MediaItemID,
+			&i.SyncType,
+			&i.SyncData,
+			&i.Priority,
+			&i.Attempts,
+			&i.MaxAttempts,
+			&i.Status,
+			&i.ErrorMessage,
+			&i.CreatedAt,
+			&i.ProcessedAt,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const GetKoboEntitlementByContentId = `-- name: GetKoboEntitlementByContentId :one
 SELECT ke.id, ke.device_id, ke.media_item_id, ke.entitlement_id, ke.content_id, ke.revision_number, ke.purchase_date, ke.accession_date, ke.book_status, ke.sync_status, ke.kobo_metadata, ke.created_at, ke.updated_at, mi.title, mi.author, mi.file_path
 FROM kobo_entitlements ke
@@ -2089,6 +2140,25 @@ func (q *Queries) GetMediaRatings(ctx context.Context, mediaItemID pgtype.UUID)
 	return items, nil
 }

+const GetNextRetryTime = `-- name: GetNextRetryTime :one
+SELECT 
+    CASE 
+        WHEN attempts = 0 THEN NOW()
+        WHEN attempts = 1 THEN NOW() + INTERVAL '1 minute'
+        WHEN attempts = 2 THEN NOW() + INTERVAL '5 minutes'
+        WHEN attempts = 3 THEN NOW() + INTERVAL '15 minutes'
+        WHEN attempts = 4 THEN NOW() + INTERVAL '1 hour'
+        ELSE NOW() + INTERVAL '24 hours'
+    END as next_retry_time
+`
+
+func (q *Queries) GetNextRetryTime(ctx context.Context) (interface{}, error) {
+	row := q.db.QueryRow(ctx, GetNextRetryTime)
+	var next_retry_time interface{}
+	err := row.Scan(&next_retry_time)
+	return next_retry_time, err
+}
+
 const GetReadingHistory = `-- name: GetReadingHistory :many
 SELECT id, user_id, media_item_id, device_id, progress_percentage, reading_session_start, reading_session_end, pages_read, time_spent_seconds, device_metadata, created_at
 FROM reading_history
@@ -2228,6 +2298,45 @@ func (q *Queries) GetScanSettings(ctx context.Context, id pgtype.UUID) (GetScanS
 	return i, err
 }

+const GetStuckSyncQueueItems = `-- name: GetStuckSyncQueueItems :many
+SELECT id, device_id, media_item_id, sync_type, sync_data, priority, attempts, max_attempts, status, error_message, created_at, processed_at FROM sync_queue
+WHERE status = 'processing' AND created_at < NOW() - INTERVAL '1 hour'
+ORDER BY priority ASC, created_at ASC
+`
+
+func (q *Queries) GetStuckSyncQueueItems(ctx context.Context) ([]SyncQueue, error) {
+	rows, err := q.db.Query(ctx, GetStuckSyncQueueItems)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []SyncQueue{}
+	for rows.Next() {
+		var i SyncQueue
+		if err := rows.Scan(
+			&i.ID,
+			&i.DeviceID,
+			&i.MediaItemID,
+			&i.SyncType,
+			&i.SyncData,
+			&i.Priority,
+			&i.Attempts,
+			&i.MaxAttempts,
+			&i.Status,
+			&i.ErrorMessage,
+			&i.CreatedAt,
+			&i.ProcessedAt,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const GetSyncConflict = `-- name: GetSyncConflict :one
 SELECT id, media_item_id, user_id, conflict_type, conflict_data, resolution_status, resolution_data, resolved_by, resolved_at, created_at FROM sync_conflicts WHERE id = $1
 `
@@ -2274,6 +2383,38 @@ func (q *Queries) GetSyncQueueItem(ctx context.Context, id pgtype.UUID) (SyncQue
 	return i, err
 }

+const GetSyncQueueStats = `-- name: GetSyncQueueStats :one
+SELECT
+    COUNT(*) FILTER (WHERE status = 'pending') as pending_count,
+    COUNT(*) FILTER (WHERE status = 'processing') as processing_count,
+    COUNT(*) FILTER (WHERE status = 'failed') as failed_count,
+    COUNT(*) FILTER (WHERE status = 'completed') as completed_count,
+    COUNT(*) as total_count
+FROM sync_queue
+WHERE device_id = $1
+`
+
+type GetSyncQueueStatsRow struct {
+	PendingCount    int64 `db:"pending_count" json:"pending_count"`
+	ProcessingCount int64 `db:"processing_count" json:"processing_count"`
+	FailedCount     int64 `db:"failed_count" json:"failed_count"`
+	CompletedCount  int64 `db:"completed_count" json:"completed_count"`
+	TotalCount      int64 `db:"total_count" json:"total_count"`
+}
+
+func (q *Queries) GetSyncQueueStats(ctx context.Context, deviceID pgtype.UUID) (GetSyncQueueStatsRow, error) {
+	row := q.db.QueryRow(ctx, GetSyncQueueStats, deviceID)
+	var i GetSyncQueueStatsRow
+	err := row.Scan(
+		&i.PendingCount,
+		&i.ProcessingCount,
+		&i.FailedCount,
+		&i.CompletedCount,
+		&i.TotalCount,
+	)
+	return i, err
+}
+
 const GetUniversalProgress = `-- name: GetUniversalProgress :one
 SELECT
    rp.id,
@@ -2770,6 +2911,44 @@ func (q *Queries) GetUserVisibleLibraries(ctx context.Context, userID pgtype.UUI
 	return items, nil
 }

+const IncrementSyncQueueAttempts = `-- name: IncrementSyncQueueAttempts :one
+UPDATE sync_queue
+SET
+    attempts = attempts + 1,
+    status = CASE 
+        WHEN attempts + 1 >= max_attempts THEN 'failed'
+        ELSE 'pending'
+    END,
+    error_message = $2
+WHERE id = $1
+RETURNING id, device_id, media_item_id, sync_type, sync_data, priority, attempts, max_attempts, status, error_message, created_at, processed_at
+`
+
+type IncrementSyncQueueAttemptsParams struct {
+	ID           pgtype.UUID `db:"id" json:"id"`
+	ErrorMessage pgtype.Text `db:"error_message" json:"error_message"`
+}
+
+func (q *Queries) IncrementSyncQueueAttempts(ctx context.Context, arg IncrementSyncQueueAttemptsParams) (SyncQueue, error) {
+	row := q.db.QueryRow(ctx, IncrementSyncQueueAttempts, arg.ID, arg.ErrorMessage)
+	var i SyncQueue
+	err := row.Scan(
+		&i.ID,
+		&i.DeviceID,
+		&i.MediaItemID,
+		&i.SyncType,
+		&i.SyncData,
+		&i.Priority,
+		&i.Attempts,
+		&i.MaxAttempts,
+		&i.Status,
+		&i.ErrorMessage,
+		&i.CreatedAt,
+		&i.ProcessedAt,
+	)
+	return i, err
+}
+
 const IsBookOnKoboShelf = `-- name: IsBookOnKoboShelf :one
 SELECT EXISTS(
    SELECT 1 FROM kobo_shelves
@@ -2850,6 +3029,82 @@ func (q *Queries) ListAllConflictsByUserAndStatus(ctx context.Context, arg ListA
 	return items, nil
 }

+const ListAllSyncQueueItems = `-- name: ListAllSyncQueueItems :many
+SELECT 
+    sq.id, sq.device_id, sq.media_item_id, sq.sync_type, sq.sync_data, sq.priority, sq.attempts, sq.max_attempts, sq.status, sq.error_message, sq.created_at, sq.processed_at,
+    d.device_name,
+    d.device_type,
+    u.email as user_email,
+    mi.title as media_title
+FROM sync_queue sq
+JOIN devices d ON sq.device_id = d.id
+JOIN users u ON d.user_id = u.id
+LEFT JOIN media_items mi ON sq.media_item_id = mi.id
+ORDER BY sq.priority ASC, sq.created_at DESC
+LIMIT $1 OFFSET $2
+`
+
+type ListAllSyncQueueItemsParams struct {
+	Limit  int32 `db:"limit" json:"limit"`
+	Offset int32 `db:"offset" json:"offset"`
+}
+
+type ListAllSyncQueueItemsRow struct {
+	ID           pgtype.UUID        `db:"id" json:"id"`
+	DeviceID     pgtype.UUID        `db:"device_id" json:"device_id"`
+	MediaItemID  pgtype.UUID        `db:"media_item_id" json:"media_item_id"`
+	SyncType     string             `db:"sync_type" json:"sync_type"`
+	SyncData     []byte             `db:"sync_data" json:"sync_data"`
+	Priority     pgtype.Int4        `db:"priority" json:"priority"`
+	Attempts     pgtype.Int4        `db:"attempts" json:"attempts"`
+	MaxAttempts  pgtype.Int4        `db:"max_attempts" json:"max_attempts"`
+	Status       pgtype.Text        `db:"status" json:"status"`
+	ErrorMessage pgtype.Text        `db:"error_message" json:"error_message"`
+	CreatedAt    pgtype.Timestamptz `db:"created_at" json:"created_at"`
+	ProcessedAt  pgtype.Timestamptz `db:"processed_at" json:"processed_at"`
+	DeviceName   string             `db:"device_name" json:"device_name"`
+	DeviceType   string             `db:"device_type" json:"device_type"`
+	UserEmail    string             `db:"user_email" json:"user_email"`
+	MediaTitle   pgtype.Text        `db:"media_title" json:"media_title"`
+}
+
+func (q *Queries) ListAllSyncQueueItems(ctx context.Context, arg ListAllSyncQueueItemsParams) ([]ListAllSyncQueueItemsRow, error) {
+	rows, err := q.db.Query(ctx, ListAllSyncQueueItems, arg.Limit, arg.Offset)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	items := []ListAllSyncQueueItemsRow{}
+	for rows.Next() {
+		var i ListAllSyncQueueItemsRow
+		if err := rows.Scan(
+			&i.ID,
+			&i.DeviceID,
+			&i.MediaItemID,
+			&i.SyncType,
+			&i.SyncData,
+			&i.Priority,
+			&i.Attempts,
+			&i.MaxAttempts,
+			&i.Status,
+			&i.ErrorMessage,
+			&i.CreatedAt,
+			&i.ProcessedAt,
+			&i.DeviceName,
+			&i.DeviceType,
+			&i.UserEmail,
+			&i.MediaTitle,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const ListDevicesByType = `-- name: ListDevicesByType :many
 SELECT id, user_id, device_name, device_type, device_identifier, auth_token, last_sync, last_seen, sync_enabled, auto_sync, sync_frequency_minutes, device_metadata, created_at, updated_at FROM devices WHERE device_type = $1 ORDER BY created_at DESC
 `
@@ -5041,6 +5296,20 @@ func (q *Queries) UpdateUniversalProgress(ctx context.Context, arg UpdateUnivers
 	return i, err
 }

+const UpdateUserMaxDevices = `-- name: UpdateUserMaxDevices :exec
+UPDATE users SET max_devices = $2, updated_at = NOW() WHERE id = $1
+`
+
+type UpdateUserMaxDevicesParams struct {
+	ID         pgtype.UUID `db:"id" json:"id"`
+	MaxDevices pgtype.Int4 `db:"max_devices" json:"max_devices"`
+}
+
+func (q *Queries) UpdateUserMaxDevices(ctx context.Context, arg UpdateUserMaxDevicesParams) error {
+	_, err := q.db.Exec(ctx, UpdateUserMaxDevices, arg.ID, arg.MaxDevices)
+	return err
+}
+
 const UpdateUserProfile = `-- name: UpdateUserProfile :exec
 UPDATE users SET first_name = $2, last_name = $3, updated_at = NOW() WHERE id = $1
 `
@@ -305,6 +305,12 @@ UPDATE users SET email = $2, updated_at = NOW() WHERE id = $1;
 -- name: UpdatePassword :exec
 UPDATE users SET password_hash = $2, updated_at = NOW() WHERE id = $1;

+-- name: UpdateUserMaxDevices :exec
+UPDATE users SET max_devices = $2, updated_at = NOW() WHERE id = $1;
+
+-- name: CountUserDevices :one
+SELECT COUNT(*) FROM devices WHERE user_id = $1;
+
 -- name: DeleteUser :exec
 DELETE FROM users WHERE id = $1;

@@ -720,6 +726,64 @@ DELETE FROM sync_queue WHERE id = $1;
 -- name: ClearDeviceSyncQueue :exec
 DELETE FROM sync_queue WHERE device_id = $1;

+-- name: GetFailedSyncQueueItems :many
+SELECT * FROM sync_queue
+WHERE status = 'failed' AND attempts < max_attempts
+ORDER BY priority ASC, created_at ASC
+LIMIT $1;
+
+-- name: GetStuckSyncQueueItems :many
+SELECT * FROM sync_queue
+WHERE status = 'processing' AND created_at < NOW() - INTERVAL '1 hour'
+ORDER BY priority ASC, created_at ASC;
+
+-- name: IncrementSyncQueueAttempts :one
+UPDATE sync_queue
+SET
+    attempts = attempts + 1,
+    status = CASE 
+        WHEN attempts + 1 >= max_attempts THEN 'failed'
+        ELSE 'pending'
+    END,
+    error_message = $2
+WHERE id = $1
+RETURNING *;
+
+-- name: GetSyncQueueStats :one
+SELECT
+    COUNT(*) FILTER (WHERE status = 'pending') as pending_count,
+    COUNT(*) FILTER (WHERE status = 'processing') as processing_count,
+    COUNT(*) FILTER (WHERE status = 'failed') as failed_count,
+    COUNT(*) FILTER (WHERE status = 'completed') as completed_count,
+    COUNT(*) as total_count
+FROM sync_queue
+WHERE device_id = $1;
+
+-- name: ListAllSyncQueueItems :many
+SELECT 
+    sq.*,
+    d.device_name,
+    d.device_type,
+    u.email as user_email,
+    mi.title as media_title
+FROM sync_queue sq
+JOIN devices d ON sq.device_id = d.id
+JOIN users u ON d.user_id = u.id
+LEFT JOIN media_items mi ON sq.media_item_id = mi.id
+ORDER BY sq.priority ASC, sq.created_at DESC
+LIMIT $1 OFFSET $2;
+
+-- name: GetNextRetryTime :one
+SELECT 
+    CASE 
+        WHEN attempts = 0 THEN NOW()
+        WHEN attempts = 1 THEN NOW() + INTERVAL '1 minute'
+        WHEN attempts = 2 THEN NOW() + INTERVAL '5 minutes'
+        WHEN attempts = 3 THEN NOW() + INTERVAL '15 minutes'
+        WHEN attempts = 4 THEN NOW() + INTERVAL '1 hour'
+        ELSE NOW() + INTERVAL '24 hours'
+    END as next_retry_time;
+
 -- Conflict Resolution
 -- name: CreateSyncConflict :one
 INSERT INTO sync_conflicts (media_item_id, user_id, conflict_type, conflict_data)