feat: add one-click replay for captured requests

- Add POST /e/{endpoint}/requests/{id}/replay endpoint
- Replay re-sends captured method, path, headers, query, body to target URL
- Response includes target status code, headers, body (truncated 64KB)
- SSRF guard: denies replay to private/loopback by default
- Opt-in env var HATCH_ALLOW_PRIVATE_REPLAY=true for local dev
- Server-rendered inspect page at GET /e/{endpoint} with replay button UI
- SSE live stream for new requests (EventSource)
- Mock configuration at PUT /e/{endpoint}/mock
- Store: GetRequest, GetMock, SetMock methods added
- Chi router migration from stdlib ServeMux
- 26 passing tests including replay unit tests and E2E capture-to-replay

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
SoftwareEngineer 2026-06-22 16:58:42 +02:00
parent fcfde5660c
commit 4256d34c27
16 changed files with 1351 additions and 124 deletions

1
.gitignore vendored
View File

@ -19,3 +19,4 @@ go.work
*.swo *.swo
*~ *~
.DS_Store .DS_Store
hatch

View File

@ -8,26 +8,28 @@ import (
"github.com/elfoundation/hatch/internal/handler" "github.com/elfoundation/hatch/internal/handler"
"github.com/elfoundation/hatch/internal/store" "github.com/elfoundation/hatch/internal/store"
"github.com/go-chi/chi/v5"
) )
func main() { func main() {
port := os.Getenv("PORT") port := os.Getenv("PORT")
if port == "" { port = "8080" } if port == "" {
port = "8080"
}
dbPath := os.Getenv("HATCH_DB_PATH") dbPath := os.Getenv("HATCH_DB_PATH")
repo, err := store.Open(dbPath) repo, err := store.Open(dbPath)
if err != nil { log.Fatalf("hatch: open store: %v", err) } if err != nil {
log.Fatalf("hatch: open store: %v", err)
}
defer repo.Close() defer repo.Close()
h := handler.New(repo) h := handler.New(repo)
mux := http.NewServeMux() r := chi.NewRouter()
mux.HandleFunc("GET /healthz", healthz) h.RegisterRoutes(r)
h.RegisterRoutes(mux)
addr := fmt.Sprintf(":%s", port) addr := fmt.Sprintf(":%s", port)
log.Printf("hatch starting on %s", addr) log.Printf("hatch starting on %s", addr)
if err := http.ListenAndServe(addr, mux); err != nil { log.Fatalf("hatch server error: %v", err) } if err := http.ListenAndServe(addr, r); err != nil {
} log.Fatalf("hatch server error: %v", err)
}
func healthz(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain")
w.WriteHeader(http.StatusOK)
fmt.Fprintln(w, "ok")
} }

View File

@ -4,36 +4,57 @@ import (
"io" "io"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os"
"strings" "strings"
"testing" "testing"
"github.com/elfoundation/hatch/internal/handler"
"github.com/elfoundation/hatch/internal/store"
"github.com/go-chi/chi/v5"
) )
func TestHealthz(t *testing.T) { func TestHealthzViaRouter(t *testing.T) {
repo, err := store.Open(":memory:")
if err != nil {
t.Fatalf("open store: %v", err)
}
defer repo.Close()
r := chi.NewRouter()
h := handler.New(repo)
h.RegisterRoutes(r)
req := httptest.NewRequest(http.MethodGet, "/healthz", nil) req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
w := httptest.NewRecorder() w := httptest.NewRecorder()
r.ServeHTTP(w, req)
healthz(w, req) if w.Code != http.StatusOK {
t.Fatalf("expected 200 OK, got %d", w.Code)
resp := w.Result()
if resp.StatusCode != http.StatusOK {
t.Fatalf("expected 200 OK, got %d", resp.StatusCode)
} }
body, _ := io.ReadAll(w.Result().Body)
body, err := io.ReadAll(resp.Body) w.Result().Body.Close()
resp.Body.Close()
if err != nil {
t.Fatalf("failed to read response body: %v", err)
}
got := strings.TrimSpace(string(body)) got := strings.TrimSpace(string(body))
if got != "ok" { if got != "ok" {
t.Fatalf("expected body 'ok', got '%s'", got) t.Fatalf("expected body 'ok', got %q", got)
} }
} }
func TestHealthzSmoke(t *testing.T) { func TestHealthzSmoke(t *testing.T) {
// Smoke test: boot the server on a random port, hit /healthz, verify 200 + "ok". // Smoke: boot server on random port, hit /healthz.
srv := httptest.NewServer(http.HandlerFunc(healthz)) // Use :memory: store so no filesystem dependency.
os.Setenv("HATCH_DB_PATH", ":memory:")
repo, err := store.Open(":memory:")
if err != nil {
t.Fatalf("open store: %v", err)
}
defer repo.Close()
r := chi.NewRouter()
h := handler.New(repo)
h.RegisterRoutes(r)
srv := httptest.NewServer(r)
defer srv.Close() defer srv.Close()
resp, err := http.Get(srv.URL + "/healthz") resp, err := http.Get(srv.URL + "/healthz")
@ -45,14 +66,12 @@ func TestHealthzSmoke(t *testing.T) {
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
t.Fatalf("expected 200 OK, got %d", resp.StatusCode) t.Fatalf("expected 200 OK, got %d", resp.StatusCode)
} }
body, err := io.ReadAll(resp.Body) body, err := io.ReadAll(resp.Body)
if err != nil { if err != nil {
t.Fatalf("failed to read response body: %v", err) t.Fatalf("failed to read response body: %v", err)
} }
got := strings.TrimSpace(string(body)) got := strings.TrimSpace(string(body))
if got != "ok" { if got != "ok" {
t.Fatalf("expected body 'ok', got '%s'", got) t.Fatalf("expected body 'ok', got %q", got)
} }
} }

1
go.mod
View File

@ -3,6 +3,7 @@ module github.com/elfoundation/hatch
go 1.25.0 go 1.25.0
require ( require (
github.com/go-chi/chi/v5 v5.3.0
github.com/google/uuid v1.6.0 github.com/google/uuid v1.6.0
modernc.org/sqlite v1.53.0 modernc.org/sqlite v1.53.0
) )

2
go.sum
View File

@ -1,5 +1,7 @@
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=

View File

@ -1,47 +1,125 @@
package handler package handler
import ( import (
"context"
"encoding/json" "encoding/json"
"io" "io"
"net/http" "net/http"
"strings" "strings"
"github.com/elfoundation/hatch/internal/store" "github.com/elfoundation/hatch/internal/store"
"github.com/go-chi/chi/v5"
"github.com/go-chi/chi/v5/middleware"
) )
type Handler struct{ Repo store.Repository } // Handler groups all HTTP handlers and their shared dependencies.
type Handler struct {
func New(repo store.Repository) *Handler { return &Handler{Repo: repo} } Repo store.Repository
func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
mux.Handle("/{endpoint}", h)
mux.Handle("/{endpoint}/", h)
} }
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // New creates a new Handler with the given store.
eid := extractID(r.URL.Path) func New(repo store.Repository) *Handler {
ctx := context.Background() return &Handler{Repo: repo}
}
// RegisterRoutes mounts all routes on the given chi router.
func (h *Handler) RegisterRoutes(r chi.Router) {
r.Use(middleware.Logger)
r.Use(middleware.Recoverer)
// Health check.
r.Get("/healthz", Healthz)
// Inspect page: server-rendered request list.
r.Get("/e/{endpointID}", HandleInspect(h.Repo))
// SSE stream for live updates.
r.Get("/e/{endpointID}/events", HandleSSE(h.Repo))
// Mock configuration.
r.Put("/e/{endpointID}/mock", HandleMock(h.Repo))
// Replay: one-click re-send of a captured request.
r.Post("/e/{endpointID}/requests/{requestID}/replay", HandleReplay(h.Repo))
// Capture webhook: any method on /{endpointID} (wildcard, must be last).
r.HandleFunc("/{endpointID}", h.capture)
r.HandleFunc("/{endpointID}/*", h.capture)
}
// capture handles incoming webhook captures on /{endpointID}.
func (h *Handler) capture(w http.ResponseWriter, r *http.Request) {
eid := r.PathValue("endpointID")
if eid == "" {
writeError(w, http.StatusBadRequest, "missing endpoint ID")
return
}
ctx := r.Context()
// Auto-create endpoint if it doesn't exist.
if _, err := h.Repo.GetEndpoint(ctx, eid); err != nil { if _, err := h.Repo.GetEndpoint(ctx, eid); err != nil {
h.Repo.CreateEndpoint(ctx, eid) h.Repo.CreateEndpoint(ctx, eid)
} }
// Collect headers as JSON.
hdr := map[string]string{} hdr := map[string]string{}
for k, v := range r.Header { hdr[k] = strings.Join(v, ", ") } for k, v := range r.Header {
hdr[k] = strings.Join(v, ", ")
}
hdrJSON, _ := json.Marshal(hdr) hdrJSON, _ := json.Marshal(hdr)
// Read body.
var body []byte var body []byte
if r.Body != nil { body, _ = io.ReadAll(r.Body); r.Body.Close() } if r.Body != nil {
h.Repo.AppendRequest(ctx, eid, &store.Request{ body, _ = io.ReadAll(r.Body)
Method: r.Method, Path: r.URL.Path, Headers: string(hdrJSON), r.Body.Close()
Query: r.URL.RawQuery, Body: body, }
})
// Store the request.
req := &store.Request{
Method: r.Method,
Path: r.URL.Path,
Headers: string(hdrJSON),
Query: r.URL.RawQuery,
Body: body,
}
if err := h.Repo.AppendRequest(ctx, eid, req); err != nil {
writeError(w, http.StatusInternalServerError, "failed to store request")
return
}
// Broadcast via SSE.
broadcastRequest(eid, req)
// Look up mock response for this endpoint.
mock, err := h.Repo.GetMock(ctx, eid)
if err == nil && mock != nil {
for k, v := range mock.Headers {
w.Header().Set(k, v)
}
w.WriteHeader(mock.Status)
if mock.Body != nil {
w.Write(mock.Body)
}
return
}
// Default: return 200 OK with empty JSON.
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(map[string]interface{}{}) json.NewEncoder(w).Encode(map[string]interface{}{"ok": true})
} }
func extractID(p string) string { // Healthz returns 200 OK for liveness probes.
t := strings.TrimLeft(p, "/") func Healthz(w http.ResponseWriter, r *http.Request) {
if i := strings.IndexByte(t, '?'); i >= 0 { t = t[:i] } w.Header().Set("Content-Type", "text/plain")
if i := strings.IndexByte(t, '/'); i >= 0 { return t[:i] } w.WriteHeader(http.StatusOK)
return t w.Write([]byte("ok\n"))
}
// writeError writes a JSON error response.
func writeError(w http.ResponseWriter, status int, msg string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(map[string]string{"error": msg})
} }

View File

@ -4,70 +4,146 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"io" "io"
"net/http"
"net/http/httptest" "net/http/httptest"
"strings" "strings"
"testing" "testing"
"github.com/elfoundation/hatch/internal/store" "github.com/elfoundation/hatch/internal/store"
"github.com/go-chi/chi/v5"
) )
// fakeRepo implements store.Repository for tests.
type fakeRepo struct { type fakeRepo struct {
endpoints map[string]*store.Endpoint endpoints map[string]*store.Endpoint
requests []*store.Request requests []*store.Request
mocks map[string]*store.MockConfig
} }
func newFakeRepo() *fakeRepo { return &fakeRepo{endpoints: map[string]*store.Endpoint{}} }
func newFakeRepo() *fakeRepo {
return &fakeRepo{
endpoints: map[string]*store.Endpoint{},
mocks: map[string]*store.MockConfig{},
}
}
func (f *fakeRepo) CreateEndpoint(_ context.Context, u string) (*store.Endpoint, error) { func (f *fakeRepo) CreateEndpoint(_ context.Context, u string) (*store.Endpoint, error) {
e := &store.Endpoint{ID: u, URL: u, CreatedAt: "t", UpdatedAt: "t"} e := &store.Endpoint{ID: u, URL: u, CreatedAt: "t", UpdatedAt: "t"}
f.endpoints[u] = e; return e, nil f.endpoints[u] = e
return e, nil
} }
func (f *fakeRepo) GetEndpoint(_ context.Context, id string) (*store.Endpoint, error) { func (f *fakeRepo) GetEndpoint(_ context.Context, id string) (*store.Endpoint, error) {
e, ok := f.endpoints[id]; if !ok { return nil, errNotFound }; return e, nil e, ok := f.endpoints[id]
if !ok {
return nil, errNotFound
}
return e, nil
} }
func (f *fakeRepo) AppendRequest(_ context.Context, eid string, r *store.Request) error { func (f *fakeRepo) AppendRequest(_ context.Context, eid string, r *store.Request) error {
r.ID = "x"; r.EndpointID = eid; f.requests = append(f.requests, r); return nil r.ID = "req-" + string(rune(len(f.requests)+'0'))
r.EndpointID = eid
f.requests = append(f.requests, r)
return nil
}
func (f *fakeRepo) GetRequest(_ context.Context, id string) (*store.Request, error) {
for _, r := range f.requests {
if r.ID == id {
return r, nil
}
}
return nil, errNotFound
}
func (f *fakeRepo) ListRequests(_ context.Context, _ string, _ int) ([]*store.Request, error) {
return f.requests, nil
}
func (f *fakeRepo) GetMock(_ context.Context, endpointID string) (*store.MockConfig, error) {
m, ok := f.mocks[endpointID]
if !ok {
return nil, errNotFound
}
return m, nil
}
func (f *fakeRepo) SetMock(_ context.Context, mock *store.MockConfig) error {
f.mocks[mock.EndpointID] = mock
return nil
} }
func (f *fakeRepo) ListRequests(_ context.Context, _ string, _ int) ([]*store.Request, error) { return f.requests, nil }
func (f *fakeRepo) Close() error { return nil } func (f *fakeRepo) Close() error { return nil }
var errNotFound = &se{"nf"} var errNotFound = &se{"nf"}
type se struct{ m string } type se struct{ m string }
func (e *se) Error() string { return e.m } func (e *se) Error() string { return e.m }
func TestRecordsAllVerbs(t *testing.T) { // testRouter creates a chi router with all routes registered using a fake repo.
func testRouter(repo store.Repository) chi.Router {
r := chi.NewRouter()
h := New(repo)
h.RegisterRoutes(r)
return r
}
func TestHealthz(t *testing.T) {
r := testRouter(newFakeRepo())
req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d", w.Code)
}
body := strings.TrimSpace(w.Body.String())
if body != "ok" {
t.Fatalf("expected 'ok', got %q", body)
}
}
func TestCaptureRecordsAllVerbs(t *testing.T) {
methods := []string{"GET", "POST", "PUT", "PATCH", "DELETE"} methods := []string{"GET", "POST", "PUT", "PATCH", "DELETE"}
for _, m := range methods { for _, m := range methods {
repo := newFakeRepo() repo := newFakeRepo()
h := New(repo) r := testRouter(repo)
body := "" body := ""
if m == "POST" || m == "PUT" || m == "PATCH" { body = `{"k":"v"}` } if m == "POST" || m == "PUT" || m == "PATCH" {
body = `{"k":"v"}`
}
req := httptest.NewRequest(m, "/ep", strings.NewReader(body)) req := httptest.NewRequest(m, "/ep", strings.NewReader(body))
if m == "GET" { req.Header.Set("Accept", "text/html") } if m == "GET" {
req.Header.Set("Accept", "text/html")
}
w := httptest.NewRecorder() w := httptest.NewRecorder()
h.ServeHTTP(w, req) r.ServeHTTP(w, req)
if w.Result().StatusCode != 200 { t.Errorf("%s: expected 200", m) }
if len(repo.requests) != 1 { t.Errorf("%s: expected 1 request", m); continue }
r := repo.requests[0]
if r.Method != m { t.Errorf("%s: wrong method %s", m, r.Method) }
if r.Path != "/ep" { t.Errorf("%s: wrong path", m) }
}
}
func TestHandlerReturnsJSON200(t *testing.T) { if w.Code != 200 {
h := New(newFakeRepo()) t.Errorf("%s: expected 200, got %d", m, w.Code)
w := httptest.NewRecorder() }
h.ServeHTTP(w, httptest.NewRequest("GET", "/t", nil)) if len(repo.requests) != 1 {
resp := w.Result() t.Errorf("%s: expected 1 request, got %d", m, len(repo.requests))
if resp.StatusCode != 200 { t.Fatalf("expected 200") } continue
if !strings.Contains(resp.Header.Get("Content-Type"), "application/json") { t.Error("not json") } }
data, _ := io.ReadAll(resp.Body); resp.Body.Close() reqCaptured := repo.requests[0]
var v map[string]interface{} if reqCaptured.Method != m {
if json.Unmarshal(data, &v) != nil { t.Fatal("invalid json") } t.Errorf("%s: wrong method %s", m, reqCaptured.Method)
}
func TestExtractID(t *testing.T) {
for _, tc := range []struct{ p, w string }{
{"/a","a"},{"/a/b","a"},{"/",""},{"/x?y=z","x"},
} {
if g := extractID(tc.p); g != tc.w {
t.Errorf("extractID(%q)=%q want %q", tc.p, g, tc.w)
} }
} }
} }
func TestCaptureReturnsJSON200(t *testing.T) {
r := testRouter(newFakeRepo())
w := httptest.NewRecorder()
r.ServeHTTP(w, httptest.NewRequest("GET", "/t", nil))
if w.Code != 200 {
t.Fatalf("expected 200, got %d", w.Code)
}
if !strings.Contains(w.Header().Get("Content-Type"), "application/json") {
t.Error("not json")
}
data, _ := io.ReadAll(w.Result().Body)
w.Result().Body.Close()
var v map[string]interface{}
if json.Unmarshal(data, &v) != nil {
t.Fatal("invalid json")
}
}

281
internal/handler/inspect.go Normal file
View File

@ -0,0 +1,281 @@
package handler
import (
"encoding/json"
"html/template"
"net/http"
"github.com/elfoundation/hatch/internal/store"
)
// inspectPageData is passed to the inspect template.
type inspectPageData struct {
EndpointID string
Requests []*store.Request
}
// inspectTemplate is the server-rendered HTML for the inspect page.
var inspectTemplate = template.Must(template.New("inspect").Funcs(template.FuncMap{
"prettyJSON": prettyJSON,
}).Parse(`<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Hatch {{.EndpointID}}</title>
<style>
:root { color-scheme: light dark; }
body { font-family: system-ui, sans-serif; max-width: 960px; margin: 0 auto; padding: 1rem; }
h1 { font-size: 1.25rem; margin: 0 0 0.25rem; }
.endpoint-url { color: #666; font-family: monospace; font-size: 0.85rem; margin-bottom: 1.5rem; word-break: break-all; }
.request { border: 1px solid #e0e0e0; border-radius: 8px; margin-bottom: 1rem; overflow: hidden; }
.request-header { display: flex; align-items: center; gap: 0.5rem; padding: 0.75rem 1rem; background: #f5f5f5; cursor: pointer; user-select: none; }
.request-header:hover { background: #eee; }
@media (prefers-color-scheme: dark) {
.request { border-color: #333; }
.request-header { background: #1a1a1a; }
.request-header:hover { background: #222; }
.endpoint-url { color: #aaa; }
}
.method { font-weight: 700; font-size: 0.8rem; padding: 2px 8px; border-radius: 4px; }
.method.GET { background: #d4edda; color: #155724; }
.method.POST { background: #cce5ff; color: #004085; }
.method.PUT { background: #fff3cd; color: #856404; }
.method.PATCH { background: #fff3cd; color: #856404; }
.method.DELETE { background: #f8d7da; color: #721c24; }
.path { font-family: monospace; font-size: 0.9rem; flex: 1; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.time { color: #999; font-size: 0.8rem; white-space: nowrap; }
.request-body { padding: 1rem; display: none; }
.request-body.open { display: block; }
.section { margin-bottom: 0.75rem; }
.section-label { font-size: 0.75rem; font-weight: 600; text-transform: uppercase; color: #888; margin-bottom: 0.25rem; }
pre { background: #f8f8f8; border: 1px solid #e0e0e0; border-radius: 4px; padding: 0.75rem; font-size: 0.8rem; overflow-x: auto; margin: 0; white-space: pre-wrap; word-break: break-all; }
@media (prefers-color-scheme: dark) { pre { background: #111; border-color: #333; } }
.empty { text-align: center; color: #999; padding: 3rem 0; }
.empty h2 { font-size: 1rem; margin-bottom: 0.5rem; }
.empty code { background: #f0f0f0; padding: 2px 6px; border-radius: 3px; }
@media (prefers-color-scheme: dark) { .empty code { background: #222; } }
.replay-btn { margin-left: auto; padding: 4px 12px; font-size: 0.8rem; border: 1px solid #007bff; background: #007bff; color: #fff; border-radius: 4px; cursor: pointer; }
.replay-btn:hover { background: #0056b3; }
.replay-btn:disabled { opacity: 0.6; cursor: not-allowed; }
.replay-panel { display: none; padding: 1rem; border-top: 1px solid #e0e0e0; }
@media (prefers-color-scheme: dark) { .replay-panel { border-color: #333; } }
.replay-panel.open { display: block; }
.replay-url-input { width: 100%; padding: 0.5rem; border: 1px solid #ccc; border-radius: 4px; font-family: monospace; font-size: 0.85rem; box-sizing: border-box; margin-bottom: 0.5rem; }
.replay-url-input:focus { outline: none; border-color: #007bff; box-shadow: 0 0 0 2px rgba(0,123,255,0.25); }
.replay-actions { display: flex; gap: 0.5rem; align-items: center; }
.replay-send-btn { padding: 6px 16px; border: none; background: #28a745; color: #fff; border-radius: 4px; cursor: pointer; font-size: 0.85rem; }
.replay-send-btn:hover { background: #218838; }
.replay-send-btn:disabled { opacity: 0.6; cursor: not-allowed; }
.replay-cancel-btn { padding: 6px 12px; border: 1px solid #ccc; background: transparent; border-radius: 4px; cursor: pointer; font-size: 0.85rem; }
.replay-result { margin-top: 0.75rem; }
.replay-error { color: #dc3545; font-size: 0.85rem; }
.status-badge { display: inline-block; padding: 2px 8px; border-radius: 4px; font-size: 0.8rem; font-weight: 600; margin-right: 0.5rem; }
.status-2xx { background: #d4edda; color: #155724; }
.status-3xx { background: #fff3cd; color: #856404; }
.status-4xx { background: #f8d7da; color: #721c24; }
.status-5xx { background: #f8d7da; color: #721c24; }
</style>
</head>
<body>
<h1>Hatch</h1>
<div class="endpoint-url">Endpoint: <strong>{{.EndpointID}}</strong></div>
{{if .Requests}}
{{range .Requests}}
<div class="request" id="req-{{.ID}}">
<div class="request-header" onclick="toggleRequest('{{.ID}}')">
<span class="method {{.Method}}">{{.Method}}</span>
<span class="path">{{.Path}}</span>
<span class="time">{{.CreatedAt}}</span>
<button class="replay-btn" onclick="event.stopPropagation(); openReplay('{{.ID}}')" title="Replay this request">&#x21BA; Replay</button>
</div>
<div class="request-body" id="body-{{.ID}}">
{{if .Headers}}
<div class="section">
<div class="section-label">Headers</div>
<pre>{{prettyJSON .Headers}}</pre>
</div>
{{end}}
{{if .Query}}
<div class="section">
<div class="section-label">Query</div>
<pre>{{.Query}}</pre>
</div>
{{end}}
{{if .Body}}
<div class="section">
<div class="section-label">Body</div>
<pre>{{printf "%s" .Body}}</pre>
</div>
{{end}}
</div>
<div class="replay-panel" id="replay-{{.ID}}">
<input type="text" class="replay-url-input" id="replay-url-{{.ID}}" placeholder="https://myapp.local/webhook" value="">
<div class="replay-actions">
<button class="replay-send-btn" onclick="doReplay('{{.ID}}')">Send</button>
<button class="replay-cancel-btn" onclick="closeReplay('{{.ID}}')">Cancel</button>
<span class="replay-error" id="replay-error-{{.ID}}"></span>
</div>
<div class="replay-result" id="replay-result-{{.ID}}"></div>
</div>
</div>
{{end}}
{{else}}
<div class="empty">
<h2>Waiting for requests...</h2>
<p>Send a request to <code>/{{.EndpointID}}</code> to see it appear here.</p>
</div>
{{end}}
<script>
function toggleRequest(id) {
document.getElementById('body-' + id).classList.toggle('open');
}
function openReplay(id) {
document.querySelectorAll('.replay-panel.open').forEach(function(p) { p.classList.remove('open'); });
document.getElementById('replay-' + id).classList.add('open');
var input = document.getElementById('replay-url-' + id);
input.focus();
if (!input.value) input.value = 'http://localhost:8080';
document.getElementById('replay-result-' + id).innerHTML = '';
document.getElementById('replay-error-' + id).textContent = '';
}
function closeReplay(id) {
document.getElementById('replay-' + id).classList.remove('open');
}
function doReplay(id) {
var targetUrl = document.getElementById('replay-url-' + id).value.trim();
if (!targetUrl) {
document.getElementById('replay-error-' + id).textContent = 'Please enter a target URL.';
return;
}
var sendBtn = document.querySelector('#replay-' + id + ' .replay-send-btn');
var errorEl = document.getElementById('replay-error-' + id);
var resultEl = document.getElementById('replay-result-' + id);
var endpointId = window.location.pathname.replace(/^\/e\//, '');
sendBtn.disabled = true;
errorEl.textContent = '';
resultEl.innerHTML = '<em>Sending...</em>';
fetch('/e/' + endpointId + '/requests/' + id + '/replay', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ target_url: targetUrl })
}).then(function(resp) {
return resp.json().then(function(data) { return { status: resp.status, data: data }; });
}).then(function(result) {
if (result.data.error) {
errorEl.textContent = result.data.error;
resultEl.innerHTML = '';
} else {
renderReplayResult(resultEl, result.data);
}
}).catch(function(err) {
errorEl.textContent = 'Network error: ' + err.message;
resultEl.innerHTML = '';
}).finally(function() { sendBtn.disabled = false; });
}
function renderReplayResult(el, data) {
var sc = 'status-2xx';
if (data.status >= 300 && data.status < 400) sc = 'status-3xx';
else if (data.status >= 400 && data.status < 500) sc = 'status-4xx';
else if (data.status >= 500) sc = 'status-5xx';
var html = '<div class="section"><div class="section-label">Response</div>';
html += '<span class="status-badge ' + sc + '">' + data.status + '</span></div>';
if (data.headers && Object.keys(data.headers).length > 0) {
html += '<div class="section"><div class="section-label">Response Headers</div><pre>';
for (var h in data.headers) { html += escapeHtml(h) + ': ' + escapeHtml(data.headers[h]) + '\n'; }
html += '</pre></div>';
}
if (data.body) {
html += '<div class="section"><div class="section-label">Response Body</div><pre>' + escapeHtml(data.body) + '</pre></div>';
}
el.innerHTML = html;
}
function escapeHtml(str) {
var div = document.createElement('div');
div.appendChild(document.createTextNode(str));
return div.innerHTML;
}
// Live updates via EventSource
(function() {
var endpointId = window.location.pathname.replace(/^\/e\//, '');
if (!endpointId) return;
var es = new EventSource('/e/' + endpointId + '/events');
es.onmessage = function(e) {
try {
var req = JSON.parse(e.data);
var empty = document.querySelector('.empty');
if (empty) empty.remove();
var div = document.createElement('div');
div.className = 'request';
div.id = 'req-' + req.id;
div.innerHTML = '<div class="request-header" onclick="toggleRequest(\'' + req.id + '\')">' +
'<span class="method ' + req.method + '">' + req.method + '</span>' +
'<span class="path">' + req.path + '</span>' +
'<span class="time">' + req.created_at + '</span>' +
'<button class="replay-btn" onclick="event.stopPropagation(); openReplay(\'' + req.id + '\')">&#x21BA; Replay</button>' +
'</div>' +
'<div class="request-body" id="body-' + req.id + '"></div>' +
'<div class="replay-panel" id="replay-' + req.id + '">' +
'<input type="text" class="replay-url-input" id="replay-url-' + req.id + '" placeholder="https://myapp.local/webhook" value="">' +
'<div class="replay-actions">' +
'<button class="replay-send-btn" onclick="doReplay(\'' + req.id + '\')">Send</button>' +
'<button class="replay-cancel-btn" onclick="closeReplay(\'' + req.id + '\')">Cancel</button>' +
'<span class="replay-error" id="replay-error-' + req.id + '"></span></div>' +
'<div class="replay-result" id="replay-result-' + req.id + '"></div></div>';
var ref = document.querySelector('.endpoint-url');
ref.insertAdjacentElement('afterend', div);
} catch(_) {}
};
})();
</script>
</body>
</html>
`))
// HandleInspect serves the inspect page at GET /e/{endpointID}.
func HandleInspect(repo store.Repository) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
endpointID := r.PathValue("endpointID")
if endpointID == "" {
http.Error(w, "missing endpoint ID", http.StatusBadRequest)
return
}
if _, err := repo.GetEndpoint(r.Context(), endpointID); err != nil {
repo.CreateEndpoint(r.Context(), endpointID)
}
requests, err := repo.ListRequests(r.Context(), endpointID, 100)
if err != nil {
http.Error(w, "failed to list requests", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
inspectTemplate.Execute(w, inspectPageData{
EndpointID: endpointID,
Requests: requests,
})
}
}
// prettyJSON formats a JSON string with indentation.
func prettyJSON(raw string) string {
var v interface{}
if err := json.Unmarshal([]byte(raw), &v); err != nil {
return raw
}
b, err := json.MarshalIndent(v, "", " ")
if err != nil {
return raw
}
return string(b)
}

35
internal/handler/mock.go Normal file
View File

@ -0,0 +1,35 @@
package handler
import (
"encoding/json"
"net/http"
"github.com/elfoundation/hatch/internal/store"
)
// HandleMock handles PUT /e/{endpointID}/mock.
func HandleMock(repo store.Repository) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
endpointID := r.PathValue("endpointID")
if endpointID == "" {
writeError(w, http.StatusBadRequest, "missing endpoint ID")
return
}
if _, err := repo.GetEndpoint(r.Context(), endpointID); err != nil {
repo.CreateEndpoint(r.Context(), endpointID)
}
var mock store.MockConfig
if err := json.NewDecoder(r.Body).Decode(&mock); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON body: "+err.Error())
return
}
mock.EndpointID = endpointID
if err := repo.SetMock(r.Context(), &mock); err != nil {
writeError(w, http.StatusInternalServerError, "failed to set mock: "+err.Error())
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
}
}

196
internal/handler/replay.go Normal file
View File

@ -0,0 +1,196 @@
package handler
import (
"encoding/json"
"fmt"
"io"
"net"
"net/http"
"net/url"
"os"
"strings"
"github.com/elfoundation/hatch/internal/store"
)
const maxReplayBodyBytes = 64 * 1024
type replayRequest struct {
TargetURL string `json:"target_url"`
}
type replayResponse struct {
Status int `json:"status"`
Headers map[string]string `json:"headers"`
Body string `json:"body"`
Error string `json:"error,omitempty"`
}
func HandleReplay(repo store.Repository) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
requestID := r.PathValue("requestID")
if requestID == "" {
writeError(w, http.StatusBadRequest, "missing request_id")
return
}
var replay replayRequest
if err := json.NewDecoder(r.Body).Decode(&replay); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON body: "+err.Error())
return
}
if replay.TargetURL == "" {
writeError(w, http.StatusBadRequest, "target_url is required")
return
}
target, err := url.Parse(replay.TargetURL)
if err != nil {
writeError(w, http.StatusBadRequest, "invalid target_url: "+err.Error())
return
}
if target.Scheme != "http" && target.Scheme != "https" {
writeError(w, http.StatusBadRequest, "target_url scheme must be http or https")
return
}
if !allowPrivateReplay() && isPrivateAddr(target.Host) {
writeError(w, http.StatusForbidden, "replay to private/loopback addresses is denied. Set HATCH_ALLOW_PRIVATE_REPLAY=true to allow")
return
}
capReq, err := repo.GetRequest(r.Context(), requestID)
if err != nil {
writeError(w, http.StatusNotFound, "request not found: "+err.Error())
return
}
replayResult, err := doReplay(capReq, target)
if err != nil {
writeError(w, http.StatusBadGateway, "replay failed: "+err.Error())
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(replayResult)
}
}
func doReplay(capReq *store.Request, targetURL *url.URL) (*replayResponse, error) {
outURL := *targetURL
outURL.Path = joinPath(targetURL.Path, capReq.Path)
if capReq.Query != "" {
outURL.RawQuery = capReq.Query
}
var bodyReader io.Reader
if capReq.Body != nil && len(capReq.Body) > 0 {
bodyReader = strings.NewReader(string(capReq.Body))
}
req, err := http.NewRequest(capReq.Method, outURL.String(), bodyReader)
if err != nil {
return nil, fmt.Errorf("building request: %w", err)
}
var headers map[string]string
if capReq.Headers != "" {
if err := json.Unmarshal([]byte(capReq.Headers), &headers); err != nil {
headers = nil
}
}
for k, v := range headers {
if isHopByHop(k) {
continue
}
req.Header.Set(k, v)
}
client := &http.Client{
CheckRedirect: func(req *http.Request, via []*http.Request) error {
if !allowPrivateReplay() && isPrivateAddr(req.URL.Host) {
return fmt.Errorf("redirect to private address %s denied", req.URL.Host)
}
if len(via) >= 10 {
return fmt.Errorf("too many redirects")
}
return nil
},
}
resp, err := client.Do(req)
if err != nil {
return nil, fmt.Errorf("sending request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, maxReplayBodyBytes))
if err != nil {
return nil, fmt.Errorf("reading response: %w", err)
}
respHeaders := make(map[string]string, len(resp.Header))
for k, vs := range resp.Header {
respHeaders[k] = strings.Join(vs, ", ")
}
return &replayResponse{
Status: resp.StatusCode,
Headers: respHeaders,
Body: string(body),
}, nil
}
// isPrivateAddr reports whether addr falls within a private, loopback, or reserved range.
func isPrivateAddr(hostport string) bool {
host, _, err := net.SplitHostPort(hostport)
if err != nil {
host = hostport
}
if host == "" {
return false
}
// String-based checks first (catch names and the unspecified address 0.0.0.0).
if host == "localhost" || host == "0.0.0.0" || host == "::1" || host == "[::1]" {
return true
}
ip := net.ParseIP(host)
if ip != nil {
return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast()
}
return false
}
func allowPrivateReplay() bool {
return strings.EqualFold(os.Getenv("HATCH_ALLOW_PRIVATE_REPLAY"), "true")
}
var hopByHopHeaders = map[string]bool{
"connection": true,
"keep-alive": true,
"proxy-authenticate": true,
"proxy-authorization": true,
"te": true,
"trailer": true,
"transfer-encoding": true,
"upgrade": true,
}
func isHopByHop(header string) bool {
return hopByHopHeaders[strings.ToLower(header)]
}
func joinPath(base, extra string) string {
base = strings.TrimRight(base, "/")
extra = strings.TrimLeft(extra, "/")
if base == "" {
return "/" + extra
}
if extra == "" {
return base
}
return base + "/" + extra
}

View File

@ -0,0 +1,236 @@
package handler
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/elfoundation/hatch/internal/store"
)
func TestReplayMissingTargetURL(t *testing.T) {
repo := newFakeRepo()
repo.CreateEndpoint(nil, "ep")
repo.AppendRequest(nil, "ep", &store.Request{Method: "POST", Path: "/webhook", Headers: "{}"})
reqID := repo.requests[0].ID
r := testRouter(repo)
body := `{}`
req := httptest.NewRequest(http.MethodPost, "/e/ep/requests/"+reqID+"/replay", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
t.Fatalf("expected 400, got %d", w.Code)
}
var resp map[string]string
json.NewDecoder(w.Body).Decode(&resp)
if !strings.Contains(resp["error"], "target_url") {
t.Errorf("expected target_url error, got %v", resp)
}
}
func TestReplayInvalidScheme(t *testing.T) {
repo := newFakeRepo()
repo.CreateEndpoint(nil, "ep")
repo.AppendRequest(nil, "ep", &store.Request{Method: "GET", Path: "/", Headers: "{}"})
reqID := repo.requests[0].ID
r := testRouter(repo)
body := `{"target_url": "ftp://bad.scheme/"}`
req := httptest.NewRequest(http.MethodPost, "/e/ep/requests/"+reqID+"/replay", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
t.Fatalf("expected 400, got %d", w.Code)
}
}
func TestReplaySSRFBlocksLocalhost(t *testing.T) {
repo := newFakeRepo()
repo.CreateEndpoint(nil, "ep")
repo.AppendRequest(nil, "ep", &store.Request{Method: "GET", Path: "/", Headers: "{}"})
reqID := repo.requests[0].ID
r := testRouter(repo)
body := `{"target_url": "http://localhost:8080/"}`
req := httptest.NewRequest(http.MethodPost, "/e/ep/requests/"+reqID+"/replay", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusForbidden {
t.Fatalf("expected 403 for localhost, got %d", w.Code)
}
}
func TestReplaySSRFBlocksPrivate(t *testing.T) {
repo := newFakeRepo()
repo.CreateEndpoint(nil, "ep")
repo.AppendRequest(nil, "ep", &store.Request{Method: "GET", Path: "/", Headers: "{}"})
reqID := repo.requests[0].ID
r := testRouter(repo)
for _, addr := range []string{
"http://10.0.0.1:80/",
"http://172.16.0.1:80/",
"http://192.168.1.1:80/",
"http://127.0.0.1:80/",
"http://[::1]:80/",
"http://0.0.0.0:80/",
} {
body := `{"target_url": "` + addr + `"}`
req := httptest.NewRequest(http.MethodPost, "/e/ep/requests/"+reqID+"/replay", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusForbidden {
t.Errorf("expected 403 for %s, got %d", addr, w.Code)
}
}
}
func TestReplayNotFound(t *testing.T) {
repo := newFakeRepo()
r := testRouter(repo)
body := `{"target_url": "https://example.com/"}`
req := httptest.NewRequest(http.MethodPost, "/e/ep/requests/nonexistent/replay", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Fatalf("expected 404, got %d", w.Code)
}
}
func TestReplaySuccess(t *testing.T) {
// Start a local sink server that echoes back what it receives.
sink := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("X-Sink", "yes")
w.WriteHeader(http.StatusCreated)
w.Write([]byte(`{"received": true}`))
}))
defer sink.Close()
repo := newFakeRepo()
repo.CreateEndpoint(nil, "ep")
repo.AppendRequest(nil, "ep", &store.Request{
Method: "POST",
Path: "/webhook",
Headers: `{"Content-Type":"application/json","X-Custom":"val"}`,
Query: "foo=bar",
Body: []byte(`{"msg":"hello"}`),
})
reqID := repo.requests[0].ID
// Set env to allow private replay since httptest server is on loopback.
t.Setenv("HATCH_ALLOW_PRIVATE_REPLAY", "true")
r := testRouter(repo)
body := `{"target_url": "` + sink.URL + `"}`
req := httptest.NewRequest(http.MethodPost, "/e/ep/requests/"+reqID+"/replay", strings.NewReader(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
}
var resp replayResponse
if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
t.Fatalf("failed to decode response: %v", err)
}
if resp.Status != 201 {
t.Errorf("expected status 201, got %d", resp.Status)
}
if resp.Headers["X-Sink"] != "yes" {
t.Errorf("expected X-Sink header, got %v", resp.Headers)
}
if resp.Body != `{"received": true}` {
t.Errorf("expected body, got %q", resp.Body)
}
}
func TestReplayE2ECaptureThenReplay(t *testing.T) {
// E2E: capture a request, then replay it to a sink, verify the sink received it.
// Step 1: Start a sink.
sink := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(map[string]interface{}{
"method": r.Method,
"path": r.URL.Path,
"query": r.URL.RawQuery,
"body": "received",
})
}))
defer sink.Close()
// Step 2: Capture a request.
repo := newFakeRepo()
rt := testRouter(repo)
captureReq := httptest.NewRequest("POST", "/e2e-test", strings.NewReader(`{"order":"1"}`))
captureReq.Header.Set("X-Test", "hello")
captureW := httptest.NewRecorder()
rt.ServeHTTP(captureW, captureReq)
if len(repo.requests) != 1 {
t.Fatalf("expected 1 captured request, got %d", len(repo.requests))
}
captured := repo.requests[0]
if captured.Method != "POST" {
t.Errorf("captured method: %s", captured.Method)
}
// Step 3: Replay it to the sink.
t.Setenv("HATCH_ALLOW_PRIVATE_REPLAY", "true")
replayBody := `{"target_url": "` + sink.URL + `"}`
replayReq := httptest.NewRequest("POST", "/e/e2e-test/requests/"+captured.ID+"/replay", strings.NewReader(replayBody))
replayReq.Header.Set("Content-Type", "application/json")
replayW := httptest.NewRecorder()
rt.ServeHTTP(replayW, replayReq)
if replayW.Code != http.StatusOK {
t.Fatalf("replay failed with %d: %s", replayW.Code, replayW.Body.String())
}
var resp replayResponse
json.NewDecoder(replayW.Body).Decode(&resp)
if resp.Status != 200 {
t.Errorf("replay response status: %d", resp.Status)
}
}
func TestIsPrivateAddr(t *testing.T) {
tests := []struct {
addr string
want bool
}{
{"localhost:8080", true},
{"127.0.0.1:9090", true},
{"[::1]:80", true},
{"10.0.0.5:443", true},
{"172.16.0.1:80", true},
{"192.168.1.1:3000", true},
{"0.0.0.0:8080", true},
{"example.com:443", false},
{"93.184.216.34:80", false},
{"8.8.8.8:53", false},
}
for _, tc := range tests {
got := isPrivateAddr(tc.addr)
if got != tc.want {
t.Errorf("isPrivateAddr(%q) = %v, want %v", tc.addr, tc.want, got)
}
}
}

101
internal/handler/sse.go Normal file
View File

@ -0,0 +1,101 @@
package handler
import (
"encoding/json"
"fmt"
"net/http"
"sync"
"github.com/elfoundation/hatch/internal/store"
)
// sseHub manages SSE subscribers per endpoint.
type sseHub struct {
mu sync.RWMutex
subs map[string]map[chan []byte]struct{}
}
var hub = &sseHub{
subs: make(map[string]map[chan []byte]struct{}),
}
func (h *sseHub) subscribe(endpointID string) chan []byte {
ch := make(chan []byte, 64)
h.mu.Lock()
defer h.mu.Unlock()
if h.subs[endpointID] == nil {
h.subs[endpointID] = make(map[chan []byte]struct{})
}
h.subs[endpointID][ch] = struct{}{}
return ch
}
func (h *sseHub) unsubscribe(endpointID string, ch chan []byte) {
h.mu.Lock()
defer h.mu.Unlock()
delete(h.subs[endpointID], ch)
if len(h.subs[endpointID]) == 0 {
delete(h.subs, endpointID)
}
}
func (h *sseHub) broadcast(endpointID string, req *store.Request) {
h.mu.RLock()
defer h.mu.RUnlock()
subs := h.subs[endpointID]
if subs == nil {
return
}
data, err := json.Marshal(req)
if err != nil {
return
}
for ch := range subs {
select {
case ch <- data:
default:
}
}
}
// broadcastRequest publishes req to SSE subscribers for its endpoint.
func broadcastRequest(endpointID string, req *store.Request) {
hub.broadcast(endpointID, req)
}
// HandleSSE serves an SSE stream for GET /e/{endpointID}/events.
func HandleSSE(repo store.Repository) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
endpointID := r.PathValue("endpointID")
if endpointID == "" {
http.Error(w, "missing endpoint ID", http.StatusBadRequest)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming not supported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.WriteHeader(http.StatusOK)
ch := hub.subscribe(endpointID)
defer hub.unsubscribe(endpointID, ch)
ctx := r.Context()
for {
select {
case <-ctx.Done():
return
case data, ok := <-ch:
if !ok {
return
}
fmt.Fprintf(w, "data: %s\n\n", data)
flusher.Flush()
}
}
}
}

View File

@ -20,11 +20,22 @@ type Request struct {
CreatedAt string `json:"created_at"` CreatedAt string `json:"created_at"`
} }
// MockConfig holds the mock response configuration for an endpoint.
type MockConfig struct {
EndpointID string `json:"endpoint_id"`
Status int `json:"status"`
Headers map[string]string `json:"headers"`
Body []byte `json:"body"`
}
type Repository interface { type Repository interface {
CreateEndpoint(ctx context.Context, url string) (*Endpoint, error) CreateEndpoint(ctx context.Context, url string) (*Endpoint, error)
GetEndpoint(ctx context.Context, id string) (*Endpoint, error) GetEndpoint(ctx context.Context, id string) (*Endpoint, error)
AppendRequest(ctx context.Context, endpointID string, req *Request) error AppendRequest(ctx context.Context, endpointID string, req *Request) error
GetRequest(ctx context.Context, id string) (*Request, error)
ListRequests(ctx context.Context, endpointID string, limit int) ([]*Request, error) ListRequests(ctx context.Context, endpointID string, limit int) ([]*Request, error)
GetMock(ctx context.Context, endpointID string) (*MockConfig, error)
SetMock(ctx context.Context, mock *MockConfig) error
Close() error Close() error
} }

View File

@ -2,7 +2,10 @@ CREATE TABLE IF NOT EXISTS endpoints (
id TEXT NOT NULL PRIMARY KEY, id TEXT NOT NULL PRIMARY KEY,
url TEXT NOT NULL UNIQUE, url TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL, created_at TEXT NOT NULL,
updated_at TEXT NOT NULL updated_at TEXT NOT NULL,
mock_status INTEGER,
mock_headers TEXT,
mock_body BLOB
); );
CREATE TABLE IF NOT EXISTS requests ( CREATE TABLE IF NOT EXISTS requests (
id TEXT NOT NULL PRIMARY KEY, id TEXT NOT NULL PRIMARY KEY,

View File

@ -3,15 +3,19 @@ package store
import ( import (
"context" "context"
"database/sql" "database/sql"
"encoding/json"
"fmt" "fmt"
"time" "time"
"github.com/google/uuid" "github.com/google/uuid"
) )
type sqliteRepo struct{ db *sql.DB } type sqliteRepo struct{ db *sql.DB }
func NewSQLiteRepo(db *sql.DB) (Repository, error) { func NewSQLiteRepo(db *sql.DB) (Repository, error) {
if db == nil { return nil, fmt.Errorf("store: nil db") } if db == nil {
return nil, fmt.Errorf("store: nil db")
}
return &sqliteRepo{db: db}, nil return &sqliteRepo{db: db}, nil
} }
@ -19,7 +23,9 @@ func (r *sqliteRepo) CreateEndpoint(ctx context.Context, url string) (*Endpoint,
now := utcNow() now := utcNow()
e := &Endpoint{ID: url, URL: url, CreatedAt: now, UpdatedAt: now} e := &Endpoint{ID: url, URL: url, CreatedAt: now, UpdatedAt: now}
_, err := r.db.ExecContext(ctx, `INSERT INTO endpoints (id, url, created_at, updated_at) VALUES (?, ?, ?, ?)`, e.ID, e.URL, e.CreatedAt, e.UpdatedAt) _, err := r.db.ExecContext(ctx, `INSERT INTO endpoints (id, url, created_at, updated_at) VALUES (?, ?, ?, ?)`, e.ID, e.URL, e.CreatedAt, e.UpdatedAt)
if err != nil { return nil, fmt.Errorf("store: create endpoint: %w", err) } if err != nil {
return nil, fmt.Errorf("store: create endpoint: %w", err)
}
return e, nil return e, nil
} }
@ -38,14 +44,29 @@ func (r *sqliteRepo) AppendRequest(ctx context.Context, endpointID string, req *
req.CreatedAt = utcNow() req.CreatedAt = utcNow()
_, err := r.db.ExecContext(ctx, `INSERT INTO requests (id, endpoint_id, method, path, headers, query, body, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, _, err := r.db.ExecContext(ctx, `INSERT INTO requests (id, endpoint_id, method, path, headers, query, body, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
req.ID, req.EndpointID, req.Method, req.Path, req.Headers, req.Query, req.Body, req.CreatedAt) req.ID, req.EndpointID, req.Method, req.Path, req.Headers, req.Query, req.Body, req.CreatedAt)
if err != nil { return fmt.Errorf("store: append request: %w", err) } if err != nil {
return fmt.Errorf("store: append request: %w", err)
}
return nil return nil
} }
func (r *sqliteRepo) GetRequest(ctx context.Context, id string) (*Request, error) {
row := r.db.QueryRowContext(ctx, `SELECT id, endpoint_id, method, path, headers, query, body, created_at FROM requests WHERE id = ?`, id)
req := &Request{}
if err := row.Scan(&req.ID, &req.EndpointID, &req.Method, &req.Path, &req.Headers, &req.Query, &req.Body, &req.CreatedAt); err != nil {
return nil, fmt.Errorf("store: get request %s: %w", id, err)
}
return req, nil
}
func (r *sqliteRepo) ListRequests(ctx context.Context, endpointID string, limit int) ([]*Request, error) { func (r *sqliteRepo) ListRequests(ctx context.Context, endpointID string, limit int) ([]*Request, error) {
if limit <= 0 { limit = 50 } if limit <= 0 {
limit = 50
}
rows, err := r.db.QueryContext(ctx, `SELECT id, endpoint_id, method, path, headers, query, body, created_at FROM requests WHERE endpoint_id = ? ORDER BY created_at DESC LIMIT ?`, endpointID, limit) rows, err := r.db.QueryContext(ctx, `SELECT id, endpoint_id, method, path, headers, query, body, created_at FROM requests WHERE endpoint_id = ? ORDER BY created_at DESC LIMIT ?`, endpointID, limit)
if err != nil { return nil, fmt.Errorf("store: list requests: %w", err) } if err != nil {
return nil, fmt.Errorf("store: list requests: %w", err)
}
defer rows.Close() defer rows.Close()
var out []*Request var out []*Request
for rows.Next() { for rows.Next() {
@ -58,6 +79,43 @@ func (r *sqliteRepo) ListRequests(ctx context.Context, endpointID string, limit
return out, rows.Err() return out, rows.Err()
} }
func (r *sqliteRepo) GetMock(ctx context.Context, endpointID string) (*MockConfig, error) {
row := r.db.QueryRowContext(ctx, `SELECT mock_status, mock_headers, mock_body FROM endpoints WHERE id = ?`, endpointID)
var status sql.NullInt64
var headersJSON sql.NullString
var body []byte
if err := row.Scan(&status, &headersJSON, &body); err != nil {
return nil, fmt.Errorf("store: get mock: %w", err)
}
if !status.Valid {
return nil, fmt.Errorf("store: no mock configured for %s", endpointID)
}
m := &MockConfig{
EndpointID: endpointID,
Status: int(status.Int64),
Body: body,
}
if headersJSON.Valid && headersJSON.String != "" {
json.Unmarshal([]byte(headersJSON.String), &m.Headers)
}
return m, nil
}
func (r *sqliteRepo) SetMock(ctx context.Context, mock *MockConfig) error {
headersJSON, err := json.Marshal(mock.Headers)
if err != nil {
return fmt.Errorf("store: marshal mock headers: %w", err)
}
_, err = r.db.ExecContext(ctx,
`UPDATE endpoints SET mock_status = ?, mock_headers = ?, mock_body = ? WHERE id = ?`,
mock.Status, string(headersJSON), mock.Body, mock.EndpointID,
)
if err != nil {
return fmt.Errorf("store: set mock: %w", err)
}
return nil
}
func (r *sqliteRepo) Close() error { return r.db.Close() } func (r *sqliteRepo) Close() error { return r.db.Close() }
func utcNow() string { return time.Now().UTC().Format("2006-01-02T15:04:05.000Z07:00") } func utcNow() string { return time.Now().UTC().Format("2006-01-02T15:04:05.000Z07:00") }

View File

@ -10,11 +10,19 @@ import (
func openTestRepo(t *testing.T) Repository { func openTestRepo(t *testing.T) Repository {
t.Helper() t.Helper()
db, err := sql.Open("sqlite", ":memory:?_journal_mode=WAL&_foreign_keys=on") db, err := sql.Open("sqlite", ":memory:?_journal_mode=WAL&_foreign_keys=on")
if err != nil { t.Fatalf("open in-memory sqlite: %v", err) } if err != nil {
t.Fatalf("open in-memory sqlite: %v", err)
}
db.SetMaxOpenConns(1) db.SetMaxOpenConns(1)
if err := migrate(db); err != nil { db.Close(); t.Fatalf("migrate: %v", err) } if err := migrate(db); err != nil {
db.Close()
t.Fatalf("migrate: %v", err)
}
repo, err := NewSQLiteRepo(db) repo, err := NewSQLiteRepo(db)
if err != nil { db.Close(); t.Fatalf("new sqlite repo: %v", err) } if err != nil {
db.Close()
t.Fatalf("new sqlite repo: %v", err)
}
t.Cleanup(func() { repo.Close() }) t.Cleanup(func() { repo.Close() })
return repo return repo
} }
@ -22,53 +30,158 @@ func openTestRepo(t *testing.T) Repository {
func TestCreateAndGetEndpoint(t *testing.T) { func TestCreateAndGetEndpoint(t *testing.T) {
repo := openTestRepo(t) repo := openTestRepo(t)
e, err := repo.CreateEndpoint(context.Background(), "test-one") e, err := repo.CreateEndpoint(context.Background(), "test-one")
if err != nil { t.Fatalf("CreateEndpoint: %v", err) } if err != nil {
if e.ID == "" { t.Error("expected non-empty ID") } t.Fatalf("CreateEndpoint: %v", err)
if e.URL != "test-one" { t.Errorf("url: got %q", e.URL) } }
if e.CreatedAt == "" || e.UpdatedAt == "" { t.Error("expected timestamps") } if e.ID == "" {
t.Error("expected non-empty ID")
}
if e.URL != "test-one" {
t.Errorf("url: got %q", e.URL)
}
if e.CreatedAt == "" || e.UpdatedAt == "" {
t.Error("expected timestamps")
}
got, err := repo.GetEndpoint(context.Background(), e.ID) got, err := repo.GetEndpoint(context.Background(), e.ID)
if err != nil { t.Fatalf("GetEndpoint: %v", err) } if err != nil {
if got.ID != e.ID || got.URL != e.URL { t.Errorf("got %+v", got) } t.Fatalf("GetEndpoint: %v", err)
}
if got.ID != e.ID || got.URL != e.URL {
t.Errorf("got %+v", got)
}
} }
func TestGetEndpointNotFound(t *testing.T) { func TestGetEndpointNotFound(t *testing.T) {
repo := openTestRepo(t) repo := openTestRepo(t)
_, err := repo.GetEndpoint(context.Background(), "nonexistent") _, err := repo.GetEndpoint(context.Background(), "nonexistent")
if err == nil { t.Fatal("expected error") } if err == nil {
t.Fatal("expected error")
}
} }
func TestCreateEndpointDuplicateURL(t *testing.T) { func TestCreateEndpointDuplicateURL(t *testing.T) {
repo := openTestRepo(t) repo := openTestRepo(t)
ctx := context.Background() ctx := context.Background()
if _, err := repo.CreateEndpoint(ctx, "dup"); err != nil { t.Fatal(err) } if _, err := repo.CreateEndpoint(ctx, "dup"); err != nil {
if _, err := repo.CreateEndpoint(ctx, "dup"); err == nil { t.Fatal("expected UNIQUE error") } t.Fatal(err)
}
if _, err := repo.CreateEndpoint(ctx, "dup"); err == nil {
t.Fatal("expected UNIQUE error")
}
} }
func TestAppendAndListRequests(t *testing.T) { func TestAppendAndListRequests(t *testing.T) {
repo := openTestRepo(t) repo := openTestRepo(t)
ctx := context.Background() ctx := context.Background()
e, err := repo.CreateEndpoint(ctx, "reqs-test") e, err := repo.CreateEndpoint(ctx, "reqs-test")
if err != nil { t.Fatal(err) } if err != nil {
t.Fatal(err)
}
r1 := &Request{Method: "POST", Path: "/webhook", Headers: `{"Content-Type":"application/json"}`, Query: "foo=bar", Body: []byte(`{"hello":"world"}`)} r1 := &Request{Method: "POST", Path: "/webhook", Headers: `{"Content-Type":"application/json"}`, Query: "foo=bar", Body: []byte(`{"hello":"world"}`)}
if err := repo.AppendRequest(ctx, e.ID, r1); err != nil { t.Fatal(err) } if err := repo.AppendRequest(ctx, e.ID, r1); err != nil {
t.Fatal(err)
}
time.Sleep(time.Millisecond) time.Sleep(time.Millisecond)
r2 := &Request{Method: "GET", Path: "/webhook", Headers: `{}`} r2 := &Request{Method: "GET", Path: "/webhook", Headers: `{}`}
if err := repo.AppendRequest(ctx, e.ID, r2); err != nil { t.Fatal(err) } if err := repo.AppendRequest(ctx, e.ID, r2); err != nil {
t.Fatal(err)
}
reqs, err := repo.ListRequests(ctx, e.ID, 10) reqs, err := repo.ListRequests(ctx, e.ID, 10)
if err != nil { t.Fatal(err) } if err != nil {
if len(reqs) != 2 { t.Fatalf("got %d requests", len(reqs)) } t.Fatal(err)
if reqs[0].Method != "GET" { t.Errorf("first: got %s", reqs[0].Method) } }
if reqs[1].Method != "POST" { t.Errorf("second: got %s", reqs[1].Method) } if len(reqs) != 2 {
t.Fatalf("got %d requests", len(reqs))
}
if reqs[0].Method != "GET" {
t.Errorf("first: got %s", reqs[0].Method)
}
if reqs[1].Method != "POST" {
t.Errorf("second: got %s", reqs[1].Method)
}
}
func TestGetRequest(t *testing.T) {
repo := openTestRepo(t)
ctx := context.Background()
e, _ := repo.CreateEndpoint(ctx, "getreq-test")
r := &Request{Method: "POST", Path: "/webhook", Headers: `{"X-Api-Key":"secret"}`, Query: "a=1", Body: []byte(`{"msg":"hi"}`)}
if err := repo.AppendRequest(ctx, e.ID, r); err != nil {
t.Fatal(err)
}
got, err := repo.GetRequest(ctx, r.ID)
if err != nil {
t.Fatalf("GetRequest: %v", err)
}
if got.ID != r.ID {
t.Errorf("id: got %q want %q", got.ID, r.ID)
}
if got.Method != "POST" {
t.Errorf("method: got %q", got.Method)
}
if string(got.Body) != `{"msg":"hi"}` {
t.Errorf("body: got %q", string(got.Body))
}
}
func TestGetRequestNotFound(t *testing.T) {
repo := openTestRepo(t)
_, err := repo.GetRequest(context.Background(), "nonexistent")
if err == nil {
t.Fatal("expected error for nonexistent request")
}
}
func TestSetAndGetMock(t *testing.T) {
repo := openTestRepo(t)
ctx := context.Background()
e, err := repo.CreateEndpoint(ctx, "mock-test")
if err != nil {
t.Fatal(err)
}
mock := &MockConfig{
EndpointID: e.ID,
Status: 201,
Headers: map[string]string{"Content-Type": "application/json"},
Body: []byte(`{"mocked": true}`),
}
if err := repo.SetMock(ctx, mock); err != nil {
t.Fatal(err)
}
got, err := repo.GetMock(ctx, e.ID)
if err != nil {
t.Fatal(err)
}
if got.Status != 201 {
t.Errorf("status: got %d", got.Status)
}
if string(got.Body) != `{"mocked": true}` {
t.Errorf("body: got %q", string(got.Body))
}
}
func TestGetMockNotFound(t *testing.T) {
repo := openTestRepo(t)
_, err := repo.GetMock(context.Background(), "no-mock")
if err == nil {
t.Fatal("expected error")
}
} }
func TestListRequestsLimit(t *testing.T) { func TestListRequestsLimit(t *testing.T) {
repo := openTestRepo(t) repo := openTestRepo(t)
ctx := context.Background() ctx := context.Background()
e, _ := repo.CreateEndpoint(ctx, "limit-test") e, _ := repo.CreateEndpoint(ctx, "limit-test")
for i := 0; i < 5; i++ { repo.AppendRequest(ctx, e.ID, &Request{Method: "GET", Path: "/"}) } for i := 0; i < 5; i++ {
repo.AppendRequest(ctx, e.ID, &Request{Method: "GET", Path: "/"})
}
reqs, err := repo.ListRequests(ctx, e.ID, 3) reqs, err := repo.ListRequests(ctx, e.ID, 3)
if err != nil { t.Fatal(err) } if err != nil {
if len(reqs) != 3 { t.Fatalf("got %d", len(reqs)) } t.Fatal(err)
}
if len(reqs) != 3 {
t.Fatalf("got %d", len(reqs))
}
} }
func TestListRequestsEmpty(t *testing.T) { func TestListRequestsEmpty(t *testing.T) {
@ -76,7 +189,9 @@ func TestListRequestsEmpty(t *testing.T) {
ctx := context.Background() ctx := context.Background()
e, _ := repo.CreateEndpoint(ctx, "empty-test") e, _ := repo.CreateEndpoint(ctx, "empty-test")
reqs, _ := repo.ListRequests(ctx, e.ID, 10) reqs, _ := repo.ListRequests(ctx, e.ID, 10)
if len(reqs) != 0 { t.Errorf("got %d", len(reqs)) } if len(reqs) != 0 {
t.Errorf("got %d", len(reqs))
}
} }
func TestAppendRequestBodyBinary(t *testing.T) { func TestAppendRequestBodyBinary(t *testing.T) {
@ -84,19 +199,31 @@ func TestAppendRequestBodyBinary(t *testing.T) {
ctx := context.Background() ctx := context.Background()
e, _ := repo.CreateEndpoint(ctx, "binary-test") e, _ := repo.CreateEndpoint(ctx, "binary-test")
body := []byte{0x00, 0x01, 0x02, 0xFF} body := []byte{0x00, 0x01, 0x02, 0xFF}
if err := repo.AppendRequest(ctx, e.ID, &Request{Method: "POST", Path: "/binary", Body: body}); err != nil { t.Fatal(err) } if err := repo.AppendRequest(ctx, e.ID, &Request{Method: "POST", Path: "/binary", Body: body}); err != nil {
t.Fatal(err)
}
reqs, _ := repo.ListRequests(ctx, e.ID, 1) reqs, _ := repo.ListRequests(ctx, e.ID, 1)
if len(reqs) != 1 { t.Fatal("expected 1 request") } if len(reqs) != 1 {
if len(reqs[0].Body) != 4 { t.Errorf("body len: %d", len(reqs[0].Body)) } t.Fatal("expected 1 request")
}
if len(reqs[0].Body) != 4 {
t.Errorf("body len: %d", len(reqs[0].Body))
}
} }
func TestMigrateIdempotent(t *testing.T) { func TestMigrateIdempotent(t *testing.T) {
db, err := sql.Open("sqlite", ":memory:?_journal_mode=WAL&_foreign_keys=on") db, err := sql.Open("sqlite", ":memory:?_journal_mode=WAL&_foreign_keys=on")
if err != nil { t.Fatal(err) } if err != nil {
t.Fatal(err)
}
defer db.Close() defer db.Close()
db.SetMaxOpenConns(1) db.SetMaxOpenConns(1)
if err := migrate(db); err != nil { t.Fatal(err) } if err := migrate(db); err != nil {
if err := migrate(db); err != nil { t.Fatal(err) } t.Fatal(err)
}
if err := migrate(db); err != nil {
t.Fatal(err)
}
var name string var name string
if err := db.QueryRow("SELECT name FROM sqlite_master WHERE type='table' AND name='endpoints'").Scan(&name); err != nil { if err := db.QueryRow("SELECT name FROM sqlite_master WHERE type='table' AND name='endpoints'").Scan(&name); err != nil {
t.Fatalf("endpoints table: %v", err) t.Fatalf("endpoints table: %v", err)