f87b954f27
- internal/extsvc: circuit breaker (§11.16) + retrying bearer JSON client - atomizer client: §5.1 contract incl. defensive coefficient normalization and extension coefficient range (0,2] - persisted jobs collection: atomic claim, panic recovery, exponential backoff (max 3), stale-running requeue, shutdown-safe bookkeeping - subdivide (async 202, atomizing status, re-run replaces unpublished children after confirm) and extend (sibling task, source budget) - failure path reverts status and notifies the consultant on final attempt - PATCH task editing with bounty recompute, budget cascade to descendants - publish single + bulk; task detail endpoint role-scoped - coder/websocket hub: multiplexed channels, origin check, 30s heartbeats; client ws.js with reconnect + 15s polling fallback - consultant atomization board: tree by root, coefficient sliders with live per-parent sum indicator, bounty preview, modals, shimmer, atomizer-down gating via /api/v1/service-health - fix: tasks external-ref unique index is partial, not sparse (compound sparse indexed every task through customerId and broke child inserts) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
104 lines
3.1 KiB
Go
104 lines
3.1 KiB
Go
package httpx
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/http"
|
|
)
|
|
|
|
// Late-bound providers: these are wired by later subsystems (atomizer
|
|
// client, work-performer client, sync manager, job runner) after the server
|
|
// is constructed. All accessors are nil-safe so the admin status panel
|
|
// degrades gracefully while subsystems are absent (e.g. in tests).
|
|
|
|
// BreakerInfo exposes a circuit breaker's state for the status panel.
|
|
type BreakerInfo interface {
|
|
BreakerState() string // closed | open | half-open
|
|
}
|
|
|
|
// StatusProvider returns arbitrary JSON-marshalable status payloads.
|
|
type StatusProvider interface {
|
|
Statuses() any
|
|
}
|
|
|
|
func (s *Server) SetAtomizerInfo(b BreakerInfo) { s.atomizer = b }
|
|
func (s *Server) SetPerformerInfo(b BreakerInfo) { s.performer = b }
|
|
|
|
// SetSyncTrigger wires the sync manager's "sync now" entry point.
|
|
func (s *Server) SetSyncTrigger(f func(customerID string)) { s.syncTrigger = f }
|
|
|
|
// Publish forwards a live event to connected clients. Until the WebSocket
|
|
// hub lands this is a metrics-only no-op, so subsystems can emit events
|
|
// unconditionally.
|
|
func (s *Server) Publish(channel, event string, payload any) {
|
|
s.metrics.Inc("events_published_total", 1)
|
|
if s.publishFn != nil {
|
|
s.publishFn(channel, event, payload)
|
|
}
|
|
}
|
|
|
|
// SetPublishFn wires the actual hub broadcast.
|
|
func (s *Server) SetPublishFn(f func(channel, event string, payload any)) { s.publishFn = f }
|
|
|
|
// SetEnqueue wires the background job queue.
|
|
func (s *Server) SetEnqueue(f func(ctx context.Context, kind string, payload any) (string, error)) {
|
|
s.enqueueFn = f
|
|
}
|
|
|
|
func (s *Server) enqueue(ctx context.Context, kind string, payload any) (string, error) {
|
|
if s.enqueueFn == nil {
|
|
return "", errors.New("job queue is not running")
|
|
}
|
|
return s.enqueueFn(ctx, kind, payload)
|
|
}
|
|
|
|
// SetWSHandler wires the WebSocket hub's connection handler.
|
|
func (s *Server) SetWSHandler(f func(w http.ResponseWriter, r *http.Request, userID string)) {
|
|
s.wsHandler = f
|
|
}
|
|
|
|
// handleWS authenticates the session and hands the connection to the hub.
|
|
func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
|
|
u := s.pageUser(r)
|
|
if u == nil {
|
|
writeError(w, http.StatusUnauthorized, "unauthenticated", "session required")
|
|
return
|
|
}
|
|
if s.wsHandler == nil {
|
|
writeError(w, http.StatusServiceUnavailable, "ws_unavailable", "live updates are not running")
|
|
return
|
|
}
|
|
s.wsHandler(w, r, u.ID)
|
|
}
|
|
func (s *Server) SetSyncStatusProvider(p StatusProvider) {
|
|
s.syncProvider = p
|
|
}
|
|
func (s *Server) SetJobsStatusProvider(p StatusProvider) {
|
|
s.jobsProvider = p
|
|
}
|
|
|
|
func (s *Server) syncStatuses() any {
|
|
if s.syncProvider == nil {
|
|
return []any{}
|
|
}
|
|
return s.syncProvider.Statuses()
|
|
}
|
|
|
|
func (s *Server) jobStatuses() any {
|
|
if s.jobsProvider == nil {
|
|
return map[string]any{}
|
|
}
|
|
return s.jobsProvider.Statuses()
|
|
}
|
|
|
|
// effectiveAtomizerBaseURL honors the admin settings override (§3) over the
|
|
// env default.
|
|
func (s *Server) effectiveAtomizerBaseURL(ctx context.Context) string {
|
|
if s.store != nil {
|
|
if st, err := s.store.GetSettings(ctx); err == nil && st.AtomizerBaseURLOverride != "" {
|
|
return st.AtomizerBaseURLOverride
|
|
}
|
|
}
|
|
return s.cfg.AtomizerBaseURL
|
|
}
|