Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .changeset/typed-paths-and-transaction-locks.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
---
"@cleverbrush/server": minor
"@cleverbrush/client": patch
"@cleverbrush/orm": minor
"@cleverbrush/schema": patch
---

Add typed endpoint pathname construction and matching, shared with client URL generation and server routing. Route parameters are validated and encoded as individual components.

Add optional PostgreSQL advisory locks to ORM transactions, acquired before callbacks and released with their transaction.

Reject non-object inputs to open object schemas with inspectable validation errors, including nested arrays, instead of throwing descriptor errors.
8 changes: 8 additions & 0 deletions libs/client/src/path.test.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,15 @@
import { string } from '@cleverbrush/schema';
import { route } from '@cleverbrush/server/contract';
import { describe, expect, test, vi } from 'vitest';
import { buildPath } from './path.js';

describe('buildPath', () => {
test('encodes real contract parameters without changing literal separators', () => {
const template = route({ id: string() })`/${p => p.id}/content`;
expect(buildPath('/files', template, { id: 'a/b%2F?#' })).toBe(
'/files/a%2Fb%252F%3F%23/content'
);
});
test('plain string template — appends to basePath', () => {
expect(buildPath('/api/todos', '')).toBe('/api/todos');
expect(buildPath('/api/todos', '/active')).toBe('/api/todos/active');
Expand Down
8 changes: 7 additions & 1 deletion libs/client/src/path.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { endpointPath } from '@cleverbrush/server/contract';
/**
* Builds the full request URL from endpoint metadata and optional path
* parameters.
Expand All @@ -24,7 +25,12 @@ export function buildPath(
}

if (typeof pathTemplate.serialize === 'function') {
return basePath + pathTemplate.serialize(params ?? {});
return typeof (pathTemplate as any).introspect === 'function'
? endpointPath(
{ introspect: () => ({ basePath, pathTemplate }) } as any,
params ?? {}
)
: basePath + pathTemplate.serialize(params ?? {});
}

return basePath;
Expand Down
24 changes: 24 additions & 0 deletions libs/orm/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -521,3 +521,27 @@ Editing a nested object or array marks the column modified; replacing it with an
equivalent document does not. `entry(entity).reset()` and `discardChanges()` restore
independent copies, so subsequent edits cannot mutate the saved snapshot. JSON
validation and encoding also apply to ORM saves and tracked updates.

### Transaction advisory locks

Use stable, namespaced keys to serialize operations that can race before a row
exists. PostgreSQL releases these locks when their owning transaction commits or
rolls back. The callback receives the same transaction-bound context as ordinary
transactions; tracked contexts retain tracking.

```ts
await db.transaction(async tx => {
// Read and write the resource under a transaction-scoped lock.
}, { advisoryLocks: [`resource:${resourceId}`] });
```

Keys must be nonempty strings. Duplicate keys are acquired once, in sorted order.
This prevents inconsistent ordering within one lock set; callers must avoid
inconsistent ordering across nested transactions. The database's configured lock
timeout applies. Unsupported dialects reject advisory locks before the callback
runs; ordinary transactions continue to work. Locks do not replace constraints or
authorization.

## License

BSD-3-Clause — see [LICENSE](../../LICENSE).
81 changes: 81 additions & 0 deletions libs/orm/integration/transaction-locks.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
import { randomUUID } from 'node:crypto';
import Knex from 'knex';
import { afterAll, describe, expect, it } from 'vitest';
import { createDb } from '../src/index.js';

const connection = process.env.QUERY_TEST_DATABASE_URL;
if (!connection) throw new Error('QUERY_TEST_DATABASE_URL is required');
const knex = Knex({ client: 'pg', connection, pool: { min: 0, max: 5 } });
const db = createDb(knex, {});
afterAll(() => knex.destroy());

describe('PostgreSQL advisory transactions', () => {
it.each(['commit', 'rollback'])(
'releases locks after %s and allows unrelated keys',
async outcome => {
const key = `transaction-test:${randomUUID()}`;
let release!: () => void;
let acquired!: () => void;
const ready = new Promise<void>(resolve => {
acquired = resolve;
});
const hold = new Promise<void>(resolve => {
release = resolve;
});
const first = db.transaction(
async () => {
acquired();
await hold;
if (outcome === 'rollback')
throw new Error('rollback fixture');
},
{ advisoryLocks: [key] }
);
const finished = first.catch(error => {
if (outcome !== 'rollback') throw error;
});
try {
await ready;
const locked = await knex.transaction(async trx => {
const result = await trx.raw(
'SELECT pg_try_advisory_xact_lock(hashtextextended(?, 0)) AS acquired',
[key]
);
return result.rows[0].acquired;
});
expect(locked).toBe(false);
expect(
await db.transaction(async () => 'independent', {
advisoryLocks: [key + ':other']
})
).toBe('independent');
} finally {
release();
await finished;
}
expect(
await db.transaction(async () => 'reacquired', {
advisoryLocks: [key]
})
).toBe('reacquired');
}
);
it('serializes callers with the same key', async () => {
const key = `transaction-test:${randomUUID()}`;
let active = 0;
let maximum = 0;
await Promise.all(
Array.from({ length: 4 }, () =>
db.transaction(
async scope => {
maximum = Math.max(maximum, ++active);
await scope.knex.raw('SELECT 1');
active--;
},
{ advisoryLocks: [key] }
)
)
);
expect(maximum).toBe(1);
});
});
44 changes: 41 additions & 3 deletions libs/orm/src/dbcontext.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,37 @@ import {
import { type DbSet, makeDbSet } from './dbset.js';
import { PendingChangesError } from './errors.js';

/** Options applied before a transaction callback runs. */
export interface TransactionOptions {
/**
* PostgreSQL transaction-scoped advisory locks. Use stable, namespaced keys.
* Keys are deduplicated and sorted to give overlapping batches a consistent
* acquisition order. Locks remain held until the owning transaction ends;
* they do not replace row constraints or authorization.
*/
readonly advisoryLocks?: readonly string[];
}

/** Acquire locks using parameter bindings on the transaction's connection. */
async function acquireTransactionLocks(
trx: Knex.Transaction,
options?: TransactionOptions
): Promise<void> {
const keys = options?.advisoryLocks ?? [];
if (!keys.length) return;
if (!['pg', 'postgres', 'postgresql'].includes(trx.client.config.client)) {
throw new Error('Advisory transaction locks require PostgreSQL.');
}
if (keys.some(key => typeof key !== 'string' || key.length === 0)) {
throw new TypeError('Advisory lock keys must be nonempty strings.');
}
for (const key of [...new Set(keys)].sort()) {
await trx.raw('SELECT pg_advisory_xact_lock(hashtextextended(?, 0))', [
key
]);
}
}

/**
* The map of entity name → entity definition passed to {@link createDb}.
*
Expand All @@ -44,7 +75,10 @@ export type DbContext<TMap extends EntityMap> = {
* Run `callback` inside a Knex transaction. The callback receives a new
* `DbContext` whose `DbSet`s are bound to the transaction.
*/
transaction<T>(callback: (db: DbContext<TMap>) => Promise<T>): Promise<T>;
transaction<T>(
callback: (db: DbContext<TMap>) => Promise<T>,
options?: TransactionOptions
): Promise<T>;

/**
* Return a new `DbContext` whose `DbSet`s are bound to `trx`. Useful
Expand Down Expand Up @@ -148,9 +182,11 @@ function buildContext<TMap extends EntityMap>(
return buildContext<TMap>(trx as unknown as Knex, entities);
},
async transaction<T>(
callback: (db: DbContext<TMap>) => Promise<T>
callback: (db: DbContext<TMap>) => Promise<T>,
options?: TransactionOptions
): Promise<T> {
return knex.transaction(async (trx: Knex.Transaction) => {
await acquireTransactionLocks(trx, options);
return callback(
buildContext<TMap>(trx as unknown as Knex, entities)
);
Expand Down Expand Up @@ -198,9 +234,11 @@ function buildTrackedContext<TMap extends EntityMap>(
},

async transaction<T>(
callback: (db: DbContext<TMap>) => Promise<T>
callback: (db: DbContext<TMap>) => Promise<T>,
options?: TransactionOptions
): Promise<T> {
return knex.transaction(async (trx: Knex.Transaction) => {
await acquireTransactionLocks(trx, options);
return callback(
buildTrackedContext<TMap>(
trx as unknown as Knex,
Expand Down
3 changes: 2 additions & 1 deletion libs/orm/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@ export type {
export type {
DbContext,
EntityMap,
TrackedDbContext
TrackedDbContext,
TransactionOptions
} from './dbcontext.js';
export { createDb } from './dbcontext.js';
export type {
Expand Down
61 changes: 61 additions & 0 deletions libs/orm/src/transaction-locks.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
import type { Knex } from 'knex';
import { describe, expect, it, vi } from 'vitest';
import { createDb } from './index.js';

function database(client = 'pg', tracking = false) {
const raw = vi.fn().mockResolvedValue(undefined);
const transaction = { client: { config: { client } }, raw };
const knex = {
transaction: vi.fn(async callback => callback(transaction))
} as unknown as Knex;
return { db: createDb(knex, {}, { tracking }), raw, transaction };
}

describe('advisory transaction options', () => {
it.each([false, true])(
'locks once in stable order before callback (tracking=%s)',
async tracking => {
const { db, raw, transaction } = database('pg', tracking);
const value = await db.transaction(
async scope => {
expect(scope.knex).toBe(transaction);
expect(raw.mock.calls.map(call => call[1])).toEqual([
['asset:a'],
['asset:z']
]);
return 'done';
},
{ advisoryLocks: ['asset:z', 'asset:a', 'asset:z'] }
);
expect(value).toBe('done');
}
);
it('preserves transactions without locking on other dialects', async () => {
const { db, raw } = database('sqlite3');
expect(await db.transaction(async () => 7)).toBe(7);
expect(raw).not.toHaveBeenCalled();
});
it('rejects unsupported dialects and invalid keys before running the callback', async () => {
const callback = vi.fn();
await expect(
database('sqlite3').db.transaction(callback, {
advisoryLocks: ['one']
})
).rejects.toThrow('PostgreSQL');
for (const keys of [[''], [7] as any]) {
await expect(
database().db.transaction(callback, { advisoryLocks: keys })
).rejects.toThrow('nonempty strings');
}
expect(callback).not.toHaveBeenCalled();
});
it('propagates lock errors without executing work', async () => {
const { db, raw } = database();
raw.mockRejectedValue(new Error('lock timeout'));
const callback = vi.fn();
await expect(
db.transaction(callback, { advisoryLocks: ['one'] })
).rejects.toThrow('lock timeout');
expect(callback).not.toHaveBeenCalled();
});
});
62 changes: 62 additions & 0 deletions libs/schema/src/builders/ObjectSchemaBuilder.boundaries.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
import { describe, expect, it } from 'vitest';
import { array, object, string } from '../index.js';

describe('open object boundaries', () => {
it.each([0, 'text', true, [], [1], null, undefined])(
'rejects %j with inspectable errors',
async value => {
const schema = object({}).acceptUnknownProps();
const result = schema.safeParse(value);
expect(result.valid).toBe(false);
const full = schema.validate(value as any, {
doNotStopOnFirstError: true
});
expect(full.valid).toBe(false);
expect(full.getErrorsFor().isValid).toBe(false);
expect((await schema.validateAsync(value as any)).valid).toBe(
false
);
const list = array(schema).validate([value] as any);
expect(list.valid).toBe(false);
expect(list.getNestedErrors()[0].getErrorsFor().isValid).toBe(
false
);
expect(
(await array(schema).validateAsync([value] as any)).valid
).toBe(false);
}
);
it('keeps nested extensions and empty arrays without mutating the input', () => {
const schema = object({
rows: array(
object({
data: object({}).acceptUnknownProps(),
items: array(string())
})
)
});
const value = {
rows: [
{
data: { extension: { empty: [], nullable: null } },
items: []
}
]
};
const result = schema.parse(value);
expect(result).toEqual(value);
expect(Object.getOwnPropertySymbols(result.rows[0].items)).toEqual([]);
expect(Object.getOwnPropertySymbols(value.rows[0].items)).toEqual([]);
});
it('preserves optional, nullable and preprocessor behavior', () => {
expect(object({}).optional().safeParse(undefined).valid).toBe(true);
expect(object({}).nullable().safeParse(null).valid).toBe(true);
expect(
object({ name: string() })
.addPreprocessor(value =>
typeof value === 'string' ? { name: value } : value
)
.parse('Ada' as any)
).toEqual({ name: 'Ada' });
});
});
Loading
Loading