Implement the database archiving target (closes #43) #84
@@ -342,33 +342,29 @@ func TestDeliverDatabase_ImmediateSuccess(
|
||||
t.Parallel()
|
||||
|
||||
db := testWebhookDB(t)
|
||||
e := testEngine(t, 1)
|
||||
|
||||
event := seedEvent(t, db, `{"db":"target"}`)
|
||||
|
||||
dlv := seedDelivery(
|
||||
t, db, event.ID, uuid.New().String(),
|
||||
database.DeliveryStatusPending,
|
||||
// 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,
|
||||
)
|
||||
|
||||
d := &database.Delivery{
|
||||
EventID: event.ID,
|
||||
TargetID: dlv.TargetID,
|
||||
Status: database.DeliveryStatusPending,
|
||||
Event: event,
|
||||
Target: database.Target{
|
||||
Name: "test-db",
|
||||
Type: database.TargetTypeDatabase,
|
||||
},
|
||||
}
|
||||
d.ID = dlv.ID
|
||||
event := seedEvent(t, db, `{"db":"target"}`)
|
||||
d := seedDatabaseTargetDelivery(t, db, event, "")
|
||||
|
||||
e.ExportDeliverDatabase(db, d)
|
||||
|
||||
var updated database.Delivery
|
||||
|
||||
require.NoError(t, db.First(
|
||||
&updated, "id = ?", dlv.ID,
|
||||
&updated, "id = ?", d.ID,
|
||||
).Error)
|
||||
|
||||
assert.Equal(t,
|
||||
@@ -379,7 +375,7 @@ func TestDeliverDatabase_ImmediateSuccess(
|
||||
var result database.DeliveryResult
|
||||
|
||||
require.NoError(t, db.Where(
|
||||
"delivery_id = ?", dlv.ID,
|
||||
"delivery_id = ?", d.ID,
|
||||
).First(&result).Error)
|
||||
|
||||
assert.True(t, result.Success)
|
||||
@@ -1158,7 +1154,19 @@ func TestProcessDelivery_RoutesToCorrectHandler(
|
||||
t.Parallel()
|
||||
|
||||
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 {
|
||||
name string
|
||||
|
||||
@@ -16,8 +16,9 @@ import (
|
||||
// already persisted in the per-webhook event DB by the time
|
||||
// delivery runs; the database target additionally writes a
|
||||
// durable long-term copy into archive-{webhookID}.db and then
|
||||
// records a single successful attempt. See archiveWriter for
|
||||
// the close/reopen, auto-recreate, and expiry semantics.
|
||||
// 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 {
|
||||
eng *Engine
|
||||
|
||||
@@ -27,10 +28,12 @@ type databaseTarget struct {
|
||||
|
||||
// Deliver implements Target. It archives the event, then
|
||||
// records one successful attempt and marks the delivery
|
||||
// delivered. Archiving errors are logged but do not fail the
|
||||
// delivery: the event is already durably stored in the
|
||||
// per-webhook event database, so the target stays
|
||||
// fire-and-forget.
|
||||
// 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(
|
||||
_ context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
@@ -46,6 +49,17 @@ func (t *databaseTarget) Deliver(
|
||||
"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(
|
||||
|
||||
@@ -37,6 +37,13 @@ var (
|
||||
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
|
||||
@@ -105,6 +112,35 @@ func parseArchiveExpiry(
|
||||
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
|
||||
|
||||
@@ -83,30 +83,14 @@ func TestDeliverDatabase_ArchivesEvent(t *testing.T) {
|
||||
|
||||
webhookDB := testWebhookDB(t)
|
||||
event := seedEvent(t, webhookDB, `{"archived":true}`)
|
||||
|
||||
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,
|
||||
},
|
||||
}
|
||||
d.ID = dlv.ID
|
||||
d := seedDatabaseTargetDelivery(t, webhookDB, event, "")
|
||||
|
||||
e.ExportDeliverDatabase(webhookDB, d)
|
||||
|
||||
var updated database.Delivery
|
||||
|
||||
require.NoError(t, webhookDB.First(
|
||||
&updated, "id = ?", dlv.ID,
|
||||
&updated, "id = ?", d.ID,
|
||||
).Error)
|
||||
assert.Equal(t,
|
||||
database.DeliveryStatusDelivered, updated.Status,
|
||||
@@ -202,9 +186,11 @@ func TestArchiveWriter_RecreatesAfterRemoval(
|
||||
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(), 20*time.Millisecond,
|
||||
path, archiveTestLogger(), 2*time.Second,
|
||||
)
|
||||
|
||||
require.NoError(t, w.Write(
|
||||
@@ -218,7 +204,7 @@ func TestArchiveWriter_ReopenDebounce(t *testing.T) {
|
||||
// initial open — no extra close/reopen.
|
||||
assert.Equal(t, 1, w.Reopens())
|
||||
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
time.Sleep(2100 * time.Millisecond)
|
||||
|
||||
require.NoError(t, w.Write(
|
||||
delivery.ExportArchivedEvent{EventID: "c"}, 0,
|
||||
@@ -290,3 +276,116 @@ func TestParseArchiveExpiry(t *testing.T) {
|
||||
)
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
// 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,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,3 +22,13 @@ func (s *Handlers) BuildSlackTargetConfigForTest(
|
||||
) (string, error) {
|
||||
return s.buildSlackTargetConfig(w, r, targetURL)
|
||||
}
|
||||
|
||||
// BuildDatabaseTargetConfigForTest exposes
|
||||
// buildDatabaseTargetConfig for use in the handlers_test
|
||||
// package.
|
||||
func (s *Handlers) BuildDatabaseTargetConfigForTest(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
) (string, error) {
|
||||
return s.buildDatabaseTargetConfig(w, r)
|
||||
}
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
@@ -186,3 +188,84 @@ func TestRenderTemplate(t *testing.T) {
|
||||
t, http.StatusInternalServerError, w.Code,
|
||||
)
|
||||
}
|
||||
|
||||
// databaseConfigRequest builds a POST request carrying the
|
||||
// given expiry as a form value, as the add-target form does.
|
||||
func databaseConfigRequest(expiry string) *http.Request {
|
||||
form := url.Values{}
|
||||
if expiry != "" {
|
||||
form.Set("expiry", expiry)
|
||||
}
|
||||
|
||||
req := httptest.NewRequestWithContext(
|
||||
context.Background(), http.MethodPost, "/",
|
||||
strings.NewReader(form.Encode()),
|
||||
)
|
||||
req.Header.Set(
|
||||
"Content-Type", "application/x-www-form-urlencoded",
|
||||
)
|
||||
|
||||
return req
|
||||
}
|
||||
|
||||
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, databaseConfigRequest(""),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, cfg)
|
||||
|
||||
// Explicit never is stored as config.
|
||||
w = httptest.NewRecorder()
|
||||
cfg, err = h.BuildDatabaseTargetConfigForTest(
|
||||
w, databaseConfigRequest("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, databaseConfigRequest("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, databaseConfigRequest(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,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/google/uuid"
|
||||
@@ -903,7 +904,9 @@ func (h *Handlers) buildTargetConfig(
|
||||
return h.buildHTTPTargetConfig(w, r, targetURL)
|
||||
case database.TargetTypeSlack:
|
||||
return h.buildSlackTargetConfig(w, r, targetURL)
|
||||
case database.TargetTypeDatabase, database.TargetTypeLog:
|
||||
case database.TargetTypeDatabase:
|
||||
return h.buildDatabaseTargetConfig(w, r)
|
||||
case database.TargetTypeLog:
|
||||
return "", nil
|
||||
default:
|
||||
http.Error(
|
||||
@@ -1013,6 +1016,47 @@ func (h *Handlers) buildSlackTargetConfig(
|
||||
return string(configBytes), nil
|
||||
}
|
||||
|
||||
// buildDatabaseTargetConfig builds config JSON for a database
|
||||
// (archive) target. The optional expiry form value 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,
|
||||
r *http.Request,
|
||||
) (string, error) {
|
||||
expiry := strings.TrimSpace(r.FormValue("expiry"))
|
||||
if expiry == "" {
|
||||
return "", nil
|
||||
}
|
||||
|
||||
err := delivery.ValidateArchiveExpiry(expiry)
|
||||
if err != nil {
|
||||
http.Error(
|
||||
w,
|
||||
"Invalid archive expiry: "+err.Error(),
|
||||
http.StatusBadRequest,
|
||||
)
|
||||
|
||||
return "", err
|
||||
}
|
||||
|
||||
cfg := map[string]any{"expiry": expiry}
|
||||
|
||||
configBytes, err := json.Marshal(cfg)
|
||||
if err != nil {
|
||||
http.Error(
|
||||
w, "Internal server error",
|
||||
http.StatusInternalServerError,
|
||||
)
|
||||
|
||||
return "", err
|
||||
}
|
||||
|
||||
return string(configBytes), nil
|
||||
}
|
||||
|
||||
// HandleEntrypointDelete handles deleting an entrypoint.
|
||||
func (h *Handlers) HandleEntrypointDelete() http.HandlerFunc {
|
||||
return h.deleteChildResource(
|
||||
|
||||
@@ -113,6 +113,10 @@
|
||||
<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>
|
||||
</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>
|
||||
</form>
|
||||
</div>
|
||||
|
||||
Reference in New Issue
Block a user