authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-01 09:14:55-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-01 16:14:55+00:00
logd15075b32d5a0670ae76e32d8e1c85f8a913dd7e
tree30a211fd57b5526c2069f9dfde94ece875447771
parent9f6686a13d308289c187c9da369f4874f0b5acc5
signature Signed by PGP key B5690EEEBB952194

[codex] migrate Effect.fn in apps/server/src/provider/Layers/EventNdjsonLogger.ts (#1636)

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

1 files changed, 152 insertions(+), 156 deletions(-)

apps/server/src/provider/Layers/EventNdjsonLogger.ts+152-156
......@@ -74,184 +74,180 @@ function resolveStreamLabel(stream: EventNdjsonStream): string {
7474 }
7575}
7676
77function toLogMessage(event: unknown): Effect.Effect<string | undefined> {
78 return Effect.gen(function* () {
79 const serialized = yield* Effect.sync(() => {
80 try {
81 return { ok: true as const, value: JSON.stringify(event) };
82 } catch (error) {
83 return { ok: false as const, error };
84 }
85 });
86
87 if (!serialized.ok) {
88 yield* logWarning("failed to serialize provider event log record", {
89 error: serialized.error,
90 });
91 return undefined;
77const toLogMessage = Effect.fn("toLogMessage")(function* (
78 event: unknown,
79): Effect.fn.Return<string | undefined> {
80 const serialized = yield* Effect.sync(() => {
81 try {
82 return { ok: true as const, value: JSON.stringify(event) };
83 } catch (error) {
84 return { ok: false as const, error };
9285 }
86 });
9387
94 if (typeof serialized.value !== "string") {
95 return undefined;
96 }
88 if (!serialized.ok) {
89 yield* logWarning("failed to serialize provider event log record", {
90 error: serialized.error,
91 });
92 return undefined;
93 }
9794
98 return serialized.value;
99 });
100}
95 if (typeof serialized.value !== "string") {
96 return undefined;
97 }
98
99 return serialized.value;
100});
101101
102function makeThreadWriter(input: {
102const makeThreadWriter = Effect.fn("makeThreadWriter")(function* (input: {
103103 readonly filePath: string;
104104 readonly maxBytes: number;
105105 readonly maxFiles: number;
106106 readonly batchWindowMs: number;
107107 readonly streamLabel: string;
108}): Effect.Effect<ThreadWriter | undefined> {
109 return Effect.gen(function* () {
110 const sinkResult = yield* Effect.sync(() => {
111 try {
112 return {
113 ok: true as const,
114 sink: new RotatingFileSink({
115 filePath: input.filePath,
116 maxBytes: input.maxBytes,
117 maxFiles: input.maxFiles,
118 throwOnError: true,
119 }),
120 };
121 } catch (error) {
122 return { ok: false as const, error };
123 }
124 });
125
126 if (!sinkResult.ok) {
127 yield* logWarning("failed to initialize provider thread log file", {
128 filePath: input.filePath,
129 error: sinkResult.error,
130 });
131 return undefined;
108}): Effect.fn.Return<ThreadWriter | undefined> {
109 const sinkResult = yield* Effect.sync(() => {
110 try {
111 return {
112 ok: true as const,
113 sink: new RotatingFileSink({
114 filePath: input.filePath,
115 maxBytes: input.maxBytes,
116 maxFiles: input.maxFiles,
117 throwOnError: true,
118 }),
119 };
120 } catch (error) {
121 return { ok: false as const, error };
132122 }
123 });
124
125 if (!sinkResult.ok) {
126 yield* logWarning("failed to initialize provider thread log file", {
127 filePath: input.filePath,
128 error: sinkResult.error,
129 });
130 return undefined;
131 }
133132
134 const sink = sinkResult.sink;
135 const scope = yield* Scope.make();
136 const lineLogger = makeLineLogger(input.streamLabel);
137 const batchedLogger = yield* Logger.batched(lineLogger, {
138 window: input.batchWindowMs,
139 flush: (messages) =>
140 Effect.gen(function* () {
141 const flushResult = yield* Effect.sync(() => {
142 try {
143 for (const message of messages) {
144 sink.write(message);
145 }
146 return { ok: true as const };
147 } catch (error) {
148 return { ok: false as const, error };
149 }
150 });
151
152 if (!flushResult.ok) {
153 yield* logWarning("provider event log batch flush failed", {
154 filePath: input.filePath,
155 error: flushResult.error,
156 });
133 const sink = sinkResult.sink;
134 const scope = yield* Scope.make();
135 const lineLogger = makeLineLogger(input.streamLabel);
136 const batchedLogger = yield* Logger.batched(lineLogger, {
137 window: input.batchWindowMs,
138 flush: Effect.fn("makeThreadWriter.flush")(function* (messages) {
139 const flushResult = yield* Effect.sync(() => {
140 try {
141 for (const message of messages) {
142 sink.write(message);
157143 }
158 }),
159 }).pipe(Effect.provideService(Scope.Scope, scope));
160
161 const loggerLayer = Logger.layer([batchedLogger], { mergeWithExisting: false });
162
163 return {
164 writeMessage(message: string) {
165 return Effect.log(message).pipe(Effect.provide(loggerLayer));
166 },
167 close() {
168 return Scope.close(scope, Exit.void);
169 },
170 } satisfies ThreadWriter;
171 });
172}
144 return { ok: true as const };
145 } catch (error) {
146 return { ok: false as const, error };
147 }
148 });
173149
174export function makeEventNdjsonLogger(
150 if (!flushResult.ok) {
151 yield* logWarning("provider event log batch flush failed", {
152 filePath: input.filePath,
153 error: flushResult.error,
154 });
155 }
156 }),
157 }).pipe(Effect.provideService(Scope.Scope, scope));
158
159 const loggerLayer = Logger.layer([batchedLogger], { mergeWithExisting: false });
160
161 return {
162 writeMessage(message: string) {
163 return Effect.log(message).pipe(Effect.provide(loggerLayer));
164 },
165 close() {
166 return Scope.close(scope, Exit.void);
167 },
168 } satisfies ThreadWriter;
169});
170
171export const makeEventNdjsonLogger = Effect.fn("makeEventNdjsonLogger")(function* (
175172 filePath: string,
176173 options: EventNdjsonLoggerOptions,
177): Effect.Effect<EventNdjsonLogger | undefined> {
178 return Effect.gen(function* () {
179 const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
180 const maxFiles = options.maxFiles ?? DEFAULT_MAX_FILES;
181 const batchWindowMs = options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS;
182 const streamLabel = resolveStreamLabel(options.stream);
183
184 const directoryReady = yield* Effect.sync(() => {
185 try {
186 fs.mkdirSync(path.dirname(filePath), { recursive: true });
187 return true;
188 } catch (error) {
189 return { ok: false as const, error };
190 }
174): Effect.fn.Return<EventNdjsonLogger | undefined> {
175 const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
176 const maxFiles = options.maxFiles ?? DEFAULT_MAX_FILES;
177 const batchWindowMs = options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS;
178 const streamLabel = resolveStreamLabel(options.stream);
179
180 const directoryReady = yield* Effect.sync(() => {
181 try {
182 fs.mkdirSync(path.dirname(filePath), { recursive: true });
183 return true;
184 } catch (error) {
185 return { ok: false as const, error };
186 }
187 });
188 if (directoryReady !== true) {
189 yield* logWarning("failed to create provider event log directory", {
190 filePath,
191 error: directoryReady.error,
191192 });
192 if (directoryReady !== true) {
193 yield* logWarning("failed to create provider event log directory", {
194 filePath,
195 error: directoryReady.error,
196 });
193 return undefined;
194 }
195
196 const threadWriters = new Map<string, ThreadWriter>();
197 const failedSegments = new Set<string>();
198
199 const resolveThreadWriter = Effect.fn("resolveThreadWriter")(function* (
200 threadSegment: string,
201 ): Effect.fn.Return<ThreadWriter | undefined> {
202 if (failedSegments.has(threadSegment)) {
197203 return undefined;
198204 }
205 const existing = threadWriters.get(threadSegment);
206 if (existing) {
207 return existing;
208 }
199209
200 const threadWriters = new Map<string, ThreadWriter>();
201 const failedSegments = new Set<string>();
202
203 const resolveThreadWriter = (threadSegment: string): Effect.Effect<ThreadWriter | undefined> =>
204 Effect.gen(function* () {
205 if (failedSegments.has(threadSegment)) {
206 return undefined;
207 }
208 const existing = threadWriters.get(threadSegment);
209 if (existing) {
210 return existing;
211 }
210 const writer = yield* makeThreadWriter({
211 filePath: path.join(path.dirname(filePath), `${threadSegment}.log`),
212 maxBytes,
213 maxFiles,
214 batchWindowMs,
215 streamLabel,
216 });
217 if (!writer) {
218 failedSegments.add(threadSegment);
219 return undefined;
220 }
212221
213 const writer = yield* makeThreadWriter({
214 filePath: path.join(path.dirname(filePath), `${threadSegment}.log`),
215 maxBytes,
216 maxFiles,
217 batchWindowMs,
218 streamLabel,
219 });
220 if (!writer) {
221 failedSegments.add(threadSegment);
222 return undefined;
223 }
222 threadWriters.set(threadSegment, writer);
223 return writer;
224 });
224225
225 threadWriters.set(threadSegment, writer);
226 return writer;
227 });
226 const write = Effect.fn("write")(function* (event: unknown, threadId: ThreadId | null) {
227 const threadSegment = resolveThreadSegment(threadId);
228 const message = yield* toLogMessage(event);
229 if (!message) {
230 return;
231 }
228232
229 return {
230 filePath,
231 write(event: unknown, threadId: ThreadId | null) {
232 return Effect.gen(function* () {
233 const threadSegment = resolveThreadSegment(threadId);
234 const message = yield* toLogMessage(event);
235 if (!message) {
236 return;
237 }
233 const writer = yield* resolveThreadWriter(threadSegment);
234 if (!writer) {
235 return;
236 }
238237
239 const writer = yield* resolveThreadWriter(threadSegment);
240 if (!writer) {
241 return;
242 }
238 yield* writer.writeMessage(message);
239 });
243240
244 yield* writer.writeMessage(message);
245 });
246 },
247 close() {
248 return Effect.gen(function* () {
249 for (const writer of threadWriters.values()) {
250 yield* writer.close();
251 }
252 threadWriters.clear();
253 });
254 },
255 } satisfies EventNdjsonLogger;
241 const close = Effect.fn("close")(function* () {
242 for (const writer of threadWriters.values()) {
243 yield* writer.close();
244 }
245 threadWriters.clear();
256246 });
257}
247
248 return {
249 filePath,
250 write,
251 close,
252 } satisfies EventNdjsonLogger;
253});