Refactor delivery targets to a Target interface (closes #77)
All checks were successful
check / check (push) Successful in 5s

Each target TYPE is now an implementation of a Target interface,
dispatched from a registry in processDelivery instead of a type
switch on TargetType. Every target owns its full delivery,
including durable retries.

- Target.Deliver receives the context, the per-webhook DB, the
  Delivery, the attempt Task, and a Scheduler for durable
  re-enqueue (the existing timer + retry queue). The target makes
  one attempt, records the DeliveryResult, updates DeliveryStatus,
  and — for retry targets — decides whether to retry, computes its
  own backoff, gates with its own circuit breaker, and reschedules
  via the Scheduler.
- httpTarget and slackTarget share a retry core (retry, backoff,
  circuit breaker). database and log targets are fire-and-forget.
- Slack retry/breaker is gated on MaxRetries: 0 stays
  fire-and-forget (existing Slack targets unchanged), >0 gets
  retry + backoff + breaker on the shared core.
- The engine keeps only the worker pool, queue/channels, restart
  recovery/sweep, the recordResult/updateDeliveryStatus helpers,
  and ScheduleRetry. Recovery/sweep hand each orphaned retrying
  delivery back to its target to recompute the backoff.
- The log target logs the entire inbound webhook: full body and
  headers, method, content type, and the webhook and entrypoint
  ids (supersedes the smaller log-summary work).
- Task gains EntrypointID, populated in the webhook handler, the
  recovery-task builder, and buildEventFromTask.

Behaviour is preserved: existing delivery tests pass with their
export_test wrappers re-pointed at the new targets; new pure
Deliver tests cover the log full-content output and the gated
Slack retry path.
This commit is contained in:
2026-08-07 21:43:21 +07:00
parent 752d6beead
commit 7b1f997194
9 changed files with 1313 additions and 856 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,7 @@
package delivery_test
import (
"bytes"
"context"
"database/sql"
"encoding/json"
@@ -1652,6 +1653,179 @@ func TestProcessDelivery_RoutesToSlack(t *testing.T) {
)
}
// newLogCaptureEngine builds a test engine whose logger
// writes to the returned buffer, for inspecting log output.
func newLogCaptureEngine(
t *testing.T,
) (*delivery.Engine, *bytes.Buffer) {
t.Helper()
var buf bytes.Buffer
log := slog.New(slog.NewTextHandler(
&buf,
&slog.HandlerOptions{Level: slog.LevelDebug},
))
e := delivery.NewTestEngine(
log, &http.Client{Timeout: 5 * time.Second}, 1,
)
return e, &buf
}
// assertLogLineComplete asserts the captured log output
// carries the full inbound webhook content and ids.
func assertLogLineComplete(
t *testing.T, out string, event database.Event,
) {
t.Helper()
assert.Contains(t, out, "log-body-marker",
"log line must contain the full request body",
)
assert.Contains(t, out, "Content-Type",
"log line must contain the full request headers",
)
assert.Contains(t, out, event.EntrypointID,
"log line must contain the entrypoint id",
)
assert.Contains(t, out, event.WebhookID,
"log line must contain the webhook id",
)
assert.Contains(t, out, "application/json",
"log line must contain the content type",
)
}
func TestDeliverLog_LogsFullContent(t *testing.T) {
t.Parallel()
db := testWebhookDB(t)
e, buf := newLogCaptureEngine(t)
event := seedEvent(
t, db, `{"log-body-marker":"abc123"}`,
)
dlv := seedDelivery(
t, db, event.ID, uuid.New().String(),
database.DeliveryStatusPending,
)
d := &database.Delivery{
EventID: event.ID,
TargetID: dlv.TargetID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-log-full",
Type: database.TargetTypeLog,
},
}
d.ID = dlv.ID
e.ExportDeliverLog(db, d)
assertLogLineComplete(t, buf.String(), event)
assertDeliveryStatus(t, db, dlv.ID,
database.DeliveryStatusDelivered,
)
}
// buildSlackRetryDelivery builds a Slack delivery whose
// target is configured with retries enabled.
func buildSlackRetryDelivery(
dlv database.Delivery,
event database.Event,
targetID, cfg string,
) *database.Delivery {
d := &database.Delivery{
EventID: event.ID,
TargetID: targetID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-slack-retry",
Type: database.TargetTypeSlack,
Config: cfg,
MaxRetries: 5,
},
}
d.ID = dlv.ID
return d
}
func TestDeliverSlack_WithRetries_SchedulesRetry(
t *testing.T,
) {
t.Parallel()
db := testWebhookDB(t)
ts := newStatusServer(t, http.StatusServiceUnavailable)
e := testEngine(t, 1)
targetID := uuid.New().String()
slackCfg, err := json.Marshal(
delivery.SlackTargetConfig{WebhookURL: ts.URL},
)
require.NoError(t, err)
event := seedEvent(t, db, `{"slack":"retry"}`)
dlv := seedDelivery(
t, db, event.ID, targetID,
database.DeliveryStatusPending,
)
d := buildSlackRetryDelivery(
dlv, event, targetID, string(slackCfg),
)
task := &delivery.Task{
DeliveryID: dlv.ID,
TargetID: targetID,
TargetType: database.TargetTypeSlack,
MaxRetries: 5,
AttemptNum: 1,
}
e.ExportProcessDelivery(context.TODO(), db, d, task)
assertDeliveryStatus(t, db, dlv.ID,
database.DeliveryStatusRetrying,
)
assertDeliveryResult(
t, db, dlv.ID, false,
http.StatusServiceUnavailable,
)
}
// newStatusServer starts a test server that always responds
// with the given status code.
func newStatusServer(
t *testing.T, code int,
) *httptest.Server {
t.Helper()
ts := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(code)
},
))
t.Cleanup(ts.Close)
return ts
}
// readAll is a small helper to avoid importing io in
// a test handler inline.
func readAll(r interface {

View File

@@ -39,37 +39,50 @@ func ExportTruncate(s string, maxLen int) string {
return truncate(s, maxLen)
}
// ExportDeliverHTTP exposes deliverHTTP for testing.
// ExportDeliverHTTP delivers via the http target for testing.
func (e *Engine) ExportDeliverHTTP(
ctx context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
) {
e.deliverHTTP(ctx, webhookDB, d, task)
e.httpTarget.Deliver(ctx, webhookDB, d, task, e)
}
// ExportDeliverDatabase exposes deliverDatabase.
// ExportDeliverDatabase delivers via the database target.
func (e *Engine) ExportDeliverDatabase(
webhookDB *gorm.DB, d *database.Delivery,
) {
e.deliverDatabase(webhookDB, d)
e.targets[database.TargetTypeDatabase].Deliver(
context.Background(), webhookDB, d, &Task{}, e,
)
}
// ExportDeliverLog exposes deliverLog for testing.
// ExportDeliverLog delivers via the log target for testing.
func (e *Engine) ExportDeliverLog(
webhookDB *gorm.DB, d *database.Delivery,
) {
e.deliverLog(webhookDB, d)
e.targets[database.TargetTypeLog].Deliver(
context.Background(), webhookDB, d, &Task{}, e,
)
}
// ExportDeliverSlack exposes deliverSlack for testing.
// ExportDeliverSlack delivers via the slack target for
// testing.
func (e *Engine) ExportDeliverSlack(
ctx context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
) {
e.deliverSlack(ctx, webhookDB, d)
task := &Task{
DeliveryID: d.ID,
TargetID: d.TargetID,
AttemptNum: 1,
}
e.targets[database.TargetTypeSlack].Deliver(
ctx, webhookDB, d, task, e,
)
}
// ExportProcessNewTask exposes processNewTask.
@@ -96,53 +109,56 @@ func (e *Engine) ExportProcessDelivery(
e.processDelivery(ctx, webhookDB, d, task)
}
// ExportGetCircuitBreaker exposes getCircuitBreaker.
// ExportGetCircuitBreaker exposes the http target's
// getCircuitBreaker.
func (e *Engine) ExportGetCircuitBreaker(
targetID string,
) *CircuitBreaker {
return e.getCircuitBreaker(targetID)
return e.httpTarget.getCircuitBreaker(targetID)
}
// ExportParseHTTPConfig exposes parseHTTPConfig.
func (e *Engine) ExportParseHTTPConfig(
configJSON string,
) (*HTTPTargetConfig, error) {
return e.parseHTTPConfig(configJSON)
return parseHTTPConfig(configJSON)
}
// ExportParseSlackConfig exposes parseSlackConfig.
func (e *Engine) ExportParseSlackConfig(
configJSON string,
) (*SlackTargetConfig, error) {
return e.parseSlackConfig(configJSON)
return parseSlackConfig(configJSON)
}
// ExportDoHTTPRequest exposes doHTTPRequest.
// ExportDoHTTPRequest exposes the http target's
// doHTTPRequest.
func (e *Engine) ExportDoHTTPRequest(
ctx context.Context,
cfg *HTTPTargetConfig,
event *database.Event,
) (int, string, int64, error) {
return e.doHTTPRequest(ctx, cfg, event)
return e.httpTarget.doHTTPRequest(ctx, cfg, event)
}
// ExportClientForConfig exposes clientForConfig.
// ExportClientForConfig exposes the http target's
// clientForConfig.
func (e *Engine) ExportClientForConfig(
cfg *HTTPTargetConfig,
) *http.Client {
return e.clientForConfig(cfg)
return e.httpTarget.clientForConfig(cfg)
}
// ExportClient returns the engine's shared HTTP client.
// ExportClient returns the http target's shared HTTP client.
func (e *Engine) ExportClient() *http.Client {
return e.client
return e.httpTarget.client
}
// ExportScheduleRetry exposes scheduleRetry.
// ExportScheduleRetry exposes ScheduleRetry.
func (e *Engine) ExportScheduleRetry(
task Task, delay time.Duration,
) {
e.scheduleRetry(task, delay)
e.ScheduleRetry(task, delay)
}
// ExportRecoverPendingDeliveries exposes
@@ -199,13 +215,15 @@ func NewTestEngine(
client *http.Client,
workers int,
) *Engine {
return &Engine{
e := &Engine{
log: log,
client: client,
deliveryCh: make(chan Task, deliveryChannelSize),
retryCh: make(chan Task, retryChannelSize),
workers: workers,
}
e.initTargets(client)
return e
}
// NewTestEngineSmallRetry creates an Engine with a tiny
@@ -213,10 +231,13 @@ func NewTestEngine(
func NewTestEngineSmallRetry(
log *slog.Logger,
) *Engine {
return &Engine{
e := &Engine{
log: log,
retryCh: make(chan Task, 1),
}
e.initTargets(nil)
return e
}
// NewTestEngineWithDB creates an Engine with a real
@@ -228,15 +249,17 @@ func NewTestEngineWithDB(
client *http.Client,
workers int,
) *Engine {
return &Engine{
e := &Engine{
database: db,
dbManager: dbMgr,
log: log,
client: client,
deliveryCh: make(chan Task, deliveryChannelSize),
retryCh: make(chan Task, retryChannelSize),
workers: workers,
}
e.initTargets(client)
return e
}
// NewTestCircuitBreaker creates a CircuitBreaker with

101
internal/delivery/target.go Normal file
View File

@@ -0,0 +1,101 @@
package delivery
import (
"context"
"net/http"
"time"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// Scheduler re-enqueues a task for a future delivery attempt.
// The engine provides one to each target so a target can own
// its retries durably: it records the attempt, marks the
// delivery retrying, and asks the Scheduler to deliver the
// next attempt after delay — exactly what the engine does for
// its own restart recovery.
type Scheduler interface {
ScheduleRetry(task Task, delay time.Duration)
}
// Target delivers an event to one target type. Each type is
// an implementation. A Target owns its whole delivery: it
// makes the attempt, records the DeliveryResult and updates
// the DeliveryStatus, and — for targets that retry — decides
// whether to retry, computes its own backoff, gates with its
// own circuit breaker, and reschedules via the injected
// Scheduler. Fire-and-forget targets simply record a single
// attempt.
type Target interface {
Deliver(
ctx context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
)
}
// rescheduler is implemented by targets that own durable
// retries. The engine's restart recovery and periodic sweep
// use it to let the target recompute the schedule for an
// orphaned retrying delivery, keeping the retry schedule
// target-owned. Fire-and-forget targets do not implement it
// and their (never-occurring) retrying deliveries are
// skipped.
type rescheduler interface {
// remainingBackoff returns how long to wait before the
// next attempt of a recovered retrying delivery.
remainingBackoff(
webhookDB *gorm.DB,
deliveryID string,
attemptNum int,
) time.Duration
// backoffElapsed reports whether the backoff window for
// the last attempt has already passed, so the periodic
// sweep can re-enqueue the delivery now.
backoffElapsed(
webhookDB *gorm.DB,
deliveryID string,
attemptNum int,
) bool
}
// attemptResult is the outcome of a single delivery attempt,
// as reported by a target's per-attempt function to the
// shared retry core.
type attemptResult struct {
statusCode int
respBody string
duration int64
success bool
errMsg string
}
// initTargets builds the target registry, wiring each target
// to the engine's persistence helpers and giving the HTTP and
// Slack targets the shared SSRF-safe client. It is called by
// both New and the test constructors so the registry is
// always populated.
func (e *Engine) initTargets(client *http.Client) {
httpT := &httpTarget{
httpCore: &httpCore{eng: e},
client: client,
}
slackT := &slackTarget{
httpCore: &httpCore{eng: e},
client: client,
}
e.httpTarget = httpT
e.targets = map[database.TargetType]Target{
database.TargetTypeHTTP: httpT,
database.TargetTypeSlack: slackT,
database.TargetTypeDatabase: &databaseTarget{eng: e},
database.TargetTypeLog: &logTarget{eng: e},
}
}

View File

@@ -0,0 +1,34 @@
package delivery
import (
"context"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// databaseTarget is a fire-and-forget target: the event is
// already persisted in the per-webhook database by the time
// delivery runs, so the target records a single successful
// attempt. (Durable archiving to a separate store is tracked
// as its own work.)
type databaseTarget struct {
eng *Engine
}
// Deliver implements Target.
func (t *databaseTarget) Deliver(
_ context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
_ *Task,
_ Scheduler,
) {
t.eng.recordResult(
webhookDB, d, 1, true, 0, "", "", 0,
)
t.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusDelivered,
)
}

View File

@@ -0,0 +1,499 @@
package delivery
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"sync"
"time"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// Sentinel errors returned by the config parsers.
var (
errEmptyTargetConfig = errors.New(
"empty target config",
)
errMissingTargetURL = errors.New(
"target URL is required",
)
)
// HTTPTargetConfig holds configuration for http target
// types.
type HTTPTargetConfig struct {
URL string `json:"url"`
Headers map[string]string `json:"headers,omitempty"`
Timeout int `json:"timeout,omitempty"`
}
// httpCore holds the retry, backoff, and circuit-breaker
// machinery shared by the HTTP and Slack targets. Each of
// those targets owns its own httpCore instance (and thus its
// own circuit breakers); the per-attempt request differs
// between them and is supplied as a closure.
type httpCore struct {
eng *Engine
// circuitBreakers stores a *CircuitBreaker per target ID.
circuitBreakers sync.Map
}
// deliver runs one delivery attempt through the retry core.
// A maxRetries of 0 is fire-and-forget: a single attempt is
// recorded and no circuit breaker is consulted. A positive
// maxRetries gates the attempt on the circuit breaker and
// schedules a backed-off retry on failure.
func (c *httpCore) deliver(
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
maxRetries int,
attempt func() attemptResult,
) {
if maxRetries == 0 {
c.fireAndForget(webhookDB, d, attempt())
return
}
c.withRetry(
webhookDB, d, task, sched, maxRetries, attempt,
)
}
func (c *httpCore) fireAndForget(
webhookDB *gorm.DB,
d *database.Delivery,
res attemptResult,
) {
c.eng.recordResult(
webhookDB, d, 1, res.success,
res.statusCode, res.respBody, res.errMsg,
res.duration,
)
if res.success {
c.eng.updateDeliveryStatus(
webhookDB, d,
database.DeliveryStatusDelivered,
)
return
}
c.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusFailed,
)
}
func (c *httpCore) withRetry(
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
maxRetries int,
attempt func() attemptResult,
) {
cb := c.getCircuitBreaker(task.TargetID)
if c.circuitBreakerBlock(webhookDB, d, task, sched, cb) {
return
}
attemptNum := task.AttemptNum
res := attempt()
c.eng.recordResult(
webhookDB, d, attemptNum, res.success,
res.statusCode, res.respBody, res.errMsg,
res.duration,
)
if res.success {
cb.RecordSuccess()
c.eng.updateDeliveryStatus(
webhookDB, d,
database.DeliveryStatusDelivered,
)
return
}
cb.RecordFailure()
c.handleRetry(
webhookDB, d, task, sched, maxRetries, attemptNum,
)
}
func (c *httpCore) circuitBreakerBlock(
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
cb *CircuitBreaker,
) bool {
if cb.Allow() {
return false
}
remaining := cb.CooldownRemaining()
c.eng.log.Info(
"circuit breaker open, skipping delivery",
"target_id", task.TargetID,
"target_name", task.TargetName,
"delivery_id", d.ID,
"cooldown_remaining", remaining,
)
c.eng.updateDeliveryStatus(
webhookDB, d,
database.DeliveryStatusRetrying,
)
retryTask := *task
sched.ScheduleRetry(retryTask, remaining)
return true
}
func (c *httpCore) handleRetry(
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
maxRetries int,
attemptNum int,
) {
if attemptNum >= maxRetries {
c.eng.updateDeliveryStatus(
webhookDB, d,
database.DeliveryStatusFailed,
)
return
}
c.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusRetrying,
)
backoff := calcBackoff(attemptNum)
retryTask := *task
retryTask.AttemptNum = attemptNum + 1
sched.ScheduleRetry(retryTask, backoff)
}
func (c *httpCore) getCircuitBreaker(
targetID string,
) *CircuitBreaker {
if val, ok := c.circuitBreakers.Load(targetID); ok {
cb, _ := val.(*CircuitBreaker)
return cb
}
fresh := NewCircuitBreaker()
actual, _ := c.circuitBreakers.LoadOrStore(
targetID, fresh,
)
cb, _ := actual.(*CircuitBreaker)
return cb
}
// remainingBackoff returns how long remains of the backoff
// window for the last attempt of a recovered retrying
// delivery. It implements rescheduler.
func (c *httpCore) remainingBackoff(
webhookDB *gorm.DB,
deliveryID string,
attemptNum int,
) time.Duration {
var lastResult database.DeliveryResult
err := webhookDB.
Where("delivery_id = ?", deliveryID).
Order("created_at DESC").
First(&lastResult).Error
if err != nil {
return 0
}
backoff := calcBackoff(attemptNum)
elapsed := time.Since(lastResult.CreatedAt)
remaining := backoff - elapsed
return max(remaining, 0)
}
// backoffElapsed reports whether the backoff window for the
// last attempt of a retrying delivery has passed. It
// implements rescheduler.
func (c *httpCore) backoffElapsed(
webhookDB *gorm.DB,
deliveryID string,
attemptNum int,
) bool {
var lastResult database.DeliveryResult
err := webhookDB.
Where("delivery_id = ?", deliveryID).
Order("created_at DESC").
First(&lastResult).Error
if err != nil {
return true
}
backoff := calcBackoff(attemptNum)
return time.Since(lastResult.CreatedAt) >= backoff
}
func calcBackoff(attemptNum int) time.Duration {
shift := max(attemptNum-1, 0)
shift = min(shift, maxBackoffShift)
return time.Duration(1<<uint(shift)) * time.Second
}
// httpTarget delivers events to http targets. It forwards the
// event body and (filtered) request headers to the configured
// URL and owns retry, backoff, and circuit breaking through
// the shared httpCore.
type httpTarget struct {
*httpCore
client *http.Client
}
// Deliver implements Target.
func (t *httpTarget) Deliver(
ctx context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
) {
cfg, err := parseHTTPConfig(d.Target.Config)
if err != nil {
t.eng.log.Error(
"invalid HTTP target config",
"target_id", d.TargetID,
"error", err,
)
t.eng.recordResult(
webhookDB, d, task.AttemptNum,
false, 0, "", err.Error(), 0,
)
t.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusFailed,
)
return
}
attempt := func() attemptResult {
return t.attempt(ctx, cfg, &d.Event)
}
t.deliver(
webhookDB, d, task, sched,
d.Target.MaxRetries, attempt,
)
}
// attempt performs a single HTTP delivery attempt and derives
// the success flag and error message the same way the engine
// did: a non-2xx response is a failure but carries no error
// string; only a transport-level error does.
func (t *httpTarget) attempt(
ctx context.Context,
cfg *HTTPTargetConfig,
event *database.Event,
) attemptResult {
statusCode, respBody, duration, reqErr :=
t.doHTTPRequest(ctx, cfg, event)
success := reqErr == nil &&
statusCode >= httpSuccessMin &&
statusCode < httpSuccessMax
errMsg := ""
if reqErr != nil {
errMsg = reqErr.Error()
}
return attemptResult{
statusCode: statusCode,
respBody: respBody,
duration: duration,
success: success,
errMsg: errMsg,
}
}
func (t *httpTarget) doHTTPRequest(
ctx context.Context,
cfg *HTTPTargetConfig,
event *database.Event,
) (int, string, int64, error) {
start := time.Now()
req, reqErr := http.NewRequestWithContext(
ctx,
http.MethodPost,
cfg.URL,
bytes.NewReader([]byte(event.Body)),
)
if reqErr != nil {
return 0, "", 0, fmt.Errorf(
"creating request: %w", reqErr,
)
}
applyRequestHeaders(req, event, cfg)
client := t.clientForConfig(cfg)
resp, doErr := executeHTTPRequest(client, req)
dur := time.Since(start).Milliseconds()
if doErr != nil {
return 0, "", dur, fmt.Errorf(
"sending request: %w", doErr,
)
}
defer func() { _ = resp.Body.Close() }()
body, readErr := io.ReadAll(
io.LimitReader(resp.Body, maxBodyLog),
)
if readErr != nil {
return resp.StatusCode, "", dur,
fmt.Errorf(
"reading response body: %w", readErr,
)
}
return resp.StatusCode, string(body), dur, nil
}
func (t *httpTarget) clientForConfig(
cfg *HTTPTargetConfig,
) *http.Client {
if cfg.Timeout > 0 {
// Reuse the shared client's SSRF-safe transport so
// a per-target timeout does not drop the
// request-time private-IP guard. Only the timeout
// is overridden.
return &http.Client{
Timeout: time.Duration(
cfg.Timeout,
) * time.Second,
Transport: t.client.Transport,
}
}
return t.client
}
func parseHTTPConfig(
configJSON string,
) (*HTTPTargetConfig, error) {
if configJSON == "" {
return nil, errEmptyTargetConfig
}
var cfg HTTPTargetConfig
err := json.Unmarshal(
[]byte(configJSON), &cfg,
)
if err != nil {
return nil, fmt.Errorf(
"parsing config JSON: %w", err,
)
}
if cfg.URL == "" {
return nil, errMissingTargetURL
}
return &cfg, nil
}
// isForwardableHeader returns true if the header should
// be forwarded to targets.
func isForwardableHeader(name string) bool {
switch http.CanonicalHeaderKey(name) {
case "Host", "Connection", "Keep-Alive",
"Transfer-Encoding", "Te", "Trailer",
"Upgrade", "Proxy-Authorization",
"Proxy-Connection", "Content-Length":
return false
default:
return true
}
}
func applyRequestHeaders(
req *http.Request,
event *database.Event,
cfg *HTTPTargetConfig,
) {
if event.ContentType != "" {
req.Header.Set(
"Content-Type", event.ContentType,
)
}
var originalHeaders map[string][]string
if event.Headers != "" {
jsonErr := json.Unmarshal(
[]byte(event.Headers),
&originalHeaders,
)
if jsonErr == nil {
for k, vals := range originalHeaders {
if isForwardableHeader(k) {
for _, v := range vals {
req.Header.Add(k, v)
}
}
}
}
}
for k, v := range cfg.Headers {
req.Header.Set(k, v)
}
req.Header.Set("User-Agent", "webhooker/1.0")
}
// executeHTTPRequest sends an HTTP request using the provided
// client. URLs are validated by the config parsers and the
// SSRF-safe transport before reaching here.
func executeHTTPRequest(
client *http.Client, req *http.Request,
) (*http.Response, error) {
return client.Do(req) //#nosec G704 -- URL validated by parseHTTPConfig/parseSlackConfig and SSRF-safe transport
}

View File

@@ -0,0 +1,47 @@
package delivery
import (
"context"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// logTarget is a fire-and-forget target that logs the entire
// inbound webhook — the full request body and headers, plus
// the method, content type, and the webhook and entrypoint
// ids — then records a single successful attempt.
type logTarget struct {
eng *Engine
}
// Deliver implements Target.
func (t *logTarget) Deliver(
_ context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
_ *Task,
_ Scheduler,
) {
t.eng.log.Info(
"webhook event delivered to log target",
"delivery_id", d.ID,
"event_id", d.EventID,
"target_id", d.TargetID,
"target_name", d.Target.Name,
"webhook_id", d.Event.WebhookID,
"entrypoint_id", d.Event.EntrypointID,
"method", d.Event.Method,
"content_type", d.Event.ContentType,
"headers", d.Event.Headers,
"body", d.Event.Body,
)
t.eng.recordResult(
webhookDB, d, 1, true, 0, "", "", 0,
)
t.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusDelivered,
)
}

View File

@@ -0,0 +1,299 @@
package delivery
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strings"
"time"
"gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database"
)
// errMissingWebhookURL is returned when a Slack target config
// omits its webhook URL.
var errMissingWebhookURL = errors.New(
"webhook_url is required",
)
// SlackTargetConfig holds configuration for slack target
// types.
type SlackTargetConfig struct {
WebhookURL string `json:"webhookUrl"`
}
// slackTarget delivers events to Slack incoming webhooks. It
// formats the event into a Slack message and posts it as
// JSON. It shares the retry core with the HTTP target: a
// MaxRetries of 0 stays single-attempt fire-and-forget
// (preserving existing Slack targets), while a positive
// MaxRetries adds backoff and circuit breaking.
type slackTarget struct {
*httpCore
client *http.Client
}
// Deliver implements Target.
func (t *slackTarget) Deliver(
ctx context.Context,
webhookDB *gorm.DB,
d *database.Delivery,
task *Task,
sched Scheduler,
) {
cfg, err := parseSlackConfig(d.Target.Config)
if err != nil {
t.eng.log.Error(
"invalid Slack target config",
"target_id", d.TargetID,
"error", err,
)
t.failConfig(webhookDB, d, err)
return
}
msg := FormatSlackMessage(&d.Event)
payload, err := json.Marshal(
map[string]string{"text": msg},
)
if err != nil {
t.eng.log.Error(
"failed to marshal Slack payload",
"target_id", d.TargetID,
"error", err,
)
t.failConfig(webhookDB, d, err)
return
}
attempt := func() attemptResult {
return t.attempt(ctx, cfg, payload)
}
t.deliver(
webhookDB, d, task, sched,
d.Target.MaxRetries, attempt,
)
}
// failConfig records a first-attempt failure for a delivery
// that could not be prepared (bad config or unmarshalable
// payload) and marks it failed.
func (t *slackTarget) failConfig(
webhookDB *gorm.DB,
d *database.Delivery,
err error,
) {
t.eng.recordResult(
webhookDB, d, 1,
false, 0, "", err.Error(), 0,
)
t.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusFailed,
)
}
// attempt performs a single Slack POST and derives its
// outcome, preserving the engine's original semantics: a
// non-2xx response records an "HTTP <code>" error string and
// a transport error records a "sending request" error.
func (t *slackTarget) attempt(
ctx context.Context,
cfg *SlackTargetConfig,
payload []byte,
) attemptResult {
start := time.Now()
req, err := http.NewRequestWithContext(
ctx,
http.MethodPost,
cfg.WebhookURL,
bytes.NewReader(payload),
)
if err != nil {
return attemptResult{
success: false,
errMsg: err.Error(),
}
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "webhooker/1.0")
resp, doErr := executeHTTPRequest(t.client, req)
durationMs := time.Since(start).Milliseconds()
if doErr != nil {
return attemptResult{
success: false,
duration: durationMs,
errMsg: fmt.Errorf(
"sending request: %w", doErr,
).Error(),
}
}
defer func() { _ = resp.Body.Close() }()
return t.readSlackResponse(resp, durationMs)
}
func (t *slackTarget) readSlackResponse(
resp *http.Response,
durationMs int64,
) attemptResult {
body, readErr := io.ReadAll(
io.LimitReader(resp.Body, maxBodyLog),
)
if readErr != nil {
t.eng.log.Error(
"failed to read Slack response body",
"error", readErr,
)
}
success := resp.StatusCode >= httpSuccessMin &&
resp.StatusCode < httpSuccessMax
errMsg := ""
if !success {
errMsg = fmt.Sprintf("HTTP %d", resp.StatusCode)
}
return attemptResult{
statusCode: resp.StatusCode,
respBody: string(body),
duration: durationMs,
success: success,
errMsg: errMsg,
}
}
func parseSlackConfig(
configJSON string,
) (*SlackTargetConfig, error) {
if configJSON == "" {
return nil, errEmptyTargetConfig
}
var cfg SlackTargetConfig
err := json.Unmarshal(
[]byte(configJSON), &cfg,
)
if err != nil {
return nil, fmt.Errorf(
"parsing config JSON: %w", err,
)
}
if cfg.WebhookURL == "" {
return nil, errMissingWebhookURL
}
return &cfg, nil
}
// FormatSlackMessage builds a Slack-compatible message
// string from a webhook event.
func FormatSlackMessage(
event *database.Event,
) string {
var b strings.Builder
b.WriteString("*Webhook Event Received*\n")
fmt.Fprintf(
&b, "*Method:* `%s`\n", event.Method,
)
fmt.Fprintf(
&b,
"*Content-Type:* `%s`\n",
event.ContentType,
)
fmt.Fprintf(
&b,
"*Timestamp:* `%s`\n",
event.CreatedAt.UTC().Format(time.RFC3339),
)
fmt.Fprintf(
&b,
"*Body Size:* %d bytes\n",
len(event.Body),
)
if event.Body == "" {
b.WriteString("\n_(empty body)_\n")
return b.String()
}
if formatted := formatJSONBody(event.Body); formatted != "" {
b.WriteString(formatted)
return b.String()
}
formatRawBody(&b, event.Body)
return b.String()
}
func formatJSONBody(body string) string {
var parsed json.RawMessage
if json.Unmarshal([]byte(body), &parsed) != nil {
return ""
}
var pretty bytes.Buffer
if json.Indent(&pretty, parsed, "", " ") != nil {
return ""
}
var b strings.Builder
b.WriteString("\n```\n")
prettyStr := pretty.String()
const maxPayloadDisplay = 3500
if len(prettyStr) > maxPayloadDisplay {
b.WriteString(prettyStr[:maxPayloadDisplay])
b.WriteString("\n... (truncated)")
} else {
b.WriteString(prettyStr)
}
b.WriteString("\n```\n")
return b.String()
}
func formatRawBody(b *strings.Builder, body string) {
b.WriteString("\n```\n")
const maxRawDisplay = 3500
if len(body) > maxRawDisplay {
b.WriteString(body[:maxRawDisplay])
b.WriteString("\n... (truncated)")
} else {
b.WriteString(body)
}
b.WriteString("\n```\n")
}

View File

@@ -329,6 +329,7 @@ func (h *Handlers) buildDeliveryTasks(
DeliveryID: dlv.ID,
EventID: event.ID,
WebhookID: entrypoint.WebhookID,
EntrypointID: entrypoint.ID,
TargetID: targets[i].ID,
TargetName: targets[i].Name,
TargetType: targets[i].Type,