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