;
export interface PreparedQuery {
cql: string;
params: P;
}
export function prepared
(cql: string, params: P): PreparedQuery
{
return {cql, params};
}
export interface QueryTemplate
{
cql: string;
bind(params: P): PreparedQuery
;
}
const clientOptions: cassandra.ClientOptions = {
contactPoints: Config.cassandra.hosts.split(','),
keyspace: Config.cassandra.keyspace,
localDataCenter: Config.cassandra.localDc,
encoding: {
map: Map,
set: Set,
useUndefinedAsUnset: false,
useBigIntAsLong: true,
useBigIntAsVarint: true,
},
};
if (Config.cassandra.username && Config.cassandra.password) {
clientOptions.credentials = {
username: Config.cassandra.username,
password: Config.cassandra.password,
};
}
const client = new cassandra.Client(clientOptions);
const DEFAULT_MAX_PARTITION_KEYS_PER_QUERY = 100;
function chunkArray(items: Array, size: number): Array> {
const chunks: Array> = [];
if (size <= 0) {
throw new Error('Chunk size must be greater than 0');
}
for (let i = 0; i < items.length; i += size) {
chunks.push(items.slice(i, i + size));
}
return chunks;
}
function isUnsafePreparedStatement(query: string): boolean {
const tokens = query.trim().split(/\s+/);
return tokens.length >= 2 && tokens[0].toLowerCase() === 'select' && tokens[1] === '*';
}
function assertNoUndefinedDeep(value: unknown, path: string): void {
if (value === undefined) {
throw new Error(
`Undefined value at "${path}". This project forbids undefined in Cassandra params; use null explicitly or omit the column via PATCH.`,
);
}
if (value === null) return;
const t = typeof value;
if (t === 'string' || t === 'number' || t === 'bigint' || t === 'boolean') return;
if (value instanceof Date) return;
if (value instanceof Buffer) return;
if (typeof value === 'object' && value && value.constructor?.name === 'LocalDate') return;
if (Array.isArray(value)) {
for (let i = 0; i < value.length; i++) {
assertNoUndefinedDeep(value[i], `${path}[${i}]`);
}
return;
}
if (value instanceof Set) {
let idx = 0;
for (const v of value.values()) {
assertNoUndefinedDeep(v, `${path}{set:${idx}}`);
idx++;
}
return;
}
if (value instanceof Map) {
let idx = 0;
for (const [k, v] of value.entries()) {
assertNoUndefinedDeep(k, `${path}{mapKey:${idx}}`);
assertNoUndefinedDeep(v, `${path}{mapVal:${idx}}`);
idx++;
}
return;
}
if (typeof value === 'object') {
for (const [k, v] of Object.entries(value as Record)) {
assertNoUndefinedDeep(v, `${path}.${k}`);
}
}
}
function assertNoUndefinedParams(params: Record): void {
for (const [k, v] of Object.entries(params)) {
assertNoUndefinedDeep(v, `:${k}`);
}
}
function normalizeExecuteArgs(
queryOrPrepared: string | PreparedQuery
,
params?: P,
): PreparedQuery
{
if (typeof queryOrPrepared === 'string') {
if (!params) {
throw new Error('Missing params object for Cassandra query execution');
}
return {cql: queryOrPrepared, params};
}
return queryOrPrepared;
}
async function executeQuery, P extends CassandraParams = CassandraParams>(
queryOrPrepared: string | PreparedQuery,
params?: P,
): Promise> {
const {cql, params: bound} = normalizeExecuteArgs(queryOrPrepared, params);
if (isUnsafePreparedStatement(cql)) {
throw new Error('Cannot prepare a statement that looks like `SELECT *`');
}
assertNoUndefinedParams(bound as Record);
const startTime = IS_DEV ? performance.now() : 0;
try {
const result = await client.execute(cql, bound, {prepare: true});
const rows = (result.rows ?? []) as Array;
if (IS_DEV) {
const durationMs = performance.now() - startTime;
logQuery(getQueryType(cql), cql, bound as Record, durationMs, rows.length);
}
return rows;
} catch (err: unknown) {
const paramSummary: Record = {};
for (const [k, v] of Object.entries(bound as Record)) {
if (typeof v === 'string') paramSummary[k] = {type: 'string', len: v.length};
else if (typeof v === 'bigint') paramSummary[k] = {type: 'bigint'};
else if (typeof v === 'number') paramSummary[k] = {type: 'number'};
else if (typeof v === 'boolean') paramSummary[k] = {type: 'boolean'};
else if (v instanceof Buffer) paramSummary[k] = {type: 'buffer', len: v.length};
else if (v instanceof Set) paramSummary[k] = {type: 'set', size: (v as Set).size};
else if (v instanceof Map) paramSummary[k] = {type: 'map', size: (v as Map).size};
else if (v instanceof Date) paramSummary[k] = {type: 'date'};
else if (Array.isArray(v)) paramSummary[k] = {type: 'array', len: v.length};
else if (v === null) paramSummary[k] = {type: 'null'};
else paramSummary[k] = {type: typeof v};
}
const errorMessage = err instanceof Error ? err.message : String(err);
Logger.warn({error: errorMessage, query: cql, params: paramSummary}, 'Cassandra query failed');
throw err;
}
}
export async function fetchOne, P extends CassandraParams = CassandraParams>(
queryOrPrepared: PreparedQuery | string,
params?: P,
): Promise {
const [row] = await executeQuery(queryOrPrepared, params);
return row ?? null;
}
export async function fetchMany, P extends CassandraParams = CassandraParams>(
queryOrPrepared: PreparedQuery | string,
params?: P,
): Promise> {
return executeQuery(queryOrPrepared, params);
}
export async function fetchManyInChunks<
T = Record,
V = unknown,
P extends CassandraParams = CassandraParams,
>(
query: QueryTemplate | PreparedQuery
| string,
values: Array,
paramsFactory: (chunk: Array) => P,
chunkSize = DEFAULT_MAX_PARTITION_KEYS_PER_QUERY,
): Promise> {
if (values.length === 0) return [];
const chunks = chunkArray(values, chunkSize);
const results = await Promise.all(
chunks.map(async (chunk) => {
const params = paramsFactory(chunk);
if (typeof query === 'string') {
return executeQuery(query, params);
}
if ((query as PreparedQuery).params !== undefined) {
return executeQuery(query as PreparedQuery);
}
return executeQuery((query as QueryTemplate).bind(params));
}),
);
return results.flat();
}
export async function upsertOne
(
queryOrPrepared: PreparedQuery
| string,
params?: P,
): Promise {
await executeQuery(queryOrPrepared, params);
}
export async function deleteOneOrMany(
queryOrPrepared: PreparedQuery
| string,
params?: P,
): Promise {
await executeQuery(queryOrPrepared, params);
}
export async function executeConditional(
queryOrPrepared: PreparedQuery
| string,
params?: P,
): Promise<{applied: boolean; rows: Array>}> {
const {cql, params: bound} = normalizeExecuteArgs(queryOrPrepared, params);
if (isUnsafePreparedStatement(cql)) {
throw new Error('Cannot prepare a statement that looks like `SELECT *`');
}
assertNoUndefinedParams(bound as Record);
const startTime = IS_DEV ? performance.now() : 0;
try {
const result = await client.execute(cql, bound, {prepare: true});
interface MaybeApplied {
wasApplied?: () => boolean;
}
const applied =
typeof (result as MaybeApplied).wasApplied === 'function' ? (result as MaybeApplied).wasApplied!() : false;
if (IS_DEV) {
const durationMs = performance.now() - startTime;
logQuery(
`${getQueryType(cql)} (LWT${applied ? ' applied' : ' not applied'})`,
cql,
bound as Record,
durationMs,
result.rows.length,
);
}
return {applied, rows: result.rows as Array>};
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
Logger.warn({error: errorMessage, query: cql}, 'Cassandra conditional query failed');
throw err;
}
}
interface BatchQuery {
query: string;
params: object;
}
async function executeBatch(queries: Array, atomic = true): Promise {
if (queries.length === 0) return;
for (const {query} of queries) {
if (isUnsafePreparedStatement(query)) {
throw new Error('Cannot prepare a statement that looks like `SELECT *`');
}
}
for (const {params} of queries) {
assertNoUndefinedParams(params as Record);
}
const options = {
prepare: true,
logged: atomic,
counter: false,
};
const startTime = IS_DEV ? performance.now() : 0;
await client.batch(
queries.map(({query, params}) => ({query, params})),
options,
);
if (IS_DEV) {
const durationMs = performance.now() - startTime;
logBatch(queries, durationMs);
}
}
export class BatchBuilder {
private queries: Array = [];
add(query: string, params: object): this {
this.queries.push({query, params});
return this;
}
addPrepared(q: PreparedQuery): this {
this.queries.push({query: q.cql, params: q.params});
return this;
}
addIf(condition: boolean, query: string, params: object): this {
if (condition) this.queries.push({query, params});
return this;
}
addPreparedIf(condition: boolean, q: PreparedQuery): this {
if (condition) this.queries.push({query: q.cql, params: q.params});
return this;
}
async execute(atomic = true): Promise {
if (this.queries.length === 0) return;
await executeBatch(this.queries, atomic);
}
getQueries(): Array {
return this.queries;
}
}
export type WhereExpr =
| {kind: 'eq'; col: ColumnName; param: string}
| {kind: 'in'; col: ColumnName; param: string}
| {kind: 'lt'; col: ColumnName; param: string}
| {kind: 'gt'; col: ColumnName; param: string}
| {kind: 'lte'; col: ColumnName; param: string}
| {kind: 'gte'; col: ColumnName; param: string}
| {kind: 'tokenGt'; col: ColumnName; param: string}
| {kind: 'tupleGt'; cols: ReadonlyArray>; params: ReadonlyArray};
export type OrderBy = {col: ColumnName; direction?: 'ASC' | 'DESC'};
export interface Table, PartKey extends ColumnName = PK> {
name: string;
columns: ReadonlyArray>;
primaryKey: ReadonlyArray;
partitionKey: ReadonlyArray;
selectCql(opts?: {
columns?: ReadonlyArray>;
where?: WhereExpr | ReadonlyArray>;
orderBy?: OrderBy;
limit?: number;
}): string;
select(opts?: {
columns?: ReadonlyArray>;
where?: WhereExpr | ReadonlyArray>;
orderBy?: OrderBy;
limit?: number;
}): QueryTemplate;
updateAllCql(): string;
paramsFromRow(row: Row): CassandraParams;
upsertAll(row: Row): PreparedQuery;
patchByPk(
pk: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
): PreparedQuery;
deleteCql(opts?: {where?: WhereExpr | ReadonlyArray>}): string;
delete(opts?: {where?: WhereExpr | ReadonlyArray>}): QueryTemplate;
deleteByPk(pk: Pick): PreparedQuery;
deletePartition(pk: Pick): PreparedQuery;
insertCql(opts?: {ttlParam?: string}): string;
insert(row: Row): PreparedQuery;
insertWithTtl(row: Row, ttlSeconds: number): PreparedQuery;
insertWithTtlParam(row: Row, ttlParamName: string): PreparedQuery;
insertIfNotExists(row: Row): PreparedQuery;
selectCountCql(opts?: {where?: WhereExpr | ReadonlyArray>}): string;
selectCount(opts?: {where?: WhereExpr | ReadonlyArray>}): QueryTemplate;
insertWithNow>(row: Omit, nowColumn: NowCol): PreparedQuery;
patchByPkWithTtl(
pk: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
ttlSeconds: number,
): PreparedQuery;
patchByPkWithTtlParam(
pk: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
ttlParamName: string,
ttlValue: number,
): PreparedQuery;
upsertAllWithTtl(row: Row, ttlSeconds: number): PreparedQuery;
upsertAllWithTtlParam(row: Row, ttlParamName: string, ttlValue: number): PreparedQuery;
patchByPkIf, PK>>(
pk: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
condition: {col: CondCol; expectedParam: string; expectedValue: RowValue},
): PreparedQuery;
where: {
eq: >(col: K, param?: string) => WhereExpr;
in: >(col: K, param: string) => WhereExpr;
lt: >(col: K, param?: string) => WhereExpr;
gt: >(col: K, param?: string) => WhereExpr;
lte: >(col: K, param?: string) => WhereExpr;
gte: >(col: K, param?: string) => WhereExpr;
tokenGt: >(col: K, param: string) => WhereExpr;
tupleGt: >(cols: ReadonlyArray, params: ReadonlyArray) => WhereExpr;
};
}
function compileWhere(w: WhereExpr): string {
switch (w.kind) {
case 'eq':
return `${w.col} = :${w.param}`;
case 'in':
return `${w.col} IN :${w.param}`;
case 'lt':
return `${w.col} < :${w.param}`;
case 'gt':
return `${w.col} > :${w.param}`;
case 'lte':
return `${w.col} <= :${w.param}`;
case 'gte':
return `${w.col} >= :${w.param}`;
case 'tokenGt':
return `TOKEN(${w.col}) > TOKEN(:${w.param})`;
case 'tupleGt': {
if (w.cols.length !== w.params.length || w.cols.length === 0) {
throw new Error('tupleGt requires equal-length non-empty cols/params');
}
const cols = `(${w.cols.join(', ')})`;
const params = `(${w.params.map((p) => `:${p}`).join(', ')})`;
return `${cols} > ${params}`;
}
default: {
const _exhaustive: never = w;
return _exhaustive;
}
}
}
function opToValue(op: DbOp): CassandraParam {
return op.kind === 'clear' ? null : (op.value as CassandraParam);
}
export function defineTable, PartKey extends ColumnName = PK>(def: {
name: string;
columns: ReadonlyArray>;
primaryKey: ReadonlyArray;
partitionKey?: ReadonlyArray;
}): Table {
const columns = [...def.columns];
const pk = [...def.primaryKey];
const partitionKey = [...(def.partitionKey ?? def.primaryKey)] as Array;
const nonPkColumns = columns.filter((c) => !pk.includes(c as PK)) as Array, PK>>;
const updateAll =
nonPkColumns.length > 0
? `UPDATE ${def.name}
SET ${nonPkColumns.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`
: `INSERT INTO ${def.name} (${columns.join(', ')}) VALUES (${columns.map((c) => `:${c}`).join(', ')});`;
function paramsFromRow(row: Row, requireAll: boolean = true): CassandraParams {
const params: CassandraParams = {};
for (const c of columns) {
const v = row[c as keyof Row];
if (v === undefined) {
if (requireAll) {
throw new Error(
`Row is missing value for "${def.name}.${c}". Full-row upserts require every column to be present (use patchByPk() for partial writes).`,
);
}
continue;
}
params[c] = v as CassandraParam;
}
return params;
}
function buildDynamicUpsertCql(row: Row): {cql: string; params: CassandraParams} {
const presentColumns: Array = [];
const params: CassandraParams = {};
for (const c of columns) {
const v = row[c as keyof Row];
if (v !== undefined) {
presentColumns.push(c);
params[c] = v as CassandraParam;
}
}
const nonPkColumns = presentColumns.filter((c) => !pk.includes(c as PK));
const cql =
nonPkColumns.length > 0
? `UPDATE ${def.name}
SET ${nonPkColumns.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`
: `INSERT INTO ${def.name} (${pk.join(', ')}) VALUES (${pk.map((c) => `:${c}`).join(', ')});`;
return {cql, params};
}
function selectCql(
opts: {
columns?: ReadonlyArray>;
where?: WhereExpr | ReadonlyArray>;
orderBy?: OrderBy;
limit?: number;
} = {},
): string {
const selectCols = (opts.columns ?? columns).join(', ');
let where = '';
if (opts.where) {
const clauses = Array.isArray(opts.where) ? opts.where : [opts.where];
if (clauses.length > 0) {
where = ` WHERE ${clauses.map((c) => compileWhere(c)).join(' AND ')}`;
}
}
const orderBy = opts.orderBy != null ? ` ORDER BY ${opts.orderBy.col} ${opts.orderBy.direction ?? 'ASC'}` : '';
const limit = typeof opts.limit === 'number' ? ` LIMIT ${opts.limit}` : '';
return `SELECT ${selectCols} FROM ${def.name}${where}${orderBy}${limit};`;
}
function select(
opts: {
columns?: ReadonlyArray>;
where?: WhereExpr | ReadonlyArray>;
orderBy?: OrderBy;
limit?: number;
} = {},
): QueryTemplate {
const cql = selectCql(opts);
return {
cql,
bind(params: CassandraParams) {
return prepared(cql, params);
},
};
}
function patchByPk(
pkValues: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
): PreparedQuery {
const patchKeys = Object.keys(patch) as Array, PK>>;
if (patchKeys.length === 0) {
throw new Error(`Refusing to execute empty PATCH update on table "${def.name}"`);
}
patchKeys.sort((a, b) => columns.indexOf(a) - columns.indexOf(b));
const cql = `UPDATE ${def.name}
SET ${patchKeys.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`;
const params: CassandraParams = {};
for (const k of pk) params[k] = pkValues[k] as CassandraParam;
for (const c of patchKeys) params[c] = opToValue(patch[c] as DbOp);
return prepared(cql, params);
}
const deleteByPkCql = `DELETE FROM ${def.name} WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};`;
function deleteCql(opts: {where?: WhereExpr | ReadonlyArray>} = {}): string {
let where = '';
if (opts.where) {
const clauses = Array.isArray(opts.where) ? opts.where : [opts.where];
if (clauses.length > 0) {
where = ` WHERE ${clauses.map((c) => compileWhere(c)).join(' AND ')}`;
}
} else {
where = ` WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')}`;
}
return `DELETE FROM ${def.name}${where};`;
}
function del(opts: {where?: WhereExpr | ReadonlyArray>} = {}): QueryTemplate {
const cql = deleteCql(opts);
return {
cql,
bind(params: CassandraParams) {
return prepared(cql, params);
},
};
}
function deleteByPk(pkValues: Pick): PreparedQuery {
const params: CassandraParams = {};
for (const k of pk) params[k] = pkValues[k] as CassandraParam;
return prepared(deleteByPkCql, params);
}
function deletePartition(partKeyValues: Pick): PreparedQuery {
if (partitionKey.length === 0) {
throw new Error(`Table "${def.name}" has empty partitionKey; cannot deletePartition()`);
}
const cql = `DELETE FROM ${def.name} WHERE ${partitionKey.map((k) => `${k} = :${k}`).join(' AND ')};`;
const params: CassandraParams = {};
for (const k of partitionKey) params[k] = (partKeyValues as Record)[k];
return prepared(cql, params);
}
const insertBaseCql = `INSERT INTO ${def.name} (${columns.join(', ')}) VALUES (${columns.map((c) => `:${c}`).join(', ')})`;
function insertCql(opts: {ttlParam?: string} = {}): string {
if (opts.ttlParam) return `${insertBaseCql} USING TTL :${opts.ttlParam};`;
return `${insertBaseCql};`;
}
function insert(row: Row): PreparedQuery {
return prepared(`${insertBaseCql};`, paramsFromRow(row));
}
function insertWithTtl(row: Row, ttlSeconds: number): PreparedQuery {
const cql = `${insertBaseCql} USING TTL ${ttlSeconds};`;
return prepared(cql, paramsFromRow(row));
}
function insertWithTtlParam(row: Row, ttlParamName: string): PreparedQuery {
const cql = `${insertBaseCql} USING TTL :${ttlParamName};`;
const params = paramsFromRow(row);
if (params[ttlParamName] === undefined) {
params[ttlParamName] = row[ttlParamName as keyof Row] as CassandraParam;
}
return prepared(cql, params as CassandraParams);
}
function insertIfNotExists(row: Row): PreparedQuery {
const cql = `${insertBaseCql} IF NOT EXISTS;`;
return prepared(cql, paramsFromRow(row));
}
function selectCountCql(opts: {where?: WhereExpr | ReadonlyArray>} = {}): string {
let where = '';
if (opts.where) {
const clauses = Array.isArray(opts.where) ? opts.where : [opts.where];
if (clauses.length > 0) {
where = ` WHERE ${clauses.map((c) => compileWhere(c)).join(' AND ')}`;
}
}
return `SELECT COUNT(*) as count FROM ${def.name}${where};`;
}
function selectCount(opts: {where?: WhereExpr | ReadonlyArray>} = {}): QueryTemplate {
const cql = selectCountCql(opts);
return {
cql,
bind(params: CassandraParams) {
return prepared(cql, params);
},
};
}
function insertWithNow>(row: Omit, nowColumn: NowCol): PreparedQuery {
const otherColumns = columns.filter((c) => c !== nowColumn);
const allCols = [...otherColumns, nowColumn];
const values = otherColumns.map((c) => `:${c}`).concat(['now()']);
const cql = `INSERT INTO ${def.name} (${allCols.join(', ')}) VALUES (${values.join(', ')});`;
const params: CassandraParams = {};
for (const c of otherColumns) {
if (c === nowColumn) continue;
const v = (row as Record)[c];
if (v === undefined) {
throw new Error(`Row is missing value for "${def.name}.${c}". INSERT requires every column to be present.`);
}
params[c] = v as CassandraParam;
}
return prepared(cql, params);
}
function patchByPkWithTtl(
pkValues: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
ttlSeconds: number,
): PreparedQuery {
const patchKeys = Object.keys(patch) as Array, PK>>;
if (patchKeys.length === 0) {
throw new Error(`Refusing to execute empty PATCH update on table "${def.name}"`);
}
patchKeys.sort((a, b) => columns.indexOf(a) - columns.indexOf(b));
const cql = `UPDATE ${def.name} USING TTL ${ttlSeconds}
SET ${patchKeys.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`;
const params: CassandraParams = {};
for (const k of pk) params[k] = pkValues[k] as CassandraParam;
for (const c of patchKeys) params[c] = opToValue(patch[c] as DbOp);
return prepared(cql, params);
}
function patchByPkWithTtlParam(
pkValues: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
ttlParamName: string,
ttlValue: number,
): PreparedQuery {
const patchKeys = Object.keys(patch) as Array, PK>>;
if (patchKeys.length === 0) {
throw new Error(`Refusing to execute empty PATCH update on table "${def.name}"`);
}
patchKeys.sort((a, b) => columns.indexOf(a) - columns.indexOf(b));
const cql = `UPDATE ${def.name} USING TTL :${ttlParamName}
SET ${patchKeys.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`;
const params: CassandraParams = {};
for (const k of pk) params[k] = pkValues[k] as CassandraParam;
for (const c of patchKeys) params[c] = opToValue(patch[c] as DbOp);
params[ttlParamName] = ttlValue;
return prepared(cql, params);
}
function upsertAllWithTtl(row: Row, ttlSeconds: number): PreparedQuery {
const cql =
nonPkColumns.length > 0
? `UPDATE ${def.name} USING TTL ${ttlSeconds}
SET ${nonPkColumns.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`
: `INSERT INTO ${def.name} (${columns.join(', ')}) VALUES (${columns.map((c) => `:${c}`).join(', ')}) USING TTL ${ttlSeconds};`;
return prepared(cql, paramsFromRow(row));
}
function upsertAllWithTtlParam(row: Row, ttlParamName: string, ttlValue: number): PreparedQuery {
const cql =
nonPkColumns.length > 0
? `UPDATE ${def.name} USING TTL :${ttlParamName}
SET ${nonPkColumns.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')};
`
: `INSERT INTO ${def.name} (${columns.join(', ')}) VALUES (${columns.map((c) => `:${c}`).join(', ')}) USING TTL :${ttlParamName};`;
const params = paramsFromRow(row);
params[ttlParamName] = ttlValue;
return prepared(cql, params);
}
function patchByPkIf, PK>>(
pkValues: Pick,
patch: Partial<{[K in Exclude, PK>]: DbOp>}>,
condition: {col: CondCol; expectedParam: string; expectedValue: RowValue},
): PreparedQuery {
const patchKeys = Object.keys(patch) as Array, PK>>;
if (patchKeys.length === 0) {
throw new Error(`Refusing to execute empty conditional PATCH update on table "${def.name}"`);
}
patchKeys.sort((a, b) => columns.indexOf(a) - columns.indexOf(b));
const cql = `UPDATE ${def.name}
SET ${patchKeys.map((c) => `${c} = :${c}`).join(', ')}
WHERE ${pk.map((k) => `${k} = :${k}`).join(' AND ')}
IF ${condition.col} = :${condition.expectedParam};
`;
const params: CassandraParams = {};
for (const k of pk) params[k] = pkValues[k] as CassandraParam;
for (const c of patchKeys) params[c] = opToValue(patch[c] as DbOp);
params[condition.expectedParam] = condition.expectedValue as CassandraParam;
return prepared(cql, params);
}
return {
name: def.name,
columns: def.columns,
primaryKey: def.primaryKey,
partitionKey: partitionKey,
selectCql,
select,
updateAllCql() {
return updateAll;
},
paramsFromRow,
upsertAll(row: Row) {
const hasAllColumns = columns.every((c) => row[c as keyof Row] !== undefined);
if (hasAllColumns) {
return prepared(updateAll, paramsFromRow(row));
}
const {cql, params} = buildDynamicUpsertCql(row);
return prepared(cql, params);
},
patchByPk,
deleteCql,
delete: del,
deleteByPk,
deletePartition,
insertCql,
insert,
insertWithTtl,
insertWithTtlParam,
insertIfNotExists,
selectCountCql,
selectCount,
insertWithNow,
patchByPkWithTtl,
patchByPkWithTtlParam,
upsertAllWithTtl,
upsertAllWithTtlParam,
patchByPkIf,
where: {
eq: (col, param) => ({kind: 'eq', col, param: param ?? col}),
in: (col, param) => ({kind: 'in', col, param}),
lt: (col, param) => ({kind: 'lt', col, param: param ?? col}),
gt: (col, param) => ({kind: 'gt', col, param: param ?? col}),
lte: (col, param) => ({kind: 'lte', col, param: param ?? col}),
gte: (col, param) => ({kind: 'gte', col, param: param ?? col}),
tokenGt: (col, param) => ({kind: 'tokenGt', col, param}),
tupleGt: (cols, params) => ({kind: 'tupleGt', cols, params}),
},
};
}
const DEFAULT_LWT_RETRIES = 8;
export type PatchObject = {[key: string]: DbOp};
export async function executeVersionedUpdate<
Row extends {version?: number | null},
PK extends ColumnName,
Patch extends PatchObject = PatchObject,
>(
fetchCurrent: () => Promise,
buildPatch: (current: Row | null) => {pk: Record; patch: Patch},
table: Table,
opts?: {maxRetries?: number; onFailure?: 'throw' | 'log' | 'silent'},
): Promise<{applied: boolean; finalVersion: number | null}> {
const maxRetries = opts?.maxRetries ?? DEFAULT_LWT_RETRIES;
for (let i = 0; i < maxRetries; i++) {
const current = await fetchCurrent();
const currentVersion = current?.version ?? null;
const newVersion = (currentVersion ?? 0) + 1;
const {pk, patch} = buildPatch(current);
const q = table.patchByPkIf(
pk as Pick,
{...patch, version: Db.set(newVersion)} as Partial<{[K in Exclude, PK>]: DbOp>}>,
{
col: 'version' as Exclude, PK>,
expectedParam: 'prev_version',
expectedValue: currentVersion as RowValue, PK>>,
},
);
const res = await executeConditional(q);
if (res.applied) {
return {applied: true, finalVersion: newVersion};
}
}
if (opts?.onFailure === 'throw') {
throw new Error('LWT update failed after max retries');
} else if (opts?.onFailure !== 'silent') {
Logger.warn({}, 'LWT update failed after max retries');
}
return {applied: false, finalVersion: null};
}
export function buildPatchFromData(
newData: Partial,
oldData: Partial | null,
columns: ReadonlyArray,
pkColumns: ReadonlyArray,
): Record> {
const patch: Record> = {};
for (const col of columns) {
if (pkColumns.includes(col)) continue;
if (col === 'version') continue;
const colName = col as string;
const newVal = (newData as Record)[colName];
if (newVal === undefined) continue;
const oldVal = oldData ? (oldData as Record)[colName] : undefined;
if (newVal === null) {
if (oldData !== null && oldVal !== null && oldVal !== undefined) {
patch[colName] = Db.clear();
}
} else {
patch[colName] = Db.set(newVal);
}
}
return patch;
}