• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

albe / node-event-storage / 23402128479

22 Mar 2026 11:31AM UTC coverage: 97.958% (+0.1%) from 97.826%
23402128479

Pull #107

github

web-flow
Merge 967ba2d7d into ea855be31
Pull Request #107: Implement auto-repair

658 of 691 branches covered (95.22%)

Branch coverage included in aggregate %.

54 of 55 new or added lines in 6 files covered. (98.18%)

2 existing lines in 2 files now uncovered.

1501 of 1513 relevant lines covered (99.21%)

1302.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.04
/src/Storage/ReadableStorage.js
1
const fs = require('fs');
4✔
2
const path = require('path');
4✔
3
const events = require('events');
4✔
4
const Partition = require('../Partition');
4✔
5
const Index = require('../Index');
4✔
6
const { assert, createHmac, matches, wrapAndCheck, buildMetadataForMatcher } = require('../util');
4✔
7

8
const DEFAULT_READ_BUFFER_SIZE = 4 * 1024;
4✔
9

10
/**
11
 * Reverses the items of an iterable
12
 * @param {Generator|Iterable} iterator
13
 * @returns {Generator<*>}
14
 */
15
function *reverse(iterator) {
16
    const items = Array.from(iterator);
64✔
17
    for (let i = items.length - 1; i >= 0; i--) {
64✔
18
        yield items[i];
544✔
19
    }
20
}
21

22
/**
23
 * @typedef {object|function(object):boolean} Matcher
24
 */
25

26
/**
27
 * An append-only storage with highly performant positional range scans.
28
 * It's highly optimized for an event-store and hence does not support compaction or data-rewrite, nor any querying
29
 */
30
class ReadableStorage extends events.EventEmitter {
31

32
    /**
33
     * @param {string} [storageName] The name of the storage.
34
     * @param {object} [config] An object with storage parameters.
35
     * @param {object} [config.serializer] A serializer object with methods serialize(document) and deserialize(data).
36
     * @param {function(object): string} config.serializer.serialize Default is JSON.stringify.
37
     * @param {function(string): object} config.serializer.deserialize Default is JSON.parse.
38
     * @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
39
     * @param {string} [config.indexDirectory] The path where the indexes should be stored. Defaults to dataDirectory.
40
     * @param {string} [config.indexFile] The name of the primary index. Default '{storageName}.index'.
41
     * @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
42
     * @param {object} [config.indexOptions] An options object that should be passed to all indexes on construction.
43
     * @param {string} [config.hmacSecret] A private key that is used to verify matchers retrieved from indexes.
44
     * @param {object} [config.metadata] A metadata object to be stored in all partitions belonging to this storage.
45
     */
46
    constructor(storageName = 'storage', config = {}) {
476!
47
        super();
944✔
48
        if (typeof storageName !== 'string') {
944✔
49
            config = storageName;
52✔
50
            storageName = undefined;
52✔
51
        }
52

53
        this.storageFile = storageName || 'storage';
944✔
54
        const defaults = {
944✔
55
            serializer: { serialize: JSON.stringify, deserialize: JSON.parse },
56
            dataDirectory: '.',
57
            indexFile: this.storageFile + '.index',
58
            indexOptions: {},
59
            hmacSecret: '',
60
            metadata: {}
61
        };
62
        config = Object.assign(defaults, config);
944✔
63
        this.serializer = config.serializer;
944✔
64

65
        this.hmac = createHmac(config.hmacSecret);
944✔
66

67
        this.dataDirectory = path.resolve(config.dataDirectory);
944✔
68

69
        this.scanPartitions(config);
944✔
70
        this.initializeIndexes(config);
944✔
71
    }
72

73
    /**
74
     * @protected
75
     * @param {string} name
76
     * @param {object} [options]
77
     * @returns {{ index: ReadableIndex, matcher?: Matcher }}
78
     */
79
    createIndex(name, options = {}) {
×
80
        /** @type ReadableIndex */
81
        const index = new Index.ReadOnly(name, options);
136✔
82
        return { index };
132✔
83
    }
84

85
    /**
86
     * @protected
87
     * @param {string} name
88
     * @param {object} [options]
89
     * @returns {ReadablePartition}
90
     */
91
    createPartition(name, options = {}) {
×
92
        return new Partition.ReadOnly(name, options);
52✔
93
    }
94

95
    /**
96
     * Create/open the primary index and build the base configuration for all secondary indexes.
97
     *
98
     * @private
99
     * @param {object} config The configuration object
100
     * @returns void
101
     */
102
    initializeIndexes(config) {
103
        this.indexDirectory = path.resolve(config.indexDirectory || this.dataDirectory);
944✔
104

105
        this.indexOptions = config.indexOptions;
944✔
106
        this.indexOptions.dataDirectory = this.indexDirectory;
944✔
107
        // Safety precaution to prevent accidentally restricting main index
108
        delete this.indexOptions.matcher;
944✔
109
        const { index } = this.createIndex(config.indexFile, this.indexOptions);
944✔
110
        this.index = index;
940✔
111
        this.secondaryIndexes = {};
940✔
112
        this.readonlyIndexes = {};
940✔
113
    }
114

115
    /**
116
     * The amount of documents in the storage.
117
     * @returns {number}
118
     */
119
    get length() {
120
        return this.index.length;
4,132✔
121
    }
122

123
    /**
124
     * Scan the data directory for all existing partitions.
125
     * Every file beginning with the storageFile name is considered a partition.
126
     *
127
     * @private
128
     * @param {object} config The configuration object containing options for the partitions.
129
     * @returns void
130
     */
131
    scanPartitions(config) {
132
        const defaults = {
944✔
133
            readBufferSize: DEFAULT_READ_BUFFER_SIZE
134
        };
135
        this.partitionConfig = Object.assign(defaults, config);
944✔
136
        this.partitions = Object.create(null);
944✔
137

138
        const files = fs.readdirSync(this.dataDirectory);
944✔
139
        for (let file of files) {
944✔
140
            if (file.substr(-6) === '.index') continue;
472✔
141
            if (file.substr(-7) === '.branch') continue;
308✔
142
            if (file.substr(-5) === '.lock') continue;
296✔
143
            if (file.substr(0, this.storageFile.length) !== this.storageFile) continue;
208✔
144

145
            const partition = this.createPartition(file, this.partitionConfig);
136✔
146
            this.partitions[partition.id] = partition;
136✔
147
        }
148
    }
149

150
    /**
151
     * Open the storage and indexes and create read and write buffers eagerly.
152
     * Will emit an 'opened' event if finished.
153
     *
154
     * @api
155
     * @returns {boolean}
156
     */
157
    open() {
158
        this.index.open();
872✔
159

160
        this.forEachSecondaryIndex(index => index.open());
872✔
161

162
        this.emit('opened');
872✔
163
        return true;
872✔
164
    }
165

166
    /**
167
     * Close the storage and frees up all resources.
168
     * Will emit a 'closed' event when finished.
169
     *
170
     * @api
171
     * @returns void
172
     */
173
    close() {
174
        this.index.close();
1,440✔
175
        this.forEachSecondaryIndex(index => index.close());
1,440✔
176
        for (let index of Object.values(this.readonlyIndexes)) {
1,440✔
177
            index.close();
44✔
178
        }
179
        this.forEachPartition(partition => partition.close());
1,440✔
180
        this.emit('closed');
1,440✔
181
    }
182

183
    /**
184
     * Get a partition either by name or by id.
185
     * If a partition with the given name does not exist, a new one will be created.
186
     * If a partition with the given id does not exist, an error is thrown.
187
     *
188
     * @protected
189
     * @param {string|number} partitionIdentifier The partition name or the partition Id
190
     * @returns {ReadablePartition}
191
     * @throws {Error} If an id is given and no such partition exists.
192
     */
193
    getPartition(partitionIdentifier) {
194
        assert(partitionIdentifier in this.partitions, `Partition #${partitionIdentifier} does not exist.`);
4,040✔
195

196
        this.partitions[partitionIdentifier].open();
4,040✔
197
        return this.partitions[partitionIdentifier];
4,040✔
198
    }
199

200
    /**
201
     * Register a handler that is called before a document is read from a partition.
202
     * The handler receives the position and the partition metadata and may throw to abort the read.
203
     * Multiple handlers can be registered; all run on every read in registration order.
204
     * Equivalent to `storage.on('preRead', hook)`.
205
     *
206
     * @api
207
     * @param {function(number, object): void} hook A function receiving (position, partitionMetadata).
208
     */
209
    preRead(hook) {
210
        this.on('preRead', hook);
12✔
211
    }
212

213
    /**
214
     * @protected
215
     * @param {number} partitionId The partition to read from.
216
     * @param {number} position The file position to read from.
217
     * @param {number} [size] The expected byte size of the document at the given position.
218
     * @returns {object} The document stored at the given position.
219
     * @throws {Error} if the document at the given position can not be deserialized.
220
     */
221
    readFrom(partitionId, position, size) {
222
        const partition = this.getPartition(partitionId);
3,972✔
223
        if (this.listenerCount('preRead') > 0) {
3,972✔
224
            this.emit('preRead', position, partition.metadata);
84✔
225
        }
226
        const data = partition.readFrom(position, size);
3,960✔
227
        return this.serializer.deserialize(data);
3,960✔
228
    }
229

230
    /**
231
     * Read a single document from the given position, in the full index or in the provided index.
232
     *
233
     * @api
234
     * @param {number} number The 1-based document number (inside the given index) to read.
235
     * @param {ReadableIndex} [index] The index to use for finding the document position.
236
     * @returns {object} The document at the given position inside the index.
237
     */
238
    read(number, index) {
239
        index = index || this.index;
432✔
240

241
        if (!index.isOpen()) {
432✔
242
            index.open();
4✔
243
        }
244

245
        const entry = index.get(number);
432✔
246
        if (entry === false) {
432✔
247
            return false;
4✔
248
        }
249

250
        return this.readFrom(entry.partition, entry.position, entry.size);
428✔
251
    }
252

253
    /**
254
     * Read a range of documents from the given position range, in the full index or in the provided index.
255
     * Returns a generator in order to reduce memory usage and be able to read lots of documents with little latency.
256
     *
257
     * @api
258
     * @param {number} from The 1-based document number (inclusive) to start reading from.
259
     * @param {number} [until] The 1-based document number (inclusive) to read until. Defaults to index.length.
260
     * @param {ReadableIndex} [index] The index to use for finding the documents in the range.
261
     * @returns {Generator<object>} A generator that will read each document in the range one by one.
262
     */
263
    *readRange(from, until = -1, index = null) {
72✔
264
        index = index || this.index;
524✔
265
        index.open();
524✔
266

267
        const readFrom = wrapAndCheck(from, index.length);
524✔
268
        const readUntil = wrapAndCheck(until, index.length);
524✔
269
        assert(readFrom > 0 && readUntil > 0, `Range scan error for range ${from} - ${until}.`);
524✔
270

271
        if (readFrom > readUntil) {
508✔
272
            const batchSize = 10;
44✔
273
            let batchUntil = readFrom;
44✔
274
            while (batchUntil >= readUntil) {
44✔
275
                const batchFrom = Math.max(readUntil, batchUntil - batchSize);
64✔
276
                yield* reverse(this.iterateRange(batchFrom, batchUntil, index));
64✔
277
                batchUntil = batchFrom - 1;
64✔
278
            }
279
            return undefined;
44✔
280
        }
281

282
        yield* this.iterateRange(readFrom, readUntil, index);
464✔
283
    }
284

285
    /**
286
     * Iterate all documents in this storage in range from to until inside the index.
287
     * @private
288
     * @param {number} from
289
     * @param {number} until
290
     * @param {ReadableIndex} index
291
     * @returns {Generator<object>}
292
     */
293
    *iterateRange(from, until, index) {
294
        const entries = index.range(from, until);
528✔
295
        for (let entry of entries) {
528✔
296
            const document = this.readFrom(entry.partition, entry.position, entry.size);
1,780✔
297
            yield document;
1,776✔
298
        }
299
    }
300

301
    /**
302
     * Open an existing readonly index for reading, without registering it in the secondary indexes write path.
303
     * Use this for indexes whose files carry a status marker (e.g. `stream-foo.closed.index`).
304
     *
305
     * @api
306
     * @param {string} name The readonly index name (e.g. 'stream-foo.closed').
307
     * @returns {ReadableIndex}
308
     * @throws {Error} if the readonly index does not exist.
309
     */
310
    openReadonlyIndex(name) {
311
        if (name in this.readonlyIndexes) {
44!
UNCOV
312
            return this.readonlyIndexes[name];
×
313
        }
314
        const indexName = this.storageFile + '.' + name + '.index';
44✔
315
        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
44✔
316
        const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions));
44✔
317
        index.open();
44✔
318
        this.readonlyIndexes[name] = index;
44✔
319
        return index;
44✔
320
    }
321

322
    /**
323
     * Open an existing index.
324
     *
325
     * @api
326
     * @param {string} name The index name.
327
     * @param {Matcher} [matcher] The matcher object or function that the index needs to have been defined with. If not given it will not be validated.
328
     * @returns {ReadableIndex}
329
     * @throws {Error} if the index with that name does not exist.
330
     * @throws {Error} if the HMAC for the matcher does not match.
331
     */
332
    openIndex(name, matcher) {
333
        if (name === '_all') {
741✔
334
            return this.index;
8✔
335
        }
336
        if (name in this.secondaryIndexes) {
733✔
337
            return this.secondaryIndexes[name].index;
629✔
338
        }
339

340
        const indexName = this.storageFile + '.' + name + '.index';
104✔
341
        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
104✔
342

343
        const metadata = buildMetadataForMatcher(matcher, this.hmac);
100✔
344
        let { index } = this.secondaryIndexes[name] = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata }));
100✔
345

346
        index.open();
88✔
347
        return index;
88✔
348
    }
349

350
    /**
351
     * Helper method to iterate over all documents.
352
     *
353
     * @protected
354
     * @param {function(object, EntryInterface)} iterationHandler
355
     */
356
    forEachDocument(iterationHandler) {
357
        /* istanbul ignore if  */
358
        if (typeof iterationHandler !== 'function') {
836✔
359
            return;
360
        }
361

362
        const entries = this.index.all();
836✔
363

364
        for (let entry of entries) {
836✔
365
            const document = this.readFrom(entry.partition, entry.position, entry.size);
1,752✔
366
            iterationHandler(document, entry);
1,752✔
367
        }
368
    }
369

370
    /**
371
     * Helper method to iterate over all secondary indexes.
372
     *
373
     * @protected
374
     * @param {function(ReadableIndex, string)} iterationHandler
375
     * @param {object} [matchDocument] If supplied, only indexes the document matches on will be iterated.
376
     */
377
    forEachSecondaryIndex(iterationHandler, matchDocument) {
378
        /* istanbul ignore if  */
379
        if (typeof iterationHandler !== 'function') {
5,052✔
380
            return;
381
        }
382

383
        for (let indexName of Object.keys(this.secondaryIndexes)) {
5,052✔
384
            if (!matchDocument || matches(matchDocument, this.secondaryIndexes[indexName].matcher)) {
4,864✔
385
                iterationHandler(this.secondaryIndexes[indexName].index, indexName);
2,584✔
386
            }
387
        }
388
    }
389

390
    /**
391
     * Helper method to iterate over all partitions.
392
     *
393
     * @protected
394
     * @param {function(ReadablePartition)} iterationHandler
395
     */
396
    forEachPartition(iterationHandler) {
397
        /* istanbul ignore if  */
398
        if (typeof iterationHandler !== 'function') {
1,576✔
399
            return;
400
        }
401

402
        for (let partition of Object.keys(this.partitions)) {
1,576✔
403
            iterationHandler(this.partitions[partition]);
1,544✔
404
        }
405
    }
406

407
}
408

409
module.exports = ReadableStorage;
4✔
410
module.exports.matches = matches;
4✔
411
module.exports.CorruptFileError = Partition.CorruptFileError;
4✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc