john-okeefe
1e05470fbb
Update all handler functions to use *echo.Context (pointer) instead of echo.Context (value) as required by Echo v5. Changes across all handler files: - analytics.go: Update handler signatures - auth.go: Update authentication handler signatures - book_matching.go: Update matching handler signatures - collections.go: Update collection handler signatures - collections_preview_test.go: Update test signatures - commonhandlers.go: Update common handler signatures - conflicts.go: Update conflict handler signatures - context.go: Update context handler signatures - dashboard.go: Update dashboard handler signatures - devices.go: Update device handler signatures - jobs.go: Update job handler signatures - kobo.go: Update Kobo handler signatures - koreader.go: Update Koreader handler signatures - library.go: Update library handler signatures - matching.go: Update matching handler signatures - media.go: Update media handler signatures - opds.go: Update OPDS handler signatures - progress.go: Update progress handler signatures - queue.go: Update queue handler signatures - refresh_token.go: Update token handler signatures - scanner.go: Update scanner handler signatures - sidecar.go: Update sidecar handler signatures - sync.go: Update sync handler signatures - system_settings.go: Update settings handler signatures - websocket.go: Update WebSocket handler signatures All handlers now properly implement Echo v5's pointer-based context pattern. This change is necessary for type safety and compatibility with Echo v5's improved context handling and WebSocket support.
331 lines
9.7 KiB
Go
331 lines
9.7 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
|
|
}
|
|
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
|
|
}
|