90 lines
2.5 KiB
Go
90 lines
2.5 KiB
Go
package flow
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
)
|
|
|
|
// RegisterBuiltinHooks adds the generic hooks every deployment can use: "log"
|
|
// (structured log line) and "webhook" (HTTP POST of the event). Domain packages
|
|
// register their own hooks (e.g. "amqp_emit") on the same registry.
|
|
func RegisterBuiltinHooks(reg *Registry) {
|
|
reg.Hook("log", logHook)
|
|
reg.Hook("noop", func(context.Context, *State, HookInfo, json.RawMessage) error { return nil })
|
|
reg.Hook("webhook", webhookHook(&http.Client{Timeout: 10 * time.Second}))
|
|
}
|
|
|
|
// logHook emits a structured log line for the step/phase. Params (optional):
|
|
//
|
|
// {"message": "custom text"}
|
|
func logHook(_ context.Context, st *State, info HookInfo, params json.RawMessage) error {
|
|
msg := "flow hook"
|
|
if len(params) > 0 {
|
|
var p struct {
|
|
Message string `json:"message"`
|
|
}
|
|
if err := json.Unmarshal(params, &p); err == nil && p.Message != "" {
|
|
msg = p.Message
|
|
}
|
|
}
|
|
args := []any{"step", info.Step, "phase", info.Phase, "id", st.ID}
|
|
if info.Err != nil {
|
|
args = append(args, "err", info.Err)
|
|
}
|
|
st.Logger.Info(msg, args...)
|
|
return nil
|
|
}
|
|
|
|
// webhookHook POSTs a JSON event to a configured URL. Params:
|
|
//
|
|
// {"url": "https://...", "headers": {"X-Token": "..."}}
|
|
//
|
|
// The body is {step, phase, id, error, vars}. Returned as a closure so the HTTP
|
|
// client (and its timeout/transport) is shared across invocations.
|
|
func webhookHook(client *http.Client) Hook {
|
|
return func(ctx context.Context, st *State, info HookInfo, params json.RawMessage) error {
|
|
var p struct {
|
|
URL string `json:"url"`
|
|
Headers map[string]string `json:"headers,omitempty"`
|
|
}
|
|
if err := json.Unmarshal(params, &p); err != nil {
|
|
return fmt.Errorf("webhook: bad params: %w", err)
|
|
}
|
|
if p.URL == "" {
|
|
return fmt.Errorf("webhook: missing url")
|
|
}
|
|
errStr := ""
|
|
if info.Err != nil {
|
|
errStr = info.Err.Error()
|
|
}
|
|
body, _ := json.Marshal(map[string]any{
|
|
"step": info.Step,
|
|
"phase": info.Phase,
|
|
"id": st.ID,
|
|
"error": errStr,
|
|
"vars": st.Vars,
|
|
})
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.URL, bytes.NewReader(body))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
for k, v := range p.Headers {
|
|
req.Header.Set(k, v)
|
|
}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode >= 300 {
|
|
return fmt.Errorf("webhook: %s returned %d", p.URL, resp.StatusCode)
|
|
}
|
|
return nil
|
|
}
|
|
}
|