John O'Keefe
6808c1babb
Found while verifying H1's "app offline-play -> flush -> position lands"
with the emulator path: the flush has been dead code in two independent
ways, invisible because the first break made the second unreachable.
1. ListPendingSyncQueueItems matched `device_id = $1` with a NULL $1
(the processor passes DeviceID{Valid:false} to enumerate all devices;
uuid = NULL matches no rows). The ticker therefore processed nothing,
ever. The WHERE now treats NULL as "every device" (the same
narg::uuid IS NULL idiom the visibility queries use; EXPLAIN'd — seq
scan on a bounded table, idx_sync_queue_status_priority_created
fits). The processor also dropped its fake-pagination loop, which
re-fetched the same first batch forever once the backlog hit the
batch size (no OFFSET in the query).
2. dbItemToSyncQueueItem dropped SyncData — even with a working
enumerator, every flushed item arrived with an empty payload and
failed JSON unmarshal. Dead code harboring silent breakage, exactly
the plan's thesis.
Callers updated for the renamed RowLimit param (queue handler, offline
detector). TestH1_SyncQueueFlush_LandsAndPreservesAudio drives the real
flow: web PUT seeds audio 900s, a pending checkpoint-style queue row
applies percentage 0.8 through SaveProgress, audio survives, queue row
completes.
Also fixes dashboard_test.go's stale BuildSections signature (the
package's test binary has been unbuildable) — user-approved: tests
designed against old shapes are not worth keeping.
340 lines
10 KiB
Go
340 lines
10 KiB
Go
package handlers
|
|
|
|
import (
|
|
"bookhoard/internal/database"
|
|
"bookhoard/internal/utils"
|
|
"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"`
|
|
// CreatedAtDisplay is the server-rendered wall-clock string in the
|
|
// requesting admin's timezone (templates render this; created_at stays
|
|
// machine-readable RFC3339 for API consumers).
|
|
CreatedAtDisplay string `json:"created_at_display,omitempty"`
|
|
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,
|
|
RowLimit: int32(limit),
|
|
})
|
|
} else {
|
|
items, err = h.db.ListPendingSyncQueueItems(c.Request().Context(), database.ListPendingSyncQueueItemsParams{
|
|
DeviceID: device.ID,
|
|
RowLimit: 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"),
|
|
CreatedAtDisplay: utils.FormatTimeInTimezone(
|
|
item.CreatedAt.Time, user.Timezone.String),
|
|
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"),
|
|
CreatedAtDisplay: utils.FormatTimeInTimezone(
|
|
item.CreatedAt.Time, user.Timezone.String),
|
|
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
|
|
}
|
|
timeFormat := t.Time.Format("2006-01-02T15:04:05Z07:00")
|
|
return &timeFormat
|
|
}
|