1/**
2 * Server-side layers for workflow proxy APIs.
3 *
4 * `layerHttpApi` connects the HTTP API group created by `WorkflowProxy` to the
5 * supplied workflows. `layerRpcHandlers` does the same for the generated RPC
6 * definitions. Both layers route execute, discard, and resume requests to the
7 * matching workflow operation, while the `WorkflowEngine` and workflow handler
8 * services stay on the server side.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import type { NonEmptyReadonlyArray } from "../Array.ts"
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import type * as HttpApi from "../http-api/HttpApi.ts"
17import * as HttpApiBuilder from "../http-api/HttpApiBuilder.ts"
18import type * as HttpApiGroup from "../http-api/HttpApiGroup.ts"
19import * as Layer from "../Layer.ts"
20import type * as Rpc from "../rpc/Rpc.ts"
21import type * as Workflow from "./Workflow.ts"
22import type { WorkflowEngine } from "./WorkflowEngine.ts"
23
24/**
25 * Creates handlers for a workflow HTTP API group, wiring execute, discard, and
26 * resume endpoints to the supplied workflows.
27 *
28 * @stability unstable
29 * @category layers
30 * @since 4.0.0
31 */
32export const layerHttpApi = <
33 ApiId extends string,
34 Groups extends HttpApiGroup.Constraint,
35 Identifier extends HttpApiGroup.Identifier<Groups>,
36 const Workflows extends NonEmptyReadonlyArray<Workflow.Any>
37>(
38 api: HttpApi.HttpApi<ApiId, Groups>,
39 identifier: Identifier,
40 workflows: Workflows
41): Layer.Layer<
42 HttpApiGroup.Service<ApiId, Identifier>,
43 never,
44 WorkflowEngine | Workflow.RequirementsHandler<Workflows[number]>
45> =>
46 HttpApiBuilder.group(
47 api,
48 identifier,
49 Effect.fnUntraced(function*(handlers: any) {
50 for (const workflow_ of workflows) {
51 const workflow = workflow_ as Workflow.AnyWithProps
52 handlers = handlers
53 .handle(
54 workflow._tag,
55 ({ payload }: { payload: any }) =>
56 workflow.execute(payload).pipe(
57 Effect.tapDefect(Effect.logError),
58 Effect.annotateLogs({
59 module: "WorkflowProxyServer",
60 method: workflow._tag
61 })
62 )
63 )
64 .handle(
65 workflow._tag + "Discard",
66 ({ payload }: { payload: any }) =>
67 workflow.execute(payload, { discard: true }).pipe(
68 Effect.tapDefect(Effect.logError),
69 Effect.annotateLogs({
70 module: "WorkflowProxyServer",
71 method: workflow._tag + "Discard"
72 })
73 )
74 )
75 .handle(
76 workflow._tag + "Resume",
77 ({ payload }: { payload: any }) =>
78 workflow.resume(payload.executionId).pipe(
79 Effect.tapDefect(Effect.logError),
80 Effect.annotateLogs({
81 module: "WorkflowProxyServer",
82 method: workflow._tag + "Resume"
83 })
84 )
85 )
86 }
87 return handlers as HttpApiBuilder.Handlers<never>
88 })
89 )
90
91/**
92 * Creates RPC handlers for the supplied workflows, wiring execute, discard,
93 * and resume RPCs to workflow operations.
94 *
95 * @stability unstable
96 * @category layers
97 * @since 4.0.0
98 */
99export const layerRpcHandlers = <
100 const Workflows extends NonEmptyReadonlyArray<Workflow.Any>,
101 const Prefix extends string = ""
102>(workflows: Workflows, options?: {
103 readonly prefix?: Prefix
104}): Layer.Layer<
105 RpcHandlers<Workflows[number], Prefix>,
106 never,
107 WorkflowEngine | Workflow.RequirementsHandler<Workflows[number]>
108> =>
109 Layer.effectContext(Effect.gen(function*() {
110 const context = yield* Effect.context<never>()
111 const prefix = options?.prefix ?? ""
112 const handlers = new Map<string, Rpc.Handler<string>>()
113 for (const workflow_ of workflows) {
114 const workflow = workflow_ as Workflow.AnyWithProps
115 const tag = `${prefix}${workflow._tag}`
116 const tagDiscard = `${tag}Discard`
117 const tagResume = `${tag}Resume`
118 const key = `effect/rpc/Rpc/${tag}`
119 const keyDiscard = `${key}Discard`
120 const keyResume = `${key}Resume`
121 handlers.set(key, {
122 context,
123 tag,
124 handler: (payload: any) => workflow.execute(payload) as any
125 } as any)
126 handlers.set(keyDiscard, {
127 context,
128 tag: tagDiscard,
129 handler: (payload: any) => workflow.execute(payload, { discard: true } as any) as any
130 } as any)
131 handlers.set(keyResume, {
132 context,
133 tag: tagResume,
134 handler: (payload: any) => workflow.resume(payload.executionId) as any
135 } as any)
136 }
137 return Context.makeUnsafe(handlers)
138 }))
139
140/**
141 * Union of RPC handler services required to serve the generated workflow
142 * execute, discard, and resume RPCs.
143 *
144 * @stability unstable
145 * @category services
146 * @since 4.0.0
147 */
148export type RpcHandlers<Workflows extends Workflow.Any, Prefix extends string> = Workflows extends Workflow.Workflow<
149 infer _Name,
150 infer _Payload,
151 infer _Success,
152 infer _Error
153> ? Rpc.Handler<`${Prefix}${_Name}`> | Rpc.Handler<`${Prefix}${_Name}Discard`> | Rpc.Handler<`${Prefix}${_Name}Resume`>
154 : never