1	# Typed RPC adapters between generated code and the byte-level transports (Srv, Cli).
2	# Reference: docs/runtime.md. Templates come first in every parameter list; the codec functions
3	# are the generated `Msg.decode(+codec, b)` / `Msg.encode(+codec, m)`, hence the `@+codec` arrows.
4	import Base
5	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
6	import ../code.bend as Code
7	import ../error.bend as E
8	import ../protocol.bend as P
9	import ./server.bend as Srv
10	import ./client.bend as Cli
11	
12	# ------------------------------------------------------------------ types
13	
14	# What a handler learns about the call it serves.
15	type Ctx is Data:
16	  Ctx{procedure: String, codec: P.Codec, headers: Map<&2, List<&2, String>>}
17	
18	# A typed server-stream sender: pass it through Rpc.send / the generated per-method send.
19	type Out is Type:
20	  Out{sink: Srv.Sink, codec: P.Codec}
21	
22	# ------------------------------------------------------------------ small helpers
23	
24	def err_reply(e: E.Error) -> Srv.Reply:
25	  Srv.Single{Fail{e}, Srv.meta.empty()}
26	
27	def bad_request(+message: String) -> Srv.Reply:
28	  Srv.fail(Code.InvalidArgument{}, message)
29	
30	def reply_io(r: Srv.Reply) -> IO(Srv.Reply):
31	  IO.pure(Srv.Reply, r)
32	
33	def no_custom(+name: String) -> Srv.Reply:
34	  Srv.fail(Code.Unimplemented{}, String.append("codec not supported: ", name))
35	
36	# Unknown procedure: the server answers 404.
37	def no_route(call: Srv.Call) -> IO(Srv.Reply):
38	  reply_io(Srv.NoRoute{})
39	
40	# Bidirectional streaming is not supported by the HTTP/1.1 transport.
41	def bidi_unimplemented(call: Srv.Call) -> IO(Srv.Reply):
42	  reply_io(Srv.fail(Code.Unimplemented{}, "bidirectional streaming is not supported"))
43	
44	def route.pick(~hit: Srv.Call -> IO(Srv.Reply), ~miss: Srv.Call -> IO(Srv.Reply), b: Bool, call: Srv.Call) -> IO(Srv.Reply):
45	  match b:
46	    case True{}:
47	      hit(call)
48	    case False{}:
49	      miss(call)
50	
51	# if call.procedure == procedure then hit(call) else miss(call); chain these, ending in no_route.
52	def route(~hit: Srv.Call -> IO(Srv.Reply), ~miss: Srv.Call -> IO(Srv.Reply), +procedure: String, call: Srv.Call) -> IO(Srv.Reply):
53	  Srv.Call{+p, kind, codec, headers, timeout_ms, messages} = call
54	  route.pick(~hit, ~miss, String.eq(p, procedure), Srv.Call{p, kind, codec, headers, timeout_ms, messages})
55	
56	# ------------------------------------------------------------------ server: unary
57	
58	def enc_res(~Res: Data, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, +codec: P.Codec, r: Result<&1, &1, E.Error, Res>) -> Result<&1, &1, E.Error, Bytes.Bytes>:
59	  match r:
60	    case Fail{e}:
61	      Fail{e}
62	    case Done{x}:
63	      Done{enc(codec, x)}
64	
65	def unary.reply(~Res: Data, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, +codec: P.Codec, r: Result<&1, &1, E.Error, Res>) -> Srv.Reply:
66	  Srv.Single{enc_res(~Res, ~enc, codec, r), Srv.meta.empty()}
67	
68	def unary.run(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> Req -> IO(Result<&1, &1, E.Error, Res>), ctx: Ctx, +codec: P.Codec, r: Result<&1, &1, E.Error, Req>) -> IO(Srv.Reply):
69	  match r:
70	    case Fail{e}:
71	      reply_io(err_reply(e))
72	    case Done{req}:
73	      do IO<Srv.Reply>:
74	        res : Result<&1, &1, E.Error, Res> <- impl(ctx, req)
75	        return unary.reply(~Res, ~enc, codec, res)
76	
77	def unary.one(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> Req -> IO(Result<&1, &1, E.Error, Res>), ctx: Ctx, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> IO(Srv.Reply):
78	  match ms:
79	    case Nil{}:
80	      reply_io(bad_request("expected exactly one request message"))
81	    case Con{m, rest}:
82	      match rest:
83	        case Nil{}:
84	          unary.run(~Req, ~Res, ~dec, ~enc, ~impl, ctx, codec, dec(codec, m))
85	        case Con{x, y}:
86	          reply_io(bad_request("expected exactly one request message"))
87	
88	# Serves a unary method: decode the single request message, run impl, encode the response.
89	# A decode error (already an E.Error from the generated decoder) is returned as is.
90	def unary(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> Req -> IO(Result<&1, &1, E.Error, Res>), call: Srv.Call) -> IO(Srv.Reply):
91	  Srv.Call{procedure, kind, +codec, headers, timeout_ms, messages} = call
92	  match codec:
93	    case P.Custom{name}:
94	      reply_io(no_custom(name))
95	    case P.Json{}:
96	      unary.one(~Req, ~Res, ~dec, ~enc, ~impl, Ctx{procedure, P.Json{}, headers}, P.Json{}, messages)
97	    case P.Proto{}:
98	      unary.one(~Req, ~Res, ~dec, ~enc, ~impl, Ctx{procedure, P.Proto{}, headers}, P.Proto{}, messages)
99	
100	# ------------------------------------------------------------------ server: server stream
101	
102	# Sends one typed message on a server stream.
103	def send.go(~T: Data, ~enc: @+codec: P.Codec -> T -> Bytes.Bytes, sink2: IO(Srv.Sink), codec: P.Codec) -> IO(Out):
104	  do IO<Out>:
105	    s : Srv.Sink <- sink2
106	    return Out{s, codec}
107	
108	def send(~T: Data, ~enc: @+codec: P.Codec -> T -> Bytes.Bytes, out: Out, msg: T) -> IO(Out):
109	  Out{sink, +codec} = out
110	  send.go(~T, ~enc, Srv.send(sink, enc(codec, msg)), codec)
111	
112	def ss.fin(r: Out & Result<&1, &1, E.Error, Unit>) -> Srv.Sink & Result<&1, &1, E.Error, Unit>:
113	  (out, res) = r
114	  Out{sink, codec} = out
115	  (sink, res)
116	
117	def ss.exec(~Req: Data, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, req: Req, sink: Srv.Sink) -> IO(Srv.Sink & Result<&1, &1, E.Error, Unit>):
118	  do IO<Srv.Sink & Result<&1, &1, E.Error, Unit>>:
119	    r : Out & Result<&1, &1, E.Error, Unit> <- impl(Ctx{procedure, codec, headers}, req, Out{sink, codec})
120	    return ss.fin(r)
121	
122	def ss.reply(~Req: Data, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, r: Result<&1, &1, E.Error, Req>) -> IO(Srv.Reply):
123	  match r:
124	    case Fail{e}:
125	      reply_io(err_reply(e))
126	    case Done{req}:
127	      reply_io(Srv.Many{sink => ss.exec(~Req, ~impl, procedure, codec, headers, req, sink), Srv.meta.empty()})
128	
129	def ss.one(~Req: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, ms: List<&1, Bytes.Bytes>) -> IO(Srv.Reply):
130	  match ms:
131	    case Nil{}:
132	      reply_io(bad_request("expected exactly one request message"))
133	    case Con{m, rest}:
134	      match rest:
135	        case Nil{}:
136	          ss.reply(~Req, ~impl, procedure, codec, headers, dec(codec, m))
137	        case Con{x, y}:
138	          reply_io(bad_request("expected exactly one request message"))
139	
140	# Serves a server-streaming method: impl gets the request and an Out to Rpc.send on, and returns
141	# the final Out with Done (stream ends cleanly) or Fail (error in the EndStream message).
142	def server_stream(~Req: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~impl: Ctx -> Req -> Out -> IO(Out & Result<&1, &1, E.Error, Unit>), call: Srv.Call) -> IO(Srv.Reply):
143	  Srv.Call{procedure, kind, +codec, headers, timeout_ms, messages} = call
144	  match codec:
145	    case P.Custom{name}:
146	      reply_io(no_custom(name))
147	    case P.Json{}:
148	      ss.one(~Req, ~dec, ~impl, procedure, P.Json{}, headers, messages)
149	    case P.Proto{}:
150	      ss.one(~Req, ~dec, ~impl, procedure, P.Proto{}, headers, messages)
151	
152	# ------------------------------------------------------------------ server: client stream
153	
154	def dec_all(~Req: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, List<&2, Req>>:
155	  match ms:
156	    case Nil{}:
157	      Done{Nil{}}
158	    case Con{m, t}:
159	      do Result<&1, &1, E.Error, List<&2, Req>>:
160	        x : Req <- dec(codec, m)
161	        rest : List<&2, Req> <- dec_all(~Req, ~dec, codec, t)
162	        return x <> rest
163	
164	def cs.run(~Req: Data, ~Res: Data, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> List<&2, Req> -> IO(Result<&1, &1, E.Error, Res>), ctx: Ctx, +codec: P.Codec, r: Result<&1, &1, E.Error, List<&2, Req>>) -> IO(Srv.Reply):
165	  match r:
166	    case Fail{e}:
167	      reply_io(err_reply(e))
168	    case Done{reqs}:
169	      do IO<Srv.Reply>:
170	        res : Result<&1, &1, E.Error, Res> <- impl(ctx, reqs)
171	        return unary.reply(~Res, ~enc, codec, res)
172	
173	# Serves a client-streaming method: all request messages are decoded first (the transport buffers
174	# them), impl answers with the single response.
175	def client_stream(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Req>, ~enc: @+codec: P.Codec -> Res -> Bytes.Bytes, ~impl: Ctx -> List<&2, Req> -> IO(Result<&1, &1, E.Error, Res>), call: Srv.Call) -> IO(Srv.Reply):
176	  Srv.Call{procedure, kind, +codec, headers, timeout_ms, messages} = call
177	  match codec:
178	    case P.Custom{name}:
179	      reply_io(no_custom(name))
180	    case P.Json{}:
181	      cs.run(~Req, ~Res, ~enc, ~impl, Ctx{procedure, P.Json{}, headers}, P.Json{}, dec_all(~Req, ~dec, P.Json{}, messages))
182	    case P.Proto{}:
183	      cs.run(~Req, ~Res, ~enc, ~impl, Ctx{procedure, P.Proto{}, headers}, P.Proto{}, dec_all(~Req, ~dec, P.Proto{}, messages))
184	
185	# ------------------------------------------------------------------ client
186	
187	def bad_response(+message: String) -> E.Error:
188	  E.new(Code.Unknown{}, message)
189	
190	def one_msg(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, Res>:
191	  match ms:
192	    case Nil{}:
193	      Fail{bad_response("no response message")}
194	    case Con{m, rest}:
195	      match rest:
196	        case Nil{}:
197	          dec(codec, m)
198	        case Con{x, y}:
199	          Fail{bad_response("more than one response message")}
200	
201	def one_res(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, r: Result<&1, &1, E.Error, Cli.Response>) -> Result<&1, &1, E.Error, Res>:
202	  match r:
203	    case Fail{e}:
204	      Fail{e}
205	    case Done{Cli.Response{h, t, ms}}:
206	      one_msg(~Res, ~dec, codec, ms)
207	
208	def dec_msgs(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, ms: List<&1, Bytes.Bytes>) -> Result<&1, &1, E.Error, List<&2, Res>>:
209	  dec_all(~Res, ~dec, codec, ms)
210	
211	def many_res(~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, +codec: P.Codec, r: Result<&1, &1, E.Error, Cli.Response>) -> Result<&1, &1, E.Error, List<&2, Res>>:
212	  match r:
213	    case Fail{e}:
214	      Fail{e}
215	    case Done{Cli.Response{h, t, ms}}:
216	      dec_msgs(~Res, ~dec, codec, ms)
217	
218	def enc_all(~Req: Data, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, +codec: P.Codec, xs: List<&2, Req>) -> List<&1, Bytes.Bytes>:
219	  match xs:
220	    case Nil{}:
221	      Nil{}
222	    case Con{x, t}:
223	      enc(codec, x) <> enc_all(~Req, ~enc, codec, t)
224	
225	# Unary call. headers: request headers (lowercase keys). Errors: transport failure -> Unavailable,
226	# non-200 -> the server's error, undecodable response -> the decoder's error.
227	def call.unary(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, req: Req) -> IO(Result<&1, &1, E.Error, Res>):
228	  do IO<Result<&1, &1, E.Error, Res>>:
229	    r : Result<&1, &1, E.Error, Cli.Response> <- Cli.unary(base_url, procedure, codec, headers, enc(codec, req))
230	    return one_res(~Res, ~dec, codec, r)
231	
232	# Server-streaming call; collects every message (an error end-stream turns the whole call into Fail).
233	def call.server_stream(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, req: Req) -> IO(Result<&1, &1, E.Error, List<&2, Res>>):
234	  do IO<Result<&1, &1, E.Error, List<&2, Res>>>:
235	    r : Result<&1, &1, E.Error, Cli.Response> <- Cli.server_stream(base_url, procedure, codec, headers, enc(codec, req))
236	    return many_res(~Res, ~dec, codec, r)
237	
238	# Client-streaming call: sends all reqs, answers the single response.
239	def call.client_stream(~Req: Data, ~Res: Data, ~dec: @+codec: P.Codec -> Bytes.Bytes -> Result<&1, &1, E.Error, Res>, ~enc: @+codec: P.Codec -> Req -> Bytes.Bytes, base_url: String, procedure: String, +codec: P.Codec, headers: Map<&2, List<&2, String>>, reqs: List<&2, Req>) -> IO(Result<&1, &1, E.Error, Res>):
240	  do IO<Result<&1, &1, E.Error, Res>>:
241	    r : Result<&1, &1, E.Error, Cli.Response> <- Cli.client_stream(base_url, procedure, codec, headers, enc_all(~Req, ~enc, codec, reqs))
242	    return one_res(~Res, ~dec, codec, r)