mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-13 04:14:56 +00:00
feat: Dynamically register/unregister providers in memory and recompute stale cron job next_run_at on startup.
This commit is contained in:
1 parent
3a85614e0b
commit
9b0dce2aa5
3 files changed
+101
No files matched your search
@@ -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"})
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user