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