swadhin / opencodehub public

feat: add src/db/adapter/dynamodb.ts

swadhinbiswas committed 8 months ago
35fcda6
1 parent d6773b9
commit 35fcda6
1 file changed
+221 -0
Showing 1 changed file+221-0
src/db/adapter/dynamodb.ts
▼src/db/adapter/dynamodb.ts
+221-0
@@ -0,0 +1,221 @@
1
+import {
2
+ BaseDatabaseAdapter,
3
+ ColumnType,
4
+ DatabaseConfig,
5
+ Migration,
6
+ MigrationStatus,
7
+ QueryOptions,
8
+ QueryResult,
9
+ TableAlterCommand,
10
+ TableSchema,
11
+} from "./types";
12
+
13
+const dynamicImport = new Function("m", "return import(m)") as (m: string) => Promise<any>;
14
+
15
+function toAttr(value: any): any {
16
+ if (value === null || value === undefined) return { NULL: true };
17
+ if (typeof value === "string") return { S: value };
18
+ if (typeof value === "number") return { N: String(value) };
19
+ if (typeof value === "boolean") return { BOOL: value };
20
+ if (Array.isArray(value)) return { L: value.map(toAttr) };
21
+ if (typeof value === "object") {
22
+ const map: Record<string, any> = {};
23
+ for (const [k, v] of Object.entries(value)) map[k] = toAttr(v);
24
+ return { M: map };
25
+ }
26
+ return { S: String(value) };
27
+}
28
+
29
+function fromAttr(attr: any): any {
30
+ if (!attr) return undefined;
31
+ if (attr.S !== undefined) return attr.S;
32
+ if (attr.N !== undefined) return Number(attr.N);
33
+ if (attr.BOOL !== undefined) return attr.BOOL;
34
+ if (attr.NULL) return null;
35
+ if (attr.L) return attr.L.map(fromAttr);
36
+ if (attr.M) {
37
+ const obj: Record<string, any> = {};
38
+ for (const [k, v] of Object.entries(attr.M)) obj[k] = fromAttr(v);
39
+ return obj;
40
+ }
41
+ return undefined;
42
+}
43
+
44
+function fromItem(item: Record<string, any>): any {
45
+ const out: Record<string, any> = {};
46
+ for (const [k, v] of Object.entries(item || {})) out[k] = fromAttr(v);
47
+ return out;
48
+}
49
+
50
+function toItem(obj: Record<string, any>): Record<string, any> {
51
+ const out: Record<string, any> = {};
52
+ for (const [k, v] of Object.entries(obj)) out[k] = toAttr(v);
53
+ return out;
54
+}
55
+
56
+export class DynamoDBAdapter extends BaseDatabaseAdapter {
57
+ private client: any;
58
+ private commands: any;
59
+
60
+ async connect(): Promise<void> {
61
+ if (this.connected) return;
62
+ const mod = await dynamicImport("@aws-sdk/client-dynamodb");
63
+ this.client = new mod.DynamoDBClient({
64
+ endpoint: this.config.host ? `${this.config.ssl ? "https" : "http"}://${this.config.host}${this.config.port ? `:${this.config.port}` : ""}` : undefined,
65
+ region: this.config.options?.region || this.config.database || "us-east-1",
66
+ credentials: this.config.user && this.config.password
67
+ ? { accessKeyId: this.config.user, secretAccessKey: this.config.password }
68
+ : undefined,
69
+ });
70
+ this.commands = mod;
71
+ this.connected = true;
72
+ }
73
+
74
+ async disconnect(): Promise<void> {
75
+ this.connected = false;
76
+ }
77
+
78
+ async findOne<T>(table: string, options?: QueryOptions): Promise<T | null> {
79
+ const rows = await this.findMany<T>(table, { ...options, limit: 1 });
80
+ return rows[0] || null;
81
+ }
82
+
83
+ async findMany<T>(table: string, options?: QueryOptions): Promise<T[]> {
84
+ const res = await this.client.send(new this.commands.ScanCommand({
85
+ TableName: table,
86
+ Limit: options?.limit,
87
+ }));
88
+ let rows = (res.Items || []).map(fromItem);
89
+ if (options?.where?.length) {
90
+ rows = rows.filter((row: any) =>
91
+ options.where!.every((w) => {
92
+ switch (w.operator) {
93
+ case "=":
94
+ case "eq":
95
+ return row[w.field] === w.value;
96
+ case "!=":
97
+ case "ne":
98
+ return row[w.field] !== w.value;
99
+ default:
100
+ return true;
101
+ }
102
+ })
103
+ );
104
+ }
105
+ if (options?.offset) rows = rows.slice(options.offset);
106
+ if (options?.limit) rows = rows.slice(0, options.limit);
107
+ return rows as T[];
108
+ }
109
+
110
+ async create<T>(table: string, data: Partial<T>): Promise<T> {
111
+ const row = { ...(data as any) };
112
+ if (!row.id) row.id = `${Date.now()}_${Math.random().toString(36).slice(2)}`;
113
+ await this.client.send(new this.commands.PutItemCommand({
114
+ TableName: table,
115
+ Item: toItem(row),
116
+ }));
117
+ return row as T;
118
+ }
119
+
120
+ async createMany<T>(table: string, data: Partial<T>[]): Promise<T[]> {
121
+ const out: T[] = [];
122
+ for (const item of data) out.push(await this.create<T>(table, item));
123
+ return out;
124
+ }
125
+
126
+ async update<T>(table: string, id: string | number, data: Partial<T>): Promise<T | null> {
127
+ const current = await this.findById<any>(table, id);
128
+ if (!current) return null;
129
+ const next = { ...current, ...(data as any), id: String(id) };
130
+ await this.client.send(new this.commands.PutItemCommand({
131
+ TableName: table,
132
+ Item: toItem(next),
133
+ }));
134
+ return next as T;
135
+ }
136
+
137
+ async updateMany<T>(table: string, options: QueryOptions, data: Partial<T>): Promise<number> {
138
+ const rows = await this.findMany<any>(table, options);
139
+ for (const row of rows) await this.update(table, row.id, data);
140
+ return rows.length;
141
+ }
142
+
143
+ async delete(table: string, id: string | number): Promise<boolean> {
144
+ await this.client.send(new this.commands.DeleteItemCommand({
145
+ TableName: table,
146
+ Key: { id: { S: String(id) } },
147
+ }));
148
+ return true;
149
+ }
150
+
151
+ async deleteMany(table: string, options?: QueryOptions): Promise<number> {
152
+ const rows = await this.findMany<any>(table, options);
153
+ for (const row of rows) await this.delete(table, row.id);
154
+ return rows.length;
155
+ }
156
+
157
+ async count(table: string, options?: QueryOptions): Promise<number> {
158
+ const rows = await this.findMany<any>(table, options);
159
+ return rows.length;
160
+ }
161
+
162
+ async rawQuery<T>(query: string, params?: unknown[]): Promise<QueryResult<T>> {
163
+ throw new Error("DynamoDB adapter does not support SQL rawQuery");
164
+ }
165
+
166
+ async beginTransaction(): Promise<void> {}
167
+ async commitTransaction(): Promise<void> {}
168
+ async rollbackTransaction(): Promise<void> {}
169
+
170
+ async tableExists(table: string): Promise<boolean> {
171
+ try {
172
+ await this.client.send(new this.commands.DescribeTableCommand({ TableName: table }));
173
+ return true;
174
+ } catch {
175
+ return false;
176
+ }
177
+ }
178
+
179
+ async createTable(table: string, schema: TableSchema): Promise<void> {
180
+ const exists = await this.tableExists(table);
181
+ if (exists) return;
182
+ await this.client.send(new this.commands.CreateTableCommand({
183
+ TableName: table,
184
+ AttributeDefinitions: [{ AttributeName: "id", AttributeType: "S" }],
185
+ KeySchema: [{ AttributeName: "id", KeyType: "HASH" }],
186
+ BillingMode: "PAY_PER_REQUEST",
187
+ }));
188
+ }
189
+
190
+ async dropTable(table: string): Promise<void> {
191
+ if (!(await this.tableExists(table))) return;
192
+ await this.client.send(new this.commands.DeleteTableCommand({ TableName: table }));
193
+ }
194
+
195
+ async alterTable(table: string, changes: TableAlterCommand[]): Promise<void> {}
196
+
197
+ async runMigration(migration: Migration): Promise<void> {
198
+ await migration.up(this);
199
+ await this.createTable("_migrations", { columns: [{ name: "id", type: "string" }], primaryKey: "id" });
200
+ await this.create("_migrations", {
201
+ id: migration.id,
202
+ name: migration.name,
203
+ executedAt: new Date().toISOString(),
204
+ } as any);
205
+ }
206
+
207
+ async getMigrationStatus(): Promise<MigrationStatus[]> {
208
+ if (!(await this.tableExists("_migrations"))) return [];
209
+ const rows = await this.findMany<any>("_migrations");
210
+ return rows.map((row) => ({
211
+ id: row.id,
212
+ name: row.name || row.id,
213
+ executedAt: row.executedAt ? new Date(row.executedAt) : null,
214
+ status: row.executedAt ? "executed" : "pending",
215
+ }));
216
+ }
217
+
218
+ protected mapColumnType(type: ColumnType): string {
219
+ return type;
220
+ }
221
+}
0
222