mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
the engine hands a composer a zip app, not a handler
A composer had one way to mount the engine: take HTTPHandler and adapt it, which puts sixty-odd operations behind one opaque wildcard the composing router cannot see into. Surface registers them as real routes, so app.Use(engine.Surface()) puts every address in the composed router and in every projection that reads it. Both renderings reach the same functions -- a call in, an answer out -- so the two transports cannot describe one operation differently, and HTTPHandler is unchanged for callers that have it. The wire is held by a test rather than by intent. TestSurfaceAnswersWhatTheMuxAnswers drives the same requests through both and compares status, body, Content-Type and Location. It replaced a probe that printed the two answers and asserted nothing, and the first thing it found was real: http.ServeMux tidies a path with an empty or dot segment and redirects to the tidied form before matching, so /namespaces//workflows was a redirect through the mux and a 404 here. The wildcard answers that redirect now, with the body http.Redirect writes and only for the methods it writes one for.
This commit is contained in:
@@ -18,6 +18,7 @@ require (
|
||||
github.com/luxfi/log v1.5.0
|
||||
github.com/luxfi/zap v1.2.7
|
||||
github.com/robfig/cron/v3 v3.0.1
|
||||
github.com/zap-proto/zip v1.36.9
|
||||
golang.org/x/time v0.15.0
|
||||
)
|
||||
|
||||
@@ -45,7 +46,7 @@ require (
|
||||
github.com/luxfi/math v1.5.1 // indirect
|
||||
github.com/luxfi/math/big v0.1.0 // indirect
|
||||
github.com/luxfi/mdns v0.1.1 // indirect
|
||||
github.com/luxfi/metric v1.8.1 // indirect
|
||||
github.com/luxfi/metric v1.10.0 // indirect
|
||||
github.com/luxfi/mock v0.1.1 // indirect
|
||||
github.com/luxfi/p2p v1.22.1 // indirect
|
||||
github.com/luxfi/sampler v1.1.0 // indirect
|
||||
@@ -72,14 +73,25 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/dop251/goja v0.0.0-20260627200808-0b76000cabdb // indirect
|
||||
github.com/andybalholm/brotli v1.2.1 // indirect
|
||||
github.com/gofiber/schema v1.7.1 // indirect
|
||||
github.com/gofiber/utils/v2 v2.0.4 // indirect
|
||||
github.com/hanzoai/csqlite v0.1.0 // indirect
|
||||
github.com/hanzoai/sqlcipher v0.1.1 // indirect
|
||||
github.com/luxfi/bft v0.1.5 // indirect
|
||||
github.com/luxfi/pq v1.1.0 // indirect
|
||||
github.com/mattn/go-colorable v0.1.15 // indirect
|
||||
github.com/mattn/go-isatty v0.0.22 // indirect
|
||||
github.com/philhofer/fwd v1.2.0 // indirect
|
||||
github.com/tinylib/msgp v1.6.4 // indirect
|
||||
github.com/valyala/bytebufferpool v1.0.0 // indirect
|
||||
github.com/valyala/fasthttp v1.70.0 // indirect
|
||||
github.com/zap-proto/fiber/v3 v3.2.1 // indirect
|
||||
github.com/zap-proto/go v1.3.0 // indirect
|
||||
github.com/zap-proto/http v0.3.5 // indirect
|
||||
github.com/zap-proto/mcp v1.0.5 // indirect
|
||||
golang.org/x/net v0.56.0 // indirect
|
||||
golang.org/x/sys v0.47.0 // indirect
|
||||
golang.org/x/text v0.40.0 // indirect
|
||||
modernc.org/sqlite v1.51.0 // indirect
|
||||
)
|
||||
|
||||
@@ -12,17 +12,12 @@ github.com/decred/dcrd/crypto/blake256 v1.1.0 h1:zPMNGQCm0g4QTY27fOCorQW7EryeQ/U
|
||||
github.com/decred/dcrd/crypto/blake256 v1.1.0/go.mod h1:2OfgNZ5wDpcsFmHmCK5gZTPcCXqlm2ArzUIkw9czNJo=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 h1:5RVFMOWjMyRy8cARdy79nAmgYw3hK/4HUq48LQ6Wwqo=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40=
|
||||
github.com/dlclark/regexp2 v1.11.5 h1:Q/sSnsKerHeCkc/jSTNq1oCm7KiVgUMZRDUoRu0JQZQ=
|
||||
github.com/dlclark/regexp2/v2 v2.2.1 h1:mf4KkFUj0gJuarK8P+LgiS+Lit7m9N1yAwEfPbee7R0=
|
||||
github.com/dlclark/regexp2/v2 v2.2.1/go.mod h1:avUrQvPaLz2DrFNHJF0taWAFFX2C1GMSSoeiqFjcBmU=
|
||||
github.com/dop251/goja v0.0.0-20260627200808-0b76000cabdb h1:qxbH9q0zvt7Fwy9UJrHdd9grPTlO07G/vUUE3eqfZ4w=
|
||||
github.com/dop251/goja v0.0.0-20260627200808-0b76000cabdb/go.mod h1:Sc+QOu1WruvaaeT/cxFez/pXHpI9ZDjg/E8QNfSVveI=
|
||||
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/evanw/esbuild v0.28.1 h1:ds+yuRyUaZGx++GR56CrCeuXh8PVhVM4xq8v7PNELFc=
|
||||
github.com/evanw/esbuild v0.28.1/go.mod h1:D2vIQZqV/vIf/VRHtViaUtViZmG7o+kKmlBfVQuRi48=
|
||||
github.com/go-sourcemap/sourcemap v2.1.4+incompatible h1:a+iTbH5auLKxaNwQFg0B+TCYl6lbukKPc7b5x0n1s6Q=
|
||||
github.com/go-sourcemap/sourcemap v2.1.4+incompatible/go.mod h1:F8jJfvm2KbVjc5NqelyYJmf/v5J0dwNLS2mL4sNA1Jg=
|
||||
github.com/fasthttp/websocket v1.5.12 h1:e4RGPpWW2HTbL3zV0Y/t7g0ub294LkiuXXUuTOUInlE=
|
||||
github.com/fasthttp/websocket v1.5.12/go.mod h1:I+liyL7/4moHojiOgUOIKEWm9EIxHqxZChS+aMFltyg=
|
||||
github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ=
|
||||
github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
|
||||
github.com/gofiber/schema v1.7.1 h1:oSJBKdgP8JeIME4TQSAqlNKTU2iBB+2RNmKi8Nsc+TI=
|
||||
github.com/gofiber/schema v1.7.1/go.mod h1:A/X5Ffyru4p9eBdp99qu+nzviHzQiZ7odLT+TwxWhbk=
|
||||
github.com/gofiber/utils/v2 v2.0.4 h1:WwAxUA7L4MW2DjdEHF234lfqvBqd2vYYuBtA9TJq2ec=
|
||||
@@ -91,8 +86,8 @@ github.com/luxfi/math/big v0.1.0 h1:Vz4c0RsZVPdIKPsHPgAJChH/R3p15WHRUz7LkLf+NIQ=
|
||||
github.com/luxfi/math/big v0.1.0/go.mod h1:BuxSu22RbO93xBLk5Eam5nldFponoJ73xDFz4uJ3Huk=
|
||||
github.com/luxfi/mdns v0.1.1 h1:g2eRr9AXcziPkkcd24M+Qu9ApEpoKKjfI79QSNqv0rQ=
|
||||
github.com/luxfi/mdns v0.1.1/go.mod h1:dbp5f3h3aE7CGzwbaWzBM9cwdcekhmSrWhQevgYhhNA=
|
||||
github.com/luxfi/metric v1.8.1 h1:v58GgPFAOLPVxSa/JiNLwqJQNEFHdWbXZV28piMXX4s=
|
||||
github.com/luxfi/metric v1.8.1/go.mod h1:R1OPAIeW4UBW3osK7j2r3/XPmczfNRFTXg4bnlemTuE=
|
||||
github.com/luxfi/metric v1.10.0 h1:Ca4N6HqukMoZ1JUx3t5BJ4cjyY3HL+HnF9kaGEAAUfA=
|
||||
github.com/luxfi/metric v1.10.0/go.mod h1:JhVJPOM10FEpGSSaP8QrGLi7w0D3pjJ2kwWwmGi+9YI=
|
||||
github.com/luxfi/mock v0.1.1 h1:0HEtIjg1J6CWz+IUyP6rsGqNWTcmxjFnSQIhaDuARwY=
|
||||
github.com/luxfi/mock v0.1.1/go.mod h1:jo35akl3Vtd8LbzDts8VJ0jmSVycrd1/eBi6g6t5hKU=
|
||||
github.com/luxfi/p2p v1.22.1 h1:M59Iy+FIJta99aFfTpG6EE70jm9uqIX+iHrGBYPHDhg=
|
||||
@@ -134,6 +129,10 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
|
||||
github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
|
||||
github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
|
||||
github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38 h1:D0vL7YNisV2yqE55+q0lFuGse6U8lxlg7fYTctlT5Gc=
|
||||
github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38/go.mod h1:sM7Mt7uEoCeFSCBM+qBrqvEo+/9vdmj19wzp3yzUhmg=
|
||||
github.com/shamaton/msgpack/v3 v3.1.0 h1:jsk0vEAqVvvS9+fTZ5/EcQ9tz860c9pWxJ4Iwecz8gU=
|
||||
github.com/shamaton/msgpack/v3 v3.1.0/go.mod h1:DcQG8jrdrQCIxr3HlMYkiXdMhK+KfN2CitkyzsQV4uc=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
github.com/supranational/blst v0.3.16 h1:bTDadT+3fK497EvLdWRQEjiGnUtzJ7jjIUMF0jqwYhE=
|
||||
@@ -144,14 +143,20 @@ github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6Kllzaw
|
||||
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
|
||||
github.com/valyala/fasthttp v1.70.0 h1:LAhMGcWk13QZWm85+eg8ZBNbrq5mnkWFGbHMUJHIdXA=
|
||||
github.com/valyala/fasthttp v1.70.0/go.mod h1:oDZEHHkJ/Buyklg6uURmYs19442zFSnCIfX3j1FY3pE=
|
||||
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
|
||||
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
|
||||
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
|
||||
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
|
||||
github.com/zap-proto/fiber/v3 v3.2.1 h1:k45oKyTwySPtGt8sPz2Ao8OUHc7pDEhai8Np2Ym6Jbg=
|
||||
github.com/zap-proto/fiber/v3 v3.2.1/go.mod h1:eDm2z+ufJrkuE4MeX0Mea4oc/p7/HpXiZjVB+BXCKOA=
|
||||
github.com/zap-proto/go v1.3.0 h1:S3rMoawwhH/BbSZ4G8zG05hJoQnMSMDPzIq75diCTqE=
|
||||
github.com/zap-proto/go v1.3.0/go.mod h1:914SNGTH6Rv3Yu1MweWJBPEN8FZlo5C39QyhaB0C7Q0=
|
||||
github.com/zap-proto/http v0.3.1 h1:A2rCPWYCX866eAsdiWuns0dvWnBmViZtGm4pwX7jwlY=
|
||||
github.com/zap-proto/http v0.3.1/go.mod h1:UYfGhDDCetgxs65XSev8Lpf65COg5vKQK+cWwZGh4zQ=
|
||||
github.com/zap-proto/zip v1.18.1 h1:DWsp9UtPwfijiRtEut7G8grex3Qprm/1c2G5TYWSpFc=
|
||||
github.com/zap-proto/zip v1.18.1/go.mod h1:EKMmUX9wCPvpkhpMBQRqa17YVXNXRYEjj7X+65Y+J9E=
|
||||
github.com/zap-proto/http v0.3.5 h1:ezSuUbF05jxFGpfmScBnzIVTeH1ZGhcDSm7d5yL8Byc=
|
||||
github.com/zap-proto/http v0.3.5/go.mod h1:UYfGhDDCetgxs65XSev8Lpf65COg5vKQK+cWwZGh4zQ=
|
||||
github.com/zap-proto/mcp v1.0.5 h1:1Hi0KNi4YlWW2zkQUMDK9n3jP3wZaRJ1rseMW9p9F0c=
|
||||
github.com/zap-proto/mcp v1.0.5/go.mod h1:cYOsq+33NyW4X8nUVtIs/qtezVovYazw1s0V7mrQBYA=
|
||||
github.com/zap-proto/zip v1.36.9 h1:8jQH9uolR4UaDzxSrEFrDGkFyTmMjRwrbukJ2KvH6R8=
|
||||
github.com/zap-proto/zip v1.36.9/go.mod h1:k0CQcj6Ro+gfNdREF3WOc9/KeXidzI0WPVnzxUjVR/s=
|
||||
go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
|
||||
go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
// Copyright © 2026 Hanzo AI. MIT License.
|
||||
|
||||
// One request as an operation reads it, one answer as an operation writes it.
|
||||
//
|
||||
// Every operation on the engine's surface is a function from a call to an
|
||||
// answer, holding no transport type of its own. Two renderers put an answer on
|
||||
// a wire — one onto a net/http response (HTTPHandler), one onto a zip response
|
||||
// (Surface) — so the same body serves both and the two cannot describe the same
|
||||
// operation differently.
|
||||
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/url"
|
||||
|
||||
"github.com/zap-proto/zip"
|
||||
)
|
||||
|
||||
// call is what an operation reads: the caller's identity context, the method,
|
||||
// the parsed query and the raw body. The path is not here — an operation is
|
||||
// reached by its route and receives the segments it needs as arguments.
|
||||
type call struct {
|
||||
ctx context.Context
|
||||
method string
|
||||
query url.Values
|
||||
body []byte
|
||||
// unread is the transport's failure to deliver a body. It reaches the
|
||||
// operation through decode, which is where a body that never arrived was
|
||||
// always reported.
|
||||
unread error
|
||||
}
|
||||
|
||||
// decode parses the body as JSON into v. An empty body is not an error: many
|
||||
// operations accept one and act on the zero value.
|
||||
func (rq call) decode(v any) error {
|
||||
if rq.unread != nil {
|
||||
return rq.unread
|
||||
}
|
||||
if len(rq.body) == 0 {
|
||||
return nil
|
||||
}
|
||||
return json.Unmarshal(rq.body, v)
|
||||
}
|
||||
|
||||
// stream parses the body as a JSON stream into v, so an EMPTY body reports
|
||||
// io.EOF instead of leaving v at its zero value. One operation reads its body
|
||||
// that way and refuses on it.
|
||||
func (rq call) stream(v any) error {
|
||||
if rq.unread != nil {
|
||||
return rq.unread
|
||||
}
|
||||
return json.NewDecoder(bytes.NewReader(rq.body)).Decode(v)
|
||||
}
|
||||
|
||||
// answer is what an operation writes: a status, at most one content type and
|
||||
// the exact bytes of the body.
|
||||
type answer struct {
|
||||
status int
|
||||
ctype string
|
||||
nosniff bool
|
||||
body []byte
|
||||
}
|
||||
|
||||
const (
|
||||
ctypeJSON = "application/json"
|
||||
ctypePlain = "text/plain; charset=utf-8"
|
||||
)
|
||||
|
||||
// data answers 200 with v as JSON, or the engine's 500 envelope when err is
|
||||
// non-nil — the shape almost every read and write on this surface returns.
|
||||
func data(err error, v any) answer {
|
||||
if err != nil {
|
||||
return fault(http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
return answer{status: http.StatusOK, ctype: ctypeJSON, body: render(v)}
|
||||
}
|
||||
|
||||
// fault answers code with the engine's error envelope, which carries `code` as
|
||||
// a NUMBER beside the message. Clients of this surface read that shape.
|
||||
func fault(code int, msg string) answer {
|
||||
return answer{status: code, ctype: ctypeJSON, body: render(map[string]any{"error": msg, "code": code})}
|
||||
}
|
||||
|
||||
// plain answers code with msg as text, the shape http.Error writes: a trailing
|
||||
// newline and the nosniff header that stops a browser guessing at the body.
|
||||
func plain(code int, msg string) answer {
|
||||
return answer{status: code, ctype: ctypePlain, nosniff: true, body: []byte(msg + "\n")}
|
||||
}
|
||||
|
||||
// absent is the answer to a path this surface routes nowhere, byte for byte
|
||||
// what http.NotFound writes.
|
||||
func absent() answer { return plain(http.StatusNotFound, "404 page not found") }
|
||||
|
||||
// empty answers code alone — no body, no content type.
|
||||
func empty(code int) answer { return answer{status: code} }
|
||||
|
||||
// render encodes v the way json.NewEncoder(w).Encode does: HTML escaped, one
|
||||
// trailing newline. A value the encoder rejects yields no bytes, which is also
|
||||
// what a caller writing straight to the response would have sent.
|
||||
func render(v any) []byte {
|
||||
var buf bytes.Buffer
|
||||
if err := json.NewEncoder(&buf).Encode(v); err != nil {
|
||||
return nil
|
||||
}
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
// write puts a on a net/http response.
|
||||
func (a answer) write(w http.ResponseWriter) {
|
||||
if a.ctype != "" {
|
||||
w.Header().Set("Content-Type", a.ctype)
|
||||
}
|
||||
if a.nosniff {
|
||||
w.Header().Set("X-Content-Type-Options", "nosniff")
|
||||
}
|
||||
w.WriteHeader(a.status)
|
||||
if len(a.body) > 0 {
|
||||
_, _ = w.Write(a.body)
|
||||
}
|
||||
}
|
||||
|
||||
// send puts a on a zip response.
|
||||
func (a answer) send(c *zip.Ctx) error {
|
||||
if a.ctype != "" {
|
||||
c.SetHeader("Content-Type", a.ctype)
|
||||
}
|
||||
if a.nosniff {
|
||||
c.SetHeader("X-Content-Type-Options", "nosniff")
|
||||
}
|
||||
return c.Bytes(a.status, a.body)
|
||||
}
|
||||
+465
-521
File diff suppressed because it is too large
Load Diff
+56
-37
@@ -11,8 +11,10 @@
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -89,50 +91,67 @@ func (b *broker) publish(e Event) {
|
||||
}
|
||||
}
|
||||
|
||||
// eventHeaders are the four an event stream carries, in the order the browser
|
||||
// reads them. Both transports set exactly these before the first frame.
|
||||
var eventHeaders = [][2]string{
|
||||
{"Content-Type", "text/event-stream"},
|
||||
{"Cache-Control", "no-cache"},
|
||||
{"Connection", "keep-alive"},
|
||||
{"X-Accel-Buffering", "no"},
|
||||
}
|
||||
|
||||
// sseHandler streams Events as text/event-stream. Disconnects unsubscribe.
|
||||
func (e *Embedded) sseHandler() http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
flusher, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
|
||||
plain(http.StatusInternalServerError, "streaming unsupported").write(w)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("Connection", "keep-alive")
|
||||
w.Header().Set("X-Accel-Buffering", "no")
|
||||
|
||||
// Hello frame so clients know the stream is live.
|
||||
fmt.Fprintf(w, ": hanzo-tasks event stream\n\n")
|
||||
flusher.Flush()
|
||||
|
||||
id, ch := e.engine.broker.subscribe()
|
||||
defer e.engine.broker.unsubscribe(id)
|
||||
|
||||
// Per-tenant filter: when the caller authenticated, only deliver
|
||||
// events tagged with their org. Empty caller-org (embedded/dev)
|
||||
// sees the full unscoped firehose — same surface as today.
|
||||
callerOrg := auth.OrgID(r.Context())
|
||||
|
||||
ctx := r.Context()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ev, ok := <-ch:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if callerOrg != "" && ev.OrgID != callerOrg {
|
||||
continue
|
||||
}
|
||||
body, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Kind, body)
|
||||
flusher.Flush()
|
||||
}
|
||||
for _, h := range eventHeaders {
|
||||
w.Header().Set(h[0], h[1])
|
||||
}
|
||||
e.feed(r.Context(), auth.OrgID(r.Context()), w, func() error { flusher.Flush(); return nil })
|
||||
})
|
||||
}
|
||||
|
||||
// feed writes the engine's event stream to w until ctx ends, the subscription
|
||||
// closes, or the client stops reading. org "" sees the full unscoped firehose
|
||||
// (embedded/dev); otherwise only events tagged with that tenant. flush pushes
|
||||
// each frame and reports a connection that can no longer take one.
|
||||
func (e *Embedded) feed(ctx context.Context, org string, w io.Writer, flush func() error) {
|
||||
// Hello frame so clients know the stream is live.
|
||||
if _, err := io.WriteString(w, ": hanzo-tasks event stream\n\n"); err != nil {
|
||||
return
|
||||
}
|
||||
if flush() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
id, ch := e.engine.broker.subscribe()
|
||||
defer e.engine.broker.unsubscribe(id)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case ev, ok := <-ch:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if org != "" && ev.OrgID != org {
|
||||
continue
|
||||
}
|
||||
body, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Kind, body); err != nil {
|
||||
return
|
||||
}
|
||||
if flush() != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+72
-80
@@ -29,39 +29,44 @@ func (e *Embedded) mcpHandler() http.Handler {
|
||||
return http.NotFoundHandler()
|
||||
}
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "POST required", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
var req mcpRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
mcpError(w, nil, -32700, "parse error: "+err.Error())
|
||||
return
|
||||
}
|
||||
switch req.Method {
|
||||
case "initialize":
|
||||
mcpResult(w, req.ID, map[string]any{
|
||||
"protocolVersion": "2025-06-18",
|
||||
"serverInfo": map[string]any{
|
||||
"name": "hanzo-tasks",
|
||||
"version": "1.40.0",
|
||||
},
|
||||
"capabilities": map[string]any{
|
||||
"tools": map[string]any{},
|
||||
},
|
||||
})
|
||||
case "tools/list":
|
||||
mcpResult(w, req.ID, map[string]any{"tools": mcpTools()})
|
||||
case "tools/call":
|
||||
e.mcpToolCall(w, r, req)
|
||||
case "ping":
|
||||
mcpResult(w, req.ID, map[string]any{})
|
||||
default:
|
||||
mcpError(w, req.ID, -32601, "method not found: "+req.Method)
|
||||
}
|
||||
e.mcp(asked(r)).write(w)
|
||||
})
|
||||
}
|
||||
|
||||
// mcp answers one JSON-RPC 2.0 call against the MCP method surface.
|
||||
func (e *Embedded) mcp(rq call) answer {
|
||||
if rq.method != http.MethodPost {
|
||||
return plain(http.StatusMethodNotAllowed, "POST required")
|
||||
}
|
||||
var req mcpRequest
|
||||
// A stream decode, so an empty body is a parse error rather than a call
|
||||
// with no method.
|
||||
if err := rq.stream(&req); err != nil {
|
||||
return mcpError(nil, -32700, "parse error: "+err.Error())
|
||||
}
|
||||
switch req.Method {
|
||||
case "initialize":
|
||||
return mcpResult(req.ID, map[string]any{
|
||||
"protocolVersion": "2025-06-18",
|
||||
"serverInfo": map[string]any{
|
||||
"name": "hanzo-tasks",
|
||||
"version": "1.40.0",
|
||||
},
|
||||
"capabilities": map[string]any{
|
||||
"tools": map[string]any{},
|
||||
},
|
||||
})
|
||||
case "tools/list":
|
||||
return mcpResult(req.ID, map[string]any{"tools": mcpTools()})
|
||||
case "tools/call":
|
||||
return e.mcpToolCall(rq, req)
|
||||
case "ping":
|
||||
return mcpResult(req.ID, map[string]any{})
|
||||
default:
|
||||
return mcpError(req.ID, -32601, "method not found: "+req.Method)
|
||||
}
|
||||
}
|
||||
|
||||
// mcpTools is the static MCP tool catalogue. Each tool maps onto an
|
||||
// engine method; arg shapes follow the JSON Schema MCP spec expects.
|
||||
func mcpTools() []map[string]any {
|
||||
@@ -137,10 +142,10 @@ func mcpTools() []map[string]any {
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Embedded) mcpToolCall(w http.ResponseWriter, r *http.Request, req mcpRequest) {
|
||||
func (e *Embedded) mcpToolCall(rq call, req mcpRequest) answer {
|
||||
// Scope every tool call to the caller's IAM-validated org. Empty org
|
||||
// (embedded/dev) returns the unscoped engine — same surface as today.
|
||||
en := e.engine.As(Org(auth.OrgID(r.Context())))
|
||||
en := e.engine.As(Org(auth.OrgID(rq.ctx)))
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Arguments map[string]any `json:"arguments"`
|
||||
@@ -155,88 +160,77 @@ func (e *Embedded) mcpToolCall(w http.ResponseWriter, r *http.Request, req mcpRe
|
||||
return ""
|
||||
}
|
||||
|
||||
emit := func(v any) {
|
||||
emit := func(v any) answer {
|
||||
text, _ := json.Marshal(v)
|
||||
mcpResult(w, req.ID, map[string]any{
|
||||
return mcpResult(req.ID, map[string]any{
|
||||
"content": []map[string]any{{"type": "text", "text": string(text)}},
|
||||
})
|
||||
}
|
||||
fail := func(code int, msg string) {
|
||||
mcpError(w, req.ID, code, msg)
|
||||
fail := func(code int, msg string) answer {
|
||||
return mcpError(req.ID, code, msg)
|
||||
}
|
||||
|
||||
switch p.Name {
|
||||
case "list_namespaces":
|
||||
rows, err := en.ListNamespaces()
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(map[string]any{"namespaces": rows})
|
||||
return emit(map[string]any{"namespaces": rows})
|
||||
case "describe_namespace":
|
||||
n, ok, err := en.DescribeNamespace(str("name"))
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
if !ok {
|
||||
fail(-32004, "namespace not found")
|
||||
return
|
||||
return fail(-32004, "namespace not found")
|
||||
}
|
||||
emit(n)
|
||||
return emit(n)
|
||||
case "list_workflows":
|
||||
rows, err := en.ListWorkflows(str("namespace"))
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(map[string]any{"executions": rows})
|
||||
return emit(map[string]any{"executions": rows})
|
||||
case "describe_workflow":
|
||||
wf, ok, err := en.DescribeWorkflow(str("namespace"), str("workflow_id"), str("run_id"))
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
if !ok {
|
||||
fail(-32004, "workflow not found")
|
||||
return
|
||||
return fail(-32004, "workflow not found")
|
||||
}
|
||||
emit(wf)
|
||||
return emit(wf)
|
||||
case "start_workflow":
|
||||
typ := TypeRef{Name: str("workflow_type")}
|
||||
wf, err := en.StartWorkflow(str("namespace"), str("workflow_id"), str("run_id"), typ, str("task_queue"), a["input"])
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(wf)
|
||||
return emit(wf)
|
||||
case "signal_workflow":
|
||||
if err := en.SignalWorkflow(str("namespace"), str("workflow_id"), str("run_id"), str("signal_name"), a["input"]); err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(map[string]string{"status": "signaled"})
|
||||
return emit(map[string]string{"status": "signaled"})
|
||||
case "cancel_workflow":
|
||||
wf, err := en.CancelWorkflow(str("namespace"), str("workflow_id"), str("run_id"))
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(wf)
|
||||
return emit(wf)
|
||||
case "terminate_workflow":
|
||||
wf, err := en.TerminateWorkflow(str("namespace"), str("workflow_id"), str("run_id"))
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(wf)
|
||||
return emit(wf)
|
||||
case "list_schedules":
|
||||
rows, err := en.ListSchedules(str("namespace"))
|
||||
if err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(map[string]any{"schedules": rows})
|
||||
return emit(map[string]any{"schedules": rows})
|
||||
case "create_schedule":
|
||||
s := Schedule{
|
||||
ScheduleId: str("schedule_id"),
|
||||
@@ -248,19 +242,17 @@ func (e *Embedded) mcpToolCall(w http.ResponseWriter, r *http.Request, req mcpRe
|
||||
},
|
||||
}
|
||||
if err := en.CreateSchedule(s); err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(s)
|
||||
return emit(s)
|
||||
case "pause_schedule":
|
||||
paused, _ := a["paused"].(bool)
|
||||
if err := en.PauseSchedule(str("namespace"), str("schedule_id"), paused, ""); err != nil {
|
||||
fail(-32000, err.Error())
|
||||
return
|
||||
return fail(-32000, err.Error())
|
||||
}
|
||||
emit(map[string]any{"paused": paused})
|
||||
return emit(map[string]any{"paused": paused})
|
||||
default:
|
||||
fail(-32601, "unknown tool: "+p.Name)
|
||||
return fail(-32601, "unknown tool: "+p.Name)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -272,18 +264,18 @@ type mcpRequest struct {
|
||||
Params any `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
func mcpResult(w http.ResponseWriter, id any, result any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
// A JSON-RPC answer carries its outcome in the body, so both of these are 200:
|
||||
// the transport delivered the call, and `error` is the call's own verdict.
|
||||
func mcpResult(id any, result any) answer {
|
||||
return data(nil, map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"result": result,
|
||||
})
|
||||
}
|
||||
|
||||
func mcpError(w http.ResponseWriter, id any, code int, msg string) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
func mcpError(id any, code int, msg string) answer {
|
||||
return data(nil, map[string]any{
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"error": map[string]any{"code": code, "message": msg},
|
||||
|
||||
@@ -0,0 +1,314 @@
|
||||
// Copyright © 2026 Hanzo AI. MIT License.
|
||||
|
||||
// The engine's surface as a zip application — the door a composer includes
|
||||
// with app.Use, holding no net/http type anywhere in the request path.
|
||||
//
|
||||
// Every address here is a REAL route: zip's router matches it and zip's own
|
||||
// projections read it, where an adapted http.Handler is one opaque wildcard
|
||||
// over sixty-odd operations the composing router never sees.
|
||||
//
|
||||
// The operations behind the routes are the same functions HTTPHandler,
|
||||
// ClusterHandler, MCPHandler and EventsHandler reach — a call in, an answer
|
||||
// out — so the two transports cannot describe one operation differently.
|
||||
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"html"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"path"
|
||||
"strings"
|
||||
|
||||
"github.com/hanzoai/tasks/pkg/auth"
|
||||
"github.com/zap-proto/zip"
|
||||
)
|
||||
|
||||
// root is where the engine answers. Every route below hangs off it.
|
||||
const root = "/v1/tasks"
|
||||
|
||||
// Surface returns the engine's routes as a zip application:
|
||||
//
|
||||
// app.Use(engine.Surface())
|
||||
//
|
||||
// Identity is read the way it always has been — auth.OrgID over the request's
|
||||
// context — so a composer that terminates the IAM boundary states it once,
|
||||
// with zip.Ctx.SetContext(auth.WithIdentity(…)), and every operation lands in
|
||||
// that tenant's shard. A request carrying no identity reads the unscoped
|
||||
// store, the same embedded/dev path the mux serves.
|
||||
//
|
||||
// The engine answers everything under /v1/tasks, including a path it routes
|
||||
// nowhere: the trailing wildcard carries that refusal, and it yields to any
|
||||
// route registered at a more specific address — a composer's own
|
||||
// /v1/tasks/health among them.
|
||||
func (e *Embedded) Surface() *zip.App {
|
||||
app := zip.New(zip.Config{})
|
||||
|
||||
// The addresses that stand on their own. The MCP endpoint and the event
|
||||
// stream are NAMED here, because a route carries its address where a bare
|
||||
// handler left the choice to whoever mounted it.
|
||||
app.Get(root+"/settings", handle(e, func(rq call, _ *engine) answer { return settings(rq) }))
|
||||
app.Get(root+"/namespaces", handle(e, namespaces))
|
||||
app.Post(root+"/namespaces", handle(e, namespaces))
|
||||
app.Get(root+"/nexus", handle(e, endpoints))
|
||||
app.Get(root+"/cluster", alone(e.clusterStatus))
|
||||
app.Get(root+"/cluster/health", alone(e.clusterHealth))
|
||||
app.Post(root+"/mcp", alone(e.mcp))
|
||||
app.Get(root+"/events", e.tail)
|
||||
|
||||
// One namespace, and the migration of one namespace.
|
||||
app.Get(root+"/namespaces/:ns", named(e, namespace))
|
||||
app.Delete(root+"/namespaces/:ns", named(e, namespace))
|
||||
app.Post(root+"/namespaces/:ns/migrate", func(c *zip.Ctx) error {
|
||||
ns := c.Param("ns")
|
||||
if a, ok := grammar(ns, nil); !ok {
|
||||
return a.send(c)
|
||||
}
|
||||
return e.migrate(taken(c), ns).send(c)
|
||||
})
|
||||
|
||||
// Everything a namespace contains.
|
||||
for _, r := range contents {
|
||||
kind, below, _ := strings.Cut(r.path, "/")
|
||||
var rest []string
|
||||
if below != "" {
|
||||
rest = strings.Split(below, "/")
|
||||
}
|
||||
on(app, r.method, root+"/namespaces/:ns/"+r.path, func(c *zip.Ctx) error {
|
||||
rq := taken(c)
|
||||
ns := c.Param("ns")
|
||||
sub := segments(c, rest)
|
||||
if a, ok := grammar(ns, sub); !ok {
|
||||
return a.send(c)
|
||||
}
|
||||
return resource(rq, e.engine.As(Org(auth.OrgID(rq.ctx))), ns, kind, sub).send(c)
|
||||
})
|
||||
}
|
||||
|
||||
// A path under /v1/tasks that no route above claims. The engine has always
|
||||
// answered it with net/http's plain-text refusal rather than a status of
|
||||
// its own, and clients read that shape.
|
||||
//
|
||||
// Before refusing, it answers the redirect a path needing cleaning earns.
|
||||
// http.ServeMux tidies a path — an empty segment, a dot segment — and
|
||||
// redirects to the tidied form before it matches anything, so a request
|
||||
// like /namespaces//workflows reaches a route there and 404s here. That is
|
||||
// a difference a caller can see, so it is reproduced rather than accepted.
|
||||
app.All(root+"/*", func(c *zip.Ctx) error {
|
||||
if to, dirty := tidy(c.Path()); dirty {
|
||||
if q := string(c.Fiber().RequestCtx().URI().QueryString()); q != "" {
|
||||
to += "?" + q
|
||||
}
|
||||
c.SetHeader("Location", to)
|
||||
// http.Redirect writes the hypertext body for a GET or a HEAD and
|
||||
// for no other method.
|
||||
switch c.Method() {
|
||||
case http.MethodGet, http.MethodHead:
|
||||
return answer{
|
||||
status: http.StatusTemporaryRedirect,
|
||||
ctype: "text/html; charset=utf-8",
|
||||
body: []byte("<a href=\"" + html.EscapeString(to) + "\">Temporary Redirect</a>.\n\n"),
|
||||
}.send(c)
|
||||
default:
|
||||
return empty(http.StatusTemporaryRedirect).send(c)
|
||||
}
|
||||
}
|
||||
return absent().send(c)
|
||||
})
|
||||
|
||||
return app
|
||||
}
|
||||
|
||||
// row is one address a namespace's contents answer: the method, and the path
|
||||
// below /v1/tasks/namespaces/:ns/. The first segment names the resource; a
|
||||
// ":name" segment is a path parameter and every other segment reaches the
|
||||
// operation as itself — which is what the mux hands it after a split.
|
||||
type row struct {
|
||||
method string
|
||||
path string
|
||||
}
|
||||
|
||||
const (
|
||||
get = http.MethodGet
|
||||
post = http.MethodPost
|
||||
del = http.MethodDelete
|
||||
)
|
||||
|
||||
// contents is the engine's route table under one namespace. Sixty-one
|
||||
// addresses, each registered on its own with its own method, which is the
|
||||
// whole difference between this and a wildcard.
|
||||
var contents = []row{
|
||||
{get, "workflows"},
|
||||
{post, "workflows"},
|
||||
{post, "workflows/signal-with-start"},
|
||||
{get, "workflows/:wf"},
|
||||
{post, "workflows/:wf/cancel"},
|
||||
{post, "workflows/:wf/terminate"},
|
||||
{post, "workflows/:wf/signal"},
|
||||
{get, "workflows/:wf/history"},
|
||||
{post, "workflows/:wf/query"},
|
||||
{post, "workflows/:wf/metadata"},
|
||||
{get, "workflows/:wf/executions"},
|
||||
{post, "workflows/:wf/reset"},
|
||||
|
||||
{get, "schedules"},
|
||||
{post, "schedules"},
|
||||
{get, "schedules/:id"},
|
||||
{post, "schedules/:id"},
|
||||
{del, "schedules/:id"},
|
||||
{post, "schedules/:id/trigger"},
|
||||
{get, "schedules/:id/matching-times"},
|
||||
{post, "schedules/:id/pause"},
|
||||
{post, "schedules/:id/unpause"},
|
||||
|
||||
{get, "batches"},
|
||||
{post, "batches"},
|
||||
{get, "batches/:id"},
|
||||
{post, "batches/:id/terminate"},
|
||||
|
||||
{get, "deployments"},
|
||||
{post, "deployments"},
|
||||
{get, "deployments/:id"},
|
||||
{post, "deployments/:id"},
|
||||
{del, "deployments/:id"},
|
||||
{post, "deployments/:id/set-current"},
|
||||
{post, "deployments/:id/versions"},
|
||||
{post, "deployments/:id/versions/:build"},
|
||||
{del, "deployments/:id/versions/:build"},
|
||||
{post, "deployments/:id/versions/:build/validate"},
|
||||
|
||||
{get, "nexus"},
|
||||
{post, "nexus"},
|
||||
{del, "nexus/:id"},
|
||||
|
||||
{get, "identities"},
|
||||
{post, "identities"},
|
||||
{del, "identities/:id"},
|
||||
|
||||
{get, "task-queues"},
|
||||
{get, "task-queues/:queue"},
|
||||
{get, "task-queues/:queue/workers"},
|
||||
{get, "task-queues/:queue/partitions"},
|
||||
|
||||
{get, "workers"},
|
||||
{get, "workers/:id"},
|
||||
|
||||
{get, "search-attributes"},
|
||||
{post, "search-attributes"},
|
||||
{del, "search-attributes/:attr"},
|
||||
|
||||
{post, "metadata"},
|
||||
|
||||
{get, "archival"},
|
||||
|
||||
{get, "activities"},
|
||||
{post, "activities"},
|
||||
{post, "activities/claim"},
|
||||
{get, "activities/:id/:run"},
|
||||
{post, "activities/:id/:run/cancel"},
|
||||
{post, "activities/:id/:run/complete"},
|
||||
{post, "activities/:id/:run/fail"},
|
||||
{post, "activities/:id/:run/heartbeat"},
|
||||
{get, "activities/:id/:run/history"},
|
||||
}
|
||||
|
||||
// on registers h at path for one method. A method outside the three the engine
|
||||
// serves is a mistake in the table, and panicking at composition is where it
|
||||
// can still be seen — a silently unregistered route answers 404 in production.
|
||||
func on(app *zip.App, method, path string, h zip.Handler) {
|
||||
switch method {
|
||||
case get:
|
||||
app.Get(path, h)
|
||||
case post:
|
||||
app.Post(path, h)
|
||||
case del:
|
||||
app.Delete(path, h)
|
||||
default:
|
||||
panic("tasks: no route method " + method)
|
||||
}
|
||||
}
|
||||
|
||||
// segments resolves a row's path below the resource name into the values the
|
||||
// operation reads: a parameter takes the segment that matched it, anything
|
||||
// else is itself.
|
||||
func segments(c *zip.Ctx, rest []string) []string {
|
||||
out := make([]string, len(rest))
|
||||
for i, s := range rest {
|
||||
if name, ok := strings.CutPrefix(s, ":"); ok {
|
||||
out[i] = c.Param(name)
|
||||
continue
|
||||
}
|
||||
out[i] = s
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// handle renders one operation that reads the org-scoped engine.
|
||||
func handle(e *Embedded, fn func(rq call, en *engine) answer) zip.Handler {
|
||||
return func(c *zip.Ctx) error {
|
||||
rq := taken(c)
|
||||
return fn(rq, e.engine.As(Org(auth.OrgID(rq.ctx)))).send(c)
|
||||
}
|
||||
}
|
||||
|
||||
// alone renders one operation that resolves whatever engine view it needs.
|
||||
func alone(fn func(rq call) answer) zip.Handler {
|
||||
return func(c *zip.Ctx) error { return fn(taken(c)).send(c) }
|
||||
}
|
||||
|
||||
// named renders one operation on a namespace, after that namespace has
|
||||
// satisfied the grammar a constructed store key depends on.
|
||||
func named(e *Embedded, fn func(rq call, en *engine, ns string) answer) zip.Handler {
|
||||
return func(c *zip.Ctx) error {
|
||||
rq := taken(c)
|
||||
ns := c.Param("ns")
|
||||
if a, ok := grammar(ns, nil); !ok {
|
||||
return a.send(c)
|
||||
}
|
||||
return fn(rq, e.engine.As(Org(auth.OrgID(rq.ctx))), ns).send(c)
|
||||
}
|
||||
}
|
||||
|
||||
// tail streams the engine's event feed. It ends when the client stops reading
|
||||
// — a flush onto a closed connection reports it — which is how a feed with no
|
||||
// natural end unsubscribes.
|
||||
func (e *Embedded) tail(c *zip.Ctx) error {
|
||||
for _, h := range eventHeaders {
|
||||
c.SetHeader(h[0], h[1])
|
||||
}
|
||||
ctx := c.Context()
|
||||
org := auth.OrgID(ctx)
|
||||
return c.SendStreamWriter(func(w *bufio.Writer) { e.feed(ctx, org, w, w.Flush) })
|
||||
}
|
||||
|
||||
// taken reads one zip request as an operation reads it. The query is parsed
|
||||
// from the raw string by the same parser net/http uses, so a repeated or
|
||||
// malformed parameter resolves identically on both transports.
|
||||
func taken(c *zip.Ctx) call {
|
||||
q, _ := url.ParseQuery(string(c.Fiber().Request().URI().QueryString()))
|
||||
return call{ctx: c.Context(), method: c.Method(), query: q, body: c.Body()}
|
||||
}
|
||||
|
||||
// tidy is the path http.ServeMux would route a request at, and whether that
|
||||
// differs from the one it arrived on. It is net/http's own cleanPath: collapse
|
||||
// the path, then put back a trailing slash the collapse removed, because
|
||||
// "/x/" and "/x" are two addresses to a mux and only the first names a subtree.
|
||||
func tidy(p string) (string, bool) {
|
||||
if p == "" {
|
||||
return "/", true
|
||||
}
|
||||
q := p
|
||||
if q[0] != '/' {
|
||||
q = "/" + q
|
||||
}
|
||||
c := path.Clean(q)
|
||||
if q[len(q)-1] == '/' && c != "/" {
|
||||
if len(q) == len(c)+1 && strings.HasPrefix(q, c) {
|
||||
c = q
|
||||
} else {
|
||||
c += "/"
|
||||
}
|
||||
}
|
||||
return c, c != p
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package tasks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/zap-proto/zip"
|
||||
)
|
||||
|
||||
// The engine hands a composer two renderings of one surface: HTTPHandler, the
|
||||
// http.ServeMux it has always exposed, and Surface, the zip app a composer mounts
|
||||
// with Use. A caller must not be able to tell which one answered it, so this
|
||||
// drives the same requests through both and compares what comes back.
|
||||
//
|
||||
// It replaces a probe that printed the two answers side by side and asserted
|
||||
// nothing — which passes however far the two drift, and is the shape of test
|
||||
// that reports coverage it does not have.
|
||||
//
|
||||
// Framing is excluded and named rather than waved at: Date is the clock, and
|
||||
// Content-Length is a property of how a body was written, not of what it says.
|
||||
// Everything a caller reads to decide what happened — status, Content-Type,
|
||||
// Location, and the body itself — is compared exactly.
|
||||
func TestSurfaceAnswersWhatTheMuxAnswers(t *testing.T) {
|
||||
e, err := Embed(context.Background(), EmbedConfig{DataDir: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatalf("Embed: %v", err)
|
||||
}
|
||||
defer e.Stop(context.Background())
|
||||
|
||||
mux := e.HTTPHandler()
|
||||
app := zip.New(zip.Config{DisableStartupMessage: true})
|
||||
app.Use(e.Surface())
|
||||
|
||||
// Each row is an address worth disagreeing about: a real read, an unknown
|
||||
// leaf, the bare noun, a deep parameterised write, an empty path segment,
|
||||
// and a path that needs cleaning.
|
||||
for _, tc := range []struct{ method, path string }{
|
||||
{http.MethodGet, "/v1/tasks/settings"},
|
||||
{http.MethodGet, "/v1/tasks/nope"},
|
||||
{http.MethodGet, "/v1/tasks"},
|
||||
{http.MethodPost, "/v1/tasks/namespaces/default/activities/claim"},
|
||||
{http.MethodGet, "/v1/tasks/namespaces//workflows"},
|
||||
{http.MethodGet, "/v1/tasks/namespaces/default/workflows/../x"},
|
||||
} {
|
||||
t.Run(tc.method+" "+tc.path, func(t *testing.T) {
|
||||
rec := httptest.NewRecorder()
|
||||
mux.ServeHTTP(rec, httptest.NewRequest(tc.method, tc.path, nil))
|
||||
want := rec.Result()
|
||||
wantBody, _ := io.ReadAll(want.Body)
|
||||
|
||||
got, err := app.Test(httptest.NewRequest(tc.method, tc.path, nil))
|
||||
if err != nil {
|
||||
t.Fatalf("zip app: %v", err)
|
||||
}
|
||||
gotBody, _ := io.ReadAll(got.Body)
|
||||
|
||||
if got.StatusCode != want.StatusCode {
|
||||
t.Errorf("status = %d, mux answers %d", got.StatusCode, want.StatusCode)
|
||||
}
|
||||
if string(gotBody) != string(wantBody) {
|
||||
t.Errorf("body = %q, mux answers %q", gotBody, wantBody)
|
||||
}
|
||||
for _, h := range []string{"Content-Type", "Location"} {
|
||||
if g, w := got.Header.Get(h), want.Header.Get(h); g != w {
|
||||
t.Errorf("%s = %q, mux answers %q", h, g, w)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user