swadhin / opencodehub public

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

swadhinbiswas committed 8 months ago

d6773b9

1 parent 7afa58e

commit d6773b9

1 file changed

+213 -0

Files

src

adapter

cassandra.ts +213

Showing 1 changed file+213-0

src/db/adapter/cassandra.ts

▼src/db/adapter/cassandra.ts

+213-0

@@ -0,0 +1,213 @@
	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 buildWhere(where?: WhereClause[]): { sql: string; params: unknown[] } {
	17	+ if (!where?.length) return { sql: "", params: [] };
	18	+ const parts: string[] = [];
	19	+ const params: unknown[] = [];
	20	+ for (const w of where) {
	21	+ switch (w.operator) {
	22	+ case "=":
	23	+ case "eq":
	24	+ parts.push(`${w.field} = ?`);
	25	+ params.push(w.value);
	26	+ break;
	27	+ case "!=":
	28	+ case "ne":
	29	+ parts.push(`${w.field} != ?`);
	30	+ params.push(w.value);
	31	+ break;
	32	+ case ">":
	33	+ case "gt":
	34	+ parts.push(`${w.field} > ?`);
	35	+ params.push(w.value);
	36	+ break;
	37	+ case ">=":
	38	+ case "gte":
	39	+ parts.push(`${w.field} >= ?`);
	40	+ params.push(w.value);
	41	+ break;
	42	+ case "<":
	43	+ case "lt":
	44	+ parts.push(`${w.field} < ?`);
	45	+ params.push(w.value);
	46	+ break;
	47	+ case "<=":
	48	+ case "lte":
	49	+ parts.push(`${w.field} <= ?`);
	50	+ params.push(w.value);
	51	+ break;
	52	+ default:
	53	+ break;
	54	+ }
	55	+ }
	56	+ return { sql: parts.length ? ` WHERE ${parts.join(" AND ")}` : "", params };
	57	+}
	58	+
	59	+export class CassandraAdapter extends BaseDatabaseAdapter {
	60	+ protected client: any;
	61	+
	62	+ async connect(): Promise<void> {
	63	+ if (this.connected) return;
	64	+ const cassandra = await dynamicImport("cassandra-driver");
	65	+ const url = new URL(this.config.url);
	66	+ this.client = new cassandra.Client({
	67	+ contactPoints: [url.hostname \|\| this.config.host \|\| "127.0.0.1"],
	68	+ localDataCenter: (this.config.options?.localDataCenter as string) \|\| "datacenter1",
	69	+ keyspace: this.config.database \|\| url.pathname.replace(/^\//, "") \|\| "opencodehub",
	70	+ credentials: this.config.user && this.config.password
	71	+ ? new cassandra.auth.PlainTextAuthProvider(this.config.user, this.config.password)
	72	+ : undefined,
	73	+ });
	74	+ await this.client.connect();
	75	+ this.connected = true;
	76	+ }
	77	+
	78	+ async disconnect(): Promise<void> {
	79	+ if (!this.connected) return;
	80	+ await this.client.shutdown();
	81	+ this.connected = false;
	82	+ }
	83	+
	84	+ async findOne<T>(table: string, options?: QueryOptions): Promise<T \| null> {
	85	+ const rows = await this.findMany<T>(table, { ...options, limit: 1 });
	86	+ return rows[0] \|\| null;
	87	+ }
	88	+
	89	+ async findMany<T>(table: string, options?: QueryOptions): Promise<T[]> {
	90	+ let sql = `SELECT * FROM ${table}`;
	91	+ const where = buildWhere(options?.where);
	92	+ sql += where.sql;
	93	+ if (options?.limit) sql += ` LIMIT ${options.limit}`;
	94	+ const result = await this.client.execute(sql, where.params, { prepare: true });
	95	+ return result.rows as T[];
	96	+ }
	97	+
	98	+ async create<T>(table: string, data: Partial<T>): Promise<T> {
	99	+ const row = { ...(data as any) };
	100	+ if (!row.id) row.id = `${Date.now()}_${Math.random().toString(36).slice(2)}`;
	101	+ const keys = Object.keys(row);
	102	+ const sql = `INSERT INTO ${table} (${keys.join(", ")}) VALUES (${keys.map(() => "?").join(", ")})`;
	103	+ await this.client.execute(sql, keys.map((k) => row[k]), { prepare: true });
	104	+ return row as T;
	105	+ }
	106	+
	107	+ async createMany<T>(table: string, data: Partial<T>[]): Promise<T[]> {
	108	+ const out: T[] = [];
	109	+ for (const item of data) out.push(await this.create<T>(table, item));
	110	+ return out;
	111	+ }
	112	+
	113	+ async update<T>(table: string, id: string \| number, data: Partial<T>): Promise<T \| null> {
	114	+ const keys = Object.keys(data);
	115	+ const sql = `UPDATE ${table} SET ${keys.map((k) => `${k} = ?`).join(", ")} WHERE id = ?`;
	116	+ await this.client.execute(sql, [...keys.map((k) => (data as any)[k]), String(id)], { prepare: true });
	117	+ return this.findById<T>(table, id);
	118	+ }
	119	+
	120	+ async updateMany<T>(table: string, options: QueryOptions, data: Partial<T>): Promise<number> {
	121	+ const rows = await this.findMany<any>(table, options);
	122	+ for (const row of rows) await this.update(table, row.id, data);
	123	+ return rows.length;
	124	+ }
	125	+
	126	+ async delete(table: string, id: string \| number): Promise<boolean> {
	127	+ await this.client.execute(`DELETE FROM ${table} WHERE id = ?`, [String(id)], { prepare: true });
	128	+ return true;
	129	+ }
	130	+
	131	+ async deleteMany(table: string, options?: QueryOptions): Promise<number> {
	132	+ const rows = await this.findMany<any>(table, options);
	133	+ for (const row of rows) await this.delete(table, row.id);
	134	+ return rows.length;
	135	+ }
	136	+
	137	+ async count(table: string, options?: QueryOptions): Promise<number> {
	138	+ const rows = await this.findMany<any>(table, options);
	139	+ return rows.length;
	140	+ }
	141	+
	142	+ async rawQuery<T>(query: string, params?: unknown[]): Promise<QueryResult<T>> {
	143	+ const result = await this.client.execute(query, params \|\| [], { prepare: true });
	144	+ return {
	145	+ rows: result.rows as T[],
	146	+ rowCount: result.rows.length,
	147	+ };
	148	+ }
	149	+
	150	+ async beginTransaction(): Promise<void> {}
	151	+ async commitTransaction(): Promise<void> {}
	152	+ async rollbackTransaction(): Promise<void> {}
	153	+
	154	+ async tableExists(table: string): Promise<boolean> {
	155	+ const result = await this.client.execute(
	156	+ "SELECT table_name FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
	157	+ [this.config.database, table],
	158	+ { prepare: true }
	159	+ );
	160	+ return result.rows.length > 0;
	161	+ }
	162	+
	163	+ async createTable(table: string, schema: TableSchema): Promise<void> {
	164	+ const columnDefs = schema.columns.map((c) => `${c.name} ${this.mapColumnType(c.type)}`);
	165	+ const pk = Array.isArray(schema.primaryKey)
	166	+ ? schema.primaryKey.join(", ")
	167	+ : schema.primaryKey \|\| "id";
	168	+ await this.client.execute(
	169	+ `CREATE TABLE IF NOT EXISTS ${table} (${columnDefs.join(", ")}, PRIMARY KEY (${pk}))`
	170	+ );
	171	+ }
	172	+
	173	+ async dropTable(table: string): Promise<void> {
	174	+ await this.client.execute(`DROP TABLE IF EXISTS ${table}`);
	175	+ }
	176	+
	177	+ async alterTable(table: string, changes: TableAlterCommand[]): Promise<void> {
	178	+ for (const change of changes) {
	179	+ if (change.type === "add_column" && change.column) {
	180	+ await this.client.execute(
	181	+ `ALTER TABLE ${table} ADD ${change.column.name} ${this.mapColumnType(change.column.type)}`
	182	+ );
	183	+ }
	184	+ }
	185	+ }
	186	+
	187	+ async runMigration(migration: Migration): Promise<void> {
	188	+ await migration.up(this);
	189	+ }
	190	+
	191	+ async getMigrationStatus(): Promise<MigrationStatus[]> {
	192	+ return [];
	193	+ }
	194	+
	195	+ protected mapColumnType(type: ColumnType): string {
	196	+ const map: Record<ColumnType, string> = {
	197	+ string: "text",
	198	+ text: "text",
	199	+ integer: "int",
	200	+ bigint: "bigint",
	201	+ float: "float",
	202	+ decimal: "decimal",
	203	+ boolean: "boolean",
	204	+ date: "date",
	205	+ datetime: "timestamp",
	206	+ timestamp: "timestamp",
	207	+ json: "text",
	208	+ blob: "blob",
	209	+ uuid: "uuid",
	210	+ };
	211	+ return map[type] \|\| "text";
	212	+ }
	213	+}
0	214