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