1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package rpc
4
5import (
6 "context"
7 "io"
8 "net/http"
9 "sync"
10
11 "bnema.dev/ermite/internal/domain"
12)
13
14// inProcessHost is the host of the URLs of in-process calls. The name is
15// never resolved.
16const inProcessHost = "ermite.invalid"
17
18// InProcessURL is the base URL to give to the generated Connect clients that
19// use InProcessClient.
20const InProcessURL = "http://" + inProcessHost
21
22// InProcessClient returns an HTTP client whose requests are served by the
23// handlers of the API in this process: no socket, no network hop. Build the
24// generated Connect clients on it:
25//
26// hc := api.InProcessClient()
27// repos := ermiteconnect.NewRepoServiceClient(hc, rpc.InProcessURL)
28//
29// The caller is the principal found in the context of each call
30// (domain.ContextWithPrincipal), anonymous when there is none; the
31// Authorization header is ignored. This is the identity of the web (P5):
32// anonymous for visitors, the owner for a session. Only code in this process
33// can put the marker that makes the interceptor trust it.
34// Unary and server-streaming calls work, with all the codecs; the request
35// log and the error mapping are those of remote calls.
36func (a *API) InProcessClient() *http.Client {
37 return &http.Client{Transport: inProcessTransport{h: a.handler}}
38}
39
40type inProcessTransport struct{ h http.Handler }
41
42// inProcessKey marks the context of a call made through InProcessClient. It is
43// unexported, so that nothing outside this package, and nothing that comes
44// from the network, can make the interceptor trust a principal.
45type inProcessKey struct{}
46
47func inProcess(ctx context.Context) bool {
48 _, ok := ctx.Value(inProcessKey{}).(struct{})
49 return ok
50}
51
52// RoundTrip serves req with the handler in a goroutine, and returns once the
53// handler wrote its status. The body of the response is a pipe the handler
54// writes to, so that streams are delivered as they are sent.
55func (t inProcessTransport) RoundTrip(req *http.Request) (*http.Response, error) {
56 ctx := req.Context()
57 p, ok := domain.PrincipalFromContext(ctx)
58 if !ok {
59 p = domain.Anonymous()
60 }
61 sreq := req.Clone(context.WithValue(domain.ContextWithPrincipal(ctx, p), inProcessKey{}, struct{}{}))
62 sreq.RemoteAddr = "in-process"
63 sreq.RequestURI = req.URL.RequestURI()
64 if sreq.Body == nil {
65 sreq.Body = http.NoBody
66 }
67 sreq.Header.Del("Authorization")
68
69 pr, pw := io.Pipe()
70 w := &pipeResponse{header: http.Header{}, pw: pw, ready: make(chan struct{})}
71 go func() {
72 defer func() {
73 if r := recover(); r != nil {
74 w.writeHeader(http.StatusInternalServerError)
75 _ = pw.CloseWithError(io.ErrUnexpectedEOF)
76 }
77 }()
78 t.h.ServeHTTP(w, sreq)
79 w.writeHeader(http.StatusOK) // a handler that wrote nothing
80 _ = pw.Close()
81 }()
82
83 select {
84 case <-w.ready:
85 case <-ctx.Done():
86 _ = pr.CloseWithError(ctx.Err())
87 return nil, ctx.Err()
88 }
89 return &http.Response{
90 Status: http.StatusText(w.status),
91 StatusCode: w.status,
92 Proto: "HTTP/1.1",
93 ProtoMajor: 1,
94 ProtoMinor: 1,
95 Header: w.sent,
96 Body: pr,
97 ContentLength: -1,
98 Request: req,
99 }, nil
100}
101
102// pipeResponse is the http.ResponseWriter of an in-process call.
103type pipeResponse struct {
104 header http.Header
105 pw *io.PipeWriter
106
107 once sync.Once
108 ready chan struct{}
109 status int
110 sent http.Header // header as of WriteHeader
111}
112
113func (w *pipeResponse) Header() http.Header { return w.header }
114
115func (w *pipeResponse) WriteHeader(code int) { w.writeHeader(code) }
116
117func (w *pipeResponse) writeHeader(code int) {
118 w.once.Do(func() {
119 w.status, w.sent = code, w.header.Clone()
120 close(w.ready)
121 })
122}
123
124func (w *pipeResponse) Write(b []byte) (int, error) {
125 w.writeHeader(http.StatusOK)
126 return w.pw.Write(b)
127}
128
129// Flush implements http.Flusher, which streaming handlers require: the pipe
130// delivers every write at once, there is nothing to flush but the header.
131func (w *pipeResponse) Flush() { w.writeHeader(http.StatusOK) }