Replaces the sequential worker with a download-manager-style pool (§0 item 4). Model: the user's connection is the limit; semantics are fair-share ACROSS feeds, not goroutine-per-book (a global FIFO would make Book B wait behind Book A's 100 files). - Atomic claim: GetNextQueuedEpisode + SetPodcastEpisodeDownloading was a read-then-write race under concurrency; ClaimQueuedPodcastEpisode converts queued→downloading in ONE guarded statement (two claimants, one winner — the loser gets ErrNoRows). SetPodcastEpisodeDownloading retired. - Queue tickets: download_queue_pos BIGINT fed by a Postgres SEQUENCE stamped at enqueue — the enqueue loop's ordering IS the claim order (the foundation §0 item 5 builds on; previously all rows in a batch shared one updated_at, so within-feed order was arbitrary). - Pool: up to podcast_download_concurrency (default 4, live range 1–16) claim-loop goroutines per drain round; the setting re-reads every round so admin changes take effect without restart. - Fair share: claim prefers the feed least-recently-served (pickFeedToServe; never-served first), earliest ticket within a feed — books proceed in parallel while chapters land in order (start listening while the rest download). - Bandwidth: shared token bucket across all workers (podcast_download_speed_mbps, default 0 = unlimited); take() reserves bandwidth and callers sleep outside the lock. - Settings: new float setting type end to end (registry defaults + clamping, save validation, admin UI number-with-step input) + the two schema seeds; worker wires the registry via SetSettings (tests run on registered defaults). - Tests: unit coverage for the bucket reservation math + fair-share pick; integration TestPodcastDownloadPool pins ticket order = enqueue order, claim exclusivity, and a full 3-episode pool drain. Dev-DB note: the column/sequence/settings were applied live (idempotent DDL, same statements the schema runs at next boot) since the running container predates them.
314 lines
9.9 KiB
Go
314 lines
9.9 KiB
Go
package handlers
|
|
|
|
import (
|
|
"bookhoard/internal/database"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/labstack/echo/v5"
|
|
)
|
|
|
|
type SystemSettingsHandler struct {
|
|
db *database.Queries
|
|
settings *database.SettingsRegistry
|
|
onScanSettingsChanged func()
|
|
}
|
|
|
|
func NewSystemSettingsHandler(db *database.Queries) *SystemSettingsHandler {
|
|
return &SystemSettingsHandler{
|
|
db: db,
|
|
}
|
|
}
|
|
|
|
// SetScanSettingsChanged registers a callback fired after scan settings are
|
|
// successfully saved on any write path, so listeners (the scanner) can drop
|
|
// their cached values immediately instead of waiting out the cache TTL.
|
|
func (h *SystemSettingsHandler) SetScanSettingsChanged(fn func()) {
|
|
h.onScanSettingsChanged = fn
|
|
}
|
|
|
|
func (h *SystemSettingsHandler) notifyScanSettingsChanged() {
|
|
if h.onScanSettingsChanged != nil {
|
|
h.onScanSettingsChanged()
|
|
}
|
|
}
|
|
|
|
// SetSettings wires the tunable settings registry. Required for the unified
|
|
// /api/system/settings endpoints and for cache invalidation after writes.
|
|
func (h *SystemSettingsHandler) SetSettings(s *database.SettingsRegistry) {
|
|
h.settings = s
|
|
}
|
|
|
|
// reload refreshes the in-memory cache after a write.
|
|
func (h *SystemSettingsHandler) reload(c *echo.Context) {
|
|
if h.settings != nil {
|
|
h.settings.Reload(c.Request().Context())
|
|
}
|
|
}
|
|
|
|
// ---- Unified /api/system/settings endpoints ----
|
|
|
|
// GetSettings handles GET /api/system/settings.
|
|
func (h *SystemSettingsHandler) GetSettings(c *echo.Context) error {
|
|
if h.settings == nil {
|
|
return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": "settings registry not initialized"})
|
|
}
|
|
return c.JSON(http.StatusOK, h.settings.All())
|
|
}
|
|
|
|
// UpdateSettingRequest is the body for PUT /api/system/settings.
|
|
type UpdateSettingRequest struct {
|
|
Key string `json:"key" form:"key"`
|
|
Value string `json:"value" form:"value"`
|
|
}
|
|
|
|
// UpdateSettingResponse mirrors a settings entry plus a reload hint.
|
|
type UpdateSettingResponse struct {
|
|
database.SettingEntry
|
|
ReloadRequired bool `json:"reload_required"`
|
|
Message string `json:"message,omitempty"`
|
|
}
|
|
|
|
// UpdateSetting handles PUT /api/system/settings.
|
|
func (h *SystemSettingsHandler) UpdateSetting(c *echo.Context) error {
|
|
if h.settings == nil {
|
|
return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": "settings registry not initialized"})
|
|
}
|
|
var req UpdateSettingRequest
|
|
if err := c.Bind(&req); err != nil {
|
|
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
|
|
}
|
|
resp, err := h.ApplySetting(c.Request().Context(), req.Key, req.Value)
|
|
if err != nil {
|
|
return c.JSON(http.StatusBadRequest, map[string]string{"error": err.Error()})
|
|
}
|
|
return c.JSON(http.StatusOK, resp)
|
|
}
|
|
|
|
// ApplySetting validates, persists, and reloads a single setting. Shared by the
|
|
// JSON API and the HTMX admin endpoint.
|
|
func (h *SystemSettingsHandler) ApplySetting(ctx context.Context, key, value string) (UpdateSettingResponse, error) {
|
|
if h.settings == nil {
|
|
return UpdateSettingResponse{}, fmt.Errorf("settings registry not initialized")
|
|
}
|
|
if key == "" {
|
|
return UpdateSettingResponse{}, fmt.Errorf("key is required")
|
|
}
|
|
def, ok := database.LookupDefault(key)
|
|
if !ok {
|
|
return UpdateSettingResponse{}, fmt.Errorf("unknown setting key: %s", key)
|
|
}
|
|
if err := validateSettingValue(def, value); err != nil {
|
|
return UpdateSettingResponse{}, err
|
|
}
|
|
|
|
desc := def.Description
|
|
rType := pgtype.Text{}
|
|
if def.Type != "" {
|
|
rType = pgtype.Text{String: def.Type, Valid: true}
|
|
}
|
|
var minP, maxP pgtype.Text
|
|
if def.Min != "" {
|
|
minP = pgtype.Text{String: def.Min, Valid: true}
|
|
}
|
|
if def.Max != "" {
|
|
maxP = pgtype.Text{String: def.Max, Valid: true}
|
|
}
|
|
if _, err := h.db.UpsertSystemSetting(ctx, database.UpsertSystemSettingParams{
|
|
SettingKey: key,
|
|
SettingValue: value,
|
|
Description: pgtype.Text{String: desc, Valid: desc != ""},
|
|
SettingType: rType,
|
|
MinValue: minP,
|
|
MaxValue: maxP,
|
|
RequiresRestart: pgtype.Bool{Bool: def.RequiresRestart, Valid: true},
|
|
Category: pgtype.Text{String: def.Category, Valid: def.Category != ""},
|
|
}); err != nil {
|
|
return UpdateSettingResponse{}, err
|
|
}
|
|
|
|
h.settings.Reload(ctx)
|
|
|
|
// Scan keys are also cached inside MediaScanner (30s TTL); let it drop
|
|
// its cache so the change takes effect on the next poll.
|
|
if key == "scan_poll_interval_seconds" || key == "auto_scan_enabled" || key == "scan_extractor_version" {
|
|
h.notifyScanSettingsChanged()
|
|
}
|
|
|
|
resp := UpdateSettingResponse{ReloadRequired: def.RequiresRestart}
|
|
for _, e := range h.settings.All() {
|
|
if e.Key == key {
|
|
resp.SettingEntry = e
|
|
break
|
|
}
|
|
}
|
|
if def.RequiresRestart {
|
|
resp.Message = "Saved. Restart the server for this change to take full effect."
|
|
} else {
|
|
resp.Message = "Saved."
|
|
}
|
|
return resp, nil
|
|
}
|
|
|
|
// validateSettingValue checks a candidate value against the setting's type and bounds.
|
|
func validateSettingValue(def database.SettingDefault, value string) error {
|
|
switch def.Type {
|
|
case database.SettingTypeInt:
|
|
n, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
return fmt.Errorf("value must be an integer")
|
|
}
|
|
if def.Min != "" {
|
|
if mn, err := strconv.Atoi(def.Min); err == nil && n < mn {
|
|
return fmt.Errorf("value must be >= %s", def.Min)
|
|
}
|
|
}
|
|
if def.Max != "" {
|
|
if mx, err := strconv.Atoi(def.Max); err == nil && n > mx {
|
|
return fmt.Errorf("value must be <= %s", def.Max)
|
|
}
|
|
}
|
|
case database.SettingTypeBool:
|
|
if _, err := strconv.ParseBool(value); err != nil {
|
|
return fmt.Errorf("value must be true or false")
|
|
}
|
|
case database.SettingTypeFloat:
|
|
f, err := strconv.ParseFloat(value, 64)
|
|
if err != nil {
|
|
return fmt.Errorf("value must be a number")
|
|
}
|
|
if def.Min != "" {
|
|
if mn, err := strconv.ParseFloat(def.Min, 64); err == nil && f < mn {
|
|
return fmt.Errorf("value must be >= %s", def.Min)
|
|
}
|
|
}
|
|
if def.Max != "" {
|
|
if mx, err := strconv.ParseFloat(def.Max, 64); err == nil && f > mx {
|
|
return fmt.Errorf("value must be <= %s", def.Max)
|
|
}
|
|
}
|
|
case database.SettingTypeString:
|
|
if value == "" {
|
|
return fmt.Errorf("value must not be empty")
|
|
}
|
|
if def.Key == "default_timezone" {
|
|
if _, err := time.LoadLocation(value); err != nil {
|
|
return fmt.Errorf("invalid timezone: %v", err)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ---- Legacy scan-settings endpoints (retained for backward compatibility) ----
|
|
|
|
type UpdateScanSettingsRequest struct {
|
|
ScanPollIntervalSeconds int32 `json:"scan_poll_interval_seconds" validate:"required,min=1,max=3600"`
|
|
AutoScanEnabled bool `json:"auto_scan_enabled"`
|
|
}
|
|
|
|
type ScanSettingsResponse struct {
|
|
ScanPollIntervalSeconds int32 `json:"scan_poll_interval_seconds"`
|
|
AutoScanEnabled bool `json:"auto_scan_enabled"`
|
|
Message string `json:"message,omitempty"`
|
|
}
|
|
|
|
func (h *SystemSettingsHandler) UpdateScanSettings(c *echo.Context) error {
|
|
var req UpdateScanSettingsRequest
|
|
if err := c.Bind(&req); err != nil {
|
|
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid request"})
|
|
}
|
|
if err := c.Validate(&req); err != nil {
|
|
return c.JSON(http.StatusBadRequest, map[string]string{"error": err.Error()})
|
|
}
|
|
|
|
scanFrequencyValue := strconv.FormatInt(int64(req.ScanPollIntervalSeconds), 10)
|
|
autoScanValue := strconv.FormatBool(req.AutoScanEnabled)
|
|
|
|
err := h.db.UpdateSystemSetting(c.Request().Context(), database.UpdateSystemSettingParams{
|
|
SettingKey: "scan_poll_interval_seconds",
|
|
SettingValue: scanFrequencyValue,
|
|
})
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return c.JSON(http.StatusNotFound, map[string]string{"error": "system setting not found"})
|
|
}
|
|
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
|
}
|
|
|
|
err = h.db.UpdateSystemSetting(c.Request().Context(), database.UpdateSystemSettingParams{
|
|
SettingKey: "auto_scan_enabled",
|
|
SettingValue: autoScanValue,
|
|
})
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return c.JSON(http.StatusNotFound, map[string]string{"error": "system setting not found"})
|
|
}
|
|
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
|
}
|
|
|
|
h.notifyScanSettingsChanged()
|
|
|
|
h.reload(c)
|
|
|
|
return c.JSON(http.StatusOK, ScanSettingsResponse{
|
|
ScanPollIntervalSeconds: req.ScanPollIntervalSeconds,
|
|
AutoScanEnabled: req.AutoScanEnabled,
|
|
Message: "scan settings updated successfully",
|
|
})
|
|
}
|
|
|
|
func (h *SystemSettingsHandler) GetScanSettings(c *echo.Context) error {
|
|
// Prefer the registry (single source of truth after Load).
|
|
if h.settings != nil {
|
|
interval := int32(h.settings.ScanPollInterval().Seconds())
|
|
return c.JSON(http.StatusOK, ScanSettingsResponse{
|
|
ScanPollIntervalSeconds: interval,
|
|
AutoScanEnabled: h.settings.AutoScanEnabled(),
|
|
})
|
|
}
|
|
|
|
scanFrequencySetting, err := h.db.GetSystemSetting(c.Request().Context(), "scan_poll_interval_seconds")
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return c.JSON(http.StatusOK, ScanSettingsResponse{
|
|
ScanPollIntervalSeconds: 1800,
|
|
AutoScanEnabled: true,
|
|
})
|
|
}
|
|
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
|
}
|
|
|
|
autoScanSetting, err := h.db.GetSystemSetting(c.Request().Context(), "auto_scan_enabled")
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return c.JSON(http.StatusOK, ScanSettingsResponse{
|
|
ScanPollIntervalSeconds: 1800,
|
|
AutoScanEnabled: true,
|
|
})
|
|
}
|
|
return c.JSON(http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
|
}
|
|
|
|
scanFrequency, err := strconv.ParseInt(scanFrequencySetting, 10, 32)
|
|
if err != nil {
|
|
return c.JSON(http.StatusInternalServerError, map[string]string{"error": "invalid scan frequency setting"})
|
|
}
|
|
|
|
autoScanEnabled, err := strconv.ParseBool(autoScanSetting)
|
|
if err != nil {
|
|
return c.JSON(http.StatusInternalServerError, map[string]string{"error": "invalid auto scan setting"})
|
|
}
|
|
|
|
return c.JSON(http.StatusOK, ScanSettingsResponse{
|
|
ScanPollIntervalSeconds: int32(scanFrequency),
|
|
AutoScanEnabled: autoScanEnabled,
|
|
})
|
|
}
|