1/**
2 * Keeps resources available across cluster entity restarts.
3 *
4 * `EntityResource` is useful for long-lived resources tied to an entity
5 * address, such as external processes, network clients, Kubernetes Pods, or
6 * other handles that should survive routine shard movement. This module
7 * includes the resource wrapper, a close scope that survives normal entity
8 * restarts, a generic resource constructor, and a Kubernetes Pod resource
9 * helper built on `K8sHttpClient`.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Context from "../Context.ts"
15import * as Duration from "../Duration.ts"
16import * as Effect from "../Effect.ts"
17import { identity } from "../Function.ts"
18import * as RcRef from "../RcRef.ts"
19import * as Scope from "../Scope.ts"
20import * as Entity from "./Entity.ts"
21import * as K8sHttpClient from "./K8sHttpClient.ts"
22import type { Pod as K8sPod } from "./K8sTypes.ts"
23import type { Sharding } from "./Sharding.ts"
24
25/**
26 * Type identifier used to brand `EntityResource` values.
27 *
28 * @stability unstable
29 * @category type IDs
30 * @since 4.0.0
31 */
32export const TypeId: TypeId = "~effect/cluster/EntityResource"
33
34/**
35 * Literal type of the `EntityResource` type identifier.
36 *
37 * @stability unstable
38 * @category type IDs
39 * @since 4.0.0
40 */
41export type TypeId = "~effect/cluster/EntityResource"
42
43/**
44 * A resource acquired inside a cluster entity and kept alive across restarts.
45 *
46 * **Details**
47 *
48 * `get` acquires or reuses the resource in the caller's scope, while `close`
49 * invalidates it so its close scope can be released.
50 *
51 * @stability unstable
52 * @category models
53 * @since 4.0.0
54 */
55export interface EntityResource<out A, out E = never> {
56 readonly [TypeId]: TypeId
57 readonly get: Effect.Effect<A, E, Scope.Scope>
58 readonly close: Effect.Effect<void>
59}
60
61/**
62 * Context service for a Scope that is only closed when the resource is explicitly closed.
63 *
64 * **When to use**
65 *
66 * Use when a cluster entity resource needs a scope that survives restarts and
67 * closes only through the resource lifecycle.
68 *
69 * **Gotchas**
70 *
71 * It is not closed during restarts, due to shard movement or node shutdowns.
72 *
73 * @stability unstable
74 * @category services
75 * @since 4.0.0
76 */
77export class CloseScope extends Context.Service<
78 CloseScope,
79 Scope.Scope
80>()("effect/cluster/EntityResource/CloseScope") {}
81
82/**
83 * Creates an `EntityResource` that can be acquired inside a cluster entity.
84 *
85 * **When to use**
86 *
87 * Use when a cluster entity should lazily share an acquired resource across
88 * messages and release it only on idle timeout or explicit close.
89 *
90 * **Details**
91 *
92 * The resource will only be fully released when the idle time to live is
93 * reached, or when the `close` effect is called.
94 *
95 * **Gotchas**
96 *
97 * By default, the `idleTimeToLive` is infinite, meaning the resource will only
98 * be released when `close` is called.
99 *
100 * @stability unstable
101 * @category constructors
102 * @since 4.0.0
103 */
104export const make: <A, E, R>(options: {
105 readonly acquire: Effect.Effect<A, E, R>
106 readonly idleTimeToLive?: Duration.Input | undefined
107 readonly acquireEagerly?: boolean | undefined
108}) => Effect.Effect<
109 EntityResource<A, E>,
110 E,
111 Scope.Scope | Exclude<R, CloseScope> | Sharding | Entity.CurrentAddress
112> = Effect.fnUntraced(function*<A, E, R>(options: {
113 readonly acquire: Effect.Effect<A, E, R>
114 readonly idleTimeToLive?: Duration.Input | undefined
115 readonly acquireEagerly?: boolean | undefined
116}) {
117 let shuttingDown = false
118
119 const ref = yield* RcRef.make({
120 acquire: Effect.gen(function*() {
121 yield* Entity.keepAlive(true)
122
123 const closeable = yield* Scope.make()
124
125 yield* Effect.addFinalizer(
126 Effect.fnUntraced(function*(exit) {
127 if (shuttingDown) return
128 yield* Scope.close(closeable, exit)
129 yield* Entity.keepAlive(false)
130 })
131 )
132
133 return yield* options.acquire.pipe(
134 Effect.provideService(CloseScope, closeable)
135 )
136 }),
137 idleTimeToLive: options.idleTimeToLive ?? Duration.infinity
138 })
139
140 yield* Effect.addFinalizer(() => {
141 shuttingDown = true
142 return Effect.void
143 })
144
145 if (options.acquireEagerly) {
146 // Initialize the resource
147 yield* Effect.scoped(RcRef.get(ref))
148 }
149
150 return identity<EntityResource<A, E>>({
151 [TypeId]: TypeId,
152 get: RcRef.get(ref),
153 close: RcRef.invalidate(ref)
154 })
155})
156
157/**
158 * Creates an `EntityResource` backed by a Kubernetes Pod.
159 *
160 * **Details**
161 *
162 * The pod is created and waited on through `K8sHttpClient`, and is kept alive
163 * until the resource is closed or its idle time to live expires.
164 *
165 * @stability unstable
166 * @category constructors
167 * @since 4.0.0
168 */
169export const makeK8sPod: (
170 spec: K8sPod,
171 options?: {
172 readonly idleTimeToLive?: Duration.Input | undefined
173 } | undefined
174) => Effect.Effect<
175 EntityResource<K8sHttpClient.PodStatus>,
176 never,
177 Scope.Scope | Sharding | Entity.CurrentAddress | K8sHttpClient.K8sHttpClient
178> = Effect.fnUntraced(function*(spec: K8sPod, options?: {
179 readonly idleTimeToLive?: Duration.Input | undefined
180}) {
181 const createPod = yield* K8sHttpClient.makeCreatePod
182 return yield* make({
183 ...options,
184 acquire: Effect.gen(function*() {
185 const scope = yield* CloseScope
186 return yield* createPod(spec).pipe(
187 Scope.provide(scope)
188 )
189 })
190 })
191})