mirror of
https://github.com/luxfi/broker.git
synced 2026-09-24 00:25:18 +00:00
feat: webhook infrastructure — 16 BD events, HMAC delivery, routes, store
This commit is contained in:
@@ -22,6 +22,7 @@ import (
|
||||
"github.com/luxfi/broker/pkg/provider"
|
||||
"github.com/luxfi/broker/pkg/provider/envconfig"
|
||||
"github.com/luxfi/broker/pkg/router"
|
||||
"github.com/luxfi/broker/pkg/webhook"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -93,6 +94,10 @@ func main() {
|
||||
srv.Mount("/compliance", compliance.NewRouter(complianceStore, compliance.WithScamDB(scamDB), compliance.WithRegistry(registry)))
|
||||
log.Info().Msg("Compliance routes mounted at /compliance")
|
||||
|
||||
webhookStore := webhook.NewMemoryStore()
|
||||
srv.Mount("/webhooks", webhook.NewRouter(webhookStore))
|
||||
log.Info().Msg("Webhook routes mounted at /webhooks")
|
||||
|
||||
// --- gRPC Server (optional) ---
|
||||
grpcAddr := os.Getenv("BROKER_GRPC_LISTEN")
|
||||
var grpcSrv *brokergrpc.Server
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MemoryStore is an in-memory Store for development and testing.
|
||||
type MemoryStore struct {
|
||||
mu sync.RWMutex
|
||||
webhooks map[string]*Webhook // id -> webhook
|
||||
deliveries map[string]*Delivery // id -> delivery
|
||||
counter int
|
||||
}
|
||||
|
||||
// NewMemoryStore returns an in-memory webhook store.
|
||||
func NewMemoryStore() *MemoryStore {
|
||||
return &MemoryStore{
|
||||
webhooks: make(map[string]*Webhook),
|
||||
deliveries: make(map[string]*Delivery),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *MemoryStore) nextID() string {
|
||||
s.counter++
|
||||
return fmt.Sprintf("wh_%d", s.counter)
|
||||
}
|
||||
|
||||
func (s *MemoryStore) ListByEvent(orgID, event string) ([]Webhook, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
var out []Webhook
|
||||
for _, wh := range s.webhooks {
|
||||
if wh.OrgID != orgID || !wh.Active {
|
||||
continue
|
||||
}
|
||||
for _, ev := range wh.Events {
|
||||
if ev == event || ev == "*" {
|
||||
out = append(out, *wh)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *MemoryStore) GetByID(orgID, id string) (*Webhook, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
wh, ok := s.webhooks[id]
|
||||
if !ok || wh.OrgID != orgID {
|
||||
return nil, fmt.Errorf("webhook not found")
|
||||
}
|
||||
return wh, nil
|
||||
}
|
||||
|
||||
func (s *MemoryStore) List(orgID string) ([]Webhook, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
var out []Webhook
|
||||
for _, wh := range s.webhooks {
|
||||
if wh.OrgID == orgID {
|
||||
out = append(out, *wh)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *MemoryStore) Save(wh *Webhook) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if wh.ID == "" {
|
||||
wh.ID = s.nextID()
|
||||
wh.CreatedAt = time.Now().UTC()
|
||||
}
|
||||
s.webhooks[wh.ID] = wh
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *MemoryStore) Delete(orgID, id string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
wh, ok := s.webhooks[id]
|
||||
if !ok || wh.OrgID != orgID {
|
||||
return fmt.Errorf("webhook not found")
|
||||
}
|
||||
delete(s.webhooks, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *MemoryStore) LogDelivery(d *Delivery) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if d.ID == "" {
|
||||
s.counter++
|
||||
d.ID = fmt.Sprintf("del_%d", s.counter)
|
||||
d.CreatedAt = time.Now().UTC()
|
||||
}
|
||||
s.deliveries[d.ID] = d
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *MemoryStore) ListDeliveries(webhookID string, limit int) ([]Delivery, error) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
var out []Delivery
|
||||
for _, d := range s.deliveries {
|
||||
if d.WebhookID == webhookID {
|
||||
out = append(out, *d)
|
||||
if limit > 0 && len(out) >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
// NewRouter creates a chi sub-router for webhook management.
|
||||
// Mount at /v1/bd/webhooks on the main router.
|
||||
//
|
||||
// POST / — register webhook
|
||||
// GET / — list webhooks
|
||||
// DELETE /{id} — delete webhook
|
||||
// GET /{id}/deliveries — list delivery history
|
||||
func NewRouter(store Store) chi.Router {
|
||||
h := &handler{store: store}
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Post("/", h.handleCreate)
|
||||
r.Get("/", h.handleList)
|
||||
r.Delete("/{id}", h.handleDelete)
|
||||
r.Get("/{id}/deliveries", h.handleListDeliveries)
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
type handler struct {
|
||||
store Store
|
||||
}
|
||||
|
||||
func (h *handler) handleCreate(w http.ResponseWriter, r *http.Request) {
|
||||
orgID := r.Header.Get("X-Org-Id")
|
||||
if orgID == "" {
|
||||
writeErr(w, http.StatusUnauthorized, "org identity required")
|
||||
return
|
||||
}
|
||||
|
||||
var req struct {
|
||||
URL string `json:"url"`
|
||||
Events []string `json:"events"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeErr(w, http.StatusBadRequest, "invalid request body")
|
||||
return
|
||||
}
|
||||
if req.URL == "" {
|
||||
writeErr(w, http.StatusBadRequest, "url is required")
|
||||
return
|
||||
}
|
||||
if len(req.Events) == 0 {
|
||||
writeErr(w, http.StatusBadRequest, "events is required")
|
||||
return
|
||||
}
|
||||
|
||||
secret := generateSecret()
|
||||
wh := &Webhook{
|
||||
OrgID: orgID,
|
||||
URL: req.URL,
|
||||
Secret: secret,
|
||||
Events: req.Events,
|
||||
Active: true,
|
||||
}
|
||||
if err := h.store.Save(wh); err != nil {
|
||||
writeErr(w, http.StatusInternalServerError, "internal error")
|
||||
return
|
||||
}
|
||||
|
||||
// Return secret only on creation so the caller can store it.
|
||||
resp := map[string]interface{}{
|
||||
"id": wh.ID,
|
||||
"url": wh.URL,
|
||||
"events": wh.Events,
|
||||
"active": wh.Active,
|
||||
"secret": secret,
|
||||
"created_at": wh.CreatedAt,
|
||||
}
|
||||
writeJSON(w, http.StatusCreated, resp)
|
||||
}
|
||||
|
||||
func (h *handler) handleList(w http.ResponseWriter, r *http.Request) {
|
||||
orgID := r.Header.Get("X-Org-Id")
|
||||
if orgID == "" {
|
||||
writeErr(w, http.StatusUnauthorized, "org identity required")
|
||||
return
|
||||
}
|
||||
|
||||
hooks, err := h.store.List(orgID)
|
||||
if err != nil {
|
||||
writeErr(w, http.StatusInternalServerError, "internal error")
|
||||
return
|
||||
}
|
||||
if hooks == nil {
|
||||
hooks = []Webhook{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, hooks)
|
||||
}
|
||||
|
||||
func (h *handler) handleDelete(w http.ResponseWriter, r *http.Request) {
|
||||
orgID := r.Header.Get("X-Org-Id")
|
||||
if orgID == "" {
|
||||
writeErr(w, http.StatusUnauthorized, "org identity required")
|
||||
return
|
||||
}
|
||||
|
||||
id := chi.URLParam(r, "id")
|
||||
if err := h.store.Delete(orgID, id); err != nil {
|
||||
writeErr(w, http.StatusNotFound, "webhook not found")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "deleted"})
|
||||
}
|
||||
|
||||
func (h *handler) handleListDeliveries(w http.ResponseWriter, r *http.Request) {
|
||||
orgID := r.Header.Get("X-Org-Id")
|
||||
if orgID == "" {
|
||||
writeErr(w, http.StatusUnauthorized, "org identity required")
|
||||
return
|
||||
}
|
||||
|
||||
id := chi.URLParam(r, "id")
|
||||
// Verify the webhook belongs to this org.
|
||||
if _, err := h.store.GetByID(orgID, id); err != nil {
|
||||
writeErr(w, http.StatusNotFound, "webhook not found")
|
||||
return
|
||||
}
|
||||
|
||||
deliveries, err := h.store.ListDeliveries(id, 100)
|
||||
if err != nil {
|
||||
writeErr(w, http.StatusInternalServerError, "internal error")
|
||||
return
|
||||
}
|
||||
if deliveries == nil {
|
||||
deliveries = []Delivery{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, deliveries)
|
||||
}
|
||||
|
||||
func generateSecret() string {
|
||||
b := make([]byte, 32)
|
||||
rand.Read(b)
|
||||
return hex.EncodeToString(b)
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, v interface{}) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
json.NewEncoder(w).Encode(v)
|
||||
}
|
||||
|
||||
func writeErr(w http.ResponseWriter, status int, msg string) {
|
||||
writeJSON(w, status, map[string]string{"error": msg})
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package webhook
|
||||
|
||||
import "time"
|
||||
|
||||
// Webhook is a registered webhook endpoint.
|
||||
type Webhook struct {
|
||||
ID string `json:"id"`
|
||||
OrgID string `json:"org_id"`
|
||||
URL string `json:"url"`
|
||||
Secret string `json:"-"` // HMAC-SHA256 signing secret; never serialized
|
||||
Events []string `json:"events"`
|
||||
Active bool `json:"active"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// Delivery is a log entry for a single webhook delivery attempt.
|
||||
type Delivery struct {
|
||||
ID string `json:"id"`
|
||||
WebhookID string `json:"webhook_id"`
|
||||
EventType string `json:"event_type"`
|
||||
EventID string `json:"event_id"`
|
||||
Status string `json:"status"` // pending, delivered, failed
|
||||
ResponseCode int `json:"response_code,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Attempts int `json:"attempts"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// Store is the persistence interface for webhook registrations and delivery logs.
|
||||
type Store interface {
|
||||
// ListByEvent returns all active webhooks for the given org subscribed to the event.
|
||||
ListByEvent(orgID, event string) ([]Webhook, error)
|
||||
|
||||
// GetByID returns a single webhook by ID, scoped to org.
|
||||
GetByID(orgID, id string) (*Webhook, error)
|
||||
|
||||
// List returns all webhooks for the given org.
|
||||
List(orgID string) ([]Webhook, error)
|
||||
|
||||
// Save creates or updates a webhook.
|
||||
Save(wh *Webhook) error
|
||||
|
||||
// Delete removes a webhook by ID, scoped to org.
|
||||
Delete(orgID, id string) error
|
||||
|
||||
// LogDelivery records a delivery attempt.
|
||||
LogDelivery(d *Delivery) error
|
||||
|
||||
// ListDeliveries returns delivery history for a webhook.
|
||||
ListDeliveries(webhookID string, limit int) ([]Delivery, error)
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
// Package webhook provides HMAC-SHA256 signed webhook delivery with
|
||||
// exponential backoff retry for the broker-dealer service.
|
||||
//
|
||||
// Consistent with the ATS (client_routes.go) and TA (webhook.go) patterns:
|
||||
// same headers, same signature format, same retry semantics.
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/hmac"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
// maxRetries is the number of delivery attempts before giving up.
|
||||
const maxRetries = 3
|
||||
|
||||
// backoffBase is the base delay for exponential backoff: 1s, 4s, 16s.
|
||||
const backoffBase = 1 * time.Second
|
||||
|
||||
var httpClient = &http.Client{Timeout: 30 * time.Second}
|
||||
|
||||
// Deliver sends a webhook event to all registered subscribers for the given
|
||||
// org and event type. Each subscriber is delivered independently; failures
|
||||
// on one do not block others. Delivery is logged via the store.
|
||||
//
|
||||
// Headers match the ATS/TA convention:
|
||||
// - X-Webhook-Signature: sha256=<hex>
|
||||
// - X-Event-Type: <event>
|
||||
// - X-Event-ID: <id>
|
||||
// - X-Webhook-Timestamp: <unix>
|
||||
func Deliver(store Store, orgID, eventType string, payload any) {
|
||||
if store == nil {
|
||||
return
|
||||
}
|
||||
|
||||
hooks, err := store.ListByEvent(orgID, eventType)
|
||||
if err != nil || len(hooks) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Str("event", eventType).Msg("webhook: marshal payload")
|
||||
return
|
||||
}
|
||||
|
||||
eventID := generateEventID()
|
||||
|
||||
for _, wh := range hooks {
|
||||
go deliver(store, wh, eventType, eventID, body)
|
||||
}
|
||||
}
|
||||
|
||||
// deliver sends a single webhook with retries.
|
||||
func deliver(store Store, wh Webhook, eventType, eventID string, body []byte) {
|
||||
delivery := &Delivery{
|
||||
WebhookID: wh.ID,
|
||||
EventType: eventType,
|
||||
EventID: eventID,
|
||||
Status: "pending",
|
||||
}
|
||||
|
||||
var lastErr error
|
||||
for attempt := 0; attempt < maxRetries; attempt++ {
|
||||
if attempt > 0 {
|
||||
backoff := time.Duration(math.Pow(4, float64(attempt))) * backoffBase
|
||||
time.Sleep(backoff)
|
||||
}
|
||||
|
||||
delivery.Attempts = attempt + 1
|
||||
code, err := send(wh.URL, wh.Secret, eventType, eventID, body)
|
||||
delivery.ResponseCode = code
|
||||
|
||||
if err == nil && code >= 200 && code < 300 {
|
||||
delivery.Status = "delivered"
|
||||
if logErr := store.LogDelivery(delivery); logErr != nil {
|
||||
log.Error().Err(logErr).Str("webhook", wh.ID).Msg("webhook: log delivery")
|
||||
}
|
||||
log.Info().
|
||||
Str("webhook", wh.ID).
|
||||
Str("event", eventType).
|
||||
Str("event_id", eventID).
|
||||
Int("attempts", attempt+1).
|
||||
Msg("webhook: delivered")
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
} else {
|
||||
lastErr = fmt.Errorf("HTTP %d", code)
|
||||
}
|
||||
}
|
||||
|
||||
delivery.Status = "failed"
|
||||
delivery.Error = lastErr.Error()
|
||||
if logErr := store.LogDelivery(delivery); logErr != nil {
|
||||
log.Error().Err(logErr).Str("webhook", wh.ID).Msg("webhook: log delivery")
|
||||
}
|
||||
log.Error().
|
||||
Err(lastErr).
|
||||
Str("webhook", wh.ID).
|
||||
Str("event", eventType).
|
||||
Str("event_id", eventID).
|
||||
Int("attempts", maxRetries).
|
||||
Msg("webhook: delivery failed")
|
||||
}
|
||||
|
||||
// send performs a single HTTP POST with HMAC-SHA256 signature.
|
||||
func send(url, secret, eventType, eventID string, body []byte) (int, error) {
|
||||
timestamp := fmt.Sprintf("%d", time.Now().Unix())
|
||||
signatureBody := timestamp + "." + string(body)
|
||||
signature := hmacSHA256(secret, signatureBody)
|
||||
|
||||
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("create request: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("X-Webhook-Signature", "sha256="+signature)
|
||||
req.Header.Set("X-Event-Type", eventType)
|
||||
req.Header.Set("X-Event-ID", eventID)
|
||||
req.Header.Set("X-Webhook-Timestamp", timestamp)
|
||||
|
||||
resp, err := httpClient.Do(req)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("http post: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
|
||||
return resp.StatusCode, nil
|
||||
}
|
||||
|
||||
// hmacSHA256 computes HMAC-SHA256 of msg using the given key.
|
||||
func hmacSHA256(key, msg string) string {
|
||||
mac := hmac.New(sha256.New, []byte(key))
|
||||
mac.Write([]byte(msg))
|
||||
return hex.EncodeToString(mac.Sum(nil))
|
||||
}
|
||||
|
||||
// VerifySignature checks that a received signature matches the expected
|
||||
// HMAC-SHA256. Receivers of BD webhooks use this for validation.
|
||||
func VerifySignature(payload []byte, timestamp, signature, secret string) bool {
|
||||
signatureBody := timestamp + "." + string(payload)
|
||||
expected := "sha256=" + hmacSHA256(secret, signatureBody)
|
||||
return hmac.Equal([]byte(expected), []byte(signature))
|
||||
}
|
||||
|
||||
func generateEventID() string {
|
||||
b := make([]byte, 16)
|
||||
rand.Read(b)
|
||||
return "evt_" + hex.EncodeToString(b)
|
||||
}
|
||||
@@ -0,0 +1,338 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestDeliver_Success(t *testing.T) {
|
||||
var deliveries atomic.Int32
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
t.Errorf("expected POST, got %s", r.Method)
|
||||
}
|
||||
if r.Header.Get("Content-Type") != "application/json" {
|
||||
t.Errorf("expected application/json, got %s", r.Header.Get("Content-Type"))
|
||||
}
|
||||
if r.Header.Get("X-Webhook-Signature") == "" {
|
||||
t.Error("missing X-Webhook-Signature")
|
||||
}
|
||||
if r.Header.Get("X-Event-Type") != "order.placed" {
|
||||
t.Errorf("expected X-Event-Type order.placed, got %s", r.Header.Get("X-Event-Type"))
|
||||
}
|
||||
if r.Header.Get("X-Event-ID") == "" {
|
||||
t.Error("missing X-Event-ID")
|
||||
}
|
||||
if r.Header.Get("X-Webhook-Timestamp") == "" {
|
||||
t.Error("missing X-Webhook-Timestamp")
|
||||
}
|
||||
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
if len(body) == 0 {
|
||||
t.Error("empty body")
|
||||
}
|
||||
|
||||
// Verify HMAC signature.
|
||||
ts := r.Header.Get("X-Webhook-Timestamp")
|
||||
sig := r.Header.Get("X-Webhook-Signature")
|
||||
if !VerifySignature(body, ts, sig, "test-secret") {
|
||||
t.Error("signature verification failed")
|
||||
}
|
||||
|
||||
deliveries.Add(1)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
store := NewMemoryStore()
|
||||
store.Save(&Webhook{
|
||||
OrgID: "org1",
|
||||
URL: srv.URL,
|
||||
Secret: "test-secret",
|
||||
Events: []string{"order.placed"},
|
||||
Active: true,
|
||||
})
|
||||
|
||||
Deliver(store, "org1", "order.placed", map[string]string{"order_id": "123"})
|
||||
|
||||
// Wait for async delivery.
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for deliveries.Load() < 1 && time.Now().Before(deadline) {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
|
||||
if deliveries.Load() != 1 {
|
||||
t.Errorf("expected 1 delivery, got %d", deliveries.Load())
|
||||
}
|
||||
|
||||
// Verify delivery was logged.
|
||||
dels, _ := store.ListDeliveries(store.webhooks["wh_1"].ID, 10)
|
||||
if len(dels) != 1 {
|
||||
t.Fatalf("expected 1 delivery log, got %d", len(dels))
|
||||
}
|
||||
if dels[0].Status != "delivered" {
|
||||
t.Errorf("expected status delivered, got %s", dels[0].Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliver_RetryOnFailure(t *testing.T) {
|
||||
var attempts atomic.Int32
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
n := attempts.Add(1)
|
||||
if n < 3 {
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
store := NewMemoryStore()
|
||||
store.Save(&Webhook{
|
||||
OrgID: "org1",
|
||||
URL: srv.URL,
|
||||
Secret: "s",
|
||||
Events: []string{"order.filled"},
|
||||
Active: true,
|
||||
})
|
||||
|
||||
Deliver(store, "org1", "order.filled", map[string]string{"order_id": "456"})
|
||||
|
||||
// Wait for retries (backoff: 1s + 4s worst case, but attempt 3 succeeds).
|
||||
deadline := time.Now().Add(15 * time.Second)
|
||||
for attempts.Load() < 3 && time.Now().Before(deadline) {
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
|
||||
if attempts.Load() < 3 {
|
||||
t.Errorf("expected at least 3 attempts, got %d", attempts.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliver_NoMatchingWebhooks(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
store.Save(&Webhook{
|
||||
OrgID: "org1",
|
||||
URL: "http://example.com",
|
||||
Secret: "s",
|
||||
Events: []string{"order.placed"},
|
||||
Active: true,
|
||||
})
|
||||
|
||||
// Different event — should not deliver.
|
||||
Deliver(store, "org1", "transfer.completed", nil)
|
||||
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
dels, _ := store.ListDeliveries("wh_1", 10)
|
||||
if len(dels) != 0 {
|
||||
t.Errorf("expected 0 deliveries, got %d", len(dels))
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeliver_WildcardSubscription(t *testing.T) {
|
||||
var deliveries atomic.Int32
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
deliveries.Add(1)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
store := NewMemoryStore()
|
||||
store.Save(&Webhook{
|
||||
OrgID: "org1",
|
||||
URL: srv.URL,
|
||||
Secret: "s",
|
||||
Events: []string{"*"},
|
||||
Active: true,
|
||||
})
|
||||
|
||||
Deliver(store, "org1", "anything.here", map[string]string{"key": "val"})
|
||||
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for deliveries.Load() < 1 && time.Now().Before(deadline) {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
|
||||
if deliveries.Load() != 1 {
|
||||
t.Errorf("expected 1 delivery, got %d", deliveries.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifySignature(t *testing.T) {
|
||||
payload := []byte(`{"event":"test"}`)
|
||||
ts := "1700000000"
|
||||
secret := "my-secret"
|
||||
|
||||
signatureBody := ts + "." + string(payload)
|
||||
sig := "sha256=" + hmacSHA256(secret, signatureBody)
|
||||
|
||||
if !VerifySignature(payload, ts, sig, secret) {
|
||||
t.Error("valid signature rejected")
|
||||
}
|
||||
if VerifySignature(payload, ts, "sha256=bad", secret) {
|
||||
t.Error("invalid signature accepted")
|
||||
}
|
||||
if VerifySignature(payload, ts, sig, "wrong-secret") {
|
||||
t.Error("wrong secret accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// --- Route tests ---
|
||||
|
||||
func TestRoutes_CreateAndList(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
r := NewRouter(store)
|
||||
|
||||
// Create a webhook.
|
||||
body := `{"url":"https://example.com/hook","events":["order.placed","order.filled"]}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w := httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusCreated {
|
||||
t.Fatalf("create: expected 201, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
var created map[string]interface{}
|
||||
json.Unmarshal(w.Body.Bytes(), &created)
|
||||
if created["secret"] == nil || created["secret"] == "" {
|
||||
t.Error("create: secret not returned")
|
||||
}
|
||||
if created["id"] == nil || created["id"] == "" {
|
||||
t.Error("create: id not returned")
|
||||
}
|
||||
|
||||
// List webhooks.
|
||||
req = httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
w = httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("list: expected 200, got %d", w.Code)
|
||||
}
|
||||
|
||||
var hooks []Webhook
|
||||
json.Unmarshal(w.Body.Bytes(), &hooks)
|
||||
if len(hooks) != 1 {
|
||||
t.Fatalf("list: expected 1 webhook, got %d", len(hooks))
|
||||
}
|
||||
if hooks[0].URL != "https://example.com/hook" {
|
||||
t.Errorf("list: unexpected URL %s", hooks[0].URL)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoutes_Delete(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
r := NewRouter(store)
|
||||
|
||||
// Create.
|
||||
body := `{"url":"https://example.com/hook","events":["*"]}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w := httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
var created map[string]interface{}
|
||||
json.Unmarshal(w.Body.Bytes(), &created)
|
||||
id := created["id"].(string)
|
||||
|
||||
// Delete.
|
||||
req = httptest.NewRequest(http.MethodDelete, "/"+id, nil)
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
w = httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("delete: expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
// List should be empty.
|
||||
req = httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
w = httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
var hooks []Webhook
|
||||
json.Unmarshal(w.Body.Bytes(), &hooks)
|
||||
if len(hooks) != 0 {
|
||||
t.Errorf("list after delete: expected 0, got %d", len(hooks))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoutes_OrgIsolation(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
r := NewRouter(store)
|
||||
|
||||
// Create webhook for org1.
|
||||
body := `{"url":"https://example.com/hook","events":["*"]}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w := httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
// List as org2 — should see nothing.
|
||||
req = httptest.NewRequest(http.MethodGet, "/", nil)
|
||||
req.Header.Set("X-Org-Id", "org2")
|
||||
w = httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
var hooks []Webhook
|
||||
json.Unmarshal(w.Body.Bytes(), &hooks)
|
||||
if len(hooks) != 0 {
|
||||
t.Errorf("org2 should see 0 webhooks, got %d", len(hooks))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoutes_Validation(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
r := NewRouter(store)
|
||||
|
||||
// Missing URL.
|
||||
body := `{"events":["order.placed"]}`
|
||||
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w := httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("missing url: expected 400, got %d", w.Code)
|
||||
}
|
||||
|
||||
// Missing events.
|
||||
body = `{"url":"https://example.com/hook"}`
|
||||
req = httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
|
||||
req.Header.Set("X-Org-Id", "org1")
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w = httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusBadRequest {
|
||||
t.Errorf("missing events: expected 400, got %d", w.Code)
|
||||
}
|
||||
|
||||
// Missing org.
|
||||
body = `{"url":"https://example.com/hook","events":["*"]}`
|
||||
req = httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w = httptest.NewRecorder()
|
||||
r.ServeHTTP(w, req)
|
||||
|
||||
if w.Code != http.StatusUnauthorized {
|
||||
t.Errorf("missing org: expected 401, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user