1import { Clock } from "../../Clock.ts"
2import * as Context from "../../Context.ts"
3import * as Effect from "../../Effect.ts"
4import * as Latch from "../../Latch.ts"
5import * as Layer from "../../Layer.ts"
6import type { EntityAddress } from "../EntityAddress.ts"
7import type { EntityId } from "../EntityId.ts"
8import type { EntityState } from "./entityManager.ts"
9import type { ResourceMap } from "./resourceMap.ts"
10
11/**
12 * @internal
13 */
14export class EntityReaper extends Context.Service<EntityReaper>()("effect/cluster/EntityReaper", {
15 make: Effect.gen(function*() {
16 let currentResolution = 30_000
17 const registered: Array<{
18 readonly maxIdleTime: number
19 readonly servers: Map<EntityId, EntityState>
20 readonly entities: ResourceMap<EntityAddress, EntityState, any>
21 }> = []
22 const latch = yield* Latch.make()
23
24 const register = (options: {
25 readonly maxIdleTime: number
26 readonly servers: Map<EntityId, EntityState>
27 readonly entities: ResourceMap<EntityAddress, EntityState, any>
28 }) =>
29 Effect.suspend(() => {
30 currentResolution = Math.max(Math.min(currentResolution, options.maxIdleTime), 5000)
31 registered.push(options)
32 return latch.open
33 })
34
35 const clock = yield* Clock
36 yield* Effect.gen(function*() {
37 while (true) {
38 yield* Effect.sleep(currentResolution)
39 const now = clock.currentTimeMillisUnsafe()
40 for (const { entities, maxIdleTime, servers } of registered) {
41 for (const state of servers.values()) {
42 const duration = now - state.lastActiveCheck
43 if (state.keepAliveEnabled || state.activeRequests.size > 0 || duration < maxIdleTime) {
44 continue
45 }
46 yield* Effect.forkChild(entities.removeIgnore(state.address))
47 }
48 }
49 }
50 }).pipe(
51 latch.whenOpen,
52 Effect.forkScoped
53 )
54
55 return { register } as const
56 })
57}) {
58 static readonly layer: Layer.Layer<EntityReaper> = Layer.effect(this)(this.make)
59}