diff --git a/internal/http/providers.go b/internal/http/providers.go index e02389f3..2dac589c 100644 --- a/internal/http/providers.go +++ b/internal/http/providers.go @@ -58,6 +58,19 @@ func maskAPIKey(p *store.LLMProviderData) { } } +// registerInMemory adds (or replaces) a provider in the in-memory registry +// so it's immediately usable for verify/chat without a gateway restart. +func (h *ProvidersHandler) registerInMemory(p *store.LLMProviderData) { + if h.providerReg == nil || !p.Enabled || p.APIKey == "" { + return + } + if p.ProviderType == "anthropic_native" { + h.providerReg.Register(providers.NewAnthropicProvider(p.APIKey)) + } else { + h.providerReg.Register(providers.NewOpenAIProvider(p.Name, p.APIKey, p.APIBase, "")) + } +} + // --- Provider CRUD --- func (h *ProvidersHandler) handleListProviders(w http.ResponseWriter, r *http.Request) { @@ -101,6 +114,9 @@ func (h *ProvidersHandler) handleCreateProvider(w http.ResponseWriter, r *http.R return } + // Register in-memory so verify/chat work without restart + h.registerInMemory(&p) + maskAPIKey(&p) writeJSON(w, http.StatusCreated, p) } @@ -168,6 +184,17 @@ func (h *ProvidersHandler) handleUpdateProvider(w http.ResponseWriter, r *http.R return } + // Sync in-memory registry with updated provider + if h.providerReg != nil { + if updated, err := h.store.GetProvider(r.Context(), id); err == nil { + if !updated.Enabled { + h.providerReg.Unregister(updated.Name) + } else { + h.registerInMemory(updated) + } + } + } + writeJSON(w, http.StatusOK, map[string]string{"status": "updated"}) } @@ -178,11 +205,21 @@ func (h *ProvidersHandler) handleDeleteProvider(w http.ResponseWriter, r *http.R return } + // Read provider name before deleting so we can unregister it + var providerName string + if p, err := h.store.GetProvider(r.Context(), id); err == nil { + providerName = p.Name + } + if err := h.store.DeleteProvider(r.Context(), id); err != nil { slog.Error("providers.delete", "error", err) writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } + if h.providerReg != nil && providerName != "" { + h.providerReg.Unregister(providerName) + } + writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"}) } diff --git a/internal/providers/registry.go b/internal/providers/registry.go index 85e690f3..d7ec0024 100644 --- a/internal/providers/registry.go +++ b/internal/providers/registry.go @@ -24,6 +24,13 @@ func (r *Registry) Register(provider Provider) { r.providers[provider.Name()] = provider } +// Unregister removes a provider from the registry. +func (r *Registry) Unregister(name string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.providers, name) +} + // Get returns a provider by name. func (r *Registry) Get(name string) (Provider, error) { r.mu.RLock() diff --git a/internal/store/pg/cron.go b/internal/store/pg/cron.go index 1fc3c775..a94fda48 100644 --- a/internal/store/pg/cron.go +++ b/internal/store/pg/cron.go @@ -323,6 +323,7 @@ func (s *PGCronStore) Start() error { } s.stop = make(chan struct{}) s.running = true + s.recomputeStaleJobs() go s.runLoop() slog.Info("pg cron service started") return nil @@ -416,6 +417,62 @@ func (s *PGCronStore) InvalidateCache() { s.mu.Unlock() } +// recomputeStaleJobs fixes enabled jobs that have next_run_at = NULL. +// This happens when the gateway was stopped/crashed while a job was executing, +// or when the previously computed next_run_at was consumed but never recomputed. +// Mirrors the startup logic in cron.Service.Start() (standalone mode). +func (s *PGCronStore) recomputeStaleJobs() { + rows, err := s.db.Query( + `SELECT id, schedule_kind, cron_expression, run_at, timezone, interval_ms + FROM cron_jobs WHERE enabled = true AND next_run_at IS NULL`) + if err != nil { + slog.Warn("cron: failed to query stale jobs", "error", err) + return + } + defer rows.Close() + + now := time.Now() + var fixed int + for rows.Next() { + var id uuid.UUID + var scheduleKind string + var cronExpr, tz *string + var runAt *time.Time + var intervalMS *int64 + + if err := rows.Scan(&id, &scheduleKind, &cronExpr, &runAt, &tz, &intervalMS); err != nil { + continue + } + + schedule := store.CronSchedule{Kind: scheduleKind} + if cronExpr != nil { + schedule.Expr = *cronExpr + } + if runAt != nil { + ms := runAt.UnixMilli() + schedule.AtMS = &ms + } + if intervalMS != nil { + schedule.EveryMS = intervalMS + } + + next := computeNextRun(&schedule, now) + if next == nil { + if scheduleKind == "at" { + s.db.Exec("UPDATE cron_jobs SET enabled = false, updated_at = $1 WHERE id = $2", now, id) + } + continue + } + + s.db.Exec("UPDATE cron_jobs SET next_run_at = $1, updated_at = $2 WHERE id = $3", *next, now, id) + fixed++ + } + + if fixed > 0 { + slog.Info("cron: recomputed stale next_run_at on startup", "fixed", fixed) + } +} + // --- Internal --- func (s *PGCronStore) scanJob(id uuid.UUID) (*store.CronJob, error) {