193 lines
6.7 KiB
Go
193 lines
6.7 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
|
|
"git.gobha.me/xcaliber/chat-switchboard/models"
|
|
)
|
|
|
|
// HealthStore implements health.Store for Postgres.
|
|
type HealthStore struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
func NewHealthStore(db *sql.DB) *HealthStore {
|
|
return &HealthStore{db: db}
|
|
}
|
|
|
|
func (s *HealthStore) UpsertWindow(ctx context.Context, w *models.ProviderHealthWindow) error {
|
|
_, err := s.db.ExecContext(ctx, `
|
|
INSERT INTO provider_health (provider_config_id, window_start,
|
|
request_count, error_count, timeout_count, rate_limit_count,
|
|
total_latency_ms, max_latency_ms, last_error, last_error_at, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now())
|
|
ON CONFLICT (provider_config_id, window_start) DO UPDATE SET
|
|
request_count = provider_health.request_count + EXCLUDED.request_count,
|
|
error_count = provider_health.error_count + EXCLUDED.error_count,
|
|
timeout_count = provider_health.timeout_count + EXCLUDED.timeout_count,
|
|
rate_limit_count = provider_health.rate_limit_count + EXCLUDED.rate_limit_count,
|
|
total_latency_ms = provider_health.total_latency_ms + EXCLUDED.total_latency_ms,
|
|
max_latency_ms = GREATEST(provider_health.max_latency_ms, EXCLUDED.max_latency_ms),
|
|
last_error = COALESCE(EXCLUDED.last_error, provider_health.last_error),
|
|
last_error_at = COALESCE(EXCLUDED.last_error_at, provider_health.last_error_at),
|
|
updated_at = now()
|
|
`, w.ProviderConfigID, w.WindowStart,
|
|
w.RequestCount, w.ErrorCount, w.TimeoutCount, w.RateLimitCount,
|
|
w.TotalLatencyMs, w.MaxLatencyMs, w.LastError, w.LastErrorAt,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *HealthStore) GetCurrentWindow(ctx context.Context, providerConfigID string) (*models.ProviderHealthWindow, error) {
|
|
windowStart := time.Now().UTC().Truncate(time.Hour)
|
|
var w models.ProviderHealthWindow
|
|
err := s.db.QueryRowContext(ctx, `
|
|
SELECT id, provider_config_id, window_start,
|
|
request_count, error_count, timeout_count, rate_limit_count,
|
|
total_latency_ms, max_latency_ms, last_error, last_error_at
|
|
FROM provider_health
|
|
WHERE provider_config_id = $1 AND window_start = $2
|
|
`, providerConfigID, windowStart).Scan(
|
|
&w.ID, &w.ProviderConfigID, &w.WindowStart,
|
|
&w.RequestCount, &w.ErrorCount, &w.TimeoutCount, &w.RateLimitCount,
|
|
&w.TotalLatencyMs, &w.MaxLatencyMs, &w.LastError, &w.LastErrorAt,
|
|
)
|
|
if err == sql.ErrNoRows {
|
|
return nil, nil
|
|
}
|
|
return &w, err
|
|
}
|
|
|
|
func (s *HealthStore) ListWindows(ctx context.Context, providerConfigID string, hours int) ([]models.ProviderHealthWindow, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, provider_config_id, window_start,
|
|
request_count, error_count, timeout_count, rate_limit_count,
|
|
total_latency_ms, max_latency_ms, last_error, last_error_at
|
|
FROM provider_health
|
|
WHERE provider_config_id = $1
|
|
ORDER BY window_start DESC
|
|
LIMIT $2
|
|
`, providerConfigID, hours)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.ProviderHealthWindow
|
|
for rows.Next() {
|
|
var w models.ProviderHealthWindow
|
|
if err := rows.Scan(
|
|
&w.ID, &w.ProviderConfigID, &w.WindowStart,
|
|
&w.RequestCount, &w.ErrorCount, &w.TimeoutCount, &w.RateLimitCount,
|
|
&w.TotalLatencyMs, &w.MaxLatencyMs, &w.LastError, &w.LastErrorAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, w)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *HealthStore) ListAllCurrent(ctx context.Context) ([]models.ProviderHealthWindow, error) {
|
|
windowStart := time.Now().UTC().Truncate(time.Hour)
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, provider_config_id, window_start,
|
|
request_count, error_count, timeout_count, rate_limit_count,
|
|
total_latency_ms, max_latency_ms, last_error, last_error_at
|
|
FROM provider_health
|
|
WHERE window_start = $1
|
|
ORDER BY provider_config_id
|
|
`, windowStart)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.ProviderHealthWindow
|
|
for rows.Next() {
|
|
var w models.ProviderHealthWindow
|
|
if err := rows.Scan(
|
|
&w.ID, &w.ProviderConfigID, &w.WindowStart,
|
|
&w.RequestCount, &w.ErrorCount, &w.TimeoutCount, &w.RateLimitCount,
|
|
&w.TotalLatencyMs, &w.MaxLatencyMs, &w.LastError, &w.LastErrorAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, w)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *HealthStore) Prune(ctx context.Context, before time.Time) (int64, error) {
|
|
result, err := s.db.ExecContext(ctx, `
|
|
DELETE FROM provider_health WHERE window_start < $1
|
|
`, before)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
// Also prune tool health
|
|
s.db.ExecContext(ctx, `DELETE FROM tool_health WHERE window_start < $1`, before)
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
// ── Tool Health (v0.22.4) ────────────────────
|
|
|
|
func (s *HealthStore) UpsertToolWindow(ctx context.Context, w *models.ToolHealthWindow) error {
|
|
_, err := s.db.ExecContext(ctx, `
|
|
INSERT INTO tool_health (tool_name, window_start,
|
|
request_count, error_count, total_latency_ms, max_latency_ms,
|
|
last_error, last_error_at, updated_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())
|
|
ON CONFLICT (tool_name, window_start) DO UPDATE SET
|
|
request_count = tool_health.request_count + EXCLUDED.request_count,
|
|
error_count = tool_health.error_count + EXCLUDED.error_count,
|
|
total_latency_ms = tool_health.total_latency_ms + EXCLUDED.total_latency_ms,
|
|
max_latency_ms = GREATEST(tool_health.max_latency_ms, EXCLUDED.max_latency_ms),
|
|
last_error = COALESCE(EXCLUDED.last_error, tool_health.last_error),
|
|
last_error_at = COALESCE(EXCLUDED.last_error_at, tool_health.last_error_at),
|
|
updated_at = now()
|
|
`, w.ToolName, w.WindowStart,
|
|
w.RequestCount, w.ErrorCount, w.TotalLatencyMs, w.MaxLatencyMs,
|
|
w.LastError, w.LastErrorAt,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *HealthStore) ListAllToolCurrent(ctx context.Context) ([]models.ToolHealthWindow, error) {
|
|
windowStart := time.Now().UTC().Truncate(time.Hour)
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT id, tool_name, window_start,
|
|
request_count, error_count, total_latency_ms, max_latency_ms,
|
|
last_error, last_error_at
|
|
FROM tool_health
|
|
WHERE window_start = $1
|
|
ORDER BY tool_name
|
|
`, windowStart)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []models.ToolHealthWindow
|
|
for rows.Next() {
|
|
var w models.ToolHealthWindow
|
|
if err := rows.Scan(
|
|
&w.ID, &w.ToolName, &w.WindowStart,
|
|
&w.RequestCount, &w.ErrorCount, &w.TotalLatencyMs, &w.MaxLatencyMs,
|
|
&w.LastError, &w.LastErrorAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, w)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
// DeactivateProvider marks a provider config as inactive (for auto-disable).
|
|
func (s *HealthStore) DeactivateProvider(ctx context.Context, configID string) error {
|
|
_, err := s.db.ExecContext(ctx, `UPDATE provider_configs SET is_active = false WHERE id = $1`, configID)
|
|
return err
|
|
}
|