1/**
2 * OrchestrationCommandReceiptRepository - Repository interface for command receipts.
3 *
4 * Owns persistence operations for deduplication and status tracking of
5 * orchestration command handling.
6 *
7 * @module OrchestrationCommandReceiptRepository
8 */
9import { CommandId, IsoDateTime, NonNegativeInt, ProjectId, ThreadId } from "@t3tools/contracts";
10import * as SqlClient from "effect/sql/SqlClient";
11import * as SqlSchema from "effect/sql/SqlSchema";
12import * as Context from "effect/Context";
13import * as Effect from "effect/Effect";
14import * as Layer from "effect/Layer";
15import * as Option from "effect/Option";
16import * as Schema from "effect/Schema";
17
18import {
19 toPersistenceSqlError,
20 type OrchestrationCommandReceiptRepositoryError,
21} from "./Errors.ts";
22
23export const OrchestrationCommandReceipt = Schema.Struct({
24 commandId: CommandId,
25 aggregateKind: Schema.Literals(["project", "thread"]),
26 aggregateId: Schema.Union([ProjectId, ThreadId]),
27 commandType: Schema.String,
28 acceptedAt: IsoDateTime,
29 resultSequence: NonNegativeInt,
30 status: Schema.Literals(["accepted", "rejected"]),
31 error: Schema.NullOr(Schema.String),
32});
33export type OrchestrationCommandReceipt = typeof OrchestrationCommandReceipt.Type;
34
35export const GetByCommandIdInput = Schema.Struct({
36 commandId: CommandId,
37});
38export type GetByCommandIdInput = typeof GetByCommandIdInput.Type;
39
40/**
41 * OrchestrationCommandReceiptRepository - Service tag for command receipt persistence.
42 */
43export class OrchestrationCommandReceiptRepository extends Context.Service<
44 OrchestrationCommandReceiptRepository,
45 {
46 readonly insertIfAbsent: (
47 receipt: OrchestrationCommandReceipt,
48 ) => Effect.Effect<boolean, OrchestrationCommandReceiptRepositoryError>;
49
50 /**
51 * Insert or replace a command receipt row.
52 *
53 * Upserts by `commandId` for idempotent command-result tracking.
54 */
55 readonly upsert: (
56 receipt: OrchestrationCommandReceipt,
57 ) => Effect.Effect<void, OrchestrationCommandReceiptRepositoryError>;
58
59 /**
60 * Read a command receipt by command id.
61 */
62 readonly getByCommandId: (
63 input: GetByCommandIdInput,
64 ) => Effect.Effect<
65 Option.Option<OrchestrationCommandReceipt>,
66 OrchestrationCommandReceiptRepositoryError
67 >;
68 }
69>()("t3/persistence/OrchestrationCommandReceipts/OrchestrationCommandReceiptRepository") {}
70
71const makeOrchestrationCommandReceiptRepository = Effect.gen(function* () {
72 const sql = yield* SqlClient.SqlClient;
73
74 const upsertReceiptRow = SqlSchema.void({
75 Request: OrchestrationCommandReceipt,
76 execute: (receipt) =>
77 sql`
78 INSERT INTO orchestration_command_receipts (
79 command_id,
80 aggregate_kind,
81 aggregate_id,
82 command_type,
83 accepted_at,
84 result_sequence,
85 status,
86 error
87 )
88 VALUES (
89 ${receipt.commandId},
90 ${receipt.aggregateKind},
91 ${receipt.aggregateId},
92 ${receipt.commandType},
93 ${receipt.acceptedAt},
94 ${receipt.resultSequence},
95 ${receipt.status},
96 ${receipt.error}
97 )
98 ON CONFLICT (command_id)
99 DO UPDATE SET
100 aggregate_kind = excluded.aggregate_kind,
101 aggregate_id = excluded.aggregate_id,
102 command_type = excluded.command_type,
103 accepted_at = excluded.accepted_at,
104 result_sequence = excluded.result_sequence,
105 status = excluded.status,
106 error = excluded.error
107 `,
108 });
109
110 const findReceiptByCommandId = SqlSchema.findOneOption({
111 Request: GetByCommandIdInput,
112 Result: OrchestrationCommandReceipt,
113 execute: ({ commandId }) =>
114 sql`
115 SELECT
116 command_id AS "commandId",
117 aggregate_kind AS "aggregateKind",
118 aggregate_id AS "aggregateId",
119 command_type AS "commandType",
120 accepted_at AS "acceptedAt",
121 result_sequence AS "resultSequence",
122 status,
123 error
124 FROM orchestration_command_receipts
125 WHERE command_id = ${commandId}
126 `,
127 });
128
129 const upsert: OrchestrationCommandReceiptRepository["Service"]["upsert"] = (receipt) =>
130 upsertReceiptRow(receipt).pipe(
131 Effect.mapError(toPersistenceSqlError("OrchestrationCommandReceiptRepository.upsert:query")),
132 );
133
134 const insertIfAbsent: OrchestrationCommandReceiptRepository["Service"]["insertIfAbsent"] = (
135 receipt,
136 ) =>
137 sql<{ readonly command_id: string }>`
138 INSERT INTO orchestration_command_receipts (
139 command_id,
140 aggregate_kind,
141 aggregate_id,
142 command_type,
143 accepted_at,
144 result_sequence,
145 status,
146 error
147 )
148 VALUES (
149 ${receipt.commandId},
150 ${receipt.aggregateKind},
151 ${receipt.aggregateId},
152 ${receipt.commandType},
153 ${receipt.acceptedAt},
154 ${receipt.resultSequence},
155 ${receipt.status},
156 ${receipt.error}
157 )
158 ON CONFLICT(command_id) DO NOTHING
159 RETURNING command_id
160 `.pipe(
161 Effect.map((rows) => rows.length === 1),
162 Effect.mapError(
163 toPersistenceSqlError("OrchestrationCommandReceiptRepository.insertIfAbsent:query"),
164 ),
165 );
166
167 const getByCommandId: OrchestrationCommandReceiptRepository["Service"]["getByCommandId"] = (
168 input,
169 ) =>
170 findReceiptByCommandId(input).pipe(
171 Effect.mapError(
172 toPersistenceSqlError("OrchestrationCommandReceiptRepository.getByCommandId:query"),
173 ),
174 );
175
176 return {
177 insertIfAbsent,
178 upsert,
179 getByCommandId,
180 } satisfies OrchestrationCommandReceiptRepository["Service"];
181});
182
183export const layer = Layer.effect(
184 OrchestrationCommandReceiptRepository,
185 makeOrchestrationCommandReceiptRepository,
186);