1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodePath from "node:path";
4import * as Schema from "effect/Schema";
5import * as Effect from "effect/Effect";
6import * as FileSystem from "effect/FileSystem";
7import * as Path from "effect/Path";
8import * as Semaphore from "effect/Semaphore";
9
10export interface RotatingFileSinkOptions {
11 readonly filePath: string;
12 readonly maxBytes: number;
13 readonly maxFiles: number;
14 readonly throwOnError?: boolean;
15}
16
17export class RotatingFileSinkConfigurationError extends Schema.TaggedError<RotatingFileSinkConfigurationError>()(
18 "RotatingFileSinkConfigurationError",
19 {
20 option: Schema.Literals(["maxBytes", "maxFiles"]),
21 received: Schema.Number,
22 minimum: Schema.Number,
23 },
24) {
25 override get message(): string {
26 return `${this.option} must be >= ${this.minimum} (received ${this.received})`;
27 }
28}
29
30export class RotatingFileSinkError extends Schema.TaggedError<RotatingFileSinkError>()(
31 "RotatingFileSinkError",
32 {
33 operation: Schema.Literals(["initialize", "read", "write", "rotate", "prune"]),
34 filePath: Schema.String,
35 cause: Schema.Defect(),
36 },
37) {
38 override get message(): string {
39 return `Failed to ${this.operation} rotating log file ${this.filePath}`;
40 }
41}
42
43const isRotatingFileSinkError = Schema.is(RotatingFileSinkError);
44
45const isFileNotFoundError = (cause: unknown): cause is NodeJS.ErrnoException =>
46 cause instanceof Error && "code" in cause && cause.code === "ENOENT";
47
48export class RotatingFileSink {
49 private readonly filePath: string;
50 private readonly maxBytes: number;
51 private readonly maxFiles: number;
52 private readonly throwOnError: boolean;
53 private currentSize = 0;
54
55 constructor(options: RotatingFileSinkOptions) {
56 if (options.maxBytes < 1) {
57 throw new RotatingFileSinkConfigurationError({
58 option: "maxBytes",
59 received: options.maxBytes,
60 minimum: 1,
61 });
62 }
63 if (options.maxFiles < 1) {
64 throw new RotatingFileSinkConfigurationError({
65 option: "maxFiles",
66 received: options.maxFiles,
67 minimum: 1,
68 });
69 }
70
71 this.filePath = options.filePath;
72 this.maxBytes = options.maxBytes;
73 this.maxFiles = options.maxFiles;
74 this.throwOnError = options.throwOnError ?? false;
75
76 try {
77 NodeFS.mkdirSync(NodePath.dirname(this.filePath), { recursive: true });
78 } catch (cause) {
79 throw new RotatingFileSinkError({
80 operation: "initialize",
81 filePath: this.filePath,
82 cause,
83 });
84 }
85 this.pruneOverflowBackups();
86 this.currentSize = this.readCurrentSize();
87 }
88
89 write(chunk: string | Buffer): void {
90 const buffer = typeof chunk === "string" ? Buffer.from(chunk) : chunk;
91 if (buffer.length === 0) return;
92
93 try {
94 if (this.currentSize > 0 && this.currentSize + buffer.length > this.maxBytes) {
95 this.rotate();
96 }
97
98 NodeFS.appendFileSync(this.filePath, buffer);
99 this.currentSize += buffer.length;
100 } catch (cause) {
101 if (isRotatingFileSinkError(cause)) {
102 throw cause;
103 }
104 if (this.throwOnError) {
105 throw new RotatingFileSinkError({
106 operation: "write",
107 filePath: this.filePath,
108 cause,
109 });
110 }
111 this.currentSize = this.readCurrentSize();
112 }
113 }
114
115 private rotate(): void {
116 try {
117 const oldest = this.withSuffix(this.maxFiles);
118 if (NodeFS.existsSync(oldest)) {
119 NodeFS.rmSync(oldest, { force: true });
120 }
121
122 for (let index = this.maxFiles - 1; index >= 1; index -= 1) {
123 const source = this.withSuffix(index);
124 const target = this.withSuffix(index + 1);
125 if (NodeFS.existsSync(source)) {
126 NodeFS.renameSync(source, target);
127 }
128 }
129
130 if (NodeFS.existsSync(this.filePath)) {
131 NodeFS.renameSync(this.filePath, this.withSuffix(1));
132 }
133
134 this.currentSize = 0;
135 } catch (cause) {
136 if (this.throwOnError) {
137 throw new RotatingFileSinkError({
138 operation: "rotate",
139 filePath: this.filePath,
140 cause,
141 });
142 }
143 this.currentSize = this.readCurrentSize();
144 }
145 }
146
147 private pruneOverflowBackups(): void {
148 try {
149 const dir = NodePath.dirname(this.filePath);
150 const baseName = NodePath.basename(this.filePath);
151 for (const entry of NodeFS.readdirSync(dir)) {
152 if (!entry.startsWith(`${baseName}.`)) continue;
153 const suffix = Number(entry.slice(baseName.length + 1));
154 if (!Number.isInteger(suffix) || suffix <= this.maxFiles) continue;
155 NodeFS.rmSync(NodePath.join(dir, entry), { force: true });
156 }
157 } catch (cause) {
158 if (this.throwOnError) {
159 throw new RotatingFileSinkError({
160 operation: "prune",
161 filePath: this.filePath,
162 cause,
163 });
164 }
165 }
166 }
167
168 private readCurrentSize(): number {
169 try {
170 return NodeFS.statSync(this.filePath).size;
171 } catch (cause) {
172 if (isFileNotFoundError(cause)) {
173 return 0;
174 }
175 throw new RotatingFileSinkError({
176 operation: "read",
177 filePath: this.filePath,
178 cause,
179 });
180 }
181 }
182
183 private withSuffix(index: number): string {
184 return `${this.filePath}.${index}`;
185 }
186}
187
188/** Serialized asynchronous writes and rotation without blocking the event loop. */
189export const makeAsyncRotatingFileSink = Effect.fnUntraced(function* (
190 options: Omit<RotatingFileSinkOptions, "throwOnError">,
191) {
192 const fs = yield* FileSystem.FileSystem;
193 const path = yield* Path.Path;
194 if (options.maxBytes < 1)
195 return yield* new RotatingFileSinkConfigurationError({
196 option: "maxBytes",
197 received: options.maxBytes,
198 minimum: 1,
199 });
200 if (options.maxFiles < 1)
201 return yield* new RotatingFileSinkConfigurationError({
202 option: "maxFiles",
203 received: options.maxFiles,
204 minimum: 1,
205 });
206 const failure = (operation: RotatingFileSinkError["operation"]) => (cause: unknown) =>
207 new RotatingFileSinkError({ operation, filePath: options.filePath, cause });
208 const directory = path.dirname(options.filePath);
209 const basename = path.basename(options.filePath);
210 yield* fs
211 .makeDirectory(directory, { recursive: true })
212 .pipe(Effect.mapError(failure("initialize")));
213 yield* Effect.gen(function* () {
214 for (const entry of yield* fs.readDirectory(directory)) {
215 if (!entry.startsWith(`${basename}.`)) continue;
216 const suffix = Number(entry.slice(basename.length + 1));
217 if (!Number.isInteger(suffix) || suffix <= options.maxFiles) continue;
218 yield* fs.remove(path.join(directory, entry), { force: true });
219 }
220 }).pipe(Effect.mapError(failure("prune")));
221 const readSize = fs.stat(options.filePath).pipe(
222 Effect.map((stat) => Number(stat.size)),
223 Effect.catch((cause) =>
224 cause.reason._tag === "NotFound" ? Effect.succeed(0) : Effect.fail(failure("read")(cause)),
225 ),
226 );
227 let currentSize = yield* readSize;
228 const semaphore = yield* Semaphore.make(1);
229 const rotate = Effect.gen(function* () {
230 yield* fs.remove(`${options.filePath}.${options.maxFiles}`, { force: true });
231 for (let index = options.maxFiles - 1; index >= 1; index--) {
232 const source = `${options.filePath}.${index}`;
233 if (yield* fs.exists(source)) yield* fs.rename(source, `${options.filePath}.${index + 1}`);
234 }
235 if (yield* fs.exists(options.filePath))
236 yield* fs.rename(options.filePath, `${options.filePath}.1`);
237 currentSize = 0;
238 }).pipe(Effect.mapError(failure("rotate")));
239 const write = (chunk: string | Buffer) =>
240 semaphore
241 .withPermits(1)(
242 Effect.gen(function* () {
243 const buffer = typeof chunk === "string" ? Buffer.from(chunk) : chunk;
244 if (buffer.length === 0) return;
245 if (currentSize > 0 && currentSize + buffer.length > options.maxBytes) yield* rotate;
246 yield* fs
247 .writeFile(options.filePath, buffer, { flag: "a" })
248 .pipe(Effect.mapError(failure("write")));
249 currentSize += buffer.length;
250 }).pipe(
251 Effect.tapError(() =>
252 readSize.pipe(
253 Effect.tap((size) =>
254 Effect.sync(() => {
255 currentSize = size;
256 }),
257 ),
258 Effect.ignore,
259 ),
260 ),
261 ),
262 )
263 .pipe(Effect.uninterruptible);
264 return { write };
265});