mpackdb

17.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 = {};
30. _initPromise = null;
31. _initState = 0;
32. _dataPath = null;
33. _dataStream = null;
34. _lockPath = null;
35. _indexManager = null;
36. _meta = {
37. nextId: 0,
38. deleted: [],
39. };
40. _debug = false;
41. _indexPersistInterval = 60000; // 60 seconds default
42. _indexPersistThreshold = 1000; // 1000 entries default
43. _processExitHandler = null;
45. /**
46. * Create a new MPackDB instance
47. *
48. * @param {string} dbFile - Path to the database file (without extension)
49. * @param {Object} options - Configuration options
50. * @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
51. * @param {PrimaryKeyType} [options.primaryKeyType] - Explicit primary key type (overrides prefix)
52. * @param {string[]} [options.indexes] - Array of field names to index. Use * prefix for numeric, @ for UUID
53. * @param {boolean} [options.debug=false] - Enable debug logging
54. * @param {number} [options.indexPersistInterval=60000] - Milliseconds between automatic index persistence (0 to disable)
55. * @param {number} [options.indexPersistThreshold=1000] - Number of changes before auto-persisting indexes
56. *
57. * @example
58. * const db = new MPackDB('data/users', {
59. * primaryKey: '*id', // Numeric auto-increment
60. * indexes: ['email', '*age'], // Index email (lexical) and age (numeric)
61. * indexPersistThreshold: 100
62. * });
63. */
64. constructor(dbFile, { primaryKey, primaryKeyType, indexes, debug, indexPersistInterval, indexPersistThreshold } = {}) {
65. this._dbFile = dbFile || this._dbFile;
66. this._debug = debug || this._debug;
67. if (indexPersistInterval !== undefined) this._indexPersistInterval = indexPersistInterval;
68. if (indexPersistThreshold !== undefined) this._indexPersistThreshold = indexPersistThreshold;
70. // Parse primary key with optional prefix
71. if (primaryKey) {
72. if (primaryKey.startsWith('*')) {
73. // *id = numeric primary key
74. this._primaryKey = primaryKey.slice(1);
75. this._primaryKeyType = PrimaryKeyType.NUMBER;
76. } else if (primaryKey.startsWith('@')) {
77. // @id = UUID primary key
78. this._primaryKey = primaryKey.slice(1);
79. this._primaryKeyType = PrimaryKeyType.UUID;
80. } else {
81. // id = string primary key (lexical)
82. this._primaryKey = primaryKey;
83. this._primaryKeyType = PrimaryKeyType.STRING;
84. }
85. // Allow explicit override
86. if (primaryKeyType !== undefined) {
87. this._primaryKeyType = primaryKeyType;
88. }
89. }
91. const allIndexes = [...new Set([
92. ...(this._primaryKey ? [this._primaryKey] : []),
93. ...(indexes || []),
94. ...(this._indexes || []),
95. ])];
96. this._indexes = [];
97. for (let index of allIndexes) {
98. let cleanIndex = index;
99. if (index.startsWith('*')) {
100. // *field = numeric index
101. cleanIndex = index.slice(1);
102. this._indexTypes[cleanIndex] = IndexType.NUMERIC;
103. } else if (index.startsWith('@')) {
104. // @field = UUID index (lexical)
105. cleanIndex = index.slice(1);
106. this._indexTypes[cleanIndex] = IndexType.LEXICAL;
107. } else if (index === this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER) {
108. // Primary key is numeric
109. this._indexTypes[index] = IndexType.NUMERIC;
110. } else {
111. // Default to lexical
112. this._indexTypes[index] = IndexType.LEXICAL;
113. }
114. this._indexes.push(cleanIndex);
115. }
116. }
118. /**
119. * Initialize the database (called automatically by other methods)
120. * Performs compaction and loads metadata
121. *
122. * @returns {Promise<MPackDB>} The database instance
123. */
124. async init() {
125. if (this._initState === 2) return this;
127. if (this._initState === 1) {
128. return this._initPromise;
129. }
131. this._initState = 1;
132. return this._initPromise = new Promise(async success => {
133. if (!this._dbFile) {
134. throw new Error('No database file specified.');
135. }
137. const dbDir = dirname(this._dbFile);
138. const baseName = basename(this._dbFile, extname(this._dbFile));
139. await mkdir(dbDir, { recursive: true });
141. this._dataPath = resolve(dbDir, `${baseName}.mpack`);
142. this._lockPath = resolve(dbDir, `${baseName}.lock`);
144. try {
145. this._meta = JSON.parse(await readFile(`${this._dbFile}.meta.json`));
146. } catch (e) {
147. this._meta = {
148. nextId: 0,
149. deleted: [],
150. };
151. }
153. this._initState = 2; // compact causes init to run again thats why we set it to 2 here already
155. await this.compact();
157. this._dataStream = createWriteStream(this._dataPath, { flags: 'a' }); // needs to be set after compact as it replaces the file with a tmp file
159. // Initialize IndexManager if indexes are specified
160. if (this._indexes.length > 0) {
161. const dbDir = dirname(this._dbFile);
162. const baseName = basename(this._dbFile, extname(this._dbFile));
163. this._indexManager = new IndexManager(dbDir, baseName, this._indexes, this._indexTypes, this._primaryKeyType);
164. await this._indexManager.init(this._dataPath);
165. // Start auto-persist with configured interval and threshold
166. this._indexManager.startAutoPersist(this._indexPersistInterval, this._indexPersistThreshold);
167. // Register signal handlers for Ctrl+C and kill signals
168. this._processExitHandler = async () => {
169. await this.close();
170. process.exit(0);
171. };
172. process.once('SIGINT', this._processExitHandler);
173. process.once('SIGTERM', this._processExitHandler);
174. }
176. this._initPromise = null;
177. success(this);
178. });
179. }
181. /**
182. * Insert a new record into the database
183. *
184. * @param {Object} record - The record to insert
185. * @param {Object} [options]
186. * @param {boolean} [options.skipPrimaryKey=false] - Skip auto-generating primary key
187. * @returns {Promise<Object>} The inserted record with primary key
188. *
189. * @example
190. * await db.insert({ name: 'Alice', age: 30 });
191. * // Returns: { id: 0, name: 'Alice', age: 30 }
192. */
193. async insert(record, { skipPrimaryKey = false } = {}) {
194. await this.init();
195. await this._acquireLock('insert', record);
196. try {
198. const recToInsert = { ...record };
200. if (this._primaryKey && !skipPrimaryKey && !recToInsert[this._primaryKey]) {
201. if (this._primaryKeyType === PrimaryKeyType.NUMBER) {
202. recToInsert[this._primaryKey] = this._meta.nextId++;
203. } else if (this._primaryKeyType === PrimaryKeyType.UUID) {
204. recToInsert[this._primaryKey] = uuid();
205. }
206. }
208. const packedBuffer = serialize(recToInsert);
209. const fileStat = await stat(this._dataPath).catch(() => ({ size: 0 }));
210. const offset = fileStat.size;
212. await new Promise(resolve => this._dataStream.write(packedBuffer, resolve));
214. const loc = [offset, packedBuffer.length];
215. // indexes
216. if (this._indexManager) {
217. this._indexManager.insert(recToInsert, loc);
218. }
220. // Persist meta if we incremented nextId
221. if (this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER && !skipPrimaryKey && !record[this._primaryKey]) {
222. await this.persistMeta();
223. }
225. return this._primaryKey ? recToInsert[this._primaryKey] : recToInsert;
226. } finally {
227. await this._releaseLock();
228. }
229. }
231. /**
232. * Update records matching a query
233. *
234. * @param {string|Function} mixed - Primary key value or query function
235. * @param {Object|Function} dataOrCallback - Data to update or callback function
236. * @param {Object} [options]
237. * @param {boolean} [options.upsert=false] - Insert if no records match
238. * @returns {Promise<Object[]>} Array of updated records
239. *
240. * @example
241. * // Update by primary key
242. * await db.update(0, { age: 31 });
243. *
244. * // Update with query function
245. * await db.update(r => r.age > 30, { status: 'senior' });
246. *
247. * // Update with callback
248. * await db.update(r => r.age > 30, r => ({ ...r, age: r.age + 1 }));
249. */
250. async update(mixed, dataOrCallback, { upsert = false } = {}) {
251. let insertedRecords = await this.delete(mixed, async record => {
252. return this.insert(typeof dataOrCallback === 'function'
253. ? await dataOrCallback(record)
254. : dataOrCallback,
255. { skipPrimaryKey: true }
256. );
257. });
258. console.log('insertedRecords', insertedRecords);
259. if (upsert && insertedRecords.length === 0) {
260. insertedRecords = await this.insert(typeof dataOrCallback === 'function'
261. ? await dataOrCallback({})
262. : dataOrCallback);
263. }
264. return insertedRecords;
265. }
267. /**
268. * Update records or insert if not found
269. *
270. * @param {string|Function} mixed - Primary key value or query function
271. * @param {Object|Function} dataOrCallback - Data to update/insert or callback function
272. * @returns {Promise<Object[]>} Array of updated/inserted records
273. */
274. async upsert(mixed, dataOrCallback) {
275. return this.update(mixed, dataOrCallback, { upsert: true });
276. }
278. /**
279. * Delete records matching a query
280. *
281. * @param {string|Function} mixed - Primary key value or query function
282. * @param {Function} [callback] - Optional callback to execute for each deleted record
283. * @returns {Promise<Object[]>} Array of deleted records
284. *
285. * @example
286. * // Delete by primary key
287. * await db.delete(0);
288. *
289. * // Delete with query function
290. * await db.delete(r => r.age < 18);
291. */
292. async delete(mixed, callback = async record => record) {
293. await this.init();
294. await this._acquireLock('delete', mixed);
295. try {
296. const promises = [];
297. for await (const [record, offset] of this.find(mixed, { mode: 'mixed' })) {
298. this._meta.deleted.push(offset);
299. if (this._indexManager) {
300. await this._indexManager.remove(record, this._primaryKey);
301. }
302. promises.push(callback(record));
303. }
304. await this.persistMeta();
305. return Promise.all(promises);
306. } finally {
307. await this._releaseLock();
308. }
309. }
311. /**
312. * Finds records in the database
313. * @param {undefined|string|function} [mixed] - Query: undefined/null for all records, string for primary key lookup, function for filter
314. * @param {Object} [options] - Query options
315. * @returns {Cursor} A cursor for iterating over results
316. * @example
317. * // Get all records
318. * for await (const user of db.find()) {
319. * console.log(user.name);
320. * }
321. */
322. find(mixed, options = {}) {
323. if (typeof mixed === 'undefined' || mixed === null || typeof mixed === 'function') {
324. return new Cursor(this, mixed, options);
325. } else {
326. if (!this._primaryKey) {
327. throw new Error('No primary key specified.');
328. }
329. return new Cursor(this, record => {
330. return record[this._primaryKey] === mixed;
331. }, options);
332. }
333. }
335. /**
336. * Generator that yields records from the database file
337. * @param {function|null} [queryFn=null] - Optional filter function
338. * @param {Object} [options] - Generator options
339. * @param {string} [options.mode='record'] - Mode: 'record', 'raw', 'offset', or 'mixed'
340. * @yields {Object|Buffer|Array} Records, buffers, or [record, offset, size] tuples depending on mode
341. */
342. async *recordGenerator(queryFn = null, { mode = 'record' } = {}) {
343. await this.init();
345. // Flush pending writes so reads see all inserted data
346. if (this._dataStream && this._dataStream.writableLength > 0) {
347. await new Promise(resolve => this._dataStream.once('drain', resolve));
348. }
350. // Check if the data file exists before attempting to read it
351. try {
352. await stat(this._dataPath);
353. } catch (error) {
354. // File doesn't exist - return empty generator (no records)
355. return;
356. }
358. // Convert deleted array to Set for O(1) lookup instead of O(n)
359. const deletedSet = new Set(this._meta.deleted);
361. const readStream = createReadStream(this._dataPath);
362. let chunks = [];
363. let totalLength = 0;
364. let processedBytes = 0;
366. for await (const chunk of readStream) {
367. chunks.push(chunk);
368. totalLength += chunk.length;
370. while (true) {
371. // Exit 1: Not enough data to even read the 4-byte size header.
372. if (totalLength < 4) {
373. break;
374. }
376. // Safely read the header, even if it's split across chunks
377. let headerBuffer;
378. if (chunks[0].length >= 4) {
379. headerBuffer = chunks[0];
380. } else {
381. // The header is fragmented, so we must concat just enough to read it.
382. headerBuffer = Buffer.concat(chunks, 4);
383. }
384. const recSize = headerBuffer.readInt32LE(0);
386. // Validate the record size to prevent infinite loops
387. // A record must be at least as large as its header (4 bytes).
388. // A size of 0 or less is invalid and indicates corruption.
389. if (recSize <= 4) {
390. throw new Error(`Invalid record size read from stream: ${recSize}`);
391. }
393. // Exit 2: We have the size, but not the full record yet.
394. if (totalLength < recSize) {
395. break;
396. }
398. // skip deleted records (only after we have the full record)
399. if (deletedSet.has(processedBytes)) {
400. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
401. continue; // Goes back to while (true)
402. }
404. switch (mode) {
405. case 'raw':
406. yield Buffer.concat(chunks, recSize).subarray(0, recSize);
407. break;
409. case 'mixed':
410. case 'record':
411. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
412. // Skip the 4-byte size header
413. const data = deserialize(recBuffer.subarray(4));
414. const rec = this._classToUse
415. ? Object.assign(new this._classToUse(), data)
416. : data;
417. if (queryFn) {
418. if (queryFn(rec)) {
419. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
420. }
421. } else {
422. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
423. }
424. break;
426. case 'offset':
427. // if someone uses a queryFn with offset, we have to deserialize it first to perform the queryFn
428. if (queryFn) {
429. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
430. const rec = deserialize(recBuffer.subarray(4));
431. if (queryFn(rec)) {
432. yield [processedBytes, recSize];
433. }
434. } else {
435. yield [processedBytes, recSize];
436. }
437. break;
439. default:
440. throw new Error(`Invalid mode: ${mode}`);
441. }
443. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
444. }
445. }
446. }
448. async persistMeta() {
449. const metaToSave = { ...this._meta };
451. // Only save nextId if we have a numeric primary key
452. if (this._primaryKeyType !== PrimaryKeyType.NUMBER) {
453. delete metaToSave.nextId;
454. }
456. await writeFile(`${this._dbFile}.meta.json`, JSON.stringify(metaToSave));
457. }
459. /**
460. * Compact the database by removing deleted records
461. * This rewrites the data file without tombstones
462. *
463. * @returns {Promise<void>}
464. */
465. async compact() {
466. await this._acquireLock('compact');
467. try {
468. const writeStream = createWriteStream(this._dataPath + '.tmp', { flags: 'w' });
470. for await (const binary of this.find(null, { mode: 'raw' })) {
471. writeStream.write(binary);
472. }
474. writeStream.end();
476. // rename is atomic, so we can just rename the file and it will replace the old one
477. await rename(this._dataPath + '.tmp', this._dataPath);
480. this._meta.deleted = [];
481. await this.persistMeta();
482. } finally {
483. await this._releaseLock();
484. }
485. }
487. async _acquireLock(operation, record, retries = 0) {
488. try {
489. await writeFile(this._lockPath, String(process.pid), { flag: 'wx' });
490. } catch (e) {
491. if (e.code === 'EEXIST') {
492. this.debug(`Waiting for lock file ${operation} retry #${retries}`, record);
493. await new Promise(resolve => setTimeout(resolve, 100));
494. return this._acquireLock(operation, record, retries + 1);
495. }
496. throw e;
497. }
498. }
500. async _releaseLock() {
501. await unlink(this._lockPath).catch(() => { });
502. }
504. /**
505. * Retrieves a document by its file offset and length
506. * @param {Array} location - [offset, length] tuple
507. * @returns {Promise<Object|null>} The deserialized document or null
508. * @private
509. * @deprecated Currently unused - may be removed in future versions
510. */
511. async _getDocByLocation([offset, length]) {
512. if (!offset || length === 0) return null;
513. const fileHandle = await open(this._dataPath, 'r');
514. try {
515. const buffer = Buffer.alloc(length);
516. await fileHandle.read(buffer, 0, length, offset);
517. return deserialize(buffer);
518. } finally {
519. await fileHandle.close();
520. }
521. }
523. /**
524. * Close the database and persist all pending changes
525. * Should be called before process exit
526. *
527. * @returns {Promise<void>}
528. */
529. async close() {
530. // Persist indexes before closing
531. if (this._indexManager) {
532. await this._indexManager.close();
533. }
535. // Close data stream
536. if (this._dataStream) {
537. await new Promise((resolve, reject) => {
538. this._dataStream.end((err) => err ? reject(err) : resolve());
539. });
540. }
542. // Remove signal handlers
543. if (this._processExitHandler) {
544. process.off('SIGINT', this._processExitHandler);
545. process.off('SIGTERM', this._processExitHandler);
546. this._processExitHandler = null;
547. }
548. }
550. debug(...args) {
551. if (this._debug) {
552. console.log(...args);
553. }
554. }
555. }
557. /**
558. * Helper function to remove a processed chunk from the buffer
559. * @param {number} recSize - Size of the record to remove
560. * @param {number} processedBytes - Current offset in the file
561. * @param {number} totalLength - Total length of buffered data
562. * @param {Buffer[]} chunks - Array of buffer chunks
563. * @returns {[number, number]} Updated [processedBytes, totalLength]
564. */
565. function removeChunk(recSize, processedBytes, totalLength, chunks) {
566. processedBytes += recSize;
567. totalLength -= recSize;
569. let bytesToRemove = recSize;
570. while (bytesToRemove > 0 && chunks.length > 0) {
571. const currentChunk = chunks[0];
572. if (bytesToRemove >= currentChunk.length) {
573. bytesToRemove -= currentChunk.length;
574. chunks.shift();
575. } else {
576. chunks[0] = currentChunk.subarray(bytesToRemove);
577. bytesToRemove = 0;
578. }
579. }
581. return [processedBytes, totalLength];
582. }
584. export default MPackDB;