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

albe / node-event-storage / 23383240597

21 Mar 2026 03:55PM UTC coverage: 97.927% (+0.1%) from 97.826%
23383240597

Pull #107

github

web-flow
Merge 2c94d3694 into eaa8ee102
Pull Request #107: Start implementing auto-repair

639 of 672 branches covered (95.09%)

Branch coverage included in aggregate %.

32 of 33 new or added lines in 5 files covered. (96.97%)

2 existing lines in 2 files now uncovered.

1487 of 1499 relevant lines covered (99.2%)

1295.2 hits per line

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

96.99
/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 = {}) {
448!
47
        super();
916✔
48
        if (typeof storageName !== 'string') {
916✔
49
            config = storageName;
52✔
50
            storageName = undefined;
52✔
51
        }
52

53
        this.storageFile = storageName || 'storage';
916✔
54
        const defaults = {
916✔
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);
916✔
63
        this.serializer = config.serializer;
916✔
64

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

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

69
        this.scanPartitions(config);
916✔
70
        this.initializeIndexes(config);
916✔
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);
916✔
104

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

115
    /**
116
     * The amount of documents in the storage.
117
     * @returns {number}
118
     */
119
    get length() {
120
        return this.index.length;
4,052✔
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 = {
916✔
133
            readBufferSize: DEFAULT_READ_BUFFER_SIZE
134
        };
135
        this.partitionConfig = Object.assign(defaults, config);
916✔
136
        this.partitions = Object.create(null);
916✔
137

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

145
            const partition = this.createPartition(file, this.partitionConfig);
124✔
146
            this.partitions[partition.id] = partition;
124✔
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();
844✔
159

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

162
        this.emit('opened');
844✔
163
        return true;
844✔
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,384✔
175
        this.forEachSecondaryIndex(index => index.close());
1,384✔
176
        for (let index of Object.values(this.readonlyIndexes)) {
1,384✔
177
            index.close();
44✔
178
        }
179
        this.forEachPartition(partition => partition.close());
1,384✔
180
        this.emit('closed');
1,384✔
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,036✔
195

196
        this.partitions[partitionIdentifier].open();
4,036✔
197
        return this.partitions[partitionIdentifier];
4,036✔
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
        this.emit('preRead', position, partition.metadata);
3,972✔
224
        const data = partition.readFrom(position, size);
3,960✔
225
        return this.serializer.deserialize(data);
3,960✔
226
    }
227

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

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

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

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

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

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

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

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

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

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

320
    /**
321
     * Open an existing index.
322
     *
323
     * @api
324
     * @param {string} name The index name.
325
     * @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.
326
     * @returns {ReadableIndex}
327
     * @throws {Error} if the index with that name does not exist.
328
     * @throws {Error} if the HMAC for the matcher does not match.
329
     */
330
    openIndex(name, matcher) {
331
        if (name === '_all') {
741✔
332
            return this.index;
8✔
333
        }
334
        if (name in this.secondaryIndexes) {
733✔
335
            return this.secondaryIndexes[name].index;
629✔
336
        }
337

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

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

344
        index.open();
88✔
345
        return index;
88✔
346
    }
347

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

360
        const entries = this.index.all();
836✔
361

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

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

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

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

400
        for (let partition of Object.keys(this.partitions)) {
1,472✔
401
            iterationHandler(this.partitions[partition]);
1,440✔
402
        }
403
    }
404

405
}
406

407
module.exports = ReadableStorage;
4✔
408
module.exports.matches = matches;
4✔
409
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