From 85120e2487fda5f2ec5388a9ccffcc92c60535b9 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 20:35:18 +0000 Subject: [PATCH 1/3] Fix functional bugs and add batch ops, scores, iteration, serializers and retention Bug fixes: - ItemRepository: replace blocking KEYS with SCAN (fanned out to all masters on Redis Cluster), avoid CROSSSLOT MGET/DEL on Cluster, match repository names literally (glob escaping), stop colliding with the index key of a SortedItemRepository with the same name, stable ordered pagination - SortedItemRepository: atomic MULTI writes/deletes (ordered, safe writes on Cluster), drop redundant hexists/zrem round trips, wall-clock monotonic scores instead of process.hrtime (meaningless across processes, lost precision) - Both: real total count on out-of-range pages, read methods no longer return hidden nulls, validate page/pageSize/n/TTL/id instead of silently selecting (or deleting) the whole repository - build script no longer fails without an existing dist/ Features: - Batch operations: setMany, getByIds, deleteByIds - Sorted repository scoring and ordering: explicit score, keepScore, desc pagination, infinite bounds plus offset/limit/order on getItemsByScore - Streaming via iterate() async iterators - Constructor options: pluggable serializer, Cluster hashTag for atomic transactions on Redis Cluster - TTL and retention: getTtl/expire/persist, deleteByScore, trimToSize, maxItems Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JLbcuU2UWtjZCuWZa1QTkE --- README.md | 191 +++++++--------- libs/model.ts | 164 +++++++++++++- libs/repository.ts | 213 ++++++++++++++--- libs/sorted-repository.ts | 277 +++++++++++++++++------ libs/utils.ts | 84 +++++++ package.json | 2 +- test/2-repository.test.ts | 3 +- test/3-sortedrepository.test.ts | 3 +- test/4-repository-features.test.ts | 155 +++++++++++++ test/5-sortedrepository-features.test.ts | 216 ++++++++++++++++++ 10 files changed, 1079 insertions(+), 229 deletions(-) create mode 100644 libs/utils.ts create mode 100644 test/4-repository-features.test.ts create mode 100644 test/5-sortedrepository-features.test.ts diff --git a/README.md b/README.md index 5ac0a0a..bc06bb7 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ This module provides classes that implement repositories for storing and retrieving data in a Redis database: - `ItemRepository`: The class provides basic functionality for storing and retrieving individual items, as well as querying all items or a subset of items in a paginated manner. The items are stored in Redis using a key-value data model with expiration support. -- `SortedItemRepository`: The class provides a more advanced functionality for storing and retrieving items in a sorted manner. The items are sorted based on a numeric score, and can be queried in a paginated manner as well. The items are stored in Redis using a sorted set data model, making it more performant and efficient at scale (see `demos/benchmark.ts`). However, this model does not supports automatic expiration. +- `SortedItemRepository`: The class provides a more advanced functionality for storing and retrieving items in a sorted manner. The items are sorted based on a numeric score, and can be queried in a paginated manner as well. The items are stored in Redis using a sorted set data model, making it more performant and efficient at scale (see `demos/benchmark.ts`). Per-item expiration is not supported, but size and age based retention is available through `maxItems`, `trimToSize` and `deleteByScore`. ```ts import { ItemRepository, SortedItemRepository, IItem } from 'item-store-redis' @@ -22,7 +22,11 @@ import { ItemRepository, SortedItemRepository, IItem } from 'item-store-redis' ### Constructor ```ts -constructor (name: string, redis?: Redis | Cluster) +constructor (name: string, redis?: Redis | Cluster, options?: IRepositoryOptions) + +interface IRepositoryOptions { + serializer?: ISerializer // defaults to MessagePack +} ``` ### Usage @@ -32,7 +36,14 @@ import Redis from 'ioredis' const redis = new Redis({ host: 'localhost', port: 6379 }) const repository = new ItemRepository('my-unsorted-repo', redis) -... + +await repository.set({ id: 'a', data }, 60) // expires in 60 seconds +await repository.setMany([{ id: 'b', data }, { id: 'c', data }]) +const [a, missing] = await repository.getByIds(['a', 'missing']) // missing === null + +for await (const item of repository.iterate({ batchSize: 500 })) { + // streams the whole repository without loading it in memory +} ``` ### Interface @@ -41,21 +52,36 @@ interface IItemRepository { name: string redis: Redis | Cluster set: (item: IItem, expirationInSeconds?: number) => Promise + setMany: (items: IItem[], expirationInSeconds?: number) => Promise getById: (id: string) => Promise | null> + getByIds: (ids: string[]) => Promise<(IItem | null)[]> getAll: () => Promise[]> getPaginated: (page: number, pageSize: number) => Promise> + iterate: (options?: { batchSize?: number }) => AsyncIterableIterator> hasItem: (id: string) => Promise deleteById: (id: string) => Promise + deleteByIds: (ids: string[]) => Promise deleteAll: () => Promise count: () => Promise + getTtl: (id: string) => Promise // -1: no expiration, null: not found + expire: (id: string, expirationInSeconds: number) => Promise + persist: (id: string) => Promise } ``` +Keys are discovered with `SCAN` (never `KEYS`), on every master node when using Redis Cluster. Pages are ordered by item key, so they are stable across calls. + ## Class SortedItemRepository ### Constructor ```ts -constructor (name: string, redis?: Redis | Cluster) +constructor (name: string, redis?: Redis | Cluster, options?: ISortedRepositoryOptions) + +interface ISortedRepositoryOptions { + serializer?: ISerializer // defaults to MessagePack + maxItems?: number // caps the repository size, the lowest scored (oldest) items are removed + hashTag?: boolean // see "Redis Cluster" below +} ``` ### Usage @@ -65,136 +91,73 @@ import Redis from 'ioredis' const redis = new Redis({ host: 'localhost', port: 6379 }) const repository = new SortedItemRepository('my-sorted-repo', redis) -... + +await repository.set({ id: 'a', data }) // scored by insertion time (microseconds since epoch) +await repository.set({ id: 'b', data }, { score: 42 }) // explicit score +await repository.set({ id: 'a', data: newData }, { keepScore: true }) // update without moving the item + +const latest = await repository.getPaginated(1, 10, 'desc') +const lastHour = await repository.getItemsByScore((Date.now() - 3_600_000) * 1000, '+inf', { limit: 100 }) +await repository.deleteByScore('-inf', (Date.now() - 86_400_000) * 1000) // retention: drop items older than a day ``` ### Interface ```ts -/** - * ISortedItemRepository is an interface that defines the methods for interacting with a - * sorted item repository that uses Redis as the storage system. - */ -export interface ISortedItemRepository { +interface ISortedItemRepository { name: string redis: Redis | Cluster - - /** - * Sets an item in the repository. - * @param item - The item to be added. - */ - set: (item: IItem) => Promise - - /** - * Retrieves an item from the repository by its ID. - * @param id - The ID of the item to retrieve. - * @returns The item if found, null otherwise. - */ + set: (item: IItem, options?: { score?: number; keepScore?: boolean }) => Promise + setMany: (items: IItem[], options?: { score?: number; keepScore?: boolean }) => Promise getById: (id: string) => Promise | null> - - /** - * Retrieves all items from the repository. - * @returns An array of all items in the repository. - */ + getByIds: (ids: string[]) => Promise<(IItem | null)[]> getAll: () => Promise[]> - - /** - * Retrieves a paginated set of items from the repository. - * @param page - The page number to retrieve. - * @param pageSize - The number of items per page. - * @returns An object containing the paginated items and the total count of items. - */ - getPaginated: (page: number, pageSize: number) => Promise> - - /** - * Retrieves the score of an item by its ID. - * @param id - The ID of the item. - * @returns The score of the item if found, null otherwise. - */ + getPaginated: (page: number, pageSize: number, order?: 'asc' | 'desc') => Promise> + iterate: (options?: { batchSize?: number; order?: 'asc' | 'desc' }) => AsyncIterableIterator> getItemScoreById: (id: string) => Promise - - /** - * Retrieves items by their scores within a specified range. - * @param min - The minimum score value. - * @param max - The maximum score value. - * @returns An array of items with scores within the specified range. - */ - getItemsByScore: (min: number, max: number) => Promise[]> - - /** - * Deletes a page of items from the repository. - * @param page - The page number of items to delete. - * @param pageSize - The number of items per page. - */ - deletePage: (page: number, pageSize: number) => Promise - - /** - * Checks if an item exists in the repository by its ID. - * @param id - The ID of the item. - * @returns A boolean indicating whether the item exists. - */ + getItemsByScore: (min: ScoreBound, max: ScoreBound, options?: IScoreRangeOptions) => Promise[]> + getFirstNItems: (n: number) => Promise[]> + getLastNItems: (n: number) => Promise[]> + getItemsInRange: (start: number, end: number) => Promise[]> + getNextNItemsGreaterThanScore: (score: number, n: number) => Promise[]> + existsInRange: (min: ScoreBound, max: ScoreBound) => Promise hasItem: (id: string) => Promise - - /** - * Deletes an item from the repository by its ID. - * @param id - The ID of the item to delete. - */ + count: () => Promise deleteById: (id: string) => Promise - - /** - * Deletes all items from the repository. - */ + deleteByIds: (ids: string[]) => Promise + deletePage: (page: number, pageSize: number) => Promise + deleteByScore: (min: ScoreBound, max: ScoreBound) => Promise + trimToSize: (maxItems: number) => Promise deleteAll: () => Promise +} - /** - * Retrieves the count of items in the repository. - * @returns The count of items. - */ - count: () => Promise +type ScoreBound = number | '-inf' | '+inf' - /** - * Retrieves the first N items from the repository. - * @param n - The number of items to retrieve. - * @returns An array of the first N items. - */ - getFirstNItems: (n: number) => Promise[]> +interface IScoreRangeOptions { + offset?: number + limit?: number + order?: 'asc' | 'desc' +} +``` - /** - * Retrieves the last N items from the repository. - * @param n - The number of items to retrieve. - * @returns An array of the last N items. - */ - getLastNItems: (n: number) => Promise[]> +See `libs/model.ts` for the documented interfaces. - /** - * Retrieves items within a specified index range. - * @param start - The starting index. - * @param end - The ending index. - * - * @returns An array of items within the specified index range. - */ - getItemsInRange: (start: number, end: number) => Promise[]> +## Validation +Pages start at `1`; `page`, `pageSize`, expirations and `n` arguments must be integers in range, otherwise a `RangeError` is thrown (previously, values like `pageSize = 0` or `n = 0` silently selected the whole repository). Item ids must be non-empty strings. - /** - * - * Checks if any items exist within a specified score range. - * @param min - The minimum score value. - * @param max - The maximum score value. - * @returns A boolean indicating whether any items exist within the specified score range. - */ - existsInRange: (min: number, max: number) => Promise - - /** - * - * Retrieves the next N items with scores greater than a specified score. - * @param score - The score to compare. - * @param n - The number of items to retrieve. - * @returns An array of the next N items with scores greater than the specified score. - */ - getNextNItemsGreaterThanScore: (score: number, n: number) => Promise[]> -} +## Redis Cluster +Both repositories support `Cluster` connections. `SortedItemRepository` stores its index and data under two keys. On standalone Redis, writes and deletes touching both keys run as a `MULTI` transaction. On Redis Cluster those keys live in different slots, so the commands run in a safe order instead (data before index on writes, index before data on deletes) and readers never observe indexed items without data. Set `hashTag: true` to keep both keys in the same slot and get atomic transactions on Cluster as well. Note that this changes the key layout (`items:{name}:`), so existing data stored without the option is not visible with it. +## Custom serialization +```ts +const repository = new ItemRepository('json-repo', redis, { + serializer: { + encode: (item) => Buffer.from(JSON.stringify(item)), + decode: (buffer) => JSON.parse(buffer.toString()) + } +}) ``` + # License ``` diff --git a/libs/model.ts b/libs/model.ts index 83ef55c..e44e38e 100644 --- a/libs/model.ts +++ b/libs/model.ts @@ -10,6 +10,83 @@ export interface IPaginatedItems { count: number } +/** + * ISerializer converts items to and from the binary payload stored in Redis. + * The default serializer uses MessagePack. + */ +export interface ISerializer { + encode: (item: IItem) => Buffer | Uint8Array + decode: (buffer: Buffer) => IItem +} + +export interface IRepositoryOptions { + /** + * Custom serializer used to encode/decode items. Defaults to MessagePack. + */ + serializer?: ISerializer +} + +export interface ISortedRepositoryOptions extends IRepositoryOptions { + /** + * Wraps the repository name in a Redis Cluster hash tag (`items:{name}:`), so the index and the data keys + * live in the same slot and multi-key writes run as atomic transactions on Redis Cluster. + * Changes the key layout: data stored without this option is not visible with it, and vice versa. + */ + hashTag?: boolean + + /** + * Maximum number of items to keep. When exceeded after a write, the lowest scored (oldest) items are removed. + */ + maxItems?: number +} + +export interface ISortedSetOptions { + /** + * Explicit score for the item. Defaults to a wall-clock based, monotonically increasing score. + */ + score?: number + + /** + * When true and no explicit score is given, an existing item keeps its current score (and position) on update. + */ + keepScore?: boolean +} + +export type SortOrder = 'asc' | 'desc' + +export type ScoreBound = number | '-inf' | '+inf' + +export interface IScoreRangeOptions { + /** + * Number of matching items to skip. + */ + offset?: number + + /** + * Maximum number of items to return. + */ + limit?: number + + /** + * Sort order of the result, ascending by default. + */ + order?: SortOrder +} + +export interface IIterateOptions { + /** + * Number of items fetched from Redis per round trip. Defaults to 100. + */ + batchSize?: number +} + +export interface ISortedIterateOptions extends IIterateOptions { + /** + * Iteration order, ascending by default. + */ + order?: SortOrder +} + export interface IItemRepository { name: string redis: Redis | Cluster @@ -21,6 +98,47 @@ export interface IItemRepository { deleteById: (id: string) => Promise deleteAll: () => Promise count: () => Promise + + /** + * Sets multiple items in one round trip. + * @param items - The items to store. + * @param expirationInSeconds - Optional expiration applied to every item. + */ + setMany: (items: IItem[], expirationInSeconds?: number) => Promise + + /** + * Retrieves multiple items by ID. The result is aligned with the input: missing items are null. + */ + getByIds: (ids: string[]) => Promise<(IItem | null)[]> + + /** + * Deletes multiple items by ID. + * @returns The number of items deleted. + */ + deleteByIds: (ids: string[]) => Promise + + /** + * Iterates over all items in batches without loading the whole repository in memory. Order is not guaranteed. + */ + iterate: (options?: IIterateOptions) => AsyncIterableIterator> + + /** + * Retrieves the remaining time to live of an item. + * @returns Seconds to live, -1 if the item does not expire, null if the item does not exist. + */ + getTtl: (id: string) => Promise + + /** + * Sets an expiration on an existing item. + * @returns true if the item exists and the expiration was set. + */ + expire: (id: string, expirationInSeconds: number) => Promise + + /** + * Removes the expiration of an item. + * @returns true if the expiration was removed. + */ + persist: (id: string) => Promise } /** @@ -34,8 +152,9 @@ export interface ISortedItemRepository { /** * Sets an item in the repository. * @param item - The item to be added. + * @param options - Optional explicit score, or keepScore to preserve the position of an existing item. */ - set: (item: IItem) => Promise + set: (item: IItem, options?: ISortedSetOptions) => Promise /** * Retrieves an item from the repository by its ID. @@ -52,11 +171,12 @@ export interface ISortedItemRepository { /** * Retrieves a paginated set of items from the repository. - * @param page - The page number to retrieve. + * @param page - The page number to retrieve (starting at 1). * @param pageSize - The number of items per page. + * @param order - Sort order, ascending by default. * @returns An object containing the paginated items and the total count of items. */ - getPaginated: (page: number, pageSize: number) => Promise> + getPaginated: (page: number, pageSize: number, order?: SortOrder) => Promise> /** * Retrieves the score of an item by its ID. @@ -69,9 +189,10 @@ export interface ISortedItemRepository { * Retrieves items by their scores within a specified range. * @param min - The minimum score value. * @param max - The maximum score value. + * @param options - Optional offset, limit and order. * @returns An array of items with scores within the specified range. */ - getItemsByScore: (min: number, max: number) => Promise[]> + getItemsByScore: (min: ScoreBound, max: ScoreBound, options?: IScoreRangeOptions) => Promise[]> /** * Deletes a page of items from the repository. @@ -134,7 +255,7 @@ export interface ISortedItemRepository { * @param max - The maximum score value. * @returns A boolean indicating whether any items exist within the specified score range. */ - existsInRange: (min: number, max: number) => Promise + existsInRange: (min: ScoreBound, max: ScoreBound) => Promise /** * @@ -144,4 +265,37 @@ export interface ISortedItemRepository { * @returns An array of the next N items with scores greater than the specified score. */ getNextNItemsGreaterThanScore: (score: number, n: number) => Promise[]> + + /** + * Sets multiple items in one round trip. Items without an explicit score keep the input order. + */ + setMany: (items: IItem[], options?: ISortedSetOptions) => Promise + + /** + * Retrieves multiple items by ID. The result is aligned with the input: missing items are null. + */ + getByIds: (ids: string[]) => Promise<(IItem | null)[]> + + /** + * Deletes multiple items by ID. + * @returns The number of items deleted. + */ + deleteByIds: (ids: string[]) => Promise + + /** + * Iterates over all items in score order, in batches, without loading the whole repository in memory. + */ + iterate: (options?: ISortedIterateOptions) => AsyncIterableIterator> + + /** + * Deletes all items with scores within the specified range. + * @returns The number of items deleted. + */ + deleteByScore: (min: ScoreBound, max: ScoreBound) => Promise + + /** + * Removes the lowest scored (oldest) items so that at most maxItems remain. + * @returns The number of items deleted. + */ + trimToSize: (maxItems: number) => Promise } diff --git a/libs/repository.ts b/libs/repository.ts index 33a9807..beb8807 100644 --- a/libs/repository.ts +++ b/libs/repository.ts @@ -1,6 +1,20 @@ import Redis, { Cluster } from 'ioredis' -import { IItem, IItemRepository, IPaginatedItems } from './model' -import { encode, decode } from '@msgpack/msgpack' +import { IItem, IItemRepository, IIterateOptions, IPaginatedItems, IRepositoryOptions, ISerializer } from './model' +import { decode } from '@msgpack/msgpack' +import { + assertId, + assertPositiveInteger, + chunk, + createMsgpackSerializer, + escapeGlob, + isCluster, + notNull, + pageBounds, + toBuffer +} from './utils' + +const DEFAULT_BATCH_SIZE = 100 +const DELETE_CHUNK_SIZE = 500 export function bufferToItem(buffer: Buffer | null): IItem | null { if (!buffer) return null @@ -11,86 +25,221 @@ export function bufferToItem(buffer: Buffer | null): IItem | null { export class ItemRepository implements IItemRepository { private readonly keyPrefix: string - - constructor(public readonly name: string, public readonly redis: Redis | Cluster = new Redis()) { + private readonly matchPattern: string + private readonly serializer: ISerializer + + constructor( + public readonly name: string, + public readonly redis: Redis | Cluster = new Redis(), + options: IRepositoryOptions = {} + ) { this.keyPrefix = `items:${name}:` + this.matchPattern = `${escapeGlob(this.keyPrefix)}*` + this.serializer = options.serializer ?? createMsgpackSerializer() } async set(item: IItem, expirationInSeconds?: number): Promise { + assertId(item.id) const key = this.getKey(item.id) - const buffer = encode(item) + const buffer = toBuffer(this.serializer.encode(item)) if (expirationInSeconds != null) { - await this.redis.setex(key, expirationInSeconds, Buffer.from(buffer)) + assertPositiveInteger(expirationInSeconds, 'expirationInSeconds') + await this.redis.set(key, buffer, 'EX', expirationInSeconds) } else { - await this.redis.set(key, Buffer.from(buffer)) + await this.redis.set(key, buffer) } } + async setMany(items: IItem[], expirationInSeconds?: number): Promise { + if (items.length === 0) return + items.forEach((item) => assertId(item.id)) + if (expirationInSeconds != null) { + assertPositiveInteger(expirationInSeconds, 'expirationInSeconds') + } + + const entries = items.map((item) => [this.getKey(item.id), toBuffer(this.serializer.encode(item))] as const) + if (isCluster(this.redis)) { + // keys of a repository are spread across slots, so they are written individually + await Promise.all( + entries.map(([key, buffer]) => + expirationInSeconds != null + ? this.redis.set(key, buffer, 'EX', expirationInSeconds) + : this.redis.set(key, buffer) + ) + ) + return + } + + const tx = this.redis.multi() + for (const [key, buffer] of entries) { + if (expirationInSeconds != null) { + tx.set(key, buffer, 'EX', expirationInSeconds) + } else { + tx.set(key, buffer) + } + } + const results = await tx.exec() + const failure = results?.find(([err]) => err) + if (failure) throw failure[0] + } + async getById(id: string): Promise | null> { + if (!id) return null const key = this.getKey(id) const result = await this.redis.getBuffer(key) - return bufferToItem(result) as IItem + return this.decode(result) + } + + async getByIds(ids: string[]): Promise<(IItem | null)[]> { + const valid = ids.filter((id) => id) + const items = await this.fetch(valid.map((id) => this.getKey(id))) + const byId = new Map(valid.map((id, i) => [id, items[i]])) + + return ids.map((id) => byId.get(id) ?? null) } async getAll(): Promise[]> { - const keys = await this.redis.keys(this.getKey('*')) - if (keys.length === 0) { - return [] - } - const buffers = await this.redis.mgetBuffer(keys) + const keys = await this.scanKeys() + const items = await this.fetch(keys) - return buffers.map((b) => bufferToItem(b) as IItem) + return items.filter(notNull) } async getPaginated(page: number, pageSize: number): Promise> { - const keys = await this.redis.keys(this.getKey('*')) + const [start, end] = pageBounds(page, pageSize) + // sorting keeps pages stable across calls, SCAN order is arbitrary + const keys = (await this.scanKeys()).sort() const count = keys.length - const start = (page - 1) * pageSize - const end = start + pageSize - 1 - const itemKeys = keys.slice(start, end + 1) - if (itemKeys.length === 0) { - return { - items: [], - count: 0 - } - } - - const buffers = await this.redis.mgetBuffer(itemKeys) - const items = buffers.map((b) => bufferToItem(b) as IItem) as IItem[] + const items = await this.fetch(keys.slice(start, end + 1)) return { - items, + items: items.filter(notNull), count } } + async *iterate(options: IIterateOptions = {}): AsyncIterableIterator> { + const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE + assertPositiveInteger(batchSize, 'batchSize') + + // SCAN may return a key more than once + const seen = new Set() + for await (const batch of this.scanKeyBatches(batchSize)) { + const keys = batch.filter((key) => !seen.has(key)) + keys.forEach((key) => seen.add(key)) + for (const item of await this.fetch(keys)) { + if (item !== null) yield item + } + } + } + async hasItem(id: string): Promise { + if (!id) return false const key = this.getKey(id) return (await this.redis.exists(key)) === 1 } async deleteById(id: string): Promise { + // an empty id would address the index key of a SortedItemRepository with the same name + if (!id) return const key = this.getKey(id) await this.redis.del(key) } + async deleteByIds(ids: string[]): Promise { + return this.deleteKeys([...new Set(ids.filter((id) => id))].map((id) => this.getKey(id))) + } + async deleteAll(): Promise { - const keys = await this.redis.keys(this.getKey('*')) - if (keys.length > 0) { - await this.redis.del(keys) + for await (const keys of this.scanKeyBatches(DELETE_CHUNK_SIZE)) { + await this.deleteKeys(keys) } } async count(): Promise { - const keys = await this.redis.keys(this.getKey('*')) + const keys = await this.scanKeys() return keys.length } + async getTtl(id: string): Promise { + if (!id) return null + const ttl = await this.redis.ttl(this.getKey(id)) + + return ttl === -2 ? null : ttl + } + + async expire(id: string, expirationInSeconds: number): Promise { + assertId(id) + assertPositiveInteger(expirationInSeconds, 'expirationInSeconds') + + return (await this.redis.expire(this.getKey(id), expirationInSeconds)) === 1 + } + + async persist(id: string): Promise { + assertId(id) + return (await this.redis.persist(this.getKey(id))) === 1 + } + private getKey(id: string): string { return `${this.keyPrefix}${id}` } + + private decode(buffer: Buffer | null): IItem | null { + return buffer ? this.serializer.decode(buffer) : null + } + + private async fetch(keys: string[]): Promise<(IItem | null)[]> { + if (keys.length === 0) return [] + + // MGET across slots fails with CROSSSLOT on Redis Cluster + const buffers = isCluster(this.redis) + ? await Promise.all(keys.map((key) => this.redis.getBuffer(key))) + : await this.redis.mgetBuffer(keys) + + return buffers.map((b) => this.decode(b)) + } + + private async deleteKeys(keys: string[]): Promise { + if (keys.length === 0) return 0 + + if (isCluster(this.redis)) { + const results = await Promise.all(keys.map((key) => this.redis.del(key))) + return results.reduce((sum, n) => sum + n, 0) + } + + let deleted = 0 + for (const batch of chunk(keys, DELETE_CHUNK_SIZE)) { + deleted += await this.redis.del(batch) + } + + return deleted + } + + private async scanKeys(): Promise { + const keys = new Set() + for await (const batch of this.scanKeyBatches(DEFAULT_BATCH_SIZE * 10)) { + batch.forEach((key) => keys.add(key)) + } + + return [...keys] + } + + private async *scanKeyBatches(count: number): AsyncGenerator { + // on Redis Cluster, every master holds a subset of the keys + const nodes = isCluster(this.redis) ? (this.redis as Cluster).nodes('master') : [this.redis as Redis] + for (const node of nodes) { + let cursor = '0' + do { + const [next, keys] = await node.scan(cursor, 'MATCH', this.matchPattern, 'COUNT', count) + cursor = next + // the bare prefix is the index key of a SortedItemRepository with the same name + const own = keys.filter((key) => key !== this.keyPrefix) + if (own.length > 0) yield own + } while (cursor !== '0') + } + } } export default ItemRepository diff --git a/libs/sorted-repository.ts b/libs/sorted-repository.ts index ede675c..f5ae924 100644 --- a/libs/sorted-repository.ts +++ b/libs/sorted-repository.ts @@ -1,100 +1,196 @@ -import { IItem, IPaginatedItems, ISortedItemRepository } from './model' +import { + IItem, + IPaginatedItems, + IScoreRangeOptions, + ISerializer, + ISortedItemRepository, + ISortedIterateOptions, + ISortedRepositoryOptions, + ISortedSetOptions, + ScoreBound, + SortOrder +} from './model' import { Redis, Cluster } from 'ioredis' -import { encode } from '@msgpack/msgpack' -import { bufferToItem } from './repository' +import { + assertId, + assertInteger, + assertNonNegativeInteger, + assertPositiveInteger, + createMsgpackSerializer, + isCluster, + nextScore, + notNull, + pageBounds, + toBuffer +} from './utils' + +type Command = [string, ...(string | number | Buffer)[]] + +const DEFAULT_BATCH_SIZE = 100 export class SortedItemRepository implements ISortedItemRepository { private readonly keyPrefix: string private readonly hashPrefix: string + private readonly serializer: ISerializer + private readonly maxItems?: number + private readonly atomic: boolean + + constructor( + public readonly name: string, + public readonly redis: Redis | Cluster = new Redis(), + options: ISortedRepositoryOptions = {} + ) { + const namespace = options.hashTag ? `{${name}}` : name + this.keyPrefix = `items:${namespace}:` + this.hashPrefix = `items-data:${namespace}:` + this.serializer = options.serializer ?? createMsgpackSerializer() + if (options.maxItems != null) { + assertPositiveInteger(options.maxItems, 'maxItems') + this.maxItems = options.maxItems + } + // without a hash tag, the index and data keys live in different cluster slots and cannot share a transaction + this.atomic = !isCluster(redis) || options.hashTag === true + } - constructor(public readonly name: string, public readonly redis: Redis | Cluster = new Redis()) { - this.keyPrefix = `items:${name}:` - this.hashPrefix = `items-data:${name}:` + async set(item: IItem, options: ISortedSetOptions = {}): Promise { + await this.setMany([item], options) } - async set(item: IItem): Promise { - if (await this.hasItem(item.id)) { - await this.redis.zrem(this.keyPrefix, item.id) + async setMany(items: IItem[], options: ISortedSetOptions = {}): Promise { + if (items.length === 0) return + items.forEach((item) => assertId(item.id)) + if (options.score != null && !Number.isFinite(options.score)) { + throw new RangeError('score must be a finite number') } - const score = Number(process.hrtime.bigint()) - const buffer = encode(item) - await Promise.all([ - await this.redis.hset(this.hashPrefix, item.id, Buffer.from(buffer)), - await this.redis.zadd(this.keyPrefix, score, item.id) + const keepScore = options.keepScore === true && options.score == null + const fields = items.flatMap((item) => [item.id, toBuffer(this.serializer.encode(item))]) + const members = items.flatMap((item) => [options.score ?? nextScore(), item.id]) + + // data is written before the index, so readers never find an indexed item without data + await this.run([ + ['hset', this.hashPrefix, ...fields], + ['zadd', this.keyPrefix, ...(keepScore ? ['NX'] : []), ...members] ]) + + if (this.maxItems != null) { + await this.trimToSize(this.maxItems) + } } async getById(id: string): Promise | null> { const payload = await this.redis.hgetBuffer(this.hashPrefix, id) - return bufferToItem(payload) + return this.decode(payload) + } + + async getByIds(ids: string[]): Promise<(IItem | null)[]> { + if (ids.length === 0) return [] + const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...ids) + + return buffers.map((b) => this.decode(b)) } async getAll(): Promise[]> { - const keys = await this.redis.zrange(this.keyPrefix, 0, -1) - if (keys.length === 0) { - return [] - } - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) + const ids = await this.redis.zrange(this.keyPrefix, 0, -1) - return buffers.map((b) => bufferToItem(b) as IItem) + return this.fetch(ids) } async getItemScoreById(id: string): Promise { const score = await this.redis.zscore(this.keyPrefix, id) - return score ? Number(score) : null + return score === null ? null : Number(score) } - async getItemsByScore(min: number, max: number): Promise[]> { - const keys = await this.redis.zrangebyscore(this.keyPrefix, min, max) - if (keys.length === 0) { - return [] + async getItemsByScore(min: ScoreBound, max: ScoreBound, options: IScoreRangeOptions = {}): Promise[]> { + const { offset = 0, limit, order = 'asc' } = options + assertNonNegativeInteger(offset, 'offset') + if (limit != null) { + assertNonNegativeInteger(limit, 'limit') + if (limit === 0) return [] } - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) - const items = buffers.map((b) => bufferToItem(b) as IItem) + const limitArgs = ['LIMIT', offset, limit ?? -1] as const + const ids = + order === 'desc' + ? await this.redis.zrevrangebyscore(this.keyPrefix, max, min, ...limitArgs) + : await this.redis.zrangebyscore(this.keyPrefix, min, max, ...limitArgs) - return items + return this.fetch(ids) } - async getPaginated(page: number, pageSize: number): Promise> { - const count = await this.redis.zcard(this.keyPrefix) - const start = (page - 1) * pageSize - const end = start + pageSize - 1 - const keys = await this.redis.zrange(this.keyPrefix, start, end) - if (keys.length === 0) { - return { - items: [], - count: 0 - } + async getPaginated(page: number, pageSize: number, order: SortOrder = 'asc'): Promise> { + const [start, end] = pageBounds(page, pageSize) + // count and range are read in one transaction, so they are consistent + const tx = this.redis.multi().zcard(this.keyPrefix) + if (order === 'desc') { + tx.zrevrange(this.keyPrefix, start, end) + } else { + tx.zrange(this.keyPrefix, start, end) } - - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) - const items = buffers.map((b) => bufferToItem(b) as IItem) + const [count, ids] = (await this.exec(tx)) as [number, string[]] return { - items, + items: await this.fetch(ids), count } } - async deletePage(page: number, pageSize: number): Promise { - const start = (page - 1) * pageSize - const end = start + pageSize - 1 - const keys = await this.redis.zrange(this.keyPrefix, start, end) - if (keys.length > 0) { - await Promise.all([this.redis.zrem(this.keyPrefix, ...keys), this.redis.hdel(this.hashPrefix, ...keys)]) + async *iterate(options: ISortedIterateOptions = {}): AsyncIterableIterator> { + const { batchSize = DEFAULT_BATCH_SIZE, order = 'asc' } = options + assertPositiveInteger(batchSize, 'batchSize') + + for (let start = 0; ; start += batchSize) { + const end = start + batchSize - 1 + const ids = + order === 'desc' + ? await this.redis.zrevrange(this.keyPrefix, start, end) + : await this.redis.zrange(this.keyPrefix, start, end) + for (const item of await this.fetch(ids)) { + yield item + } + if (ids.length < batchSize) return } } + async deletePage(page: number, pageSize: number): Promise { + const [start, end] = pageBounds(page, pageSize) + const ids = await this.redis.zrange(this.keyPrefix, start, end) + await this.deleteByIds(ids) + } + async hasItem(id: string): Promise { return (await this.redis.hexists(this.hashPrefix, id)) === 1 } async deleteById(id: string): Promise { - await Promise.all([this.redis.zrem(this.keyPrefix, id), this.redis.hdel(this.hashPrefix, id)]) + await this.deleteByIds([id]) + } + + async deleteByIds(ids: string[]): Promise { + if (ids.length === 0) return 0 + + // the index is cleaned before the data, so readers never find an indexed item without data + const [removed, deleted] = (await this.run([ + ['zrem', this.keyPrefix, ...ids], + ['hdel', this.hashPrefix, ...ids] + ])) as [number, number] + + return Math.max(removed, deleted) + } + + async deleteByScore(min: ScoreBound, max: ScoreBound): Promise { + const ids = await this.redis.zrangebyscore(this.keyPrefix, min, max) + + return this.deleteByIds(ids) + } + + async trimToSize(maxItems: number): Promise { + assertNonNegativeInteger(maxItems, 'maxItems') + const ids = await this.redis.zrange(this.keyPrefix, 0, -(maxItems + 1)) + + return this.deleteByIds(ids) } async count(): Promise { @@ -104,53 +200,84 @@ export class SortedItemRepository implements ISortedItemRepository { } async deleteAll(): Promise { - await Promise.all([await this.redis.del(this.hashPrefix), await this.redis.del(this.keyPrefix)]) + await this.run([ + ['del', this.keyPrefix], + ['del', this.hashPrefix] + ]) } async getFirstNItems(n: number): Promise[]> { - const keys = await this.redis.zrange(this.keyPrefix, 0, n - 1) - if (keys.length === 0) { - return [] - } - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) + assertNonNegativeInteger(n, 'n') + if (n === 0) return [] + const ids = await this.redis.zrange(this.keyPrefix, 0, n - 1) - return buffers.map((b) => bufferToItem(b) as IItem) + return this.fetch(ids) } async getLastNItems(n: number): Promise[]> { - const keys = await this.redis.zrevrange(this.keyPrefix, 0, n - 1) - if (keys.length === 0) { - return [] - } - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) + assertNonNegativeInteger(n, 'n') + if (n === 0) return [] + const ids = await this.redis.zrevrange(this.keyPrefix, 0, n - 1) - return buffers.map((b) => bufferToItem(b) as IItem).reverse() + return (await this.fetch(ids)).reverse() } async getItemsInRange(start: number, end: number): Promise[]> { - const keys = await this.redis.zrange(this.keyPrefix, start, end) - if (keys.length === 0) { - return [] - } - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) + assertInteger(start, 'start') + assertInteger(end, 'end') + const ids = await this.redis.zrange(this.keyPrefix, start, end) - return buffers.map((b) => bufferToItem(b) as IItem) + return this.fetch(ids) } - async existsInRange(min: number, max: number): Promise { + async existsInRange(min: ScoreBound, max: ScoreBound): Promise { const count = await this.redis.zcount(this.keyPrefix, min, max) return count > 0 } async getNextNItemsGreaterThanScore(score: number, n: number): Promise[]> { - const keys = await this.redis.zrangebyscore(this.keyPrefix, `(${score}`, '+inf', 'LIMIT', 0, n) - if (keys.length === 0) { - return [] + assertNonNegativeInteger(n, 'n') + if (n === 0) return [] + const ids = await this.redis.zrangebyscore(this.keyPrefix, `(${score}`, '+inf', 'LIMIT', 0, n) + + return this.fetch(ids) + } + + private decode(buffer: Buffer | null): IItem | null { + return buffer ? this.serializer.decode(buffer) : null + } + + // Loads the data of indexed ids, skipping ids whose data is missing. + private async fetch(ids: string[]): Promise[]> { + return (await this.getByIds(ids)).filter(notNull) + } + + /** + * Runs commands touching the index and data keys. They run as a MULTI transaction, unless the keys live in + * different Redis Cluster slots: then they run sequentially in the given order. + */ + private async run(commands: Command[]): Promise { + if (this.atomic) { + return this.exec(this.redis.multi(commands)) } - const buffers = await this.redis.hmgetBuffer(this.hashPrefix, ...keys) - return buffers.map((b) => bufferToItem(b) as IItem) + const results: unknown[] = [] + for (const [command, ...args] of commands) { + results.push(await this.redis.call(command, ...args)) + } + + return results + } + + private async exec(tx: ReturnType): Promise { + const results = await tx.exec() + if (!results) throw new Error('Redis transaction aborted') + + return results.map(([err, result]) => { + if (err) throw err + return result + }) } } diff --git a/libs/utils.ts b/libs/utils.ts new file mode 100644 index 0000000..a6733be --- /dev/null +++ b/libs/utils.ts @@ -0,0 +1,84 @@ +import { Redis, Cluster } from 'ioredis' +import { encode, decode } from '@msgpack/msgpack' +import { IItem, ISerializer } from './model' + +export function createMsgpackSerializer(): ISerializer { + return { + encode: (item: IItem) => { + const bytes = encode(item) + return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength) + }, + decode: (buffer: Buffer) => decode(buffer) as IItem + } +} + +export function toBuffer(value: Buffer | Uint8Array): Buffer { + return Buffer.isBuffer(value) ? value : Buffer.from(value.buffer, value.byteOffset, value.byteLength) +} + +export function isCluster(redis: Redis | Cluster): boolean { + return redis.isCluster === true +} + +// Escapes Redis glob special characters so a repository name is matched literally. +export function escapeGlob(value: string): string { + return value.replace(/[*?[\]\\]/g, '\\$&') +} + +let lastScore = 0 + +/** + * Returns a wall-clock based score (microseconds since epoch) that is strictly increasing within the process. + * Stays below Number.MAX_SAFE_INTEGER, so it is stored by Redis without precision loss. + */ +export function nextScore(): number { + const now = Date.now() * 1000 + lastScore = now > lastScore ? now : lastScore + 1 + + return lastScore +} + +export function assertId(id: unknown): void { + if (typeof id !== 'string' || id.length === 0) { + throw new TypeError('Item id must be a non-empty string') + } +} + +export function assertPositiveInteger(value: number, name: string): void { + if (!Number.isInteger(value) || value < 1) { + throw new RangeError(`${name} must be a positive integer`) + } +} + +export function assertNonNegativeInteger(value: number, name: string): void { + if (!Number.isInteger(value) || value < 0) { + throw new RangeError(`${name} must be a non-negative integer`) + } +} + +export function assertInteger(value: number, name: string): void { + if (!Number.isInteger(value)) { + throw new RangeError(`${name} must be an integer`) + } +} + +export function pageBounds(page: number, pageSize: number): [number, number] { + assertPositiveInteger(page, 'page') + assertPositiveInteger(pageSize, 'pageSize') + const start = (page - 1) * pageSize + + return [start, start + pageSize - 1] +} + +export function notNull(value: V | null): value is V { + return value !== null +} + +export function chunk(values: V[], size: number): V[][] { + const chunks: V[][] = [] + for (let i = 0; i < values.length; i += size) { + chunks.push(values.slice(i, i + size)) + } + + return chunks +} diff --git a/package.json b/package.json index 4e3fe48..d9d40d6 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "main": "dist/libs/index.js", "types": "dist/libs/index.d.ts", "scripts": { - "build": "rm -r dist/ && tsc", + "build": "rm -rf dist/ && tsc", "lint": "bun x rome ci .", "format": "bun x rome format . --write", "test": "bun test test/*.test.ts" diff --git a/test/2-repository.test.ts b/test/2-repository.test.ts index 0954a61..5bbbc76 100644 --- a/test/2-repository.test.ts +++ b/test/2-repository.test.ts @@ -14,7 +14,7 @@ describe('ItemRepository', () => { }) beforeEach(() => { - const name = `repo-${Date.now()}` + const name = `repo-${crypto.randomUUID()}` repository = new ItemRepository(name, redis) }) @@ -57,6 +57,7 @@ describe('ItemRepository', () => { } const result = await repository.getAll() + expect(result.length).toBe(items.length) for (const item of result) { expect(items.find((i) => i.id === item.id)).toEqual(item) } diff --git a/test/3-sortedrepository.test.ts b/test/3-sortedrepository.test.ts index 1c0af37..6fea787 100644 --- a/test/3-sortedrepository.test.ts +++ b/test/3-sortedrepository.test.ts @@ -14,7 +14,7 @@ describe('SortedItemRepository', () => { }) beforeEach(() => { - const name = `repo-${Date.now()}` + const name = `repo-${crypto.randomUUID()}` repository = new SortedItemRepository(name, redis) }) @@ -46,6 +46,7 @@ describe('SortedItemRepository', () => { } const result = await repository.getAll() + expect(result.length).toBe(items.length) for (const item of result) { expect(items.find((i) => i.id === item.id)).toEqual(item) } diff --git a/test/4-repository-features.test.ts b/test/4-repository-features.test.ts new file mode 100644 index 0000000..569b39d --- /dev/null +++ b/test/4-repository-features.test.ts @@ -0,0 +1,155 @@ +import { IItem, ItemRepository, SortedItemRepository, ISerializer } from '../libs' +import { describe, expect, test, beforeAll, beforeEach, afterAll, afterEach } from 'bun:test' +import Redis from 'ioredis' + +const toItems = (n: number): IItem[] => + Array.from({ length: n }, (_, i) => ({ id: `item-${String(i).padStart(3, '0')}`, data: `data-${i}` })) + +const expectRejection = async (promise: Promise, type: new (...args: never[]) => Error): Promise => { + let error: unknown + try { + await promise + } catch (err) { + error = err + } + expect(error instanceof type).toBe(true) +} + +describe('ItemRepository bug fixes and features', () => { + let repository: ItemRepository + let name: string + const redis: Redis = new Redis({ + host: '127.0.0.1', + port: 6379 + }) + + beforeAll(async () => { + await redis.flushall() + }) + + beforeEach(() => { + name = `repo-${crypto.randomUUID()}` + repository = new ItemRepository(name, redis) + }) + + afterEach(async () => { + await redis.flushall() + }) + + afterAll(() => { + redis.disconnect() + }) + + test('getPaginated returns the total count on an out of range page', async () => { + await repository.setMany(toItems(5)) + + const result = await repository.getPaginated(3, 5) + expect(result.items).toEqual([]) + expect(result.count).toBe(5) + }) + + test('getPaginated returns stable, non overlapping pages', async () => { + const items = toItems(25) + await repository.setMany(items) + + const seen: string[] = [] + for (let page = 1; page <= 3; page++) { + const { items: pageItems } = await repository.getPaginated(page, 10) + seen.push(...pageItems.map((i) => i.id)) + } + expect(seen).toEqual(items.map((i) => i.id)) + }) + + test('rejects invalid pagination arguments and expirations', async () => { + await expectRejection(repository.getPaginated(0, 10), RangeError) + await expectRejection(repository.getPaginated(1, 0), RangeError) + await expectRejection(repository.getPaginated(1.5, 10), RangeError) + await expectRejection(repository.set({ id: 'a', data: 'a' }, 0), RangeError) + await expectRejection(repository.set({ id: '', data: 'a' }), TypeError) + }) + + test('repository names are matched literally, not as glob patterns', async () => { + const other = new ItemRepository('other', redis) + await other.set({ id: 'x', data: 'x' }) + const wildcard = new ItemRepository('*', redis) + await wildcard.set({ id: 'y', data: 'y' }) + + expect(await wildcard.count()).toBe(1) + await wildcard.deleteAll() + expect(await other.getById('x')).toEqual({ id: 'x', data: 'x' }) + }) + + test('does not collide with a SortedItemRepository using the same name', async () => { + const sorted = new SortedItemRepository(name, redis) + await sorted.set({ id: 's', data: 's' }) + await repository.set({ id: 'a', data: 'a' }) + + expect(await repository.count()).toBe(1) + expect(await repository.getAll()).toEqual([{ id: 'a', data: 'a' }]) + expect(await repository.hasItem('')).toBe(false) + + await repository.deleteAll() + expect(await sorted.getAll()).toEqual([{ id: 's', data: 's' }]) + }) + + test('setMany, getByIds and deleteByIds', async () => { + const items = toItems(3) + await repository.setMany(items) + + expect(await repository.getByIds(['item-002', 'missing', 'item-000'])).toEqual([items[2], null, items[0]]) + expect(await repository.deleteByIds(['item-000', 'item-001', 'missing'])).toBe(2) + expect(await repository.getAll()).toEqual([items[2]]) + }) + + test('setMany applies the expiration to every item', async () => { + await repository.setMany(toItems(2), 100) + + expect(await repository.getTtl('item-000')).toBeGreaterThan(0) + expect(await repository.getTtl('item-001')).toBeGreaterThan(0) + }) + + test('iterate streams all items in batches', async () => { + const items = toItems(250) + await repository.setMany(items) + + const ids: string[] = [] + for await (const item of repository.iterate({ batchSize: 20 })) { + ids.push(item.id) + } + expect(ids.sort()).toEqual(items.map((i) => i.id)) + }) + + test('deleteAll removes large repositories', async () => { + await repository.setMany(toItems(1200)) + await repository.deleteAll() + + expect(await repository.count()).toBe(0) + }) + + test('getTtl, expire and persist', async () => { + await repository.set({ id: 'a', data: 'a' }) + + expect(await repository.getTtl('a')).toBe(-1) + expect(await repository.getTtl('missing')).toBeNull() + + expect(await repository.expire('a', 100)).toBe(true) + const ttl = (await repository.getTtl('a')) as number + expect(ttl > 0 && ttl <= 100).toBe(true) + + expect(await repository.persist('a')).toBe(true) + expect(await repository.getTtl('a')).toBe(-1) + expect(await repository.expire('missing', 100)).toBe(false) + }) + + test('supports a custom serializer', async () => { + const serializer: ISerializer = { + encode: (item) => Buffer.from(JSON.stringify(item)), + decode: (buffer) => JSON.parse(buffer.toString()) + } + const jsonRepository = new ItemRepository(name, redis, { serializer }) + await jsonRepository.set({ id: 'a', data: 'a' }) + + expect(JSON.parse((await redis.get(`items:${name}:a`)) as string)).toEqual({ id: 'a', data: 'a' }) + expect(await jsonRepository.getById('a')).toEqual({ id: 'a', data: 'a' }) + }) +}) diff --git a/test/5-sortedrepository-features.test.ts b/test/5-sortedrepository-features.test.ts new file mode 100644 index 0000000..2531e9f --- /dev/null +++ b/test/5-sortedrepository-features.test.ts @@ -0,0 +1,216 @@ +import { IItem, ISerializer, SortedItemRepository } from '../libs' +import { describe, expect, test, beforeAll, beforeEach, afterAll, afterEach } from 'bun:test' +import Redis from 'ioredis' + +const toItems = (n: number): IItem[] => + Array.from({ length: n }, (_, i) => ({ id: `item-${String(i).padStart(3, '0')}`, data: `data-${i}` })) + +const ids = (items: IItem[]): string[] => items.map((i) => i.id) + +const expectRejection = async (promise: Promise, type: new (...args: never[]) => Error): Promise => { + let error: unknown + try { + await promise + } catch (err) { + error = err + } + expect(error instanceof type).toBe(true) +} + +describe('SortedItemRepository bug fixes and features', () => { + let repository: SortedItemRepository + let name: string + const redis: Redis = new Redis({ + host: '127.0.0.1', + port: 6379 + }) + + beforeAll(async () => { + await redis.flushall() + }) + + beforeEach(() => { + name = `repo-${crypto.randomUUID()}` + repository = new SortedItemRepository(name, redis) + }) + + afterEach(async () => { + await redis.flushall() + }) + + afterAll(() => { + redis.disconnect() + }) + + test('default scores are wall-clock based and safe integers', async () => { + const before = Date.now() * 1000 + await repository.set({ id: 'a', data: 'a' }) + const score = (await repository.getItemScoreById('a')) as number + + expect(score).toBeGreaterThanOrEqual(before) + expect(score).toBeLessThanOrEqual(Date.now() * 1000 + 1000) + expect(Number.isSafeInteger(score)).toBe(true) + }) + + test('scores are ordered across repository instances', async () => { + const other = new SortedItemRepository(name, redis) + await repository.set({ id: 'z', data: 'z' }) + await other.set({ id: 'a', data: 'a' }) + + expect(ids(await repository.getAll())).toEqual(['z', 'a']) + }) + + test('getPaginated returns the total count on an out of range page', async () => { + await repository.setMany(toItems(5)) + + const result = await repository.getPaginated(2, 5) + expect(result.items).toEqual([]) + expect(result.count).toBe(5) + }) + + test('rejects invalid pagination and n arguments instead of returning everything', async () => { + await repository.setMany(toItems(5)) + + await expectRejection(repository.getPaginated(0, 2), RangeError) + await expectRejection(repository.getPaginated(1, 0), RangeError) + await expectRejection(repository.deletePage(1, 0), RangeError) + await expectRejection(repository.getFirstNItems(-1), RangeError) + await expectRejection(repository.getNextNItemsGreaterThanScore(0, -1), RangeError) + expect(await repository.getFirstNItems(0)).toEqual([]) + expect(await repository.getLastNItems(0)).toEqual([]) + expect(await repository.getNextNItemsGreaterThanScore(0, 0)).toEqual([]) + expect(await repository.count()).toBe(5) + }) + + test('read methods skip indexed ids without data', async () => { + await repository.setMany(toItems(3)) + await redis.hdel(`items-data:${name}:`, 'item-001') + + expect(ids(await repository.getAll())).toEqual(['item-000', 'item-002']) + expect(ids((await repository.getPaginated(1, 10)).items)).toEqual(['item-000', 'item-002']) + expect(ids(await repository.getFirstNItems(3))).toEqual(['item-000', 'item-002']) + }) + + test('set keeps index and data consistent under concurrent set and delete', async () => { + await Promise.all( + Array.from({ length: 200 }, (_, i) => + i % 2 === 0 ? repository.set({ id: 'same', data: `${i}` }) : repository.deleteById('same') + ) + ) + + const indexed = await redis.zcard(`items:${name}:`) + const stored = await redis.hlen(`items-data:${name}:`) + expect(indexed).toBe(stored) + }) + + test('set with an explicit score', async () => { + await repository.set({ id: 'b', data: 'b' }, { score: 20 }) + await repository.set({ id: 'a', data: 'a' }, { score: 10 }) + + expect(await repository.getItemScoreById('a')).toBe(10) + expect(ids(await repository.getAll())).toEqual(['a', 'b']) + await expectRejection(repository.set({ id: 'c', data: 'c' }, { score: NaN }), RangeError) + }) + + test('set with keepScore preserves the position of updated items', async () => { + await repository.setMany(toItems(3)) + await repository.set({ id: 'item-000', data: 'updated' }, { keepScore: true }) + + const all = await repository.getAll() + expect(ids(all)).toEqual(['item-000', 'item-001', 'item-002']) + expect(all[0].data).toBe('updated') + + await repository.set({ id: 'item-000', data: 'moved' }) + expect(ids(await repository.getAll())).toEqual(['item-001', 'item-002', 'item-000']) + }) + + test('getPaginated in descending order', async () => { + await repository.setMany(toItems(5)) + + const result = await repository.getPaginated(1, 2, 'desc') + expect(ids(result.items)).toEqual(['item-004', 'item-003']) + expect(result.count).toBe(5) + }) + + test('getItemsByScore supports infinite bounds, limit, offset and order', async () => { + const items = toItems(5) + for (const [i, item] of items.entries()) { + await repository.set(item, { score: i }) + } + + expect(ids(await repository.getItemsByScore('-inf', '+inf'))).toEqual(ids(items)) + expect(ids(await repository.getItemsByScore(1, '+inf', { offset: 1, limit: 2 }))).toEqual(['item-002', 'item-003']) + expect(ids(await repository.getItemsByScore(0, 4, { order: 'desc', limit: 2 }))).toEqual(['item-004', 'item-003']) + expect(await repository.getItemsByScore(0, 4, { limit: 0 })).toEqual([]) + }) + + test('setMany, getByIds and deleteByIds', async () => { + const items = toItems(4) + await repository.setMany(items) + + expect(ids(await repository.getAll())).toEqual(ids(items)) + expect(await repository.getByIds(['item-003', 'missing'])).toEqual([items[3], null]) + expect(await repository.deleteByIds(['item-000', 'item-002', 'missing'])).toBe(2) + expect(ids(await repository.getAll())).toEqual(['item-001', 'item-003']) + expect(await redis.hlen(`items-data:${name}:`)).toBe(2) + }) + + test('iterate streams all items in order', async () => { + const items = toItems(25) + await repository.setMany(items) + + const asc: string[] = [] + for await (const item of repository.iterate({ batchSize: 10 })) { + asc.push(item.id) + } + expect(asc).toEqual(ids(items)) + + const desc: string[] = [] + for await (const item of repository.iterate({ batchSize: 5, order: 'desc' })) { + desc.push(item.id) + } + expect(desc).toEqual(ids(items).reverse()) + }) + + test('deleteByScore and trimToSize', async () => { + const items = toItems(10) + for (const [i, item] of items.entries()) { + await repository.set(item, { score: i }) + } + + expect(await repository.deleteByScore(0, 2)).toBe(3) + expect(await repository.trimToSize(4)).toBe(3) + expect(ids(await repository.getAll())).toEqual(['item-006', 'item-007', 'item-008', 'item-009']) + expect(await repository.trimToSize(10)).toBe(0) + expect(await redis.hlen(`items-data:${name}:`)).toBe(4) + }) + + test('maxItems caps the repository size', async () => { + const capped = new SortedItemRepository(name, redis, { maxItems: 3 }) + await capped.setMany(toItems(5)) + await capped.set({ id: 'last', data: 'last' }) + + expect(ids(await capped.getAll())).toEqual(['item-003', 'item-004', 'last']) + expect(await capped.hasItem('item-000')).toBe(false) + }) + + test('hashTag option wraps the name in a cluster hash tag', async () => { + const tagged = new SortedItemRepository(name, redis, { hashTag: true }) + await tagged.set({ id: 'a', data: 'a' }) + + expect(await redis.exists(`items:{${name}}:`, `items-data:{${name}}:`)).toBe(2) + expect(await tagged.getById('a')).toEqual({ id: 'a', data: 'a' }) + }) + + test('supports a custom serializer', async () => { + const serializer: ISerializer = { + encode: (item) => Buffer.from(JSON.stringify(item)), + decode: (buffer) => JSON.parse(buffer.toString()) + } + const jsonRepository = new SortedItemRepository(name, redis, { serializer }) + await jsonRepository.set({ id: 'a', data: 'a' }) + + expect(await redis.hget(`items-data:${name}:`, 'a')).toBe(JSON.stringify({ id: 'a', data: 'a' })) + expect(await jsonRepository.getAll()).toEqual([{ id: 'a', data: 'a' }]) + }) +}) From 592e1566ca21dba9eec145dac4e699cb0a18df28 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 24 Oct 2026 14:51:50 +0000 Subject: [PATCH 2/3] v0.4.0 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JLbcuU2UWtjZCuWZa1QTkE --- package-lock.json | 4 ++-- package.json | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index 64cb9e8..f4747d9 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "item-store-redis", - "version": "0.2.1", + "version": "0.4.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "item-store-redis", - "version": "0.2.1", + "version": "0.4.0", "license": "MIT", "dependencies": { "@msgpack/msgpack": "^2.8.0", diff --git a/package.json b/package.json index d9d40d6..a2af7d8 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "item-store-redis", - "version": "0.3.0", + "version": "0.4.0", "description": "This module provides classes that implement repositories for storing and retrieving data in a Redis database", "main": "dist/libs/index.js", "types": "dist/libs/index.d.ts", From 6c50def4407b3f6844f918cfaab4f0251038d90a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 24 Oct 2026 14:54:29 +0000 Subject: [PATCH 3/3] Add bun.lock so CI installs the Rome binary for Linux package-lock.json was generated on darwin-arm64 and only records that platform's @rometools/cli package, so the bun install in CI skipped the linux-x64 binary and `bun run lint` failed with "Cannot find module '@rometools/cli-linux-x64/rome'". bun.lock takes precedence for bun and records every platform's binary; all dependency versions match package-lock.json. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JLbcuU2UWtjZCuWZa1QTkE --- bun.lock | 77 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 bun.lock diff --git a/bun.lock b/bun.lock new file mode 100644 index 0000000..7b45414 --- /dev/null +++ b/bun.lock @@ -0,0 +1,77 @@ +{ + "lockfileVersion": 2, + "configVersion": 0, + "workspaces": { + "": { + "name": "item-store-redis", + "dependencies": { + "@msgpack/msgpack": "^2.8.0", + "ioredis": "^5.3.1", + }, + "devDependencies": { + "@types/msgpack-lite": "^0.1.8", + "bun-types": "^0.5.6", + "msgpack-lite": "^0.1.26", + "rome": "^11.0.0", + "typescript": "^4.9.5", + }, + }, + }, + "packages": { + "@ioredis/commands": ["@ioredis/commands@1.2.0", "", {}, ""], + + "@msgpack/msgpack": ["@msgpack/msgpack@2.8.0", "", {}, ""], + + "@rometools/cli-darwin-arm64": ["@rometools/cli-darwin-arm64@11.0.0", "", { "os": "darwin", "cpu": "arm64" }, "sha512-F3vkdY+s3FLIEnAjSbyHTuIPB88cLpccimW4ecid5I7S6GzGG3iUJI4xT00JhH73K4P/qW20/9r+kH1T9Du8Xg=="], + + "@rometools/cli-darwin-x64": ["@rometools/cli-darwin-x64@11.0.0", "", { "os": "darwin", "cpu": "x64" }, "sha512-X6jhtS6Iml4GOzgNtnLwIp/KXXhSdqeVyfv69m/AHnIzx3gQAjPZ7BPnJLvTCbhe4SKHL+uTZYFSCJpkUUKE6w=="], + + "@rometools/cli-linux-arm64": ["@rometools/cli-linux-arm64@11.0.0", "", { "os": "linux", "cpu": "arm64" }, "sha512-dktTJJlTpmycBZ2TwhJBcAO8ztK8DdevdyZnFFxdYRvtmJgTjIsC2UFayf/SbKew8B8q1IhI0it+D6ihAeIpeg=="], + + "@rometools/cli-linux-x64": ["@rometools/cli-linux-x64@11.0.0", "", { "os": "linux", "cpu": "x64" }, "sha512-WVcnXPNdWGUWo0p4NU8YzuthjYR7q+b4vRcjdxtP1DlpphZmSsoC/RSE85nEqRAz8hChcKUansVzOPM8BSsuGA=="], + + "@rometools/cli-win32-arm64": ["@rometools/cli-win32-arm64@11.0.0", "", { "os": "win32", "cpu": "arm64" }, "sha512-tPj6RThQzS7Q45jqQll7NlTYvNcsg/BEP3LYiiazqSh9FAFnMkrV6ewUcMPKWyAfiyLs7jlz4rRvdNRUSygzfQ=="], + + "@rometools/cli-win32-x64": ["@rometools/cli-win32-x64@11.0.0", "", { "os": "win32", "cpu": "x64" }, "sha512-bmBai8WHxYjsGk1+je7ZTfCUCWq30WJI3pQM8pzTA674lfGTZ9ymJoZwTaIMSO4rL5V9mlO6uLunsBKso9VqOg=="], + + "@types/msgpack-lite": ["@types/msgpack-lite@0.1.8", "", { "dependencies": { "@types/node": "*" } }, ""], + + "@types/node": ["@types/node@18.15.0", "", {}, ""], + + "bun-types": ["bun-types@0.5.6", "", {}, ""], + + "cluster-key-slot": ["cluster-key-slot@1.1.2", "", {}, ""], + + "debug": ["debug@4.3.4", "", { "dependencies": { "ms": "2.1.2" } }, ""], + + "denque": ["denque@2.1.0", "", {}, ""], + + "event-lite": ["event-lite@0.1.3", "", {}, ""], + + "ieee754": ["ieee754@1.2.1", "", {}, ""], + + "int64-buffer": ["int64-buffer@0.1.10", "", {}, ""], + + "ioredis": ["ioredis@5.3.1", "", { "dependencies": { "@ioredis/commands": "^1.1.1", "cluster-key-slot": "^1.1.0", "debug": "^4.3.4", "denque": "^2.1.0", "lodash.defaults": "^4.2.0", "lodash.isarguments": "^3.1.0", "redis-errors": "^1.2.0", "redis-parser": "^3.0.0", "standard-as-callback": "^2.1.0" } }, ""], + + "isarray": ["isarray@1.0.0", "", {}, ""], + + "lodash.defaults": ["lodash.defaults@4.2.0", "", {}, ""], + + "lodash.isarguments": ["lodash.isarguments@3.1.0", "", {}, ""], + + "ms": ["ms@2.1.2", "", {}, ""], + + "msgpack-lite": ["msgpack-lite@0.1.26", "", { "dependencies": { "event-lite": "^0.1.1", "ieee754": "^1.1.8", "int64-buffer": "^0.1.9", "isarray": "^1.0.0" }, "bin": { "msgpack": "bin/msgpack" } }, ""], + + "redis-errors": ["redis-errors@1.2.0", "", {}, ""], + + "redis-parser": ["redis-parser@3.0.0", "", { "dependencies": { "redis-errors": "^1.0.0" } }, ""], + + "rome": ["rome@11.0.0", "", { "optionalDependencies": { "@rometools/cli-darwin-arm64": "11.0.0", "@rometools/cli-darwin-x64": "11.0.0", "@rometools/cli-linux-arm64": "11.0.0", "@rometools/cli-linux-x64": "11.0.0", "@rometools/cli-win32-arm64": "11.0.0", "@rometools/cli-win32-x64": "11.0.0" }, "bin": { "rome": "bin/rome" } }, "sha512-rRo6JOwpMLc3OkeTDRXkrmrDqnxDvZ75GS4f0jLDBNmRgDXWbu0F8eVnJoRn+VbK2AE7vWvhVOMBjnWowcopkQ=="], + + "standard-as-callback": ["standard-as-callback@2.1.0", "", {}, ""], + + "typescript": ["typescript@4.9.5", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, ""], + } +}