feat: implement core webhook engine, delivery system, and management UI (Phase 2)
All checks were successful
check / check (push) Successful in 1m49s
All checks were successful
check / check (push) Successful in 1m49s
- Webhook reception handler: look up entrypoint by UUID, verify active,
capture full HTTP request (method, headers, body, content-type), create
Event record, queue Delivery records for each active Target, return 200 OK.
Handles edge cases: unknown UUID → 404, inactive → 410, oversized → 413.
- Delivery engine (internal/delivery): fx-managed background goroutine that
polls for pending/retrying deliveries and dispatches to target type handlers.
Graceful shutdown via context cancellation.
- Target type implementations:
- HTTP: fire-and-forget POST with original headers forwarding
- Retry: exponential backoff (1s, 2s, 4s...) up to max_retries
- Database: immediate success (event already stored)
- Log: slog output with event details
- Webhook management pages with Tailwind CSS + Alpine.js:
- List (/sources): webhooks with entrypoint/target/event counts
- Create (/sources/new): form with auto-created default entrypoint
- Detail (/source/{id}): config, entrypoints, targets, recent events
- Edit (/source/{id}/edit): name, description, retention_days
- Delete (/source/{id}/delete): soft-delete with child records
- Add Entrypoint (/source/{id}/entrypoints): inline form
- Add Target (/source/{id}/targets): type-aware form
- Event Log (/source/{id}/logs): paginated with delivery status
- Updated README: marked completed items, updated naming conventions
table, added delivery engine to package layout and DI docs, updated
column names to reflect entity rename.
- Rebuilt Tailwind CSS for new template classes.
Part of: #15
This commit is contained in:
@@ -1,69 +1,520 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/go-chi/chi"
|
||||
"github.com/google/uuid"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
)
|
||||
|
||||
// HandleSourceList shows a list of user's webhooks
|
||||
// WebhookListItem holds data for the webhook list view.
|
||||
type WebhookListItem struct {
|
||||
database.Webhook
|
||||
EntrypointCount int64
|
||||
TargetCount int64
|
||||
EventCount int64
|
||||
}
|
||||
|
||||
// HandleSourceList shows a list of user's webhooks.
|
||||
func (h *Handlers) HandleSourceList() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook list page
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
var webhooks []database.Webhook
|
||||
if err := h.db.DB().Where("user_id = ?", userID).Order("created_at DESC").Find(&webhooks).Error; err != nil {
|
||||
h.log.Error("failed to list webhooks", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Build list items with counts
|
||||
items := make([]WebhookListItem, len(webhooks))
|
||||
for i := range webhooks {
|
||||
items[i].Webhook = webhooks[i]
|
||||
h.db.DB().Model(&database.Entrypoint{}).Where("webhook_id = ?", webhooks[i].ID).Count(&items[i].EntrypointCount)
|
||||
h.db.DB().Model(&database.Target{}).Where("webhook_id = ?", webhooks[i].ID).Count(&items[i].TargetCount)
|
||||
h.db.DB().Model(&database.Event{}).Where("webhook_id = ?", webhooks[i].ID).Count(&items[i].EventCount)
|
||||
}
|
||||
|
||||
data := map[string]interface{}{
|
||||
"Webhooks": items,
|
||||
}
|
||||
h.renderTemplate(w, r, "sources_list.html", data)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceCreate shows the form to create a new webhook
|
||||
// HandleSourceCreate shows the form to create a new webhook.
|
||||
func (h *Handlers) HandleSourceCreate() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook creation form
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
data := map[string]interface{}{
|
||||
"Error": "",
|
||||
}
|
||||
h.renderTemplate(w, r, "sources_new.html", data)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceCreateSubmit handles the webhook creation form submission
|
||||
// HandleSourceCreateSubmit handles the webhook creation form submission.
|
||||
func (h *Handlers) HandleSourceCreateSubmit() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook creation logic
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
if err := r.ParseForm(); err != nil {
|
||||
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
name := r.FormValue("name")
|
||||
description := r.FormValue("description")
|
||||
retentionStr := r.FormValue("retention_days")
|
||||
|
||||
if name == "" {
|
||||
data := map[string]interface{}{
|
||||
"Error": "Name is required",
|
||||
}
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
h.renderTemplate(w, r, "sources_new.html", data)
|
||||
return
|
||||
}
|
||||
|
||||
retentionDays := 30
|
||||
if retentionStr != "" {
|
||||
if v, err := strconv.Atoi(retentionStr); err == nil && v > 0 {
|
||||
retentionDays = v
|
||||
}
|
||||
}
|
||||
|
||||
tx := h.db.DB().Begin()
|
||||
if tx.Error != nil {
|
||||
h.log.Error("failed to begin transaction", "error", tx.Error)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
webhook := &database.Webhook{
|
||||
UserID: userID,
|
||||
Name: name,
|
||||
Description: description,
|
||||
RetentionDays: retentionDays,
|
||||
}
|
||||
|
||||
if err := tx.Create(webhook).Error; err != nil {
|
||||
tx.Rollback()
|
||||
h.log.Error("failed to create webhook", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Auto-create one entrypoint
|
||||
entrypoint := &database.Entrypoint{
|
||||
WebhookID: webhook.ID,
|
||||
Path: uuid.New().String(),
|
||||
Description: "Default entrypoint",
|
||||
Active: true,
|
||||
}
|
||||
|
||||
if err := tx.Create(entrypoint).Error; err != nil {
|
||||
tx.Rollback()
|
||||
h.log.Error("failed to create entrypoint", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
if err := tx.Commit().Error; err != nil {
|
||||
h.log.Error("failed to commit transaction", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Info("webhook created",
|
||||
"webhook_id", webhook.ID,
|
||||
"name", name,
|
||||
"user_id", userID,
|
||||
)
|
||||
|
||||
http.Redirect(w, r, "/source/"+webhook.ID, http.StatusSeeOther)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceDetail shows details for a specific webhook
|
||||
// HandleSourceDetail shows details for a specific webhook.
|
||||
func (h *Handlers) HandleSourceDetail() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook detail page
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
var entrypoints []database.Entrypoint
|
||||
h.db.DB().Where("webhook_id = ?", webhook.ID).Find(&entrypoints)
|
||||
|
||||
var targets []database.Target
|
||||
h.db.DB().Where("webhook_id = ?", webhook.ID).Find(&targets)
|
||||
|
||||
// Recent events with delivery info
|
||||
var events []database.Event
|
||||
h.db.DB().Where("webhook_id = ?", webhook.ID).Order("created_at DESC").Limit(20).Find(&events)
|
||||
|
||||
// Build host URL for display
|
||||
host := r.Host
|
||||
scheme := "https"
|
||||
if r.TLS == nil {
|
||||
scheme = "http"
|
||||
}
|
||||
// Check X-Forwarded headers
|
||||
if fwdProto := r.Header.Get("X-Forwarded-Proto"); fwdProto != "" {
|
||||
scheme = fwdProto
|
||||
}
|
||||
|
||||
data := map[string]interface{}{
|
||||
"Webhook": webhook,
|
||||
"Entrypoints": entrypoints,
|
||||
"Targets": targets,
|
||||
"Events": events,
|
||||
"BaseURL": scheme + "://" + host,
|
||||
}
|
||||
h.renderTemplate(w, r, "source_detail.html", data)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceEdit shows the form to edit a webhook
|
||||
// HandleSourceEdit shows the form to edit a webhook.
|
||||
func (h *Handlers) HandleSourceEdit() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook edit form
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
data := map[string]interface{}{
|
||||
"Webhook": webhook,
|
||||
"Error": "",
|
||||
}
|
||||
h.renderTemplate(w, r, "source_edit.html", data)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceEditSubmit handles the webhook edit form submission
|
||||
// HandleSourceEditSubmit handles the webhook edit form submission.
|
||||
func (h *Handlers) HandleSourceEditSubmit() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook update logic
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
if err := r.ParseForm(); err != nil {
|
||||
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
name := r.FormValue("name")
|
||||
if name == "" {
|
||||
data := map[string]interface{}{
|
||||
"Webhook": webhook,
|
||||
"Error": "Name is required",
|
||||
}
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
h.renderTemplate(w, r, "source_edit.html", data)
|
||||
return
|
||||
}
|
||||
|
||||
webhook.Name = name
|
||||
webhook.Description = r.FormValue("description")
|
||||
if retStr := r.FormValue("retention_days"); retStr != "" {
|
||||
if v, err := strconv.Atoi(retStr); err == nil && v > 0 {
|
||||
webhook.RetentionDays = v
|
||||
}
|
||||
}
|
||||
|
||||
if err := h.db.DB().Save(&webhook).Error; err != nil {
|
||||
h.log.Error("failed to update webhook", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
http.Redirect(w, r, "/source/"+webhook.ID, http.StatusSeeOther)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceDelete handles webhook deletion
|
||||
// HandleSourceDelete handles webhook deletion (soft delete).
|
||||
func (h *Handlers) HandleSourceDelete() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook deletion logic
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
tx := h.db.DB().Begin()
|
||||
if tx.Error != nil {
|
||||
h.log.Error("failed to begin transaction", "error", tx.Error)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Soft-delete child records
|
||||
tx.Where("webhook_id = ?", webhook.ID).Delete(&database.Entrypoint{})
|
||||
tx.Where("webhook_id = ?", webhook.ID).Delete(&database.Target{})
|
||||
tx.Where("webhook_id = ?", webhook.ID).Delete(&database.Event{})
|
||||
tx.Delete(&webhook)
|
||||
|
||||
if err := tx.Commit().Error; err != nil {
|
||||
h.log.Error("failed to commit deletion", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Info("webhook deleted", "webhook_id", webhook.ID, "user_id", userID)
|
||||
http.Redirect(w, r, "/sources", http.StatusSeeOther)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSourceLogs shows the request/response logs for a webhook
|
||||
// HandleSourceLogs shows the request/response logs for a webhook.
|
||||
func (h *Handlers) HandleSourceLogs() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: Implement webhook logs page
|
||||
http.Error(w, "Not implemented", http.StatusNotImplemented)
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
// Pagination
|
||||
page := 1
|
||||
if p := r.URL.Query().Get("page"); p != "" {
|
||||
if v, err := strconv.Atoi(p); err == nil && v > 0 {
|
||||
page = v
|
||||
}
|
||||
}
|
||||
perPage := 25
|
||||
offset := (page - 1) * perPage
|
||||
|
||||
var totalEvents int64
|
||||
h.db.DB().Model(&database.Event{}).Where("webhook_id = ?", webhook.ID).Count(&totalEvents)
|
||||
|
||||
var events []database.Event
|
||||
h.db.DB().Where("webhook_id = ?", webhook.ID).
|
||||
Order("created_at DESC").
|
||||
Offset(offset).
|
||||
Limit(perPage).
|
||||
Find(&events)
|
||||
|
||||
// Load deliveries for each event
|
||||
type EventWithDeliveries struct {
|
||||
database.Event
|
||||
Deliveries []database.Delivery
|
||||
}
|
||||
eventsWithDeliveries := make([]EventWithDeliveries, len(events))
|
||||
for i := range events {
|
||||
eventsWithDeliveries[i].Event = events[i]
|
||||
h.db.DB().Where("event_id = ?", events[i].ID).Preload("Target").Find(&eventsWithDeliveries[i].Deliveries)
|
||||
}
|
||||
|
||||
totalPages := int(totalEvents) / perPage
|
||||
if int(totalEvents)%perPage != 0 {
|
||||
totalPages++
|
||||
}
|
||||
|
||||
data := map[string]interface{}{
|
||||
"Webhook": webhook,
|
||||
"Events": eventsWithDeliveries,
|
||||
"Page": page,
|
||||
"TotalPages": totalPages,
|
||||
"TotalEvents": totalEvents,
|
||||
"HasPrev": page > 1,
|
||||
"HasNext": page < totalPages,
|
||||
"PrevPage": page - 1,
|
||||
"NextPage": page + 1,
|
||||
}
|
||||
h.renderTemplate(w, r, "source_logs.html", data)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleEntrypointCreate handles adding a new entrypoint to a webhook.
|
||||
func (h *Handlers) HandleEntrypointCreate() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
// Verify ownership
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
if err := r.ParseForm(); err != nil {
|
||||
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
description := r.FormValue("description")
|
||||
|
||||
entrypoint := &database.Entrypoint{
|
||||
WebhookID: webhook.ID,
|
||||
Path: uuid.New().String(),
|
||||
Description: description,
|
||||
Active: true,
|
||||
}
|
||||
|
||||
if err := h.db.DB().Create(entrypoint).Error; err != nil {
|
||||
h.log.Error("failed to create entrypoint", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
http.Redirect(w, r, "/source/"+webhook.ID, http.StatusSeeOther)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleTargetCreate handles adding a new target to a webhook.
|
||||
func (h *Handlers) HandleTargetCreate() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
userID, ok := h.getUserID(r)
|
||||
if !ok {
|
||||
http.Redirect(w, r, "/pages/login", http.StatusSeeOther)
|
||||
return
|
||||
}
|
||||
|
||||
sourceID := chi.URLParam(r, "sourceID")
|
||||
|
||||
var webhook database.Webhook
|
||||
if err := h.db.DB().Where("id = ? AND user_id = ?", sourceID, userID).First(&webhook).Error; err != nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
|
||||
if err := r.ParseForm(); err != nil {
|
||||
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
name := r.FormValue("name")
|
||||
targetType := database.TargetType(r.FormValue("type"))
|
||||
url := r.FormValue("url")
|
||||
maxRetriesStr := r.FormValue("max_retries")
|
||||
|
||||
if name == "" {
|
||||
http.Error(w, "Name is required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate target type
|
||||
switch targetType {
|
||||
case database.TargetTypeHTTP, database.TargetTypeRetry, database.TargetTypeDatabase, database.TargetTypeLog:
|
||||
// valid
|
||||
default:
|
||||
http.Error(w, "Invalid target type", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Build config JSON for HTTP-based targets
|
||||
var configJSON string
|
||||
if targetType == database.TargetTypeHTTP || targetType == database.TargetTypeRetry {
|
||||
if url == "" {
|
||||
http.Error(w, "URL is required for HTTP targets", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
cfg := map[string]interface{}{
|
||||
"url": url,
|
||||
}
|
||||
configBytes, err := json.Marshal(cfg)
|
||||
if err != nil {
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
configJSON = string(configBytes)
|
||||
}
|
||||
|
||||
maxRetries := 5
|
||||
if maxRetriesStr != "" {
|
||||
if v, err := strconv.Atoi(maxRetriesStr); err == nil && v > 0 {
|
||||
maxRetries = v
|
||||
}
|
||||
}
|
||||
|
||||
target := &database.Target{
|
||||
WebhookID: webhook.ID,
|
||||
Name: name,
|
||||
Type: targetType,
|
||||
Active: true,
|
||||
Config: configJSON,
|
||||
MaxRetries: maxRetries,
|
||||
}
|
||||
|
||||
if err := h.db.DB().Create(target).Error; err != nil {
|
||||
h.log.Error("failed to create target", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
http.Redirect(w, r, "/source/"+webhook.ID, http.StatusSeeOther)
|
||||
}
|
||||
}
|
||||
|
||||
// getUserID extracts the user ID from the session.
|
||||
func (h *Handlers) getUserID(r *http.Request) (string, bool) {
|
||||
sess, err := h.session.Get(r)
|
||||
if err != nil {
|
||||
return "", false
|
||||
}
|
||||
if !h.session.IsAuthenticated(sess) {
|
||||
return "", false
|
||||
}
|
||||
return h.session.GetUserID(sess)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user