1import { CommandId, NonNegativeInt, ProjectId, ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as DateTime from "effect/DateTime";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
7import * as Schema from "effect/Schema";
8import * as OrchestrationCommandReceipts from "../persistence/OrchestrationCommandReceipts.ts";
9
10export class CommandReceiptStoreWriteError extends Schema.TaggedError<CommandReceiptStoreWriteError>()(
11 "CommandReceiptStoreWriteError",
12 {
13 commandId: CommandId,
14 cause: Schema.optional(Schema.Defect()),
15 },
16) {
17 override get message(): string {
18 return `Failed to write orchestration V2 command receipt ${this.commandId}.`;
19 }
20}
21
22export class CommandReceiptStoreReadError extends Schema.TaggedError<CommandReceiptStoreReadError>()(
23 "CommandReceiptStoreReadError",
24 {
25 commandId: CommandId,
26 cause: Schema.optional(Schema.Defect()),
27 },
28) {
29 override get message(): string {
30 return `Failed to read orchestration V2 command receipt ${this.commandId}.`;
31 }
32}
33
34export const CommandReceiptStoreV2Error = Schema.Union([
35 CommandReceiptStoreWriteError,
36 CommandReceiptStoreReadError,
37]);
38export type CommandReceiptStoreV2Error = typeof CommandReceiptStoreV2Error.Type;
39
40export const CommandReceiptV2Status = Schema.Literals(["accepted", "rejected"]);
41export type CommandReceiptV2Status = typeof CommandReceiptV2Status.Type;
42
43export const CommandReceiptV2 = Schema.Struct({
44 commandId: CommandId,
45 threadId: ThreadId,
46 commandType: Schema.String,
47 acceptedAt: Schema.DateTimeUtc,
48 resultSequence: NonNegativeInt,
49 status: CommandReceiptV2Status,
50 error: Schema.NullOr(Schema.String),
51});
52export type CommandReceiptV2 = typeof CommandReceiptV2.Type;
53
54export const ProjectCommandReceiptV2 = Schema.Struct({
55 commandId: CommandId,
56 projectId: ProjectId,
57 commandType: Schema.String,
58 acceptedAt: Schema.DateTimeUtc,
59 resultSequence: NonNegativeInt,
60 status: CommandReceiptV2Status,
61 error: Schema.NullOr(Schema.String),
62});
63export type ProjectCommandReceiptV2 = typeof ProjectCommandReceiptV2.Type;
64
65type AnyCommandReceiptV2 = CommandReceiptV2 | ProjectCommandReceiptV2;
66
67export interface CommandReceiptStoreV2Shape {
68 readonly insertIfAbsent: (
69 receipt: AnyCommandReceiptV2,
70 ) => Effect.Effect<boolean, CommandReceiptStoreV2Error>;
71 readonly upsert: (
72 receipt: AnyCommandReceiptV2,
73 ) => Effect.Effect<void, CommandReceiptStoreV2Error>;
74 /** None for project command ids. */
75 readonly getByCommandId: (
76 commandId: CommandId,
77 ) => Effect.Effect<Option.Option<CommandReceiptV2>, CommandReceiptStoreV2Error>;
78 /** None for thread command ids. */
79 readonly getProjectByCommandId: (
80 commandId: CommandId,
81 ) => Effect.Effect<Option.Option<ProjectCommandReceiptV2>, CommandReceiptStoreV2Error>;
82}
83
84export class CommandReceiptStoreV2 extends Context.Service<
85 CommandReceiptStoreV2,
86 CommandReceiptStoreV2Shape
87>()("t3/orchestration-v2/CommandReceiptStore/CommandReceiptStoreV2") {}
88
89const decodeReceipt = Schema.decodeUnknownEffect(
90 CommandReceiptV2.mapFields((fields) => ({
91 ...fields,
92 acceptedAt: Schema.DateTimeUtcFromString,
93 })),
94);
95const decodeProjectReceipt = Schema.decodeUnknownEffect(
96 ProjectCommandReceiptV2.mapFields((fields) => ({
97 ...fields,
98 acceptedAt: Schema.DateTimeUtcFromString,
99 })),
100);
101
102function fromApplicationReceipt(receipt: OrchestrationCommandReceipts.OrchestrationCommandReceipt) {
103 return decodeReceipt({
104 commandId: receipt.commandId,
105 threadId: receipt.aggregateId,
106 commandType: receipt.commandType,
107 acceptedAt: receipt.acceptedAt,
108 resultSequence: receipt.resultSequence,
109 status: receipt.status,
110 error: receipt.error,
111 });
112}
113
114function toApplicationReceipt(
115 receipt: AnyCommandReceiptV2,
116): OrchestrationCommandReceipts.OrchestrationCommandReceipt {
117 return {
118 commandId: receipt.commandId,
119 ...("projectId" in receipt
120 ? { aggregateKind: "project" as const, aggregateId: receipt.projectId }
121 : { aggregateKind: "thread" as const, aggregateId: receipt.threadId }),
122 commandType: receipt.commandType,
123 acceptedAt: DateTime.formatIso(receipt.acceptedAt),
124 resultSequence: receipt.resultSequence,
125 status: receipt.status,
126 error: receipt.error,
127 };
128}
129
130const layerBase: Layer.Layer<
131 CommandReceiptStoreV2,
132 never,
133 OrchestrationCommandReceipts.OrchestrationCommandReceiptRepository
134> = Layer.effect(
135 CommandReceiptStoreV2,
136 Effect.gen(function* () {
137 const receipts = yield* OrchestrationCommandReceipts.OrchestrationCommandReceiptRepository;
138
139 return CommandReceiptStoreV2.of({
140 insertIfAbsent: (receipt) =>
141 receipts.insertIfAbsent(toApplicationReceipt(receipt)).pipe(
142 Effect.mapError(
143 (cause) =>
144 new CommandReceiptStoreWriteError({
145 commandId: receipt.commandId,
146 cause,
147 }),
148 ),
149 ),
150 upsert: (receipt) =>
151 receipts.upsert(toApplicationReceipt(receipt)).pipe(
152 Effect.mapError(
153 (cause) =>
154 new CommandReceiptStoreWriteError({
155 commandId: receipt.commandId,
156 cause,
157 }),
158 ),
159 ),
160 getByCommandId: (commandId) =>
161 receipts.getByCommandId({ commandId }).pipe(
162 Effect.flatMap(
163 Option.match({
164 onNone: () => Effect.succeedNone,
165 onSome: (receipt) =>
166 receipt.aggregateKind !== "thread"
167 ? Effect.succeedNone
168 : fromApplicationReceipt(receipt).pipe(Effect.asSome),
169 }),
170 ),
171 Effect.mapError(
172 (cause) =>
173 new CommandReceiptStoreReadError({
174 commandId,
175 cause,
176 }),
177 ),
178 ),
179 getProjectByCommandId: (commandId) =>
180 receipts.getByCommandId({ commandId }).pipe(
181 Effect.flatMap((receipt) =>
182 Option.isNone(receipt) || receipt.value.aggregateKind !== "project"
183 ? Effect.succeedNone
184 : decodeProjectReceipt({
185 commandId: receipt.value.commandId,
186 projectId: receipt.value.aggregateId,
187 commandType: receipt.value.commandType,
188 acceptedAt: receipt.value.acceptedAt,
189 resultSequence: receipt.value.resultSequence,
190 status: receipt.value.status,
191 error: receipt.value.error,
192 }).pipe(Effect.asSome),
193 ),
194 Effect.mapError((cause) => new CommandReceiptStoreReadError({ commandId, cause })),
195 ),
196 } satisfies CommandReceiptStoreV2Shape);
197 }),
198);
199
200export const layer = layerBase.pipe(Layer.provide(OrchestrationCommandReceipts.layer));
201
202export const layerFromApplicationReceipts = layerBase;