mpackdb

26.9 KB

1. import { dirname, basename, extname, resolve } from 'path';
2. import { createReadStream, createWriteStream } from 'fs';
3. import { mkdir, open, stat, readFile, writeFile, rename, unlink } from 'fs/promises';
4. import { serialize, deserialize, uuid, PrimaryKeyType, IndexType } from './mpack.js';
5. import { Cursor } from './Cursor.js';
6. import { IndexManager } from './IndexManager.js';
8. export { PrimaryKeyType, IndexType };
10. /**
11. * MPackDB - A fast, local, append-only JSON database with MessagePack serialization
12. *
13. * Features:
14. * - Append-only writes for high performance
15. * - MessagePack binary serialization
16. * - Optional indexes (numeric and lexical)
17. * - File-based locking for concurrent access
18. * - Auto-compaction on startup
19. * - Auto-persistence of indexes
20. */
21. export class MPackDB {
22. _dbFile = null;
23. _primaryKeyType = null;
24. _primaryKey = null;
25. _classToUse = null;
27. _indexes = [];
28. _indexTypes = {};
29. _uniqueIndexes = new Set();
31. _initPromise = null;
32. _initState = 0;
33. _dataPath = null;
34. _dataStream = null;
35. _lockPath = null;
36. _lockDepth = 0;
37. _indexManager = null;
38. _meta = {
39. nextId: 0,
40. deleted: [],
41. };
42. _debug = false;
43. _indexPersistInterval = 60000; // 60 seconds default
44. _indexPersistThreshold = 1000; // 1000 entries default
45. _processExitHandler = null;
47. /**
48. * Create a new MPackDB instance
49. *
50. * @param {string} dbFile - Path to the database file (without extension)
51. * @param {Object} options - Configuration options
52. * @param {string} [options.primaryKey] - Primary key field name. Prefix with * for numeric (e.g., '*id'), @ for UUID (e.g., '@uuid'), or no prefix for string
53. * @param {PrimaryKeyType} [options.primaryKeyType] - Explicit primary key type (overrides prefix)
54. * @param {string[]} [options.indexes] - Array of field names to index. Use * prefix for numeric, @ for UUID
55. * @param {boolean} [options.debug=false] - Enable debug logging
56. * @param {number} [options.indexPersistInterval=60000] - Milliseconds between automatic index persistence (0 to disable)
57. * @param {number} [options.indexPersistThreshold=1000] - Number of changes before auto-persisting indexes
58. * @param {boolean} [options.compact=true] - Run compaction on init (set false for read-only / secondary instances)
59. *
60. * @example
61. * const db = new MPackDB('data/users', {
62. * primaryKey: '*id', // Numeric auto-increment
63. * indexes: ['email', '*age'], // Index email (lexical) and age (numeric)
64. * indexPersistThreshold: 100
65. * });
66. */
67. _compact = true;
69. constructor(dbFile, { primaryKey, primaryKeyType, indexes, debug, indexPersistInterval, indexPersistThreshold, compact } = {}) {
70. this._dbFile = dbFile || this._dbFile;
71. this._debug = debug || this._debug;
72. if (compact !== undefined) this._compact = compact;
73. if (indexPersistInterval !== undefined) this._indexPersistInterval = indexPersistInterval;
74. if (indexPersistThreshold !== undefined) this._indexPersistThreshold = indexPersistThreshold;
76. // Parse primary key with optional prefix
77. if (primaryKey) {
78. if (primaryKey.startsWith('*')) {
79. // *id = numeric primary key
80. this._primaryKey = primaryKey.slice(1);
81. this._primaryKeyType = PrimaryKeyType.NUMBER;
82. } else if (primaryKey.startsWith('@')) {
83. // @id = UUID primary key
84. this._primaryKey = primaryKey.slice(1);
85. this._primaryKeyType = PrimaryKeyType.UUID;
86. } else {
87. // id = string primary key (lexical)
88. this._primaryKey = primaryKey;
89. this._primaryKeyType = PrimaryKeyType.STRING;
90. }
91. // Allow explicit override
92. if (primaryKeyType !== undefined) {
93. this._primaryKeyType = primaryKeyType;
94. }
95. }
97. // Primary key is always unique
98. if (this._primaryKey) {
99. this._uniqueIndexes.add(this._primaryKey);
100. }
102. const allIndexes = [...new Set([
103. ...(this._primaryKey ? [this._primaryKey] : []),
104. ...(indexes || []),
105. ...(this._indexes || []),
106. ])];
107. this._indexes = [];
108. for (let index of allIndexes) {
109. let cleanIndex = index;
110. // !field = unique index (can combine with * and @: !*field, !@field)
111. let isUnique = false;
112. if (cleanIndex.startsWith('!')) {
113. isUnique = true;
114. cleanIndex = cleanIndex.slice(1);
115. }
116. if (cleanIndex.startsWith('*')) {
117. // *field = numeric index
118. cleanIndex = cleanIndex.slice(1);
119. this._indexTypes[cleanIndex] = IndexType.NUMERIC;
120. } else if (cleanIndex.startsWith('@')) {
121. // @field = UUID index (lexical)
122. cleanIndex = cleanIndex.slice(1);
123. this._indexTypes[cleanIndex] = IndexType.LEXICAL;
124. } else if (cleanIndex === this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER) {
125. // Primary key is numeric
126. this._indexTypes[cleanIndex] = IndexType.NUMERIC;
127. } else {
128. // Default to lexical
129. this._indexTypes[cleanIndex] = IndexType.LEXICAL;
130. }
131. if (isUnique) {
132. this._uniqueIndexes.add(cleanIndex);
133. }
134. this._indexes.push(cleanIndex);
135. }
136. }
138. /**
139. * Initialize the database (called automatically by other methods)
140. * Performs compaction and loads metadata
141. *
142. * @returns {Promise<MPackDB>} The database instance
143. */
144. async init() {
145. if (this._initState === 2) return this;
147. if (this._initState === 1) {
148. return this._initPromise;
149. }
151. this._initState = 1;
152. return this._initPromise = new Promise(async success => {
153. if (!this._dbFile) {
154. throw new Error('No database file specified.');
155. }
157. const dbDir = dirname(this._dbFile);
158. const baseName = basename(this._dbFile, extname(this._dbFile));
159. await mkdir(dbDir, { recursive: true });
161. this._dataPath = resolve(dbDir, `${baseName}.mpack`);
162. this._metaPath = resolve(dbDir, `${baseName}.meta.json`);
163. this._lockPath = resolve(dbDir, `${baseName}.lock`);
165. try {
166. this._meta = JSON.parse(await readFile(this._metaPath));
167. } catch (e) {
168. this._meta = {
169. nextId: 0,
170. deleted: [],
171. };
172. }
174. this._initState = 2; // compact causes init to run again thats why we set it to 2 here already
176. if (this._compact) await this.compact();
178. this._dataStream = createWriteStream(this._dataPath, { flags: 'a' }); // needs to be set after compact as it replaces the file with a tmp file
180. // Initialize IndexManager if indexes are specified
181. if (this._indexes.length > 0) {
182. const dbDir = dirname(this._dbFile);
183. const baseName = basename(this._dbFile, extname(this._dbFile));
184. this._indexManager = new IndexManager(dbDir, baseName, this._indexes, this._indexTypes, this._primaryKeyType);
185. await this._indexManager.init(this._dataPath);
186. // Start auto-persist with configured interval and threshold
187. this._indexManager.startAutoPersist(this._indexPersistInterval, this._indexPersistThreshold);
188. // Register signal handlers for Ctrl+C and kill signals
189. this._processExitHandler = async () => {
190. await this.close();
191. process.exit(0);
192. };
193. process.once('SIGINT', this._processExitHandler);
194. process.once('SIGTERM', this._processExitHandler);
195. }
197. this._initPromise = null;
198. success(this);
199. });
200. }
202. /**
203. * Insert a new record into the database
204. *
205. * @param {Object} record - The record to insert
206. * @param {Object} [options]
207. * @param {boolean} [options.skipPrimaryKey=false] - Skip auto-generating primary key
208. * @returns {Promise<Object>} The inserted record with primary key
209. *
210. * @example
211. * await db.insert({ name: 'Alice', age: 30 });
212. * // Returns: { id: 0, name: 'Alice', age: 30 }
213. */
214. async insert(record, { skipPrimaryKey = false } = {}) {
215. await this.init();
216. await this._acquireLock('insert', record);
217. try {
219. const recToInsert = this._classToUse
220. ? Object.assign(new this._classToUse(), record)
221. : { ...record };
223. if (this._primaryKey && !skipPrimaryKey && !recToInsert[this._primaryKey]) {
224. if (this._primaryKeyType === PrimaryKeyType.NUMBER) {
225. recToInsert[this._primaryKey] = this._meta.nextId++;
226. } else if (this._primaryKeyType === PrimaryKeyType.UUID) {
227. recToInsert[this._primaryKey] = uuid();
228. }
229. }
231. // Check unique index constraints (skip auto-generated primary keys — guaranteed unique)
232. if (this._indexManager && this._uniqueIndexes.size > 0) {
233. const autoGenPK = this._primaryKey && !skipPrimaryKey && !record[this._primaryKey]
234. && (this._primaryKeyType === PrimaryKeyType.NUMBER || this._primaryKeyType === PrimaryKeyType.UUID);
235. const deletedSet = new Set(this._meta.deleted);
236. for (const field of this._uniqueIndexes) {
237. if (autoGenPK && field === this._primaryKey) continue;
238. const value = recToInsert[field];
239. if (value === undefined) continue;
240. const existing = await this._indexManager.get(field, value);
241. const live = existing.filter(e => !deletedSet.has(e.loc[0]));
242. if (live.length > 0) {
243. const err = new Error(`Duplicate key: ${field}=${value}`);
244. err.code = 'DUPLICATE_KEY';
245. err.field = field;
246. err.value = value;
247. throw err;
248. }
249. }
250. }
252. const packedBuffer = serialize(recToInsert);
253. const fileStat = await stat(this._dataPath).catch(() => ({ size: 0 }));
254. const offset = fileStat.size;
256. await new Promise(resolve => this._dataStream.write(packedBuffer, resolve));
258. const loc = [offset, packedBuffer.length];
259. // indexes
260. if (this._indexManager) {
261. this._indexManager.insert(recToInsert, loc);
262. }
264. // Persist meta if we incremented nextId
265. if (this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER && !skipPrimaryKey && !record[this._primaryKey]) {
266. await this.persistMeta();
267. }
269. return this._primaryKey ? recToInsert[this._primaryKey] : recToInsert;
270. } finally {
271. await this._releaseLock();
272. }
273. }
275. /**
276. * Update records matching a query.
277. * The callback receives the old record and must return the new record.
278. *
279. * @param {string|number|Function} mixed - Primary key value or query function
280. * @param {Function} callback - Receives old record, must return new record
281. * @param {Object} [options]
282. * @param {boolean} [options.upsert=false] - Insert if no records match
283. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads
284. * @returns {Promise<Object[]>} Array of updated records
285. *
286. * @example
287. * await db.update(0, record => {
288. * record.age = 32;
289. * delete record.address;
290. * return record;
291. * });
292. */
293. async update(mixed, callback, { upsert = false, index } = {}) {
294. if (typeof callback !== 'function') {
295. throw new Error('update requires a callback function as second argument');
296. }
297. let insertedRecords = await this.delete(mixed, {
298. index,
299. callback: async record => {
300. return this.insert(await callback(record), { skipPrimaryKey: true });
301. }
302. });
303. if (upsert && insertedRecords.length === 0) {
304. insertedRecords = await this.insert(await callback({}));
305. }
306. return insertedRecords;
307. }
309. /**
310. * Update records or insert if not found
311. *
312. * @param {string|number|Function} mixed - Primary key value or query function
313. * @param {Function} callback - Receives old record (or {} if inserting), must return new record
314. * @param {Object} [options]
315. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads
316. * @returns {Promise<Object[]>} Array of updated/inserted records
317. */
318. async upsert(mixed, callback, { index } = {}) {
319. return this.update(mixed, callback, { upsert: true, index });
320. }
322. /**
323. * Delete records matching a query
324. *
325. * @param {string|Function} mixed - Primary key value or query function
326. * @param {Object} [options]
327. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads
328. * @param {Function} [options.callback] - Internal callback per deleted record (used by update)
329. * @returns {Promise<Object[]>} Array of deleted records
330. *
331. * @example
332. * // Delete by primary key
333. * await db.delete(0);
334. *
335. * // Delete with query function
336. * await db.delete(r => r.age < 18);
337. *
338. * // Delete with index hint
339. * await db.delete(r => r.status === 'inactive', { index: { field: 'status', value: 'inactive' } });
340. */
341. async delete(mixed, { index, callback = async record => record } = {}) {
342. await this.init();
343. await this._acquireLock('delete', mixed);
344. try {
345. const promises = [];
346. for await (const [record, offset] of this.find(mixed, { mode: 'mixed', index })) {
347. this._meta.deleted.push(offset);
348. if (this._indexManager) {
349. await this._indexManager.remove(record, this._primaryKey);
350. }
351. promises.push(callback(record));
352. }
353. await this.persistMeta();
354. return Promise.all(promises);
355. } finally {
356. await this._releaseLock();
357. }
358. }
360. /**
361. * Finds records in the database
362. * @param {undefined|string|number|function} [mixed] - Query: undefined/null for all, primary key value for PK lookup, function for filter
363. * @param {Object} [options] - Query options
364. * @param {Object|Array} [options.index] - Index hint(s) to narrow disk reads before filtering
365. * @returns {Cursor} A cursor for iterating over results
366. * @example
367. * // All records
368. * for await (const user of db.find()) { ... }
369. *
370. * // Primary key lookup
371. * const [user] = await db.find(68);
372. *
373. * // Filter with index range — only reads records in x 100+
374. * for await (const doc of db.find(r => r.x <= 200, { index: { field: 'x', from: 100 } })) { ... }
375. *
376. * // Index intersection — intersects offsets first, then streams matches
377. * for await (const doc of db.find(r => r.x <= 200, {
378. * index: [
379. * { field: 'x', from: 100 },
380. * { field: 'status', value: 'active' }
381. * ]
382. * })) { ... }
383. */
384. find(mixed, options = {}) {
385. if (typeof mixed === 'undefined' || mixed === null || typeof mixed === 'function') {
386. return new Cursor(this, mixed, options);
387. } else {
388. if (!this._primaryKey) {
389. throw new Error('No primary key specified.');
390. }
391. // Use indexed lookup when available
392. if (this._indexManager) {
393. return new Cursor(this, null, {
394. ...options,
395. _indexLookup: { field: this._primaryKey, value: mixed }
396. });
397. }
398. return new Cursor(this, record => {
399. return record[this._primaryKey] === mixed;
400. }, options);
401. }
402. }
404. /**
405. * Find records within a bounding box defined by 4 corners.
406. * Requires numeric indexes on x and y fields.
407. * Corners can be in any order — min/max are extracted automatically.
408. *
409. * @param {Array<{x: number, y: number}>} corners - 4 corner coordinates
410. * @param {function} [filter] - Optional additional filter function
411. * @returns {Cursor}
412. * @example
413. * const results = await db.boundingBox([
414. * { x: -5, y: 5 }, { x: 5, y: 5 },
415. * { x: 5, y: -5 }, { x: -5, y: -5 }
416. * ]);
417. */
418. boundingBox(corners, filter) {
419. const xs = corners.map(c => c.x);
420. const ys = corners.map(c => c.y);
421. const minX = Math.min(...xs), maxX = Math.max(...xs);
422. const minY = Math.min(...ys), maxY = Math.max(...ys);
424. return this.find(filter || null, {
425. index: [
426. { field: 'x', from: minX, to: maxX },
427. { field: 'y', from: minY, to: maxY }
428. ]
429. });
430. }
432. /**
433. * Execute a callback while holding the database lock.
434. * The lock is re-entrant: find/insert/delete/update called inside
435. * the callback reuse the same lock instead of deadlocking.
436. *
437. * Use this for compound operations that must be atomic, e.g.
438. * find-then-insert (login pattern).
439. *
440. * @param {Function} callback - Async function to execute under lock
441. * @returns {Promise<any>} The return value of the callback
442. *
443. * @example
444. * const user = await db.withLock(async () => {
445. * const [existing] = await db.find(u => u.email === email);
446. * if (existing) return existing;
447. * return db.insert({ email });
448. * });
449. */
450. async withLock(callback) {
451. await this.init();
452. await this._acquireLock('withLock');
453. try {
454. return await callback();
455. } finally {
456. await this._releaseLock();
457. }
458. }
460. /**
461. * Generator that yields records from the database file
462. * @param {function|null} [queryFn=null] - Optional filter function
463. * @param {Object} [options] - Generator options
464. * @param {string} [options.mode='record'] - Mode: 'record', 'raw', 'offset', or 'mixed'
465. * @yields {Object|Buffer|Array} Records, buffers, or [record, offset, size] tuples depending on mode
466. */
467. async *recordGenerator(queryFn = null, { mode = 'record', _indexLookup, index } = {}) {
468. await this.init();
469. await this.refresh();
471. // Indexed primary key lookup — O(log n) instead of full scan
472. if (_indexLookup && this._indexManager) {
473. yield* this._indexedLookup(_indexLookup.field, _indexLookup.value, mode);
474. return;
475. }
477. // Index-based find: collect offsets from index(es), then stream only those records
478. if (index && this._indexManager) {
479. yield* this._indexedStream(index, queryFn, mode);
480. return;
481. }
483. // Flush pending writes so reads see all inserted data
484. if (this._dataStream && this._dataStream.writableLength > 0) {
485. await new Promise(resolve => this._dataStream.once('drain', resolve));
486. }
488. // Check if the data file exists before attempting to read it
489. try {
490. await stat(this._dataPath);
491. } catch (error) {
492. // File doesn't exist - return empty generator (no records)
493. return;
494. }
496. // Convert deleted array to Set for O(1) lookup instead of O(n)
497. const deletedSet = new Set(this._meta.deleted);
499. const readStream = createReadStream(this._dataPath);
500. let chunks = [];
501. let totalLength = 0;
502. let processedBytes = 0;
504. for await (const chunk of readStream) {
505. chunks.push(chunk);
506. totalLength += chunk.length;
508. while (true) {
509. // Exit 1: Not enough data to even read the 4-byte size header.
510. if (totalLength < 4) {
511. break;
512. }
514. // Safely read the header, even if it's split across chunks
515. let headerBuffer;
516. if (chunks[0].length >= 4) {
517. headerBuffer = chunks[0];
518. } else {
519. // The header is fragmented, so we must concat just enough to read it.
520. headerBuffer = Buffer.concat(chunks, 4);
521. }
522. const recSize = headerBuffer.readInt32LE(0);
524. // Validate the record size to prevent infinite loops
525. // A record must be at least as large as its header (4 bytes).
526. // A size of 0 or less is invalid and indicates corruption.
527. if (recSize <= 4) {
528. throw new Error(`Invalid record size read from stream: ${recSize}`);
529. }
531. // Exit 2: We have the size, but not the full record yet.
532. if (totalLength < recSize) {
533. break;
534. }
536. // skip deleted records (only after we have the full record)
537. if (deletedSet.has(processedBytes)) {
538. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
539. continue; // Goes back to while (true)
540. }
542. switch (mode) {
543. case 'raw':
544. yield Buffer.concat(chunks, recSize).subarray(0, recSize);
545. break;
547. case 'mixed':
548. case 'record':
549. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
550. // Skip the 4-byte size header
551. const data = deserialize(recBuffer.subarray(4));
552. const rec = this._classToUse
553. ? Object.assign(new this._classToUse(), data)
554. : data;
555. if (queryFn) {
556. if (queryFn(rec)) {
557. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
558. }
559. } else {
560. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
561. }
562. break;
564. case 'offset':
565. // if someone uses a queryFn with offset, we have to deserialize it first to perform the queryFn
566. if (queryFn) {
567. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
568. const rec = deserialize(recBuffer.subarray(4));
569. if (queryFn(rec)) {
570. yield [processedBytes, recSize];
571. }
572. } else {
573. yield [processedBytes, recSize];
574. }
575. break;
577. default:
578. throw new Error(`Invalid mode: ${mode}`);
579. }
581. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
582. }
583. }
584. }
586. /**
587. * Indexed lookup — reads records directly by offset from the index.
588. * Uses binary search on the index file for O(log n) lookups.
589. * @private
590. */
591. async *_indexedLookup(field, value, mode = 'record') {
592. const entries = await this._indexManager.get(field, value);
593. if (entries.length === 0) return;
595. const deletedSet = new Set(this._meta.deleted);
596. const fileHandle = await open(this._dataPath, 'r');
598. try {
599. for (const entry of entries) {
600. const [offset, length] = entry.loc;
601. if (deletedSet.has(offset)) continue;
603. const buffer = Buffer.alloc(length);
604. await fileHandle.read(buffer, 0, length, offset);
606. switch (mode) {
607. case 'raw':
608. yield buffer;
609. break;
610. case 'mixed': {
611. const data = deserialize(buffer.subarray(4));
612. const rec = this._classToUse
613. ? Object.assign(new this._classToUse(), data)
614. : data;
615. yield [rec, offset, length];
616. break;
617. }
618. case 'record':
619. default: {
620. const data = deserialize(buffer.subarray(4));
621. const rec = this._classToUse
622. ? Object.assign(new this._classToUse(), data)
623. : data;
624. yield rec;
625. break;
626. }
627. }
628. }
629. } finally {
630. await fileHandle.close();
631. }
632. }
634. /**
635. * Collect offsets from one or more indexes, optionally intersect, then stream records.
636. * @param {Object|Array} index - Single index hint or array of hints
637. * @param {function|null} queryFn - Optional filter function
638. * @param {string} mode - Output mode: 'record', 'raw', or 'mixed'
639. * @private
640. */
641. async *_indexedStream(index, queryFn, mode = 'record') {
642. const hints = Array.isArray(index) ? index : [index];
643. const deletedSet = new Set(this._meta.deleted);
645. // Collect offset sets from each index hint
646. const offsetSets = [];
647. for (const hint of hints) {
648. const offsets = new Map(); // offset → [offset, length]
649. if (hint.value !== undefined) {
650. // Exact match via get()
651. const entries = await this._indexManager.get(hint.field, hint.value);
652. for (const e of entries) {
653. if (!deletedSet.has(e.loc[0])) offsets.set(e.loc[0], e.loc);
654. }
655. } else {
656. // Range scan via entries()
657. for await (const e of this._indexManager.entries(hint.field, { from: hint.from, to: hint.to, direction: hint.direction })) {
658. if (!deletedSet.has(e.loc[0])) offsets.set(e.loc[0], e.loc);
659. }
660. }
661. offsetSets.push(offsets);
662. }
664. // Intersect: keep only offsets present in ALL sets
665. let locations;
666. if (offsetSets.length === 1) {
667. locations = Array.from(offsetSets[0].values());
668. } else {
669. // Start with smallest set for efficiency
670. offsetSets.sort((a, b) => a.size - b.size);
671. const [smallest, ...rest] = offsetSets;
672. locations = [];
673. for (const [offset, loc] of smallest) {
674. if (rest.every(s => s.has(offset))) locations.push(loc);
675. }
676. }
678. // Stream records from the intersected locations
679. const fileHandle = await open(this._dataPath, 'r');
680. try {
681. for (const [offset, length] of locations) {
682. const buffer = Buffer.alloc(length);
683. await fileHandle.read(buffer, 0, length, offset);
685. if (mode === 'raw') {
686. yield buffer;
687. } else {
688. const data = deserialize(buffer.subarray(4));
689. const rec = this._classToUse
690. ? Object.assign(new this._classToUse(), data)
691. : data;
693. if (queryFn && !queryFn(rec)) continue;
695. if (mode === 'mixed') {
696. yield [rec, offset, length];
697. } else {
698. yield rec;
699. }
700. }
701. }
702. } finally {
703. await fileHandle.close();
704. }
705. }
707. /**
708. * Re-read meta.json from disk so this instance sees changes made by other processes.
709. * Called automatically before every read operation.
710. */
711. async refresh() {
712. try {
713. this._meta = JSON.parse(await readFile(this._metaPath));
714. } catch (e) {
715. // file missing or corrupt — keep current meta
716. }
717. }
719. async persistMeta() {
720. const metaToSave = { ...this._meta };
722. // Only save nextId if we have a numeric primary key
723. if (this._primaryKeyType !== PrimaryKeyType.NUMBER) {
724. delete metaToSave.nextId;
725. }
727. await writeFile(this._metaPath, JSON.stringify(metaToSave));
728. }
730. /**
731. * Compact the database by removing deleted records
732. * This rewrites the data file without tombstones
733. *
734. * @returns {Promise<void>}
735. */
736. async compact() {
737. await this._acquireLock('compact');
738. try {
739. const writeStream = createWriteStream(this._dataPath + '.tmp', { flags: 'w' });
741. for await (const binary of this.find(null, { mode: 'raw' })) {
742. writeStream.write(binary);
743. }
745. await new Promise((resolve, reject) => {
746. writeStream.end((err) => err ? reject(err) : resolve());
747. });
749. // rename is atomic, so we can just rename the file and it will replace the old one
750. await rename(this._dataPath + '.tmp', this._dataPath);
753. this._meta.deleted = [];
754. await this.persistMeta();
756. // Rebuild indexes — offsets changed after compaction
757. if (this._indexManager) {
758. await this._indexManager.init(this._dataPath, { forceRebuild: true });
759. }
760. } finally {
761. await this._releaseLock();
762. }
763. }
765. async _acquireLock(operation, record, retries = 0) {
766. if (this._lockDepth > 0) {
767. this._lockDepth++;
768. return;
769. }
770. try {
771. await writeFile(this._lockPath, String(process.pid), { flag: 'wx' });
772. this._lockDepth = 1;
773. } catch (e) {
774. if (e.code === 'EEXIST') {
775. this.debug(`Waiting for lock file ${operation} retry #${retries}`, record);
776. await new Promise(resolve => setTimeout(resolve, 100));
777. return this._acquireLock(operation, record, retries + 1);
778. }
779. throw e;
780. }
781. }
783. async _releaseLock() {
784. this._lockDepth--;
785. if (this._lockDepth <= 0) {
786. this._lockDepth = 0;
787. await unlink(this._lockPath).catch(() => { });
788. }
789. }
791. /**
792. * Retrieves a document by its file offset and length
793. * @param {Array} location - [offset, length] tuple
794. * @returns {Promise<Object|null>} The deserialized document or null
795. * @private
796. * @deprecated Currently unused - may be removed in future versions
797. */
798. async _getDocByLocation([offset, length]) {
799. if (!offset || length === 0) return null;
800. const fileHandle = await open(this._dataPath, 'r');
801. try {
802. const buffer = Buffer.alloc(length);
803. await fileHandle.read(buffer, 0, length, offset);
804. return deserialize(buffer);
805. } finally {
806. await fileHandle.close();
807. }
808. }
810. /**
811. * Close the database and persist all pending changes
812. * Should be called before process exit
813. *
814. * @returns {Promise<void>}
815. */
816. async close() {
817. // Persist indexes before closing
818. if (this._indexManager) {
819. await this._indexManager.close();
820. }
822. // Close data stream
823. if (this._dataStream) {
824. await new Promise((resolve, reject) => {
825. this._dataStream.end((err) => err ? reject(err) : resolve());
826. });
827. }
829. // Remove signal handlers
830. if (this._processExitHandler) {
831. process.off('SIGINT', this._processExitHandler);
832. process.off('SIGTERM', this._processExitHandler);
833. this._processExitHandler = null;
834. }
835. }
837. debug(...args) {
838. if (this._debug) {
839. console.log(...args);
840. }
841. }
842. }
844. /**
845. * Helper function to remove a processed chunk from the buffer
846. * @param {number} recSize - Size of the record to remove
847. * @param {number} processedBytes - Current offset in the file
848. * @param {number} totalLength - Total length of buffered data
849. * @param {Buffer[]} chunks - Array of buffer chunks
850. * @returns {[number, number]} Updated [processedBytes, totalLength]
851. */
852. function removeChunk(recSize, processedBytes, totalLength, chunks) {
853. processedBytes += recSize;
854. totalLength -= recSize;
856. let bytesToRemove = recSize;
857. while (bytesToRemove > 0 && chunks.length > 0) {
858. const currentChunk = chunks[0];
859. if (bytesToRemove >= currentChunk.length) {
860. bytesToRemove -= currentChunk.length;
861. chunks.shift();
862. } else {
863. chunks[0] = currentChunk.subarray(bytesToRemove);
864. bytesToRemove = 0;
865. }
866. }
868. return [processedBytes, totalLength];
869. }
871. export default MPackDB;