1	# ezhttp/src/wire: one request out and one response back, over
2	# bend-kit-wire's sockets. Its effects park on the runtime's event loop, so
3	# exchanges run side by side under IO.fork. bend-kit-dns resolves the host;
4	# TLS is bend-kit-wire's (OpenSSL 3 loaded at run time, BEND_LIBSSL to point
5	# at it).
6	#
7	# An exchange's answer is a status on its first line, then text: "0" and the
8	# raw response, or a non-zero status and the reason (wirespec.bend
9	# wire.fail). Every decision is wirespec.bend's: the deadline (wire.ms.of),
10	# the octets sent (wire.payload), what a failed call is reported as
11	# (wire.fail) and when the response is whole (wire.read.next), so a server
12	# that keeps the connection open after a complete response is not waited on
13	# (SPEC.md HTTP-CLI-1, 6, 10, 13; HTTP-TRUST-3, 5).
14	import Base
15	import bend-kit-wire@0.4.6.1/wire.bend as W
16	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
17	import bend-kit-dns@0.6.0.2/dns.bend as Dns
18	import ./wirespec.bend as Spec
19	
20	# the most octets one receive asks for
21	def wire.max() -> U32:
22	  65536
23	
24	# the most receives one exchange makes
25	def wire.reads() -> Nat:
26	  65536n
27	
28	# --- the socket calls, plain or over TLS ---
29	
30	# close a socket, ending its TLS session when it has one
31	def wire.close(secure: Bool, s: Socket) -> IO(Unit):
32	  match secure:
33	    case True{}:
34	      W.tls.close(s)
35	    case False{}:
36	      Socket.close(s)
37	
38	# the next octets a socket has received, packed four to a word
39	def wire.recv(secure: Bool, s: Socket, ms: U32)
40	  -> IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
41	  match secure:
42	    case True{}:
43	      W.tls.recv.words(s, wire.max(), ms)
44	    case False{}:
45	      W.recv.words(s, wire.max(), ms)
46	
47	# a send.words.deadline answer without its count of octets accepted
48	def wire.sent(r: Socket & (U32 & Result<&1, &1, U32 & String, Unit>))
49	  -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
50	  (s, (n, res)) = r
51	  IO.pure(Socket & Result<&1, &1, U32 & String, Unit>, (s, res))
52	
53	# send octets, the plain send bounded by the deadline
54	def wire.send(secure: Bool, s: Socket, +len: U32, buf: Array<U32>, ms: U32)
55	  -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
56	  match secure:
57	    case True{}:
58	      W.tls.send.words(s, len, buf)
59	    case False{}:
60	      IO.bind(Socket & (U32 & Result<&1, &1, U32 & String, Unit>),
61	        Socket & Result<&1, &1, U32 & String, Unit>,
62	        W.send.words.deadline(s, len, buf, ms), wire.sent)
63	
64	# the TLS handshake with the host, when the exchange is secure
65	def wire.shake(secure: Bool, s: Socket, host: String, ms: U32)
66	  -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
67	  match secure:
68	    case True{}:
69	      W.tls.connect(s, host, ms)
70	    case False{}:
71	      IO.pure(Socket & Result<&1, &1, U32 & String, Unit>, (s, Done{Unit{}}))
72	
73	# --- reading the answer ---
74	
75	# a judged read: read again, or close and answer
76	def wire.loop.on(+secure: Bool, s: Socket, r: Spec.Read,
77	  more: Socket -> Spec.Got -> IO(String)) -> IO(String):
78	  match r:
79	    case Spec.Again{acc}:
80	      more(s)(acc)
81	    case Spec.Heard{answer}:
82	      IO.bind(Unit, String, wire.close(secure, s),
83	        _u => IO.pure(String, answer))
84	
85	# a receive's octets, or the server's close (no octets)
86	def wire.loop.got(+secure: Bool, s: Socket, acc: Spec.Got, len: U32,
87	  buf: Array<U32>, more: Socket -> Spec.Got -> IO(String)) -> IO(String):
88	  wire.loop.on(secure, s,
89	    Spec.wire.read.next(acc, "0\n" ++ Bytes.to_string(Bytes.Bytes{len, buf})),
90	    more)
91	
92	# a receive's outcome: a failure is the answer
93	def wire.loop.un(+secure: Bool, +host: String, +port: U32, +ms: U32,
94	  acc: Spec.Got,
95	  r: Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>,
96	  more: Socket -> Spec.Got -> IO(String)) -> IO(String):
97	  (s, res) = r
98	  match res:
99	    case Fail{(n, why)}:
100	      IO.bind(Unit, String, wire.close(secure, s),
101	        _u => IO.pure(String,
102	          Spec.wire.fail(Spec.Receive{}, host, port, ms, n, why)))
103	    case Done{(len, buf)}:
104	      wire.loop.got(secure, s, acc, len, buf, more)
105	
106	# receive until wirespec.bend says the answer is whole
107	def wire.loop(fuel: Nat, +secure: Bool, s: Socket, +host: String, +port: U32,
108	  +ms: U32, acc: Spec.Got) -> IO(String):
109	  match fuel:
110	    case 0n:
111	      IO.bind(Unit, String, wire.close(secure, s),
112	        _u => IO.pure(String, Spec.wire.spent()))
113	    case 1n+p:
114	      IO.bind(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>, String,
115	        wire.recv(secure, s, ms),
116	        r => wire.loop.un(secure, host, port, ms, acc, r,
117	          s2 => a2 => wire.loop(p, secure, s2, host, port, ms, a2)))
118	
119	# --- one exchange ---
120	
121	# after the send: read the answer, or the send's failure
122	def wire.after.send(+secure: Bool, +host: String, +port: U32, +ms: U32,
123	  +method: String, r: Socket & Result<&1, &1, U32 & String, Unit>)
124	  -> IO(String):
125	  (s, res) = r
126	  match res:
127	    case Fail{(n, why)}:
128	      IO.bind(Unit, String, wire.close(secure, s),
129	        _u => IO.pure(String,
130	          Spec.wire.fail(Spec.Send{}, host, port, ms, n, why)))
131	    case Done{_u}:
132	      wire.loop(wire.reads(), secure, s, host, port, ms,
133	        Spec.wire.got(method, []))
134	
135	# send the octets an exchange carries, then read its answer
136	def wire.put(+secure: Bool, s: Socket, +host: String, +port: U32, +ms: U32,
137	  +method: String, payload: Bytes.Bytes) -> IO(String):
138	  Bytes.Bytes{+len, buf} = payload
139	  IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
140	    wire.send(secure, s, len, buf, ms),
141	    r => wire.after.send(secure, host, port, ms, method, r))
142	
143	# after the handshake: send, or the handshake's failure (the socket closed
144	# plainly, as no TLS session was made)
145	def wire.after.shake(+secure: Bool, +host: String, +port: U32, +ms: U32,
146	  +method: String, payload: Bytes.Bytes,
147	  r: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(String):
148	  (s, res) = r
149	  match res:
150	    case Fail{(n, why)}:
151	      IO.bind(Unit, String, Socket.close(s),
152	        _u => IO.pure(String,
153	          Spec.wire.fail(Spec.Handshake{}, host, port, ms, n, why)))
154	    case Done{_u}:
155	      wire.put(secure, s, host, port, ms, method, payload)
156	
157	# after the connect: the handshake, or the connect's failure
158	def wire.after.connect(+secure: Bool, +host: String, +port: U32, +ms: U32,
159	  +method: String, payload: Bytes.Bytes,
160	  r: Result<&1, &1, U32 & String, Socket>) -> IO(String):
161	  match r:
162	    case Fail{(n, why)}:
163	      IO.pure(String, Spec.wire.fail(Spec.Connect{}, host, port, ms, n, why))
164	    case Done{s}:
165	      IO.bind(Socket & Result<&1, &1, U32 & String, Unit>, String,
166	        wire.shake(secure, s, host, ms),
167	        r2 => wire.after.shake(secure, host, port, ms, method, payload, r2))
168	
169	# a connect that failed tries the next address; one that worked is kept
170	def wire.dial.next(r: Result<&1, &1, U32 & String, Socket>,
171	  more: Unit -> IO(Result<&1, &1, U32 & String, Socket>))
172	  -> IO(Result<&1, &1, U32 & String, Socket>):
173	  match r:
174	    case Done{s}:
175	      IO.pure(Result<&1, &1, U32 & String, Socket>, Done{s})
176	    case Fail{e}:
177	      more(Unit{})
178	
179	# connect to each address in turn until one connects; the last one's
180	# failure when none does
181	def wire.dial(rest: List<&2, String>, +first: String, +port: U32, +ms: U32)
182	  -> IO(Result<&1, &1, U32 & String, Socket>):
183	  match rest:
184	    case []:
185	      W.connect(first, port, ms)
186	    case ip <> t:
187	      IO.bind(Result<&1, &1, U32 & String, Socket>,
188	        Result<&1, &1, U32 & String, Socket>,
189	        W.connect(first, port, ms),
190	        r => wire.dial.next(r, _u => wire.dial(t, ip, port, ms)))
191	
192	# after the resolve: connect, or the resolve's failure (no address at all is
193	# ENOENT, as bend-kit-dns reports a name with none)
194	def wire.after.resolve(+secure: Bool, +host: String, +port: U32, +ms: U32,
195	  +method: String, payload: Bytes.Bytes,
196	  r: Result<&1, &1, U32 & String, List<&2, String>>) -> IO(String):
197	  match r:
198	    case Fail{(n, why)}:
199	      IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, n, why))
200	    case Done{[]}:
201	      IO.pure(String, Spec.wire.fail(Spec.Resolve{}, host, port, ms, 2, ""))
202	    case Done{ip <> t}:
203	      IO.bind(Result<&1, &1, U32 & String, Socket>, String,
204	        wire.dial(t, ip, port, ms),
205	        c => wire.after.connect(secure, host, port, ms, method, payload, c))
206	
207	# the OS resolver's addresses, NUL-separated, as a list
208	def wire.looked(r: Result<&1, &1, U32 & String, String>)
209	  -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
210	  match r:
211	    case Fail{e}:
212	      IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Fail{e})
213	    case Done{text}:
214	      IO.pure(Result<&1, &1, U32 & String, List<&2, String>>,
215	        Done{Dns.lookup.split(text)})
216	
217	# a host's addresses: a numeric address or localhost as it is, else the OS
218	# resolver's answer
219	def wire.resolve(+host: String, literal: List<&2, String>)
220	  -> IO(Result<&1, &1, U32 & String, List<&2, String>>):
221	  match literal:
222	    case []:
223	      IO.bind(Result<&1, &1, U32 & String, String>,
224	        Result<&1, &1, U32 & String, List<&2, String>>,
225	        Dns.lookup.all(String.to_lower(host)), wire.looked)
226	    case ip <> t:
227	      IO.pure(Result<&1, &1, U32 & String, List<&2, String>>, Done{ip <> t})
228	
229	# the exchange once its octets are packed
230	def wire.run.with(+secure: Bool, +ms: U32, +host: String, +port: U32,
231	  +method: String, payload: Bytes.Bytes) -> IO(String):
232	  IO.bind(Result<&1, &1, U32 & String, List<&2, String>>, String,
233	    wire.resolve(host, Dns.resolve.literal.all(host)),
234	    r => wire.after.resolve(secure, host, port, ms, method, payload, r))
235	
236	# a request with this method, sent to a host as wirespec.bend says, with
237	# its tail of octets after it, every call bounded by the deadline; its whole
238	# answer
239	def wire.run(+secure: Bool, +ms: U32, +host: String, +port: U32,
240	  +method: String, req: String, tail: Bytes.Bytes) -> IO(String):
241	  wire.run.with(secure, ms, host, port, method,
242	    Bytes.append(Bytes.from_string(Spec.wire.payload(req)), tail))
243	
244	# a request sent to a host, its tail of octets after it, and its whole answer
245	# read back, with the default deadline
246	def wire.talk(secure: Bool, host: String, port: U32, method: String,
247	  req: String, tail: Bytes.Bytes) -> IO(String):
248	  wire.run(secure, Spec.wire.ms.of(None{}), host, port, method, req, tail)
249	
250	# the same exchange, bounded by a timeout in milliseconds
251	def wire.talk.timed(secure: Bool, ms: U32, host: String, port: U32,
252	  method: String, req: String, tail: Bytes.Bytes) -> IO(String):
253	  wire.run(secure, Spec.wire.ms.of(Some{ms}), host, port, method, req, tail)