-
Notifications
You must be signed in to change notification settings - Fork 78
/
client.ts
252 lines (216 loc) · 6.63 KB
/
client.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
import { createClient } from 'redis';
import { Repository } from './repository';
import { JsonRepository, HashRepository } from './repository';
import { Entity } from './entity/entity';
import { Schema } from './schema/schema';
import { RedisError } from './errors';
export type RedisConnection = ReturnType<typeof createClient>;
/**
* Alias for a JavaScript object used by HSET.
* @internal
*/
export type RedisHashData = { [key: string]: string };
/**
* Alias for any old JavaScript object used by JSON.SET.
* @internal
*/
export type RedisJsonData = { [key: string]: any };
/** The type of data structure in Redis to map objects to. */
export type SearchDataStructure = 'HASH' | 'JSON';
/** @internal */
export type CreateIndexOptions = {
indexName: string,
dataStructure: SearchDataStructure,
schema: Array<string>,
prefix: string,
stopWords?: Array<string>
}
/** @internal */
export type LimitOptions = {
offset: number;
count: number;
}
/** @internal */
export type SortOptions = {
field: string;
order: 'ASC' | 'DESC';
}
/** @internal */
export type SearchOptions = {
indexName: string,
query: string,
limit?: LimitOptions,
sort?: SortOptions,
keysOnly?: boolean
}
/**
* A Client is the starting point for working with Redis OM. Clients manage the
* connection to Redis and provide limited functionality for executing Redis commands.
* Create a client and open it before you use it:
*
* ```typescript
* const client = new Client();
* await client.open();
* ```
*
* A Client is primarily used by a {@link Repository} which requires a client in
* its constructor.
*/
export class Client {
/** @internal */
protected redis?: RedisConnection
/**
* Attaches an existing Node Redis connection to this Redis OM client. Closes
* any existing connection.
* @param connection An existing Node Redis client.
* @returns This {@link Client} instance.
*/
async use(connection: RedisConnection): Promise<Client> {
// close existing connection
await this.close()
this.redis = connection
return this;
}
/**
* Open a connection to Redis at the provided URL.
* @param url A URL to Redis as defined with the [IANA](https://www.iana.org/assignments/uri-schemes/prov/redis).
* @returns This {@link Client} instance.
*/
async open(url: string = 'redis://localhost:6379'): Promise<Client> {
if (!this.isOpen()) {
this.redis = createClient({ url })
await this.redis.connect();
}
return this;
}
/**
* Execute an arbitrary Redis command.
* @template TResult Expect result type such as `string`, `Array<string>`, or whatever complex type Redis returns.
* @param command The command to execute.
* @returns The raw results of calling the Redis command.
*/
async execute(command: Array<string | number | boolean>): Promise<unknown> {
this.validateRedisOpen();
return this.redis.sendCommand<any>(command.map(arg => {
if (arg === false) return '0';
if (arg === true) return '1';
return arg.toString();
}));
}
/**
* Creates a repository for the given schema.
* @template TEntity The entity type for this {@link Schema} and {@link Repository}.
* @param schema The schema.
* @returns A repository for the provided schema.
*/
fetchRepository<TEntity extends Entity>(schema: Schema<TEntity>): Repository<TEntity> {
this.validateRedisOpen();
if (schema.dataStructure === 'JSON') {
return new JsonRepository(schema, this);
} else {
return new HashRepository(schema, this);
}
}
/**
* Close the connection to Redis.
*/
async close() {
if (this.redis) {
await this.redis.quit()
}
this.redis = undefined;
}
/** @internal */
async createIndex(options: CreateIndexOptions) {
this.validateRedisOpen();
const { indexName, dataStructure, prefix, schema, stopWords } = options;
const command = [
'FT.CREATE', indexName,
'ON', dataStructure,
'PREFIX', '1', `${prefix}`];
if (stopWords !== undefined)
command.push('STOPWORDS', `${stopWords.length}`, ...stopWords);
command.push('SCHEMA', ...schema);
await this.redis.sendCommand(command);
}
/** @internal */
async dropIndex(indexName: string) {
this.validateRedisOpen();
await this.redis.sendCommand(['FT.DROPINDEX', indexName]);
}
/** @internal */
async search(options: SearchOptions) {
this.validateRedisOpen();
const { indexName, query, limit, sort, keysOnly } = options
const command = ['FT.SEARCH', indexName, query];
if (limit !== undefined)
command.push('LIMIT', limit.offset.toString(), limit.count.toString());
if (sort !== undefined)
command.push('SORTBY', sort.field, sort.order);
if (keysOnly) command.push('RETURN', '0');
return this.redis.sendCommand<any[]>(command);
}
/** @internal */
async unlink(...keys: string[]) {
this.validateRedisOpen();
if (keys.length > 0) await this.redis.unlink(keys);
}
/** @internal */
async expire(key: string, ttl: number) {
this.validateRedisOpen();
await this.redis.expire(key, ttl);
}
/** @internal */
async get(key: string): Promise<string | null> {
this.validateRedisOpen();
return this.redis.get(key);
}
/** @internal */
async set(key: string, value: string) {
this.validateRedisOpen();
await this.redis.set(key, value);
}
/** @internal */
async hgetall(key: string): Promise<RedisHashData> {
this.validateRedisOpen();
return this.redis.hGetAll(key);
}
/** @internal */
async hsetall(key: string, data: RedisHashData) {
this.validateRedisOpen();
try {
await this.redis.executeIsolated(async isolatedClient => {
await isolatedClient.watch(key);
await isolatedClient
.multi()
.unlink(key)
.hSet(key, data)
.exec();
});
} catch (error: any) {
if (error.name === 'WatchError') throw new RedisError("Watch error when setting HASH.");
throw error;
}
}
/** @internal */
async jsonget(key: string): Promise<RedisJsonData> {
this.validateRedisOpen();
const json = await this.redis.sendCommand<string>(['JSON.GET', key, '.']);
return JSON.parse(json);
}
/** @internal */
async jsonset(key: string, data: RedisJsonData) {
this.validateRedisOpen();
const json = JSON.stringify(data);
await this.redis.sendCommand<string>(['JSON.SET', key, '.', json]);
}
/**
* @returns Whether a connection is already open.
*/
isOpen() {
return !!this.redis;
}
private validateRedisOpen(): asserts this is { redis: RedisConnection } {
if (!this.redis) throw new RedisError("Redis connection needs to be open.");
}
}