Skip to content

Commit 0dd7e7f

Browse files
robin-biallyclaude
andcommitted
Add auto (record-on-miss) mode and multi-instance support
auto mode: serves from cassette on match, forwards + records on miss. Cassette path is optional in all modes — defaults to <target-host>.json. Multi-instance: set BUFFR_0_TARGET, BUFFR_1_TARGET, … to run multiple proxies in a single process, each on its own port with its own cassette. Useful for containerized setups where the app points at multiple APIs. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent 08b827e commit 0dd7e7f

3 files changed

Lines changed: 234 additions & 11 deletions

File tree

‎cmd/buffr/main.go‎

Lines changed: 159 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -44,11 +44,19 @@ func main() {
4444
usage()
4545
os.Exit(2)
4646
}
47+
// Multi-instance mode: BUFFR_0_TARGET (and optionally BUFFR_1_TARGET, …)
48+
// takes precedence over any subcommand.
49+
if instances := loadInstances(); len(instances) > 0 {
50+
os.Exit(runInstances(instances))
51+
}
52+
4753
switch args[0] {
4854
case "record":
4955
os.Exit(runRecord(args[1:]))
5056
case "replay":
5157
os.Exit(runReplay(args[1:]))
58+
case "auto":
59+
os.Exit(runAuto(args[1:]))
5260
case "-h", "--help", "help":
5361
usage()
5462
os.Exit(0)
@@ -63,19 +71,25 @@ func usage() {
6371
fmt.Fprintln(os.Stderr, `buffr — record/replay HTTP + WebSocket proxy
6472
6573
Usage:
66-
buffr record --target <url> --port <p> --cassette <file>
67-
buffr replay --cassette <file> --port <p>
74+
buffr record --target <url> [--port <p>] [--cassette <file>]
75+
buffr replay [--cassette <file>] [--port <p>]
76+
buffr auto --target <url> [--port <p>] [--cassette <file>]
6877
6978
Environment variables (flags take precedence):
70-
BUFFR_MODE subcommand (record|replay) — used when no CLI argument is given
71-
BUFFR_TARGET upstream URL for record mode
79+
BUFFR_MODE subcommand (record|replay|auto) — used when no CLI argument is given
80+
BUFFR_TARGET upstream URL for record/auto mode
7281
BUFFR_PORT local port (default 8080)
73-
BUFFR_CASSETTE cassette file path
82+
BUFFR_CASSETTE cassette file path (auto-generated from target host if omitted)
83+
84+
Multi-instance (one process, many ports):
85+
BUFFR_0_TARGET=https://api.openai.com BUFFR_0_PORT=8081
86+
BUFFR_1_TARGET=https://api.anthropic.com BUFFR_1_PORT=8082
87+
… (BUFFR_N_CASSETTE and BUFFR_N_MODE are optional per instance)
7488
7589
Examples:
76-
buffr record --target https://api.openai.com --port 8080 --cassette session.json
90+
buffr auto --target https://api.openai.com --port 8080
7791
buffr replay --cassette session.json --port 8080
78-
BUFFR_MODE=replay BUFFR_CASSETTE=/data/session.json buffr`)
92+
BUFFR_MODE=auto BUFFR_TARGET=https://api.openai.com buffr`)
7993
}
8094

8195
// envStr returns the value of the environment variable key, or fallback if unset.
@@ -103,22 +117,23 @@ func runRecord(args []string) int {
103117
port := fs.Int("port", envInt("BUFFR_PORT", 8080), "local port to listen on [BUFFR_PORT]")
104118
cass := fs.String("cassette", envStr("BUFFR_CASSETTE", ""), "cassette file to write (required) [BUFFR_CASSETTE]")
105119
_ = fs.Parse(args)
106-
if *target == "" || *cass == "" {
107-
fmt.Fprintln(os.Stderr, "record: --target and --cassette are required")
120+
if *target == "" {
121+
fmt.Fprintln(os.Stderr, "record: --target is required")
108122
return 2
109123
}
110124
u, err := url.Parse(*target)
111125
if err != nil || u.Scheme == "" || u.Host == "" {
112126
fmt.Fprintf(os.Stderr, "record: invalid --target %q\n", *target)
113127
return 2
114128
}
115-
rec := proxy.NewRecorder(*cass)
129+
cassPath := cassettePath(*cass, u.Host)
130+
rec := proxy.NewRecorder(cassPath)
116131
mux := http.NewServeMux()
117132
mux.Handle("/", routeUpgrade(
118133
proxy.RecordWSHandler(u, rec),
119134
proxy.RecordHandler(u, rec),
120135
))
121-
return serve(*port, mux, "record", *cass)
136+
return serve(*port, mux, "record", cassPath)
122137
}
123138

124139
func runReplay(args []string) int {
@@ -159,6 +174,139 @@ func routeUpgrade(wsHandler, httpHandler http.Handler) http.Handler {
159174
})
160175
}
161176

177+
// cassettePath returns path if non-empty, otherwise derives a filename from
178+
// the target host (e.g. "api.openai.com.json") in the current directory.
179+
func cassettePath(path, targetHost string) string {
180+
if path != "" {
181+
return path
182+
}
183+
return targetHost + ".json"
184+
}
185+
186+
func runAuto(args []string) int {
187+
fs := flag.NewFlagSet("auto", flag.ExitOnError)
188+
target := fs.String("target", envStr("BUFFR_TARGET", ""), "upstream URL to proxy to (required) [BUFFR_TARGET]")
189+
port := fs.Int("port", envInt("BUFFR_PORT", 8080), "local port to listen on [BUFFR_PORT]")
190+
cass := fs.String("cassette", envStr("BUFFR_CASSETTE", ""), "cassette file (auto-generated if omitted) [BUFFR_CASSETTE]")
191+
_ = fs.Parse(args)
192+
if *target == "" {
193+
fmt.Fprintln(os.Stderr, "auto: --target is required")
194+
return 2
195+
}
196+
u, err := url.Parse(*target)
197+
if err != nil || u.Scheme == "" || u.Host == "" {
198+
fmt.Fprintf(os.Stderr, "auto: invalid --target %q\n", *target)
199+
return 2
200+
}
201+
cassPath := cassettePath(*cass, u.Host)
202+
rec, existing := proxy.NewAutoRecorder(cassPath)
203+
m := matcher.New(existing, matcher.JSONBodyNormalizer)
204+
rep := proxy.NewWSReplayer(existing)
205+
mux := http.NewServeMux()
206+
mux.Handle("/", routeUpgrade(
207+
proxy.AutoWSHandler(u, rec, rep),
208+
proxy.AutoHandler(u, rec, m),
209+
))
210+
return serve(*port, mux, "auto", cassPath)
211+
}
212+
213+
// instanceConfig holds the parsed configuration for one proxy instance.
214+
type instanceConfig struct {
215+
mode string
216+
target *url.URL
217+
port int
218+
cassette string
219+
}
220+
221+
// loadInstances reads BUFFR_0_TARGET, BUFFR_1_TARGET, … from the environment
222+
// and returns one instanceConfig per indexed group found. Returns nil when no
223+
// indexed variables are set, signalling single-instance mode.
224+
func loadInstances() []instanceConfig {
225+
var out []instanceConfig
226+
for i := 0; ; i++ {
227+
pfx := fmt.Sprintf("BUFFR_%d_", i)
228+
rawTarget := os.Getenv(pfx + "TARGET")
229+
if rawTarget == "" {
230+
break
231+
}
232+
u, err := url.Parse(rawTarget)
233+
if err != nil || u.Scheme == "" || u.Host == "" {
234+
fmt.Fprintf(os.Stderr, "buffr: invalid %sTARGET=%q, skipping\n", pfx, rawTarget)
235+
continue
236+
}
237+
mode := os.Getenv(pfx + "MODE")
238+
if mode == "" {
239+
mode = "auto"
240+
}
241+
port := envInt(pfx+"PORT", 8080+i)
242+
cass := cassettePath(os.Getenv(pfx+"CASSETTE"), u.Host)
243+
out = append(out, instanceConfig{mode: mode, target: u, port: port, cassette: cass})
244+
}
245+
return out
246+
}
247+
248+
// runInstances starts one HTTP server per instanceConfig in separate goroutines
249+
// and blocks until SIGINT/SIGTERM, then shuts them all down gracefully.
250+
func runInstances(instances []instanceConfig) int {
251+
var servers []*http.Server
252+
for _, cfg := range instances {
253+
cfg := cfg
254+
var mux *http.ServeMux
255+
switch cfg.mode {
256+
case "record":
257+
rec := proxy.NewRecorder(cfg.cassette)
258+
mux = http.NewServeMux()
259+
mux.Handle("/", routeUpgrade(
260+
proxy.RecordWSHandler(cfg.target, rec),
261+
proxy.RecordHandler(cfg.target, rec),
262+
))
263+
case "replay":
264+
c, err := cassette.Load(cfg.cassette)
265+
if err != nil {
266+
fmt.Fprintf(os.Stderr, "buffr: failed to load cassette %s: %v\n", cfg.cassette, err)
267+
return 1
268+
}
269+
m := matcher.New(c, matcher.JSONBodyNormalizer)
270+
rep := proxy.NewWSReplayer(c)
271+
mux = http.NewServeMux()
272+
mux.Handle("/", routeUpgrade(
273+
proxy.ReplayWSHandler(rep),
274+
proxy.ReplayHandler(m),
275+
))
276+
default: // "auto"
277+
rec, existing := proxy.NewAutoRecorder(cfg.cassette)
278+
m := matcher.New(existing, matcher.JSONBodyNormalizer)
279+
rep := proxy.NewWSReplayer(existing)
280+
mux = http.NewServeMux()
281+
mux.Handle("/", routeUpgrade(
282+
proxy.AutoWSHandler(cfg.target, rec, rep),
283+
proxy.AutoHandler(cfg.target, rec, m),
284+
))
285+
}
286+
addr := fmt.Sprintf(":%d", cfg.port)
287+
srv := &http.Server{Addr: addr, Handler: mux}
288+
servers = append(servers, srv)
289+
go func() {
290+
fmt.Fprintf(os.Stderr, "buffr %s on %s → target=%s cassette=%s\n",
291+
cfg.mode, addr, cfg.target, cfg.cassette)
292+
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
293+
fmt.Fprintf(os.Stderr, "buffr: server error on %s: %v\n", addr, err)
294+
}
295+
}()
296+
}
297+
298+
stop := make(chan os.Signal, 1)
299+
signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
300+
<-stop
301+
fmt.Fprintln(os.Stderr, "buffr: shutting down…")
302+
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
303+
defer cancel()
304+
for _, srv := range servers {
305+
_ = srv.Shutdown(ctx)
306+
}
307+
return 0
308+
}
309+
162310
func serve(port int, h http.Handler, mode, cass string) int {
163311
addr := fmt.Sprintf(":%d", port)
164312
srv := &http.Server{Addr: addr, Handler: h}

‎internal/proxy/http.go‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,67 @@ func (r *Recorder) Append(it cassette.Interaction) error {
4242
return cassette.Save(r.path, r.cassette)
4343
}
4444

45+
// NewAutoRecorder creates a Recorder pre-seeded from an existing cassette file.
46+
// If the file does not exist yet, an empty cassette is used. The returned
47+
// cassette snapshot can be handed to matcher.New so the auto handler can
48+
// match against already-recorded interactions.
49+
func NewAutoRecorder(path string) (*Recorder, *cassette.Cassette) {
50+
c, err := cassette.Load(path)
51+
if err != nil {
52+
c = &cassette.Cassette{Version: cassette.CurrentVersion}
53+
}
54+
return &Recorder{cassette: c, path: path}, c
55+
}
56+
57+
// AutoHandler is the record-on-miss handler: it serves from the cassette when
58+
// a match exists, and falls back to forwarding + recording when it does not.
59+
// This lets tests run fully offline once every interaction has been captured,
60+
// while still accumulating new cassette entries transparently on first run.
61+
func AutoHandler(target *url.URL, rec *Recorder, m *matcher.Matcher) http.Handler {
62+
record := RecordHandler(target, rec)
63+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
64+
body, err := io.ReadAll(r.Body)
65+
if err != nil {
66+
http.Error(w, "buffr: failed to read request body: "+err.Error(), http.StatusInternalServerError)
67+
return
68+
}
69+
r.Body.Close()
70+
71+
ex := m.Take(r.Method, r.URL.Path, string(body))
72+
if ex == nil {
73+
// Cache miss — forward to upstream and record.
74+
r.Body = io.NopCloser(bytes.NewReader(body))
75+
record.ServeHTTP(w, r)
76+
return
77+
}
78+
79+
// Cache hit — replay from cassette.
80+
for k, vs := range ex.Response.Headers {
81+
for _, v := range vs {
82+
w.Header().Add(k, v)
83+
}
84+
}
85+
w.WriteHeader(ex.Response.Status)
86+
flusher, _ := w.(http.Flusher)
87+
88+
if len(ex.Response.BodyChunks) > 0 {
89+
for _, c := range ex.Response.BodyChunks {
90+
if c.DelayMs > 0 {
91+
time.Sleep(time.Duration(c.DelayMs) * time.Millisecond)
92+
}
93+
if _, werr := w.Write([]byte(c.Data)); werr != nil {
94+
return
95+
}
96+
if flusher != nil {
97+
flusher.Flush()
98+
}
99+
}
100+
return
101+
}
102+
_, _ = w.Write([]byte(ex.Response.Body))
103+
})
104+
}
105+
45106
// RecordHandler returns an http.Handler that proxies every request to `target`
46107
// and writes the round-trip to the recorder. It supports `text/event-stream`
47108
// responses by streaming chunks through and capturing each chunk with the

‎internal/proxy/websocket.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,20 @@ var upgrader = websocket.Upgrader{
2727
HandshakeTimeout: 30 * time.Second,
2828
}
2929

30+
// AutoWSHandler replays the next recorded WebSocket session when one is
31+
// available, and falls back to forwarding + recording when the cassette is
32+
// exhausted. Matches the record-on-miss semantics of AutoHandler.
33+
func AutoWSHandler(target *url.URL, rec *Recorder, rep *wsReplayer) http.Handler {
34+
record := RecordWSHandler(target, rec)
35+
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
36+
if rep.Remaining() > 0 {
37+
ReplayWSHandler(rep).ServeHTTP(w, r)
38+
return
39+
}
40+
record.ServeHTTP(w, r)
41+
})
42+
}
43+
3044
// RecordWSHandler proxies WebSocket connections to the upstream target and
3145
// records every frame in both directions to the recorder.
3246
//

0 commit comments

Comments
 (0)