Refactor delivery targets to a Target interface (closes #77) (#81)
All checks were successful
check / check (push) Successful in 2m42s
All checks were successful
check / check (push) Successful in 2m42s
Refactors the delivery engine so each target TYPE is an implementation of a `Target` interface, dispatched from a registry, with each target owning its full delivery including durable retries. Implements the authoritative design from issue #77 (the corrected "hand the DB + Scheduler to the target" design). ## The new interface ```go type Scheduler interface { ScheduleRetry(task Task, delay time.Duration) } type Target interface { Deliver(ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, task *Task, sched Scheduler) } ``` `Deliver` receives everything a target needs to be autonomous and durable: the request context, the per-webhook `*gorm.DB`, the `*database.Delivery`, the attempt `*Task`, and a `Scheduler` (the engine) for durable re-enqueue. The target makes one attempt, writes 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 injected `Scheduler`. `processDelivery` collapses to a registry lookup (`map[database.TargetType]Target`) and a `Deliver` call; an unknown target type still fails the delivery as before. ## Per-target ownership - `httpTarget` and `slackTarget` share a retry core (`httpCore`) that owns retry, exponential backoff, and the per-target circuit breaker. The core is fire-and-forget when `MaxRetries == 0` and adds breaker-gated backed-off retries when `MaxRetries > 0`. The per-attempt request differs (HTTP forwards the body + filtered headers; Slack posts a formatted message) and is supplied as a closure, so each keeps its exact recording semantics (e.g. HTTP records no error string for a non-2xx, Slack records `HTTP <code>`). - `databaseTarget` and `logTarget` are fire-and-forget: they record a single successful attempt. Moved wholesale into the http/slack targets: `deliverHTTP*`, `handleHTTPRetry`, `circuitBreakerBlock`, `calcBackoff` / `calcRemainingBackoff` / `backoffElapsed`, the circuit-breaker `sync.Map` + `getCircuitBreaker`, `clientForConfig`, `doHTTPRequest`, `applyRequestHeaders`, and the config parsers. The engine keeps `recordResult`, `updateDeliveryStatus`, and `ScheduleRetry`. ## Slack MaxRetries gating Slack is now on the same shared core as HTTP, with retry + breaker gated on `MaxRetries`. A `MaxRetries` of 0 stays single-attempt fire-and-forget, so **every existing Slack target is unchanged**; a Slack target configured with retries gets backoff + circuit breaker. ## Log-target full content `logTarget` now logs the ENTIRE inbound webhook — full request body and full request headers, plus method, content type, and the webhook id and entrypoint id — rather than a summary line. This supersedes the smaller log-summary work (#70). ## `Task.EntrypointID` To carry the entrypoint id to the log target, `Task` gains an `EntrypointID` field, populated in the webhook handler's `buildDeliveryTasks`, the engine's recovery-task builder, and `buildEventFromTask`. ## Durability / recovery The crash-durable async retry model is preserved unchanged: one attempt per worker turn; on failure the status is set `retrying`, backoff is computed, and the task is re-enqueued via `ScheduleRetry` (a `time.AfterFunc` onto the retry channel). On restart, `recoverRetryingDeliveries` and the 60s sweep hand each orphaned `retrying` delivery back to its target to recompute the remaining backoff and reschedule (targets that own retries implement an internal `rescheduler`; fire-and-forget targets, which never produce `retrying` deliveries, are skipped). ## How behaviour is preserved No external behaviour changes except the two called out above (log target full content; Slack gaining `MaxRetries`-gated retries). All existing delivery tests pass with only their `export_test.go` wrappers re-pointed at the new structure — `ExportDeliverHTTP/Slack/Database/Log` now call the targets, `ExportGetCircuitBreaker` / `ExportClient` / `ExportClientForConfig` / `ExportDoHTTPRequest` resolve against the HTTP target's shared client and breaker map, and `ExportParseHTTPConfig` / `ExportParseSlackConfig` call the relocated free functions. Added: a `logTarget` test asserting the log line contains the full body, headers, and ids, and a Slack `MaxRetries`-gated retry test. `docker build .` is green (fmt-check, lint, test, static build all pass). Closes #77 Co-authored-by: sneak <sneak@sneak.berlin> Reviewed-on: #81 Co-authored-by: clawbot <clawbot@noreply.example.org> Co-committed-by: clawbot <clawbot@noreply.example.org>
This commit was merged in pull request #81.
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@@ -1,6 +1,7 @@
|
|||||||
package delivery_test
|
package delivery_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"encoding/json"
|
"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
|
// readAll is a small helper to avoid importing io in
|
||||||
// a test handler inline.
|
// a test handler inline.
|
||||||
func readAll(r interface {
|
func readAll(r interface {
|
||||||
|
|||||||
@@ -39,37 +39,50 @@ func ExportTruncate(s string, maxLen int) string {
|
|||||||
return truncate(s, maxLen)
|
return truncate(s, maxLen)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportDeliverHTTP exposes deliverHTTP for testing.
|
// ExportDeliverHTTP delivers via the http target for testing.
|
||||||
func (e *Engine) ExportDeliverHTTP(
|
func (e *Engine) ExportDeliverHTTP(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
webhookDB *gorm.DB,
|
webhookDB *gorm.DB,
|
||||||
d *database.Delivery,
|
d *database.Delivery,
|
||||||
task *Task,
|
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(
|
func (e *Engine) ExportDeliverDatabase(
|
||||||
webhookDB *gorm.DB, d *database.Delivery,
|
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(
|
func (e *Engine) ExportDeliverLog(
|
||||||
webhookDB *gorm.DB, d *database.Delivery,
|
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(
|
func (e *Engine) ExportDeliverSlack(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
webhookDB *gorm.DB,
|
webhookDB *gorm.DB,
|
||||||
d *database.Delivery,
|
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.
|
// ExportProcessNewTask exposes processNewTask.
|
||||||
@@ -96,53 +109,56 @@ func (e *Engine) ExportProcessDelivery(
|
|||||||
e.processDelivery(ctx, webhookDB, d, task)
|
e.processDelivery(ctx, webhookDB, d, task)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportGetCircuitBreaker exposes getCircuitBreaker.
|
// ExportGetCircuitBreaker exposes the http target's
|
||||||
|
// getCircuitBreaker.
|
||||||
func (e *Engine) ExportGetCircuitBreaker(
|
func (e *Engine) ExportGetCircuitBreaker(
|
||||||
targetID string,
|
targetID string,
|
||||||
) *CircuitBreaker {
|
) *CircuitBreaker {
|
||||||
return e.getCircuitBreaker(targetID)
|
return e.httpTarget.getCircuitBreaker(targetID)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportParseHTTPConfig exposes parseHTTPConfig.
|
// ExportParseHTTPConfig exposes parseHTTPConfig.
|
||||||
func (e *Engine) ExportParseHTTPConfig(
|
func (e *Engine) ExportParseHTTPConfig(
|
||||||
configJSON string,
|
configJSON string,
|
||||||
) (*HTTPTargetConfig, error) {
|
) (*HTTPTargetConfig, error) {
|
||||||
return e.parseHTTPConfig(configJSON)
|
return parseHTTPConfig(configJSON)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportParseSlackConfig exposes parseSlackConfig.
|
// ExportParseSlackConfig exposes parseSlackConfig.
|
||||||
func (e *Engine) ExportParseSlackConfig(
|
func (e *Engine) ExportParseSlackConfig(
|
||||||
configJSON string,
|
configJSON string,
|
||||||
) (*SlackTargetConfig, error) {
|
) (*SlackTargetConfig, error) {
|
||||||
return e.parseSlackConfig(configJSON)
|
return parseSlackConfig(configJSON)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportDoHTTPRequest exposes doHTTPRequest.
|
// ExportDoHTTPRequest exposes the http target's
|
||||||
|
// doHTTPRequest.
|
||||||
func (e *Engine) ExportDoHTTPRequest(
|
func (e *Engine) ExportDoHTTPRequest(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
cfg *HTTPTargetConfig,
|
cfg *HTTPTargetConfig,
|
||||||
event *database.Event,
|
event *database.Event,
|
||||||
) (int, string, int64, error) {
|
) (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(
|
func (e *Engine) ExportClientForConfig(
|
||||||
cfg *HTTPTargetConfig,
|
cfg *HTTPTargetConfig,
|
||||||
) *http.Client {
|
) *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 {
|
func (e *Engine) ExportClient() *http.Client {
|
||||||
return e.client
|
return e.httpTarget.client
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportScheduleRetry exposes scheduleRetry.
|
// ExportScheduleRetry exposes ScheduleRetry.
|
||||||
func (e *Engine) ExportScheduleRetry(
|
func (e *Engine) ExportScheduleRetry(
|
||||||
task Task, delay time.Duration,
|
task Task, delay time.Duration,
|
||||||
) {
|
) {
|
||||||
e.scheduleRetry(task, delay)
|
e.ScheduleRetry(task, delay)
|
||||||
}
|
}
|
||||||
|
|
||||||
// ExportRecoverPendingDeliveries exposes
|
// ExportRecoverPendingDeliveries exposes
|
||||||
@@ -199,13 +215,15 @@ func NewTestEngine(
|
|||||||
client *http.Client,
|
client *http.Client,
|
||||||
workers int,
|
workers int,
|
||||||
) *Engine {
|
) *Engine {
|
||||||
return &Engine{
|
e := &Engine{
|
||||||
log: log,
|
log: log,
|
||||||
client: client,
|
|
||||||
deliveryCh: make(chan Task, deliveryChannelSize),
|
deliveryCh: make(chan Task, deliveryChannelSize),
|
||||||
retryCh: make(chan Task, retryChannelSize),
|
retryCh: make(chan Task, retryChannelSize),
|
||||||
workers: workers,
|
workers: workers,
|
||||||
}
|
}
|
||||||
|
e.initTargets(client)
|
||||||
|
|
||||||
|
return e
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewTestEngineSmallRetry creates an Engine with a tiny
|
// NewTestEngineSmallRetry creates an Engine with a tiny
|
||||||
@@ -213,10 +231,13 @@ func NewTestEngine(
|
|||||||
func NewTestEngineSmallRetry(
|
func NewTestEngineSmallRetry(
|
||||||
log *slog.Logger,
|
log *slog.Logger,
|
||||||
) *Engine {
|
) *Engine {
|
||||||
return &Engine{
|
e := &Engine{
|
||||||
log: log,
|
log: log,
|
||||||
retryCh: make(chan Task, 1),
|
retryCh: make(chan Task, 1),
|
||||||
}
|
}
|
||||||
|
e.initTargets(nil)
|
||||||
|
|
||||||
|
return e
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewTestEngineWithDB creates an Engine with a real
|
// NewTestEngineWithDB creates an Engine with a real
|
||||||
@@ -228,15 +249,17 @@ func NewTestEngineWithDB(
|
|||||||
client *http.Client,
|
client *http.Client,
|
||||||
workers int,
|
workers int,
|
||||||
) *Engine {
|
) *Engine {
|
||||||
return &Engine{
|
e := &Engine{
|
||||||
database: db,
|
database: db,
|
||||||
dbManager: dbMgr,
|
dbManager: dbMgr,
|
||||||
log: log,
|
log: log,
|
||||||
client: client,
|
|
||||||
deliveryCh: make(chan Task, deliveryChannelSize),
|
deliveryCh: make(chan Task, deliveryChannelSize),
|
||||||
retryCh: make(chan Task, retryChannelSize),
|
retryCh: make(chan Task, retryChannelSize),
|
||||||
workers: workers,
|
workers: workers,
|
||||||
}
|
}
|
||||||
|
e.initTargets(client)
|
||||||
|
|
||||||
|
return e
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewTestCircuitBreaker creates a CircuitBreaker with
|
// NewTestCircuitBreaker creates a CircuitBreaker with
|
||||||
|
|||||||
101
internal/delivery/target.go
Normal file
101
internal/delivery/target.go
Normal 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},
|
||||||
|
}
|
||||||
|
}
|
||||||
34
internal/delivery/target_database.go
Normal file
34
internal/delivery/target_database.go
Normal 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,
|
||||||
|
)
|
||||||
|
}
|
||||||
499
internal/delivery/target_http.go
Normal file
499
internal/delivery/target_http.go
Normal 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
|
||||||
|
}
|
||||||
47
internal/delivery/target_log.go
Normal file
47
internal/delivery/target_log.go
Normal 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,
|
||||||
|
)
|
||||||
|
}
|
||||||
299
internal/delivery/target_slack.go
Normal file
299
internal/delivery/target_slack.go
Normal 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")
|
||||||
|
}
|
||||||
@@ -329,6 +329,7 @@ func (h *Handlers) buildDeliveryTasks(
|
|||||||
DeliveryID: dlv.ID,
|
DeliveryID: dlv.ID,
|
||||||
EventID: event.ID,
|
EventID: event.ID,
|
||||||
WebhookID: entrypoint.WebhookID,
|
WebhookID: entrypoint.WebhookID,
|
||||||
|
EntrypointID: entrypoint.ID,
|
||||||
TargetID: targets[i].ID,
|
TargetID: targets[i].ID,
|
||||||
TargetName: targets[i].Name,
|
TargetName: targets[i].Name,
|
||||||
TargetType: targets[i].Type,
|
TargetType: targets[i].Type,
|
||||||
|
|||||||
Reference in New Issue
Block a user