1/**
2 * Connects typed RPC clients to atoms.
3 *
4 * The service created here exposes a flattened RPC client plus atom-based query
5 * and mutation helpers. Query atoms call RPCs and track asynchronous or
6 * streaming results, while mutations run RPC calls that can invalidate
7 * reactivity keys after success. Query atoms can also use request headers,
8 * time-to-live settings, and serialization keys for hydration.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Context from "../Context.ts"
14import * as Duration from "../Duration.ts"
15import * as Effect from "../Effect.ts"
16import * as Headers from "../http/Headers.ts"
17import * as Layer from "../Layer.ts"
18import type { ReadonlyRecord } from "../Record.ts"
19import type * as Rpc from "../rpc/Rpc.ts"
20import * as RpcClient from "../rpc/RpcClient.ts"
21import { RpcClientError } from "../rpc/RpcClientError.ts"
22import type * as RpcGroup from "../rpc/RpcGroup.ts"
23import type { RequestId } from "../rpc/RpcMessage.ts"
24import * as RpcSchema from "../rpc/RpcSchema.ts"
25import * as Schema from "../Schema.ts"
26import type { Scope } from "../Scope.ts"
27import * as Stream from "../Stream.ts"
28import type { Mutable, NoInfer } from "../Types.ts"
29import * as AsyncResult from "./AsyncResult.ts"
30import * as Atom from "./Atom.ts"
31import * as Reactivity from "./Reactivity.ts"
32
33/**
34 * A `Context.Service` for a flattened RPC client integrated with atom reactivity.
35 *
36 * **Details**
37 *
38 * It exposes the RPC client, an atom runtime, mutation helpers that return `AtomResultFn`s, and query helpers that
39 * return atoms or pull atoms for RPC results.
40 *
41 * @stability unstable
42 * @category services
43 * @since 4.0.0
44 */
45export interface AtomRpcClient<Self, Id extends string, Rpcs extends Rpc.Any> extends
46 Context.Service<
47 Self,
48 RpcClient.RpcClient.Flat<Rpcs, RpcClientError>
49 >
50{
51 new(_: never): Context.ServiceClass.Shape<
52 Id,
53 RpcClient.RpcClient.Flat<Rpcs, RpcClientError>
54 >
55
56 readonly runtime: Atom.AtomRuntime<Self>
57
58 readonly mutation: <Tag extends Rpc.Tag<Rpcs>>(
59 arg: Tag
60 ) => Rpcs extends Rpc.Rpc<
61 infer _Tag extends Tag,
62 infer _Payload,
63 infer _Success,
64 infer _Error,
65 infer _Middleware,
66 infer _Requires
67 > ? [_Success] extends [RpcSchema.Stream<infer _A, infer _E>] ? never
68 : Atom.AtomResultFn<
69 {
70 readonly payload: Rpc.PayloadConstructor<Rpcs>
71 readonly reactivityKeys?:
72 | ReadonlyArray<unknown>
73 | ReadonlyRecord<string, ReadonlyArray<unknown>>
74 | undefined
75 readonly headers?: Headers.Input | undefined
76 },
77 _Success["Type"],
78 _Error["Type"] | RpcClientError | _Middleware["error"]["Type"] | _Middleware["~ClientError"]
79 >
80 : never
81
82 readonly query: <Tag extends Rpc.Tag<Rpcs>>(
83 tag: Tag,
84 payload: Rpc.PayloadConstructor<Rpc.ExtractTag<Rpcs, Tag>>,
85 options?: {
86 readonly headers?: Headers.Input | undefined
87 readonly reactivityKeys?:
88 | ReadonlyArray<unknown>
89 | ReadonlyRecord<string, ReadonlyArray<unknown>>
90 | undefined
91 readonly timeToLive?: Duration.Input | undefined
92 readonly serializationKey?: string | undefined
93 }
94 ) => Rpcs extends Rpc.Rpc<
95 infer _Tag extends Tag,
96 infer _Payload,
97 infer _Success,
98 infer _Error,
99 infer _Middleware,
100 infer _Requires
101 > ? [_Success] extends [RpcSchema.Stream<infer _A, infer _E>] ? Atom.Writable<
102 Atom.PullResult<
103 _A["Type"],
104 _E["Type"] | _Error["Type"] | RpcClientError | _Middleware["error"]["Type"] | _Middleware["~ClientError"]
105 >,
106 void
107 >
108 : Atom.Atom<
109 AsyncResult.AsyncResult<
110 _Success["Type"],
111 _Error["Type"] | RpcClientError | _Middleware["error"]["Type"] | _Middleware["~ClientError"]
112 >
113 >
114 : never
115}
116
117declare global {
118 interface ErrorConstructor {
119 stackTraceLimit: number
120 }
121}
122
123/**
124 * Creates a `Context.Service` class for an RPC client backed by an atom runtime.
125 *
126 * **Details**
127 *
128 * The options provide the RPC group, protocol layer, tracing options, request id generation, optional custom client
129 * effect, and runtime factory used by the query and mutation helpers.
130 *
131 * @stability unstable
132 * @category constructors
133 * @since 4.0.0
134 */
135export const Service = <Self>() =>
136<
137 const Id extends string,
138 Rpcs extends Rpc.Any,
139 ER,
140 RM =
141 | RpcClient.Protocol
142 | Rpc.MiddlewareClient<NoInfer<Rpcs>>
143 | Rpc.ServicesClient<NoInfer<Rpcs>>
144>(
145 id: Id,
146 options: {
147 readonly group: RpcGroup.RpcGroup<Rpcs>
148 readonly protocol:
149 | Layer.Layer<Exclude<NoInfer<RM>, Scope>, ER>
150 | ((get: Atom.AtomContext) => Layer.Layer<Exclude<NoInfer<RM>, Scope>, ER>)
151 readonly spanPrefix?: string | undefined
152 readonly spanAttributes?: Record<string, unknown> | undefined
153 readonly generateRequestId?: (() => RequestId) | undefined
154 readonly disableTracing?: boolean | undefined
155 readonly makeEffect?:
156 | Effect.Effect<
157 RpcClient.RpcClient.Flat<Rpcs, RpcClientError>,
158 never,
159 RM
160 >
161 | undefined
162 readonly runtime?: Atom.RuntimeFactory | undefined
163 }
164): AtomRpcClient<Self, Id, Rpcs> => {
165 const self: Mutable<AtomRpcClient<Self, Id, Rpcs>> = Context.Service<
166 Self,
167 RpcClient.RpcClient.Flat<Rpcs, RpcClientError>
168 >()(id) as any
169
170 const layer = Layer.effect(
171 self,
172 options.makeEffect ??
173 (RpcClient.make(options.group, {
174 ...options,
175 flatten: true
176 }) as Effect.Effect<
177 RpcClient.RpcClient.Flat<Rpcs, RpcClientError>,
178 never,
179 RM
180 >)
181 )
182 const runtimeFactory = options.runtime ?? Atom.runtime
183 self.runtime = runtimeFactory(
184 typeof options.protocol === "function" ?
185 (get) =>
186 Layer.provide(
187 layer,
188 Layer.orDie(
189 (options.protocol as ((get: Atom.AtomContext) => Layer.Layer<Exclude<NoInfer<RM>, Scope>, ER>))(get)
190 )
191 ) :
192 Layer.provide(layer, Layer.orDie(options.protocol))
193 )
194
195 self.mutation = Atom.family(<Tag extends Rpc.Tag<Rpcs>>(tag: Tag) => {
196 const rpc = options.group.requests.get(tag)! as any as Rpc.AnyWithProps
197 return self.runtime.fn<{
198 readonly payload: Rpc.PayloadConstructor<Rpc.ExtractTag<Rpcs, Tag>>
199 readonly reactivityKeys?:
200 | ReadonlyArray<unknown>
201 | ReadonlyRecord<string, ReadonlyArray<unknown>>
202 | undefined
203 readonly headers?: Headers.Input | undefined
204 }>()(
205 Effect.fnUntraced(function*({ headers, payload, reactivityKeys }) {
206 const client = yield* self
207 const effect = client(tag, payload, { headers } as any)
208 return yield* (reactivityKeys
209 ? Reactivity.mutation(effect, reactivityKeys) as Effect.Effect<any>
210 : effect as any as Effect.Effect<any>)
211 })
212 ).pipe(
213 Atom.serializable({
214 key: `AtomRpc:mutation:${tag}`,
215 schema: AsyncResult.Schema({
216 success: rpc.successSchema,
217 error: makeErrorSchema(rpc)
218 }) as any
219 })
220 )
221 }) as any
222
223 const queryFamily = Atom.family(
224 (key: QueryKey) => {
225 const { headers, payload, reactivityKeys, tag, timeToLive } = key
226 const rpc = options.group.requests.get(tag)! as any as Rpc.AnyWithProps
227 const isStream = RpcSchema.isStreamSchema(rpc.successSchema)
228 let atom = isStream
229 ? self.runtime.pull(
230 Stream.unwrap(
231 self.use((client) =>
232 Effect.succeed(
233 client(tag, payload, { headers } as any) as any
234 )
235 )
236 )
237 )
238 : self.runtime.atom(
239 self.use((client) => client(tag, payload, { headers } as any)) as any
240 )
241 if (reactivityKeys) {
242 atom = self.runtime.factory.withReactivity(reactivityKeys)(atom)
243 }
244 if (!isStream && key.serializationKey) {
245 atom = Atom.serializable(atom, {
246 key: `AtomRpc:${key.tag}:${key.serializationKey}`,
247 schema: AsyncResult.Schema({
248 success: rpc.successSchema,
249 error: makeErrorSchema(rpc)
250 }) as any
251 })
252 }
253 if (timeToLive) {
254 atom = Duration.isFinite(timeToLive)
255 ? Atom.setIdleTTL(atom, timeToLive)
256 : Atom.keepAlive(atom)
257 }
258 return atom
259 }
260 )
261
262 self.query = <Tag extends Rpc.Tag<Rpcs>>(
263 tag: Tag,
264 payload: Rpc.PayloadConstructor<Rpc.ExtractTag<Rpcs, Tag>>,
265 options?: {
266 readonly headers?: Headers.Input | undefined
267 readonly reactivityKeys?:
268 | ReadonlyArray<unknown>
269 | ReadonlyRecord<string, ReadonlyArray<unknown>>
270 | undefined
271 readonly timeToLive?: Duration.Input | undefined
272 readonly serializationKey?: string | undefined
273 }
274 ) => {
275 const key: QueryKey = {
276 tag,
277 payload,
278 headers: options?.headers
279 ? Headers.fromInput(options.headers)
280 : undefined,
281 reactivityKeys: options?.reactivityKeys,
282 timeToLive: options?.timeToLive !== undefined
283 ? Duration.fromInputUnsafe(options.timeToLive)
284 : undefined,
285 serializationKey: options?.serializationKey
286 }
287 return queryFamily(key) as any
288 }
289
290 return self as AtomRpcClient<Self, Id, Rpcs>
291}
292
293interface QueryKey {
294 tag: string
295 payload: any
296 headers: Headers.Headers | undefined
297 reactivityKeys:
298 | ReadonlyArray<unknown>
299 | ReadonlyRecord<string, ReadonlyArray<unknown>>
300 | undefined
301 timeToLive: Duration.Duration | undefined
302 serializationKey: string | undefined
303}
304
305const makeErrorSchema = (rpc: Rpc.AnyWithProps): Schema.Top =>
306 Schema.Union([
307 rpc.errorSchema,
308 ...Array.from(rpc.middlewares, (middleware) => middleware.error),
309 RpcClientError
310 ])