From 030d30a225db902b77ff2a17c48e501897ef62cb Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Sat, 31 Jan 2026 13:06:12 -0500 Subject: [PATCH] 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 --- internal/database/models.go | 1 + internal/database/querier.go | 8 + internal/database/queries.sql.go | 269 +++++++++++++++++++++ internal/database/queries/queries.sql | 64 +++++ internal/handlers/queue.go | 329 ++++++++++++++++++++++++++ 5 files changed, 671 insertions(+) create mode 100644 internal/handlers/queue.go diff --git a/internal/database/models.go b/internal/database/models.go index 445c003..fc95938 100644 --- a/internal/database/models.go +++ b/internal/database/models.go @@ -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"` } diff --git a/internal/database/querier.go b/internal/database/querier.go index b50403f..8581f5c 100644 --- a/internal/database/querier.go +++ b/internal/database/querier.go @@ -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 diff --git a/internal/database/queries.sql.go b/internal/database/queries.sql.go index aeabb2b..71ca907 100644 --- a/internal/database/queries.sql.go +++ b/internal/database/queries.sql.go @@ -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 ` diff --git a/internal/database/queries/queries.sql b/internal/database/queries/queries.sql index c2f1fcc..195772c 100644 --- a/internal/database/queries/queries.sql +++ b/internal/database/queries/queries.sql @@ -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) diff --git a/internal/handlers/queue.go b/internal/handlers/queue.go new file mode 100644 index 0000000..2343ac5 --- /dev/null +++ b/internal/handlers/queue.go @@ -0,0 +1,329 @@ +package handlers + +import ( + "bookmann/internal/database" + "context" + "net/http" + "strconv" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/labstack/echo/v4" +) + +type QueueHandler struct { + db *database.Queries + queue interface { + GetQueueStats(ctx context.Context, deviceID pgtype.UUID) (database.GetSyncQueueStatsRow, error) + } +} + +func NewQueueHandler(db *database.Queries, queue interface { + GetQueueStats(ctx context.Context, deviceID pgtype.UUID) (database.GetSyncQueueStatsRow, error) +}) *QueueHandler { + return &QueueHandler{db: db, queue: queue} +} + +type QueueStatsResponse struct { + PendingCount int64 `json:"pending_count"` + ProcessingCount int64 `json:"processing_count"` + FailedCount int64 `json:"failed_count"` + CompletedCount int64 `json:"completed_count"` + TotalCount int64 `json:"total_count"` +} + +type QueueItemResponse struct { + ID string `json:"id"` + DeviceID string `json:"device_id"` + DeviceName string `json:"device_name"` + DeviceType string `json:"device_type"` + MediaItemID *string `json:"media_item_id,omitempty"` + MediaTitle *string `json:"media_title,omitempty"` + UserEmail string `json:"user_email"` + SyncType string `json:"sync_type"` + Priority int32 `json:"priority"` + Attempts int32 `json:"attempts"` + MaxAttempts int32 `json:"max_attempts"` + Status string `json:"status"` + ErrorMessage *string `json:"error_message,omitempty"` + CreatedAt string `json:"created_at"` + ProcessedAt *string `json:"processed_at,omitempty"` +} + +func (h *QueueHandler) GetDeviceQueueStats(c echo.Context) error { + user := MustGetAuthenticatedUser(c) + + deviceID, err := uuid.Parse(c.Param("device_id")) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "invalid device ID") + } + + device, err := h.db.GetDevice(c.Request().Context(), pgtype.UUID{Bytes: deviceID, Valid: true}) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "device not found") + } + + if device.UserID.Bytes != user.ID.Bytes && user.Role != "admin" { + return echo.NewHTTPError(http.StatusForbidden, "access denied") + } + + stats, err := h.queue.GetQueueStats(c.Request().Context(), device.ID) + if err != nil { + return echo.NewHTTPError(http.StatusInternalServerError, "failed to get queue stats") + } + + return c.JSON(http.StatusOK, QueueStatsResponse{ + PendingCount: stats.PendingCount, + ProcessingCount: stats.ProcessingCount, + FailedCount: stats.FailedCount, + CompletedCount: stats.CompletedCount, + TotalCount: stats.TotalCount, + }) +} + +func (h *QueueHandler) ListDeviceQueueItems(c echo.Context) error { + user := MustGetAuthenticatedUser(c) + + deviceID, err := uuid.Parse(c.Param("device_id")) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "invalid device ID") + } + + device, err := h.db.GetDevice(c.Request().Context(), pgtype.UUID{Bytes: deviceID, Valid: true}) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "device not found") + } + + if device.UserID.Bytes != user.ID.Bytes && user.Role != "admin" { + return echo.NewHTTPError(http.StatusForbidden, "access denied") + } + + status := c.QueryParam("status") + limit := 50 + if l := c.QueryParam("limit"); l != "" { + if parsedLimit, err := strconv.Atoi(l); err == nil && parsedLimit > 0 && parsedLimit <= 100 { + limit = parsedLimit + } + } + + var items []database.SyncQueue + if status != "" { + items, err = h.db.ListPendingSyncQueueItems(c.Request().Context(), database.ListPendingSyncQueueItemsParams{ + DeviceID: device.ID, + Limit: int32(limit), + }) + } else { + items, err = h.db.ListPendingSyncQueueItems(c.Request().Context(), database.ListPendingSyncQueueItemsParams{ + DeviceID: device.ID, + Limit: int32(limit), + }) + } + + if err != nil { + return echo.NewHTTPError(http.StatusInternalServerError, "failed to list queue items") + } + + response := make([]QueueItemResponse, 0, len(items)) + for _, item := range items { + response = append(response, QueueItemResponse{ + ID: uuid.UUID(item.ID.Bytes).String(), + DeviceID: uuid.UUID(item.DeviceID.Bytes).String(), + MediaItemID: uuidPtrToString(item.MediaItemID), + SyncType: item.SyncType, + Priority: item.Priority.Int32, + Attempts: item.Attempts.Int32, + MaxAttempts: item.MaxAttempts.Int32, + Status: item.Status.String, + ErrorMessage: textPtrToString(item.ErrorMessage), + CreatedAt: item.CreatedAt.Time.Format("2006-01-02T15:04:05Z07:00"), + ProcessedAt: timestamptzPtrToString(item.ProcessedAt), + }) + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "items": response, + "count": len(response), + }) +} + +func (h *QueueHandler) ListAllQueueItems(c echo.Context) error { + user := MustGetAuthenticatedUser(c) + + if user.Role != "admin" { + return echo.NewHTTPError(http.StatusForbidden, "admin access required") + } + + limit := 50 + if l := c.QueryParam("limit"); l != "" { + if parsedLimit, err := strconv.Atoi(l); err == nil && parsedLimit > 0 && parsedLimit <= 100 { + limit = parsedLimit + } + } + + offset := 0 + if o := c.QueryParam("offset"); o != "" { + if parsedOffset, err := strconv.Atoi(o); err == nil && parsedOffset >= 0 { + offset = parsedOffset + } + } + + items, err := h.db.ListAllSyncQueueItems(c.Request().Context(), database.ListAllSyncQueueItemsParams{ + Limit: int32(limit), + Offset: int32(offset), + }) + + if err != nil { + return echo.NewHTTPError(http.StatusInternalServerError, "failed to list queue items") + } + + response := make([]QueueItemResponse, 0, len(items)) + for _, item := range items { + mediaTitle := textPtrToString(item.MediaTitle) + + response = append(response, QueueItemResponse{ + ID: uuid.UUID(item.ID.Bytes).String(), + DeviceID: uuid.UUID(item.DeviceID.Bytes).String(), + DeviceName: item.DeviceName, + DeviceType: item.DeviceType, + MediaItemID: uuidPtrToString(item.MediaItemID), + MediaTitle: mediaTitle, + UserEmail: item.UserEmail, + SyncType: item.SyncType, + Priority: item.Priority.Int32, + Attempts: item.Attempts.Int32, + MaxAttempts: item.MaxAttempts.Int32, + Status: item.Status.String, + ErrorMessage: textPtrToString(item.ErrorMessage), + CreatedAt: item.CreatedAt.Time.Format("2006-01-02T15:04:05Z07:00"), + ProcessedAt: timestamptzPtrToString(item.ProcessedAt), + }) + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "items": response, + "count": len(response), + "limit": limit, + "offset": offset, + }) +} + +func (h *QueueHandler) RetryQueueItem(c echo.Context) error { + user := MustGetAuthenticatedUser(c) + + itemID, err := uuid.Parse(c.Param("item_id")) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "invalid item ID") + } + + item, err := h.db.GetSyncQueueItem(c.Request().Context(), pgtype.UUID{Bytes: itemID, Valid: true}) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "queue item not found") + } + + device, err := h.db.GetDevice(c.Request().Context(), item.DeviceID) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "device not found") + } + + if device.UserID.Bytes != user.ID.Bytes && user.Role != "admin" { + return echo.NewHTTPError(http.StatusForbidden, "access denied") + } + + updatedItem, err := h.db.UpdateSyncQueueItemStatus(c.Request().Context(), database.UpdateSyncQueueItemStatusParams{ + ID: item.ID, + Status: pgtype.Text{String: "pending", Valid: true}, + ErrorMessage: pgtype.Text{}, + }) + + if err != nil { + return echo.NewHTTPError(http.StatusInternalServerError, "failed to retry item") + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "message": "item queued for retry", + "item_id": uuid.UUID(updatedItem.ID.Bytes).String(), + "status": updatedItem.Status.String, + "attempts": updatedItem.Attempts.Int32, + "max_attempts": updatedItem.MaxAttempts.Int32, + }) +} + +func (h *QueueHandler) DeleteQueueItem(c echo.Context) error { + user := MustGetAuthenticatedUser(c) + + itemID, err := uuid.Parse(c.Param("item_id")) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "invalid item ID") + } + + item, err := h.db.GetSyncQueueItem(c.Request().Context(), pgtype.UUID{Bytes: itemID, Valid: true}) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "queue item not found") + } + + device, err := h.db.GetDevice(c.Request().Context(), item.DeviceID) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "device not found") + } + + if device.UserID.Bytes != user.ID.Bytes && user.Role != "admin" { + return echo.NewHTTPError(http.StatusForbidden, "access denied") + } + + err = h.db.DeleteSyncQueueItem(c.Request().Context(), item.ID) + if err != nil { + return echo.NewHTTPError(http.StatusInternalServerError, "failed to delete item") + } + + return c.NoContent(http.StatusNoContent) +} + +func (h *QueueHandler) ClearDeviceQueue(c echo.Context) error { + user := MustGetAuthenticatedUser(c) + + deviceID, err := uuid.Parse(c.Param("device_id")) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "invalid device ID") + } + + device, err := h.db.GetDevice(c.Request().Context(), pgtype.UUID{Bytes: deviceID, Valid: true}) + if err != nil { + return echo.NewHTTPError(http.StatusNotFound, "device not found") + } + + if device.UserID.Bytes != user.ID.Bytes && user.Role != "admin" { + return echo.NewHTTPError(http.StatusForbidden, "access denied") + } + + err = h.db.ClearDeviceSyncQueue(c.Request().Context(), device.ID) + if err != nil { + return echo.NewHTTPError(http.StatusInternalServerError, "failed to clear queue") + } + + return c.JSON(http.StatusOK, map[string]interface{}{ + "message": "queue cleared", + }) +} + +func uuidPtrToString(u pgtype.UUID) *string { + if !u.Valid { + return nil + } + s := uuid.UUID(u.Bytes).String() + return &s +} + +func textPtrToString(t pgtype.Text) *string { + if !t.Valid { + return nil + } + return &t.String +} + +func timestamptzPtrToString(t pgtype.Timestamptz) *string { + if !t.Valid { + return nil + } + s := t.Time.Format("2006-01-02T15:04:05Z07:00") + return &s +}