authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 09:34:52-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 16:34:52+00:00
log9132c37c86cf5c382391dc65f4703ad407e8717e
treeb1a132236a4b7ed07dc8e3659b1419a1dec532bf
parentb9c7ea0581e171d2d4aad716f8391c670d5e3276
signature Signed by PGP key B5690EEEBB952194

[codex] migrate Effect.fn in apps/server/src/persistence/NodeSqliteClient.ts (#1633)

Co-authored-by: codex <codex@users.noreply.github.com>

1 files changed, 124 insertions(+), 131 deletions(-)

apps/server/src/persistence/NodeSqliteClient.ts+124-131
......@@ -72,159 +72,152 @@ const checkNodeSqliteCompat = () => {
7272 return Effect.void;
7373};
7474
75const makeWithDatabase = (
75const makeWithDatabase = Effect.fn("makeWithDatabase")(function* (
7676 options: SqliteClientConfig,
7777 openDatabase: () => DatabaseSync,
78): Effect.Effect<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> =>
79 Effect.gen(function* () {
80 yield* checkNodeSqliteCompat();
78): Effect.fn.Return<Client.SqlClient, never, Scope.Scope | Reactivity.Reactivity> {
79 yield* checkNodeSqliteCompat();
8180
82 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames);
83 const transformRows = options.transformResultNames
84 ? Statement.defaultTransforms(options.transformResultNames).array
85 : undefined;
81 const compiler = Statement.makeCompilerSqlite(options.transformQueryNames);
82 const transformRows = options.transformResultNames
83 ? Statement.defaultTransforms(options.transformResultNames).array
84 : undefined;
8685
87 const makeConnection = Effect.gen(function* () {
88 const scope = yield* Effect.scope;
89 const db = openDatabase();
90 yield* Scope.addFinalizer(
91 scope,
92 Effect.sync(() => db.close()),
93 );
86 const makeConnection = Effect.gen(function* () {
87 const scope = yield* Effect.scope;
88 const db = openDatabase();
89 yield* Scope.addFinalizer(
90 scope,
91 Effect.sync(() => db.close()),
92 );
9493
95 const statementReaderCache = new WeakMap<StatementSync, boolean>();
96 const hasRows = (statement: StatementSync): boolean => {
97 const cached = statementReaderCache.get(statement);
98 if (cached !== undefined) {
99 return cached;
94 const statementReaderCache = new WeakMap<StatementSync, boolean>();
95 const hasRows = (statement: StatementSync): boolean => {
96 const cached = statementReaderCache.get(statement);
97 if (cached !== undefined) {
98 return cached;
99 }
100 const value = statement.columns().length > 0;
101 statementReaderCache.set(statement, value);
102 return value;
103 };
104
105 const prepareCache = yield* Cache.make({
106 capacity: options.prepareCacheSize ?? 200,
107 timeToLive: options.prepareCacheTTL ?? Duration.minutes(10),
108 lookup: (sql: string) =>
109 Effect.try({
110 try: () => db.prepare(sql),
111 catch: (cause) =>
112 new SqlError({
113 reason: classifySqliteError(cause, {
114 message: "Failed to prepare statement",
115 operation: "prepare",
116 }),
117 }),
118 }),
119 });
120
121 const runStatement = (statement: StatementSync, params: ReadonlyArray<unknown>, raw: boolean) =>
122 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
123 statement.setReadBigInts(Boolean(ServiceMap.get(fiber.services, Client.SafeIntegers)));
124 try {
125 if (hasRows(statement)) {
126 return Effect.succeed(statement.all(...(params as any)));
127 }
128 const result = statement.run(...(params as any));
129 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);
130 } catch (cause) {
131 return Effect.fail(
132 new SqlError({
133 reason: classifySqliteError(cause, {
134 message: "Failed to execute statement",
135 operation: "execute",
136 }),
137 }),
138 );
100139 }
101 const value = statement.columns().length > 0;
102 statementReaderCache.set(statement, value);
103 return value;
104 };
140 });
141
142 const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) =>
143 Effect.flatMap(Cache.get(prepareCache, sql), (s) => runStatement(s, params, raw));
105144
106 const prepareCache = yield* Cache.make({
107 capacity: options.prepareCacheSize ?? 200,
108 timeToLive: options.prepareCacheTTL ?? Duration.minutes(10),
109 lookup: (sql: string) =>
145 const runValues = (sql: string, params: ReadonlyArray<unknown>) =>
146 Effect.acquireUseRelease(
147 Cache.get(prepareCache, sql),
148 (statement) =>
110149 Effect.try({
111 try: () => db.prepare(sql),
150 try: () => {
151 if (hasRows(statement)) {
152 statement.setReturnArrays(true);
153 // Safe to cast to array after we've setReturnArrays(true)
154 return statement.all(...(params as any)) as unknown as ReadonlyArray<
155 ReadonlyArray<unknown>
156 >;
157 }
158 statement.run(...(params as any));
159 return [];
160 },
112161 catch: (cause) =>
113162 new SqlError({
114163 reason: classifySqliteError(cause, {
115 message: "Failed to prepare statement",
116 operation: "prepare",
164 message: "Failed to execute statement",
165 operation: "execute",
117166 }),
118167 }),
119168 }),
120 });
121
122 const runStatement = (
123 statement: StatementSync,
124 params: ReadonlyArray<unknown>,
125 raw: boolean,
126 ) =>
127 Effect.withFiber<ReadonlyArray<any>, SqlError>((fiber) => {
128 statement.setReadBigInts(Boolean(ServiceMap.get(fiber.services, Client.SafeIntegers)));
129 try {
169 (statement) =>
170 Effect.sync(() => {
130171 if (hasRows(statement)) {
131 return Effect.succeed(statement.all(...(params as any)));
172 statement.setReturnArrays(false);
132173 }
133 const result = statement.run(...(params as any));
134 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);
135 } catch (cause) {
136 return Effect.fail(
137 new SqlError({
138 reason: classifySqliteError(cause, {
139 message: "Failed to execute statement",
140 operation: "execute",
141 }),
142 }),
143 );
144 }
145 });
146
147 const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) =>
148 Effect.flatMap(Cache.get(prepareCache, sql), (s) => runStatement(s, params, raw));
149
150 const runValues = (sql: string, params: ReadonlyArray<unknown>) =>
151 Effect.acquireUseRelease(
152 Cache.get(prepareCache, sql),
153 (statement) =>
154 Effect.try({
155 try: () => {
156 if (hasRows(statement)) {
157 statement.setReturnArrays(true);
158 // Safe to cast to array after we've setReturnArrays(true)
159 return statement.all(...(params as any)) as unknown as ReadonlyArray<
160 ReadonlyArray<unknown>
161 >;
162 }
163 statement.run(...(params as any));
164 return [];
165 },
166 catch: (cause) =>
167 new SqlError({
168 reason: classifySqliteError(cause, {
169 message: "Failed to execute statement",
170 operation: "execute",
171 }),
172 }),
173 }),
174 (statement) =>
175 Effect.sync(() => {
176 if (hasRows(statement)) {
177 statement.setReturnArrays(false);
178 }
179 }),
180 );
174 }),
175 );
181176
182 return identity<Connection>({
183 execute(sql, params, rowTransform) {
184 return rowTransform ? Effect.map(run(sql, params), rowTransform) : run(sql, params);
185 },
186 executeRaw(sql, params) {
187 return run(sql, params, true);
188 },
189 executeValues(sql, params) {
190 return runValues(sql, params);
191 },
192 executeUnprepared(sql, params, rowTransform) {
193 const effect = runStatement(db.prepare(sql), params ?? [], false);
194 return rowTransform ? Effect.map(effect, rowTransform) : effect;
195 },
196 executeStream(_sql, _params) {
197 return Stream.die("executeStream not implemented");
198 },
199 });
177 return identity<Connection>({
178 execute(sql, params, rowTransform) {
179 return rowTransform ? Effect.map(run(sql, params), rowTransform) : run(sql, params);
180 },
181 executeRaw(sql, params) {
182 return run(sql, params, true);
183 },
184 executeValues(sql, params) {
185 return runValues(sql, params);
186 },
187 executeUnprepared(sql, params, rowTransform) {
188 const effect = runStatement(db.prepare(sql), params ?? [], false);
189 return rowTransform ? Effect.map(effect, rowTransform) : effect;
190 },
191 executeStream(_sql, _params) {
192 return Stream.die("executeStream not implemented");
193 },
200194 });
195 });
201196
202 const semaphore = yield* Semaphore.make(1);
203 const connection = yield* makeConnection;
197 const semaphore = yield* Semaphore.make(1);
198 const connection = yield* makeConnection;
204199
205 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection));
206 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
207 const fiber = Fiber.getCurrent()!;
208 const scope = ServiceMap.getUnsafe(fiber.services, Scope.Scope);
209 return Effect.as(
210 Effect.tap(restore(semaphore.take(1)), () =>
211 Scope.addFinalizer(scope, semaphore.release(1)),
212 ),
213 connection,
214 );
215 });
200 const acquirer = semaphore.withPermits(1)(Effect.succeed(connection));
201 const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
202 const fiber = Fiber.getCurrent()!;
203 const scope = ServiceMap.getUnsafe(fiber.services, Scope.Scope);
204 return Effect.as(
205 Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
206 connection,
207 );
208 });
216209
217 return yield* Client.make({
218 acquirer,
219 compiler,
220 transactionAcquirer,
221 spanAttributes: [
222 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
223 [ATTR_DB_SYSTEM_NAME, "sqlite"],
224 ],
225 transformRows,
226 });
210 return yield* Client.make({
211 acquirer,
212 compiler,
213 transactionAcquirer,
214 spanAttributes: [
215 ...(options.spanAttributes ? Object.entries(options.spanAttributes) : []),
216 [ATTR_DB_SYSTEM_NAME, "sqlite"],
217 ],
218 transformRows,
227219 });
220});
228221
229222const make = (
230223 options: SqliteClientConfig,