1import { type Cause, Context, Effect, Latch, Layer, MutableRef, Option, Queue, Schedule, Schema, Stream } from "effect"
2import type { Envelope } from "effect/cluster"
3import { ClusterSchema, Entity } from "effect/cluster"
4import { MemoryTransaction } from "effect/cluster/MessageStorage"
5import type { RpcGroup } from "effect/rpc"
6import { Rpc, RpcSchema } from "effect/rpc"
7
8export class User extends Schema.Class<User>("User")({
9 id: Schema.Number,
10 name: Schema.String
11}) {}
12
13export class BoomError extends Schema.TaggedError<BoomError>()("BoomError", {
14 cause: Schema.Unknown
15}) {}
16
17export class StreamWithKey extends Rpc.make("StreamWithKey", {
18 success: RpcSchema.Stream(Schema.Number, Schema.Never),
19 payload: { key: Schema.String },
20 primaryKey: ({ key }) => key
21}) {}
22
23export const TestEntity = Entity.make("TestEntity", [
24 Rpc.make("GetUser", {
25 success: User,
26 payload: { id: Schema.Number }
27 }),
28 Rpc.make("GetUserVolatile", {
29 success: User,
30 payload: { id: Schema.Number }
31 }).annotate(ClusterSchema.Persisted, false),
32 Rpc.make("Never"),
33 Rpc.make("Fail", { error: BoomError }),
34 Rpc.make("NeverFork"),
35 Rpc.make("NeverVolatile").annotate(ClusterSchema.Persisted, false),
36 Rpc.make("RequestWithKey", {
37 payload: { key: Schema.String },
38 primaryKey: ({ key }) => key
39 }),
40 StreamWithKey,
41 Rpc.make("GetAllUsers", {
42 success: User,
43 payload: { ids: Schema.Array(Schema.Number) },
44 stream: true
45 }),
46 Rpc.make("WithTransaction", {
47 success: Schema.Boolean,
48 payload: { id: Schema.Number }
49 }).annotate(ClusterSchema.Dynamic, Context.add(ClusterSchema.WithTransaction, true))
50]).annotateRpcs(ClusterSchema.Persisted, true)
51
52export class TestEntityState extends Context.Service<TestEntityState>()("TestEntityState", {
53 make: Effect.gen(function*() {
54 const messages = yield* Queue.make<void>()
55 const streamMessages = yield* Queue.make<void, Cause.Done>()
56 const envelopes = yield* Queue.make<
57 RpcGroup.Rpcs<typeof TestEntity.protocol> extends infer R ? R extends Rpc.Any ? Envelope.Request<R> : never
58 : never
59 >()
60 const interrupts = yield* Queue.make<
61 RpcGroup.Rpcs<typeof TestEntity.protocol> extends infer R ? R extends Rpc.Any ? Envelope.Request<R> : never
62 : never
63 >()
64 const defectTrigger = MutableRef.make(false)
65 const layerBuilds = MutableRef.make(0)
66 const buildLatch = Latch.makeUnsafe(true)
67
68 return {
69 messages,
70 streamMessages,
71 envelopes,
72 interrupts,
73 defectTrigger,
74 layerBuilds,
75 buildLatch
76 } as const
77 })
78}) {
79 static readonly layer = Layer.effect(this)(this.make)
80}
81
82export const TestEntityNoState = TestEntity.toLayer(
83 Effect.gen(function*() {
84 const state = yield* TestEntityState
85
86 MutableRef.update(state.layerBuilds, (count) => count + 1)
87 yield* state.buildLatch.await
88
89 const never = (envelope: any) =>
90 Effect.suspend(() => {
91 Queue.offerUnsafe(state.envelopes, envelope)
92 return Effect.never
93 }).pipe(Effect.onInterrupt(() => {
94 Queue.offerUnsafe(state.interrupts, envelope)
95 return Effect.void
96 }))
97 return TestEntity.of({
98 GetUser: (envelope) =>
99 Effect.sync(() => {
100 Queue.offerUnsafe(state.envelopes, envelope)
101 if (state.defectTrigger.current) {
102 MutableRef.set(state.defectTrigger, false)
103 throw new Error("User not found")
104 }
105 return new User({ id: envelope.payload.id, name: `User ${envelope.payload.id}` })
106 }),
107 GetUserVolatile: (envelope) =>
108 Effect.sync(() => {
109 Queue.offerUnsafe(state.envelopes, envelope)
110 return new User({ id: envelope.payload.id, name: `User ${envelope.payload.id}` })
111 }),
112 Never: never,
113 Fail: () => Effect.fail(new BoomError({ cause: new Error("boom") })),
114 NeverFork: (envelope) => Rpc.fork(never(envelope)),
115 NeverVolatile: never,
116 RequestWithKey: (envelope) => {
117 Queue.offerUnsafe(state.envelopes, envelope)
118 return Queue.take(state.messages)
119 },
120 StreamWithKey: (envelope) => {
121 let sequence = Option.match(envelope.lastSentChunkValue, {
122 onNone: () => 0,
123 onSome: (value) => value + 1
124 })
125 return Stream.fromQueue(state.streamMessages).pipe(
126 Stream.map(() => sequence++)
127 )
128 },
129 GetAllUsers: (envelope) => {
130 Queue.offerUnsafe(state.envelopes, envelope)
131 return Stream.fromIterable(envelope.payload.ids.map((id) => new User({ id, name: `User ${id}` }))).pipe(
132 Stream.rechunk(1)
133 )
134 },
135 WithTransaction: () => MemoryTransaction
136 })
137 }),
138 { defectRetryPolicy: Schedule.forever }
139)
140
141export const TestEntityLayer = TestEntityNoState.pipe(Layer.provideMerge(TestEntityState.layer))
142
143export const CallerId = Context.Service<never, string>(
144 "effect/test/cluster/CallerId"
145)
146
147export const ContextBleedEntity = Entity.make("ContextBleedEntity", [
148 Rpc.make("ReadCaller", { success: Schema.String }).annotate(ClusterSchema.Persisted, false),
149 Rpc.make("ReadCallerPersisted", { success: Schema.String }).annotate(ClusterSchema.Persisted, true)
150])
151
152const readCaller = () => Effect.map(Effect.serviceOption(CallerId), Option.getOrElse(() => "none"))
153
154export const ContextBleedLayer = ContextBleedEntity.toLayer({
155 ReadCaller: readCaller,
156 ReadCallerPersisted: readCaller
157})