Documentation
¶
Index ¶
- Variables
- func RunAdapter(ctx context.Context, adapter FrameworkAdapter, opts RuntimeOptions) error
- type AccountQueue
- type Aquifer
- func (a *Aquifer) Enqueue(req JobRequest) (EnqueueResult, error)
- func (a *Aquifer) GetJob(id string) (*Job, error)
- func (a *Aquifer) HandleL8Challenge(req L8ChallengeReq) (*L8ChallengeResp, error)
- func (a *Aquifer) Health() map[string]any
- func (a *Aquifer) L8Metadata(host string) L8Meta
- func (a *Aquifer) SubscribeJob(id string) (*Job, <-chan SSEEvent, func(), error)
- type Broker
- type Config
- type EnqueueResult
- type FrameworkAdapter
- type HTTPAdapter
- type Job
- type JobRequest
- type L8ChallengeReq
- type L8ChallengeResp
- type L8Meta
- type L8Registry
- type MCPStdioAdapter
- type MetricsAdapter
- type NoopMetricsAdapter
- func (NoopMetricsAdapter) FlowRate(upstream string, rps float64)
- func (NoopMetricsAdapter) JobCompleted(userID, upstream string, durationMs int64)
- func (NoopMetricsAdapter) JobDispatched(userID, upstream string)
- func (NoopMetricsAdapter) JobFailed(userID, upstream string, reason string)
- func (NoopMetricsAdapter) JobQueued(userID, upstream string)
- func (NoopMetricsAdapter) QueueDepth(upstream string, depth int)
- func (NoopMetricsAdapter) WebhookDelivered(url string, attempt int)
- func (NoopMetricsAdapter) WebhookFailed(url string, attempts int)
- type RateConfig
- type Registry
- type Runtime
- type RuntimeOptions
- type SSEEvent
- type Server
- type Status
- type Store
- func (s *Store) CheckOrInsert(job *Job) (string, bool)
- func (s *Store) Counts() StoreCounts
- func (s *Store) GetJob(jobID string) *Job
- func (s *Store) GetQueuedJobs() []*Job
- func (s *Store) MarkInFlight(jobID string)
- func (s *Store) Path() string
- func (s *Store) RecoverInFlight(queueKey string) []*Job
- func (s *Store) SetQueueKey(jobID, queueKey string)
- func (s *Store) UpdateStatus(jobID string, status Status)
- type StoreCounts
- type URLWorker
Constants ¶
This section is empty.
Variables ¶
View Source
var ErrJobNotFound = errors.New("job not found")
Functions ¶
func RunAdapter ¶
func RunAdapter(ctx context.Context, adapter FrameworkAdapter, opts RuntimeOptions) error
Types ¶
type AccountQueue ¶
type AccountQueue struct {
// contains filtered or unexported fields
}
func NewAccountQueue ¶
func NewAccountQueue(key, upstream string, rps float64, maxConc int, store *Store, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, onIdle func(string)) *AccountQueue
func (*AccountQueue) Enqueue ¶
func (q *AccountQueue) Enqueue(job *Job)
func (*AccountQueue) RPS ¶
func (q *AccountQueue) RPS() float64
type Aquifer ¶
type Aquifer struct {
// contains filtered or unexported fields
}
func NewAquifer ¶
func NewAquifer(store *Store, registry *Registry, broker *Broker, l8 *L8Registry) *Aquifer
func (*Aquifer) Enqueue ¶
func (a *Aquifer) Enqueue(req JobRequest) (EnqueueResult, error)
func (*Aquifer) HandleL8Challenge ¶
func (a *Aquifer) HandleL8Challenge(req L8ChallengeReq) (*L8ChallengeResp, error)
func (*Aquifer) L8Metadata ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker is the pub/sub layer for SSE streams. Each job gets its own set of subscriber channels.
type Config ¶
type Config struct {
Defaults RateConfig `yaml:"defaults"`
Upstreams map[string]RateConfig `yaml:"upstreams"`
}
func LoadConfig ¶
func (*Config) ForURL ¶
func (c *Config) ForURL(rawURL string) RateConfig
type EnqueueResult ¶
type FrameworkAdapter ¶
type HTTPAdapter ¶
type HTTPAdapter struct {
// contains filtered or unexported fields
}
func NewHTTPAdapter ¶
func NewHTTPAdapter(addr string) *HTTPAdapter
func (*HTTPAdapter) Name ¶
func (a *HTTPAdapter) Name() string
type Job ¶
type Job struct {
ID string `json:"id"`
UserID string `json:"user_id"`
IdempotentKey string `json:"idempotent_key"`
URL string `json:"url"`
Method string `json:"method"`
Headers map[string]string `json:"headers,omitempty"`
Body string `json:"body,omitempty"`
WebhookURL string `json:"webhook_url"`
Status Status `json:"status"`
CreatedAt int64 `json:"created_at"`
}
func NewJob ¶
func NewJob(r *JobRequest) *Job
type JobRequest ¶
type JobRequest struct {
UserID string `json:"user_id"`
IdempotentKey string `json:"idempotent_key"`
URL string `json:"url"`
Method string `json:"method"`
Headers map[string]string `json:"headers,omitempty"`
Body string `json:"body,omitempty"`
WebhookURL string `json:"webhook_url"`
}
func (*JobRequest) Validate ¶
func (r *JobRequest) Validate() string
type L8ChallengeReq ¶
type L8ChallengeResp ¶
type L8Meta ¶
type L8Meta struct {
ProtocolVersion string `json:"protocol_version"`
ServiceName string `json:"service_name"`
PublicKey string `json:"public_key"`
ChallengeEndpoint string `json:"challenge_endpoint"`
SupportedAlgos []string `json:"supported_algorithms"`
Capabilities []string `json:"capabilities"`
SpecURL string `json:"spec_url"`
}
type L8Registry ¶
type L8Registry struct {
PubB64 string // exported so server can embed in responses
// contains filtered or unexported fields
}
func NewL8Registry ¶
func NewL8Registry(keyPath, trustDir string) *L8Registry
func (*L8Registry) EnsureTrust ¶
func (r *L8Registry) EnsureTrust(webhookURL string)
EnsureTrust runs the L8 handshake with the webhook domain if not already trusted. Silently does nothing if the receiver doesn't support L8 — delivery still proceeds unsigned.
func (*L8Registry) HandleChallenge ¶
func (r *L8Registry) HandleChallenge(req L8ChallengeReq) (*L8ChallengeResp, error)
HandleChallenge verifies the sender's ownership proof and returns Aquifer's signed response.
func (*L8Registry) IsTrusted ¶
func (r *L8Registry) IsTrusted(webhookURL string) bool
IsTrusted returns true if the L8 handshake has been completed for this webhook domain.
func (*L8Registry) Meta ¶
func (r *L8Registry) Meta(host string) L8Meta
func (*L8Registry) SignHeaders ¶
func (r *L8Registry) SignHeaders(body []byte) map[string]string
SignHeaders returns X-L8-* headers to attach to an outgoing webhook delivery.
type MCPStdioAdapter ¶
type MCPStdioAdapter struct {
// contains filtered or unexported fields
}
func NewMCPStdioAdapter ¶
func NewMCPStdioAdapter(in io.Reader, out io.Writer) *MCPStdioAdapter
func (*MCPStdioAdapter) Name ¶
func (a *MCPStdioAdapter) Name() string
type MetricsAdapter ¶
type MetricsAdapter interface {
JobQueued(userID, upstream string)
JobDispatched(userID, upstream string)
JobCompleted(userID, upstream string, durationMs int64)
JobFailed(userID, upstream string, reason string)
WebhookDelivered(url string, attempt int)
WebhookFailed(url string, attempts int)
QueueDepth(upstream string, depth int)
FlowRate(upstream string, rps float64)
}
type NoopMetricsAdapter ¶
type NoopMetricsAdapter struct{}
func (NoopMetricsAdapter) FlowRate ¶
func (NoopMetricsAdapter) FlowRate(upstream string, rps float64)
func (NoopMetricsAdapter) JobCompleted ¶
func (NoopMetricsAdapter) JobCompleted(userID, upstream string, durationMs int64)
func (NoopMetricsAdapter) JobDispatched ¶
func (NoopMetricsAdapter) JobDispatched(userID, upstream string)
func (NoopMetricsAdapter) JobFailed ¶
func (NoopMetricsAdapter) JobFailed(userID, upstream string, reason string)
func (NoopMetricsAdapter) JobQueued ¶
func (NoopMetricsAdapter) JobQueued(userID, upstream string)
func (NoopMetricsAdapter) QueueDepth ¶
func (NoopMetricsAdapter) QueueDepth(upstream string, depth int)
func (NoopMetricsAdapter) WebhookDelivered ¶
func (NoopMetricsAdapter) WebhookDelivered(url string, attempt int)
func (NoopMetricsAdapter) WebhookFailed ¶
func (NoopMetricsAdapter) WebhookFailed(url string, attempts int)
type RateConfig ¶
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
func NewRegistry ¶
func NewRegistry(store *Store, cfg *Config, broker *Broker, l8 *L8Registry, metrics MetricsAdapter) *Registry
func (*Registry) JobDispatched ¶
func (r *Registry) JobDispatched()
type Runtime ¶
type Runtime struct {
Aquifer *Aquifer
Store *Store
Broker *Broker
Registry *Registry
L8 *L8Registry
Config *Config
}
func NewRuntime ¶
func NewRuntime(opts RuntimeOptions) *Runtime
func (*Runtime) RecoverQueuedJobs ¶
type RuntimeOptions ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func (*Store) Counts ¶
func (s *Store) Counts() StoreCounts
func (*Store) GetQueuedJobs ¶
func (*Store) MarkInFlight ¶
func (*Store) RecoverInFlight ¶
func (*Store) SetQueueKey ¶
func (*Store) UpdateStatus ¶
type StoreCounts ¶
type URLWorker ¶
type URLWorker struct {
// contains filtered or unexported fields
}
func NewURLWorker ¶
func NewURLWorker(domain string, rps float64, maxConc int, store *Store, broker *Broker, l8 *L8Registry, metrics MetricsAdapter, onIdle func(string)) *URLWorker
Source Files
¶
Click to show internal directories.
Click to hide internal directories.