This repository has no description
Something went wrong. Try again.
9.1 kB · 279 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280// SPDX-License-Identifier: AGPL-3.0-or-later
import {getKvMeta} from '../database/CassandraMetaRegistry';import type {CassandraQueryExecutorForTesting} from '../database/CassandraQueryExecution';import type {CassandraParams, KvQueryMeta, PreparedQuery, WhereExpr} from '../database/CassandraTypes';
type Row = Record<string, unknown>;
function normalizeCql(cql: string): string { return cql.replace(/\s+/g, ' ').trim();}
function compareValues(a: unknown, b: unknown): number { if (typeof a === 'bigint' || typeof b === 'bigint') { const av = typeof a === 'bigint' ? a : BigInt(a as number | string); const bv = typeof b === 'bigint' ? b : BigInt(b as number | string); if (av === bv) return 0; return av < bv ? -1 : 1; } const av = a instanceof Date ? a.getTime() : typeof a === 'bigint' ? Number(a) : a; const bv = b instanceof Date ? b.getTime() : typeof b === 'bigint' ? Number(b) : b; if (av === bv) return 0; return (av as number | string) < (bv as number | string) ? -1 : 1;}
function valuesEqual(a: unknown, b: unknown): boolean { if (a == null && b == null) return true; if (a instanceof Date && b instanceof Date) return a.getTime() === b.getTime(); if (Buffer.isBuffer(a) && Buffer.isBuffer(b)) return a.equals(b); return a === b;}
function cloneValue<T>(value: T): T { if (value instanceof Date) return new Date(value.getTime()) as T; if (value instanceof Set) return new Set([...value].map((entry) => cloneValue(entry))) as T; if (value instanceof Map) { return new Map([...value.entries()].map(([key, entry]) => [cloneValue(key), cloneValue(entry)])) as T; } if (Array.isArray(value)) return value.map((entry) => cloneValue(entry)) as T; if (value && typeof value === 'object' && Object.getPrototypeOf(value) === Object.prototype) { return {...(value as object)} as T; } return value;}
function cloneRow(row: Row): Row { const cloned: Row = {}; for (const [key, value] of Object.entries(row)) { cloned[key] = cloneValue(value); } return cloned;}
function pkKey(meta: KvQueryMeta, row: Row): string { return meta.table.primaryKey.map((column) => valueKey(row[column])).join('\u0000');}
function valueKey(value: unknown): string { if (Buffer.isBuffer(value)) return `buffer:${value.toString('base64')}`; if (value instanceof Date) return `date:${value.toISOString()}`; if (typeof value === 'bigint') return `bigint:${value.toString()}`; return `${typeof value}:${String(value)}`;}
function getParam(params: CassandraParams, param: string): unknown { return params[param];}
function matchesWhere(row: Row, where: ReadonlyArray<WhereExpr<Row>> | undefined, params: CassandraParams): boolean { for (const clause of where ?? []) { switch (clause.kind) { case 'eq': if (!valuesEqual(row[clause.col], getParam(params, clause.param))) return false; break; case 'in': { const values = getParam(params, clause.param) as ReadonlyArray<unknown> | Set<unknown> | undefined; const haystack = values instanceof Set ? [...values] : (values ?? []); if (!haystack.some((value) => valuesEqual(row[clause.col], value))) return false; break; } case 'lt': if (compareValues(row[clause.col], getParam(params, clause.param)) >= 0) return false; break; case 'lte': if (compareValues(row[clause.col], getParam(params, clause.param)) > 0) return false; break; case 'gt': if (compareValues(row[clause.col], getParam(params, clause.param)) <= 0) return false; break; case 'gte': if (compareValues(row[clause.col], getParam(params, clause.param)) < 0) return false; break; case 'tokenGt': break; case 'tupleGt': { const left = clause.cols.map((column) => row[column]); const right = clause.params.map((param) => getParam(params, param)); let greater = false; for (let i = 0; i < left.length; i += 1) { const cmp = compareValues(left[i], right[i]); if (cmp > 0) { greater = true; break; } if (cmp < 0) break; } if (!greater) return false; break; } } } return true;}
function projectRow(row: Row, columns: ReadonlyArray<string> | undefined): Row { if (!columns) return cloneRow(row); const projected: Row = {}; for (const column of columns) { projected[column] = cloneValue(row[column]); } return projected;}
function parseRawMeta(cql: string): KvQueryMeta<Row> | null { const normalized = normalizeCql(cql).replace(/;$/, ''); const update = /^UPDATE\s+([a-zA-Z0-9_]+)\s+SET\s+(.+?)\s+WHERE\s+(.+)$/i.exec(normalized); if (update) { const table = update[1]!; const patchKeys = update[2]! .split(',') .map((part) => part.trim().split(/\s*=\s*/)[0]) .filter(Boolean); const where = update[3]! .split(/\s+AND\s+/i) .map((part) => part.trim().split(/\s*=\s*/)[0]) .filter(Boolean); return { action: 'patch', table: {name: table, columns: [...new Set([...where, ...patchKeys])], primaryKey: where, partitionKey: where}, where: where.map((col) => ({kind: 'eq', col, param: col})), patchKeys, pkColumns: where, }; } const insert = /^INSERT\s+INTO\s+([a-zA-Z0-9_]+)\s*\((.+?)\)\s+VALUES/i.exec(normalized); if (insert) { const columns = insert[2]!.split(',').map((part) => part.trim()); const pk = columns.includes('event_id') ? ['event_id'] : [columns[0]!]; return { action: 'upsert', table: {name: insert[1]!, columns, primaryKey: pk, partitionKey: pk}, }; } return null;}
export class InMemoryCassandraQueryExecutor implements CassandraQueryExecutorForTesting { private readonly tables = new Map<string, Map<string, Row>>();
reset(): void { this.tables.clear(); }
private table(meta: KvQueryMeta): Map<string, Row> { let table = this.tables.get(meta.table.name); if (!table) { table = new Map(); this.tables.set(meta.table.name, table); } return table; }
private meta(query: PreparedQuery): KvQueryMeta<Row> { const meta = (query.kvMeta ?? getKvMeta(query.cql) ?? parseRawMeta(query.cql)) as KvQueryMeta<Row> | null; if (!meta) { throw new Error(`In-memory Cassandra does not understand query: ${query.cql}`); } return meta; }
async executeQuery<T = Row>(query: PreparedQuery): Promise<Array<T>> { const meta = this.meta(query); switch (meta.action) { case 'select': return this.select(meta, query.params) as Array<T>; case 'count': return [{count: this.select(meta, query.params).length}] as Array<T>; case 'upsert': if (meta.ifNotExists) { return [{'[applied]': this.upsert(meta, query.params, true)}] as Array<T>; } this.upsert(meta, query.params); return []; case 'patch': this.patch(meta, query.params); return []; case 'delete': this.delete(meta, query.params); return []; case 'batch': return []; } return []; }
async executeBatch(queries: Array<{query: string; params: object; meta?: KvQueryMeta}>): Promise<void> { for (const query of queries) { await this.executeQuery({ cql: query.query, params: query.params as CassandraParams, kvMeta: query.meta, }); } }
private rowFromParams(meta: KvQueryMeta, params: CassandraParams): Row { const row: Row = {}; for (const column of meta.table.columns) { if (column in params) { row[column] = cloneValue(params[column]); } } if (meta.nowColumn) { row[meta.nowColumn] = new Date(); } return row; }
private keyFromParams(meta: KvQueryMeta, params: CassandraParams): string { const row: Row = {}; for (const column of meta.pkColumns ?? meta.table.primaryKey) { row[column] = params[column]; } return pkKey(meta, row); }
private upsert(meta: KvQueryMeta, params: CassandraParams, ifNotExists = false): boolean { const table = this.table(meta); const next = this.rowFromParams(meta, params); const key = pkKey(meta, next); if (ifNotExists && table.has(key)) { return false; } table.set(key, {...(table.get(key) ?? {}), ...next}); return true; }
private patch(meta: KvQueryMeta, params: CassandraParams): void { const table = this.table(meta); const key = this.keyFromParams(meta, params); const row = table.get(key) ?? this.rowFromParams({...meta, table: {...meta.table, columns: meta.table.primaryKey}}, params); for (const column of meta.patchKeys ?? []) { row[column] = column in params ? cloneValue(params[column]) : null; } table.set(key, row); }
private delete(meta: KvQueryMeta, params: CassandraParams): void { const table = this.table(meta); for (const [key, row] of table.entries()) { if (matchesWhere(row, meta.where, params)) { table.delete(key); } } }
private select(meta: KvQueryMeta, params: CassandraParams): Array<Row> { let rows = [...this.table(meta).values()].filter((row) => matchesWhere(row, meta.where, params)); if (meta.orderBy) { const direction = meta.orderBy.direction === 'DESC' ? -1 : 1; rows = rows.sort((a, b) => compareValues(a[meta.orderBy!.col], b[meta.orderBy!.col]) * direction); } if (typeof meta.limit === 'number') { rows = rows.slice(0, meta.limit); } return rows.map((row) => projectRow(row, meta.columns)); }}