john-okeefe
8ac6e1ac79
Simplify pointer creation across kobo, koreader, and queue handlers by replacing the two-step pattern (assign to local, then take address) with inline new() calls. This reduces verbosity without changing behavior: Before: remaining := int(a - b) pagesRemaining = &remaining After: pagesRemaining = new(int(a - b)) Covers page calculations, chapter/progress fields, UUID formatting, and timestamp string conversions.
329 lines
9.6 KiB
Go
329 lines
9.6 KiB
Go
package handlers
|
|
|
|
import (
|
|
"bookhoard/internal/database"
|
|
"context"
|
|
"net/http"
|
|
"strconv"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/labstack/echo/v5"
|
|
)
|
|
|
|
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) GetQueueData(c *echo.Context) ([]QueueItemResponse, error) {
|
|
user := MustGetAuthenticatedUser(c)
|
|
|
|
if user.Role != "admin" {
|
|
return nil, 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 nil, err
|
|
}
|
|
|
|
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 response, nil
|
|
}
|
|
|
|
func (h *QueueHandler) ListAllQueueItems(c *echo.Context) error {
|
|
response, err := h.GetQueueData(c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return c.JSON(http.StatusOK, map[string]interface{}{
|
|
"items": response,
|
|
"count": len(response),
|
|
})
|
|
}
|
|
|
|
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
|
|
}
|
|
return new(uuid.UUID(u.Bytes).String())
|
|
}
|
|
|
|
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
|
|
}
|
|
return new(t.Time.Format("2006-01-02T15:04:05Z07:00"))
|
|
}
|