1	# ezhttp/src/server: listen, accept, parse one request, write one response.
2	# Uses Base TCP (no TLS in v0). One exchange per connection with
3	# Connection: close, one connection at a time. The handler returns IO; a pure
4	# one is lifted. Every decision after an effect is step.bend's, so a failed
5	# accept, recv or send loses that connection, logs one line and the server
6	# takes the next one (HTTP-SRV-4).
7	import Base
8	import ./http.bend as Http
9	import ./cors.bend as Cors
10	import ./step.bend as Step
11	
12	# default recv budget for one request message
13	def server.max() -> U32:
14	  65536
15	
16	# listen on a host address and port, e.g. "127.0.0.1" or "0.0.0.0"
17	# (RFC 9112 connection establishment stays at the socket)
18	def server.listen(host: String, port: U32) -> IO(Listener):
19	  IO.try(Listener, TCP.listen(host, port))
20	
21	# accept one connection; a failed accept is a lost connection
22	def server.accept.un(
23	  pair: Listener & Result<&1, &1, U32 & String, Socket>
24	) -> IO(Listener & Step.Got<Socket>):
25	  (lst, res) = pair
26	  IO.pure(Listener & Step.Got<Socket>, (lst, Step.step.got(Socket, "accept", res)))
27	
28	def server.accept(listener: Listener) -> IO(Listener & Step.Got<Socket>):
29	  IO.bind(
30	    Listener & Result<&1, &1, U32 & String, Socket>,
31	    Listener & Step.Got<Socket>,
32	    TCP.accept(listener),
33	    server.accept.un)
34	
35	# the short-body refusal, so a partial recv can continue
36	def server.read.why() -> String:
37	  "the request body stopped before its Content-Length"
38	
39	# whether this refusal is a body that has not arrived yet
40	def server.read.short(why: String) -> Bool:
41	  String.eq(why, server.read.why())
42	
43	# a parsed request still wants more bytes
44	def server.read.need(req: Http.Request) -> Bool:
45	  match req:
46	    case Http.Bad{why}:
47	      server.read.short(why)
48	    case Http.Request{method, target, headers, body}:
49	      False{}
50	
51	# how many extra recvs to chase a Content-Length
52	def server.read.fuel() -> Nat:
53	  64n
54	
55	# parse what has arrived
56	def server.read.done(sock: Socket, got: String) -> IO(Socket & Step.Got<Http.Request>):
57	  IO.pure(Socket & Step.Got<Http.Request>, (sock, Step.Kept{Http.ask(got)}))
58	
59	# stop on a finished message, or ask for another recv
60	def server.read.next(
61	  sock: Socket,
62	  got: String,
63	  req: Http.Request,
64	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
65	  more: Bool
66	) -> IO(Socket & Step.Got<Http.Request>):
67	  match more:
68	    case False{}:
69	      IO.pure(Socket & Step.Got<Http.Request>, (sock, Step.Kept{req}))
70	    case True{}:
71	      rest(sock)(got)
72	
73	# append a chunk and decide whether it completes the request
74	def server.read.grew(
75	  sock: Socket,
76	  +got: String,
77	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>)
78	) -> IO(Socket & Step.Got<Http.Request>):
79	  +req = Http.ask(got)
80	  server.read.next(sock, got, req, rest, server.read.need(req))
81	
82	# peer closed, or the chunk extends the buffer
83	def server.read.empty(
84	  sock: Socket,
85	  got: String,
86	  chunk: String,
87	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
88	  closed: Bool
89	) -> IO(Socket & Step.Got<Http.Request>):
90	  match closed:
91	    case True{}:
92	      server.read.done(sock, got)
93	    case False{}:
94	      server.read.grew(sock, got ++ chunk, rest)
95	
96	# one chunk of the request message
97	def server.read.use(
98	  sock: Socket,
99	  got: String,
100	  +chunk: String,
101	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>)
102	) -> IO(Socket & Step.Got<Http.Request>):
103	  server.read.empty(sock, got, chunk, rest, String.eq(chunk, ""))
104	
105	# None{} is the peer closing its write side; Some is the bytes that arrived
106	def server.read.got(
107	  sock: Socket,
108	  got: String,
109	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
110	  chunk: Maybe<&1, String>
111	) -> IO(Socket & Step.Got<Http.Request>):
112	  match chunk:
113	    case None{}:
114	      server.read.done(sock, got)
115	    case Some{text}:
116	      server.read.use(sock, got, text, rest)
117	
118	# the bytes from one recv; a failed recv loses the connection
119	def server.read.pass(
120	  sock: Socket,
121	  got: String,
122	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
123	  chunk: Step.Got<Maybe<&1, String>>
124	) -> IO(Socket & Step.Got<Http.Request>):
125	  match chunk:
126	    case Step.Lost{line}:
127	      IO.pure(Socket & Step.Got<Http.Request>, (sock, Step.Lost{line}))
128	    case Step.Kept{text}:
129	      server.read.got(sock, got, rest, text)
130	
131	# unpack recv's socket and result
132	def server.read.un(
133	  got: String,
134	  rest: Socket -> String -> IO(Socket & Step.Got<Http.Request>),
135	  pair: Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>
136	) -> IO(Socket & Step.Got<Http.Request>):
137	  (sock, r) = pair
138	  server.read.pass(sock, got, rest,
139	    Step.step.got(Maybe<&1, String>, "read", r))
140	
141	# recv until Content-Length is met, the peer closes, or fuel runs out.
142	# fuel is first so the recursive call is a decreasing self-call.
143	def server.read.go(fuel: Nat, sock: Socket, got: String) -> IO(Socket & Step.Got<Http.Request>):
144	  match fuel:
145	    case 0n:
146	      server.read.done(sock, got)
147	    case 1n+f:
148	      IO.bind(
149	        Socket & Result<&1, &1, U32 & String, Maybe<&1, String>>,
150	        Socket & Step.Got<Http.Request>,
151	        TCP.recv(sock, server.max()),
152	        pair => server.read.un(got, s => g => server.read.go(f, s, g), pair))
153	
154	# recv then parse an HTTP request (RFC 9112 §3)
155	def server.read(sock: Socket) -> IO(Socket & Step.Got<Http.Request>):
156	  server.read.go(server.read.fuel(), sock, "")
157	
158	# write a structured response; a failed send loses the connection
159	def server.write.un(
160	  pair: Socket & Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>
161	) -> IO(Socket & Step.Got<Unit>):
162	  (sock, res) = pair
163	  IO.pure(Socket & Step.Got<Unit>, (sock, Step.step.sent(res)))
164	
165	def server.write(sock: Socket, reply: Http.Reply) -> IO(Socket & Step.Got<Unit>):
166	  IO.bind(
167	    Socket & Result<&1, &1, (U32 & String) & List<&2, U32>, Unit>,
168	    Socket & Step.Got<Unit>,
169	    TCP.send(sock, Http.respond.of(reply)),
170	    server.write.un)
171	
172	# turn a bad request into 400; otherwise call the handler
173	def server.dispatch(~handle: Http.Request -> Http.Reply, req: Http.Request)
174	  -> Http.Reply:
175	  match req:
176	    case Http.Bad{why}:
177	      Http.Reply{400, [], why}
178	    case Http.Request{method, target, headers, body}:
179	      handle(Http.Request{method, target, headers, body})
180	
181	# apply HEAD body stripping once the method token is known
182	def server.reply.built(~handle: Http.Request -> Http.Reply, method: String,
183	  req: Http.Request) -> Http.Reply:
184	  Http.reply.for(method, server.dispatch(handle, req))
185	
186	# HEAD responses are written with an empty body (RFC 9110 §9.3.2)
187	def server.reply(~handle: Http.Request -> Http.Reply, req: Http.Request)
188	  -> Http.Reply:
189	  match req:
190	    case Http.Bad{why}:
191	      server.dispatch(handle, Http.Bad{why})
192	    case Http.Request{+method, target, headers, body}:
193	      server.reply.built(handle, method,
194	        Http.Request{method, target, headers, body})
195	
196	# CORS preflight or simple headers, after HEAD bodies are dropped
197	def server.cors_reply(
198	  ~handle: Http.Request -> Http.Reply,
199	  cfg: Cors.Cfg,
200	  +req: Http.Request
201	) -> Http.Reply:
202	  Cors.cors.on(cfg, req, server.reply(handle, req))
203	
204	# a lost connection logs its line and closes
205	def server.lose(sock: Socket, line: String) -> IO(Unit):
206	  IO.bind(Unit, Unit, IO.print_err(line), _u => Socket.close(sock))
207	
208	# after the write: close, logging first when the send failed
209	def server.sent.got(sock: Socket, got: Step.Got<Unit>) -> IO(Unit):
210	  match got:
211	    case Step.Lost{line}:
212	      server.lose(sock, line)
213	    case Step.Kept{_u}:
214	      Socket.close(sock)
215	
216	def server.sent(pair: Socket & Step.Got<Unit>) -> IO(Unit):
217	  (sock, got) = pair
218	  server.sent.got(sock, got)
219	
220	# write a reply shaped for the request's method (HTTP-SRV-3), then close
221	def server.answer(sock: Socket, method: String, reply: Http.Reply) -> IO(Unit):
222	  IO.bind(Socket & Step.Got<Unit>, Unit,
223	    server.write(sock, Http.reply.for(method, reply)), server.sent)
224	
225	# answer now, or call the handler: step.bend decides which (HTTP-SRV-1, 2)
226	def server.step(
227	  ~handle: Http.Request -> IO(Http.Reply),
228	  sock: Socket,
229	  +method: String,
230	  step: Step.Step
231	) -> IO(Unit):
232	  match step:
233	    case Step.Answer{reply}:
234	      server.answer(sock, method, reply)
235	    case Step.Call{req}:
236	      IO.bind(Http.Reply, Unit, handle(req),
237	        reply => server.answer(sock, method, reply))
238	
239	# after the read: a lost connection, or a request to step
240	def server.exchange.got(
241	  ~handle: Http.Request -> IO(Http.Reply),
242	  sock: Socket,
243	  got: Step.Got<Http.Request>
244	) -> IO(Unit):
245	  match got:
246	    case Step.Lost{line}:
247	      server.lose(sock, line)
248	    case Step.Kept{+req}:
249	      server.step(handle, sock, Http.method.of(req), Step.step.of(req))
250	
251	def server.exchange.un(
252	  ~handle: Http.Request -> IO(Http.Reply),
253	  pair: Socket & Step.Got<Http.Request>
254	) -> IO(Unit):
255	  (sock, got) = pair
256	  server.exchange.got(handle, sock, got)
257	
258	# one connection: read, decide, maybe handle, write, close
259	def server.exchange(
260	  ~handle: Http.Request -> IO(Http.Reply),
261	  sock: Socket
262	) -> IO(Unit):
263	  IO.bind(Socket & Step.Got<Http.Request>, Unit, server.read(sock),
264	    pair => server.exchange.un(handle, pair))
265	
266	# how long a failed accept waits before the next, so EMFILE does not spin
267	def server.pause() -> U32:
268	  100
269	
270	# after accept: serve the connection, or log a failed accept and pause, then
271	# call `again` with the listener either way. `again` is a runtime closure so
272	# it can carry the fuel still left.
273	def server.loop.got(
274	  ~handle: Http.Request -> IO(Http.Reply),
275	  again: Listener -> IO(Unit),
276	  lst: Listener,
277	  got: Step.Got<Socket>
278	) -> IO(Unit):
279	  match got:
280	    case Step.Lost{line}:
281	      IO.bind(Unit, Unit, IO.print_err(line), _u =>
282	        IO.bind(Unit, Unit, IO.sleep(server.pause()), _v => again(lst)))
283	    case Step.Kept{sock}:
284	      IO.bind(Unit, Unit, server.exchange(handle, sock), _u => again(lst))
285	
286	def server.loop.cont(
287	  ~handle: Http.Request -> IO(Http.Reply),
288	  again: Listener -> IO(Unit),
289	  pair: Listener & Step.Got<Socket>
290	) -> IO(Unit):
291	  (lst, got) = pair
292	  server.loop.got(handle, again, lst, got)
293	
294	# fuel-bounded accept loop (self-recursive); a failed accept spends fuel too
295	def server.loop.go(
296	  ~handle: Http.Request -> IO(Http.Reply),
297	  fuel: Nat,
298	  listener: Listener
299	) -> IO(Unit):
300	  match fuel:
301	    case 0n:
302	      Listener.close(listener)
303	    case 1n+f:
304	      IO.bind(Listener & Step.Got<Socket>, Unit, server.accept(listener),
305	        pair => server.loop.cont(handle,
306	          lst => server.loop.go(handle, f, lst), pair))
307	
308	# listen and serve up to `limit` connections with an IO handler
309	def server.serve_io(
310	  ~handle: Http.Request -> IO(Http.Reply),
311	  host: String,
312	  port: U32,
313	  limit: U32
314	) -> IO(Unit):
315	  IO.bind(Listener, Unit, server.listen(host, port),
316	    lst => server.loop.go(handle, U32.to_nat(limit), lst))
317	
318	# a pure handler as an IO one
319	def server.lift(
320	  ~handle: Http.Request -> Http.Reply,
321	  req: Http.Request
322	) -> IO(Http.Reply):
323	  IO.pure(Http.Reply, handle(req))
324	
325	# listen and serve up to `limit` connections
326	def server.serve(
327	  ~handle: Http.Request -> Http.Reply,
328	  host: String,
329	  port: U32,
330	  limit: U32
331	) -> IO(Unit):
332	  server.serve_io(~(req => server.lift(handle, req)), host, port, limit)
333	
334	# serve exactly one connection then stop
335	def server.once(
336	  ~handle: Http.Request -> Http.Reply,
337	  host: String,
338	  port: U32
339	) -> IO(Unit):
340	  server.serve(handle, host, port, 1)
341	
342	# the accept loop with no fuel: it runs until the process is stopped
343	# (HTTP-TRUST-2)
344	@unsafe def server.forever.go(
345	  ~handle: Http.Request -> IO(Http.Reply),
346	  listener: Listener
347	) -> IO(Unit):
348	  IO.bind(Listener & Step.Got<Socket>, Unit, server.accept(listener),
349	    pair => server.loop.cont(handle,
350	      lst => server.forever.go(handle, lst), pair))
351	
352	# listen and serve with an IO handler until the process is stopped
353	@unsafe def server.serve_forever(
354	  ~handle: Http.Request -> IO(Http.Reply),
355	  host: String,
356	  port: U32
357	) -> IO(Unit):
358	  IO.bind(Listener, Unit, server.listen(host, port),
359	    lst => server.forever.go(handle, lst))