2 Commits

Author SHA1 Message Date
36a1bacf11 Update golangci-lint to v2.12.2 with canonical config
All checks were successful
check / check (push) Successful in 2m48s
Bump the golangci-lint Docker image pin in Dockerfile and the
release-archive sha256 pins in script/bootstrap from 2.11.3 to
2.12.2, and replace .golangci.yml with the canonical config. The
canonical config moves lll/funlen/cyclop/dupl settings from the
top-level linters-settings key (ignored by the v2 schema) to
linters.settings, so those thresholds now actually apply.

Fix all findings the newly applied thresholds surfaced:

- lll: wrap or shorten seven over-length lines (struct tag
  comments, test logger construction, a func signature, and a
  nosec comment)
- goconst: use http.MethodPost/http.MethodPut and new shared
  constants for repeated test strings; add tmplKeyError and
  tmplKeyWebhook constants for template data keys in handlers
- dupl: merge buildHTTPTargetConfig and buildSlackTargetConfig
  into a parameterized buildURLTargetConfig; drop the duplicate
  iWebhookDB test helper in favor of testWebhookDB; extract
  shared helpers in middleware and session tests
2026-08-07 20:55:31 +00:00
ee7c626071 Implement the database archiving target (closes #43) (#84)
All checks were successful
check / check (push) Successful in 4s
Implements the `databaseTarget` as a real archiving target, replacing the always-successful stub. Delivering to a `database` target now writes the full event into a per-webhook archive SQLite file for long-term storage.

## Archive-writer semantics

- **Separate file:** each webhook's full events are written as rows into `archive-{webhookID}.db` under the data dir, distinct from the per-webhook event DB (`events-{webhookID}.db`). The file and its schema are created on first write if missing. Each row carries the full event: body, headers, method, content type, webhook id, entrypoint id, event id, and an archived-at timestamp.
- **Close/reopen with debounce:** after each write the archive handle is closed and reopened, unless the last (re)open was less than one second ago. This lets an operator move the archive file away for offline archiving while bounding file churn under load. A per-webhook `archiveWriter` owns this debounce state and serialises writes.
- **Auto-recreate:** the file is opened create-if-missing (`mode=rwc`) and its schema re-migrated on every open, so if the archive was moved or removed since the last open, the next write recreates it. The writer also detects a missing file before writing and reopens first, so a moved-away file is recreated rather than lost.
- **Optional expiry, validated at creation:** an optional `expiry` in the target's config JSON (e.g. `{"expiry":"720h"}`) is validated when the target is created (`ValidateArchiveExpiry`; bad values are rejected with a 400 at the add-target form, the Slack URL precedent). The default (missing, empty, or `"never"`) keeps rows forever with no pruning. When a positive duration is set, rows older than it (measured from each row's archived-at time) are pruned on every (re)open; because the file is reopened after writes, prune-on-open keeps the archive swept without a separate background sweeper. A set-but-invalid expiry in a stored config (unparseable, zero, or negative) is an error at delivery time too — never a silent default.
- **No-retry, fail-loud:** the target performs a single attempt with no retries. On success it records one successful attempt and marks the delivery delivered. If the archive write fails, the attempt is recorded as failed with the error and the delivery is marked failed — archiving errors never report success.

## Scope

- `internal/delivery/target_database.go` — the `databaseTarget` (no-retry) archives via a per-webhook writer registry; an archive error records a failed attempt and marks the delivery failed.
- `internal/delivery/target_database_archive.go` (new) — the `archiveWriter`, the archived-row model, config/expiry parsing (fail-loud on set-but-invalid values), `ValidateArchiveExpiry`, and prune-on-open.
- `internal/handlers/source_management.go` — database targets get a creation-validated `expiry` config (`buildDatabaseTargetConfig`); the expiry form value is read where the request body is bounded and bad values are rejected with a 400 at target creation.
- `templates/source_detail.html` — the add-target form shows an expiry field for database targets.
- `README.md` — the database-target documentation describes the archiving semantics.
- `internal/delivery/export_test.go`, `internal/delivery/target_database_test.go`, `internal/handlers` tests — tests and their exported shims.

No changes to the `Target` interface or other targets.

## Tests

- a row is archived (both at the writer level and end-to-end through `Deliver`)
- a forced archive failure (bad stored expiry config) yields a `Failed` delivery with a non-success `DeliveryResult` carrying the error and no archive file created
- the file is recreated after removal, with only the post-removal row
- the one-second reopen debounce (rapid writes reopen once; a write after the window reopens again)
- expiry pruning removes rows older than the configured expiry
- expiry config parsing (empty / `never` / duration accepted; unparseable, zero, and negative values error)
- expiry validation at target creation (`TestValidateArchiveExpiry`; valid values build the config, bad values get a 400)

## Validation

`docker build .` exits 0 (fmt-check, lint, test, build all pass).

Closes #43

Co-authored-by: sneak <sneak@sneak.berlin>
Reviewed-on: #84
Co-authored-by: clawbot <clawbot@noreply.example.org>
Co-committed-by: clawbot <clawbot@noreply.example.org>
2026-08-07 22:50:08 +02:00
33 changed files with 1258 additions and 577 deletions

View File

@@ -1,5 +1,9 @@
version: "2" version: "2"
# Config schema uses the golangci-lint v2 layout (settings live under
# linters.settings, not top-level linters-settings) so that the
# thresholds below are actually applied by golangci-lint >= v2.
run: run:
timeout: 5m timeout: 5m
modules-download-mode: readonly modules-download-mode: readonly
@@ -14,8 +18,7 @@ linters:
- wsl # Deprecated, replaced by wsl_v5 - wsl # Deprecated, replaced by wsl_v5
- wrapcheck # Too verbose for internal packages - wrapcheck # Too verbose for internal packages
- varnamelen # Short names like db, id are idiomatic Go - varnamelen # Short names like db, id are idiomatic Go
settings:
linters-settings:
lll: lll:
line-length: 88 line-length: 88
funlen: funlen:
@@ -27,6 +30,5 @@ linters-settings:
threshold: 100 threshold: 100
issues: issues:
exclude-use-default: false
max-issues-per-linter: 0 max-issues-per-linter: 0
max-same-issues: 0 max-same-issues: 0

View File

@@ -1,8 +1,8 @@
# Lint stage # Lint stage
# golangci/golangci-lint:v2.11.3 (Debian-based), 2026-03-17 # golangci/golangci-lint:v2.12.2 (Debian-based), 2026-08-07
# Using Debian-based image because mattn/go-sqlite3 (CGO) does not # Using Debian-based image because mattn/go-sqlite3 (CGO) does not
# compile on Alpine musl (off64_t is a glibc type). # compile on Alpine musl (off64_t is a glibc type).
FROM golangci/golangci-lint:v2.11.3@sha256:e838e8ab68aaefe83e2408691510867ade9329c0e0b895a3fb35eb93d1c2a4ba AS lint FROM golangci/golangci-lint:v2.12.2@sha256:5cceeef04e53efe1470638d4b4b4f5ceefd574955ab3941b2d9a68a8c9ad5240 AS lint
RUN apt-get update && apt-get install -y --no-install-recommends make && rm -rf /var/lib/apt/lists/* RUN apt-get update && apt-get install -y --no-install-recommends make && rm -rf /var/lib/apt/lists/*

View File

@@ -92,7 +92,6 @@ TTY detection, and security headers are always applied.
| `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` | | `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` |
| `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` | | `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` |
| `SENTRY_DSN` | Sentry error reporting DSN | `""` | | `SENTRY_DSN` | Sentry error reporting DSN | `""` |
| `RECEIVER_RATE_LIMIT` | Receiver requests/minute per IP per entrypoint | `120` |
On first startup, webhooker automatically generates a cryptographically On first startup, webhooker automatically generates a cryptographically
secure session encryption key and stores it in the database. This key secure session encryption key and stores it in the database. This key
@@ -364,10 +363,12 @@ events should be forwarded.
greater than 0, failed deliveries are retried with exponential backoff greater than 0, failed deliveries are retried with exponential backoff
up to `max_retries` attempts, protected by a per-target circuit up to `max_retries` attempts, protected by a per-target circuit
breaker. breaker.
- **`database`** — Confirm the event is stored in the webhook's - **`database`** — Archive the full event as a row into a separate
per-webhook database (no external delivery). Since events are always per-webhook archive database (`archive-{webhookID}.db`) for long-term
written to the per-webhook DB on ingestion, this target marks delivery retention, with an optional creation-validated expiry (default: keep
as immediately successful. Useful for ensuring durable event archival. forever). No external delivery and no retries; an archive write
failure fails the delivery. See the database target section under
"Per-Webhook Event Databases" for the full semantics.
- **`log`** — Write the event to the application log (stdout). Useful - **`log`** — Write the event to the application log (stdout). Useful
for debugging. for debugging.
@@ -513,11 +514,22 @@ This separation provides:
page cache, and its own lock, so concurrent event ingestion across page cache, and its own lock, so concurrent event ingestion across
webhooks won't contend. webhooks won't contend.
The **database target type** leverages this architecture: since events The **database target type** builds on this architecture to provide
are already stored in the per-webhook database by design, the database long-term archiving, separate from the per-webhook event database (which
target simply marks the delivery as immediately successful. The may prune events under its own retention). Delivering to a database
per-webhook DB IS the dedicated event database — that's the whole point target writes the full event — body, headers, method, content type, and
of the database target type. webhook/entrypoint/event identifiers — as a row into a dedicated archive
database, `archive-{webhookID}.db`, stored under the data directory
beside the event database. After each write the archive handle is closed
and reopened, debounced to at most once per second, so an operator can
move the archive file away for offline archiving without stopping the
service; a moved or removed archive file is recreated automatically on
the next write. An optional `expiry` in the target's config JSON (e.g.
`{"expiry":"720h"}`) is validated when the target is created — the
default (unset or the literal `never`) keeps rows forever — and rows
older than the expiry are pruned each time the archive is (re)opened. An
archive write failure is never silent success: the delivery records a
failed attempt with the error and is marked failed.
The **Slack target type** sends webhook events as formatted messages to The **Slack target type** sends webhook events as formatted messages to
any Slack-compatible incoming webhook URL (works with Slack, Mattermost, any Slack-compatible incoming webhook URL (works with Slack, Mattermost,
@@ -677,24 +689,17 @@ just delayed until the target is healthy again.
### Rate Limiting ### Rate Limiting
Global blanket rate limiting middleware (e.g., a per-IP throttle shared Global rate limiting middleware (e.g., per-IP throttling applied at the
with the web UI) **must not** apply to webhook receiver endpoints. router level) **must not** apply to webhook receiver endpoints. Webhook
Webhook endpoints receive automated traffic from external services at endpoints receive automated traffic from external services at
unpredictable rates, and blanket limits shared with other routes would unpredictable rates, and blanket rate limits would cause legitimate
cause legitimate deliveries to be dropped. deliveries to be dropped.
The receiver instead has its own dedicated abuse limit, scoped to the Instead, each webhook has its own individually configurable rate limit,
`/webhook/{uuid}` route only and keyed per client IP per entrypoint: one applied within the webhook handler itself. By default, no rate limit is
misbehaving sender is throttled without affecting other senders of the applied — webhook endpoints accept traffic as fast as it arrives. Rate
same entrypoint or the same sender's other entrypoints. The limit is limits can be configured per-webhook when needed (e.g., to protect
`RECEIVER_RATE_LIMIT` requests per minute (default 120, generous for against a misbehaving sender).
legitimate webhook senders). Requests over the limit receive HTTP 429
with a `Retry-After` header. A set-but-invalid `RECEIVER_RATE_LIMIT`
value aborts startup rather than silently falling back to the default.
Finer-grained per-webhook rate limits (configured in the web UI and
enforced in the webhook handler) can layer on top of this env-level
abuse limit later; they are tracked as future work.
### API Endpoints ### API Endpoints

36
TODO.md
View File

@@ -10,27 +10,29 @@
# Status # Status
pre-1.0. No git tags exist. main (81413c5) is a working webhook proxy pre-1.0. No git tags exist. main (afe88c6) is a working webhook proxy
with auth, CSRF/SSRF protections, login rate limiting, Slack target, with auth, CSRF/SSRF protections, login rate limiting, Slack target,
policy compliance (#6), pinned lint tooling (#55), a per-webhook event policy compliance (#6), and pinned lint tooling (#55). Note: TODO.md was
retention reaper (#63), and delivery targets behind a Target interface deliberately deleted from this repo in f9a9569 (2026-03-01, #6); its
(#77). Work is tracked as Gitea issues (the authoritative TODO); this content was folded into the README TODO section, which this draft
file is a summary. Note: TODO.md was deliberately deleted from this reconstructs as of 2026-07-06.
repo in f9a9569 (2026-03-01, #6); its content was folded into the
README TODO section, which this draft reconstructs as of 2026-07-06.
# Next Step # Next Step
Manual event redelivery from the web UI (replay is a core promised Implement automatic event retention cleanup based on retention_days: a
capability in the README rationale). periodic maintenance job that deletes Events, Deliveries, and
DeliveryResults older than the parent webhook's retention_days from each
per-webhook event database. The field exists on the Webhook model and
the README promises the behavior, but nothing enforces it, so event
databases currently grow without bound.
# Completed Steps # Completed Steps
- 2026-08-07 Rate-limit the public webhook receiver per client IP per - 2026-08-07 Update golangci-lint to v2.12.2 (Docker image digest in
entrypoint, env-configurable with fail-loud parsing (#64) `Dockerfile`, release-archive sha256 pins in `script/bootstrap`),
- 2026-08-07 Per-webhook event retention reaper (#63); NoCache adopt the canonical `.golangci.yml` (v2 `linters.settings` layout so
middleware for authenticated pages (#61); Target interface refactor `lll`/`funlen`/`cyclop`/`dupl` thresholds actually apply), and fix
(#77) all newly surfaced lint findings
- 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, - 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints,
Makefile shims, README Entrypoints section Makefile shims, README Entrypoints section
- 2026-03-25 pin golangci-lint Docker image for linting (#55) - 2026-03-25 pin golangci-lint Docker image for linting (#55)
@@ -54,10 +56,12 @@ capability in the README rationale).
# Future Steps # Future Steps
- Manual event redelivery from the web UI (replay is a core promised
capability in the README rationale)
- Delivery status and retry management UI - Delivery status and retry management UI
- Per-webhook rate limiting in the receiver handler (per-webhook config - Per-webhook rate limiting in the receiver handler (per-webhook config
plus handler enforcement, layered on the env-level receiver limit plus handler enforcement; global limits must not apply to receiver
from #64; global limits must not apply to receiver endpoints) endpoints)
- Webhook signature verification for GitHub and Stripe HMAC formats - Webhook signature verification for GitHub and Stripe HMAC formats
- API key authentication for programmatic access (APIKey model exists; - API key authentication for programmatic access (APIKey model exists;
Bearer token middleware does not) Bearer token middleware does not)

View File

@@ -31,23 +31,12 @@ const (
// defaultRetentionSweepInterval is how often the retention // defaultRetentionSweepInterval is how often the retention
// reaper deletes events older than each webhook's RetentionDays. // reaper deletes events older than each webhook's RetentionDays.
defaultRetentionSweepInterval = time.Hour defaultRetentionSweepInterval = time.Hour
// defaultReceiverRateLimit is the default number of requests
// per minute each client IP may send to a single webhook
// receiver entrypoint. Generous for legitimate webhook
// senders while bounding abuse of the one unauthenticated,
// internet-exposed endpoint.
defaultReceiverRateLimit = 120
) )
// ErrInvalidEnvironment is returned when WEBHOOKER_ENVIRONMENT // ErrInvalidEnvironment is returned when WEBHOOKER_ENVIRONMENT
// contains an unrecognised value. // contains an unrecognised value.
var ErrInvalidEnvironment = errors.New("invalid environment") var ErrInvalidEnvironment = errors.New("invalid environment")
// ErrNonPositiveValue is returned when an environment variable that
// requires a positive integer is set to zero or a negative number.
var ErrNonPositiveValue = errors.New("value must be positive")
//nolint:revive // ConfigParams is a standard fx naming convention. //nolint:revive // ConfigParams is a standard fx naming convention.
type ConfigParams struct { type ConfigParams struct {
fx.In fx.In
@@ -71,10 +60,6 @@ type Config struct {
// RetentionSweepInterval is how often the retention reaper runs. // RetentionSweepInterval is how often the retention reaper runs.
RetentionSweepInterval time.Duration RetentionSweepInterval time.Duration
// ReceiverRateLimit is the number of requests per minute each
// client IP may send to a single webhook receiver entrypoint.
ReceiverRateLimit int
params *ConfigParams params *ConfigParams
log *slog.Logger log *slog.Logger
} }
@@ -119,38 +104,6 @@ func envInt(key string, defaultValue int) int {
return defaultValue return defaultValue
} }
// envPositiveInt returns the value of the named environment variable
// parsed as a positive integer. Returns defaultValue if not set. If
// the variable is set but cannot be parsed, or parses to less than
// one, it returns a wrapped error naming the key and the bad value,
// so startup fails loudly rather than silently falling back to the
// default.
func envPositiveInt(
key string,
defaultValue int,
) (int, error) {
v := os.Getenv(key)
if v == "" {
return defaultValue, nil
}
i, err := strconv.Atoi(v)
if err != nil {
return 0, fmt.Errorf(
"invalid integer for %s: %q: %w", key, v, err,
)
}
if i < 1 {
return 0, fmt.Errorf(
"%w: %s must be at least 1, got %q",
ErrNonPositiveValue, key, v,
)
}
return i, nil
}
// envDuration returns the value of the named environment variable // envDuration returns the value of the named environment variable
// parsed as a Go duration (e.g. "1h", "30m"). Returns defaultValue if // parsed as a Go duration (e.g. "1h", "30m"). Returns defaultValue if
// not set. If the variable is set but cannot be parsed, it returns a // not set. If the variable is set but cannot be parsed, it returns a
@@ -209,17 +162,6 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
return nil, err return nil, err
} }
// Parse the receiver rate limit; a set-but-unparseable or
// non-positive value is a hard error so fx aborts startup
// rather than silently using the default.
receiverRateLimit, err := envPositiveInt(
"RECEIVER_RATE_LIMIT",
defaultReceiverRateLimit,
)
if err != nil {
return nil, err
}
// Load configuration values from environment variables // Load configuration values from environment variables
s := &Config{ s := &Config{
DataDir: envString("DATA_DIR"), DataDir: envString("DATA_DIR"),
@@ -231,7 +173,6 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
Port: envInt("PORT", defaultPort), Port: envInt("PORT", defaultPort),
SentryDSN: envString("SENTRY_DSN"), SentryDSN: envString("SENTRY_DSN"),
RetentionSweepInterval: retentionSweepInterval, RetentionSweepInterval: retentionSweepInterval,
ReceiverRateLimit: receiverRateLimit,
log: log, log: log,
params: &params, params: &params,
} }
@@ -256,7 +197,6 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
"maintenanceMode", s.MaintenanceMode, "maintenanceMode", s.MaintenanceMode,
"dataDir", s.DataDir, "dataDir", s.DataDir,
"retentionSweepInterval", s.RetentionSweepInterval.String(), "retentionSweepInterval", s.RetentionSweepInterval.String(),
"receiverRateLimit", s.ReceiverRateLimit,
"hasSentryDSN", s.SentryDSN != "", "hasSentryDSN", s.SentryDSN != "",
"hasMetricsAuth", "hasMetricsAuth",
s.MetricsUsername != "" && s.MetricsPassword != "", s.MetricsUsername != "" && s.MetricsPassword != "",

View File

@@ -258,109 +258,3 @@ func TestDefaultDataDir(t *testing.T) {
}) })
} }
} }
func TestReceiverRateLimit(t *testing.T) {
tests := []struct {
name string
set bool
value string
expectError bool
expected int
}{
{
name: "unset uses default",
set: false,
expected: 120,
},
{
name: "valid value is parsed",
set: true,
value: "30",
expected: 30,
},
{
name: "unparseable value fails startup",
set: true,
value: "not-a-number",
expectError: true,
},
{
name: "zero fails startup",
set: true,
value: "0",
expectError: true,
},
{
name: "negative fails startup",
set: true,
value: "-5",
expectError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Cannot use t.Parallel() here because t.Setenv
// is incompatible with parallel subtests.
t.Setenv("WEBHOOKER_ENVIRONMENT", "dev")
if tt.set {
t.Setenv("RECEIVER_RATE_LIMIT", tt.value)
} else {
require.NoError(t, os.Unsetenv(
"RECEIVER_RATE_LIMIT",
))
}
if tt.expectError {
testReceiverRateLimitError(t)
} else {
testReceiverRateLimitSuccess(t, tt.expected)
}
})
}
}
func testReceiverRateLimitError(t *testing.T) {
t.Helper()
var cfg *config.Config
app := fx.New(
fx.NopLogger,
fx.Provide(
globals.New,
logger.New,
config.New,
),
fx.Populate(&cfg),
)
assert.Error(t, app.Err())
}
func testReceiverRateLimitSuccess(
t *testing.T,
expected int,
) {
t.Helper()
var cfg *config.Config
app := fxtest.New(
t,
fx.Provide(
globals.New,
logger.New,
config.New,
),
fx.Populate(&cfg),
)
require.NoError(t, app.Err())
app.RequireStart()
defer app.RequireStop()
assert.Equal(t, expected, cfg.ReceiverRateLimit)
}

View File

@@ -11,6 +11,15 @@ import (
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
const (
// testAppname is the Globals.Appname used in tests.
testAppname = "webhooker-test"
// testVersion is the Globals.Version used in tests.
testVersion = "test"
// testContentType is the event content type used in tests.
testContentType = "application/json"
)
func setupTestDB( func setupTestDB(
t *testing.T, t *testing.T,
) (*database.Database, *fxtest.Lifecycle) { ) (*database.Database, *fxtest.Lifecycle) {
@@ -19,8 +28,8 @@ func setupTestDB(
lc := fxtest.NewLifecycle(t) lc := fxtest.NewLifecycle(t)
g := &globals.Globals{ g := &globals.Globals{
Appname: "webhooker-test", Appname: testAppname,
Version: "test", Version: testVersion,
} }
l, err := logger.New( l, err := logger.New(

View File

@@ -5,7 +5,10 @@ type Entrypoint struct {
BaseModel BaseModel
WebhookID string `gorm:"type:uuid;not null" json:"webhookId"` WebhookID string `gorm:"type:uuid;not null" json:"webhookId"`
Path string `gorm:"uniqueIndex;not null" json:"path"` // URL path for this entrypoint
// Path is the URL path for this entrypoint.
Path string `gorm:"uniqueIndex;not null" json:"path"`
Description string `json:"description"` Description string `json:"description"`
Active bool `gorm:"default:true" json:"active"` Active bool `gorm:"default:true" json:"active"`

View File

@@ -23,7 +23,8 @@ type Target struct {
// Configuration fields (JSON stored based on type) // Configuration fields (JSON stored based on type)
Config string `gorm:"type:text" json:"config"` // JSON configuration Config string `gorm:"type:text" json:"config"` // JSON configuration
// For HTTP targets (max_retries=0 means fire-and-forget, >0 enables retries with backoff) // For HTTP targets (max_retries=0 means fire-and-forget,
// >0 enables retries with backoff)
MaxRetries int `json:"maxRetries,omitempty"` MaxRetries int `json:"maxRetries,omitempty"`
MaxQueueSize int `json:"maxQueueSize,omitempty"` MaxQueueSize int `json:"maxQueueSize,omitempty"`

View File

@@ -7,7 +7,9 @@ type Webhook struct {
UserID string `gorm:"type:uuid;not null" json:"userId"` UserID string `gorm:"type:uuid;not null" json:"userId"`
Name string `gorm:"not null" json:"name"` Name string `gorm:"not null" json:"name"`
Description string `json:"description"` Description string `json:"description"`
RetentionDays int `gorm:"default:30" json:"retentionDays"` // Days to retain events
// RetentionDays is the number of days to retain events.
RetentionDays int `gorm:"default:30" json:"retentionDays"`
// Relations // Relations
User User `json:"user,omitzero"` User User `json:"user,omitzero"`

View File

@@ -2,6 +2,7 @@ package database_test
import ( import (
"context" "context"
"net/http"
"testing" "testing"
"time" "time"
@@ -30,8 +31,8 @@ func setupRetentionTest(t *testing.T) *retentionTestEnv {
lc := fxtest.NewLifecycle(t) lc := fxtest.NewLifecycle(t)
g := &globals.Globals{ g := &globals.Globals{
Appname: "webhooker-test", Appname: testAppname,
Version: "test", Version: testVersion,
} }
l, err := logger.New(lc, logger.LoggerParams{Globals: g}) l, err := logger.New(lc, logger.LoggerParams{Globals: g})
@@ -117,9 +118,9 @@ func seedEventChain(
event := &database.Event{ event := &database.Event{
WebhookID: webhookID, WebhookID: webhookID,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Body: `{"seed": true}`, Body: `{"seed": true}`,
ContentType: "application/json", ContentType: testContentType,
} }
event.CreatedAt = createdAt event.CreatedAt = createdAt
require.NoError(t, db.Create(event).Error) require.NoError(t, db.Create(event).Error)

View File

@@ -14,7 +14,10 @@ import (
func NewTestDatabase(db *gorm.DB) *Database { func NewTestDatabase(db *gorm.DB) *Database {
return &Database{ return &Database{
db: db, db: db,
log: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})), log: slog.New(slog.NewTextHandler(
os.Stderr,
&slog.HandlerOptions{Level: slog.LevelDebug},
)),
} }
} }
@@ -23,6 +26,9 @@ func NewTestDatabase(db *gorm.DB) *Database {
func NewTestWebhookDBManager(dataDir string) *WebhookDBManager { func NewTestWebhookDBManager(dataDir string) *WebhookDBManager {
return &WebhookDBManager{ return &WebhookDBManager{
dataDir: dataDir, dataDir: dataDir,
log: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})), log: slog.New(slog.NewTextHandler(
os.Stderr,
&slog.HandlerOptions{Level: slog.LevelDebug},
)),
} }
} }

View File

@@ -2,6 +2,7 @@ package database_test
import ( import (
"context" "context"
"net/http"
"os" "os"
"path/filepath" "path/filepath"
"testing" "testing"
@@ -25,8 +26,8 @@ func setupTestWebhookDBManager(
lc := fxtest.NewLifecycle(t) lc := fxtest.NewLifecycle(t)
g := &globals.Globals{ g := &globals.Globals{
Appname: "webhooker-test", Appname: testAppname,
Version: "test", Version: testVersion,
} }
l, err := logger.New( l, err := logger.New(
@@ -83,10 +84,10 @@ func TestWebhookDBManager_CreateAndGetDB(t *testing.T) {
event := &database.Event{ event := &database.Event{
WebhookID: webhookID, WebhookID: webhookID,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Headers: `{"Content-Type":["application/json"]}`, Headers: `{"Content-Type":["application/json"]}`,
Body: `{"test": true}`, Body: `{"test": true}`,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db.Create(event).Error) require.NoError(t, db.Create(event).Error)
assert.NotEmpty(t, event.ID) assert.NotEmpty(t, event.ID)
@@ -99,7 +100,7 @@ func TestWebhookDBManager_CreateAndGetDB(t *testing.T) {
db.First(&readEvent, "id = ?", event.ID).Error, db.First(&readEvent, "id = ?", event.ID).Error,
) )
assert.Equal(t, webhookID, readEvent.WebhookID) assert.Equal(t, webhookID, readEvent.WebhookID)
assert.Equal(t, "POST", readEvent.Method) assert.Equal(t, http.MethodPost, readEvent.Method)
assert.Equal(t, `{"test": true}`, readEvent.Body) assert.Equal(t, `{"test": true}`, readEvent.Body)
} }
@@ -123,9 +124,9 @@ func TestWebhookDBManager_DeleteDB(t *testing.T) {
event := &database.Event{ event := &database.Event{
WebhookID: webhookID, WebhookID: webhookID,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Body: `{"test": true}`, Body: `{"test": true}`,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db.Create(event).Error) require.NoError(t, db.Create(event).Error)
@@ -196,10 +197,10 @@ func seedDeliveryWorkflow(
event := &database.Event{ event := &database.Event{
WebhookID: webhookID, WebhookID: webhookID,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Headers: `{"Content-Type":["application/json"]}`, Headers: `{"Content-Type":["application/json"]}`,
Body: `{"payload": "test"}`, Body: `{"payload": "test"}`,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db.Create(event).Error) require.NoError(t, db.Create(event).Error)
@@ -231,7 +232,7 @@ func verifyPendingDeliveries(
) )
require.Len(t, pending, 1) require.Len(t, pending, 1)
assert.Equal(t, event.ID, pending[0].EventID) assert.Equal(t, event.ID, pending[0].EventID)
assert.Equal(t, "POST", pending[0].Event.Method) assert.Equal(t, http.MethodPost, pending[0].Event.Method)
} }
func completeDelivery( func completeDelivery(
@@ -303,16 +304,16 @@ func TestWebhookDBManager_MultipleWebhooks(t *testing.T) {
event1 := &database.Event{ event1 := &database.Event{
WebhookID: webhook1, WebhookID: webhook1,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Body: `{"webhook": 1}`, Body: `{"webhook": 1}`,
ContentType: "application/json", ContentType: testContentType,
} }
event2 := &database.Event{ event2 := &database.Event{
WebhookID: webhook2, WebhookID: webhook2,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "PUT", Method: http.MethodPut,
Body: `{"webhook": 2}`, Body: `{"webhook": 2}`,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db1.Create(event1).Error) require.NoError(t, db1.Create(event1).Error)

View File

@@ -126,36 +126,6 @@ func iHTTPConfig(url string) string {
return string(data) return string(data)
} }
func iWebhookDB(t *testing.T) *gorm.DB {
t.Helper()
dbPath := filepath.Join(
t.TempDir(), "events-test.db",
)
dsn := fmt.Sprintf(
"file:%s?cache=shared&mode=rwc", dbPath,
)
sqlDB, err := sql.Open("sqlite", dsn)
require.NoError(t, err)
t.Cleanup(func() { _ = sqlDB.Close() })
db, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
require.NoError(t, err)
require.NoError(t, db.AutoMigrate(
&database.Event{},
&database.Delivery{},
&database.DeliveryResult{},
))
return db
}
func iEngine( func iEngine(
t *testing.T, workers int, t *testing.T, workers int,
) *delivery.Engine { ) *delivery.Engine {
@@ -182,10 +152,10 @@ func iSeedEvent(
event := database.Event{ event := database.Event{
WebhookID: webhookID, WebhookID: webhookID,
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Headers: `{}`, Headers: `{}`,
Body: body, Body: body,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db.Create(&event).Error) require.NoError(t, db.Create(&event).Error)
@@ -935,7 +905,7 @@ func TestDeliverHTTP_CustomTargetHeaders(t *testing.T) {
func TestDeliverHTTP_TargetTimeout(t *testing.T) { func TestDeliverHTTP_TargetTimeout(t *testing.T) {
t.Parallel() t.Parallel()
db := iWebhookDB(t) db := testWebhookDB(t)
e := iEngine(t, 1) e := iEngine(t, 1)
ts := httptest.NewServer( ts := httptest.NewServer(
@@ -987,10 +957,10 @@ func iSeedEventAndDelivery(
event := database.Event{ event := database.Event{
WebhookID: uuid.New().String(), WebhookID: uuid.New().String(),
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Headers: `{"Content-Type":["application/json"]}`, Headers: `{"Content-Type":["application/json"]}`,
Body: body, Body: body,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db.Create(&event).Error) require.NoError(t, db.Create(&event).Error)
@@ -1067,7 +1037,7 @@ func iAssertResultFailed(
func TestDeliverHTTP_InvalidConfig(t *testing.T) { func TestDeliverHTTP_InvalidConfig(t *testing.T) {
t.Parallel() t.Parallel()
db := iWebhookDB(t) db := testWebhookDB(t)
e := iEngine(t, 1) e := iEngine(t, 1)
event, del := iSeedEventAndDelivery( event, del := iSeedEventAndDelivery(

View File

@@ -27,6 +27,9 @@ import (
"sneak.berlin/go/webhooker/internal/delivery" "sneak.berlin/go/webhooker/internal/delivery"
) )
// testContentType is the event content type used in tests.
const testContentType = "application/json"
func testWebhookDB(t *testing.T) *gorm.DB { func testWebhookDB(t *testing.T) *gorm.DB {
t.Helper() t.Helper()
@@ -94,10 +97,10 @@ func seedEvent(
event := database.Event{ event := database.Event{
WebhookID: uuid.New().String(), WebhookID: uuid.New().String(),
EntrypointID: uuid.New().String(), EntrypointID: uuid.New().String(),
Method: "POST", Method: http.MethodPost,
Headers: `{"Content-Type":["application/json"]}`, Headers: `{"Content-Type":["application/json"]}`,
Body: body, Body: body,
ContentType: "application/json", ContentType: testContentType,
} }
require.NoError(t, db.Create(&event).Error) require.NoError(t, db.Create(&event).Error)
@@ -342,33 +345,29 @@ func TestDeliverDatabase_ImmediateSuccess(
t.Parallel() t.Parallel()
db := testWebhookDB(t) db := testWebhookDB(t)
e := testEngine(t, 1)
event := seedEvent(t, db, `{"db":"target"}`) // The database target archives for real now, so the engine
// needs a webhook DB manager to locate the data directory.
dlv := seedDelivery( e := delivery.NewTestEngineWithDB(
t, db, event.ID, uuid.New().String(), nil,
database.DeliveryStatusPending, database.NewTestWebhookDBManager(t.TempDir()),
slog.New(slog.NewTextHandler(
os.Stderr,
&slog.HandlerOptions{Level: slog.LevelDebug},
)),
&http.Client{Timeout: 5 * time.Second},
1,
) )
d := &database.Delivery{ event := seedEvent(t, db, `{"db":"target"}`)
EventID: event.ID, d := seedDatabaseTargetDelivery(t, db, event, "")
TargetID: dlv.TargetID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-db",
Type: database.TargetTypeDatabase,
},
}
d.ID = dlv.ID
e.ExportDeliverDatabase(db, d) e.ExportDeliverDatabase(db, d)
var updated database.Delivery var updated database.Delivery
require.NoError(t, db.First( require.NoError(t, db.First(
&updated, "id = ?", dlv.ID, &updated, "id = ?", d.ID,
).Error) ).Error)
assert.Equal(t, assert.Equal(t,
@@ -379,7 +378,7 @@ func TestDeliverDatabase_ImmediateSuccess(
var result database.DeliveryResult var result database.DeliveryResult
require.NoError(t, db.Where( require.NoError(t, db.Where(
"delivery_id = ?", dlv.ID, "delivery_id = ?", d.ID,
).First(&result).Error) ).First(&result).Error)
assert.True(t, result.Success) assert.True(t, result.Success)
@@ -1117,10 +1116,10 @@ func TestDoHTTPRequest_ForwardsHeaders(t *testing.T) {
} }
event := &database.Event{ event := &database.Event{
Method: "POST", Method: http.MethodPost,
Headers: `{"X-Custom":["value1"],"Content-Type":["application/json"]}`, Headers: `{"X-Custom":["value1"],"Content-Type":["application/json"]}`,
Body: `{"test":true}`, Body: `{"test":true}`,
ContentType: "application/json", ContentType: testContentType,
} }
statusCode, _, _, err := e.ExportDoHTTPRequest( statusCode, _, _, err := e.ExportDoHTTPRequest(
@@ -1142,7 +1141,7 @@ func TestDoHTTPRequest_ForwardsHeaders(t *testing.T) {
) )
assert.Equal(t, assert.Equal(t,
"application/json", testContentType,
receivedHeaders.Get("Content-Type"), receivedHeaders.Get("Content-Type"),
) )
@@ -1158,7 +1157,19 @@ func TestProcessDelivery_RoutesToCorrectHandler(
t.Parallel() t.Parallel()
db := testWebhookDB(t) db := testWebhookDB(t)
e := testEngine(t, 1)
// The database target archives for real now, so the engine
// needs a webhook DB manager to locate the data directory.
e := delivery.NewTestEngineWithDB(
nil,
database.NewTestWebhookDBManager(t.TempDir()),
slog.New(slog.NewTextHandler(
os.Stderr,
&slog.HandlerOptions{Level: slog.LevelDebug},
)),
&http.Client{Timeout: 5 * time.Second},
1,
)
tests := []struct { tests := []struct {
name string name string
@@ -1289,8 +1300,8 @@ func TestFormatSlackMessage_JSONBody(t *testing.T) {
t.Parallel() t.Parallel()
event := &database.Event{ event := &database.Event{
Method: "POST", Method: http.MethodPost,
ContentType: "application/json", ContentType: testContentType,
Body: `{"action":"push",` + Body: `{"action":"push",` +
`"repo":"test/repo",` + `"repo":"test/repo",` +
`"ref":"refs/heads/main"}`, `"ref":"refs/heads/main"}`,
@@ -1315,7 +1326,7 @@ func TestFormatSlackMessage_NonJSONBody(t *testing.T) {
t.Parallel() t.Parallel()
event := &database.Event{ event := &database.Event{
Method: "POST", Method: http.MethodPost,
ContentType: "text/plain", ContentType: "text/plain",
Body: "hello world plain text", Body: "hello world plain text",
} }
@@ -1338,8 +1349,8 @@ func TestFormatSlackMessage_EmptyBody(t *testing.T) {
t.Parallel() t.Parallel()
event := &database.Event{ event := &database.Event{
Method: "POST", Method: http.MethodPost,
ContentType: "application/json", ContentType: testContentType,
Body: "", Body: "",
} }
event.CreatedAt = time.Date( event.CreatedAt = time.Date(
@@ -1367,8 +1378,8 @@ func TestFormatSlackMessage_LargeJSONTruncated(
require.NoError(t, err) require.NoError(t, err)
event := &database.Event{ event := &database.Event{
Method: "POST", Method: http.MethodPost,
ContentType: "application/json", ContentType: testContentType,
Body: string(largeJSON), Body: string(largeJSON),
} }
event.CreatedAt = time.Date( event.CreatedAt = time.Date(
@@ -1697,7 +1708,7 @@ func assertLogLineComplete(
"log line must contain the webhook id", "log line must contain the webhook id",
) )
assert.Contains(t, out, "application/json", assert.Contains(t, out, testContentType,
"log line must contain the content type", "log line must contain the content type",
) )
} }

View File

@@ -273,3 +273,64 @@ func NewTestCircuitBreaker(
cooldown: cooldown, cooldown: cooldown,
} }
} }
// ExportArchivedEvent aliases the archive row type so black-box
// tests can construct and read archive rows.
type ExportArchivedEvent = archivedEvent
// ExportArchiveWriter wraps an archiveWriter so black-box tests
// can exercise the per-webhook archive file mechanics.
type ExportArchiveWriter struct {
w *archiveWriter
}
// NewExportArchiveWriter builds an archive writer for tests,
// optionally overriding the reopen debounce (a non-positive
// debounce keeps the production default).
func NewExportArchiveWriter(
path string, log *slog.Logger, debounce time.Duration,
) *ExportArchiveWriter {
w := newArchiveWriter(path, log)
if debounce > 0 {
w.debounce = debounce
}
return &ExportArchiveWriter{w: w}
}
// Write archives a row through the writer.
func (e *ExportArchiveWriter) Write(
row ExportArchivedEvent, expiry time.Duration,
) error {
return e.w.write(row, expiry)
}
// Open opens the archive file, pruning when expiry is positive.
func (e *ExportArchiveWriter) Open(expiry time.Duration) error {
return e.w.open(expiry)
}
// Reopen closes and reopens the archive file.
func (e *ExportArchiveWriter) Reopen(
expiry time.Duration,
) error {
return e.w.reopen(expiry)
}
// Reopens reports how many times the file has been opened.
func (e *ExportArchiveWriter) Reopens() int {
return e.w.reopens
}
// DB returns the writer's current open handle for row
// inspection in tests.
func (e *ExportArchiveWriter) DB() *gorm.DB {
return e.w.db
}
// ExportParseArchiveExpiry exposes parseArchiveExpiry.
func ExportParseArchiveExpiry(
configJSON string,
) (time.Duration, error) {
return parseArchiveExpiry(configJSON)
}

View File

@@ -2,21 +2,38 @@ package delivery
import ( import (
"context" "context"
"fmt"
"path/filepath"
"sync"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
) )
// databaseTarget is a fire-and-forget target: the event is // databaseTarget is a no-retry target that archives the
// already persisted in the per-webhook database by the time // full inbound event into a per-webhook archive SQLite file,
// delivery runs, so the target records a single successful // separate from the per-webhook event database. The event is
// attempt. (Durable archiving to a separate store is tracked // already persisted in the per-webhook event DB by the time
// as its own work.) // delivery runs; the database target additionally writes a
// durable long-term copy into archive-{webhookID}.db and then
// records a single attempt whose outcome reflects whether the
// archive write succeeded. See archiveWriter for the
// close/reopen, auto-recreate, and expiry semantics.
type databaseTarget struct { type databaseTarget struct {
eng *Engine eng *Engine
mu sync.Mutex
writers map[string]*archiveWriter
} }
// Deliver implements Target. // Deliver implements Target. It archives the event, then
// records one successful attempt and marks the delivery
// delivered. An archiving error fails the delivery: the
// attempt is recorded as failed with the error and the
// delivery is marked failed, so a target that could not do
// its one job (archiving) never reports success. The target
// does not retry; the event remains durably stored in the
// per-webhook event database.
func (t *databaseTarget) Deliver( func (t *databaseTarget) Deliver(
_ context.Context, _ context.Context,
webhookDB *gorm.DB, webhookDB *gorm.DB,
@@ -24,6 +41,27 @@ func (t *databaseTarget) Deliver(
_ *Task, _ *Task,
_ Scheduler, _ Scheduler,
) { ) {
err := t.archive(d)
if err != nil {
t.eng.log.Error(
"failed to archive event to database target",
"delivery_id", d.ID,
"event_id", d.EventID,
"error", err,
)
t.eng.recordResult(
webhookDB, d, 1, false, 0, "",
err.Error(), 0,
)
t.eng.updateDeliveryStatus(
webhookDB, d, database.DeliveryStatusFailed,
)
return
}
t.eng.recordResult( t.eng.recordResult(
webhookDB, d, 1, true, 0, "", "", 0, webhookDB, d, 1, true, 0, "", "", 0,
) )
@@ -32,3 +70,68 @@ func (t *databaseTarget) Deliver(
webhookDB, d, database.DeliveryStatusDelivered, webhookDB, d, database.DeliveryStatusDelivered,
) )
} }
// archive writes the full event as a row into the webhook's
// archive database, honouring the optional per-target expiry
// parsed from the target config JSON.
func (t *databaseTarget) archive(d *database.Delivery) error {
webhookID := d.Event.WebhookID
if webhookID == "" {
return errArchiveMissingWebhookID
}
expiry, err := parseArchiveExpiry(d.Target.Config)
if err != nil {
return err
}
w, err := t.writerFor(webhookID)
if err != nil {
return err
}
row := archivedEvent{
EventID: d.Event.ID,
WebhookID: webhookID,
EntrypointID: d.Event.EntrypointID,
Method: d.Event.Method,
Headers: d.Event.Headers,
Body: d.Event.Body,
ContentType: d.Event.ContentType,
}
return w.write(row, expiry)
}
// writerFor returns the archiveWriter for a webhook, creating
// and caching it on first use. Each webhook has one writer so
// its close/reopen debounce state is shared across concurrent
// deliveries. The archive file lives beside the per-webhook
// event database in the data directory.
func (t *databaseTarget) writerFor(
webhookID string,
) (*archiveWriter, error) {
if t.eng.dbManager == nil {
return nil, errArchiveNoDataDir
}
dir := filepath.Dir(t.eng.dbManager.DBPath(webhookID))
path := filepath.Join(
dir, fmt.Sprintf("archive-%s.db", webhookID),
)
t.mu.Lock()
defer t.mu.Unlock()
if t.writers == nil {
t.writers = make(map[string]*archiveWriter)
}
w, ok := t.writers[webhookID]
if !ok {
w = newArchiveWriter(path, t.eng.log)
t.writers[webhookID] = w
}
return w, nil
}

View File

@@ -0,0 +1,312 @@
package delivery
import (
"database/sql"
"encoding/json"
"errors"
"fmt"
"log/slog"
"os"
"sync"
"time"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
// archiveExpiryNever is the expiry sentinel (and default) that
// disables pruning so archived rows are kept forever.
const archiveExpiryNever = "never"
// archiveReopenDebounce bounds how often an archive file is
// closed and reopened. After each write the handle is closed
// and reopened so an operator can move the file away for
// offline archiving, but never more than once per this window.
const archiveReopenDebounce = time.Second
var (
// errArchiveMissingWebhookID is returned when an event to
// archive has no webhook id to key its archive file on.
errArchiveMissingWebhookID = errors.New(
"cannot archive event without a webhook id",
)
// errArchiveNoDataDir is returned when the database target
// has no webhook database manager and so cannot locate the
// data directory for archive files.
errArchiveNoDataDir = errors.New(
"database target has no data directory",
)
// errArchiveExpiryNotPositive is returned when a
// user-supplied archive expiry parses as a duration but is
// zero or negative; "never" is the way to disable pruning.
errArchiveExpiryNotPositive = errors.New(
"expiry must be a positive duration or \"never\"",
)
)
// databaseTargetConfig is the optional per-target JSON config
// for a database (archive) target.
type databaseTargetConfig struct {
// Expiry is a Go duration (e.g. "720h") after which
// archived rows are pruned, or "never" (the default) to
// keep them forever.
Expiry string `json:"expiry"`
}
// archivedEvent is one fully captured webhook event stored in a
// per-webhook archive database for long-term retention. It is a
// self-contained copy — independent of the per-webhook event
// database, which may prune events under its own retention.
type archivedEvent struct {
ID uint `gorm:"primaryKey;autoIncrement"`
EventID string `gorm:"index"`
WebhookID string
EntrypointID string
Method string
Headers string
Body string
ContentType string
// ArchivedAt is when the row was archived and is the age
// basis for expiry pruning.
ArchivedAt time.Time `gorm:"index"`
}
// parseArchiveExpiry reads the optional expiry from a database
// target's config JSON. An empty config, an empty expiry, or
// the literal "never" all mean keep forever, returned as a zero
// duration. Any other value must parse as a positive Go
// duration; a set-but-invalid value (unparseable, zero, or
// negative) is an error rather than a silent default, matching
// ValidateArchiveExpiry at target creation.
func parseArchiveExpiry(
configJSON string,
) (time.Duration, error) {
if configJSON == "" {
return 0, nil
}
var cfg databaseTargetConfig
err := json.Unmarshal([]byte(configJSON), &cfg)
if err != nil {
return 0, fmt.Errorf(
"parsing database target config: %w", err,
)
}
if cfg.Expiry == "" || cfg.Expiry == archiveExpiryNever {
return 0, nil
}
dur, err := time.ParseDuration(cfg.Expiry)
if err != nil {
return 0, fmt.Errorf(
"parsing archive expiry %q: %w", cfg.Expiry, err,
)
}
if dur <= 0 {
return 0, fmt.Errorf(
"%w: %q", errArchiveExpiryNotPositive, cfg.Expiry,
)
}
return dur, nil
}
// ValidateArchiveExpiry checks a user-supplied archive expiry
// for a database target at configuration time. Valid values are
// empty, "never" (both meaning keep forever), or a positive Go
// duration such as "720h". Anything else is an error, so a bad
// expiry is rejected when the target is created rather than
// failing every subsequent delivery.
func ValidateArchiveExpiry(expiry string) error {
if expiry == "" || expiry == archiveExpiryNever {
return nil
}
dur, err := time.ParseDuration(expiry)
if err != nil {
return fmt.Errorf(
"expiry must be %q or a Go duration "+
"such as \"720h\": %w",
archiveExpiryNever, err,
)
}
if dur <= 0 {
return fmt.Errorf(
"%w: %q", errArchiveExpiryNotPositive, expiry,
)
}
return nil
}
// archiveWriter owns one per-webhook archive SQLite file. It
// serialises writes, and after each write closes and reopens
// the file (debounced to at most once per debounce window) so
// an operator can move the file away for offline archiving. The
// next write recreates a moved or removed file, because the
// file is opened create-if-missing and its schema is migrated
// on every open.
type archiveWriter struct {
mu sync.Mutex
path string
log *slog.Logger
debounce time.Duration
db *gorm.DB
lastReopen time.Time
reopens int
}
// newArchiveWriter builds an archiveWriter for a file path with
// the default reopen debounce.
func newArchiveWriter(
path string, log *slog.Logger,
) *archiveWriter {
return &archiveWriter{
path: path,
log: log,
debounce: archiveReopenDebounce,
}
}
// write appends the event as a row, then applies the debounced
// close/reopen. It recreates the archive file if it was moved
// or removed since the last open. A positive expiry prunes rows
// older than it on each (re)open.
func (w *archiveWriter) write(
row archivedEvent, expiry time.Duration,
) error {
w.mu.Lock()
defer w.mu.Unlock()
if w.db == nil || !fileExists(w.path) {
err := w.reopen(expiry)
if err != nil {
return err
}
}
row.ArchivedAt = time.Now()
err := w.db.Create(&row).Error
if err != nil {
return fmt.Errorf(
"archiving event to %s: %w", w.path, err,
)
}
if time.Since(w.lastReopen) >= w.debounce {
return w.reopen(expiry)
}
return nil
}
// open opens (creating if missing) the archive file, migrates
// its schema, records the reopen time, and prunes expired rows
// when expiry is positive.
func (w *archiveWriter) open(expiry time.Duration) error {
dbURL := fmt.Sprintf("file:%s?mode=rwc", w.path)
sqlDB, err := sql.Open("sqlite", dbURL)
if err != nil {
return fmt.Errorf(
"opening archive database %s: %w", w.path, err,
)
}
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
if err != nil {
_ = sqlDB.Close()
return fmt.Errorf(
"connecting to archive database %s: %w",
w.path, err,
)
}
err = gdb.AutoMigrate(&archivedEvent{})
if err != nil {
_ = sqlDB.Close()
return fmt.Errorf(
"migrating archive database %s: %w", w.path, err,
)
}
w.db = gdb
w.lastReopen = time.Now()
w.reopens++
if expiry > 0 {
w.prune(expiry)
}
return nil
}
// reopen closes any open handle and opens the file afresh. The
// fresh open recreates the file if it was moved away.
func (w *archiveWriter) reopen(expiry time.Duration) error {
w.close()
return w.open(expiry)
}
// close closes the underlying handle, if any.
func (w *archiveWriter) close() {
if w.db == nil {
return
}
sqlDB, err := w.db.DB()
if err == nil {
_ = sqlDB.Close()
}
w.db = nil
}
// prune deletes archived rows older than expiry, measured from
// each row's archived time. It runs on every (re)open, and
// because the file is reopened after writes this keeps the
// archive swept without a separate background sweeper. Failures
// are logged, not fatal: a prune error must not stop archiving.
func (w *archiveWriter) prune(expiry time.Duration) {
cutoff := time.Now().Add(-expiry)
res := w.db.Where("archived_at < ?", cutoff).
Delete(&archivedEvent{})
if res.Error != nil {
w.log.Error(
"failed to prune expired archive rows",
"path", w.path,
"error", res.Error,
)
return
}
if res.RowsAffected > 0 {
w.log.Info(
"pruned expired archive rows",
"path", w.path,
"rows_deleted", res.RowsAffected,
)
}
}
// fileExists reports whether a path currently exists.
func fileExists(path string) bool {
_, err := os.Stat(path)
return err == nil
}

View File

@@ -0,0 +1,395 @@
package delivery_test
import (
"database/sql"
"fmt"
"log/slog"
"net/http"
"os"
"path/filepath"
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
_ "modernc.org/sqlite" // Pure Go SQLite driver.
"sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/delivery"
)
func archiveTestLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(
os.Stderr,
&slog.HandlerOptions{Level: slog.LevelDebug},
))
}
// openArchiveDBForRead opens an archive file read-only so a
// test can inspect the rows the writer persisted.
func openArchiveDBForRead(
t *testing.T, path string,
) *gorm.DB {
t.Helper()
sqlDB, err := sql.Open(
"sqlite",
fmt.Sprintf("file:%s?mode=ro", path),
)
require.NoError(t, err)
t.Cleanup(func() { _ = sqlDB.Close() })
gdb, err := gorm.Open(
sqlite.Dialector{Conn: sqlDB}, &gorm.Config{},
)
require.NoError(t, err)
return gdb
}
// removeArchiveFiles simulates an operator moving the archive
// away by deleting the SQLite file and its sidecar files.
func removeArchiveFiles(t *testing.T, path string) {
t.Helper()
for _, suffix := range []string{
"", "-wal", "-shm", "-journal",
} {
err := os.Remove(path + suffix)
if err != nil && !os.IsNotExist(err) {
t.Fatalf("removing %s%s: %v", path, suffix, err)
}
}
}
// TestDeliverDatabase_ArchivesEvent verifies that delivering to
// a database target marks the delivery delivered and archives
// the full event into a separate per-webhook archive file.
func TestDeliverDatabase_ArchivesEvent(t *testing.T) {
t.Parallel()
dataDir := t.TempDir()
dbMgr := database.NewTestWebhookDBManager(dataDir)
e := delivery.NewTestEngineWithDB(
nil, dbMgr,
archiveTestLogger(),
&http.Client{Timeout: 5 * time.Second},
1,
)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"archived":true}`)
d := seedDatabaseTargetDelivery(t, webhookDB, event, "")
e.ExportDeliverDatabase(webhookDB, d)
var updated database.Delivery
require.NoError(t, webhookDB.First(
&updated, "id = ?", d.ID,
).Error)
assert.Equal(t,
database.DeliveryStatusDelivered, updated.Status,
"database target should mark the delivery delivered",
)
archivePath := filepath.Join(
dataDir,
fmt.Sprintf("archive-%s.db", event.WebhookID),
)
assert.FileExists(t, archivePath)
rdb := openArchiveDBForRead(t, archivePath)
var rows []delivery.ExportArchivedEvent
require.NoError(t, rdb.Find(&rows).Error)
require.Len(t, rows, 1)
assert.Equal(t, event.ID, rows[0].EventID)
assert.Equal(t, event.WebhookID, rows[0].WebhookID)
assert.Equal(t, event.Method, rows[0].Method)
assert.JSONEq(t, `{"archived":true}`, rows[0].Body)
}
func TestArchiveWriter_WritesRow(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
row := delivery.ExportArchivedEvent{
EventID: "ev-1",
WebhookID: "wh-1",
EntrypointID: "ep-1",
Method: "POST",
Headers: `{"X":"Y"}`,
Body: `{"hello":"world"}`,
ContentType: "application/json",
}
require.NoError(t, w.Write(row, 0))
assert.FileExists(t, path)
var got []delivery.ExportArchivedEvent
require.NoError(t, w.DB().Find(&got).Error)
require.Len(t, got, 1)
assert.Equal(t, "ev-1", got[0].EventID)
assert.Equal(t, "wh-1", got[0].WebhookID)
assert.Equal(t, "ep-1", got[0].EntrypointID)
assert.Equal(t, row.Method, got[0].Method)
assert.Equal(t, row.ContentType, got[0].ContentType)
assert.JSONEq(t, `{"hello":"world"}`, got[0].Body)
assert.False(t, got[0].ArchivedAt.IsZero())
}
func TestArchiveWriter_RecreatesAfterRemoval(
t *testing.T,
) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "a"}, 0,
))
assert.FileExists(t, path)
// The operator moves the archive away while the handle is
// still open.
removeArchiveFiles(t, path)
require.NoFileExists(t, path)
// The next write recreates the file with a fresh schema and
// only the new row.
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "b"}, 0,
))
assert.FileExists(t, path)
var got []delivery.ExportArchivedEvent
require.NoError(t, w.DB().Find(&got).Error)
require.Len(t, got, 1)
assert.Equal(t, "b", got[0].EventID)
}
func TestArchiveWriter_ReopenDebounce(t *testing.T) {
t.Parallel()
// A generous debounce keeps the two rapid writes inside
// the window even on a heavily loaded test machine.
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 2*time.Second,
)
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "a"}, 0,
))
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "b"}, 0,
))
// Two writes inside the debounce window trigger only the
// initial open — no extra close/reopen.
assert.Equal(t, 1, w.Reopens())
time.Sleep(2100 * time.Millisecond)
require.NoError(t, w.Write(
delivery.ExportArchivedEvent{EventID: "c"}, 0,
))
// A write after the window elapses closes and reopens once.
assert.Equal(t, 2, w.Reopens())
}
func TestArchiveWriter_ExpiryPrune(t *testing.T) {
t.Parallel()
path := filepath.Join(t.TempDir(), "archive-wh.db")
w := delivery.NewExportArchiveWriter(
path, archiveTestLogger(), 0,
)
require.NoError(t, w.Open(0))
old := delivery.ExportArchivedEvent{
EventID: "old",
ArchivedAt: time.Now().Add(-2 * time.Hour),
}
fresh := delivery.ExportArchivedEvent{
EventID: "fresh",
ArchivedAt: time.Now(),
}
require.NoError(t, w.DB().Create(&old).Error)
require.NoError(t, w.DB().Create(&fresh).Error)
// Reopening with a one-hour expiry prunes the old row.
require.NoError(t, w.Reopen(time.Hour))
var got []delivery.ExportArchivedEvent
require.NoError(t, w.DB().Find(&got).Error)
require.Len(t, got, 1)
assert.Equal(t, "fresh", got[0].EventID)
}
func TestParseArchiveExpiry(t *testing.T) {
t.Parallel()
cases := []struct {
name string
in string
want time.Duration
wantErr bool
}{
{"empty config", "", 0, false},
{"explicit never", `{"expiry":"never"}`, 0, false},
{"empty expiry", `{"expiry":""}`, 0, false},
{"duration", `{"expiry":"1h"}`, time.Hour, false},
{"unparseable", `{"expiry":"nonsense"}`, 0, true},
{"zero duration", `{"expiry":"0s"}`, 0, true},
{"negative duration", `{"expiry":"-5h"}`, 0, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
got, err := delivery.ExportParseArchiveExpiry(tc.in)
if tc.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tc.want, got)
})
}
}
// seedDatabaseTargetDelivery seeds a pending delivery for a
// database target with the given config JSON and returns the
// in-memory delivery the target handler is invoked with.
func seedDatabaseTargetDelivery(
t *testing.T,
webhookDB *gorm.DB,
event database.Event,
config string,
) *database.Delivery {
t.Helper()
dlv := seedDelivery(
t, webhookDB, 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-db",
Type: database.TargetTypeDatabase,
Config: config,
},
}
d.ID = dlv.ID
return d
}
// TestDeliverDatabase_ArchiveFailureFailsDelivery verifies that
// an archive error (here: an unparseable expiry in the target
// config) fails the delivery loudly: the attempt is recorded as
// failed with the error and the delivery is marked failed, not
// delivered.
func TestDeliverDatabase_ArchiveFailureFailsDelivery(
t *testing.T,
) {
t.Parallel()
dataDir := t.TempDir()
e := delivery.NewTestEngineWithDB(
nil, database.NewTestWebhookDBManager(dataDir),
archiveTestLogger(),
&http.Client{Timeout: 5 * time.Second},
1,
)
webhookDB := testWebhookDB(t)
event := seedEvent(t, webhookDB, `{"archived":false}`)
d := seedDatabaseTargetDelivery(
t, webhookDB, event, `{"expiry":"nonsense"}`,
)
e.ExportDeliverDatabase(webhookDB, d)
var updated database.Delivery
require.NoError(t, webhookDB.First(
&updated, "id = ?", d.ID,
).Error)
assert.Equal(t,
database.DeliveryStatusFailed, updated.Status,
"archive failure must mark the delivery failed",
)
var results []database.DeliveryResult
require.NoError(t, webhookDB.Where(
"delivery_id = ?", d.ID,
).Find(&results).Error)
require.Len(t, results, 1)
assert.False(t,
results[0].Success,
"the attempt must be recorded as failed",
)
assert.Contains(t,
results[0].Error, "nonsense",
"the archive error must be recorded on the attempt",
)
assert.NoFileExists(t,
filepath.Join(
dataDir,
fmt.Sprintf("archive-%s.db", event.WebhookID),
),
"no archive file should exist for a failed config",
)
}
func TestValidateArchiveExpiry(t *testing.T) {
t.Parallel()
valid := []string{"", "never", "1h", "720h", "30m"}
for _, in := range valid {
require.NoError(t,
delivery.ValidateArchiveExpiry(in),
"expiry %q should be accepted", in,
)
}
invalid := []string{"nonsense", "7d", "-5h", "0s", "0"}
for _, in := range invalid {
require.Error(t,
delivery.ValidateArchiveExpiry(in),
"expiry %q should be rejected", in,
)
}
}

View File

@@ -495,5 +495,5 @@ func applyRequestHeaders(
func executeHTTPRequest( func executeHTTPRequest(
client *http.Client, req *http.Request, client *http.Client, req *http.Request,
) (*http.Response, error) { ) (*http.Response, error) {
return client.Do(req) //#nosec G704 -- URL validated by parseHTTPConfig/parseSlackConfig and SSRF-safe transport return client.Do(req) //#nosec G704 -- validated URL, SSRF-safe transport
} }

View File

@@ -19,7 +19,7 @@ func (h *Handlers) HandleLoginPage() http.HandlerFunc {
// Render login page // Render login page
data := map[string]any{ data := map[string]any{
"Error": "", tmplKeyError: "",
} }
h.renderTemplate(w, r, "login.html", data) h.renderTemplate(w, r, "login.html", data)
@@ -86,7 +86,7 @@ func (h *Handlers) renderLoginError(
status int, status int,
) { ) {
data := map[string]any{ data := map[string]any{
"Error": msg, tmplKeyError: msg,
} }
w.WriteHeader(status) w.WriteHeader(status)

View File

@@ -13,12 +13,26 @@ func (s *Handlers) RenderTemplateForTest(
s.renderTemplate(w, r, pageTemplate, data) s.renderTemplate(w, r, pageTemplate, data)
} }
// BuildSlackTargetConfigForTest exposes buildSlackTargetConfig // BuildSlackTargetConfigForTest exposes buildURLTargetConfig
// for use in the handlers_test package. // with the Slack target parameters for use in the
// handlers_test package.
func (s *Handlers) BuildSlackTargetConfigForTest( func (s *Handlers) BuildSlackTargetConfigForTest(
w http.ResponseWriter, w http.ResponseWriter,
r *http.Request, r *http.Request,
targetURL string, targetURL string,
) (string, error) { ) (string, error) {
return s.buildSlackTargetConfig(w, r, targetURL) return s.buildURLTargetConfig(
w, r, targetURL, "webhookUrl",
"Webhook URL is required for Slack targets",
)
}
// BuildDatabaseTargetConfigForTest exposes
// buildDatabaseTargetConfig for use in the handlers_test
// package.
func (s *Handlers) BuildDatabaseTargetConfigForTest(
w http.ResponseWriter,
expiry string,
) (string, error) {
return s.buildDatabaseTargetConfig(w, expiry)
} }

View File

@@ -30,6 +30,11 @@ const (
defaultRetentionDays = 30 defaultRetentionDays = 30
// paginationPerPage is the number of items per page. // paginationPerPage is the number of items per page.
paginationPerPage = 25 paginationPerPage = 25
// tmplKeyError is the template data key for an error message.
tmplKeyError = "Error"
// tmplKeyWebhook is the template data key for a webhook.
tmplKeyWebhook = "Webhook"
) )
// errInvalidPassword is returned when a password does not match. // errInvalidPassword is returned when a password does not match.

View File

@@ -186,3 +186,57 @@ func TestRenderTemplate(t *testing.T) {
t, http.StatusInternalServerError, w.Code, t, http.StatusInternalServerError, w.Code,
) )
} }
func TestBuildDatabaseTargetConfig_Valid(t *testing.T) {
t.Parallel()
var h *handlers.Handlers
app := newTestApp(t, &h)
app.RequireStart()
t.Cleanup(app.RequireStop)
// Empty expiry: the keep-forever default, empty config.
w := httptest.NewRecorder()
cfg, err := h.BuildDatabaseTargetConfigForTest(w, "")
require.NoError(t, err)
assert.Empty(t, cfg)
// Explicit never is stored as config.
w = httptest.NewRecorder()
cfg, err = h.BuildDatabaseTargetConfigForTest(w, "never")
require.NoError(t, err)
assert.JSONEq(t, `{"expiry":"never"}`, cfg)
// A positive duration is stored as config.
w = httptest.NewRecorder()
cfg, err = h.BuildDatabaseTargetConfigForTest(w, "720h")
require.NoError(t, err)
assert.JSONEq(t, `{"expiry":"720h"}`, cfg)
}
func TestBuildDatabaseTargetConfig_RejectsBadExpiry(
t *testing.T,
) {
t.Parallel()
var h *handlers.Handlers
app := newTestApp(t, &h)
app.RequireStart()
t.Cleanup(app.RequireStop)
for _, bad := range []string{"nonsense", "7d", "-5h"} {
w := httptest.NewRecorder()
cfg, err := h.BuildDatabaseTargetConfigForTest(w, bad)
require.Error(t, err, "expiry %q", bad)
assert.Empty(t, cfg)
assert.Equal(
t, http.StatusBadRequest, w.Code,
"expiry %q should be rejected with 400", bad,
)
}
}

View File

@@ -5,6 +5,7 @@ import (
"errors" "errors"
"net/http" "net/http"
"strconv" "strconv"
"strings"
"github.com/go-chi/chi" "github.com/go-chi/chi"
"github.com/google/uuid" "github.com/google/uuid"
@@ -106,7 +107,7 @@ func (h *Handlers) buildWebhookListItems(
func (h *Handlers) HandleSourceCreate() http.HandlerFunc { func (h *Handlers) HandleSourceCreate() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
data := map[string]any{ data := map[string]any{
"Error": "", tmplKeyError: "",
} }
h.renderTemplate(w, r, "sources_new.html", data) h.renderTemplate(w, r, "sources_new.html", data)
@@ -145,7 +146,7 @@ func (h *Handlers) HandleSourceCreateSubmit() http.HandlerFunc {
if name == "" { if name == "" {
data := map[string]any{ data := map[string]any{
"Error": "Name is required", tmplKeyError: "Name is required",
} }
w.WriteHeader(http.StatusBadRequest) w.WriteHeader(http.StatusBadRequest)
@@ -315,7 +316,7 @@ func (h *Handlers) renderSourceDetail(
} }
data := map[string]any{ data := map[string]any{
"Webhook": webhook, tmplKeyWebhook: webhook,
"Entrypoints": entrypoints, "Entrypoints": entrypoints,
"Targets": targets, "Targets": targets,
"Events": events, "Events": events,
@@ -351,8 +352,8 @@ func (h *Handlers) HandleSourceEdit() http.HandlerFunc {
} }
data := map[string]any{ data := map[string]any{
"Webhook": webhook, tmplKeyWebhook: webhook,
"Error": "", tmplKeyError: "",
} }
h.renderTemplate(w, r, "source_edit.html", data) h.renderTemplate(w, r, "source_edit.html", data)
@@ -415,8 +416,8 @@ func (h *Handlers) applyWebhookEdit(
name := r.FormValue("name") name := r.FormValue("name")
if name == "" { if name == "" {
data := map[string]any{ data := map[string]any{
"Webhook": *webhook, tmplKeyWebhook: *webhook,
"Error": "Name is required", tmplKeyError: "Name is required",
} }
w.WriteHeader(http.StatusBadRequest) w.WriteHeader(http.StatusBadRequest)
@@ -589,7 +590,7 @@ func (h *Handlers) HandleSourceLogs() http.HandlerFunc {
} }
data := map[string]any{ data := map[string]any{
"Webhook": webhook, tmplKeyWebhook: webhook,
"Events": evts, "Events": evts,
"Page": page, "Page": page,
"TotalPages": totalPages, "TotalPages": totalPages,
@@ -815,6 +816,7 @@ func (h *Handlers) processTargetCreate(
targetType := database.TargetType(r.FormValue("type")) targetType := database.TargetType(r.FormValue("type"))
targetURL := r.FormValue("url") targetURL := r.FormValue("url")
maxRetriesStr := r.FormValue("max_retries") maxRetriesStr := r.FormValue("max_retries")
expiry := r.FormValue("expiry")
if name == "" { if name == "" {
http.Error( http.Error(
@@ -834,7 +836,7 @@ func (h *Handlers) processTargetCreate(
} }
configJSON, err := h.buildTargetConfig( configJSON, err := h.buildTargetConfig(
w, r, targetType, targetURL, w, r, targetType, targetURL, expiry,
) )
if err != nil { if err != nil {
return return
@@ -892,18 +894,28 @@ func parseNonNegativeInt(s string) int {
} }
// buildTargetConfig builds the JSON config string for a target. // buildTargetConfig builds the JSON config string for a target.
// The expiry form value is read by the caller (which bounds the
// request body) and applies to database targets only.
func (h *Handlers) buildTargetConfig( func (h *Handlers) buildTargetConfig(
w http.ResponseWriter, w http.ResponseWriter,
r *http.Request, r *http.Request,
targetType database.TargetType, targetType database.TargetType,
targetURL string, targetURL, expiry string,
) (string, error) { ) (string, error) {
switch targetType { switch targetType {
case database.TargetTypeHTTP: case database.TargetTypeHTTP:
return h.buildHTTPTargetConfig(w, r, targetURL) return h.buildURLTargetConfig(
w, r, targetURL, "url",
"URL is required for HTTP targets",
)
case database.TargetTypeSlack: case database.TargetTypeSlack:
return h.buildSlackTargetConfig(w, r, targetURL) return h.buildURLTargetConfig(
case database.TargetTypeDatabase, database.TargetTypeLog: w, r, targetURL, "webhookUrl",
"Webhook URL is required for Slack targets",
)
case database.TargetTypeDatabase:
return h.buildDatabaseTargetConfig(w, expiry)
case database.TargetTypeLog:
return "", nil return "", nil
default: default:
http.Error( http.Error(
@@ -915,16 +927,18 @@ func (h *Handlers) buildTargetConfig(
} }
} }
// buildHTTPTargetConfig builds config JSON for an HTTP target. // buildURLTargetConfig builds config JSON for a target whose
func (h *Handlers) buildHTTPTargetConfig( // configuration is a single SSRF-validated URL stored under
// configKey. missingMsg is the error shown when no URL is given.
func (h *Handlers) buildURLTargetConfig(
w http.ResponseWriter, w http.ResponseWriter,
r *http.Request, r *http.Request,
targetURL string, targetURL, configKey, missingMsg string,
) (string, error) { ) (string, error) {
if targetURL == "" { if targetURL == "" {
http.Error( http.Error(
w, w,
"URL is required for HTTP targets", missingMsg,
http.StatusBadRequest, http.StatusBadRequest,
) )
@@ -949,7 +963,7 @@ func (h *Handlers) buildHTTPTargetConfig(
return "", err return "", err
} }
cfg := map[string]any{"url": targetURL} cfg := map[string]any{configKey: targetURL}
configBytes, err := json.Marshal(cfg) configBytes, err := json.Marshal(cfg)
if err != nil { if err != nil {
@@ -964,41 +978,33 @@ func (h *Handlers) buildHTTPTargetConfig(
return string(configBytes), nil return string(configBytes), nil
} }
// buildSlackTargetConfig builds config JSON for a Slack target. // buildDatabaseTargetConfig builds config JSON for a database
func (h *Handlers) buildSlackTargetConfig( // (archive) target. The optional expiry (a form value read by
// the caller, which bounds the request body) is validated here,
// at creation time, so an unparseable value is rejected with a
// 400 instead of failing every subsequent delivery. An empty
// expiry yields an empty config (the keep-forever default).
func (h *Handlers) buildDatabaseTargetConfig(
w http.ResponseWriter, w http.ResponseWriter,
r *http.Request, expiry string,
targetURL string,
) (string, error) { ) (string, error) {
if targetURL == "" { expiry = strings.TrimSpace(expiry)
http.Error( if expiry == "" {
w, return "", nil
"Webhook URL is required for Slack targets",
http.StatusBadRequest,
)
return "", errMissingURL
} }
err := delivery.ValidateTargetURL( err := delivery.ValidateArchiveExpiry(expiry)
r.Context(), targetURL,
)
if err != nil { if err != nil {
h.log.Warn(
"target URL blocked by SSRF protection",
"url", targetURL,
"error", err,
)
http.Error( http.Error(
w, w,
"Invalid target URL: "+err.Error(), "Invalid archive expiry: "+err.Error(),
http.StatusBadRequest, http.StatusBadRequest,
) )
return "", err return "", err
} }
cfg := map[string]any{"webhookUrl": targetURL} cfg := map[string]any{"expiry": expiry}
configBytes, err := json.Marshal(cfg) configBytes, err := json.Marshal(cfg)
if err != nil { if err != nil {

View File

@@ -484,8 +484,13 @@ func metricsAuthMiddleware(
return middleware.NewForTest(log, cfg, sessManager) return middleware.NewForTest(log, cfg, sessManager)
} }
func TestMetricsAuth_ValidCredentials(t *testing.T) { // runMetricsAuthRequest sends a GET /metrics request with the
t.Parallel() // given basic-auth password through MetricsAuth and reports
// whether the wrapped handler ran plus the recorded response.
func runMetricsAuthRequest(
t *testing.T, password string,
) (bool, *httptest.ResponseRecorder) {
t.Helper()
m := metricsAuthMiddleware(t) m := metricsAuthMiddleware(t)
@@ -503,12 +508,20 @@ func TestMetricsAuth_ValidCredentials(t *testing.T) {
context.Background(), context.Background(),
http.MethodGet, "/metrics", nil, http.MethodGet, "/metrics", nil,
) )
req.SetBasicAuth("admin", "secret") req.SetBasicAuth("admin", password)
w := httptest.NewRecorder() w := httptest.NewRecorder()
handler.ServeHTTP(w, req) handler.ServeHTTP(w, req)
return called, w
}
func TestMetricsAuth_ValidCredentials(t *testing.T) {
t.Parallel()
called, w := runMetricsAuthRequest(t, "secret")
assert.True( assert.True(
t, called, t, called,
"handler should be called with valid basic auth", "handler should be called with valid basic auth",
@@ -519,27 +532,7 @@ func TestMetricsAuth_ValidCredentials(t *testing.T) {
func TestMetricsAuth_InvalidCredentials(t *testing.T) { func TestMetricsAuth_InvalidCredentials(t *testing.T) {
t.Parallel() t.Parallel()
m := metricsAuthMiddleware(t) called, w := runMetricsAuthRequest(t, "wrong-password")
var called bool
handler := m.MetricsAuth()(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
called = true
w.WriteHeader(http.StatusOK)
},
))
req := httptest.NewRequestWithContext(
context.Background(),
http.MethodGet, "/metrics", nil,
)
req.SetBasicAuth("admin", "wrong-password")
w := httptest.NewRecorder()
handler.ServeHTTP(w, req)
assert.False( assert.False(
t, called, t, called,

View File

@@ -14,11 +14,6 @@ const (
// loginRateInterval is the time window for the rate limit. // loginRateInterval is the time window for the rate limit.
loginRateInterval = 1 * time.Minute loginRateInterval = 1 * time.Minute
// receiverRateInterval is the time window for the webhook
// receiver rate limit. The configured limit is expressed in
// requests per minute.
receiverRateInterval = 1 * time.Minute
) )
// LoginRateLimit returns middleware that enforces per-IP rate // LoginRateLimit returns middleware that enforces per-IP rate
@@ -67,37 +62,3 @@ func (m *Middleware) LoginRateLimit() func(http.Handler) http.Handler {
}) })
} }
} }
// ReceiverRateLimit returns middleware that rate-limits the
// public webhook receiver endpoint per client IP per request
// path (the path contains the entrypoint UUID, so each sender
// is limited per entrypoint without affecting other senders or
// other entrypoints). The limit is Config.ReceiverRateLimit
// requests per minute. Requests over the limit receive a 429;
// httprate adds the Retry-After header (RFC 6585). IP
// extraction honours X-Forwarded-For, X-Real-IP, and
// True-Client-IP headers for reverse-proxy setups.
func (m *Middleware) ReceiverRateLimit() func(http.Handler) http.Handler {
return httprate.Limit(
m.params.Config.ReceiverRateLimit,
receiverRateInterval,
httprate.WithKeyFuncs(
httprate.KeyByRealIP,
httprate.KeyByEndpoint,
),
httprate.WithLimitHandler(http.HandlerFunc(
func(w http.ResponseWriter, r *http.Request) {
m.log.Warn(
"webhook receiver rate limit exceeded",
"path", r.URL.Path,
)
http.Error(
w,
"Too many requests. "+
"Please slow down.",
http.StatusTooManyRequests,
)
},
)),
)
}

View File

@@ -2,10 +2,8 @@ package middleware_test
import ( import (
"context" "context"
"log/slog"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os"
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -147,94 +145,3 @@ func TestLoginRateLimit_IndependentPerIP(t *testing.T) {
"different IP should not be affected", "different IP should not be affected",
) )
} }
// receiverLimitedHandler builds a ReceiverRateLimit-wrapped
// handler with the given per-minute limit.
func receiverLimitedHandler(
t *testing.T, limit int,
) http.Handler {
t.Helper()
log := slog.New(slog.NewTextHandler(
os.Stderr,
&slog.HandlerOptions{Level: slog.LevelDebug},
))
m := middleware.NewForTest(
log,
&config.Config{ReceiverRateLimit: limit},
nil,
)
return m.ReceiverRateLimit()(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
},
))
}
// receiverPost sends one POST to the handler from the given IP
// and path and returns the recorder.
func receiverPost(
handler http.Handler, ip, path string,
) *httptest.ResponseRecorder {
req := httptest.NewRequestWithContext(
context.Background(),
http.MethodPost, path, nil,
)
req.RemoteAddr = ip
w := httptest.NewRecorder()
handler.ServeHTTP(w, req)
return w
}
func TestReceiverRateLimit_LimitsPerIPAndPath(t *testing.T) {
t.Parallel()
const limit = 3
handler := receiverLimitedHandler(t, limit)
// The first limit requests from one IP to one entrypoint
// pass.
for i := range limit {
w := receiverPost(
handler, "9.9.9.9:1234", "/webhook/uuid-a",
)
assert.Equal(
t, http.StatusOK, w.Code,
"request %d should pass", i,
)
}
// The next request over the limit is rejected with a 429
// carrying a Retry-After header.
w := receiverPost(
handler, "9.9.9.9:1234", "/webhook/uuid-a",
)
assert.Equal(t, http.StatusTooManyRequests, w.Code)
assert.NotEmpty(
t, w.Header().Get("Retry-After"),
"429 must carry a Retry-After header",
)
// The same IP is not limited on a different entrypoint.
w = receiverPost(
handler, "9.9.9.9:1234", "/webhook/uuid-b",
)
assert.Equal(
t, http.StatusOK, w.Code,
"a different entrypoint must not be affected",
)
// A different IP is not limited on the same entrypoint.
w = receiverPost(
handler, "8.8.8.8:1234", "/webhook/uuid-a",
)
assert.Equal(
t, http.StatusOK, w.Code,
"a different client IP must not be affected",
)
}

View File

@@ -159,7 +159,7 @@ func (s *Server) setupSourceRoutes() {
} }
func (s *Server) setupWebhookRoutes() { func (s *Server) setupWebhookRoutes() {
s.router.With(s.mw.ReceiverRateLimit()).HandleFunc( s.router.HandleFunc(
"/webhook/{uuid}", "/webhook/{uuid}",
s.h.HandleWebhook(), s.h.HandleWebhook(),
) )

View File

@@ -173,8 +173,18 @@ func TestSetUser_SetsAllFields(t *testing.T) {
) )
} }
func TestGetUserID(t *testing.T) { // testSessionGetter exercises a session string getter before and
t.Parallel() // after SetUser: it must report false with an empty value on a
// fresh session, then true with the expected value once
// SetUser(sess, "user-xyz", "bob") has run.
func testSessionGetter(
t *testing.T,
get func(
*session.Session, *sessions.Session,
) (string, bool),
expected string,
) {
t.Helper()
s := testSession(t) s := testSession(t)
@@ -185,44 +195,46 @@ func TestGetUserID(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
// Before setting user // Before setting user
userID, ok := s.GetUserID(sess) val, ok := get(s, sess)
assert.False( assert.False(
t, ok, "should return false when no user ID is set", t, ok, "should return false before SetUser",
) )
assert.Empty(t, userID) assert.Empty(t, val)
// After setting user // After setting user
s.SetUser(sess, "user-xyz", "bob") s.SetUser(sess, "user-xyz", "bob")
userID, ok = s.GetUserID(sess) val, ok = get(s, sess)
assert.True(t, ok) assert.True(t, ok)
assert.Equal(t, "user-xyz", userID) assert.Equal(t, expected, val)
}
func TestGetUserID(t *testing.T) {
t.Parallel()
testSessionGetter(
t,
func(
s *session.Session, sess *sessions.Session,
) (string, bool) {
return s.GetUserID(sess)
},
"user-xyz",
)
} }
func TestGetUsername(t *testing.T) { func TestGetUsername(t *testing.T) {
t.Parallel() t.Parallel()
s := testSession(t) testSessionGetter(
t,
req := httptest.NewRequestWithContext( func(
context.Background(), http.MethodGet, "/", nil) s *session.Session, sess *sessions.Session,
) (string, bool) {
sess, err := s.Get(req) return s.GetUsername(sess)
require.NoError(t, err) },
"bob",
// Before setting user
username, ok := s.GetUsername(sess)
assert.False(
t, ok, "should return false when no username is set",
) )
assert.Empty(t, username)
// After setting user
s.SetUser(sess, "user-xyz", "bob")
username, ok = s.GetUsername(sess)
assert.True(t, ok)
assert.Equal(t, "bob", username)
} }
// --- IsAuthenticated Tests --- // --- IsAuthenticated Tests ---

View File

@@ -12,7 +12,12 @@ import (
// middleware and handler tests to use real session functionality. The key // middleware and handler tests to use real session functionality. The key
// parameter is the raw 32-byte authentication key used for session encryption // parameter is the raw 32-byte authentication key used for session encryption
// and CSRF cookie signing. // and CSRF cookie signing.
func NewForTest(store *sessions.CookieStore, cfg *config.Config, log *slog.Logger, key []byte) *Session { func NewForTest(
store *sessions.CookieStore,
cfg *config.Config,
log *slog.Logger,
key []byte,
) *Session {
return &Session{ return &Session{
store: store, store: store,
key: key, key: key,

View File

@@ -10,11 +10,11 @@ set -eu
ROOT="$(cd "$(dirname "$0")/.." && pwd -P)" ROOT="$(cd "$(dirname "$0")/.." && pwd -P)"
# Pinned versions, 2026-07-07. Never "latest"; exact versions only. # Pinned versions, 2026-08-07. Never "latest"; exact versions only.
GOLANGCI_LINT_VERSION="2.11.3" GOLANGCI_LINT_VERSION="2.12.2"
# sha256 of golangci-lint-2.11.3-linux-<arch>.tar.gz release archives # sha256 of golangci-lint-2.12.2-linux-<arch>.tar.gz release archives
GOLANGCI_LINT_SHA256_AMD64="87bb8cddbcc825d5778b64e8a91b46c0526b247f4e2f2904dea74ec7450475d1" GOLANGCI_LINT_SHA256_AMD64="8df580d2670fed8fa984aac0507099af8df275e665215f5c7a2ae3943893a553"
GOLANGCI_LINT_SHA256_ARM64="ee3d95f301359e7d578e6d99c8ad5aeadbabc5a13009a30b2b0df11c8058afe9" GOLANGCI_LINT_SHA256_ARM64="44cd40a8c76c86755375adfeea52cfd3533cb43d7bd647771e0ae065e166df3a"
PKGMGR="" PKGMGR=""
SUDO="" SUDO=""

View File

@@ -113,6 +113,10 @@
<input type="url" name="url" placeholder="https://hooks.slack.com/services/..." :disabled="targetType !== 'slack'" class="input text-sm"> <input type="url" name="url" placeholder="https://hooks.slack.com/services/..." :disabled="targetType !== 'slack'" class="input text-sm">
<p class="text-xs text-gray-500 mt-1">Slack or Mattermost incoming webhook URL. Payloads are pretty-printed in code blocks.</p> <p class="text-xs text-gray-500 mt-1">Slack or Mattermost incoming webhook URL. Payloads are pretty-printed in code blocks.</p>
</div> </div>
<div x-show="targetType === 'database'">
<input type="text" name="expiry" placeholder="never" :disabled="targetType !== 'database'" class="input text-sm">
<p class="text-xs text-gray-500 mt-1">Archive expiry: "never" (default) keeps rows forever, or a duration like "720h" prunes older rows.</p>
</div>
<button type="submit" class="btn-primary text-sm">Add Target</button> <button type="submit" class="btn-primary text-sm">Add Target</button>
</form> </form>
</div> </div>