mpackdb

18.0 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 = this._classToUse
199. ? Object.assign(new this._classToUse(), record)
200. : { ...record };
202. if (this._primaryKey && !skipPrimaryKey && !recToInsert[this._primaryKey]) {
203. if (this._primaryKeyType === PrimaryKeyType.NUMBER) {
204. recToInsert[this._primaryKey] = this._meta.nextId++;
205. } else if (this._primaryKeyType === PrimaryKeyType.UUID) {
206. recToInsert[this._primaryKey] = uuid();
207. }
208. }
210. const packedBuffer = serialize(recToInsert);
211. const fileStat = await stat(this._dataPath).catch(() => ({ size: 0 }));
212. const offset = fileStat.size;
214. await new Promise(resolve => this._dataStream.write(packedBuffer, resolve));
216. const loc = [offset, packedBuffer.length];
217. // indexes
218. if (this._indexManager) {
219. this._indexManager.insert(recToInsert, loc);
220. }
222. // Persist meta if we incremented nextId
223. if (this._primaryKey && this._primaryKeyType === PrimaryKeyType.NUMBER && !skipPrimaryKey && !record[this._primaryKey]) {
224. await this.persistMeta();
225. }
227. return this._primaryKey ? recToInsert[this._primaryKey] : recToInsert;
228. } finally {
229. await this._releaseLock();
230. }
231. }
233. /**
234. * Update records matching a query
235. *
236. * @param {string|Function} mixed - Primary key value or query function
237. * @param {Object|Function} dataOrCallback - Data to update or callback function
238. * @param {Object} [options]
239. * @param {boolean} [options.upsert=false] - Insert if no records match
240. * @returns {Promise<Object[]>} Array of updated records
241. *
242. * @example
243. * // Update by primary key
244. * await db.update(0, { age: 31 });
245. *
246. * // Update with query function
247. * await db.update(r => r.age > 30, { status: 'senior' });
248. *
249. * // Update with callback
250. * await db.update(r => r.age > 30, r => ({ ...r, age: r.age + 1 }));
251. */
252. async update(mixed, dataOrCallback, { upsert = false } = {}) {
253. let insertedRecords = await this.delete(mixed, async record => {
254. return this.insert(typeof dataOrCallback === 'function'
255. ? await dataOrCallback(record)
256. : dataOrCallback,
257. { skipPrimaryKey: true }
258. );
259. });
260. console.log('insertedRecords', insertedRecords);
261. if (upsert && insertedRecords.length === 0) {
262. insertedRecords = await this.insert(typeof dataOrCallback === 'function'
263. ? await dataOrCallback({})
264. : dataOrCallback);
265. }
266. return insertedRecords;
267. }
269. /**
270. * Update records or insert if not found
271. *
272. * @param {string|Function} mixed - Primary key value or query function
273. * @param {Object|Function} dataOrCallback - Data to update/insert or callback function
274. * @returns {Promise<Object[]>} Array of updated/inserted records
275. */
276. async upsert(mixed, dataOrCallback) {
277. return this.update(mixed, dataOrCallback, { upsert: true });
278. }
280. /**
281. * Delete records matching a query
282. *
283. * @param {string|Function} mixed - Primary key value or query function
284. * @param {Function} [callback] - Optional callback to execute for each deleted record
285. * @returns {Promise<Object[]>} Array of deleted records
286. *
287. * @example
288. * // Delete by primary key
289. * await db.delete(0);
290. *
291. * // Delete with query function
292. * await db.delete(r => r.age < 18);
293. */
294. async delete(mixed, callback = async record => record) {
295. await this.init();
296. await this._acquireLock('delete', mixed);
297. try {
298. const promises = [];
299. for await (const [record, offset] of this.find(mixed, { mode: 'mixed' })) {
300. this._meta.deleted.push(offset);
301. if (this._indexManager) {
302. await this._indexManager.remove(record, this._primaryKey);
303. }
304. promises.push(callback(record));
305. }
306. await this.persistMeta();
307. return Promise.all(promises);
308. } finally {
309. await this._releaseLock();
310. }
311. }
313. /**
314. * Finds records in the database
315. * @param {undefined|string|function} [mixed] - Query: undefined/null for all records, string for primary key lookup, function for filter
316. * @param {Object} [options] - Query options
317. * @returns {Cursor} A cursor for iterating over results
318. * @example
319. * // Get all records
320. * for await (const user of db.find()) {
321. * console.log(user.name);
322. * }
323. */
324. find(mixed, options = {}) {
325. if (typeof mixed === 'undefined' || mixed === null || typeof mixed === 'function') {
326. return new Cursor(this, mixed, options);
327. } else {
328. if (!this._primaryKey) {
329. throw new Error('No primary key specified.');
330. }
331. return new Cursor(this, record => {
332. return record[this._primaryKey] === mixed;
333. }, options);
334. }
335. }
337. /**
338. * Generator that yields records from the database file
339. * @param {function|null} [queryFn=null] - Optional filter function
340. * @param {Object} [options] - Generator options
341. * @param {string} [options.mode='record'] - Mode: 'record', 'raw', 'offset', or 'mixed'
342. * @yields {Object|Buffer|Array} Records, buffers, or [record, offset, size] tuples depending on mode
343. */
344. async *recordGenerator(queryFn = null, { mode = 'record' } = {}) {
345. await this.init();
347. // Flush pending writes so reads see all inserted data
348. if (this._dataStream && this._dataStream.writableLength > 0) {
349. await new Promise(resolve => this._dataStream.once('drain', resolve));
350. }
352. // Check if the data file exists before attempting to read it
353. try {
354. await stat(this._dataPath);
355. } catch (error) {
356. // File doesn't exist - return empty generator (no records)
357. return;
358. }
360. // Convert deleted array to Set for O(1) lookup instead of O(n)
361. const deletedSet = new Set(this._meta.deleted);
363. const readStream = createReadStream(this._dataPath);
364. let chunks = [];
365. let totalLength = 0;
366. let processedBytes = 0;
368. for await (const chunk of readStream) {
369. chunks.push(chunk);
370. totalLength += chunk.length;
372. while (true) {
373. // Exit 1: Not enough data to even read the 4-byte size header.
374. if (totalLength < 4) {
375. break;
376. }
378. // Safely read the header, even if it's split across chunks
379. let headerBuffer;
380. if (chunks[0].length >= 4) {
381. headerBuffer = chunks[0];
382. } else {
383. // The header is fragmented, so we must concat just enough to read it.
384. headerBuffer = Buffer.concat(chunks, 4);
385. }
386. const recSize = headerBuffer.readInt32LE(0);
388. // Validate the record size to prevent infinite loops
389. // A record must be at least as large as its header (4 bytes).
390. // A size of 0 or less is invalid and indicates corruption.
391. if (recSize <= 4) {
392. throw new Error(`Invalid record size read from stream: ${recSize}`);
393. }
395. // Exit 2: We have the size, but not the full record yet.
396. if (totalLength < recSize) {
397. break;
398. }
400. // skip deleted records (only after we have the full record)
401. if (deletedSet.has(processedBytes)) {
402. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
403. continue; // Goes back to while (true)
404. }
406. switch (mode) {
407. case 'raw':
408. yield Buffer.concat(chunks, recSize).subarray(0, recSize);
409. break;
411. case 'mixed':
412. case 'record':
413. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
414. // Skip the 4-byte size header
415. const data = deserialize(recBuffer.subarray(4));
416. const rec = this._classToUse
417. ? Object.assign(new this._classToUse(), data)
418. : data;
419. if (queryFn) {
420. if (queryFn(rec)) {
421. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
422. }
423. } else {
424. yield mode === 'mixed' ? [rec, processedBytes, recSize] : rec;
425. }
426. break;
428. case 'offset':
429. // if someone uses a queryFn with offset, we have to deserialize it first to perform the queryFn
430. if (queryFn) {
431. const recBuffer = Buffer.concat(chunks, recSize).subarray(0, recSize);
432. const rec = deserialize(recBuffer.subarray(4));
433. if (queryFn(rec)) {
434. yield [processedBytes, recSize];
435. }
436. } else {
437. yield [processedBytes, recSize];
438. }
439. break;
441. default:
442. throw new Error(`Invalid mode: ${mode}`);
443. }
445. [processedBytes, totalLength] = removeChunk(recSize, processedBytes, totalLength, chunks);
446. }
447. }
448. }
450. async persistMeta() {
451. const metaToSave = { ...this._meta };
453. // Only save nextId if we have a numeric primary key
454. if (this._primaryKeyType !== PrimaryKeyType.NUMBER) {
455. delete metaToSave.nextId;
456. }
458. await writeFile(`${this._dbFile}.meta.json`, JSON.stringify(metaToSave));
459. }
461. /**
462. * Compact the database by removing deleted records
463. * This rewrites the data file without tombstones
464. *
465. * @returns {Promise<void>}
466. */
467. async compact() {
468. await this._acquireLock('compact');
469. try {
470. const writeStream = createWriteStream(this._dataPath + '.tmp', { flags: 'w' });
472. for await (const binary of this.find(null, { mode: 'raw' })) {
473. writeStream.write(binary);
474. }
476. writeStream.end();
478. // rename is atomic, so we can just rename the file and it will replace the old one
479. await rename(this._dataPath + '.tmp', this._dataPath);
482. this._meta.deleted = [];
483. await this.persistMeta();
484. } finally {
485. await this._releaseLock();
486. }
487. }
489. async _acquireLock(operation, record, retries = 0) {
490. try {
491. await writeFile(this._lockPath, String(process.pid), { flag: 'wx' });
492. } catch (e) {
493. if (e.code === 'EEXIST') {
494. this.debug(`Waiting for lock file ${operation} retry #${retries}`, record);
495. await new Promise(resolve => setTimeout(resolve, 100));
496. return this._acquireLock(operation, record, retries + 1);
497. }
498. throw e;
499. }
500. }
502. async _releaseLock() {
503. await unlink(this._lockPath).catch(() => { });
504. }
506. /**
507. * Retrieves a document by its file offset and length
508. * @param {Array} location - [offset, length] tuple
509. * @returns {Promise<Object|null>} The deserialized document or null
510. * @private
511. * @deprecated Currently unused - may be removed in future versions
512. */
513. async _getDocByLocation([offset, length]) {
514. if (!offset || length === 0) return null;
515. const fileHandle = await open(this._dataPath, 'r');
516. try {
517. const buffer = Buffer.alloc(length);
518. await fileHandle.read(buffer, 0, length, offset);
519. return deserialize(buffer);
520. } finally {
521. await fileHandle.close();
522. }
523. }
525. /**
526. * Close the database and persist all pending changes
527. * Should be called before process exit
528. *
529. * @returns {Promise<void>}
530. */
531. async close() {
532. // Persist indexes before closing
533. if (this._indexManager) {
534. await this._indexManager.close();
535. }
537. // Close data stream
538. if (this._dataStream) {
539. await new Promise((resolve, reject) => {
540. this._dataStream.end((err) => err ? reject(err) : resolve());
541. });
542. }
544. // Remove signal handlers
545. if (this._processExitHandler) {
546. process.off('SIGINT', this._processExitHandler);
547. process.off('SIGTERM', this._processExitHandler);
548. this._processExitHandler = null;
549. }
550. }
552. debug(...args) {
553. if (this._debug) {
554. console.log(...args);
555. }
556. }
557. }
559. /**
560. * Helper function to remove a processed chunk from the buffer
561. * @param {number} recSize - Size of the record to remove
562. * @param {number} processedBytes - Current offset in the file
563. * @param {number} totalLength - Total length of buffered data
564. * @param {Buffer[]} chunks - Array of buffer chunks
565. * @returns {[number, number]} Updated [processedBytes, totalLength]
566. */
567. function removeChunk(recSize, processedBytes, totalLength, chunks) {
568. processedBytes += recSize;
569. totalLength -= recSize;
571. let bytesToRemove = recSize;
572. while (bytesToRemove > 0 && chunks.length > 0) {
573. const currentChunk = chunks[0];
574. if (bytesToRemove >= currentChunk.length) {
575. bytesToRemove -= currentChunk.length;
576. chunks.shift();
577. } else {
578. chunks[0] = currentChunk.subarray(bytesToRemove);
579. bytesToRemove = 0;
580. }
581. }
583. return [processedBytes, totalLength];
584. }
586. export default MPackDB;