swadhin / opencodehub public

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

swadhinbiswas committed 8 months ago
bbd7a22
1 parent 91332f2
commit bbd7a22
1 file changed
+276 -0
Showing 1 changed file+276-0
src/db/adapter/mongodb.ts
▼src/db/adapter/mongodb.ts
+276-0
@@ -0,0 +1,276 @@
1
+import {
2
+ BaseDatabaseAdapter,
3
+ ColumnType,
4
+ DatabaseConfig,
5
+ Migration,
6
+ MigrationStatus,
7
+ QueryOptions,
8
+ QueryResult,
9
+ TableAlterCommand,
10
+ TableSchema,
11
+ WhereClause,
12
+} from "./types";
13
+
14
+const dynamicImport = new Function("m", "return import(m)") as (m: string) => Promise<any>;
15
+
16
+function toMongoFilter(where?: WhereClause[]): Record<string, unknown> {
17
+ if (!where?.length) return {};
18
+ const filter: Record<string, unknown> = {};
19
+
20
+ for (const clause of where) {
21
+ const field = clause.field === "id" ? "_id" : clause.field;
22
+ switch (clause.operator) {
23
+ case "=":
24
+ case "eq":
25
+ filter[field] = clause.value;
26
+ break;
27
+ case "!=":
28
+ case "ne":
29
+ filter[field] = { $ne: clause.value };
30
+ break;
31
+ case ">":
32
+ case "gt":
33
+ filter[field] = { $gt: clause.value };
34
+ break;
35
+ case ">=":
36
+ case "gte":
37
+ filter[field] = { $gte: clause.value };
38
+ break;
39
+ case "<":
40
+ case "lt":
41
+ filter[field] = { $lt: clause.value };
42
+ break;
43
+ case "<=":
44
+ case "lte":
45
+ filter[field] = { $lte: clause.value };
46
+ break;
47
+ case "in":
48
+ filter[field] = { $in: Array.isArray(clause.value) ? clause.value : [clause.value] };
49
+ break;
50
+ case "not in":
51
+ filter[field] = { $nin: Array.isArray(clause.value) ? clause.value : [clause.value] };
52
+ break;
53
+ case "like":
54
+ case "ilike":
55
+ filter[field] = { $regex: String(clause.value), $options: "i" };
56
+ break;
57
+ case "is null":
58
+ case "isNull":
59
+ filter[field] = null;
60
+ break;
61
+ case "is not null":
62
+ case "isNotNull":
63
+ filter[field] = { $ne: null };
64
+ break;
65
+ default:
66
+ break;
67
+ }
68
+ }
69
+
70
+ return filter;
71
+}
72
+
73
+function normalizeDoc<T>(doc: any): T {
74
+ if (!doc) return doc;
75
+ if (doc._id !== undefined) {
76
+ doc.id = typeof doc._id === "string" ? doc._id : String(doc._id);
77
+ }
78
+ return doc as T;
79
+}
80
+
81
+export class MongoDBAdapter extends BaseDatabaseAdapter {
82
+ private client: any;
83
+ private db: any;
84
+ private session: any = null;
85
+
86
+ async connect(): Promise<void> {
87
+ if (this.connected) return;
88
+
89
+ const mongodb = await dynamicImport("mongodb");
90
+ const MongoClient = mongodb.MongoClient;
91
+
92
+ this.client = new MongoClient(this.config.url, {
93
+ ignoreUndefined: true,
94
+ });
95
+ await this.client.connect();
96
+
97
+ if (this.config.database) {
98
+ this.db = this.client.db(this.config.database);
99
+ } else {
100
+ const parsed = new URL(this.config.url);
101
+ const dbName = parsed.pathname.replace(/^\//, "") || "opencodehub";
102
+ this.db = this.client.db(dbName);
103
+ }
104
+
105
+ this.connected = true;
106
+ }
107
+
108
+ async disconnect(): Promise<void> {
109
+ if (!this.connected) return;
110
+ await this.client?.close();
111
+ this.connected = false;
112
+ this.session = null;
113
+ }
114
+
115
+ private collection(table: string) {
116
+ return this.db.collection(table);
117
+ }
118
+
119
+ async findOne<T>(table: string, options?: QueryOptions): Promise<T | null> {
120
+ const filter = toMongoFilter(options?.where);
121
+ const doc = await this.collection(table).findOne(filter, this.session ? { session: this.session } : undefined);
122
+ return doc ? normalizeDoc<T>(doc) : null;
123
+ }
124
+
125
+ async findMany<T>(table: string, options?: QueryOptions): Promise<T[]> {
126
+ const filter = toMongoFilter(options?.where);
127
+ let cursor = this.collection(table).find(filter, this.session ? { session: this.session } : undefined);
128
+
129
+ if (options?.orderBy?.length) {
130
+ const sort: Record<string, 1 | -1> = {};
131
+ for (const o of options.orderBy) {
132
+ sort[o.field === "id" ? "_id" : o.field] = o.direction === "asc" ? 1 : -1;
133
+ }
134
+ cursor = cursor.sort(sort);
135
+ }
136
+ if (options?.offset) cursor = cursor.skip(options.offset);
137
+ if (options?.limit) cursor = cursor.limit(options.limit);
138
+
139
+ const docs = await cursor.toArray();
140
+ return docs.map((d: any) => normalizeDoc<T>(d));
141
+ }
142
+
143
+ async create<T>(table: string, data: Partial<T>): Promise<T> {
144
+ const doc: Record<string, unknown> = { ...(data as Record<string, unknown>) };
145
+ if (doc.id !== undefined) {
146
+ doc._id = String(doc.id);
147
+ delete doc.id;
148
+ }
149
+
150
+ await this.collection(table).insertOne(doc, this.session ? { session: this.session } : undefined);
151
+ return normalizeDoc<T>(doc);
152
+ }
153
+
154
+ async createMany<T>(table: string, data: Partial<T>[]): Promise<T[]> {
155
+ if (!data.length) return [];
156
+ const docs = data.map((item) => {
157
+ const doc: Record<string, unknown> = { ...(item as Record<string, unknown>) };
158
+ if (doc.id !== undefined) {
159
+ doc._id = String(doc.id);
160
+ delete doc.id;
161
+ }
162
+ return doc;
163
+ });
164
+ await this.collection(table).insertMany(docs, this.session ? { session: this.session } : undefined);
165
+ return docs.map((d) => normalizeDoc<T>(d));
166
+ }
167
+
168
+ async update<T>(table: string, id: string | number, data: Partial<T>): Promise<T | null> {
169
+ const patch = { ...(data as Record<string, unknown>) };
170
+ delete patch.id;
171
+ await this.collection(table).updateOne(
172
+ { _id: String(id) },
173
+ { $set: patch },
174
+ this.session ? { session: this.session } : undefined
175
+ );
176
+ const updated = await this.collection(table).findOne({ _id: String(id) }, this.session ? { session: this.session } : undefined);
177
+ return updated ? normalizeDoc<T>(updated) : null;
178
+ }
179
+
180
+ async updateMany<T>(table: string, options: QueryOptions, data: Partial<T>): Promise<number> {
181
+ const patch = { ...(data as Record<string, unknown>) };
182
+ delete patch.id;
183
+ const result = await this.collection(table).updateMany(
184
+ toMongoFilter(options.where),
185
+ { $set: patch },
186
+ this.session ? { session: this.session } : undefined
187
+ );
188
+ return result.modifiedCount || 0;
189
+ }
190
+
191
+ async delete(table: string, id: string | number): Promise<boolean> {
192
+ const result = await this.collection(table).deleteOne({ _id: String(id) }, this.session ? { session: this.session } : undefined);
193
+ return (result.deletedCount || 0) > 0;
194
+ }
195
+
196
+ async deleteMany(table: string, options?: QueryOptions): Promise<number> {
197
+ const result = await this.collection(table).deleteMany(
198
+ toMongoFilter(options?.where),
199
+ this.session ? { session: this.session } : undefined
200
+ );
201
+ return result.deletedCount || 0;
202
+ }
203
+
204
+ async count(table: string, options?: QueryOptions): Promise<number> {
205
+ return this.collection(table).countDocuments(toMongoFilter(options?.where), this.session ? { session: this.session } : undefined);
206
+ }
207
+
208
+ async rawQuery<T>(query: string, params?: unknown[]): Promise<QueryResult<T>> {
209
+ throw new Error("MongoDB rawQuery is not supported via SQL string");
210
+ }
211
+
212
+ async beginTransaction(): Promise<void> {
213
+ if (this.session) return;
214
+ this.session = this.client.startSession();
215
+ this.session.startTransaction();
216
+ }
217
+
218
+ async commitTransaction(): Promise<void> {
219
+ if (!this.session) return;
220
+ await this.session.commitTransaction();
221
+ await this.session.endSession();
222
+ this.session = null;
223
+ }
224
+
225
+ async rollbackTransaction(): Promise<void> {
226
+ if (!this.session) return;
227
+ await this.session.abortTransaction();
228
+ await this.session.endSession();
229
+ this.session = null;
230
+ }
231
+
232
+ async tableExists(table: string): Promise<boolean> {
233
+ const cols = await this.db.listCollections({ name: table }).toArray();
234
+ return cols.length > 0;
235
+ }
236
+
237
+ async createTable(table: string, schema: TableSchema): Promise<void> {
238
+ const exists = await this.tableExists(table);
239
+ if (!exists) await this.db.createCollection(table);
240
+ }
241
+
242
+ async dropTable(table: string): Promise<void> {
243
+ const exists = await this.tableExists(table);
244
+ if (exists) await this.collection(table).drop();
245
+ }
246
+
247
+ async alterTable(table: string, changes: TableAlterCommand[]): Promise<void> {
248
+ // No-op by default for document model; schema evolves dynamically.
249
+ }
250
+
251
+ async runMigration(migration: Migration): Promise<void> {
252
+ await migration.up(this);
253
+ const migrations = this.collection("_migrations");
254
+ await migrations.updateOne(
255
+ { _id: migration.id },
256
+ { $set: { name: migration.name, executedAt: new Date().toISOString() } },
257
+ { upsert: true }
258
+ );
259
+ }
260
+
261
+ async getMigrationStatus(): Promise<MigrationStatus[]> {
262
+ const exists = await this.tableExists("_migrations");
263
+ if (!exists) return [];
264
+ const rows = await this.collection("_migrations").find({}).toArray();
265
+ return rows.map((row: any) => ({
266
+ id: String(row._id),
267
+ name: row.name || String(row._id),
268
+ executedAt: row.executedAt ? new Date(row.executedAt) : null,
269
+ status: row.executedAt ? "executed" : "pending",
270
+ }));
271
+ }
272
+
273
+ protected mapColumnType(type: ColumnType): string {
274
+ return type;
275
+ }
276
+}
0
277